From 4f72d0ea9493d91b140d7683c6e11989b7e23d7c Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Tue, 3 Nov 2026 22:16:47 +0800 Subject: [PATCH] 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") + } + }) + } + } + } +}