fix(ilm): keep tier data when expiry only adds a delete marker

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

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 = `<LifecycleConfiguration><Rule><ID>exp</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter>` +
+		`<Expiration><Date>2020-01-01T00:00:00Z</Date></Expiration></Rule></LifecycleConfiguration>`
+
+	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("<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>")
+			}
+			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")
+				}
+			}
+		})
+	}
+}