minio/cmd/erasure-server-pool-tier-ownership_test.go

// Copyright (c) 2026 PGSTY
// SPDX-License-Identifier: AGPL-3.0-or-later

package cmd

import (
	"context"
	"errors"
	"slices"
	"testing"

	"github.com/minio/minio/internal/bucket/lifecycle"
	objectlock "github.com/minio/minio/internal/bucket/object/lock"
	xhttp "github.com/minio/minio/internal/http"
)

func rewriteTierTestMeta(t *testing.T, disk StorageAPI, bucket, key string, change func(*xlMetaV2)) {
	t.Helper()
	raw, err := disk.ReadAll(t.Context(), bucket, pathJoin(key, xlStorageFormatFile))
	if err != nil {
		t.Fatal(err)
	}
	var xl xlMetaV2
	if err = xl.LoadOrConvert(raw); err != nil {
		t.Fatal(err)
	}
	change(&xl)
	raw, err = xl.AppendTo(nil)
	if err != nil {
		t.Fatal(err)
	}
	if err = disk.WriteAll(t.Context(), bucket, pathJoin(key, xlStorageFormatFile), raw); err != nil {
		t.Fatal(err)
	}
}

func TestTierCleanupRequiresDurableOwnership(t *testing.T) {
	for _, kind := range []string{"absent", "older-live", "minority-live", "unknown", "remote-failure", "same-id-other-tuple"} {
		t.Run(kind, func(t *testing.T) {
			f := newFreeVersionTest(t, "")
			const key = "shared"
			oi := f.tiered(key, "remote", daysAgo(4))
			tuple := tierRemoteTuple{Tier: "WARM", Object: f.remote(key, "remote")}
			live := slices.Clone(f.xlMeta(0, key))
			f.deleteVersion(oi)
			disks := f.z.serverPools[0].getHashedSet(key).getDisks()
			other := f.z.serverPools[1].getHashedSet(key).getDisks()
			switch kind {
			case "absent":
				// A stale task with a different remote identity has no owning free record.
				tuple.Object += "-stale"
				f.tier.objects[tuple.Object] = []byte("must survive")
			case "older-live", "minority-live":
				for i, disk := range other {
					if kind == "minority-live" && i > 0 {
						break
					}
					if err := disk.WriteAll(t.Context(), f.bucket, pathJoin(key, xlStorageFormatFile), live); err != nil {
						t.Fatal(err)
					}
				}
				// Hide the live remote reference beneath a newer hot version.
				putConsistencyObject(t, f.z, f.bucket, key, 1, "hot", ObjectOptions{Versioned: true})
			case "unknown":
				if err := other[0].WriteAll(t.Context(), f.bucket, pathJoin(key, xlStorageFormatFile), []byte("corrupt")); err != nil {
					t.Fatal(err)
				}
			case "remote-failure":
				f.tier.removeErr = errors.New("remote unavailable")
			case "same-id-other-tuple":
				rewriteTierTestMeta(t, disks[len(disks)-1], f.bucket, key, func(xl *xlMetaV2) {
					ver, err := xl.getIdx(0)
					if err != nil {
						t.Fatal(err)
					}
					ver.DeleteMarker.MetaSys[metaTierObjName] = []byte("another-remote")
					if err := xl.setIdx(0, *ver); err != nil {
						t.Fatal(err)
					}
				})
				f.tier.objects["another-remote"] = []byte("other")
			}
			err := f.z.deleteTierTuple(t.Context(), f.bucket, key, tuple)
			switch kind {
			case "older-live", "minority-live":
				if !errors.Is(err, errTierStillReferenced) {
					t.Fatalf("cleanup: %v", err)
				}
			case "unknown":
				if !errors.Is(err, errTierOwnershipUnavailable) {
					t.Fatalf("cleanup: %v", err)
				}
			case "remote-failure":
				if !errors.Is(err, f.tier.removeErr) {
					t.Fatalf("cleanup: %v", err)
				}
			default:
				if err != nil {
					t.Fatal(err)
				}
			}
			if kind != "same-id-other-tuple" {
				if len(f.tier.removed) != 0 {
					t.Fatalf("deleted without sole durable ownership: %v", f.tier.removed)
				}
				if got := f.stored(0, key); len(got) != 1 || got[0] != "free" {
					t.Fatalf("lost retry record: %v", got)
				}
			} else {
				raw, err := disks[len(disks)-1].ReadAll(t.Context(), f.bucket, pathJoin(key, xlStorageFormatFile))
				if err != nil {
					t.Fatal(err)
				}
				versions, err := getFileInfoVersions(raw, f.bucket, key, true)
				if err != nil {
					t.Fatal(err)
				}
				if len(versions.Versions) != 1 || versions.Versions[0].TransitionedObjName != "another-remote" {
					t.Fatalf("removed different tuple: %+v", versions)
				}
				if len(f.tier.removed) != 1 || f.tier.removed[0] != tuple.Object {
					t.Fatalf("remote deletions: %v", f.tier.removed)
				}
			}
			if kind == "remote-failure" {
				f.tier.removeErr = nil
				if err := f.z.deleteTierTuple(t.Context(), f.bucket, key, tuple); err != nil {
					t.Fatal(err)
				}
				if got := f.stored(0, key); len(got) != 0 {
					t.Fatalf("retry left %v", got)
				}
				if err := f.z.deleteTierTuple(t.Context(), f.bucket, key, tuple); err != nil {
					t.Fatal(err)
				}
				if len(f.tier.removed) != 1 {
					t.Fatalf("stale retry deleted again: %v", f.tier.removed)
				}
			}
		})
	}
}

