diff --git a/pkg/storage/fs/posix/posix.go b/pkg/storage/fs/posix/posix.go index ecb68fe2cd2..0164bd349da 100644 --- a/pkg/storage/fs/posix/posix.go +++ b/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) diff --git a/pkg/storage/uploads.go b/pkg/storage/uploads.go index 44b70274534..3156638616e 100644 --- a/pkg/storage/uploads.go +++ b/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 diff --git a/pkg/storage/utils/decomposedfs/upload.go b/pkg/storage/utils/decomposedfs/upload.go index 7ae2034e908..cff407587a5 100644 --- a/pkg/storage/utils/decomposedfs/upload.go +++ b/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 diff --git a/pkg/storage/utils/middleware/middleware.go b/pkg/storage/utils/middleware/middleware.go index dd57f73cae4..81ce879d363 100644 --- a/pkg/storage/utils/middleware/middleware.go +++ b/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) diff --git a/pkg/upload/coordinator.go b/pkg/upload/coordinator.go index 801856218ee..9af3bab176c 100644 --- a/pkg/upload/coordinator.go +++ b/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 diff --git a/pkg/upload/events_test.go b/pkg/upload/events_test.go index 0f84a90f045..b6e14c4d106 100644 --- a/pkg/upload/events_test.go +++ b/pkg/upload/events_test.go @@ -257,8 +257,9 @@ 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}) @@ -266,6 +267,67 @@ var _ = Describe("coordinator events", func() { 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) diff --git a/pkg/upload/fakefs_test.go b/pkg/upload/fakefs_test.go index deb6557d805..3755e9c65d8 100644 --- a/pkg/upload/fakefs_test.go +++ b/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()] +}