[s3] Cleaning uploads can cause corruption [LOW RISK]

diff --git a/weed/s3api/s3api_multipart_finalize_txn.go b/weed/s3api/s3api_multipart_finalize_txn.go
new file mode 100644
--- /dev/null
+++ b/weed/s3api/s3api_multipart_finalize_txn.go
@@ -0,0 +1,139 @@
+package s3api
+
+import (
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+)
+
+// routedMultipartFinalize commits a completed multipart upload in ONE
+// ObjectTransaction on the owner filer, instead of three separate RPCs:
+//
+//  1. PUT     <object>.versions/<versionFile> (the finished object)
+//  2. DELETE  <bucket>/.uploads/<uploadId>   (metadata only)
+//  3. RECOMPUTE_LATEST on <object>.versions   (publish it as current)
+//
+// Why this exists rather than extending routedVersionedFinalize: that helper is
+// shared with CopyObject and createDeleteMarker, neither of which has an upload
+// directory or writes its entry inside the transaction. Keeping the multipart
+// shape here leaves those call sites untouched.
+//
+// Why the upload directory is removed with IsDeleteData=false: a multipart
+// object's chunks ARE the part chunks. Completion references the fids, it does
+// not copy bytes, so freeing them would destroy the object being published.
+// After this transaction the upload directory no longer exists, so
+// s3.clean.uploads has nothing to mistake for an abandoned upload.
+//
+// ORDER MATTERS, and not for the usual reason. ObjectTransaction holds one
+// per-path lock for the whole request (filer_grpc_server.go:305-307), which
+// gives isolation from other writers of the same object — but the mutation loop
+// applies each mutation in turn and returns on the first error with NO rollback
+// and no store-level transaction (filer_grpc_server.go:333-339). So a partial
+// application is reachable, and the order decides which partial states exist:
+//
+//	PUT, DELETE, RECOMPUTE   (this order)
+//	  fail at 1 -> nothing happened; the client can retry the completion
+//	  fail at 2 -> version file written but not current, upload directory still
+//	               there: BOTH reference the chunks, so nothing is unreferenced.
+//	               The caller rolls the version file back and the state is exactly
+//	               stock's "upload not completed" -- fully retryable.
+//	  fail at 3 -> version exists but is not current; the read-path heal and the
+//	               background reconciler both promote it
+//
+// Two orderings are wrong, for opposite reasons:
+//
+//	RECOMPUTE first leaves "object published, upload directory still present",
+//	which is exactly the bug this is fixing.
+//
+//	DELETE first leaves, between mutations 1 and 2, chunks referenced by NOTHING.
+//	The per-path lock does not help there: volume.fsck computes orphans from its
+//	own filer scan and never consults that lock, so a concurrent
+//	`volume.fsck -reallyDeleteFromVolume` could classify the parts as orphans and
+//	delete them, and mutation 2 would then publish an object over freed needles.
+//	Narrow (two store operations) and gated behind an expert-only flag and fsck's
+//	5h -cutoffTimeAgo, but it is a window stock does NOT have: stock keeps the
+//	parts referenced by .uploads/<id> until after the object is committed.
+//	Putting the PUT first removes it entirely.
+//
+// None of the reachable partial states loses a byte the client was told was
+// durable, and none leaves a chunk unreferenced.
+func (s3a *S3ApiServer) routedMultipartFinalize(
+	owner pb.ServerAddress,
+	bucket, object string,
+	useInvertedFormat bool,
+	versionDir, versionFileName string,
+	chunks []*filer_pb.FileChunk,
+	decorate func(*filer_pb.Entry),
+	uploadsDir, uploadID string,
+) s3err.ErrorCode {
+
+	// Same shape filer_pb.MkFile builds (filer_client.go:272-283); the owner
+	// applies collection/replication/TTL defaults itself in applyObjectMutation.
+	now := time.Now().Unix()
+	versionEntry := &filer_pb.Entry{
+		Name:        versionFileName,
+		IsDirectory: false,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime:    now,
+			Crtime:   now,
+			FileMode: uint32(0770),
+			Uid:      filer_pb.OS_UID,
+			Gid:      filer_pb.OS_GID,
+		},
+		Chunks: chunks,
+	}
+	if decorate != nil {
+		decorate(versionEntry)
+	}
+
+	mutations := []*filer_pb.ObjectMutation{
+		{
+			Type:      filer_pb.ObjectMutation_PUT,
+			Directory: versionDir,
+			Name:      versionFileName,
+			Entry:     versionEntry,
+		},
+	}
+	if uploadsDir != "" && uploadID != "" {
+		mutations = append(mutations, &filer_pb.ObjectMutation{
+			Type:         filer_pb.ObjectMutation_DELETE,
+			Directory:    uploadsDir,
+			Name:         uploadID,
+			IsDeleteData: false,
+			IsRecursive:  true,
+		})
+	}
+	mutations = append(mutations,
+		s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true),
+	)
+
+	// With no ring view there is no owner to route to, but ObjectTransaction
+	// only requires LockKey: an empty RouteKey makes the receiving filer apply
+	// the mutations locally under its own per-path lock instead of forwarding
+	// (filer_grpc_server.go:277). That is no weaker than the three unlocked RPCs
+	// this path used to make, and it still gets the ordering guarantee, so the
+	// upload directory can never outlive the published object.
+	routeKey := ""
+	if owner != "" {
+		routeKey = s3a.objectRouteKey(bucket, object)
+	}
+	req := &filer_pb.ObjectTransactionRequest{
+		LockKey:   s3a.toFilerPath(bucket, object),
+		RouteKey:  routeKey,
+		Mutations: mutations,
+	}
+	resp, err := s3a.objectTxnOnFiler(owner, req)
+	switch {
+	case err != nil:
+		glog.Errorf("routedMultipartFinalize: %s/%s upload %s on %s: %v", bucket, object, uploadID, owner, err)
+		return s3err.ErrInternalError
+	case resp.Error != "":
+		glog.Errorf("routedMultipartFinalize: %s/%s upload %s: %s", bucket, object, uploadID, resp.Error)
+		return s3err.ErrInternalError
+	default:
+		return s3err.ErrNone
+	}
+}
diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go
--- a/weed/s3api/filer_multipart.go
+++ b/weed/s3api/filer_multipart.go
@@ -236,6 +236,7 @@
 	manifestsReferenced     bool                  // failed rollback left an entry holding newManifestChunks
 	supersededPartManifests []*filer_pb.FileChunk // part-entry blobs replaced by flattening; deleted after commit
 	metadataOnlyCleanup     bool                  // deleteEntries share chunks with the live object; keep their data