func TestTierSweeperPreservesOtherPoolOwner(t *testing.T) {
	f := newFreeVersionTest(t, "")
	const key = "overwrite"
	oi := f.tiered(key, "remote", daysAgo(4))
	queue := f.expiryQueue()
	sweeper := newObjSweeper(f.bucket, key).WithVersion(oi.VersionID).WithVersioning(true, false)
	sweeper.SetTransitionState(TransitionedObject{Status: lifecycle.TransitionComplete, Tier: "WARM", Name: f.remote(key, "remote")})
	// Overwrite in the other pool while the draining pool retains its owner.
	putConsistencyObject(t, f.z, f.bucket, key, 1, "replacement", ObjectOptions{Versioned: true})
	sweeper.Sweep()
	close(queue)
	globalExpiryState.Worker(queue)
	if len(f.tier.removed) != 0 {
		t.Fatalf("sweeper deleted a live remote: %v", f.tier.removed)
	}
	if got := migrationTestBody(t, f.z.GetObjectNInfo, f.bucket, key, oi.VersionID); got != key+" remote" {
		t.Fatalf("old version body: %q", got)
	}
}

func TestDecommissionMovesTieredDeleteAll(t *testing.T) {
	for _, kind := range []string{"complete", "pending", "minority-other-pool"} {
		t.Run(kind, func(t *testing.T) {
			f := newFreeVersionTest(t, purgeRules)
			const key = "ea/tiered"
			oi := f.tiered(key, "v1", daysAgo(4))
			if kind == "pending" {
				for _, disk := range f.z.serverPools[0].getHashedSet(key).getDisks() {
					rewriteTierTestMeta(t, disk, f.bucket, key, func(xl *xlMetaV2) {
						i, ver, err := xl.findVersionStr(oi.VersionID)
						if err != nil {
							t.Fatal(err)
						}
						ver.ObjectV2.MetaSys[metaTierStatus] = []byte(lifecycle.TransitionPending)
						if err := xl.setIdx(i, *ver); err != nil {
							t.Fatal(err)
						}
					})
				}
			}
			if kind == "minority-other-pool" {
				raw := slices.Clone(f.xlMeta(0, key))
				// A newer hot copy wins the logical merge, while one other-pool disk
				// still holds the older tier revision with the same version ID.
				putConsistencyObject(t, f.z, f.bucket, key, 0, "hot-v1", ObjectOptions{Versioned: true, VersionID: oi.VersionID, MTime: daysAgo(3)})
				disk := f.z.serverPools[1].getHashedSet(key).getDisks()[0]
				if err := disk.WriteAll(t.Context(), f.bucket, pathJoin(key, xlStorageFormatFile), raw); err != nil {
					t.Fatal(err)
				}
			}
			f.put(key, "v2", daysAgo(2))
			// Verify the precheck uses per-drive evidence, including minority data.
			var history migrationHistory
			latest, err := f.z.GetObjectInfo(t.Context(), f.bucket, key, ObjectOptions{})
			if err != nil {
				t.Fatal(err)
			}
			if f.z.migrationExpiryApplies(t.Context(), latest, lifecycle.Event{Action: lifecycle.DeleteAllVersionsAction}, &history) {
				t.Fatal("tiered purge selected instead of migration")
			}
			if kind == "pending" {
				return
			} // Pending data is tested at the precheck; a real transition owns its data move.
			if state := f.decommission(); !state.Complete || state.Failed {
				t.Fatalf("decommission: %+v", state)
			}
			if got := f.stored(0, key); len(got) != 0 {
				t.Fatalf("source retained %v", got)
			}
			if got := f.stored(1, key); len(got) < 2 || !slices.Equal(got[:2], []string{"v2", "v1"}) {
				t.Fatalf("destination versions: %v", got)
			}
			if slices.Contains(f.tier.removed, f.remote(key, "v1")) {
				t.Fatal("deleted live tier object during migration")
			}
		})
	}
}

