fix(restore): rehydrate the selected version under its object lock

Preserve handler version context, resolve the requested version's pool, and
return on the initial read failure without writing metadata. Propagate owned
NoLock to single-part writes, multipart completion and failure cleanup in
the same patch. Preserve the original modification time on cleanup.

Route the restore status copy by version and allow internal multipart
rehydration only for a matching completed transition. Single-pool restores
own the metadata snapshot through download and cleanup. Keep public
completion in object-before-upload order, using the target set's object
lock for single-pool calls, so concurrent completion cannot invert locks.

Cover real signed POSTs, null/UUID and cross-pool identity, download failures,
multipart overwrite guards, Object Lock updates and concurrent completion.
This package depends on the restored-copy cleanup safety delivered in R2.

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/erasure-multipart.go

@@ -34,6 +34,7 @@ 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"
@@ -1697,13 +1698,18 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 				return oi, toObjectErr(err, bucket, object)
 			}
 		} else {
-			_, gerr := getObjectInfo(ctx, bucket, object, ObjectOptions{
+			current, gerr := getObjectInfo(ctx, bucket, object, ObjectOptions{
 				VersionID:        fi.VersionID,
 				Versioned:        opts.Versioned,
 				VersionSuspended: opts.VersionSuspended,
 				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, errors.As(gerr, &MethodNotAllowed{}):
 				return oi, toObjectErr(errAuthentication, bucket, object)
 			case isErrVersionNotFound(gerr) || isErrObjectNotFound(gerr):

cmd/erasure-object.go

@@ -2644,6 +2644,8 @@ func (er erasureObjects) updateRestoreMetadata(ctx context.Context, bucket, obje
 		VersionID: oi.VersionID,
 	}, ObjectOptions{
 		VersionID: oi.VersionID,
+		MTime:     oi.ModTime,
+		NoLock:    opts.NoLock,
 	}); err != nil {
 		storageLogIf(ctx, fmt.Errorf("Unable to update transition restore metadata for %s/%s(%s): %s", bucket, object, oi.VersionID, err))
 		return err
@@ -2654,6 +2656,19 @@ func (er erasureObjects) updateRestoreMetadata(ctx context.Context, bucket, obje
 // restoreTransitionedObject for multipart object chunks the file stream from remote tier into the same number of parts
 // as in the xl.meta for this version and rehydrates the part.n into the fi.DataDir for this version as in the xl.meta
 func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket string, object string, opts ObjectOptions) error {
+	if !opts.NoLock {
+		// Single-pool and direct callers must protect the metadata snapshot
+		// through download, rehydration and failure cleanup, just like pools.
+		lk := er.NewNSLock(bucket, object)
+		lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout)
+		if err != nil {
+			return err
+		}
+		ctx = lkctx.Context()
+		defer lk.Unlock(lkctx)
+		opts.NoLock = true
+	}
+
 	setRestoreHeaderFn := func(oi ObjectInfo, rerr error) error {
 		if rerr == nil {
 			return nil // nothing to do; restore object was successful
@@ -2665,11 +2680,13 @@ func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket s
 	// get the file info on disk for transitioned object
 	actualfi, _, _, err := er.getObjectFileInfo(ctx, bucket, object, opts, false)
 	if err != nil {
-		return setRestoreHeaderFn(oi, toObjectErr(err, bucket, object))
+		// No version was read, so there is no restore metadata to update.
+		return toObjectErr(err, bucket, object)
 	}
 
 	oi = actualfi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended)
 	ropts := putRestoreOpts(bucket, object, opts.Transition.RestoreRequest, oi)
+	ropts.NoLock = opts.NoLock
 	if len(oi.Parts) == 1 {
 		var rs *HTTPRangeSpec
 		gr, err := getTransitionedObjectReader(ctx, bucket, object, rs, http.Header{}, oi, opts)
@@ -2719,7 +2736,11 @@ func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket s
 		})
 	}
 	_, err = er.CompleteMultipartUpload(ctx, bucket, object, res.UploadID, uploadedParts, ObjectOptions{
-		MTime: oi.ModTime,
+		MTime:            oi.ModTime,
+		NoLock:           opts.NoLock,
+		Versioned:        opts.Versioned,
+		VersionSuspended: opts.VersionSuspended,
+		Transition:       opts.Transition,
 	})
 	return setRestoreHeaderFn(oi, err)
 }

cmd/erasure-server-pool.go

@@ -1625,7 +1625,17 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
 		return oi, err
 	}
 
-	poolIdx, err := z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size)
+	var poolIdx int
+	if dstOpts.MetadataChg && cpSrcDstSame && srcInfo.metadataOnly && srcOpts.VersionID == dstOpts.VersionID {
+		// Restore status updates address an existing version. Keep ordinary
+		// copy/write allocation unchanged, including its latest-version lookup.
+		poolIdx, err = z.getPoolIdxExistingWithOpts(ctx, dstBucket, dstObject, ObjectOptions{
+			VersionID: dstOpts.VersionID, MetadataChg: true, NoLock: true,
+			SkipDecommissioned: true, SkipRebalancing: true,
+		})
+	} else {
+		poolIdx, err = z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size)
+	}
 	if err != nil {
 		return objInfo, err
 	}
