diff --git a/cmd/bucket-lifecycle.go b/cmd/bucket-lifecycle.go
index 866343bc1..4b22130b8 100644
--- a/cmd/bucket-lifecycle.go
+++ b/cmd/bucket-lifecycle.go
@@ -632,13 +632,25 @@ func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *Ob
}
// Delete remote object from warm-tier
- err := deleteObjectFromRemoteTier(ctx, oi.TransitionedObject.Name, oi.TransitionedObject.VersionID, oi.TransitionedObject.Tier)
- if err == nil {
- // Skip adding free version since we successfully deleted the
- // remote object
- opts.SkipFreeVersion = true
- } else {
- transitionLogIf(ctx, err)
+ //
+ // Unless the expiry only puts a delete marker on top of the version, as
+ // for an object that is not transitioned: with versioning enabled, or
+ // suspended and the version is not the null version, the version stays
+ // as a noncurrent version and must remain readable. Its remote object is
+ // removed later, through the free version added when the version itself
+ // is deleted.
+ opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name)
+ keepRemote := opts.VersionID == "" &&
+ (opts.Versioned || (opts.VersionSuspended && oi.VersionID != "" && oi.VersionID != nullVersionID))
+ if !keepRemote {
+ err := deleteObjectFromRemoteTier(ctx, oi.TransitionedObject.Name, oi.TransitionedObject.VersionID, oi.TransitionedObject.Tier)
+ if err == nil {
+ // Skip adding free version since we successfully deleted the
+ // remote object
+ opts.SkipFreeVersion = true
+ } else {
+ transitionLogIf(ctx, err)
+ }
}
// Now, delete object from hot-tier namespace
diff --git a/cmd/bucket-lifecycle_test.go b/cmd/bucket-lifecycle_test.go
index a37e3492d..952369a55 100644
--- a/cmd/bucket-lifecycle_test.go
+++ b/cmd/bucket-lifecycle_test.go
@@ -19,11 +19,17 @@ package cmd
import (
"bytes"
+ "context"
+ "io"
"net/http"
+ "slices"
+ "strings"
+ "sync"
"testing"
"time"
"github.com/minio/minio/internal/bucket/lifecycle"
+ objectlock "github.com/minio/minio/internal/bucket/object/lock"
xhttp "github.com/minio/minio/internal/http"
)
@@ -245,3 +251,180 @@ func TestValidateTransitionTier(t *testing.T) {
}
}
}
+
+// tierExpiryTestBackend is a remote tier held in memory.
+type tierExpiryTestBackend struct {
+ mu sync.Mutex
+ objects map[string][]byte
+ removed []string
+}
+
+func (b *tierExpiryTestBackend) Put(ctx context.Context, object string, r io.Reader, length int64) (remoteVersionID, error) {
+ return b.PutWithMeta(ctx, object, r, length, nil)
+}
+
+func (b *tierExpiryTestBackend) PutWithMeta(_ context.Context, object string, r io.Reader, _ int64, _ map[string]string) (remoteVersionID, error) {
+ data, err := io.ReadAll(r)
+ if err != nil {
+ return "", err
+ }
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ b.objects[object] = data
+ return "", nil
+}
+
+func (b *tierExpiryTestBackend) Get(_ context.Context, object string, _ remoteVersionID, opts WarmBackendGetOpts) (io.ReadCloser, error) {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ data, ok := b.objects[object]
+ if !ok {
+ return nil, ObjectNotFound{Bucket: "tier", Object: object}
+ }
+ if opts.length > 0 {
+ data = data[opts.startOffset : opts.startOffset+opts.length]
+ }
+ return io.NopCloser(bytes.NewReader(data)), nil
+}
+
+func (b *tierExpiryTestBackend) Remove(_ context.Context, object string, _ remoteVersionID) error {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ delete(b.objects, object)
+ b.removed = append(b.removed, object)
+ return nil
+}
+
+func (b *tierExpiryTestBackend) InUse(context.Context) (bool, error) { return false, nil }
+
+// The expiry of a transitioned latest version that only puts a delete marker
+// on top of it must keep its remote object, so the version stays readable as
+// a noncurrent version. The remote object goes when the version is deleted.
+func TestExpireTransitionedObjectKeepsNoncurrentData(t *testing.T) {
+ z, _ := migrationTestPools(t)
+ ctx := t.Context()
+ tier := &tierExpiryTestBackend{objects: map[string][]byte{}}
+ prevTier := globalTierConfigMgr
+ globalTierConfigMgr = NewTierConfigMgr()
+ globalTierConfigMgr.drivercache["WARM"] = tier
+ t.Cleanup(func() { globalTierConfigMgr = prevTier })
+
+ const expire = `expEnabled` +
+ `2020-01-01T00:00:00Z`
+
+ testCases := []struct {
+ name string
+ versioning string // "", Enabled or Suspended
+ nullLatest bool // the latest version is the null version
+ keep bool // the latest version survives as noncurrent
+ want []string // versions after the expiry, newest first
+ }{
+ {name: "versioned", versioning: "Enabled", keep: true, want: []string{"DM", "v2", "v1"}},
+ {name: "suspended", versioning: "Suspended", keep: true, want: []string{"DM", "v2", "v1"}},
+ {name: "suspended-null", versioning: "Suspended", nullLatest: true, want: []string{"DM", "v1"}},
+ {name: "unversioned", nullLatest: true, want: nil},
+ }
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ bucket := "tier-expiry-" + tc.name
+ if err := z.MakeBucket(ctx, bucket, MakeBucketOptions{VersioningEnabled: tc.versioning != ""}); err != nil {
+ t.Fatal(err)
+ }
+ meta, err := globalBucketMetadataSys.Get(bucket)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if tc.versioning == "Suspended" {
+ meta.VersioningConfigXML = []byte("Suspended")
+ }
+ meta.LifecycleConfigXML = []byte(expire)
+ if err := meta.Save(ctx, z); err != nil {
+ t.Fatal(err)
+ }
+ globalBucketMetadataSys.Set(bucket, meta)
+
+ // tiered writes a version whose data is in the remote tier.
+ labels := map[string]string{}
+ tiered := func(label string, opts ObjectOptions) ObjectInfo {
+ t.Helper()
+ oi := putConsistencyObject(t, z, bucket, "obj", 0, "obj "+label, opts)
+ vid := migrationTestVersion(oi)
+ infos, errs := readAllFileInfo(ctx, z.serverPools[0].getHashedSet("obj").getDisks(), "", bucket, "obj", vid, false, false)
+ if errs[0] != nil {
+ t.Fatal(errs[0])
+ }
+ fi := infos[0]
+ fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", bucket+"/"+label
+ if _, err := tier.Put(ctx, fi.TransitionedObjName, strings.NewReader("obj "+label), -1); err != nil {
+ t.Fatal(err)
+ }
+ if err := z.serverPools[0].DecomTieredObject(ctx, bucket, "obj", fi, ObjectOptions{}); err != nil {
+ t.Fatal(err)
+ }
+ labels[vid] = label
+ return oi
+ }
+ var latest ObjectInfo
+ if tc.versioning != "" {
+ tiered("v1", ObjectOptions{Versioned: true})
+ }
+ if tc.nullLatest {
+ latest = tiered("v2", ObjectOptions{})
+ } else {
+ latest = tiered("v2", ObjectOptions{Versioned: true})
+ }
+
+ // Expire the latest version as the expiry worker does for the scanner.
+ getOpts := ObjectOptions{Versioned: tc.versioning == "Enabled", VersionSuspended: tc.versioning == "Suspended"}
+ oi, err := z.GetObjectInfo(ctx, bucket, "obj", getOpts)
+ if err != nil {
+ t.Fatal(err)
+ }
+ lc, err := globalLifecycleSys.Get(bucket)
+ if err != nil {
+ t.Fatal(err)
+ }
+ evt := evalActionFromLifecycle(ctx, *lc, objectlock.Retention{}, nil, oi)
+ if evt.Action != lifecycle.DeleteAction || oi.TransitionedObject.Status != lifecycle.TransitionComplete {
+ t.Fatalf("fixture: action %v, transition status %q", evt.Action, oi.TransitionedObject.Status)
+ }
+ if err := expireTransitionedObject(ctx, z, &oi, evt, lcEventSrc_Scanner); err != nil {
+ t.Errorf("expiry: %v", err)
+ }
+
+ var got []string
+ for _, set := range z.serverPools[0].sets {
+ err := set.listObjectsToDecommission(ctx, decomBucketInfo{Name: bucket}, func(entry metaCacheEntry) {
+ fivs, err := entry.fileInfoVersions(bucket)
+ if err != nil {
+ t.Error(err)
+ return
+ }
+ for _, fi := range fivs.Versions {
+ if fi.Deleted {
+ got = append(got, "DM")
+ } else {
+ got = append(got, labels[migrationTestVersion(fi.ToObjectInfo(bucket, "obj", true))])
+ }
+ }
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ if !slices.Equal(got, tc.want) {
+ t.Errorf("versions after expiry: got %v, want %v", got, tc.want)
+ }
+ remote := bucket + "/v2"
+ if removed := slices.Contains(tier.removed, remote); removed == tc.keep {
+ t.Errorf("remote object of the expired version removed: %v, want %v", removed, !tc.keep)
+ }
+ if tc.keep {
+ body := migrationTestBody(t, z.GetObjectNInfo, bucket, "obj", migrationTestVersion(latest))
+ if body != "obj v2" {
+ t.Errorf("v2 reads %q after the expiry, want %q", body, "obj v2")
+ }
+ }
+ })
+ }
+}