// expireRestoredFaultDisk fails the removal of the restored copy of one key,
// as a drive that refuses that write does. Every other operation passes.
type expireRestoredFaultDisk struct {
	StorageAPI
	bucket, object string
}

func (d expireRestoredFaultDisk) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) error {
	if volume == d.bucket && path == d.object && fi.ExpireRestored {
		return errDiskFull
	}
	return d.StorageAPI.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts)
}

// restoredPastDeadline turns the version oi of key in pool 0 into a tiered
// version with a restored copy whose restore deadline has passed: its data
// stays on the drives, its metadata records the completed transition and the
// deadline. Past the deadline the version reads as remote. It returns the
// name of the remote object.
func (f *freeVersionTest) restoredPastDeadline(key string, oi ObjectInfo) string {
	f.t.Helper()
	ctx := f.t.Context()
	label := f.labels[oi.VersionID]
	remote := f.remote(key, label)
	infos, errs := readAllFileInfo(ctx, f.z.serverPools[0].getHashedSet(key).getDisks(), "", f.bucket, key, oi.VersionID, false, false)
	if errs[0] != nil {
		f.t.Fatal(errs[0])
	}
	fi := infos[0]
	fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", remote
	fi.Metadata[xhttp.AmzRestore] = completedRestoreObj(daysAgo(1)).String()
	f.tier.objects[remote] = []byte(key + " " + label)
	if err := f.z.serverPools[0].DecomTieredObject(ctx, f.bucket, key, fi, ObjectOptions{}); err != nil {
		f.t.Fatal(err)
	}
	cur, err := f.z.GetObjectInfo(ctx, f.bucket, key, ObjectOptions{VersionID: oi.VersionID})
	if err != nil || cur.RestoreExpires.IsZero() || !UTCNow().After(cur.RestoreExpires) {
		f.t.Fatalf("fixture: restore expires %v, %v", cur.RestoreExpires, err)
	}
	lc, err := globalLifecycleSys.Get(f.bucket)
	if err != nil {
		f.t.Fatal(err)
	}
	if evt := evalActionFromLifecycle(ctx, *lc, objectlock.Retention{}, nil, cur); !evt.Action.DeleteRestored() {
		f.t.Fatalf("fixture: lifecycle action %v, want the restored copy to expire", evt.Action)
	}
	return remote
}

// tieredVersion fails the test unless pool holds the version oi of key with
// its tier metadata of remote and its restore deadline, and returns it.
func (f *freeVersionTest) tieredVersion(pool int, key, remote string, oi ObjectInfo) ObjectInfo {
	f.t.Helper()
	got, err := f.z.serverPools[pool].GetObjectInfo(f.t.Context(), f.bucket, key, ObjectOptions{VersionID: oi.VersionID})
	if err != nil {
		f.t.Fatalf("pool %d: %v", pool, err)
	}
	if tr := got.TransitionedObject; tr.Status != lifecycle.TransitionComplete || tr.Tier != "WARM" || tr.Name != remote {
		f.t.Errorf("pool %d: the version lost its tier metadata: %+v", pool, tr)
	}
	if got.RestoreExpires.IsZero() || !UTCNow().After(got.RestoreExpires) {
		f.t.Errorf("pool %d: the version lost its restore deadline: %v", pool, got.RestoreExpires)
	}
	if slices.Contains(f.tier.removed, remote) {
		f.t.Errorf("remote object removed: %v", f.tier.removed)
	}
	return got
}

