diff --git a/changelog/unreleased/fix-orphaned-upload-sessions.md b/changelog/unreleased/fix-orphaned-upload-sessions.md new file mode 100644 index 00000000000..fdea6c4aeb0 --- /dev/null +++ b/changelog/unreleased/fix-orphaned-upload-sessions.md @@ -0,0 +1,26 @@ +Bugfix: Release the quota of upload sessions with unreadable node metadata + +When an upload's target node lost its metadata, e.g. because an ancestor was +moved to the trash while the upload was still in flight, the node file remained +on disk without a readable `.mpk`. Reading such a node fails with +`Missing parent ID on node`, so the upload could never finish postprocessing. It +stayed in "Processing" forever, could not be downloaded or deleted, and kept +consuming the space quota. + +Cleaning these sessions up did not work either. `Cleanup` removed the upload +bytes and the session info file *before* attempting to revert the node, then +bailed out on the failing node read without ever releasing the quota. That +destroyed both the only copy of the uploaded data and the session metadata +needed to repair the node, while freeing nothing. + +Cleanup now reverts the node before removing anything irreversible and falls +back to the parent id recorded in the session when the node metadata cannot be +read, so the quota is released and the orphaned node is removed. If the quota +cannot be released the upload is kept so it can be retried instead of being lost. +Sessions whose node is unreadable are now also cleaned up when postprocessing +finishes, instead of being left behind to be retried indefinitely. + +A new `Orphaned` upload session filter allows listing the affected sessions. It +is only evaluated when set, as it reads the node metadata of every session. + +https://github.com/owncloud/reva/pull/692 diff --git a/pkg/storage/uploads.go b/pkg/storage/uploads.go index 59ff5088550..44b70274534 100644 --- a/pkg/storage/uploads.go +++ b/pkg/storage/uploads.go @@ -87,4 +87,8 @@ type UploadSessionFilter struct { Processing *bool Expired *bool HasVirus *bool + // Orphaned filters sessions by whether their target node can still be + // resolved. Evaluating it requires reading the node metadata of every + // session, so it is only evaluated when set. + Orphaned *bool } diff --git a/pkg/storage/utils/decomposedfs/decomposedfs.go b/pkg/storage/utils/decomposedfs/decomposedfs.go index 93e5630abea..4ad46fe3695 100644 --- a/pkg/storage/utils/decomposedfs/decomposedfs.go +++ b/pkg/storage/utils/decomposedfs/decomposedfs.go @@ -296,7 +296,12 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { n, err := session.Node(ctx) if err != nil { - sublog.Error().Err(err).Msg("could not read node") + // The node metadata is unreadable, so this upload can never finish: + // the destination cannot be resolved. Clean the session up instead of + // leaving it behind to be retried forever. Cleanup falls back to the + // session metadata to release the quota. + sublog.Error().Err(err).Msg("could not read node, cleaning up orphaned session") + session.Cleanup(true, true, true, false) continue } sublog = log.With().Str("spaceid", session.SpaceID()).Str("nodeid", session.NodeID()).Logger() diff --git a/pkg/storage/utils/decomposedfs/upload.go b/pkg/storage/utils/decomposedfs/upload.go index fac96b70632..af38d4ab958 100644 --- a/pkg/storage/utils/decomposedfs/upload.go +++ b/pkg/storage/utils/decomposedfs/upload.go @@ -411,6 +411,11 @@ func (fs *Decomposedfs) ListUploadSessions(ctx context.Context, filter storage.U 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 && *filter.Orphaned != session.IsOrphaned(ctx) { + continue + } filteredSessions = append(filteredSessions, session) } return filteredSessions, nil diff --git a/pkg/storage/utils/decomposedfs/upload/session.go b/pkg/storage/utils/decomposedfs/upload/session.go index 8d7a58ce68a..73c0e280347 100644 --- a/pkg/storage/utils/decomposedfs/upload/session.go +++ b/pkg/storage/utils/decomposedfs/upload/session.go @@ -34,6 +34,7 @@ import ( typespb "github.com/cs3org/go-cs3apis/cs3/types/v1beta1" "github.com/owncloud/reva/v2/pkg/appctx" ctxpkg "github.com/owncloud/reva/v2/pkg/ctx" + "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/reva/v2/pkg/storage/utils/decomposedfs/node" "github.com/owncloud/reva/v2/pkg/utils" ) @@ -166,6 +167,43 @@ func (s *OcisSession) Node(ctx context.Context) (*node.Node, error) { return node.ReadNode(ctx, s.store.lu, s.SpaceID(), s.info.Storage["NodeId"], false, nil, true) } +// IsOrphaned returns true if the session's target node can no longer be +// resolved. This happens when the node file still exists but its metadata is +// gone, e.g. because an ancestor was moved to the trash while the upload was in +// flight. Such a session can never finish postprocessing: reading the node +// fails before the destination can be determined. +func (s *OcisSession) IsOrphaned(ctx context.Context) bool { + _, err := s.Node(ctx) + return err != nil +} + +// syntheticNode builds a node from the session metadata alone, without reading +// the node from disk. It is used to clean up sessions whose node metadata is +// unreadable: the parent id is still recorded in the session, which is all that +// is needed to walk up the tree and revert the size propagation. +func (s *OcisSession) syntheticNode(ctx context.Context) (*node.Node, error) { + if s.NodeID() == "" || s.NodeParentID() == "" { + return nil, errtypes.InternalError("session has no node and parent id") + } + n := node.New( + s.SpaceID(), + s.NodeID(), + s.NodeParentID(), + s.Filename(), + s.Size(), + s.ID(), + provider.ResourceType_RESOURCE_TYPE_FILE, + nil, + s.store.lu, + ) + spaceRoot, err := node.ReadNode(ctx, s.store.lu, s.SpaceID(), s.SpaceID(), false, nil, false) + if err != nil { + return nil, err + } + n.SpaceRoot = spaceRoot + return n, nil +} + // ID returns the upload session id func (s *OcisSession) ID() string { return s.info.ID diff --git a/pkg/storage/utils/decomposedfs/upload/upload.go b/pkg/storage/utils/decomposedfs/upload/upload.go index 98f41db563e..62b0dfdce57 100644 --- a/pkg/storage/utils/decomposedfs/upload/upload.go +++ b/pkg/storage/utils/decomposedfs/upload/upload.go @@ -330,10 +330,71 @@ func (session *OcisSession) removeNode(ctx context.Context) { } } +// revertNode undoes the node changes made when the upload was initiated. For a +// readable node this restores the previous revision. When the node metadata can +// no longer be read the node is orphaned and can never finish postprocessing; in +// that case the node is removed and the optimistic size propagation is reverted +// using the parent id recorded in the session, so the space quota is released. +func (session *OcisSession) revertNode(ctx context.Context) error { + n, err := session.Node(ctx) + if err == nil { + curUpload, perr := n.ProcessingID(ctx) + if perr == nil && curUpload == session.ID() { + if rerr := n.RevertCurrentRevision(ctx); rerr != nil { + return rerr + } + } + return nil + } + + // The node is unreadable. Fall back to the session metadata, which still + // carries the node and parent ids needed to release the quota. + log := appctx.GetLogger(ctx) + log.Info().Err(err).Str("sessionid", session.ID()).Msg("node unreadable, cleaning up orphaned upload") + + sn, serr := session.syntheticNode(ctx) + if serr != nil { + return serr + } + + if sizeDiff := session.SizeDiff(); sizeDiff != 0 { + if perr := session.store.tp.Propagate(ctx, sn, -sizeDiff); perr != nil { + // Without the propagation the quota would stay consumed. Stop here + // so the session can be retried instead of losing the upload. + return perr + } + } + + // The orphaned node file cannot be resolved by any other means, remove it + // together with its metadata files. A missing node is not an error here: + // the node may never have been created. + nodePath := sn.InternalPath() + if rerr := utils.RemoveItem(nodePath); rerr != nil && !errors.Is(rerr, fs.ErrNotExist) { + log.Error().Err(rerr).Str("nodepath", nodePath).Msg("removing orphaned node failed") + } + if perr := session.store.lu.MetadataBackend().Purge(ctx, nodePath); perr != nil && !errors.Is(perr, fs.ErrNotExist) { + log.Error().Err(perr).Str("nodepath", nodePath).Msg("purging orphaned node metadata failed") + } + + return nil +} + // cleanup cleans up after the upload is finished func (session *OcisSession) Cleanup(revertNodeMetadata, cleanBin, cleanInfo, unmarkPostprocessing bool) { ctx := session.Context(context.Background()) + if revertNodeMetadata { + // Revert before removing the bin and info files. Both are needed to + // recover from a failure here: the bin file holds the only copy of the + // uploaded data as long as the blob has not been written, and the info + // file is the only remaining source of the node's parent id once the + // node metadata is gone. + if err := session.revertNode(ctx); err != nil { + appctx.GetLogger(ctx).Error().Err(err).Str("sessionid", session.ID()).Msg("reverting node failed, keeping upload") + return + } + } + if cleanBin { if err := os.Remove(session.binPath()); err != nil && !errors.Is(err, fs.ErrNotExist) { appctx.GetLogger(ctx).Error().Str("path", session.binPath()).Err(err).Msg("removing upload failed") @@ -346,22 +407,6 @@ func (session *OcisSession) Cleanup(revertNodeMetadata, cleanBin, cleanInfo, unm } } - if revertNodeMetadata { - n, err := session.Node(ctx) - if err != nil { - appctx.GetLogger(ctx).Error().Err(err).Str("sessionid", session.ID()).Msg("reading node for session failed") - return - } - - curUpload, err := n.ProcessingID(ctx) - if err == nil && curUpload == session.ID() { - if err := n.RevertCurrentRevision(ctx); err != nil { - appctx.GetLogger(ctx).Error().Err(err).Str("nodepath", n.InternalPath()).Msg("reverting node metadata failed") - return - } - } - } - if unmarkPostprocessing && !revertNodeMetadata { // node reverting automatically unmarks processing n, err := session.Node(ctx) if err != nil { diff --git a/pkg/storage/utils/decomposedfs/upload_async_test.go b/pkg/storage/utils/decomposedfs/upload_async_test.go index a78e701eab3..3f72eface91 100644 --- a/pkg/storage/utils/decomposedfs/upload_async_test.go +++ b/pkg/storage/utils/decomposedfs/upload_async_test.go @@ -71,7 +71,7 @@ var _ = Describe("Async file uploads", Ordered, func() { ctx context.Context - pub chan interface{} + pub chan interface{} con chan interface{} uploadID string @@ -294,6 +294,45 @@ var _ = Describe("Async file uploads", Ordered, func() { Expect(err).ToNot(BeNil()) }) + It("releases the quota and removes the node when the node metadata is unreadable", func() { + // node is created and the optimistic size has been propagated + resources, err := fs.ListFolder(ctx, rootRef, []string{}, []string{}) + Expect(err).ToNot(HaveOccurred()) + Expect(len(resources)).To(Equal(1)) + Expect(parentSize()).To(Equal(len(firstContent))) + + // simulate an orphaned node: the node file is still there but its + // metadata is gone, e.g. because an ancestor was trashed while the + // upload was in flight. Reading the node now fails. Purge instead of + // removing the file directly, so the cached attributes go as well. + nodePath := lu.InternalPath(ref.GetResourceId().GetSpaceId(), resources[0].GetId().GetOpaqueId()) + Expect(lu.MetadataBackend().Purge(ctx, nodePath)).To(Succeed()) + _, err = node.ReadNode(ctx, lu, ref.GetResourceId().GetSpaceId(), resources[0].GetId().GetOpaqueId(), false, nil, true) + Expect(err).To(HaveOccurred(), "node should be unreadable after purging its metadata") + + // No UploadReady event is published for an orphaned session: there is + // no node left to report on. Wait for the bytes to be cleaned up + // instead of for an event that will never arrive. + con <- events.PostprocessingFinished{ + UploadID: uploadID, + Outcome: events.PPOutcomeContinue, + } + Eventually(func() bool { + _, err := os.Stat(filepath.Join(o.Root, "uploads", uploadID)) + return err != nil + }).Should(BeTrue(), "the upload bytes should be cleaned up") + + // the blob was never written + bs.AssertNumberOfCalls(GinkgoT(), "Upload", 0) + + // the orphaned node is gone ... + _, err = os.Stat(nodePath) + Expect(err).ToNot(BeNil()) + + // ... and most importantly the quota has been released + Eventually(parentSize).Should(Equal(0)) + }) + It("deletes node and keeps the bytes when instructed", func() { // node is created resources, err := fs.ListFolder(ctx, rootRef, []string{}, []string{})