test(ilm): cover the history majority, approval check and tier ownership gates

Three safeguards of the lifecycle and tier cleanup paths had no test
that failed when they were removed:

- authoritativeHistory accepts a revision only on more than half of the
  drives. TestPoolsLifecycleHalfDriveMarker holds a delete marker on
  exactly half of sixteen drives, which is the read quorum of a marker,
  and checks that the history is refused, that the expiry of the
  version under it and of the key are refused and every drive keeps what
  it has, and that one more drive decides. With the guard removed the
  strict merge at the same majority still refuses the history; with the
  merge relaxed as well the test fails, so it pins the refusal, not one
  of its two layers.
- deleteAllApproved refuses a purge whose placement read differs from
  the versions it approved. deleteAllBeforePlacement is a test-only
  injection point between the two reads, nil in production.
  TestPurgeAllVersionsRefusesChangedApproval writes a new version and
  rewrites the latest one there, as a writer that bypasses the object
  lock would, and checks the write quorum refusal, the
  history_unavailable count, that nothing is deleted and that the next
  purge removes the key. With the check disabled the purge deletes the
  versions it never approved.
- readTierHistory refuses the ownership of a remote object while a
  drive of any pool cannot be read, and xlMetaV2.DeleteVersion deletes
  a free record only with the remote tuple it names.
  TestTierOwnershipReadFailsClosed fails one drive of the pool that
  holds nothing of the key as offline, not found, faulty and
  unformatted, and checks that the expiry worker's and the
  decommission's cleanup are refused with a read quorum error and
  succeed once the drive reads again.
  TestFreeRecordDeleteNeedsMatchingTuple deletes a free record with
  another tier, remote object and remote version, in memory and on an
  xlStorage drive, and checks the record stays until the tuple matches.
  With either check removed the remote object or the record is deleted.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/erasure-server-pool-delete-all.go

