feat(upload): [OCISDEV-1382] let the coordinator serve the orphaned session

pkg/storage/fs/posix/posix.go

@@ -27,6 +27,7 @@ import (
 	"os"
 	"syscall"
 
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/rs/zerolog"
 	tusd "github.com/tus/tusd/v2/pkg/handler"
 	microstore "go-micro.dev/v4/store"
@@ -185,6 +186,11 @@ func (fs *posixFS) ListUploadSessions(ctx context.Context, filter storage.Upload
 	return fs.FS.(storage.UploadSessionLister).ListUploadSessions(ctx, filter)
 }
 
+// IsOrphaned reports whether the referenced resource exists but its metadata is unreadable.
+func (fs *posixFS) IsOrphaned(ctx context.Context, ref *provider.Reference) bool {
+	return fs.FS.(storage.OrphanChecker).IsOrphaned(ctx, ref)
+}
+
 // UseIn tells the tus upload middleware which extensions it supports.
 func (fs *posixFS) UseIn(composer *tusd.StoreComposer) {
 	fs.FS.(storage.ComposableFS).UseIn(composer)

pkg/storage/uploads.go

@@ -51,6 +51,12 @@ type UploadSessionLister interface {
 	ListUploadSessions(ctx context.Context, filter UploadSessionFilter) ([]UploadSession, error)
 }
 
+// OrphanChecker defines the interface for FS implementations that can resolve a resource's metadata.
+type OrphanChecker interface {
+	// IsOrphaned reports whether the referenced resource exists but its metadata is unreadable.
+	IsOrphaned(ctx context.Context, ref *provider.Reference) bool
+}
+
 // UploadSession is the interface that storage drivers need to return whan listing upload sessions.
 type UploadSession interface {
 	// ID returns the upload id

pkg/storage/utils/decomposedfs/upload.go

@@ -830,6 +830,12 @@ func (fs *Decomposedfs) ListUploadSessions(ctx context.Context, filter storage.U
 	return filteredSessions, nil
 }
 
+// IsOrphaned reports whether the referenced resource exists but its metadata is unreadable.
+func (fs *Decomposedfs) IsOrphaned(ctx context.Context, ref *provider.Reference) bool {
+	_, err := fs.lu.NodeFromResource(ctx, ref)
+	return err != nil
+}
+
 // AsTerminatableUpload returns a TerminatableUpload
 // To implement the termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination
 // the storage needs to implement AsTerminatableUpload

pkg/storage/utils/middleware/middleware.go

@@ -55,6 +55,11 @@ func (f *FS) ListUploadSessions(ctx context.Context, filter storage.UploadSessio
 	return f.next.(storage.UploadSessionLister).ListUploadSessions(ctx, filter)
 }
 
+// IsOrphaned reports whether the referenced resource exists but its metadata is unreadable.
+func (f *FS) IsOrphaned(ctx context.Context, ref *provider.Reference) bool {
+	return f.next.(storage.OrphanChecker).IsOrphaned(ctx, ref)
+}
+
 // UseIn tells the tus upload middleware which extensions it supports.
 func (f *FS) UseIn(composer *tusd.StoreComposer) {
 	f.next.(storage.ComposableFS).UseIn(composer)

pkg/upload/coordinator.go

@@ -851,10 +851,14 @@ func rewriteChunkedRef(ref *provider.Reference) (*provider.Reference, string, er
 
 // ListUploadSessions returns the upload sessions matching the given filter.
 func (c *coordinator) ListUploadSessions(ctx context.Context, filter storage.UploadSessionFilter) ([]storage.UploadSession, error) {
-	// Only the driver can resolve a session's node, so refuse rather than
-	// silently report every session as a match.
+	// Only the driver can resolve a session's node, so the orphaned filter needs a
+	// driver that answers the question. Assert once rather than per session.
+	var orphanChecker storage.OrphanChecker
 	if filter.Orphaned != nil {
-		return nil, errtypes.NotSupported("coordinator: the orphaned filter is not supported")
+		var ok bool
+		if orphanChecker, ok = c.fs.(storage.OrphanChecker); !ok {
+			return nil, errtypes.NotSupported("coordinator: the orphaned filter is not supported")
+		}
 	}
 
 	var sessions []Session
@@ -896,6 +900,14 @@ func (c *coordinator) ListUploadSessions(ctx context.Context, filter storage.Upl
 				continue
 			}
 		}
+		// evaluated last: unlike the other filters this reads the node metadata
+		// from disk, so it is only done for sessions that passed all other filters
+		if filter.Orphaned != nil {
+			ref := session.Reference()
+			if *filter.Orphaned != orphanChecker.IsOrphaned(ctx, &ref) {
+				continue
+			}
+		}
 		filtered = append(filtered, session)
 	}
 	return filtered, nil

pkg/upload/events_test.go

@@ -257,15 +257,77 @@ var _ = Describe("coordinator events", func() {
 	})
 
 	Describe("ListUploadSessions", func() {
-		// Only the driver can resolve a session's node.
-		It("refuses the orphaned filter", func() {
+		// Only the driver can resolve a session's node, so a driver that does not
+		// implement storage.OrphanChecker cannot answer the question at all.
+		It("refuses the orphaned filter when the driver cannot answer it", func() {
 			orphaned := true
 
 			_, err := c.ListUploadSessions(ctx, storage.UploadSessionFilter{Orphaned: &orphaned})
 
 			Expect(err).To(BeAssignableToTypeOf(errtypes.NotSupported("")))
 		})
 
+		It("filters on whether the node is orphaned when the driver can answer it", func() {
+			healthy := stagedSession(ctx, store, true)
+			broken := stagedSession(ctx, store, true)
+			broken.SetStorageValue("NodeId", "orphaned-node")
+			Expect(broken.Persist(ctx)).To(Succeed())
+			c = NewCoordinator(&fakeOrphanFS{
+				fakeFS:   fs,
+				orphaned: map[string]bool{"orphaned-node": true},
+			}, store, "", nil)
+
+			isOrphaned := true
+			found, err := c.ListUploadSessions(ctx, storage.UploadSessionFilter{Orphaned: &isOrphaned})
+			Expect(err).ToNot(HaveOccurred())
+			Expect(found).To(HaveLen(1))
+			Expect(found[0].ID()).To(Equal(broken.ID()))
+
+			isOrphaned = false
+			rest, err := c.ListUploadSessions(ctx, storage.UploadSessionFilter{Orphaned: &isOrphaned})
+			Expect(err).ToNot(HaveOccurred())
+			Expect(rest).To(HaveLen(1))
+			Expect(rest[0].ID()).To(Equal(healthy.ID()))
+		})
+
+		// The driver resolves the node by id, so a reference missing the space or the
+		// node would silently report every session as healthy.
+		It("asks the driver about the session's own node", func() {
+			stagedSession(ctx, store, true)
+			orphanFS := &fakeOrphanFS{fakeFS: fs}
+			c = NewCoordinator(orphanFS, store, "", nil)
+
+			isOrphaned := false
+			_, err := c.ListUploadSessions(ctx, storage.UploadSessionFilter{Orphaned: &isOrphaned})
+
+			Expect(err).ToNot(HaveOccurred())
+			Expect(orphanFS.refs).To(HaveLen(1))
+			Expect(orphanFS.refs[0].GetResourceId().GetSpaceId()).To(Equal(spaceRoot))
+			Expect(orphanFS.refs[0].GetResourceId().GetOpaqueId()).To(Equal(nodeID))
+		})
+
+		// The filter reads node metadata, so it must not run for sessions another
+		// filter already excluded.
+		It("evaluates the orphaned filter only for sessions that passed the others", func() {
+			complete := stagedSession(ctx, store, true)
+			partial := stagedSession(ctx, store, true)
+			partial.SetSize(bodyLen * 2)
+			Expect(partial.Persist(ctx)).To(Succeed())
+			orphanFS := &fakeOrphanFS{fakeFS: fs}
+			c = NewCoordinator(orphanFS, store, "", nil)
+
+			processing, isOrphaned := true, false
+			sessions, err := c.ListUploadSessions(ctx, storage.UploadSessionFilter{
+				Processing: &processing,
+				Orphaned:   &isOrphaned,
+			})
+
+			Expect(err).ToNot(HaveOccurred())
+			Expect(sessions).To(HaveLen(1))
+			Expect(sessions[0].ID()).To(Equal(complete.ID()))
+			Expect(orphanFS.refs).To(HaveLen(1), "the incomplete session should not have been resolved")
+		})
+
 		It("returns every session when no filter is given", func() {
 			first := stagedSession(ctx, store, false)
 			second := stagedSession(ctx, store, false)

pkg/upload/fakefs_test.go

@@ -292,3 +292,20 @@ func (s *brokenStore) List(ctx context.Context) ([]Session, error) {
 	}
 	return s.SessionStore.List(ctx)
 }
+
+// fakeOrphanFS is a fakeFS that also implements storage.OrphanChecker. fakeFS
+// deliberately does not, so the specs can cover a driver that cannot answer.
+type fakeOrphanFS struct {
+	*fakeFS
+
+	// orphaned maps a node id to the verdict the driver reports for it.
+	orphaned map[string]bool
+
+	// refs records what the coordinator asked about, in order.
+	refs []*provider.Reference
+}
+
+func (f *fakeOrphanFS) IsOrphaned(_ context.Context, ref *provider.Reference) bool {
+	f.refs = append(f.refs, ref)
+	return f.orphaned[ref.GetResourceId().GetOpaqueId()]
+}