// expireRestoredCopy removes the restored copy of oi, as the scanner of the
// pool that holds the version does once its deadline has passed, and fails
// the test unless the version then reads as remote without a deadline and
// the remote object stays.
func (f *freeVersionTest) expireRestoredCopy(pool int, key, remote string, oi ObjectInfo) {
	f.t.Helper()
	ctx := f.t.Context()
	lc, err := globalLifecycleSys.Get(f.bucket)
	if err != nil {
		f.t.Fatal(err)
	}
	evt := evalActionFromLifecycle(ctx, *lc, objectlock.Retention{}, nil, oi)
	if !evt.Action.DeleteRestored() {
		f.t.Fatalf("lifecycle action in pool %d: %v", pool, evt.Action)
	}
	if !applyExpiryOnTransitionedObject(ctx, f.z, oi, evt, lcEventSrc_Scanner) {
		f.t.Fatalf("the restored copy did not expire in pool %d", pool)
	}
	after, err := f.z.serverPools[pool].GetObjectInfo(ctx, f.bucket, key, ObjectOptions{VersionID: oi.VersionID})
	if err != nil || !after.IsRemote() || !after.RestoreExpires.IsZero() || after.TransitionedObject.Name != remote {
		f.t.Fatalf("version after the restored copy expired: remote=%v expires=%v %+v, %v", after.IsRemote(), after.RestoreExpires, after.TransitionedObject, err)
	}
	if slices.Contains(f.tier.removed, remote) {
		f.t.Errorf("remote object removed with the restored copy: %v", f.tier.removed)
	}
}

// A decommission removes a restored copy that is past its deadline before it
// moves the version. When the drives refuse that removal, the version moves
// like any other tiered version, with its tier metadata and its deadline: the
// remote object stays, the source copy retires with the key, and the scanner
// of the pool that holds the version removes the copy by that deadline.
// Skipping the version instead leaves the key behind with a
// PreConditionFailed and fails the decommission on every retry.
func TestDecommissionMovesRestoredCopyPastDeadline(t *testing.T) {
	f := newFreeVersionTest(t, freeVersionNoncurrent)
	const key = "obj-restored"
	oi := f.put(key, "v1", daysAgo(4))
	remote := f.restoredPastDeadline(key, oi)

	set := f.z.serverPools[0].getHashedSet(key)
	getDisks := set.getDisks
	faulty := slices.Clone(getDisks())
	for i := range faulty {
		faulty[i] = expireRestoredFaultDisk{StorageAPI: faulty[i], bucket: f.bucket, object: key}
	}
	set.getDisks = func() []StorageAPI { return faulty }
	t.Cleanup(func() { set.getDisks = getDisks })

	info := f.decommission()
	if !info.Complete || info.Failed || info.ItemsDecommissionFailed != 0 {
		t.Fatalf("decommission with a refused restored-copy removal: complete=%v failed=%v items failed=%d, pool 0 holds %v, pool 1 holds %v",
			info.Complete, info.Failed, info.ItemsDecommissionFailed, f.stored(0, key), f.stored(1, key))
	}
	if got := f.stored(0, key); len(got) != 0 {
		t.Errorf("pool 0 still holds %v", got)
	}
	if got := f.stored(1, key); !slices.Equal(got, []string{"v1"}) {
		t.Fatalf("pool 1 holds %v, want [v1]", got)
	}
	moved := f.tieredVersion(1, key, remote, oi)
	t.Logf("the moved version reads as remote: %v", moved.IsRemote())
	f.expireRestoredCopy(1, key, remote, moved)
}

// Control: a rebalance moves no tiered version, and a restored copy past its
// deadline reads as remote. The walk leaves the version in the source pool
// without counting it as deleted, so the key is not retired, and the scanner
// of that pool removes the copy by its deadline.
func TestRebalanceLeavesRestoredCopyPastDeadline(t *testing.T) {
	f := newFreeVersionTest(t, freeVersionNoncurrent)
	const key = "obj-restored"
	oi := f.put(key, "v1", daysAgo(4))
	remote := f.restoredPastDeadline(key, oi)

	if failed := migrationTestWalk(t, f.z, f.bucket, 0, "rebalance"); failed != 0 {
		t.Fatalf("%d versions failed to move", failed)
	}
	if got := f.stored(0, key); !slices.Equal(got, []string{"v1"}) {
		t.Fatalf("pool 0 holds %v, want [v1]", got)
	}
	if got := f.stored(1, key); len(got) != 0 {
		t.Errorf("pool 1 holds %v", got)
	}
	left := f.tieredVersion(0, key, remote, oi)
	f.expireRestoredCopy(0, key, remote, left)
}

// tierReadFaultDisk fails the raw metadata read of one key with err, as a
// drive that is not reachable or faulty does.
type tierReadFaultDisk struct {
	StorageAPI
	bucket, object string
	err            error
}