@@ -91,6 +91,12 @@ var deleteAllBudget = struct {
 // quorum is repeated on the drives that failed it.
 const deleteAllRetries = 2
 
+// deleteAllBeforePlacement, when a test sets it, runs after a purge approved
+// the versions of a key and before it reads their placement, with the bucket
+// and the encoded key: a write that reaches the key then, as one that
+// bypasses the object lock would, must refuse the purge.
+var deleteAllBeforePlacement func(bucket, object string)
+
 // deleteAllMatch identifies the latest version that a batch expiry matched.
 type deleteAllMatch struct {
 	VersionID string
@@ -295,6 +301,9 @@ func (z *erasureServerPools) deleteAllVersions(ctx context.Context, bucket, obje
 	}
 	// The placement of the approved versions. This read may only confirm
 	// them: whatever differs from the approval refuses the key.
+	if f := deleteAllBeforePlacement; f != nil {
+		f(bucket, object)
+	}
 	history, err := z.deleteAllHistory(ctx, bucket, object)
 	if err != nil {
 		return fail(err)

cmd/erasure-server-pool-delete-all_test.go

@@ -723,6 +723,73 @@ func TestPurgeAllVersionsRefusesStaleDecision(t *testing.T) {
 	}
 }
 
+// The purge reads the placement of the versions it approved and may only
+// confirm them. A write that reaches the key between the approval and that
+// read, as one that bypasses the object lock would, changes what the purge
+// is about to delete: a version it did not approve, or another write of an
+// approved one. The purge refuses the key with a write quorum error, which
+// counts as an unavailable history, and deletes nothing; the next purge,
+// which approves what it finds, removes the key.
+func TestPurgeAllVersionsRefusesChangedApproval(t *testing.T) {
+	for _, kind := range []string{"new-version", "latest-rewritten"} {
+		t.Run(kind, func(t *testing.T) {
+			z := purgeSinglePool(t, 16)
+			p := newPurgeTest(t, z, "purge-approval", MakeBucketOptions{VersioningEnabled: true}, purgeRules)
+			ctx := t.Context()
+			const key = "ea/approval"
+			p.version(0, key, "v1", 5)
+			latest := p.version(0, key, "v2", 4)
+			ops := p.scanQueue(0, key)
+
+			// Each purge finds the history changed since its approval.
+			var writes int
+			deleteAllBeforePlacement = func(bucket, object string) {
+				if bucket != p.bucket || object != key {
+					return
+				}
+				writes++
+				label := fmt.Sprintf("v%d", 2+writes)
+				opts := ObjectOptions{Versioned: true, NoLock: true}
+				mtime := daysAgo(3)
+				if kind == "latest-rewritten" {
+					label = fmt.Sprintf("v2-rewrite-%d", writes)
+					opts.VersionID, mtime = latest.VersionID, latest.ModTime
+				}
+				p.put(0, key, label, mtime, opts)
+			}
+			t.Cleanup(func() { deleteAllBeforePlacement = nil })
+
+			_, err := z.DeleteObject(ctx, p.bucket, key, ObjectOptions{DeleteAllVersions: true, Expiration: ExpirationOptions{Expire: true}})
+			if !isErrWriteQuorum(err) {
+				t.Fatalf("purge of a changed history: %v, want a write quorum error", err)
+			}
+			delta := incompleteDelta(func() { p.run(ops) })
+			if delta[ilmIncompleteHistoryUnavailable] != 1 {
+				t.Errorf("counted %v, want one %s", delta, ilmIncompleteHistoryUnavailable)
+			}
+			if writes != 2 {
+				t.Fatalf("the history changed %d times, want 2", writes)
+			}
+			want, read := []string{"hot", "hot", "hot", "hot"}, "v4"
+			if kind == "latest-rewritten" {
+				want, read = []string{"hot", "hot"}, "v2-rewrite-2"
+			}
+			if got := p.held(0, key); !slices.Equal(got, want) {
+				t.Fatalf("the refused purges left %v, want %v", got, want)
+			}
+			if got := p.get(key); got != read {
+				t.Errorf("the key reads %s, want %s", got, read)
+			}
+
+			deleteAllBeforePlacement = nil
+			p.expire(0, key)
+			if got := p.held(0, key); got != nil {
+				t.Errorf("the purge of the unchanged history left %v", got)
+			}
+		})
+	}
+}
+
 // Control: on two pools the stale decision was already refused.
 func TestPurgeAllVersionsStaleDecisionTwoPoolsControl(t *testing.T) {
 	z, _ := consistencyPools(t)

cmd/erasure-server-pool-lifecycle-authority_test.go

@@ -382,6 +382,99 @@ func TestPoolsLifecycleRevisionIdentity(t *testing.T) {
 	})
 }
 
+// authorityMarker encodes a delete marker of version id at modTime.
+func authorityMarker(t *testing.T, id [16]byte, modTime int64) xlMetaV2ShallowVersion {
+	t.Helper()
+	v := xlMetaV2Version{Type: DeleteType, DeleteMarker: &xlMetaV2DeleteMarker{VersionID: id, ModTime: modTime, MetaSys: map[string][]byte{}}}
+	meta, err := v.MarshalMsg(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return xlMetaV2ShallowVersion{header: v.header(), meta: meta}
+}
+
+// A revision counts only on more than half of the drives, whatever read
+// quorum its metadata has. A delete marker that a DELETE left on exactly
+// half of the drives, which is the read quorum of a marker, is not part of
+// the history: the drives cannot tell whether the version it hides is the
+// latest one, and no expiry is decided on them. One more drive decides.
+func TestPoolsLifecycleHalfDriveMarker(t *testing.T) {
+	markerID, objectID := [16]byte{2}, [16]byte{1}
+	history := func(drives int) [][]xlMetaV2ShallowVersion {
+		metas := make([][]xlMetaV2ShallowVersion, 16)
+		for i := range metas {
+			if i < drives {
+				metas[i] = append(metas[i], authorityMarker(t, markerID, 2))
+			}
+			metas[i] = append(metas[i], authorityRevision(t, objectID, 1, i+1, ""))
+		}
+		return metas
+	}
+	metaVs := slices.Repeat([]uint8{xlMetaVersion}, 16)
+	readQuorum := func(ver xlMetaV2ShallowVersion, _ uint8) int {
+		if ver.header.Type == DeleteType {
+			return 8
+		}
+		return 12
+	}
+	t.Run("half", func(t *testing.T) {
+		if merged, err := authoritativeHistory(t.Context(), history(8), metaVs, xlMetaVersion, 9, readQuorum); !errors.Is(err, errErasureReadQuorum) {
+			t.Fatalf("marker on 8 of 16 drives: %d versions, err=%v", len(merged), err)
+		}
+	})
+	t.Run("majority", func(t *testing.T) {
+		merged, err := authoritativeHistory(t.Context(), history(9), metaVs, xlMetaVersion, 9, readQuorum)
+		if err != nil || len(merged) != 2 || merged[0].header.VersionID != markerID || merged[1].header.VersionID != objectID {
+			t.Fatalf("marker on 9 of 16 drives: %d versions, err=%v", len(merged), err)
+		}
+	})
+
+	// On the drives: an expired version under a marker that reached half of
+	// them. The expiry of the version, which the marker hides on those
+	// drives, and of the key are refused, and every drive keeps what it has.
+	t.Run("drives", func(t *testing.T) {
+		authorityParity(t)
+		s := newSplitILM(t, versioning.Enabled, splitILMRules)
+		ctx := t.Context()
+		const key = "exp-half-marker"
+		s.write(key, splitAt("v1", 5, 0))
+		s.write(key, splitMarker("dm", 0.01, 0))
+		id, marker := s.ids[key+" v1"], s.ids[key+" dm"]
+		removeAuthorityCopies(t, s, 0, key, marker, authorityDrives(8, 16))
+
+		if _, _, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, key); !isErrReadQuorum(err) {
+			t.Errorf("history with a marker on half of the drives: %v", err)
+		}
+		if err := authorityExpireVersion(t, s, key, id); !isErrWriteQuorum(err) {
+			t.Errorf("expiry of the version under the marker: %v", err)
+		}
+		if _, err := s.z.DeleteObject(ctx, s.bucket, key, authorityDeleteAll); !isErrWriteQuorum(err) {
+			t.Errorf("expiry of all versions: %v", err)
+		}
+		authorityPresent(t, s, 0, key, id)
+		for i, disk := range s.z.serverPools[0].getHashedSet(key).getDisks() {
+			_, err := disk.ReadVersion(ctx, "", s.bucket, key, marker, ReadOptions{})
+			if i < 8 && err != nil {
+				t.Fatalf("drive %d lost the marker: %v", i, err)
+			}
+			if i >= 8 && !errors.Is(err, errFileVersionNotFound) {
+				t.Fatalf("drive %d: marker %v", i, err)
+			}
+		}
+
+		// Control: the marker on one more drive decides, and the version
+		// under it is noncurrent, which the rule of exp- does not expire.
+		const whole = "exp-majority-marker"
+		s.write(whole, splitAt("v1", 5, 0))
+		s.write(whole, splitMarker("dm", 0.01, 0))
+		removeAuthorityCopies(t, s, 0, whole, s.ids[whole+" dm"], authorityDrives(9, 16))
+		versions, events, err := s.z.lifecycleEventsAllPools(ctx, s.bucket, whole)
+		if err != nil || len(versions) != 2 || !versions[0].DeleteMarker || events[1].Action != lifecycle.NoneAction {
+			t.Fatalf("history with a marker on 9 drives: %d versions, events %v, err=%v", len(versions), events, err)
+		}
+	})
+}
+
 // authorityMovedHold serves a drive whose metadata of version id carries the
 // legal hold in the system map instead of the user map, under the header it
 // had: the evaluator no longer reads the hold there.

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

@@ -376,3 +376,198 @@ func TestRebalanceLeavesRestoredCopyPastDeadline(t *testing.T) {
 	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)
+		}
+	})
+}