Ocisdev 900 pr6d orphan rollback (#720)

* fix: release the quota of an orphaned upload through the driver seam

* feat: cleanup

* feat: apply review comments

---------

Co-authored-by: Firas Frikha <firas.frikha@kiteworks.com>

Author: LarsJurgensen

pkg/ocm/storage/received/ocm.go

@@ -716,6 +716,6 @@ func (d *driver) PrepareUpload(_ context.Context, _ *provider.Reference, _ strin
 	return &storage.PrepareUploadResult{VersionCreated: info.NodeExisted}, nil
 }
 
-func (d *driver) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ bool, _ int64) error {
+func (d *driver) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error {
 	return nil
 }

pkg/storage/fs/cephfs/upload.go

@@ -159,7 +159,7 @@ func (fs *cephfs) PrepareUpload(_ context.Context, _ *provider.Reference, _ stri
 	return &storage.PrepareUploadResult{VersionCreated: info.NodeExisted}, nil
 }
 
-func (fs *cephfs) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ bool, _ int64) error {
+func (fs *cephfs) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error {
 	return nil
 }
 

pkg/storage/fs/hello/unimplemented.go

@@ -98,7 +98,7 @@ func (fs *hellofs) PrepareUpload(_ context.Context, _ *provider.Reference, _ str
 	return &storage.PrepareUploadResult{VersionCreated: info.NodeExisted}, nil
 }
 
-func (fs *hellofs) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ bool, _ int64) error {
+func (fs *hellofs) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error {
 	return nil
 }
 

pkg/storage/fs/kiteworks/kiteworks.go

@@ -279,7 +279,7 @@ func (d *Driver) PrepareUpload(_ context.Context, _ *provider.Reference, _ strin
 	return &storage.PrepareUploadResult{VersionCreated: info.NodeExisted}, nil
 }
 
-func (d *Driver) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ bool, _ int64) error {
+func (d *Driver) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error {
 	return nil
 }
 

pkg/storage/fs/nextcloud/nextcloud.go

@@ -453,7 +453,7 @@ func (nc *StorageDriver) PrepareUpload(_ context.Context, _ *provider.Reference,
 	return &storage.PrepareUploadResult{VersionCreated: info.NodeExisted}, nil
 }
 
-func (nc *StorageDriver) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ bool, _ int64) error {
+func (nc *StorageDriver) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error {
 	return nil
 }
 

pkg/storage/fs/owncloudsql/upload.go

@@ -181,7 +181,7 @@ func (fs *owncloudsqlfs) PrepareUpload(_ context.Context, _ *provider.Reference,
 	return &storage.PrepareUploadResult{VersionCreated: info.NodeExisted}, nil
 }
 
-func (fs *owncloudsqlfs) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ bool, _ int64) error {
+func (fs *owncloudsqlfs) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error {
 	return nil
 }
 

pkg/storage/fs/s3/upload.go

@@ -86,6 +86,6 @@ func (fs *s3FS) PrepareUpload(_ context.Context, _ *provider.Reference, _ string
 	return &storage.PrepareUploadResult{VersionCreated: info.NodeExisted}, nil
 }
 
-func (fs *s3FS) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ bool, _ int64) error {
+func (fs *s3FS) RollbackUpload(_ context.Context, _ *provider.Reference, _ string, _ storage.RollbackInfo) error {
 	return nil
 }

pkg/storage/storage.go

@@ -90,6 +90,18 @@ type PrepareUploadResult struct {
 	SizeDiff       int64
 }
 
+// RollbackInfo carries what a driver needs to undo PrepareUpload. NodeID and
+// ParentID come from the upload session rather than the node, so a rollback can
+// still release the quota of a node whose own metadata has become unreadable.
+type RollbackInfo struct {
+	NodeExisted bool // true when the target node had a prior version; drivers with nothing to undo for new nodes may no-op
+	SizeDiff    int64
+	NodeID      string
+	ParentID    string
+	Filename    string
+	Size        int64
+}
+
 // FS is the interface to implement access to the storage.
 type FS interface {
 	// Minimal set for a readonly storage driver
@@ -149,9 +161,9 @@ type FS interface {
 	// It is the inverse of PrepareUpload: restores previous metadata and reverts the optimistic
 	// size propagation. The caller (coordinator) is responsible for unmarking the processing flag
 	// and deleting the upload session files. Drivers that performed no work in PrepareUpload may return nil.
-	// nodeExisted indicates whether the target node had a prior version; drivers that have nothing
-	// to undo for new nodes may no-op when nodeExisted is false.
-	RollbackUpload(ctx context.Context, ref *provider.Reference, sessionID string, nodeExisted bool, sizeDiff int64) error
+	// Callers must keep the session on a returned error: the rollback is retryable and info
+	// carries state the driver cannot recover once the session files are gone.
+	RollbackUpload(ctx context.Context, ref *provider.Reference, sessionID string, info RollbackInfo) error
 
 	// Revisions
 

pkg/storage/utils/decomposedfs/rollback_upload_test.go

@@ -1,6 +1,8 @@
 package decomposedfs_test
 
 import (
+	"os"
+
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -68,7 +70,7 @@ var _ = Describe("RollbackUpload", func() {
 			Expect(revsBefore).To(HaveLen(1))
 
 			Expect(env.Fs.MarkProcessing(env.Ctx, ref, true, "session-overwrite")).To(Succeed())
-			Expect(env.Fs.RollbackUpload(env.Ctx, ref, "session-overwrite", true, result.SizeDiff)).To(Succeed())
+			Expect(env.Fs.RollbackUpload(env.Ctx, ref, "session-overwrite", storage.RollbackInfo{NodeExisted: true, SizeDiff: result.SizeDiff})).To(Succeed())
 
 			revsAfter, err := env.Fs.ListRevisions(env.Ctx, ref)
 			Expect(err).ToNot(HaveOccurred())
@@ -104,7 +106,7 @@ var _ = Describe("RollbackUpload", func() {
 			Expect(err).ToNot(HaveOccurred())
 
 			Expect(env.Fs.MarkProcessing(env.Ctx, ref, true, "session-b")).To(Succeed())
-			Expect(env.Fs.RollbackUpload(env.Ctx, ref, "session-a", true, 10)).To(Succeed())
+			Expect(env.Fs.RollbackUpload(env.Ctx, ref, "session-a", storage.RollbackInfo{NodeExisted: true, SizeDiff: 10})).To(Succeed())
 
 			n, err := env.Lookup.NodeFromResource(env.Ctx, ref)
 			Expect(err).ToNot(HaveOccurred())
@@ -120,7 +122,117 @@ var _ = Describe("RollbackUpload", func() {
 				ResourceId: env.SpaceRootRes,
 				Path:       "/dir1/does-not-exist.txt",
 			}
-			Expect(env.Fs.RollbackUpload(env.Ctx, missingRef, "any-session-id", false, 0)).To(Succeed())
+			Expect(env.Fs.RollbackUpload(env.Ctx, missingRef, "any-session-id", storage.RollbackInfo{})).To(Succeed())
+		})
+	})
+
+	// A node whose metadata is gone but whose file remains, e.g. because an
+	// ancestor was trashed while the upload was in flight. ReadNode fails on the
+	// missing parent id, so the rollback has only the session to go on.
+	Context("orphaned node: the metadata is unreadable", func() {
+		var (
+			idRef    *provider.Reference
+			nodePath string
+			dir1Ref  *provider.Reference
+			info     storage.RollbackInfo
+		)
+
+		// parentSize is the size dir1 accounts for, which the propagation moves.
+		parentSize := func() uint64 {
+			parentInfo, err := env.Fs.GetMD(env.Ctx, dir1Ref, []string{}, []string{})
+			Expect(err).ToNot(HaveOccurred())
+			return parentInfo.Size
+		}
+
+		// orphan prepares an upload, then purges the target's metadata. It returns
+		// the parent size from before the optimistic propagation.
+		orphan := func() uint64 {
+			dir1Ref = &provider.Reference{ResourceId: env.SpaceRootRes, Path: "/dir1"}
+			sizeBefore := parentSize()
+
+			_, err := env.Fs.TouchFile(env.Ctx, ref, false, "")
+			Expect(err).ToNot(HaveOccurred())
+
+			n, err := env.Lookup.NodeFromResource(env.Ctx, ref)
+			Expect(err).ToNot(HaveOccurred())
+			nodePath = n.InternalPath()
+			// Address the node by id: the path walk needs the parent metadata the
+			// purge below destroys.
+			idRef = &provider.Reference{ResourceId: &provider.ResourceId{
+				StorageId: env.SpaceRootRes.GetStorageId(),
+				SpaceId:   env.SpaceRootRes.GetSpaceId(),
+				OpaqueId:  n.ID,
+			}}
+
+			result, err := env.Fs.PrepareUpload(env.Ctx, ref, "session-orphan", storage.UploadInfo{
+				NodeExisted: false,
+				Size:        30,
+			})
+			Expect(err).ToNot(HaveOccurred())
+			Expect(env.Fs.MarkProcessing(env.Ctx, ref, true, "session-orphan")).To(Succeed())
+			Expect(parentSize()).To(Equal(sizeBefore+30), "the optimistic size should be propagated")
+
+			// Purge, not remove: the cached attributes have to go too.
+			Expect(env.Lookup.MetadataBackend().Purge(env.Ctx, nodePath)).To(Succeed())
+			_, err = env.Lookup.NodeFromResource(env.Ctx, idRef)
+			Expect(err).To(HaveOccurred(), "the node should be unreadable")
+
+			info = storage.RollbackInfo{
+				SizeDiff: result.SizeDiff,
+				NodeID:   idRef.GetResourceId().GetOpaqueId(),
+				ParentID: n.ParentID,
+				Filename: "rollback-target.txt",
+				Size:     30,
+			}
+			return sizeBefore
+		}
+
+		It("releases the quota and removes the unreachable node", func() {
+			sizeBefore := orphan()
+
+			Expect(env.Fs.RollbackUpload(env.Ctx, idRef, "session-orphan", info)).To(Succeed())
+
+			Expect(parentSize()).To(Equal(sizeBefore))
+
+			_, err := os.Stat(nodePath)
+			Expect(err).To(HaveOccurred(), "the orphaned node should be gone")
+		})
+
+		// Without them there is no way to reach the parent, so report the failure
+		// rather than pretending the upload was rolled back.
+		It("reports the lookup failure when the session carries no ids", func() {
+			orphan()
+
+			err := env.Fs.RollbackUpload(env.Ctx, idRef, "session-orphan", storage.RollbackInfo{SizeDiff: 30})
+
+			Expect(err).To(MatchError(ContainSubstring("node lookup failed")))
+			_, statErr := os.Stat(nodePath)
+			Expect(statErr).ToNot(HaveOccurred(), "the node should be left for a retry")
+		})
+
+		// The space is what the propagation walks up to, so there is nothing to
+		// walk without it.
+		It("reports the lookup failure when the reference names no space", func() {
+			orphan()
+
+			err := env.Fs.RollbackUpload(env.Ctx, &provider.Reference{Path: "/dir1/rollback-target.txt"}, "session-orphan", info)
+
+			Expect(err).To(MatchError(ContainSubstring("node lookup failed")))
+			_, statErr := os.Stat(nodePath)
+			Expect(statErr).ToNot(HaveOccurred(), "the node should be left for a retry")
+		})
+
+		// Removing the node first would destroy the parent id the retry needs.
+		It("keeps the node when the quota cannot be released", func() {
+			orphan()
+			// A parent id pointing nowhere fails the walk up the tree.
+			info.ParentID = "does-not-exist"
+
+			err := env.Fs.RollbackUpload(env.Ctx, idRef, "session-orphan", info)
+
+			Expect(err).To(MatchError(ContainSubstring("could not revert propagate")))
+			_, statErr := os.Stat(nodePath)
+			Expect(statErr).ToNot(HaveOccurred(), "the node should be left for a retry")
 		})
 	})
 })

pkg/storage/utils/decomposedfs/upload.go

@@ -21,6 +21,7 @@ package decomposedfs
 import (
 	"context"
 	"fmt"
+	iofs "io/fs"
 	"os"
 	"path/filepath"
 	"strings"
@@ -648,10 +649,12 @@ func (fs *Decomposedfs) PrepareUpload(ctx context.Context, ref *provider.Referen
 // RollbackUpload reverts the node state written by PrepareUpload after a failed or aborted
 // postprocessing run. It restores the previous revision (or purges the node if versioning is
 // disabled and no prior version exists) and reverts the optimistic size propagation.
-func (fs *Decomposedfs) RollbackUpload(ctx context.Context, ref *provider.Reference, sessionID string, nodeExisted bool, sizeDiff int64) error {
+func (fs *Decomposedfs) RollbackUpload(ctx context.Context, ref *provider.Reference, sessionID string, info storage.RollbackInfo) error {
 	n, err := fs.lu.NodeFromResource(ctx, ref)
 	if err != nil {
-		return fmt.Errorf("RollbackUpload: node lookup failed: %w", err)
+		// The node metadata is unreadable, so the upload can never finish and its
+		// quota would stay consumed forever.
+		return fs.rollbackOrphaned(ctx, ref, sessionID, info, err)
 	}
 	if !n.Exists {
 		return nil // nothing was written yet
@@ -669,7 +672,7 @@ func (fs *Decomposedfs) RollbackUpload(ctx context.Context, ref *provider.Refere
 		return nil
 	}
 
-	if nodeExisted {
+	if info.NodeExisted {
 		if err := n.RevertCurrentRevision(ctx, false); err != nil {
 			return err
 		}
@@ -681,14 +684,53 @@ func (fs *Decomposedfs) RollbackUpload(ctx context.Context, ref *provider.Refere
 		}
 	}
 
-	if sizeDiff != 0 {
-		if err := fs.tp.Propagate(ctx, n, -sizeDiff); err != nil {
+	if info.SizeDiff != 0 {
+		if err := fs.tp.Propagate(ctx, n, -info.SizeDiff); err != nil {
 			appctx.GetLogger(ctx).Error().Err(err).Msg("RollbackUpload: could not revert propagate")
 		}
 	}
 	return nil
 }
 
+// node metadata corrupt, use session ids to release quota
+func (fs *Decomposedfs) rollbackOrphaned(ctx context.Context, ref *provider.Reference, sessionID string, info storage.RollbackInfo, lookupErr error) error {
+	if info.NodeID == "" || info.ParentID == "" {
+		return fmt.Errorf("RollbackUpload: node lookup failed: %w", lookupErr)
+	}
+	spaceID := ref.GetResourceId().GetSpaceId()
+	if spaceID == "" {
+		return fmt.Errorf("RollbackUpload: node lookup failed: %w", lookupErr)
+	}
+	appctx.GetLogger(ctx).Info().Err(lookupErr).Str("sessionid", sessionID).Str("nodeid", info.NodeID).
+		Msg("node unreadable, rolling back orphaned upload")
+	n := node.New(spaceID, info.NodeID, info.ParentID, info.Filename, info.Size, sessionID,
+		provider.ResourceType_RESOURCE_TYPE_FILE, nil, fs.lu)
+	spaceRoot, err := node.ReadNode(ctx, fs.lu, spaceID, spaceID, false, nil, false)
+	if err != nil {
+		return fmt.Errorf("RollbackUpload: space root lookup failed: %w", err)
+	}
+	n.SpaceRoot = spaceRoot
+	if info.SizeDiff != 0 {
+		// return error so caller keeps session for retry
+		if err := fs.tp.Propagate(ctx, n, -info.SizeDiff); err != nil {
+			return fmt.Errorf("RollbackUpload: could not revert propagate: %w", err)
+		}
+	}
+	nodePath := n.InternalPath()
+	if err := utils.RemoveItem(nodePath); err != nil && !errors.Is(err, iofs.ErrNotExist) {
+		appctx.GetLogger(ctx).Error().Err(err).Str("nodepath", nodePath).Msg("RollbackUpload: removing orphaned node failed")
+	}
+	if err := fs.lu.MetadataBackend().Purge(ctx, nodePath); err != nil && !errors.Is(err, iofs.ErrNotExist) {
+		appctx.GetLogger(ctx).Error().Err(err).Str("nodepath", nodePath).Msg("RollbackUpload: purging orphaned node metadata failed")
+	}
+	// parent holds a child entry pointing to the now-removed node
+	childEntry := filepath.Join(n.ParentPath(), n.Name)
+	if err := os.Remove(childEntry); err != nil && !errors.Is(err, iofs.ErrNotExist) {
+		appctx.GetLogger(ctx).Error().Err(err).Str("path", childEntry).Msg("RollbackUpload: removing orphaned child entry failed")
+	}
+	return nil
+}
+
 func validateChecksums(ctx context.Context, lu node.PathLookup, n *node.Node, versionPath string) error {
 	for _, t := range []string{"md5", "sha1", "adler32"} {
 		key := prefixes.ChecksumPrefix + t