func (d tierReadFaultDisk) ReadXL(ctx context.Context, volume, path string, readData bool) (RawFileInfo, error) {
	if volume == d.bucket && path == d.object {
		return RawFileInfo{}, d.err
	}
	return d.StorageAPI.ReadXL(ctx, volume, path, readData)
}

// freeRecord returns the free version of key in pool and the remote tuple
// it names.
func (f *freeVersionTest) freeRecord(pool int, key string) (FileInfo, tierRemoteTuple) {
	f.t.Helper()
	fivs, err := getFileInfoVersions(f.xlMeta(pool, key), f.bucket, key, false)
	if err != nil {
		f.t.Fatal(err)
	}
	if len(fivs.FreeVersions) != 1 {
		f.t.Fatalf("pool %d holds %d free versions of %s, want 1", pool, len(fivs.FreeVersions), key)
	}
	fv := fivs.FreeVersions[0]
	return fv, tierRemoteTuple{Tier: fv.TransitionTier, Object: fv.TransitionedObjName, VersionID: fv.TransitionVersionID}
}

// A drive that cannot be read, in any pool, may hold a live reference to the
// remote object: its ownership is unknown. The cleanup of the expiry worker
// and of a decommission is refused with a read quorum error, the free record
// and the remote object stay, and both go once the drive reads again.
func TestTierOwnershipReadFailsClosed(t *testing.T) {
	for _, tc := range []struct {
		name string
		err  error // nil: the drive is offline
	}{
		{"offline", nil},
		{"disk-not-found", errDiskNotFound},
		{"faulty-disk", errFaultyDisk},
		{"unformatted-disk", errUnformattedDisk},
	} {
		t.Run(tc.name, func(t *testing.T) {
			f := newFreeVersionTest(t, "")
			ctx := t.Context()
			const key = "obj-02" // only a free version, in pool 0
			remote := f.remote(key, "v1")
			fv, tuple := f.freeRecord(0, key)
			if tuple.Object != remote {
				t.Fatalf("free record names %+v", tuple)
			}
			// The drive is in the pool that holds nothing of the key.
			set := f.z.serverPools[1].getHashedSet(key)
			getDisks := set.getDisks
			faulty := slices.Clone(getDisks())
			if tc.err == nil {
				faulty[3] = nil
			} else {
				faulty[3] = tierReadFaultDisk{StorageAPI: faulty[3], bucket: f.bucket, object: key, err: tc.err}
			}
			set.getDisks = func() []StorageAPI { return faulty }
			restore := func() { set.getDisks = getDisks }
			defer restore()

			err := f.z.deleteTierTuple(ctx, f.bucket, key, tuple)
			var rq InsufficientReadQuorum
			if !errors.Is(err, errTierOwnershipUnavailable) || !errors.As(err, &rq) {
				t.Fatalf("cleanup with an unreadable drive: %v", err)
			}
			if err := f.z.removeFreeVersions(ctx, f.bucket, key, 0, []FileInfo{fv}); !errors.Is(err, errTierOwnershipUnavailable) {
				t.Fatalf("decommission cleanup with an unreadable drive: %v", err)
			}
			if len(f.tier.removed) != 0 {
				t.Fatalf("remote object removed with a drive unread: %v", f.tier.removed)
			}
			if got := f.stored(0, key); !slices.Equal(got, []string{"free"}) {
				t.Fatalf("pool 0 holds %v, want the free record", got)
			}

			restore()
			if err := f.z.deleteTierTuple(ctx, f.bucket, key, tuple); err != nil {
				t.Fatal(err)
			}
			if !slices.Equal(f.tier.removed, []string{remote}) {
				t.Fatalf("remote objects removed once the drive reads: %v", f.tier.removed)
			}
			if got := f.stored(0, key); len(got) != 0 {
				t.Fatalf("pool 0 still holds %v", got)
			}
		})
	}
}

// freeRecordDelete returns the FileInfo with which the tier cleanup deletes
// the free record id of the remote tuple.
func freeRecordDelete(id string, tuple tierRemoteTuple) FileInfo {
	fi := FileInfo{VersionID: id, TransitionTier: tuple.Tier, TransitionedObjName: tuple.Object, TransitionVersionID: tuple.VersionID}
	fi.SetTierFreeVersion()
	return fi
}