@@ -2312,18 +2322,24 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
 		}
 	}()
 
-	if !z.SinglePool() {
+	// Keep object -> upload lock ordering for single-pool completion too:
-		if !opts.NoLock {
+	// native restore holds the object lock while publishing upload parts.
-			lk := z.NewNSLock(bucket, encodeDirObject(object))
+	if !opts.NoLock {
-			lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
+		lk := z.NewNSLock(bucket, encodeDirObject(object))
-			if err != nil {
+		if z.SinglePool() {
-				return objInfo, err
+			// Preserve the set-local lock domain used by single-pool writes.
-			}
+			lk = z.serverPools[0].getHashedSet(object).NewNSLock(bucket, encodeDirObject(object))
-			ctx = lkctx.Context()
-			defer lk.Unlock(lkctx)
 		}
-		opts.NoLock = true
+		lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
+		if err != nil {
+			return objInfo, err
+		}
+		ctx = lkctx.Context()
+		defer lk.Unlock(lkctx)
+	}
+	opts.NoLock = true
 
+	if !z.SinglePool() {
 		// A conditional completion must be evaluated against the logical
 		// latest object across pools, under the object write lock held for
 		// this operation. The pool hosting the upload may only hold a stale
@@ -3330,7 +3346,10 @@ func (z *erasureServerPools) RestoreTransitionedObject(ctx context.Context, buck
 	// Avoid restoring object from a pool being decommissioned.
 	opts.SkipDecommissioned = true
 	opts.NoLock = true
-	idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, opts)
+	// Restore an existing version, not the latest version's write location.
+	lookupOpts := opts
+	lookupOpts.MetadataChg = true
+	idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, lookupOpts)
 	if err != nil {
 		return err
 	}

cmd/object-handlers.go

@@ -4006,8 +4006,9 @@ func (api objectAPIHandlers) PostRestoreObjectHandler(w http.ResponseWriter, r *
 		if _, err := objectAPI.CopyObject(GlobalContext, bucket, object, bucket, object, objInfo, ObjectOptions{
 			VersionID: objInfo.VersionID,
 		}, ObjectOptions{
-			VersionID: objInfo.VersionID,
+			VersionID:   objInfo.VersionID,
-			MTime:     objInfo.ModTime,
+			MTime:       objInfo.ModTime,
+			MetadataChg: true,
 		}); err != nil {
 			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidObjectState), r.URL)
 			return
@@ -4080,14 +4081,16 @@ func (api objectAPIHandlers) PostRestoreObjectHandler(w http.ResponseWriter, r *
 			rreq.SelectParameters.Close()
 			return
 		}
-		opts := ObjectOptions{
+		restoreOpts := ObjectOptions{
 			Transition: TransitionOptions{
 				RestoreRequest: rreq,
 				RestoreExpiry:  restoreExpiry,
 			},
-			VersionID: objInfo.VersionID,
+			VersionID:        objInfo.VersionID,
+			Versioned:        opts.Versioned,
+			VersionSuspended: opts.VersionSuspended,
 		}
-		if err := objectAPI.RestoreTransitionedObject(rctx, bucket, object, opts); err != nil {
+		if err := objectAPI.RestoreTransitionedObject(rctx, bucket, object, restoreOpts); err != nil {
 			s3LogIf(ctx, fmt.Errorf("Unable to restore transitioned bucket/object %s/%s: %w", bucket, object, err))
 			return
 		}

cmd/restore-native_test.go

@@ -0,0 +1,709 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// This file is part of Silo Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"crypto/md5"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"reflect"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/bucket/lifecycle"
+	"github.com/minio/minio/internal/bucket/versioning"
+	"github.com/minio/minio/internal/event"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/minio/internal/pubsub"
+	"github.com/minio/mux"
+)
+
+func nativeRestoreSeed(t *testing.T, r *restoreTest, pool int, null, multipart bool) (ObjectInfo, []byte) {
+	t.Helper()
+	ctx := t.Context()
+	opts := ObjectOptions{Versioned: !null, MTime: UTCNow().Add(-2 * time.Hour), UserDefined: map[string]string{
+		"content-type": "application/x-restore-test", "x-amz-meta-preserve": "keep-me",
+	}}
+	body := bytes.Repeat([]byte("restore-payload"), 100000)
+	var oi ObjectInfo
+	var err error
+	if multipart {
+		parts := [][]byte{bytes.Repeat([]byte("a"), 5<<20), []byte("last-part")}
+		body = bytes.Join(parts, nil)
+		upload, err := r.z.serverPools[pool].NewMultipartUpload(ctx, r.bucket, r.object, opts)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var complete []CompletePart
+		for i, data := range parts {
+			p, err := r.z.serverPools[pool].PutObjectPart(ctx, r.bucket, r.object, upload.UploadID, i+1,
+				mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			complete = append(complete, CompletePart{PartNumber: i + 1, ETag: p.ETag})
+		}
+		oi, err = r.z.serverPools[pool].CompleteMultipartUpload(ctx, r.bucket, r.object, upload.UploadID, complete, opts)
+		if err == nil && len(oi.Parts) != 2 {
+			t.Fatalf("fixture has %d parts, want 2", len(oi.Parts))
+		}
+	} else {
+		oi, err = r.z.serverPools[pool].PutObject(ctx, r.bucket, r.object,
+			mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), opts)
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	oi.VersionID = restoreTestVersionKey(oi.VersionID)
+	topts := r.opts(oi.VersionID)
+	topts.MTime = oi.ModTime
+	topts.Transition = TransitionOptions{Tier: "WARM", ETag: oi.ETag}
+	if err := r.z.serverPools[pool].TransitionObject(ctx, r.bucket, r.object, topts); err != nil {
+		t.Fatal(err)
+	}
+	// Use the production metadata writer to mark the request ongoing.
+	topts.Transition = TransitionOptions{}
+	topts.EvalMetadataFn = func(o *ObjectInfo, _ error) (ReplicateDecision, error) {
+		o.UserDefined[xhttp.AmzRestore] = ongoingRestoreObj().String()
+		return ReplicateDecision{}, nil
+	}
+	if _, err := r.z.PutObjectMetadata(ctx, r.bucket, r.object, topts); err != nil {
+		t.Fatal(err)
+	}
+	return oi, body
+}
+
+func nativeRestoreSnapshot(t *testing.T, r *restoreTest) [][]FileInfo {
+	t.Helper()
+	result := make([][]FileInfo, len(r.z.serverPools))
+	for p, pool := range r.z.serverPools {
+		raw, err := pool.getHashedSet(r.object).getDisks()[0].ReadXL(t.Context(), r.bucket, r.object, false)
+		if errors.Is(err, errFileNotFound) {
+			continue
+		}
+		if err != nil {
+			t.Fatal(err)
+		}
+		infos, err := getFileInfoVersions(raw.Buf, r.bucket, r.object, false)
+		if err != nil {
+			t.Fatal(err)
+		}
+		result[p] = append(infos.Versions, infos.FreeVersions...)
+	}
+	return result
+}
+
+func nativeRestoreOptions(r *restoreTest, id string) ObjectOptions {
+	opts := r.opts(id)
+	opts.Transition = TransitionOptions{RestoreRequest: &RestoreObjectRequest{Days: 1}}
+	return opts
+}
+
+type nativeRestoreFailTier struct {
+	WarmBackend
+	failure error
+}
+
+func (b nativeRestoreFailTier) Get(context.Context, string, remoteVersionID, WarmBackendGetOpts) (io.ReadCloser, error) {
+	return nil, b.failure
+}
+
+func TestNativeRestorePools(t *testing.T) {
+	for _, pools := range []int{1, 2} {
+		for _, multipart := range []bool{false, true} {
+			for _, fail := range []bool{false, true} {
+				t.Run(fmt.Sprintf("pools=%d/multipart=%v/failure=%v", pools, multipart, fail), func(t *testing.T) {
+					r := newRestoreSafetyTest(t, pools, versioning.Enabled)
+					oi, body := nativeRestoreSeed(t, r, 0, false, multipart)
+					before := nativeRestoreSnapshot(t, r)
+					tier := globalTierConfigMgr.drivercache["WARM"].(*restoreTestTier)
+					failure := errors.New("injected remote download failure")
+					if fail {
+						globalTierConfigMgr.drivercache["WARM"] = nativeRestoreFailTier{tier, failure}
+					}
+					ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second)
+					defer cancel()
+					err := r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID))
+					if fail {
+						if !errors.Is(err, failure) {
+							t.Fatalf("restore error=%v, want remote failure", err)
+						}
+						got, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID))
+						if err != nil || got.UserDefined[xhttp.AmzRestore] != "" {
+							t.Fatalf("failed restore did not clear ongoing status: %v, %q", err, got.UserDefined[xhttp.AmzRestore])
+						}
+					} else {
+						if err != nil {
+							t.Fatal(err)
+						}
+						reads := tier.gets.Load()
+						_, got, err := r.read(oi.VersionID)
+						if err != nil || !bytes.Equal(got, body) {
+							t.Fatalf("restored contents differ: %v", err)
+						}
+						if tier.gets.Load() != reads {
+							t.Error("restored read still fetched the remote tier")
+						}
+					}
+					after := nativeRestoreSnapshot(t, r)
+					if len(after[0]) != len(before[0]) || after[0][0].VersionID != before[0][0].VersionID || after[0][0].TransitionedObjName != before[0][0].TransitionedObjName {
+						t.Fatal("restore changed the version stack or remote object identity")
+					}
+					if after[0][0].Metadata["content-type"] != "application/x-restore-test" || after[0][0].Metadata["x-amz-meta-preserve"] != "keep-me" {
+						t.Error("restore lost original metadata")
+					}
+				})
+			}
+		}
+	}
+}
+
+func TestNativeRestoreVersionRouting(t *testing.T) {
+	for _, pool := range []int{0, 1} {
+		for _, null := range []bool{false, true} {
+			t.Run(fmt.Sprintf("target-pool=%d/null=%v", pool, null), func(t *testing.T) {
+				r := newRestoreSafetyTest(t, 2, versioning.Enabled)
+				oi, body := nativeRestoreSeed(t, r, pool, null, false)
+				latest := putConsistencyObject(t, r.z, r.bucket, r.object, 1-pool, "latest-stays", ObjectOptions{Versioned: true, MTime: oi.ModTime.Add(time.Hour)})
+				before := nativeRestoreSnapshot(t, r)
+				ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second)
+				defer cancel()
+				if err := r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID)); err != nil {
+					t.Fatal(err)
+				}
+				_, got, err := r.read(oi.VersionID)
+				if err != nil || !bytes.Equal(got, body) {
+					t.Fatalf("requested version not restored: %v", err)
+				}
+				after := nativeRestoreSnapshot(t, r)
+				if !reflect.DeepEqual(before[1-pool], after[1-pool]) {
+					t.Error("restore changed the latest version in another pool")
+				}
+				if len(after[pool]) != 1 || after[pool][0].VersionID != before[pool][0].VersionID {
+					t.Error("restore created a version alias")
+				}
+				gotInfo, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, ObjectOptions{})
+				if err != nil || gotInfo.VersionID != latest.VersionID {
+					t.Errorf("latest changed: %v", err)
+				}
+				before = nativeRestoreSnapshot(t, r)
+				if err := r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, mustGetUUID())); !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+					t.Errorf("missing version: %v", err)
+				}
+				if !reflect.DeepEqual(before, nativeRestoreSnapshot(t, r)) {
+					t.Error("missing version restore mutated metadata")
+				}
+			})
+		}
+	}
+}
+
+func TestNativeRestoreInitialFailurePreservesMetadata(t *testing.T) {
+	r := newRestoreSafetyTest(t, 1, versioning.Enabled)
+	oi := putConsistencyObject(t, r.z, r.bucket, r.object, 0, "original-latest", ObjectOptions{Versioned: true, UserDefined: map[string]string{"content-type": "application/x-original", "x-amz-meta-preserve": "keep"}})
+	before := nativeRestoreSnapshot(t, r)
+	for _, missing := range []string{mustGetUUID(), nullVersionID} {
+		err := r.z.RestoreTransitionedObject(t.Context(), r.bucket, r.object, nativeRestoreOptions(r, missing))
+		if !isErrVersionNotFound(err) {
+			t.Errorf("missing restore %s: %v", missing, err)
+		}
+		if !reflect.DeepEqual(before, nativeRestoreSnapshot(t, r)) {
+			t.Error("initial restore failure changed stored metadata or version stack")
+		}
+	}
+	_, body, err := r.read(oi.VersionID)
+	if err != nil || string(body) != "original-latest" {
+		t.Errorf("initial restore failure changed content: %v", err)
+	}
+}
+
+func TestNativeRestoreWithoutOuterLock(t *testing.T) {
+	for _, mode := range []string{"single-part", "multipart", "failure-metadata"} {
+		t.Run(mode, func(t *testing.T) {
+			r := newRestoreSafetyTest(t, 1, versioning.Enabled)
+			oi, _ := nativeRestoreSeed(t, r, 0, false, mode == "multipart")
+			tier := globalTierConfigMgr.drivercache["WARM"]
+			failure := errors.New("injected remote failure")
+			if mode == "failure-metadata" {
+				globalTierConfigMgr.drivercache["WARM"] = nativeRestoreFailTier{tier, failure}
+			}
+			er := r.z.serverPools[0].getHashedSet(r.object)
+			lock := er.NewNSLock(r.bucket, r.object)
+			lkctx, err := lock.GetLock(t.Context(), globalOperationTimeout)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer lock.Unlock(lkctx)
+			before := nativeRestoreSnapshot(t, r)
+			ctx, cancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
+			defer cancel()
+			err = er.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID))
+			if err == nil {
+				t.Error("restore wrote through another caller's lock")
+			}
+			if !reflect.DeepEqual(before, nativeRestoreSnapshot(t, r)) {
+				t.Error("NoLock=false failed to protect the stored version")
+			}
+		})
+	}
+}
+
+type nativeRestoreHTTPResult struct {
+	opts ObjectOptions
+	err  error
+}
+type nativeRestoreHTTPObserver struct {
+	ObjectLayer
+	done chan nativeRestoreHTTPResult
+}
+
+func (o nativeRestoreHTTPObserver) RestoreTransitionedObject(ctx context.Context, bucket, object string, opts ObjectOptions) error {
+	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+	err := o.ObjectLayer.RestoreTransitionedObject(ctx, bucket, object, opts)
+	o.done <- nativeRestoreHTTPResult{opts, err}
+	return err
+}
+
+func TestNativeRestorePOSTVersionIdentity(t *testing.T) {
+	for _, state := range []versioning.State{versioning.Enabled, versioning.Suspended} {
+		for _, target := range []int{0, 1} {
+			for _, split := range []bool{false, true} {
+				for _, multipart := range []bool{false, true} {
+					t.Run(fmt.Sprintf("%s/target=%d/split=%v/multipart=%v", state, target, split, multipart), func(t *testing.T) {
+						r := newRestoreSafetyTest(t, 2, state)
+						initConfigSubsystem(t.Context(), r.z)
+						oi, body := nativeRestoreSeed(t, r, target, !multipart, multipart)
+						metaOpts := r.opts(oi.VersionID)
+						metaOpts.MTime = oi.ModTime
+						metaOpts.EvalMetadataFn = func(o *ObjectInfo, _ error) (ReplicateDecision, error) {
+							o.UserDefined[xhttp.AmzRestore] = ""
+							return ReplicateDecision{}, nil
+						}
+						if _, err := r.z.PutObjectMetadata(t.Context(), r.bucket, r.object, metaOpts); err != nil {
+							t.Fatal(err)
+						}
+						if split {
+							putConsistencyObject(t, r.z, r.bucket, r.object, 1-target, "newer-version", ObjectOptions{Versioned: true, MTime: oi.ModTime.Add(time.Hour)})
+						}
+						before := nativeRestoreSnapshot(t, r)
+						observed := nativeRestoreHTTPObserver{r.z, make(chan nativeRestoreHTTPResult, 1)}
+						api := objectAPIHandlers{ObjectAPI: func() ObjectLayer { return observed }}
+						events := make(chan event.Event, 1)
+						done := make(chan struct{})
+						defer close(done)
+						if err := globalHTTPListen.Subscribe(pubsub.MaskFromMaskable(event.ObjectRestoreCompleted), events, done, nil); err != nil {
+							t.Fatal(err)
+						}
+						bodyXML := []byte(`<RestoreRequest xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Days>1</Days></RestoreRequest>`)
+						path := "/" + r.bucket + "/" + r.object + "?restore"
+						if split {
+							path += "&versionId=" + url.QueryEscape(oi.VersionID)
+						}
+						req, err := newTestSignedRequestV4(http.MethodPost, path, int64(len(bodyXML)), bytes.NewReader(bodyXML), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+						if err != nil {
+							t.Fatal(err)
+						}
+						if err = req.ParseForm(); err != nil {
+							t.Fatal(err)
+						}
+						req = mux.SetURLVars(req, map[string]string{"bucket": r.bucket, "object": r.object})
+						rec := httptest.NewRecorder()
+						api.PostRestoreObjectHandler(rec, req)
+						if rec.Code != http.StatusOK {
+							t.Fatalf("POST restore: %d %s", rec.Code, rec.Body.String())
+						}
+						select {
+						case result := <-observed.done:
+							if result.err != nil {
+								t.Fatal(result.err)
+							}
+							if result.opts.VersionID != oi.VersionID || result.opts.Versioned != (state == versioning.Enabled) || result.opts.VersionSuspended != (state == versioning.Suspended) {
+								t.Errorf("background options lost version context: %+v", result.opts)
+							}
+						case <-time.After(5 * time.Second):
+							t.Fatal("background restore did not return")
+						}
+						select {
+						case <-events:
+						case <-time.After(5 * time.Second):
+							t.Fatal("no restore completion event")
+						}
+						after := nativeRestoreSnapshot(t, r)
+						if len(after[target]) != 1 || after[target][0].VersionID != before[target][0].VersionID || after[target][0].TransitionedObjName != before[target][0].TransitionedObjName {
+							t.Fatal("POST restore added a version alias")
+						}
+						if !reflect.DeepEqual(before[1-target], after[1-target]) {
+							t.Error("POST changed another pool's version")
+						}
+						_, got, err := r.read(oi.VersionID)
+						if err != nil || !bytes.Equal(got, body) {
+							t.Fatalf("POST-restored contents differ: %v", err)
+						}
+					})
+				}
+			}
+		}
+	}
+}
+
+// A restore request is internal context, not an exception for normal S3
+// multipart overwrites or for a version that has never completed transition.
+func TestNativeRestoreMultipartOverwriteBoundary(t *testing.T) {
+	for _, mode := range []string{"matching-restore", "ordinary-complete", "not-tiered", "changed-time", "changed-size", "changed-etag"} {
+		t.Run(mode, func(t *testing.T) {
+			r := newRestoreSafetyTest(t, 1, versioning.Enabled)
+			oi, body := nativeRestoreSeed(t, r, 0, false, true)
+			pool := r.z.serverPools[0]
+			if mode == "not-tiered" {
+				er := pool.getHashedSet(r.object)
+				fi, _, _, err := er.getObjectFileInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID), false)
+				if err != nil {
+					t.Fatal(err)
+				}
+				fi.TransitionStatus = lifecycle.TransitionPending
+				for _, disk := range er.getDisks() {
+					if err := disk.WriteMetadata(t.Context(), "", r.bucket, r.object, fi); err != nil {
+						t.Fatal(err)
+					}
+				}
+			}
+			current, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID))
+			if err != nil {
+				t.Fatal(err)
+			}
+			opts := putRestoreOpts(r.bucket, r.object, &RestoreObjectRequest{Days: 1}, current)
+			opts.Transition.RestoreRequest = &RestoreObjectRequest{Days: 1}
+			if mode == "ordinary-complete" {
+				opts.Transition.RestoreRequest = nil
+			}
+			if mode == "changed-time" {
+				opts.MTime = opts.MTime.Add(time.Second)
+			}
+			// Hold the ETag constant for the size case, so only size can reject it.
+			if mode == "changed-size" {
+				opts.UserDefined["etag"] = current.ETag
+				body = append(body, 'x')
+			}
+			if mode == "changed-etag" {
+				opts.UserDefined["etag"] = "different-etag"
+			}
+			upload, err := pool.NewMultipartUpload(t.Context(), r.bucket, r.object, opts)
+			if err != nil {
+				t.Fatal(err)
+			}
+			var complete []CompletePart
+			for i, data := range [][]byte{body[:5<<20], body[5<<20:]} {
+				part, err := pool.PutObjectPart(t.Context(), r.bucket, r.object, upload.UploadID, i+1,
+					mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{})
+				if err != nil {
+					t.Fatal(err)
+				}
+				complete = append(complete, CompletePart{PartNumber: i + 1, ETag: part.ETag})
+			}
+			if mode != "changed-size" && getCompleteMultipartMD5(complete) != current.ETag {
+				t.Fatal("fixture multipart ETag does not match original")
+			}
+			before := nativeRestoreSnapshot(t, r)
+			_, err = pool.CompleteMultipartUpload(t.Context(), r.bucket, r.object, upload.UploadID, complete, opts)
+			if mode == "matching-restore" {
+				if err != nil {
+					t.Fatalf("matching restore rejected: %v", err)
+				}
+				_, got, err := r.read(oi.VersionID)
+				if err != nil || !bytes.Equal(got, body) {
+					t.Fatalf("matching restore contents: %v", err)
+				}
+				return
+			}
+			if !errors.Is(err, errAuthentication) {
+				t.Fatalf("existing version overwrite: %v", err)
+			}
+			if !reflect.DeepEqual(before, nativeRestoreSnapshot(t, r)) {
+				t.Error("rejected completion changed the target")
+			}
+		})
+	}
+}
+
+// Hold the download after the restore snapshot has been read. Object Lock
+// updates must serialize with both successful rehydration and failure cleanup.
+type nativeRestoreBlockedTier struct {
+	WarmBackend
+	entered        chan struct{}
+	release        chan struct{}
+	once           sync.Once
+	failure        error
+	afterFirstPart bool
+}
+
+func (b *nativeRestoreBlockedTier) Get(ctx context.Context, object string, version remoteVersionID, opts WarmBackendGetOpts) (io.ReadCloser, error) {
+	if b.afterFirstPart {
+		r, err := b.WarmBackend.Get(ctx, object, version, opts)
+		if err != nil {
+			return nil, err
+		}
+		return &nativeRestorePartGateReader{ReadCloser: r, gate: b, ctx: ctx, remaining: 5 << 20}, nil
+	}
+	b.once.Do(func() { close(b.entered) })
+	select {
+	case <-b.release:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	if b.failure != nil {
+		return nil, b.failure
+	}
+	return b.WarmBackend.Get(ctx, object, version, opts)
+}
+
+func nativeRestoreLockRequest(t *testing.T, ctx context.Context, r *restoreTest, id, kind string, body []byte) *httptest.ResponseRecorder {
+	t.Helper()
+	sum := md5.Sum(body)
+	req, err := newTestSignedRequestV4(http.MethodPut, "/"+r.bucket+"/"+r.object+"?"+kind+"&versionId="+url.QueryEscape(id), int64(len(body)), bytes.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, map[string]string{xhttp.ContentMD5: base64.StdEncoding.EncodeToString(sum[:])})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := req.ParseForm(); err != nil {
+		t.Fatal(err)
+	}
+	req = mux.SetURLVars(req.WithContext(ctx), map[string]string{"bucket": r.bucket, "object": r.object})
+	rec := httptest.NewRecorder()
+	api := objectAPIHandlers{ObjectAPI: func() ObjectLayer { return r.z }}
+	if kind == "legal-hold" {
+		api.PutObjectLegalHoldHandler(rec, req)
+	} else {
+		api.PutObjectRetentionHandler(rec, req)
+	}
+	return rec
+}
+
+func TestNativeRestoreSerializesObjectLock(t *testing.T) {
+	for _, kind := range []string{"legal-hold", "retention"} {
+		for _, fail := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/failure=%v", kind, fail), func(t *testing.T) {
+				r := newRestoreSafetyTest(t, 1, versioning.Enabled)
+				initConfigSubsystem(t.Context(), r.z)
+				enableBucketObjectLock(t, r.bucket)
+				oi, _ := nativeRestoreSeed(t, r, 0, false, true)
+				initial := []byte(`<LegalHold xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>OFF</Status></LegalHold>`)
+				updated := []byte(`<LegalHold xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>ON</Status></LegalHold>`)
+				keys := []string{strings.ToLower(xhttp.AmzObjectLockLegalHold), ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp}
+				if kind == "retention" {
+					initial = retentionBody("COMPLIANCE", UTCNow().Add(24*time.Hour))
+					updated = retentionBody("COMPLIANCE", UTCNow().Add(48*time.Hour))
+					keys = []string{strings.ToLower(xhttp.AmzObjectLockMode), strings.ToLower(xhttp.AmzObjectLockRetainUntilDate), ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp}
+				}
+				if rec := nativeRestoreLockRequest(t, t.Context(), r, oi.VersionID, kind, initial); rec.Code != http.StatusOK {
+					t.Fatalf("initial protection: %d %s", rec.Code, rec.Body.String())
+				}
+				before, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID))
+				if err != nil {
+					t.Fatal(err)
+				}
+				blocked := &nativeRestoreBlockedTier{WarmBackend: globalTierConfigMgr.drivercache["WARM"], entered: make(chan struct{}), release: make(chan struct{})}
+				if fail {
+					blocked.failure = errors.New("injected download failure after snapshot")
+				}
+				globalTierConfigMgr.drivercache["WARM"] = blocked
+				var release sync.Once
+				unblock := func() { release.Do(func() { close(blocked.release) }) }
+				defer unblock()
+				ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+				defer cancel()
+				result := make(chan error, 1)
+				go func() {
+					result <- r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID))
+				}()
+				select {
+				case <-blocked.entered:
+				case <-ctx.Done():
+					t.Fatal("restore never started download")
+				}
+				updateCtx, updateCancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
+				rec := nativeRestoreLockRequest(t, updateCtx, r, oi.VersionID, kind, updated)
+				updateCancel()
+				if rec.Code == http.StatusOK {
+					t.Error("Object Lock update committed during restore's metadata snapshot")
+					// On the unfixed path this update commits, and the restore
+					// must not subsequently erase the acknowledged protection.
+					readOpts := r.opts(oi.VersionID)
+					readOpts.NoLock = true
+					before, err = r.z.GetObjectInfo(t.Context(), r.bucket, r.object, readOpts)
+					if err != nil {
+						t.Fatal(err)
+					}
+				}
+				if rec.Code != http.StatusOK && (rec.Code != http.StatusServiceUnavailable || !strings.Contains(rec.Body.String(), "<Code>RequestTimeout</Code>")) {
+					t.Errorf("update failed for an unrelated reason: %d %s", rec.Code, rec.Body.String())
+				}
+				unblock()
+				select {
+				case err := <-result:
+					if !errors.Is(err, blocked.failure) {
+						t.Fatalf("restore returned %v, want %v", err, blocked.failure)
+					}
+				case <-ctx.Done():
+					t.Fatal("restore failed to finish")
+				}
+				after, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID))
+				if err != nil {
+					t.Fatal(err)
+				}
+				for _, key := range keys {
+					if before.UserDefined[key] == "" || after.UserDefined[key] != before.UserDefined[key] {
+						t.Errorf("restore changed protection %s: %q -> %q", key, before.UserDefined[key], after.UserDefined[key])
+					}
+				}
+				if rec := nativeRestoreLockRequest(t, t.Context(), r, oi.VersionID, kind, updated); rec.Code != http.StatusOK {
+					t.Fatalf("protection update after restore: %d %s", rec.Code, rec.Body.String())
+				}
+				final, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID))
+				if err != nil {
+					t.Fatal(err)
+				}
+				if !final.ModTime.Equal(before.ModTime) || final.ETag != before.ETag || final.Size != before.Size {
+					t.Error("Object Lock update changed content identity")
+				}
+				changed := keys[0]
+				if kind == "retention" {
+					changed = keys[1]
+				}
+				if final.UserDefined[changed] == before.UserDefined[changed] || final.UserDefined[keys[len(keys)-1]] == before.UserDefined[keys[len(keys)-1]] {
+					t.Error("new protection or its timestamp was not stored")
+				}
+			})
+		}
+	}
+}
+
+func TestNativeRestoreConcurrentCompletion(t *testing.T) {
+	for _, separateSets := range []bool{false, true} {
+		t.Run(fmt.Sprintf("separate-set-locks=%v", separateSets), func(t *testing.T) {
+			r := newRestoreSafetyTest(t, 1, versioning.Enabled)
+			if separateSets {
+				// Local production sets share one map. Independent maps model the
+				// distinct locker groups of distributed sets while reusing test disks.
+				pool := r.z.serverPools[0]
+				sets, count := pool.sets, pool.setCount
+				t.Cleanup(func() { pool.sets, pool.setCount = sets, count })
+				first, second := *sets[0], *sets[0]
+				first.nsMutex, second.nsMutex = newNSLock(false), newNSLock(false)
+				second.setIndex = 1
+				pool.sets, pool.setCount = []*erasureObjects{&first, &second}, 2
+				for i := 0; ; i++ {
+					r.object = fmt.Sprintf("second-set-%d", i)
+					if pool.getHashedSetIndex(r.object) == 1 {
+						break
+					}
+				}
+			}
+			oi, body := nativeRestoreSeed(t, r, 0, false, true)
+			blocked := &nativeRestoreBlockedTier{WarmBackend: globalTierConfigMgr.drivercache["WARM"], entered: make(chan struct{}), release: make(chan struct{}), afterFirstPart: true}
+			globalTierConfigMgr.drivercache["WARM"] = blocked
+			var once sync.Once
+			unblock := func() { once.Do(func() { close(blocked.release) }) }
+			defer unblock()
+			ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second)
+			defer cancel()
+			restored := make(chan error, 1)
+			go func() {
+				restored <- r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID))
+			}()
+			select {
+			case <-blocked.entered:
+			case <-ctx.Done():
+				t.Fatal("restore never started")
+			}
+			// A client's exact-prefix listing can discover the internal upload. It can
+			// try to complete the first part while native restore owns the object.
+			uploads, err := r.z.ListMultipartUploads(t.Context(), r.bucket, r.object, "", "", "", 10)
+			if err != nil || len(uploads.Uploads) != 1 {
+				t.Fatalf("restore upload discovery: %+v, %v", uploads, err)
+			}
+			id := uploads.Uploads[0].UploadID
+			parts, err := r.z.ListObjectParts(t.Context(), r.bucket, r.object, id, 0, 10, ObjectOptions{})
+			if err != nil || len(parts.Parts) != 1 {
+				t.Fatalf("restore must have published its first part: %+v, %v", parts, err)
+			}
+			part := parts.Parts[0]
+			completed := make(chan struct{})
+			var completeErr error
+			go func() {
+				_, completeErr = r.z.CompleteMultipartUpload(ctx, r.bucket, r.object, id, []CompletePart{{PartNumber: 1, ETag: part.ETag}}, r.opts(oi.VersionID))
+				close(completed)
+			}()
+			waitForListingReadLock(t, r.z.serverPools[0].getHashedSet(r.object).nsMutex, r.bucket, r.object, completed)
+			unblock()
+			select {
+			case err := <-restored:
+				if err != nil {
+					t.Errorf("restore blocked by concurrent completion: %v", err)
+				}
+			case <-time.After(5 * time.Second):
+				t.Fatal("restore did not finish")
+			}
+			select {
+			case <-completed:
+			case <-time.After(5 * time.Second):
+				t.Fatal("completion did not finish")
+			}
+			if !errors.As(completeErr, &InvalidUploadID{}) {
+				t.Errorf("completion after native restore: %v, want InvalidUploadID", completeErr)
+			}
+			_, got, err := r.read(oi.VersionID)
+			if err != nil || !bytes.Equal(got, body) {
+				t.Fatalf("restore contents after concurrent completion: %v", err)
+			}
+		})
+	}
+}
+
+type nativeRestorePartGateReader struct {
+	io.ReadCloser
+	gate      *nativeRestoreBlockedTier
+	ctx       context.Context
+	remaining int
+}
+
+func (r *nativeRestorePartGateReader) Read(p []byte) (int, error) {
+	if r.remaining > 0 {
+		if len(p) > r.remaining {
+			p = p[:r.remaining]
+		}
+		n, err := r.ReadCloser.Read(p)
+		r.remaining -= n
+		return n, err
+	}
+	r.gate.once.Do(func() { close(r.gate.entered) })
+	select {
+	case <-r.gate.release:
+	case <-r.ctx.Done():
+		return 0, r.ctx.Err()
+	}
+	return r.ReadCloser.Read(p)
+}