+	uploadDirRemoved        bool                  // .uploads/<id> was removed inside the finalize transaction
 }
 
 func completeMultipartResult(r *http.Request, input *s3.CompleteMultipartUploadInput, etag string, entry *filer_pb.Entry) *CompleteMultipartUploadResult {
@@ -662,8 +663,9 @@
 			versionMtime := time.Now().Unix()
 			amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
 
-			// Create the version file in the .versions directory
-			if err := s3a.mkFile(versionDir, versionFileName, completionState.finalParts, func(versionEntry *filer_pb.Entry) {
+			// How the version entry is filled in. On the routed path this runs
+			// inside the finalize transaction; on the fallback it runs in mkFile.
+			decorateVersionEntry := func(versionEntry *filer_pb.Entry) {
 				if versionEntry.Extended == nil {
 					versionEntry.Extended = make(map[string][]byte)
 				}
@@ -712,9 +714,6 @@
 				}
 				versionEntry.Attributes.FileSize = uint64(completionState.offset)
 				versionEntry.Attributes.Mtime = versionMtime
-			}); err != nil {
-				glog.Errorf("completeMultipartUpload: failed to create version %s: %v", versionId, err)
-				return s3err.ErrInternalError
 			}
 
 			// Construct entry with metadata for caching in .versions directory
@@ -737,21 +736,44 @@
 			// Pass entry to cache its metadata for single-scan list efficiency
 			// Route the pointer flip to the owner (off the lock) via
 			// RECOMPUTE_LATEST; the just-written version file is the newest.
-			if owner != "" {
-				if code := s3a.routedVersionedFinalize(owner, *input.Bucket, *input.Key, useInvertedFormat); code != s3err.ErrNone {
+			// objectTxnOnFiler needs either a routed owner or a filerClient to
+			// pick a filer from; only the initialization/testing case where both
+			// are absent still needs the legacy three-RPC sequence.
+			if owner != "" || s3a.filerClient != nil {
+				// The transaction removes the whole upload directory, so the
+				// superseded part entries -- whose chunks the object does NOT
+				// reference -- have to be freed first or they leak.
+				for _, deleteEntry := range completionState.deleteEntries {
+					if err := s3a.rm(uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil {
+						glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", *input.Bucket, *input.UploadId, deleteEntry.Name, err)
+					}
+				}
+				// One transaction: drop the upload directory, write the version
+				// entry, publish it. See routedMultipartFinalize on why the order
+				// is what it is.
+				if code := s3a.routedMultipartFinalize(owner, *input.Bucket, *input.Key, useInvertedFormat,
+					versionDir, versionFileName, completionState.finalParts, decorateVersionEntry,
+					s3a.genUploadsFolder(*input.Bucket), *input.UploadId); code != s3err.ErrNone {
 					if rollbackErr := s3a.rollbackMultipartVersion(versionDir, versionFileName); rollbackErr != nil {
 						glog.Errorf("completeMultipartUpload: failed to rollback version %s for %s/%s after routed finalize error: %v", versionId, *input.Bucket, *input.Key, rollbackErr)
 						completionState.manifestsReferenced = true
 					}
 					return code
 				}
-			} else if err := s3a.updateLatestVersionInDirectory(*input.Bucket, *input.Key, versionId, versionFileName, versionEntryForCache); err != nil {
-				if rollbackErr := s3a.rollbackMultipartVersion(versionDir, versionFileName); rollbackErr != nil {
-					glog.Errorf("completeMultipartUpload: failed to rollback version %s for %s/%s after latest pointer update error: %v", versionId, *input.Bucket, *input.Key, rollbackErr)
-					completionState.manifestsReferenced = true
-				}
-				glog.Errorf("completeMultipartUpload: failed to update latest version in directory: %v", err)
-				return s3err.ErrInternalError
+				completionState.uploadDirRemoved = true
+			} else {
+				if err := s3a.mkFile(versionDir, versionFileName, completionState.finalParts, decorateVersionEntry); err != nil {
+					glog.Errorf("completeMultipartUpload: failed to create version %s: %v", versionId, err)
+					return s3err.ErrInternalError
+				}
+				if err := s3a.updateLatestVersionInDirectory(*input.Bucket, *input.Key, versionId, versionFileName, versionEntryForCache); err != nil {
+					if rollbackErr := s3a.rollbackMultipartVersion(versionDir, versionFileName); rollbackErr != nil {
+						glog.Errorf("completeMultipartUpload: failed to rollback version %s for %s/%s after latest pointer update error: %v", versionId, *input.Bucket, *input.Key, rollbackErr)
+						completionState.manifestsReferenced = true
+					}
+					glog.Errorf("completeMultipartUpload: failed to update latest version in directory: %v", err)
+					return s3err.ErrInternalError
+				}
 			}
 
 			// For versioned buckets, all content is stored in .versions directory
@@ -936,13 +958,15 @@
 	}
 
 	if completionState != nil {
-		for _, deleteEntry := range completionState.deleteEntries {
-			if err := s3a.rm(uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil {
-				glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", *input.Bucket, *input.UploadId, deleteEntry.Name, err)
-			}
-		}
-		if err := s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil {
-			glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
+		if !completionState.uploadDirRemoved {
+			for _, deleteEntry := range completionState.deleteEntries {
+				if err := s3a.rm(uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil {
+					glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", *input.Bucket, *input.UploadId, deleteEntry.Name, err)
+				}
+			}
+			if err := s3a.rm(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil {
+				glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
+			}
 		}
 		if len(completionState.supersededPartManifests) > 0 {
 			s3a.deleteOrphanedChunks(completionState.supersededPartManifests)

#!/bin/bash
# Reproduce: s3.clean.uploads destroys a completed object's data.
#
# Runs entirely on a STOCK weed binary. No patches, no instrumentation.
#
#   ./repro-clean-uploads-dataloss.sh /path/to/weed [attempts]
#
# THE BUG
#   A multipart object's chunks ARE the part chunks -- completion references the
#   fids, it does not copy bytes. Completion therefore removes the upload
#   directory metadata-only (filer_multipart.go, `s3a.rm(..., false, true)`), and
#   that removal is best-effort: its error is logged at V(1) and ignored. Any
#   filer error or a crash in that window leaves .uploads/<uploadId>/ behind
#   while the object is committed and serving. s3.clean.uploads then purges the
#   leftover with `?recursive=true` and NO skipChunkDeletion, freeing the live
#   object's needles.
#
# HOW THIS HITS THE WINDOW WITHOUT PATCHING ANYTHING
#   The window -- object published, upload directory not yet removed -- is small:
#   measured at ~2.2ms on a loopback single-node cluster with the default leveldb
#   filer store. It widens with anything that slows a filer metadata write, so a
#   remote/SQL filer store, a busy machine, or slower storage all make this
#   EASIER to hit, not harder.
#
#   Landing a SIGKILL in ~2ms needs a sampler faster than the window. A shell
#   poll loop cannot do it: each iteration spawns curl and python, ~80ms. This
#   script uses one Python process with a persistent HTTP connection polling the
#   .versions pointer every 2ms. Each attempt is then roughly a coin flip, and
#   attempts are cheap, so it converges in a handful.
#
#   NOISE_PARTS uploads extra parts the completion does not list, and it is what
#   makes this land. Not for the reason you might expect: those parts are skipped
#   at filer_multipart.go:431-434 and never reach the cleanup loop. What they
#   lengthen is the recursive delete itself -- doBatchDeleteFolderMetaAndData
#   lists the directory and fires a notification PER CHILD before the single
#   DeleteFolderChildren store call, so more children means a longer phase in
#   which the part entries still exist and the parent is still there. That is the
#   state the cleaner has to find. Measured: 0/20 attempts at NOISE_PARTS=0,
#   hits within a few attempts at 200.
#
# WHAT YOU SHOULD SEE
#   leftover .uploads/<uploadId>/ next to a healthy, readable object;
#   then after the cleaner + a vacuum: HEAD says 10485760, GET returns 0 bytes.
set -u

WEED=${1:?usage: $0 /path/to/weed [attempts]}
ATTEMPTS=${2:-8}
DIR=/tmp/sw-repro-$$
# -s3.port.iceberg=0 below disables the Iceberg REST catalog, which otherwise
# binds a FIXED 8181