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
new file mode 100644
index 000000000..6917f539d
--- /dev/null
+++ b/cmd/bucket-lifecycle-restore_test.go
@@ -0,0 +1,453 @@
+// 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: "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}}},
+ }
+
+ 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)
+ 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()
+ 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..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
@@ -622,6 +623,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 {
@@ -636,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 8ceae79d4..71857bf91 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: