fix: [OCISDEV-1392][stable-8.0] release quota of orphaned upload sessions

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