From 001078e9cae9c1e036941f7421cf52474f1c7b9e Mon Sep 17 00:00:00 2001 From: Denis Babchuk Date: Mon, 2 Nov 2026 21:41:37 +0300 Subject: [PATCH 1/2] fix(ilm): expire a restored copy by the version it belongs to Signed-off-by: Denis Babchuk --- cmd/bucket-lifecycle-restore_test.go | 451 +++++++++++++++++++++++++++ cmd/bucket-lifecycle.go | 11 + 2 files changed, 462 insertions(+) create mode 100644 cmd/bucket-lifecycle-restore_test.go diff --git a/cmd/bucket-lifecycle-restore_test.go b/cmd/bucket-lifecycle-restore_test.go new file mode 100644 index 000000000..e435a1f9e --- /dev/null +++ b/cmd/bucket-lifecycle-restore_test.go @@ -0,0 +1,451 @@ +// Copyright (c) 2026 Denis Babchuk +// +// 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" + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/minio/minio/internal/bucket/lifecycle" + "github.com/minio/minio/internal/bucket/versioning" + xhttp "github.com/minio/minio/internal/http" +) + +// restoreTestTier is a remote tier held in memory that counts its reads. +type restoreTestTier struct { + *tierExpiryTestBackend + gets atomic.Int64 +} + +func (b *restoreTestTier) Get(ctx context.Context, object string, rv remoteVersionID, opts WarmBackendGetOpts) (io.ReadCloser, error) { + b.gets.Add(1) + return b.tierExpiryTestBackend.Get(ctx, object, rv, opts) +} + +// restoreTest is a cluster with a remote tier WARM and a bucket whose +// versions are transitioned to it and restored. +type restoreTest struct { + t *testing.T + z *erasureServerPools + bucket string + object string + queue chan expiryOp + labels map[string]string // version ID ("null" for the null version) -> label + bodies map[string][]byte // label -> data +} + +func restoreTestVersionKey(versionID string) string { + if versionID == "" { + return nullVersionID + } + return versionID +} + +// put writes a version of the object to the first pool. A restored version +// is 1 MiB, so that its restored copy has a data directory of its own. +func (r *restoreTest) put(label string, null, restored bool) ObjectInfo { + r.t.Helper() + body := []byte(r.object + " " + label) + if restored { + body = bytes.Repeat(body, (1<<20)/len(body)+1) + } + opts := ObjectOptions{Versioned: !null} + if null { + opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(r.bucket, r.object) + } + oi, err := r.z.serverPools[0].PutObject(r.t.Context(), r.bucket, r.object, + mustGetPutObjReader(r.t, bytes.NewReader(body), int64(len(body)), "", ""), opts) + if err != nil { + r.t.Fatal(err) + } + r.labels[restoreTestVersionKey(oi.VersionID)] = label + r.bodies[label] = body + return oi +} + +// opts returns the options of a request on a version of the object, as the +// S3 handlers set them. +func (r *restoreTest) opts(versionID string) ObjectOptions { + return ObjectOptions{ + VersionID: versionID, + Versioned: globalBucketVersioningSys.PrefixEnabled(r.bucket, r.object), + VersionSuspended: globalBucketVersioningSys.PrefixSuspended(r.bucket, r.object), + } +} + +// restore transitions a version to the tier, restores it, and moves the +// expiry date of the restore into the past. +func (r *restoreTest) restore(oi ObjectInfo) { + r.t.Helper() + ctx := r.t.Context() + vopts := r.opts(oi.VersionID) + topts := vopts + topts.MTime = oi.ModTime + topts.Transition = TransitionOptions{Tier: "WARM", ETag: oi.ETag} + if err := r.z.TransitionObject(ctx, r.bucket, r.object, topts); err != nil { + r.t.Fatal(err) + } + ropts := vopts + ropts.Transition = TransitionOptions{RestoreRequest: &RestoreObjectRequest{Days: 1}} + // Through the pool, as with a single pool: with more than one pool the + // pools layer holds the object lock that the write of the restored copy + // waits for. + if err := r.z.serverPools[0].RestoreTransitionedObject(ctx, r.bucket, r.object, ropts); err != nil { + r.t.Fatal(err) + } + expired := completedRestoreObj(time.Now().Add(-24 * time.Hour).UTC()).String() + mopts := vopts + mopts.MTime = oi.ModTime + mopts.EvalMetadataFn = func(o *ObjectInfo, _ error) (ReplicateDecision, error) { + o.UserDefined[xhttp.AmzRestore] = expired + return ReplicateDecision{}, nil + } + if _, err := r.z.PutObjectMetadata(ctx, r.bucket, r.object, mopts); err != nil { + r.t.Fatal(err) + } +} + +// fileInfo returns the version of the object from a drive of the first pool. +func (r *restoreTest) fileInfo(versionID string) FileInfo { + r.t.Helper() + infos, errs := readAllFileInfo(r.t.Context(), r.z.serverPools[0].getHashedSet(r.object).getDisks(), "", r.bucket, r.object, restoreTestVersionKey(versionID), false, false) + if errs[0] != nil { + r.t.Fatal(errs[0]) + } + return infos[0] +} + +// dataDirs returns how many drives of the first pool hold the data +// directory of a version. +func (r *restoreTest) dataDirs(dataDir string) int { + r.t.Helper() + n := 0 + for _, disk := range r.z.serverPools[0].getHashedSet(r.object).getDisks() { + _, err := os.Stat(filepath.Join(disk.Endpoint().Path, r.bucket, r.object, dataDir)) + if err == nil { + n++ + } else if !os.IsNotExist(err) { + r.t.Fatal(err) + } + } + return n +} + +// versions returns the versions of the object in each pool, newest first, +// and its free versions in all pools. +func (r *restoreTest) versions() (views []string, free int) { + r.t.Helper() + for _, pool := range r.z.serverPools { + rf, err := pool.getHashedSet(r.object).getDisks()[0].ReadXL(r.t.Context(), r.bucket, r.object, false) + if errors.Is(err, errFileNotFound) { + views = append(views, "") + continue + } + if err != nil { + r.t.Fatal(err) + } + fivs, err := getFileInfoVersions(rf.Buf, r.bucket, r.object, false) + if err != nil { + r.t.Fatal(err) + } + var labels []string + for _, fi := range fivs.Versions { + if fi.Deleted { + labels = append(labels, "DM") + } else { + labels = append(labels, r.labels[restoreTestVersionKey(fi.VersionID)]) + } + } + views = append(views, strings.Join(labels, " ")) + free += len(fivs.FreeVersions) + } + return views, free +} + +// read returns the label and the data of a version, or of the latest version +// when versionID is empty, or the error of the read. +func (r *restoreTest) read(versionID string) (string, []byte, error) { + r.t.Helper() + gr, err := r.z.GetObjectNInfo(r.t.Context(), r.bucket, r.object, nil, nil, ObjectOptions{VersionID: versionID}) + if err != nil { + return "", nil, err + } + defer gr.Close() + data, err := io.ReadAll(gr) + if err != nil { + return "", nil, err + } + return r.labels[restoreTestVersionKey(gr.ObjInfo.VersionID)], data, nil +} + +// runQueue runs the queued expiries with an expiry worker. +func (r *restoreTest) runQueue() { + done := make(chan struct{}) + go func() { + defer close(done) + globalExpiryState.Worker(r.queue) + }() + r.queue <- nil // the worker returns once it ran every expiry queued before + <-done +} + +// objInfos returns the versions of the object as a drive of the +// first pool holds them, as the scanner and the walks of decommission and +// the drive heal see them. +func (r *restoreTest) objInfos() []ObjectInfo { + r.t.Helper() + rf, err := r.z.serverPools[0].getHashedSet(r.object).getDisks()[0].ReadXL(r.t.Context(), r.bucket, r.object, false) + if err != nil { + r.t.Fatal(err) + } + fivs, err := getFileInfoVersions(rf.Buf, r.bucket, r.object, false) + if err != nil { + r.t.Fatal(err) + } + versioned := globalBucketVersioningSys.PrefixEnabled(r.bucket, r.object) || globalBucketVersioningSys.PrefixSuspended(r.bucket, r.object) + objInfos := make([]ObjectInfo, len(fivs.Versions)) + for i, fi := range fivs.Versions { + objInfos[i] = fi.ToObjectInfo(r.bucket, r.object, versioned) + } + return objInfos +} + +// The restored copy of a transitioned version expires on the expiry date of +// its restore. The expiry removes the copy and nothing else: the version +// stays where it is in the version stack, transitioned, and is read from the +// tier again. Every path that applies ILM runs the expiry: the scanner and a +// GET or HEAD queue it for an expiry worker, decommission and the drive heal +// run it at once. The expiry of the restored copy of the latest version used +// to delete the object instead: a versioned bucket got a delete marker that +// hid it. With versioning suspended it went to the null version, removed its +// data if there was one and failed if not, and the restored copy stayed. +func TestExpireRestoredCopy(t *testing.T) { + prevTier := globalTierConfigMgr + t.Cleanup(func() { globalTierConfigMgr = prevTier }) + prevExpiry := globalExpiryState + t.Cleanup(func() { globalExpiryState = prevExpiry }) + + // Nothing here is old enough to transition; the rule makes the bucket + // one with a tier, as restored versions are in. + const rules = `tierEnabled` + + `1000WARM` + + type version struct { + label string + null bool // the null version + restored bool // transitioned and restored, the restore has expired + } + testCases := []struct { + name string + versioning versioning.State // "", Enabled or Suspended + versions []version // oldest first + }{ + {name: "versioned", versioning: versioning.Enabled, versions: []version{{label: "v1"}, {label: "v2", restored: true}}}, + {name: "versioned-noncurrent", versioning: versioning.Enabled, versions: []version{{label: "v1", restored: true}, {label: "v2"}}}, + {name: "versioned-null", versioning: versioning.Enabled, versions: []version{{label: "v1", null: true, restored: true}}}, + {name: "suspended", versioning: versioning.Suspended, versions: []version{{label: "v1", null: true}, {label: "v2", restored: true}}}, + {name: "suspended-no-null", versioning: versioning.Suspended, versions: []version{{label: "v1"}, {label: "v2", restored: true}}}, + {name: "suspended-null", versioning: versioning.Suspended, versions: []version{{label: "v1"}, {label: "v2", null: true, restored: true}}}, + {name: "unversioned", versions: []version{{label: "v1", null: true, restored: true}}}, + } + + sources := []struct { + name string + expire func(r *restoreTest, oi ObjectInfo, evt lifecycle.Event) + }{ + {"scanner", func(r *restoreTest, _ ObjectInfo, _ lifecycle.Event) { + lc, err := globalLifecycleSys.Get(r.bucket) + if err != nil { + r.t.Fatal(err) + } + lr, err := globalBucketObjectLockSys.Get(r.bucket) + if err != nil { + r.t.Fatal(err) + } + item := scannerItem{bucket: r.bucket, objectName: r.object, lifeCycle: lc} + item.applyActions(r.t.Context(), r.z, r.objInfos(), lr, &sizeSummary{}, nil) + r.runQueue() + }}, + {"head", func(r *restoreTest, oi ObjectInfo, evt lifecycle.Event) { + applyExpiryRule(evt, lcEventSrc_s3HeadObject, oi) + r.runQueue() + }}, + {"list", func(r *restoreTest, _ ObjectInfo, _ lifecycle.Event) { + // ListObjects of the key with max-keys=1 reads its latest + // version without the versioning of the bucket, a listing of + // versions reads them all. + if _, err := r.z.ListObjects(r.t.Context(), r.bucket, r.object, "", "", 1); err != nil { + r.t.Fatal(err) + } + if _, err := r.z.ListObjectVersions(r.t.Context(), r.bucket, r.object, "", "", "", 100); err != nil { + r.t.Fatal(err) + } + r.runQueue() + }}, + {"decom", func(r *restoreTest, oi ObjectInfo, evt lifecycle.Event) { + for _, o := range r.objInfos() { + if o.VersionID == oi.VersionID { + applyExpiryOnTransitionedObject(r.t.Context(), r.z, o, evt, lcEventSrc_Decom) + return + } + } + r.t.Fatal("version not found on the drive") + }}, + } + + for _, pools := range []int{1, 2} { + t.Run(fmt.Sprintf("pools=%d", pools), func(t *testing.T) { + ctx := t.Context() + dirs, err := getRandomDisks(4 * pools) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { removeRoots(dirs) }) + var endpoints EndpointServerPools + for i := range pools { + endpoints = append(endpoints, mustGetPoolEndpoints(i, dirs[4*i:4*i+4]...)...) + } + obj, _, err := initObjectLayer(ctx, endpoints) + if err != nil { + t.Fatal(err) + } + z := obj.(*erasureServerPools) + // After initObjectLayer, which sets up a new tier config. + tier := &restoreTestTier{tierExpiryTestBackend: &tierExpiryTestBackend{objects: map[string][]byte{}}} + globalTierConfigMgr = NewTierConfigMgr() + globalTierConfigMgr.drivercache["WARM"] = tier + + for _, tc := range testCases { + for _, src := range sources { + t.Run(tc.name+"/"+src.name, func(t *testing.T) { + r := &restoreTest{ + t: t, z: z, + bucket: "restore-" + tc.name + "-" + src.name, object: "obj", + queue: make(chan expiryOp, 100), + labels: map[string]string{}, bodies: map[string][]byte{}, + } + if err := z.MakeBucket(ctx, r.bucket, MakeBucketOptions{VersioningEnabled: tc.versioning != ""}); err != nil { + t.Fatal(err) + } + meta, err := globalBucketMetadataSys.Get(r.bucket) + if err != nil { + t.Fatal(err) + } + if tc.versioning == versioning.Suspended { + meta.VersioningConfigXML = []byte("Suspended") + } + meta.LifecycleConfigXML = []byte(rules) + if err := meta.Save(ctx, z); err != nil { + t.Fatal(err) + } + globalBucketMetadataSys.Set(r.bucket, meta) + es := &expiryState{ctx: ctx, objAPI: z} + workers := []chan expiryOp{r.queue} + es.workers.Store(&workers) + globalExpiryState = es + + var restored ObjectInfo + for _, v := range tc.versions { + oi := r.put(v.label, v.null, v.restored) + if v.restored { + r.restore(oi) + restored = oi + } + } + rlabel := r.labels[restoreTestVersionKey(restored.VersionID)] + + // The fixture: the restored copy is on the drives, + // its restore has expired, ILM expires it. + fi := r.fileInfo(restored.VersionID) + if fi.TransitionStatus != lifecycle.TransitionComplete || fi.DataDir == "" || r.dataDirs(fi.DataDir) == 0 { + t.Fatalf("fixture: %s transition status %q, restored copy on %d drives", rlabel, fi.TransitionStatus, r.dataDirs(fi.DataDir)) + } + oi, err := z.GetObjectInfo(ctx, r.bucket, r.object, r.opts(restored.VersionID)) + if err != nil { + t.Fatal(err) + } + lc, err := globalLifecycleSys.Get(r.bucket) + if err != nil { + t.Fatal(err) + } + lr, err := globalBucketObjectLockSys.Get(r.bucket) + if err != nil { + t.Fatal(err) + } + evt := evalActionFromLifecycle(ctx, *lc, lr, nil, oi) + if !evt.Action.DeleteRestored() || oi.RestoreExpires.IsZero() { + t.Fatalf("fixture: %s restore expires %v, action %v", rlabel, oi.RestoreExpires, evt.Action) + } + before, _ := r.versions() + for key, label := range r.labels { + if _, data, err := r.read(key); err != nil || !bytes.Equal(data, r.bodies[label]) { + t.Fatalf("fixture: GET %s: %v", label, err) + } + } + + src.expire(r, oi, evt) + + after, free := r.versions() + if fmt.Sprint(after) != fmt.Sprint(before) || free != 0 { + t.Errorf("versions in each pool: before %q, after %q with %d free versions", before, after, free) + } + if label, _, err := r.read(""); err != nil || label != tc.versions[len(tc.versions)-1].label { + t.Errorf("GET reads %q, %v; want %q", label, err, tc.versions[len(tc.versions)-1].label) + } + if n := r.dataDirs(fi.DataDir); n != 0 { + t.Errorf("restored copy of %s still on %d drives", rlabel, n) + } + oi, err = z.GetObjectInfo(ctx, r.bucket, r.object, r.opts(restored.VersionID)) + if err != nil { + t.Fatalf("HEAD %s: %v", rlabel, err) + } + if oi.TransitionedObject.Status != lifecycle.TransitionComplete || oi.UserDefined[xhttp.AmzRestore] != "" || !oi.RestoreExpires.IsZero() { + t.Errorf("HEAD %s: transition status %q, x-amz-restore %q", rlabel, oi.TransitionedObject.Status, oi.UserDefined[xhttp.AmzRestore]) + } + for _, v := range tc.versions { + for key, label := range r.labels { + if label != v.label { + continue + } + gets := tier.gets.Load() + _, data, err := r.read(key) + switch { + case err != nil: + t.Errorf("GET %s: %v", v.label, err) + case !bytes.Equal(data, r.bodies[v.label]): + t.Errorf("GET %s reads %d bytes that differ", v.label, len(data)) + case v.restored && tier.gets.Load() == gets: + t.Errorf("GET %s does not read the tier", v.label) + } + } + } + }) + } + } + }) + } +} diff --git a/cmd/bucket-lifecycle.go b/cmd/bucket-lifecycle.go index e7a6d99e5..015c9c410 100644 --- a/cmd/bucket-lifecycle.go +++ b/cmd/bucket-lifecycle.go @@ -622,6 +622,17 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob // delete locally restored copy of object or object version // from the source, while leaving metadata behind. The data on // transitioned tier lies untouched and still accessible + // + // The restored copy belongs to this version, also when it is the + // latest one (DeleteRestoredAction). Without its version ID the + // delete applies to the object: a versioned bucket gets a delete + // marker that hides it, a suspended one strips the null version. + // An empty version ID is the null version, as callers that read + // it without the versioning of the bucket leave it. + opts.VersionID = oi.VersionID + if opts.VersionID == "" { + opts.VersionID = nullVersionID + } opts.Transition.ExpireRestored = true _, err := objectAPI.DeleteObject(ctx, oi.Bucket, oi.Name, opts) if err == nil { From e7103d77cb1d520005a592aec99014d792269d16 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Tue, 3 Nov 2026 20:55:22 +0800 Subject: [PATCH 2/2] fix(ilm): constrain restored-copy cleanup to verified tiered versions Signed-off-by: Feng Ruohang --- cmd/bucket-lifecycle-restore-safety_test.go | 284 ++++++++++++++++++++ cmd/bucket-lifecycle-restore_test.go | 2 + cmd/bucket-lifecycle.go | 6 +- cmd/erasure-object.go | 4 +- cmd/xl-storage-format-v2.go | 5 + 5 files changed, 297 insertions(+), 4 deletions(-) create mode 100644 cmd/bucket-lifecycle-restore-safety_test.go diff --git a/cmd/bucket-lifecycle-restore-safety_test.go b/cmd/bucket-lifecycle-restore-safety_test.go new file mode 100644 index 000000000..9facbfd0d --- /dev/null +++ b/cmd/bucket-lifecycle-restore-safety_test.go @@ -0,0 +1,284 @@ +// 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" + "context" + "errors" + "fmt" + "os" + "path/filepath" + "reflect" + "sync/atomic" + "testing" + "time" + + "github.com/minio/minio/internal/bucket/lifecycle" + "github.com/minio/minio/internal/bucket/versioning" + xhttp "github.com/minio/minio/internal/http" +) + +const restoreSafetyRules = `tierEnabled1000WARM` + +// Reuse the original restore-copy fixture with a small, isolated real cluster. +func newRestoreSafetyTest(t *testing.T, pools int, state versioning.State) *restoreTest { + t.Helper() + prevTier := globalTierConfigMgr + t.Cleanup(func() { globalTierConfigMgr = prevTier }) + dirs, err := getRandomDisks(4 * pools) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { removeRoots(dirs) }) + var endpoints EndpointServerPools + for i := range pools { + endpoints = append(endpoints, mustGetPoolEndpoints(i, dirs[4*i:4*i+4]...)...) + } + obj, _, err := initObjectLayer(t.Context(), endpoints) + if err != nil { + t.Fatal(err) + } + z := obj.(*erasureServerPools) + t.Cleanup(func() { z.Shutdown(context.Background()) }) + globalTierConfigMgr = NewTierConfigMgr() + globalTierConfigMgr.drivercache["WARM"] = &restoreTestTier{tierExpiryTestBackend: &tierExpiryTestBackend{objects: map[string][]byte{}}} + r := &restoreTest{t: t, z: z, bucket: getRandomBucketName(), object: "obj", labels: map[string]string{}, bodies: map[string][]byte{}} + if err := z.MakeBucket(t.Context(), r.bucket, MakeBucketOptions{VersioningEnabled: state != ""}); err != nil { + t.Fatal(err) + } + meta, err := globalBucketMetadataSys.Get(r.bucket) + if err != nil { + t.Fatal(err) + } + if state == versioning.Suspended { + meta.VersioningConfigXML = []byte(`Suspended`) + } + meta.LifecycleConfigXML = []byte(restoreSafetyRules) + if err := meta.Save(t.Context(), z); err != nil { + t.Fatal(err) + } + globalBucketMetadataSys.Set(r.bucket, meta) + return r +} + +type restoreCleanupFaultDisk struct { + consistencyReadFaultDisk + deletes *atomic.Int64 +} + +func (d restoreCleanupFaultDisk) DeleteVersion(ctx context.Context, bucket, object string, fi FileInfo, force bool, opts DeleteOptions) error { + if bucket == d.bucket && object == d.object { + d.deletes.Add(1) + } + return d.StorageAPI.DeleteVersion(ctx, bucket, object, fi, force, opts) +} + +func TestExpireRestoredRequiresFreshMetadata(t *testing.T) { + for _, restored := range []bool{true, false} { + t.Run(fmt.Sprintf("expire-restored=%v", restored), func(t *testing.T) { + r := newRestoreSafetyTest(t, 1, "") + oi := r.put("v1", true, true) + r.restore(oi) + before := r.fileInfo(oi.VersionID) + er := r.z.serverPools[0].getHashedSet(r.object) + original := er.getDisks + disks := original() + var deletes atomic.Int64 + faulty := make([]StorageAPI, len(disks)) + for i, d := range disks { + faulty[i] = restoreCleanupFaultDisk{consistencyReadFaultDisk{d, r.bucket, r.object}, &deletes} + } + er.getDisks = func() []StorageAPI { return faulty } + // Restore the callback before any fixture reads and on a fatal assertion. + defer func() { er.getDisks = original }() + opts := ObjectOptions{VersionID: nullVersionID, Transition: TransitionOptions{ExpireRestored: restored}, Expiration: ExpirationOptions{Expire: restored}} + _, err := er.DeleteObject(t.Context(), r.bucket, r.object, opts) + er.getDisks = original + if restored { + var quorum InsufficientWriteQuorum + if !errors.As(err, &quorum) { + t.Errorf("cleanup without fresh metadata: got %v, want write-quorum error", err) + } + if n := deletes.Load(); n != 0 { + t.Errorf("cleanup attempted %d blind disk deletes", n) + } + if after := r.fileInfo(oi.VersionID); !reflect.DeepEqual(before, after) { + t.Error("failed metadata read changed the stored version") + } + if n := r.dataDirs(before.DataDir); n != len(disks) { + t.Errorf("cleanup removed data: %d/%d directories remain", n, len(disks)) + } + } else { + // Ordinary unversioned DELETE retains the existing tryDel behavior. + if err != nil || deletes.Load() != int64(len(disks)) { + t.Fatalf("ordinary DELETE: err=%v, deletes=%d", err, deletes.Load()) + } + if _, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, ObjectOptions{}); !isErrObjectNotFound(err) { + t.Fatalf("ordinary DELETE left the object: %v", err) + } + } + }) + } +} + +func TestExpireRestoredRenewedOrMissing(t *testing.T) { + r := newRestoreSafetyTest(t, 1, versioning.Enabled) + oi := r.put("v1", false, true) + r.restore(oi) + stale, err := r.z.GetObjectInfo(t.Context(), r.bucket, r.object, r.opts(oi.VersionID)) + if err != nil { + t.Fatal(err) + } + before := r.fileInfo(oi.VersionID) + renewed := completedRestoreObj(time.Now().Add(48 * time.Hour)).String() + opts := r.opts(oi.VersionID) + opts.EvalMetadataFn = func(o *ObjectInfo, _ error) (ReplicateDecision, error) { + o.UserDefined[xhttp.AmzRestore] = renewed + return ReplicateDecision{}, nil + } + if _, err := r.z.PutObjectMetadata(t.Context(), r.bucket, r.object, opts); err != nil { + t.Fatal(err) + } + err = expireTransitionedObject(t.Context(), r.z, &stale, lifecycle.Event{Action: lifecycle.DeleteRestoredAction}, lcEventSrc_Scanner) + if !isErrVersionNotFound(err) { + t.Errorf("stale task should be rejected, got %v", err) + } + if got := r.fileInfo(oi.VersionID); got.Metadata[xhttp.AmzRestore] != renewed || got.DataDir != before.DataDir { + t.Error("stale task changed renewed restore") + } + if n := r.dataDirs(before.DataDir); n != 4 { + t.Errorf("renewed copy has %d directories, want 4", n) + } + if _, data, err := r.read(oi.VersionID); err != nil || !bytes.Equal(data, r.bodies["v1"]) { + t.Fatalf("renewed copy not readable: %v", err) + } + + for _, id := range []string{mustGetUUID(), nullVersionID} { + opts := r.opts(id) + opts.Expiration.Expire = true + opts.Transition.ExpireRestored = true + if _, err := r.z.DeleteObject(t.Context(), r.bucket, r.object, opts); !isErrVersionNotFound(err) { + t.Errorf("missing version %s: %v", id, err) + } + } + missing := r.opts(nullVersionID) + missing.Expiration.Expire = true + missing.Transition.ExpireRestored = true + if _, err := r.z.DeleteObject(t.Context(), r.bucket, "missing-object", missing); !isErrVersionNotFound(err) && !isErrObjectNotFound(err) { + t.Errorf("missing object: %v", err) + } + if got := r.fileInfo(oi.VersionID); got.Metadata[xhttp.AmzRestore] != renewed { + t.Error("missing-target cleanup changed existing restore") + } +} + +func TestExpireRestoredStorageBoundary(t *testing.T) { + for _, status := range []string{"", lifecycle.TransitionPending, lifecycle.TransitionComplete} { + for _, inline := range []bool{false, true} { + for _, requestStatus := range []string{"", lifecycle.TransitionComplete} { + t.Run(fmt.Sprintf("stored=%s/inline=%v/request=%s", status, inline, requestStatus), func(t *testing.T) { + disk, _, err := newXLStorageTestSetup(t) + if err != nil { + t.Fatal(err) + } + const bucket, object = "restore-storage", "obj" + ctx := t.Context() + if err := disk.MakeVol(ctx, bucket); err != nil { + t.Fatal(err) + } + fi := newFileInfo(object, 2, 2) + fi.VersionID, fi.DataDir = mustGetUUID(), mustGetUUID() + fi.ModTime, fi.Size = time.Now().UTC(), 1024 + fi.TransitionStatus = status + fi.AddObjectPart(1, "", fi.Size, fi.Size, fi.ModTime, nil, nil) + fi.Metadata = map[string]string{xhttp.AmzRestore: completedRestoreObj(time.Now().Add(-time.Hour)).String(), "x-amz-meta-preserve": "keep"} + payload := []byte("sole local contents") + if inline { + fi.Data = payload + fi.SetInlineData() + } + if err := disk.WriteMetadata(ctx, "", bucket, object, fi); err != nil { + t.Fatal(err) + } + dir := filepath.Join(disk.Endpoint().Path, bucket, object, fi.DataDir) + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "part.1"), payload, 0o600); err != nil { + t.Fatal(err) + } + raw, err := disk.ReadXL(ctx, bucket, object, true) + if err != nil { + t.Fatal(err) + } + var xl xlMetaV2 + if err := xl.Load(raw.Buf); err != nil { + t.Fatal(err) + } + del := FileInfo{VersionID: fi.VersionID, ExpireRestored: true, TransitionStatus: requestStatus} + dataDir, err := xl.DeleteVersion(del) + if err != nil { + t.Fatal(err) + } + wantDir := "" + if status == lifecycle.TransitionComplete { + wantDir = fi.DataDir + } + if dataDir != wantDir { + t.Errorf("format authorized directory %q, want %q", dataDir, wantDir) + } + before, err := getFileInfo(raw.Buf, bucket, object, fi.VersionID, fileInfoOpts{Data: true}) + if err != nil { + t.Fatal(err) + } + if err := disk.DeleteVersion(ctx, bucket, object, del, false, DeleteOptions{}); err != nil { + t.Fatal(err) + } + afterRaw, err := disk.ReadXL(ctx, bucket, object, true) + if err != nil { + t.Fatal(err) + } + after, err := getFileInfo(afterRaw.Buf, bucket, object, fi.VersionID, fileInfoOpts{Data: true}) + if err != nil { + t.Fatal(err) + } + data, dirErr := os.ReadFile(filepath.Join(dir, "part.1")) + if status != lifecycle.TransitionComplete { + if !reflect.DeepEqual(before, after) { + t.Error("non-complete cleanup changed metadata or inline data") + } + if dirErr != nil || !bytes.Equal(data, payload) { + t.Errorf("sole local contents removed: %v", dirErr) + } + } else { + if !os.IsNotExist(dirErr) { + t.Errorf("completed restore directory still present: %v", dirErr) + } + if len(after.Data) != 0 || after.Metadata[xhttp.AmzRestore] != "" { + t.Error("completed restore kept cached data or restore headers") + } + if after.TransitionStatus != status || after.VersionID != before.VersionID || after.Metadata["x-amz-meta-preserve"] != "keep" { + t.Error("cleanup changed the transitioned version identity or user metadata") + } + } + }) + } + } + } +} diff --git a/cmd/bucket-lifecycle-restore_test.go b/cmd/bucket-lifecycle-restore_test.go index e435a1f9e..6917f539d 100644 --- a/cmd/bucket-lifecycle-restore_test.go +++ b/cmd/bucket-lifecycle-restore_test.go @@ -270,6 +270,7 @@ func TestExpireRestoredCopy(t *testing.T) { {name: "suspended", versioning: versioning.Suspended, versions: []version{{label: "v1", null: true}, {label: "v2", restored: true}}}, {name: "suspended-no-null", versioning: versioning.Suspended, versions: []version{{label: "v1"}, {label: "v2", restored: true}}}, {name: "suspended-null", versioning: versioning.Suspended, versions: []version{{label: "v1"}, {label: "v2", null: true, restored: true}}}, + {name: "suspended-null-noncurrent", versioning: versioning.Suspended, versions: []version{{label: "v1", null: true, restored: true}, {label: "v2"}}}, {name: "unversioned", versions: []version{{label: "v1", null: true, restored: true}}}, } @@ -334,6 +335,7 @@ func TestExpireRestoredCopy(t *testing.T) { t.Fatal(err) } z := obj.(*erasureServerPools) + t.Cleanup(func() { z.Shutdown(context.Background()) }) // After initObjectLayer, which sets up a new tier config. tier := &restoreTestTier{tierExpiryTestBackend: &tierExpiryTestBackend{objects: map[string][]byte{}}} globalTierConfigMgr = NewTierConfigMgr() diff --git a/cmd/bucket-lifecycle.go b/cmd/bucket-lifecycle.go index 015c9c410..5c5597587 100644 --- a/cmd/bucket-lifecycle.go +++ b/cmd/bucket-lifecycle.go @@ -611,8 +611,9 @@ func enqueueTransitionImmediate(obj ObjectInfo, src lcEventSrc) { func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *ObjectInfo, lcEvent lifecycle.Event, src lcEventSrc) error { traceFn := globalLifecycleSys.trace(*oi) opts := ObjectOptions{ - Versioned: globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name), - Expiration: ExpirationOptions{Expire: true}, + Versioned: globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name), + VersionSuspended: globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name), + Expiration: ExpirationOptions{Expire: true}, } if lcEvent.Action.DeleteVersioned() { opts.VersionID = oi.VersionID @@ -647,7 +648,6 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob // touching the remote tier. If the lifecycle configuration changed after // the scanner queued this task, DeleteObject rejects the stale action while // the remote object is still protected. - opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name) keepRemote := opts.VersionID == "" && (opts.Versioned || (opts.VersionSuspended && oi.VersionID != "" && oi.VersionID != nullVersionID)) if lcEvent.Action == lifecycle.DeleteAction || lcEvent.Action == lifecycle.DeleteVersionAction { diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index 5c9ee5d80..56737e962 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -2114,7 +2114,9 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string tryDel := false if gerr != nil && goi.Name == "" { if _, ok := gerr.(InsufficientReadQuorum); ok { - if opts.Versioned || opts.VersionSuspended || countOnlineDisks(storageDisks) < len(storageDisks)/2+1 { + // A restored-copy expiry needs fresh metadata to confirm its target + // and deadline; ordinary unversioned DELETE may still try a blind delete. + if opts.Transition.ExpireRestored || opts.Versioned || opts.VersionSuspended || countOnlineDisks(storageDisks) < len(storageDisks)/2+1 { // Add an MRF heal for next time. er.addPartial(bucket, object, opts.VersionID) return objInfo, InsufficientWriteQuorum{} diff --git a/cmd/xl-storage-format-v2.go b/cmd/xl-storage-format-v2.go index 5a054f86e..69465c9ab 100644 --- a/cmd/xl-storage-format-v2.go +++ b/cmd/xl-storage-format-v2.go @@ -1498,6 +1498,11 @@ func (x *xlMetaV2) DeleteVersion(fi FileInfo) (string, error) { } switch { case fi.ExpireRestored: + // Only a completed transition has a remote copy. Trust the selected + // stored version, not the transition status supplied by the caller. + if string(ver.ObjectV2.MetaSys[metaTierStatus]) != lifecycle.TransitionComplete { + return "", nil + } ver.ObjectV2.RemoveRestoreHdrs() err = x.setIdx(i, *ver) case fi.TransitionStatus == lifecycle.TransitionComplete: