From 4f72d0ea9493d91b140d7683c6e11989b7e23d7c Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Tue, 3 Nov 2026 22:16:47 +0800 Subject: [PATCH 1/3] fix(restore): preserve encoded object identity and part layout Signed-off-by: Feng Ruohang --- cmd/erasure-object.go | 20 ++- cmd/restore-native-encoding_test.go | 195 ++++++++++++++++++++++++++++ 2 files changed, 211 insertions(+), 4 deletions(-) create mode 100644 cmd/restore-native-encoding_test.go diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index 0451d1c08..b8a610b7b 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -2687,14 +2687,23 @@ 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 - if len(oi.Parts) == 1 { + // 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: + // encrypted multipart data uses that number to derive its decryption key. + if len(oi.Parts) == 1 && !oi.isMultipart() { + actualSize, err := oi.GetActualSize() + if err != nil { + return setRestoreHeaderFn(oi, err) + } + ropts.IndexCB = func() []byte { return oi.Parts[0].Index } var rs *HTTPRangeSpec gr, err := getTransitionedObjectReader(ctx, bucket, object, rs, http.Header{}, oi, opts) if err != nil { return setRestoreHeaderFn(oi, toObjectErr(err, bucket, object)) } defer gr.Close() - hashReader, err := hash.NewReader(ctx, gr, gr.ObjInfo.Size, "", "", gr.ObjInfo.Size) + hashReader, err := hash.NewReader(ctx, gr, gr.ObjInfo.Size, "", "", actualSize) if err != nil { return setRestoreHeaderFn(oi, toObjectErr(err, bucket, object)) } @@ -2719,11 +2728,14 @@ func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket s // rehydrate the parts back on disk as per the original xl.meta prior to transition for _, partInfo := range oi.Parts { - hr, err := hash.NewReader(ctx, io.LimitReader(gr, partInfo.Size), partInfo.Size, "", "", partInfo.Size) + // Stored and logical lengths differ for compressed/encrypted data. + hr, err := hash.NewReader(ctx, io.LimitReader(gr, partInfo.Size), partInfo.Size, "", "", partInfo.ActualSize) if err != nil { return setRestoreHeaderFn(oi, err) } - pInfo, err := er.PutObjectPart(ctx, bucket, object, res.UploadID, partInfo.Number, NewPutObjReader(hr), ObjectOptions{}) + pInfo, err := er.PutObjectPart(ctx, bucket, object, res.UploadID, partInfo.Number, NewPutObjReader(hr), ObjectOptions{ + IndexCB: func() []byte { return partInfo.Index }, + }) if err != nil { return setRestoreHeaderFn(oi, err) } diff --git a/cmd/restore-native-encoding_test.go b/cmd/restore-native-encoding_test.go new file mode 100644 index 000000000..1ee83a927 --- /dev/null +++ b/cmd/restore-native-encoding_test.go @@ -0,0 +1,195 @@ +// 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 . + +package cmd + +import ( + "bytes" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/minio/minio/internal/bucket/versioning" + "github.com/minio/minio/internal/event" + "github.com/minio/minio/internal/hash" + xhttp "github.com/minio/minio/internal/http" + "github.com/minio/minio/internal/kms" + "github.com/minio/minio/internal/pubsub" + "github.com/minio/mux" +) + +// Exercise the stored representation produced by real S3 uploads. Native +// restore copies encoded bytes, but must retain logical sizes and identity. +func TestNativeRestoreEncodedObjects(t *testing.T) { + for _, pools := range []int{1, 2} { + for _, encoding := range []string{"plain", "compressed", "ssec", "s3-compressed"} { + for _, layout := range []string{"single", "multipart", "sparse-one-part"} { + t.Run(fmt.Sprintf("pools=%d/%s/%s", pools, encoding, layout), func(t *testing.T) { + r := newRestoreSafetyTest(t, pools, versioning.Enabled) + initConfigSubsystem(t.Context(), r.z) + r.object = "restore-encoded.txt" + previousKMS, previousTLS := GlobalKMS, globalIsTLS + t.Cleanup(func() { GlobalKMS, globalIsTLS = previousKMS, previousTLS }) + globalIsTLS = true + headers := make(map[string]string) + readHeaders := make(map[string]string) + if encoding == "compressed" || encoding == "s3-compressed" { + enableAttributesPartsCompression(t) + } + if encoding == "ssec" { + headers = ssecTestHeaders(0x53) + readHeaders = ssecTestHeaders(0x53) + } + if encoding == "s3-compressed" { + var err error + GlobalKMS, err = kms.NewBuiltin("restore-test-key", bytes.Repeat([]byte{0x59}, 32)) + if err != nil { + t.Fatal(err) + } + headers[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionAES + } + previousObjectAPI := newObjectLayerFn() + t.Cleanup(func() { setObjectLayer(previousObjectAPI) }) + router := initTestAPIEndPoints(r.z, []string{"NewMultipart", "PutObjectPart", "CompleteMultipart", "PutObject", "GetObject"}) + bodies := [][]byte{bytes.Repeat([]byte("encoded-restore-"), compMinIndexSize/16+1)} + partNumbers := []int{1} + switch layout { + case "multipart": + bodies = append(bodies, []byte("last encoded part")) + partNumbers = []int{1, 2} + case "sparse-one-part": + partNumbers = []int{5} + } + want := bytes.Join(bodies, nil) + if layout == "single" { + headers[xhttp.AmzChecksumCRC32] = mustChecksum(t, hash.ChecksumCRC32, want) + putCopyChecksumSource(t, router, globalActiveCred, r.bucket, r.object, want, headers) + } else { + attributesPartsUpload(t, router, globalActiveCred, r.bucket, r.object, headers, bodies, partNumbers) + } + before, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if before.IsCompressed() != (encoding == "compressed" || encoding == "s3-compressed") { + t.Fatalf("unexpected compression state: %v", before.IsCompressed()) + } + if before.IsCompressed() && len(before.Parts[0].Index) == 0 { + t.Fatal("fixture must have a compression index") + } + if layout == "single" && len(before.Checksum) == 0 { + t.Fatal("fixture must have a stored checksum") + } + objectURL := getGetObjectURL("", r.bucket, r.object) + "?versionId=" + url.QueryEscape(before.VersionID) + beforeGet := attributesPartsSignedRequest(t, router, globalActiveCred, http.MethodGet, objectURL, nil, readHeaders) + if beforeGet.Code != http.StatusOK || !bytes.Equal(beforeGet.Body.Bytes(), want) { + t.Fatalf("fixture GET: %d bytes=%d want=%d", beforeGet.Code, beforeGet.Body.Len(), len(want)) + } + beforeRaw := nativeRestoreSnapshot(t, r) + topts := r.opts(before.VersionID) + topts.MTime = before.ModTime + topts.Transition = TransitionOptions{Tier: "WARM", ETag: before.ETag} + if err := r.z.TransitionObject(t.Context(), r.bucket, r.object, topts); err != nil { + t.Fatal(err) + } + observed := nativeRestoreHTTPObserver{r.z, make(chan nativeRestoreHTTPResult, 1)} + api := objectAPIHandlers{ObjectAPI: func() ObjectLayer { return observed }} + events, done := make(chan event.Event, 1), make(chan struct{}) + defer close(done) + if err := globalHTTPListen.Subscribe(pubsub.MaskFromMaskable(event.ObjectRestoreCompleted), events, done, nil); err != nil { + t.Fatal(err) + } + restoreXML := []byte(`1`) + req, err := newTestSignedRequestV4(http.MethodPost, objectURL+"&restore", int64(len(restoreXML)), bytes.NewReader(restoreXML), 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("restore POST: %d %s", rec.Code, rec.Body.String()) + } + select { + case result := <-observed.done: + if result.err != nil { + t.Fatalf("background restore: %v", result.err) + } + case <-time.After(5 * time.Second): + t.Fatal("restore did not finish") + } + select { + case <-events: + case <-time.After(5 * time.Second): + t.Fatal("restore completion event missing") + } + after, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(before.VersionID)) + if err != nil { + t.Fatal(err) + } + if after.ETag != before.ETag || after.Size != before.Size || !after.ModTime.Equal(before.ModTime) { + t.Error("restore changed content identity") + } + afterRaw := nativeRestoreSnapshot(t, r) + for pool := range beforeRaw { + if len(afterRaw[pool]) != len(beforeRaw[pool]) { + t.Fatalf("pool %d version count changed", pool) + } + for i, fi := range beforeRaw[pool] { + if !bytes.Equal(fi.Checksum, afterRaw[pool][i].Checksum) { + t.Error("restore lost stored checksum") + } + } + } + if len(after.Parts) != len(before.Parts) { + t.Fatal("restore changed part count") + } + for i, part := range after.Parts { + old := before.Parts[i] + if part.Number != old.Number || part.Size != old.Size || part.ActualSize != old.ActualSize || !bytes.Equal(part.Index, old.Index) { + t.Errorf("part %d layout or compression index changed", i) + } + } + tier := globalTierConfigMgr.drivercache["WARM"].(*restoreTestTier) + reads := tier.gets.Load() + got := attributesPartsSignedRequest(t, router, globalActiveCred, http.MethodGet, objectURL, nil, readHeaders) + if got.Code != http.StatusOK || !bytes.Equal(got.Body.Bytes(), want) { + t.Errorf("restored GET: %d bytes=%d want=%d", got.Code, got.Body.Len(), len(want)) + } + if got.Header().Get(xhttp.ETag) != beforeGet.Header().Get(xhttp.ETag) { + t.Error("restore changed client ETag") + } + for i, number := range partNumbers { + part := attributesPartsSignedRequest(t, router, globalActiveCred, http.MethodGet, objectURL+fmt.Sprintf("&partNumber=%d", number), nil, readHeaders) + if part.Code != http.StatusPartialContent || !bytes.Equal(part.Body.Bytes(), bodies[i]) { + t.Errorf("restored part %d: %d bytes=%d want=%d", number, part.Code, part.Body.Len(), len(bodies[i])) + } + } + if tier.gets.Load() != reads { + t.Error("restored GET still used remote tier") + } + }) + } + } + } +} From d9837fb319868690d20fca18e114ab25b76c0a5b Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Tue, 3 Nov 2026 22:32:55 +0800 Subject: [PATCH 2/3] fix(restore): retain legacy part-size compatibility Signed-off-by: Feng Ruohang --- cmd/erasure-object.go | 10 ++++-- cmd/restore-native-encoding_test.go | 55 ++++++++++++++++++++++++++++- 2 files changed, 62 insertions(+), 3 deletions(-) diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index b8a610b7b..f3cd7e562 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -2728,8 +2728,14 @@ func (er erasureObjects) restoreTransitionedObject(ctx context.Context, bucket s // rehydrate the parts back on disk as per the original xl.meta prior to transition for _, partInfo := range oi.Parts { - // Stored and logical lengths differ for compressed/encrypted data. - hr, err := hash.NewReader(ctx, io.LimitReader(gr, partInfo.Size), partInfo.Size, "", "", partInfo.ActualSize) + // Old uncompressed parts may not record ActualSize. Encrypted parts + // derive it from the written ciphertext in PutObjectPart; compressed + // parts need their recorded logical length, including valid zero sizes. + actualSize := partInfo.Size + if oi.IsCompressed() { + actualSize = partInfo.ActualSize + } + hr, err := hash.NewReader(ctx, io.LimitReader(gr, partInfo.Size), partInfo.Size, "", "", actualSize) if err != nil { return setRestoreHeaderFn(oi, err) } diff --git a/cmd/restore-native-encoding_test.go b/cmd/restore-native-encoding_test.go index 1ee83a927..6a4fa3ba9 100644 --- a/cmd/restore-native-encoding_test.go +++ b/cmd/restore-native-encoding_test.go @@ -19,6 +19,7 @@ package cmd import ( "bytes" + "context" "fmt" "net/http" "net/http/httptest" @@ -102,6 +103,10 @@ func TestNativeRestoreEncodedObjects(t *testing.T) { if beforeGet.Code != http.StatusOK || !bytes.Equal(beforeGet.Body.Bytes(), want) { t.Fatalf("fixture GET: %d bytes=%d want=%d", beforeGet.Code, beforeGet.Body.Len(), len(want)) } + beforeETags := beforeGet.Header()[xhttp.ETag] + if len(beforeETags) != 1 || beforeETags[0] == "" { + t.Fatal("fixture GET must expose a nonempty ETag") + } beforeRaw := nativeRestoreSnapshot(t, r) topts := r.opts(before.VersionID) topts.MTime = before.ModTime @@ -176,7 +181,8 @@ func TestNativeRestoreEncodedObjects(t *testing.T) { if got.Code != http.StatusOK || !bytes.Equal(got.Body.Bytes(), want) { t.Errorf("restored GET: %d bytes=%d want=%d", got.Code, got.Body.Len(), len(want)) } - if got.Header().Get(xhttp.ETag) != beforeGet.Header().Get(xhttp.ETag) { + afterETags := got.Header()[xhttp.ETag] + if len(afterETags) != 1 || afterETags[0] == "" || afterETags[0] != beforeETags[0] { t.Error("restore changed client ETag") } for i, number := range partNumbers { @@ -193,3 +199,50 @@ func TestNativeRestoreEncodedObjects(t *testing.T) { } } } + +// XL metadata predating compression has no ActualSize. Conversion and healing +// preserve those zero values; a plain multipart restore must derive the size. +func TestNativeRestoreLegacyPartSizes(t *testing.T) { + r := newRestoreSafetyTest(t, 1, "") + oi, body := nativeRestoreSeed(t, r, 0, true, true) + for _, disk := range r.z.serverPools[0].getHashedSet(r.object).getDisks() { + fi, err := disk.ReadVersion(t.Context(), r.bucket, r.bucket, r.object, oi.VersionID, ReadOptions{}) + if err != nil { + t.Fatal(err) + } + for i := range fi.Parts { + fi.Parts[i].ActualSize = 0 + } + delete(fi.Metadata, ReservedMetadataPrefix+"actual-size") + if err := disk.WriteMetadata(t.Context(), "", r.bucket, r.object, fi); err != nil { + t.Fatal(err) + } + } + before, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID)) + if err != nil { + t.Fatal(err) + } + for _, part := range before.Parts { + if part.ActualSize != 0 { + t.Fatal("legacy fixture was normalized") + } + } + _, got, err := r.read(oi.VersionID) + if err != nil || !bytes.Equal(got, body) { + t.Fatalf("legacy fixture read: %v", err) + } + 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) + } + tier := globalTierConfigMgr.drivercache["WARM"].(*restoreTestTier) + reads := tier.gets.Load() + _, got, err = r.read(oi.VersionID) + if err != nil || !bytes.Equal(got, body) { + t.Fatalf("legacy restored read: %v", err) + } + if tier.gets.Load() != reads { + t.Error("legacy restored read used the remote tier") + } +} From 496eab4b7637d8c8a7cac9a95511d96c080b4bf1 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Tue, 3 Nov 2026 22:46:03 +0800 Subject: [PATCH 3/3] 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") + } + }) + } + } +}