From 496eab4b7637d8c8a7cac9a95511d96c080b4bf1 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Tue, 3 Nov 2026 22:46:03 +0800 Subject: [PATCH] fix(restore): preserve safe read context through nested lookups Carry restore context through POST status reads, metadata copy routing, null-version Object Lock checks and multipart completion. A failed read must preserve the target and must not enqueue its remote object for garbage collection. Cover initial/status POST failures and metadata loss during single and multipart rehydration on real disks. Signed-off-by: Feng Ruohang --- cmd/bucket-object-lock.go | 6 +- cmd/erasure-multipart.go | 3 +- cmd/erasure-object.go | 12 +- cmd/erasure-server-pool.go | 8 +- cmd/object-handlers.go | 7 +- cmd/object-retention-overwrite_test.go | 2 +- cmd/restore-native_test.go | 192 +++++++++++++++++++++++++ 7 files changed, 218 insertions(+), 12 deletions(-) diff --git a/cmd/bucket-object-lock.go b/cmd/bucket-object-lock.go index 397b039ef..293042fc3 100644 --- a/cmd/bucket-object-lock.go +++ b/cmd/bucket-object-lock.go @@ -251,8 +251,9 @@ func enforceRetentionBypassForPut(ctx context.Context, r *http.Request, oi Objec // checkObjectLockedForOverwrite re-checks under the object write lock whether an // in-place write to the null version (such as a directory object) would -// overwrite a locked version in an Object Lock bucket. -func checkObjectLockedForOverwrite(ctx context.Context, bucket, object string, getObjectInfoFn GetObjectInfoFn) error { +// overwrite a locked version in an Object Lock bucket. Restore context keeps +// failed reads from deleting the version being rehydrated. +func checkObjectLockedForOverwrite(ctx context.Context, bucket, object string, getObjectInfoFn GetObjectInfoFn, transition TransitionOptions) error { if isMinioMetaBucketName(bucket) { return nil } @@ -265,6 +266,7 @@ func checkObjectLockedForOverwrite(ctx context.Context, bucket, object string, g } objInfo, err := getObjectInfoFn(ctx, bucket, object, ObjectOptions{ VersionID: nullVersionID, + Transition: transition, NoLock: true, NoAuditLog: true, }) diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index 4b2893c08..c68fbc446 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -1694,7 +1694,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str getObjectInfo = opts.replicaObjectInfo } if fi.VersionID == "" || fi.VersionID == nullVersionID { - if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo); err != nil { + if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo, opts.Transition); err != nil { return oi, toObjectErr(err, bucket, object) } } else { @@ -1702,6 +1702,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str VersionID: fi.VersionID, Versioned: opts.Versioned, VersionSuspended: opts.VersionSuspended, + Transition: opts.Transition, NoLock: true, }) switch { diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index abeeee82d..a8b9ba1f2 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -156,7 +156,7 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d if dstOpts.replicaObjectInfo != nil { getObjectInfo = dstOpts.replicaObjectInfo } - if err := checkObjectLockedForOverwrite(ctx, dstBucket, dstObject, getObjectInfo); err != nil { + if err := checkObjectLockedForOverwrite(ctx, dstBucket, dstObject, getObjectInfo, dstOpts.Transition); err != nil { return oi, toObjectErr(err, dstBucket, dstObject) } } @@ -1673,7 +1673,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st if opts.replicaObjectInfo != nil { getObjectInfo = opts.replicaObjectInfo } - if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo); err != nil { + if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo, opts.Transition); err != nil { return ObjectInfo{}, toObjectErr(err, bucket, object) } } @@ -2655,9 +2655,10 @@ func (er erasureObjects) updateRestoreMetadata(ctx context.Context, bucket, obje VersionID: oi.VersionID, Transition: opts.Transition, }, ObjectOptions{ - VersionID: oi.VersionID, - MTime: oi.ModTime, - NoLock: opts.NoLock, + VersionID: oi.VersionID, + MTime: oi.ModTime, + NoLock: opts.NoLock, + Transition: opts.Transition, }); err != nil { storageLogIf(ctx, fmt.Errorf("Unable to update transition restore metadata for %s/%s(%s): %s", bucket, object, oi.VersionID, err)) return err @@ -2699,6 +2700,7 @@ func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket s oi = actualfi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended) ropts := putRestoreOpts(bucket, object, opts.Transition.RestoreRequest, oi) ropts.NoLock = opts.NoLock + ropts.Transition = opts.Transition // Rehydration copies encoded bytes without changing their content identity. ropts.PreserveETag = oi.ETag // Even a one-part multipart upload must retain its original part number: diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index db2973717..52949f849 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1630,8 +1630,12 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec // 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, + VersionID: dstOpts.VersionID, + MetadataChg: true, + NoLock: true, + Transition: dstOpts.Transition, + SkipDecommissioned: true, + SkipRebalancing: true, }) } else { poolIdx, err = z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size) diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go index 2843f42d1..42b4cd0e0 100644 --- a/cmd/object-handlers.go +++ b/cmd/object-handlers.go @@ -3951,6 +3951,8 @@ func (api objectAPIHandlers) PostRestoreObjectHandler(w http.ResponseWriter, r * return } + // Preserve a failed restore read even before the request body is parsed. + opts.Transition.RestoreRequest = &RestoreObjectRequest{} objInfo, err := getObjectInfo(ctx, bucket, object, opts) if err != nil { writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL) @@ -3976,6 +3978,7 @@ func (api objectAPIHandlers) PostRestoreObjectHandler(w http.ResponseWriter, r * writeErrorResponse(ctx, w, apiErr, r.URL) return } + opts.Transition.RestoreRequest = rreq statusCode := http.StatusOK alreadyRestored := false if err == nil { @@ -4004,9 +4007,11 @@ func (api objectAPIHandlers) PostRestoreObjectHandler(w http.ResponseWriter, r * } objInfo.UserDefined = metadata if _, err := objectAPI.CopyObject(GlobalContext, bucket, object, bucket, object, objInfo, ObjectOptions{ - VersionID: objInfo.VersionID, + VersionID: objInfo.VersionID, + Transition: opts.Transition, }, ObjectOptions{ VersionID: objInfo.VersionID, + Transition: opts.Transition, MTime: objInfo.ModTime, MetadataChg: true, }); err != nil { diff --git a/cmd/object-retention-overwrite_test.go b/cmd/object-retention-overwrite_test.go index 83ce43ebe..7c864e733 100644 --- a/cmd/object-retention-overwrite_test.go +++ b/cmd/object-retention-overwrite_test.go @@ -57,7 +57,7 @@ func TestObjectLockOverwriteConfigUnavailable(t *testing.T) { err := checkObjectLockedForOverwrite(t.Context(), "lock-config-unavailable", "object", func(context.Context, string, string, ObjectOptions) (ObjectInfo, error) { t.Fatal("object must not be read without a bucket configuration") return ObjectInfo{}, nil - }) + }, TransitionOptions{}) if !errors.Is(err, errServerNotInitialized) { t.Fatalf("unavailable lock configuration must fail closed, got %v", err) } diff --git a/cmd/restore-native_test.go b/cmd/restore-native_test.go index dfe7fe6f4..b01656867 100644 --- a/cmd/restore-native_test.go +++ b/cmd/restore-native_test.go @@ -28,9 +28,11 @@ import ( "net/http" "net/http/httptest" "net/url" + "path/filepath" "reflect" "strings" "sync" + "sync/atomic" "testing" "time" @@ -742,3 +744,193 @@ func TestMultipartCompletionLockAlias(t *testing.T) { } } } + +type nativeRestoreReadHook struct { + io.ReadCloser + remaining int + mutate func() +} + +func (r *nativeRestoreReadHook) Read(p []byte) (int, error) { + n, err := r.ReadCloser.Read(p) + r.remaining -= n + if r.remaining == 0 && r.mutate != nil { + f := r.mutate + r.mutate = nil + f() + } + return n, err +} + +type nativeRestoreReadHookTier struct { + WarmBackend + size int + mutate func() +} + +func (b nativeRestoreReadHookTier) Get(ctx context.Context, object string, version remoteVersionID, opts WarmBackendGetOpts) (io.ReadCloser, error) { + rd, err := b.WarmBackend.Get(ctx, object, version, opts) + if err != nil { + return nil, err + } + return &nativeRestoreReadHook{rd, b.size, b.mutate}, nil +} + +// Remove only the addressed entry from a majority of disks, keeping shards +// and the healthy newer version. Return errors to the synchronous caller. +func nativeRestoreLoseMetadata(ctx context.Context, disks []StorageAPI, bucket, object, id string) error { + for _, disk := range disks[1:] { + raw, err := disk.ReadXL(ctx, bucket, object, true) + if err != nil { + return err + } + var meta xlMetaV2 + if err := meta.Load(raw.Buf); err != nil { + return err + } + del := FileInfo{VersionID: id} + del.SetSkipTierFreeVersion() + if _, err := meta.DeleteVersion(del); err != nil { + return err + } + data, err := meta.AppendTo(nil) + if err != nil { + return err + } + if err := disk.WriteAll(ctx, bucket, filepath.Join(object, xlStorageFormatFile), data); err != nil { + return err + } + } + return nil +} + +// A target can lose metadata quorum after the restore snapshot. Nested +// completion/retention reads must not turn rehydration into remote GC. +func TestNativeRestoreCompletionReadSafety(t *testing.T) { + for _, null := range []bool{false, true} { + for _, multipart := range []bool{false, true} { + t.Run(fmt.Sprintf("null=%v/multipart=%v", null, multipart), func(t *testing.T) { + r := newRestoreSafetyTest(t, 1, versioning.Enabled) + oi, body := nativeRestoreSeed(t, r, 0, null, multipart) + latest := r.put("newer", false, false) + enableBucketObjectLock(t, r.bucket) + disks := r.z.serverPools[0].getHashedSet(r.object).getDisks() + var mutationErr error + mutated := false + mutate := func() { + mutated = true + mutationErr = nativeRestoreLoseMetadata(t.Context(), disks, r.bucket, r.object, oi.VersionID) + } + tier := globalTierConfigMgr.drivercache["WARM"].(*restoreTestTier) + globalTierConfigMgr.drivercache["WARM"] = nativeRestoreReadHookTier{tier, len(body), mutate} + 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 ctx.Err() != nil { + t.Fatalf("restore timed out: %v", err) + } + if !mutated || mutationErr != nil { + t.Fatalf("invalid injection: mutated=%v err=%v", mutated, mutationErr) + } + t.Logf("restore after addressed metadata lost: %v", err) + for i, disk := range disks { + raw, e := disk.ReadXL(t.Context(), r.bucket, r.object, true) + if e != nil { + t.Fatal(e) + } + versions, e := getFileInfoVersions(raw.Buf, r.bucket, r.object, false) + if e != nil { + t.Fatal(e) + } + if len(versions.FreeVersions) > 0 { + t.Errorf("disk %d restore manufactured %d remote GC entries", i, len(versions.FreeVersions)) + } + } + if err == nil { + reads := tier.gets.Load() + _, got, readErr := r.read(oi.VersionID) + if readErr != nil || !bytes.Equal(got, body) || tier.gets.Load() != reads { + t.Errorf("successful restore did not rehydrate its own content: %v", readErr) + } + } + _, got, readErr := r.read(latest.VersionID) + if readErr != nil || !bytes.Equal(got, r.bodies["newer"]) { + t.Errorf("restore changed the other version: %v", readErr) + } + }) + } + } +} + +type nativeRestoreStatusHook struct { + ObjectLayer + beforeCopy func() + copyCalled bool + backgroundCalled atomic.Bool +} + +func (h *nativeRestoreStatusHook) CopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, info ObjectInfo, srcOpts, dstOpts ObjectOptions) (ObjectInfo, error) { + h.copyCalled = true + h.beforeCopy() + return h.ObjectLayer.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, info, srcOpts, dstOpts) +} + +func (h *nativeRestoreStatusHook) RestoreTransitionedObject(context.Context, string, string, ObjectOptions) error { + h.backgroundCalled.Store(true) + return errors.New("unexpected background restore after failed status update") +} + +func TestNativeRestorePOSTReadSafety(t *testing.T) { + for _, pools := range []int{1, 2} { + for _, stage := range []string{"initial", "status-copy"} { + t.Run(fmt.Sprintf("pools=%d/%s", pools, stage), func(t *testing.T) { + r := newRestoreSafetyTest(t, pools, versioning.Enabled) + initConfigSubsystem(t.Context(), r.z) + oi, _ := nativeRestoreSeed(t, r, 0, false, false) + 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) + } + r.put("newer", false, false) + disks := r.z.serverPools[0].getHashedSet(r.object).getDisks() + var before [][]byte + mutate := func() { + if err := nativeRestoreLoseMetadata(t.Context(), disks, r.bucket, r.object, oi.VersionID); err != nil { + t.Fatal(err) + } + before = restoreBoundarySnapshot(t, r) + } + hook := &nativeRestoreStatusHook{ObjectLayer: r.z, beforeCopy: mutate} + if stage == "initial" { + mutate() + } + api := objectAPIHandlers{ObjectAPI: func() ObjectLayer { return hook }} + body := []byte(`1`) + req, err := newTestSignedRequestV4(http.MethodPost, "/"+r.bucket+"/"+r.object+"?restore&versionId="+url.QueryEscape(oi.VersionID), int64(len(body)), bytes.NewReader(body), 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 < 400 { + t.Fatalf("POST acknowledged failed read: %d %s", rec.Code, rec.Body.String()) + } + if hook.copyCalled != (stage == "status-copy") || hook.backgroundCalled.Load() { + t.Fatal("request followed the wrong failure path") + } + if !reflect.DeepEqual(before, restoreBoundarySnapshot(t, r)) { + t.Error("failed POST read changed stored metadata or free versions") + } + }) + } + } +}