fix(restore): give each native restore an owner and a private upload

A native restore (POST ?restore, issue #281) marks the version ongoing, and a background worker rehydrates it from the tier. The worker, its upload and every status write could be reached or confused by other callers. A public client could write into or complete the restore upload. POST copied a whole metadata snapshot over newer protection changes. An old or failed attempt could act on a newer one. Restore, expiry and heal decided on whichever drive's metadata they read. Each POST that starts a rehydration now writes an attempt ID into the version's internal metadata, with the identity it resolved: version ID, ETag, size and ModTime. The worker, its upload, its part writes and its completion carry the attempt. A worker whose attempt no longer owns the version stops without writing, completing or resetting. Every write uses the versioning state and explicit version ID captured by POST, so a null target stays null. The restore upload carries an internal marker, decided over all drives. Public multipart calls answer NoSuchUpload and listings hide it. An unknown marker fails closed. A completion holds a commit lock from staging its commit in the upload until the commit ends. Stale cleanup decides on every upload past the stale age under that lock, and keeps it until its rename ends, also past the drive deadline. Abort of a committing upload takes the lock and removes a failed commit's leftover only when every copy is staged and older than one hour. POST, the worker, the reset, the ExpireRestored recheck and heal all decide on the authoritative revision of the exact version. That is the revision held by max(n/2+1, read quorum) drives, identified by the version header and the normalized metadata digest. Status writes patch only the restore fields of that revision. They write only the drives that hold it, and put the old revision back on failure. Heal copies restore fields only when a quorum holds the same fields. A failed attempt resets itself with bounded in-memory retries. A later POST takes over a stranded marker when no attempt runs. Otherwise it answers RestoreAlreadyInProgress within a few seconds. Older nodes do not know the marker. A new grid capability handler lets the serving node ask every peer. POST answers 503 ServiceUnavailable and writes nothing until all peers support native restore, and a queued worker asks again before it creates its upload. The single-pool lock-domain split between DELETE/CopyObject and restore is closed by its own fix, which this change depends on. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>
2026-11-11 01:55:30 +03:00 · 2026-11-05 06:47:38 +08:00
parent 9f7b49369d
commit 7f38bf6282
20 changed files with 4543 additions and 162 deletions
@@ -34,7 +34,6 @@ import (
 	"github.com/google/uuid"
 	"github.com/klauspost/readahead"
 	"github.com/minio/minio-go/v7/pkg/set"
-	"github.com/minio/minio/internal/bucket/lifecycle"
 	"github.com/minio/minio/internal/config/storageclass"
 	"github.com/minio/minio/internal/crypto"
 	"github.com/minio/minio/internal/hash"
@@ -72,6 +71,7 @@ func (er erasureObjects) getMultipartSHADir(bucket, object string) string {
 }

 // checkUploadIDExists - verify if a given uploadID exists and is valid.
+// The metadata read from each drive is returned also with an error.
 func (er erasureObjects) checkUploadIDExists(ctx context.Context, bucket, object, uploadID string, write bool) (fi FileInfo, metArr []FileInfo, err error) {
 	defer func() {
 		if errors.Is(err, errFileNotFound) {
@@ -89,15 +89,15 @@ func (er erasureObjects) checkUploadIDExists(ctx context.Context, bucket, object

 	readQuorum, writeQuorum, err := objectQuorumFromMeta(ctx, partsMetadata, errs, er.defaultParityCount)
 	if err != nil {
-		return fi, nil, err
+		return fi, partsMetadata, err
 	}

 	if readQuorum < 0 {
-		return fi, nil, errErasureReadQuorum
+		return fi, partsMetadata, errErasureReadQuorum
 	}

 	if writeQuorum < 0 {
-		return fi, nil, errErasureWriteQuorum
+		return fi, partsMetadata, errErasureWriteQuorum
 	}

 	quorum := readQuorum
@@ -114,7 +114,7 @@ func (er erasureObjects) checkUploadIDExists(ctx context.Context, bucket, object
 		err = reduceReadQuorumErrs(ctx, errs, objectOpIgnoredErrs, readQuorum)
 	}
 	if err != nil {
-		return fi, nil, err
+		return fi, partsMetadata, err
 	}

 	// Pick one from the first valid metadata.
@@ -217,12 +217,55 @@ func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk Sto
 				hasSkippedUploads = true
 				return nil
 			}
+			// A completion holds this lock from staging its commit in the
+			// upload until the commit ends: decide on a removal only while
+			// holding it, and skip an upload whose commit is running. Only
+			// uploads past the stale age pay for the lock and the read below.
+			lk := er.restoreCommitLock(uploadIDPath)
+			lkctx, err := lk.GetLock(ctx, newDynamicTimeout(time.Second, time.Second))
+			if err != nil {
+				hasSkippedUploads = true
+				return nil
+			}
+			// The rename below releases the lock when it ends, also when it
+			// outlives its deadline; every other exit releases it here.
+			renaming := false
+			defer func() {
+				if !renaming {
+					lk.Unlock(lkctx)
+				}
+			}()
+			// A native restore can run past the stale age: keep its upload
+			// while its attempt still owns the version, and keep any upload
+			// whose marker the drives cannot establish.
+			view := er.readRestoreUpload(lkctx.Context(), "", uploadIDPath)
+			if view.committing {
+				// What a commit staged, with no commit running: only a staged
+				// copy itself shows when it staged, only an old one is a
+				// leftover of a failed commit, and a copy not staged yet
+				// tells nothing.
+				local, err := disk.ReadVersion(lkctx.Context(), "", minioMetaMultipartBucket, uploadIDPath, "", ReadOptions{})
+				st, serr := os.Stat(pathJoin(drivePath, minioMetaMultipartBucket, uploadIDPath, xlStorageFormatFile))
+				if err != nil || serr != nil || !restoreUploadStaged(local) || time.Since(st.ModTime()) < restoreCommitGrace {
+					hasSkippedUploads = true
+					return nil
+				}
+			} else if view.live && (!view.known || (view.marker != nil && er.restoreUploadActive(lkctx.Context(), view))) {
+				hasSkippedUploads = true
+				return nil
+			}
+			renaming = true
 			w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
 			return w.Run(func() error {
 				wait := deleteMultipartCleanupSleeper.Timer(ctx)
 				pathUUID := mustGetUUID()
 				targetPath := pathJoin(drivePath, minioMetaTmpDeletedBucket, pathUUID)
-				renameAll(pathJoin(drivePath, minioMetaMultipartBucket, uploadIDPath), targetPath, pathJoin(drivePath, minioMetaBucket))
+				err := renameAll(pathJoin(drivePath, minioMetaMultipartBucket, uploadIDPath), targetPath, pathJoin(drivePath, minioMetaBucket))
+				lk.Unlock(lkctx)
+				if err != nil {
+					// The next cleanup round tries again.
+					storageLogIf(ctx, fmt.Errorf("Unable to remove stale upload %s on %s: %w", uploadIDPath, drivePath, err))
+				}
 				wait()
 				return nil
 			})
@@ -463,6 +506,13 @@ func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket,
 		if populatedUploadIDs.Contains(uploadID) {
 			continue
 		}
+		// A native restore's upload is private to its attempt; list only an
+		// upload that the drives establish as an ordinary one.
+		metas, errs := readAllFileInfo(ctx, er.getDisks(), bucket, minioMetaMultipartBucket,
+			pathJoin(er.getMultipartSHADir(bucket, object), uploadID), "", false, false)
+		if !restoreUploadState(metas, errs).visible(ObjectOptions{}) {
+			continue
+		}
 		var fallback time.Time
 		if _, ok := multipartUploadTime(uploadID); !ok {
 			fi, err := disk.ReadVersion(ctx, bucket, minioMetaMultipartBucket,
@@ -562,6 +612,11 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 	if opts.PreserveETag != "" {
 		userDefined["etag"] = opts.PreserveETag
 	}
+	// Only a native restore marks its upload private, never caller metadata.
+	delete(userDefined, restoreUploadKey)
+	if a := opts.Transition.attempt; a != nil {
+		userDefined[restoreUploadKey] = a.String()
+	}
 	onlineDisks := er.getDisks()

 	// Get parity and data drive count based on storage class metadata
@@ -743,7 +798,10 @@ func (er erasureObjects) PutObjectPart(ctx context.Context, bucket, object, uplo

 	uploadIDPath := er.getUploadIDDir(bucket, object, uploadID)
 	// Validates if upload ID exists.
-	fi, _, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, true)
+	fi, metas, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, true)
+	if err == nil && !restoreUploadVisible(metas, opts) {
+		err = errUploadIDNotFound
+	}
 	if err != nil {
 		if errors.Is(err, errVolumeNotFound) {
 			return pi, toObjectErr(err, bucket)
@@ -980,7 +1038,10 @@ func (er erasureObjects) GetMultipartInfo(ctx context.Context, bucket, object, u
 		UploadID: uploadID,
 	}

-	fi, _, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
+	fi, metas, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
+	if err == nil && !restoreUploadVisible(metas, opts) {
+		err = errUploadIDNotFound
+	}
 	if err != nil {
 		if errors.Is(err, errVolumeNotFound) {
 			return result, toObjectErr(err, bucket)
@@ -1058,7 +1119,10 @@ func (er erasureObjects) ListObjectParts(ctx context.Context, bucket, object, up
 		auditObjectErasureSet(ctx, "ListObjectParts", object, &er)
 	}

-	fi, _, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
+	fi, metas, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
+	if err == nil && !restoreUploadVisible(metas, opts) {
+		err = errUploadIDNotFound
+	}
 	if err != nil {
 		return result, toObjectErr(err, bucket, object, uploadID)
 	}
@@ -1271,7 +1335,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 		auditObjectErasureSet(ctx, "CompleteMultipartUpload", object, &er)
 	}

-	if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile {
+	if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile || opts.Transition.attempt != nil {
 		if !opts.NoLock {
 			ns := er.NewNSLock(bucket, object)
 			lkctx, err := ns.GetLock(ctx, globalOperationTimeout)
@@ -1304,7 +1368,24 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 		}
 	}

+	if opts.Transition.attempt != nil {
+		// A native restore holds the upload write lock from validation
+		// through commit, after the object lock and in the lock domain of
+		// PutObjectPart, which takes a part lock and then this lock to read.
+		// NoLock only says that the caller holds the object lock.
+		uploadIDLock := er.NewNSLock(bucket, pathJoin(object, uploadID))
+		ulkctx, err := uploadIDLock.GetLock(ctx, globalOperationTimeout)
+		if err != nil {
+			return oi, err
+		}
+		ctx = ulkctx.Context()
+		defer uploadIDLock.Unlock(ulkctx)
+	}
+
 	fi, partsMetadata, err := er.checkUploadIDExists(ctx, bucket, object, uploadID, true)
+	if err == nil && !restoreUploadVisible(partsMetadata, opts) {
+		err = errUploadIDNotFound
+	}
 	if err != nil {
 		if errors.Is(err, errVolumeNotFound) {
 			return oi, toObjectErr(err, bucket)
@@ -1618,6 +1699,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 	delete(fi.Metadata, hash.MinIOMultipartChecksumType)
 	delete(fi.Metadata, multipartMetaBucket)
 	delete(fi.Metadata, multipartMetaObject)
+	delete(fi.Metadata, restoreUploadKey)

 	// Save the final object size and modtime.
 	fi.Size = objectSize
@@ -1697,6 +1779,20 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 			if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo, opts.Transition); err != nil {
 				return oi, toObjectErr(err, bucket, object)
 			}
+			if a := opts.Transition.attempt; a != nil {
+				// A native restore writes back only the null version it targets.
+				current, gerr := getObjectInfo(ctx, bucket, object, ObjectOptions{
+					VersionID:  nullVersionID,
+					Transition: opts.Transition,
+					NoLock:     true,
+				})
+				if gerr != nil {
+					return oi, toObjectErr(gerr, bucket, object)
+				}
+				if !a.rehydrates(current, fi) {
+					return oi, toObjectErr(errAuthentication, bucket, object)
+				}
+			}
 		} else {
 			current, gerr := getObjectInfo(ctx, bucket, object, ObjectOptions{
 				VersionID:        fi.VersionID,
@@ -1706,11 +1802,10 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 				NoLock:           true,
 			})
 			switch {
-			case gerr == nil && opts.Transition.RestoreRequest != nil &&
-				current.TransitionedObject.Status == lifecycle.TransitionComplete &&
-				current.ETag == fi.Metadata["etag"] && current.Size == fi.Size && current.ModTime.Equal(fi.ModTime):
-				// The internal restore path rehydrates this same tiered version.
-				// Ordinary multipart completion must still reject a version overwrite.
+			case gerr == nil && opts.Transition.attempt.rehydrates(current, fi):
+				// A native restore's own private upload rehydrates this same
+				// tiered version. Any other completion must still reject a
+				// version overwrite.
 			case gerr == nil, errors.As(gerr, &MethodNotAllowed{}):
 				return oi, toObjectErr(errAuthentication, bucket, object)
 			case isErrVersionNotFound(gerr) || isErrObjectNotFound(gerr):
@@ -1720,6 +1815,16 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 		}
 	}

+	// Stale cleanup and abort judge an upload whose commit staged object
+	// metadata in it by this lock: hold it from staging through the commit.
+	commitLock := er.restoreCommitLock(uploadIDPath)
+	clkctx, err := commitLock.GetLock(ctx, globalOperationTimeout)
+	if err != nil {
+		return oi, err
+	}
+	ctx = clkctx.Context()
+	defer commitLock.Unlock(clkctx)
+
 	er.cleanupMultipartPath(ctx, paths...) // cleanup all part.N.meta, and skipped part.N's before final rename().

 	defer func() {
@@ -1791,7 +1896,38 @@ func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, objec
 	if !ok || internalID == "" || internalID == "." || internalID == ".." || strings.ContainsAny(internalID, "/\\") {
 		return false, InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID}
 	}
-	defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID))
+	// Authorize before the unconditional cleanup below. A native restore's
+	// upload is absent to every other caller, and an upload whose marker the
+	// drives cannot establish is not removed by anyone.
+	uploadIDPath := er.getUploadIDDir(bucket, object, uploadID)
+	view := er.readRestoreUpload(ctx, bucket, uploadIDPath)
+	if view.committing {
+		// What a failed commit left behind, also after stale cleanup could
+		// not remove it: only with no commit running, and once every copy
+		// shows a staging older than the grace.
+		lk := er.restoreCommitLock(uploadIDPath)
+		lkctx, err := lk.GetLock(ctx, newDynamicTimeout(restoreTakeoverWait, restoreTakeoverWait))
+		if err != nil {
+			return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID)
+		}
+		defer lk.Unlock(lkctx)
+		if !er.restoreCommitAbandoned(lkctx.Context(), bucket, uploadIDPath) {
+			return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID)
+		}
+		er.deleteAll(lkctx.Context(), minioMetaMultipartBucket, uploadIDPath)
+		return true, nil
+	}
+	switch {
+	case view.visible(opts):
+	case view.known && view.marked:
+		return false, nil
+	case view.known && opts.Transition.attempt != nil:
+		// A native restore's own abort, of an upload no longer its own.
+		return false, nil
+	default:
+		return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID)
+	}
+	defer er.deleteAll(ctx, minioMetaMultipartBucket, uploadIDPath)
 	_, _, err = er.checkUploadIDExists(ctx, bucket, object, uploadID, false)
 	err = toObjectErr(err, bucket, object, uploadID)
 	if _, absent := err.(InvalidUploadID); absent {