// A free record is deleted only with the remote tuple it names. The storage
// layer checks the record under the version ID: a cleanup that names another
// tier, remote object or remote version, as a stale task or a record rewritten
// on one drive would, finds no such version and leaves the record.
func TestFreeRecordDeleteNeedsMatchingTuple(t *testing.T) {
	const id = "00000000-0000-0000-0000-0000000000f1"
	tuple := tierRemoteTuple{Tier: "WARM", Object: "remote/obj/v1", VersionID: "rv1"}
	record := func() *xlMetaV2 {
		t.Helper()
		var xl xlMetaV2
		fi := FileInfo{
			Volume: "bucket", Name: "obj", VersionID: mustGetUUID(), ModTime: UTCNow(), DataDir: mustGetUUID(),
			Erasure: ErasureInfo{Algorithm: ReedSolomon.String(), DataBlocks: 4, ParityBlocks: 2, BlockSize: 10000, Index: 1, Distribution: []int{1, 2, 3, 4, 5, 6, 7, 8}},
		}
		if err := xl.AddVersion(fi); err != nil {
			t.Fatal(err)
		}
		fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName, fi.TransitionVersionID = lifecycle.TransitionComplete, tuple.Tier, tuple.Object, tuple.VersionID
		if _, err := xl.DeleteVersion(fi); err != nil {
			t.Fatal(err)
		}
		fi.TransitionStatus = ""
		fi.SetTierFreeVersionID(id)
		if _, err := xl.DeleteVersion(fi); err != nil {
			t.Fatal(err)
		}
		free, err := xl.listFreeVersions("bucket", "obj")
		if err != nil || len(free) != 1 || free[0].VersionID != id || free[0].TransitionedObjName != tuple.Object {
			t.Fatalf("fixture: free versions %+v, %v", free, err)
		}
		return &xl
	}
	others := map[string]func(*tierRemoteTuple){
		"tier":    func(tp *tierRemoteTuple) { tp.Tier = "COLD" },
		"object":  func(tp *tierRemoteTuple) { tp.Object = "remote/obj/v2" },
		"version": func(tp *tierRemoteTuple) { tp.VersionID = "rv2" },
	}
	for name, change := range others {
		t.Run(name, func(t *testing.T) {
			xl := record()
			other := tuple
			change(&other)
			if _, err := xl.DeleteVersion(freeRecordDelete(id, other)); !errors.Is(err, errFileVersionNotFound) {
				t.Fatalf("delete with another %s: %v, want %v", name, err, errFileVersionNotFound)
			}
			if free, err := xl.listFreeVersions("bucket", "obj"); err != nil || len(free) != 1 {
				t.Fatalf("free versions after a refused delete: %+v, %v", free, err)
			}
		})
	}
	t.Run("matching", func(t *testing.T) {
		xl := record()
		if _, err := xl.DeleteVersion(freeRecordDelete(id, tuple)); err != nil {
			t.Fatal(err)
		}
		if free, err := xl.listFreeVersions("bucket", "obj"); err != nil || len(free) != 0 {
			t.Fatalf("free versions after the delete: %+v, %v", free, err)
		}
	})
	t.Run("storage", func(t *testing.T) {
		ctx := t.Context()
		disk, _, err := newXLStorageTestSetup(t)
		if err != nil {
			t.Fatal(err)
		}
		if err := disk.MakeVol(ctx, "bucket"); err != nil {
			t.Fatal(err)
		}
		buf, err := record().AppendTo(nil)
		if err != nil {
			t.Fatal(err)
		}
		if err := disk.WriteAll(ctx, "bucket", pathJoin("obj", xlStorageFormatFile), buf); err != nil {
			t.Fatal(err)
		}
		other := tuple
		other.Object = "remote/obj/v2"
		if err := disk.DeleteVersion(ctx, "bucket", "obj", freeRecordDelete(id, other), false, DeleteOptions{}); !errors.Is(err, errFileVersionNotFound) {
			t.Fatalf("delete with another remote object: %v, want %v", err, errFileVersionNotFound)
		}
		if fi, err := disk.ReadVersion(ctx, "", "bucket", "obj", id, ReadOptions{InclFreeVersions: true}); err != nil || !fi.TierFreeVersion() || fi.TransitionedObjName != tuple.Object {
			t.Fatalf("free record after a refused delete: %+v, %v", fi, err)
		}
		if err := disk.DeleteVersion(ctx, "bucket", "obj", freeRecordDelete(id, tuple), false, DeleteOptions{}); err != nil {
			t.Fatal(err)
		}
		if _, err := disk.ReadAll(ctx, "bucket", pathJoin("obj", xlStorageFormatFile)); !errors.Is(err, errFileNotFound) {
			t.Fatalf("metadata after the last record is deleted: %v", err)
		}
	})
}