pkg/storage: release quota of orphaned upload sessions (#692) (#730)

When an upload's target node lost its metadata, e.g. because an ancestor was
trashed while the upload was in flight, the node file remained without a
readable .mpk. Reading it fails with "Missing parent ID on node", so the upload
never finished postprocessing: it stayed in "Processing", could not be
downloaded or deleted, and kept consuming the quota.

Cleanup made this worse. It removed the upload bytes and the session info file
before attempting to revert the node, then bailed out on the failing node read
without releasing the quota - destroying both the only copy of the data and the
parent id needed to repair the node, while freeing nothing.

Revert the node before anything irreversible and fall back to the parent id
recorded in the session when the node cannot be read, so the quota is released
and the orphaned node removed. Keep the upload if the quota cannot be released.
Also clean up such sessions when postprocessing finishes instead of returning
early and leaving them to be retried forever.

Add an Orphaned upload session filter to list affected sessions. It is only
evaluated when set, as it reads the node metadata of every session.

Signed-off-by: Julian Koberg <julian.koberg@kiteworks.com>
Co-authored-by: kobergj <juliankoberg@googlemail.com>

Author: mklos-kw

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

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
 }

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()

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

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

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 {

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{})