fix(ilm): trace refused expiries with their error; harden the tests

The implementation review of this round's public fixes found one gap in
the code and four in the tests.

- A refused expiry was traced with its reason but without its error,
  so the trace did not say what refused it. It now carries the error, as
  a failed one does.
- The restore MRF test replaced the live MRF channel, which the heal
  routines of the test servers read: a race, and a consumer could take
  the entry under test. addPartialOp now offers an atomic hook that
  receives the operation in place of the queue. Production never sets
  it.
- Two fixtures restored the storage class with Update, which marks an
  uninitialized configuration initialized and so changes the default
  parity (-1) into an explicit 0 for later tests.
  setTestStandardParity restores the whole saved value under
  storageclass.ConfigLock.
- The v3 metric test built its own descriptor map, so dropping the
  counter from the registered ILM collector still passed. It gathers
  through the collector the handler registers.
- The audit test read one extra entry or waited 500 ms. It drains every
  entry of the bucket up to the removal audited last, and checks one
  failure audit, one removal audit and no refusal audit per object.

Each corrected test fails with the corresponding mutation.

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

cmd/bucket-lifecycle-incomplete.go

@@ -140,7 +140,7 @@ func reportILMExpiryNotDone(ctx context.Context, oi ObjectInfo, lcEvent lifecycl
 	}
 	globalILMIncomplete.add(reason)
 	if ilmExpiryRefused(reason) {
-		traceFn(ILMExpiry, map[string]string{"result": "refused", "reason": reason, "version-id": oi.VersionID}, nil)
+		traceFn(ILMExpiry, map[string]string{"result": "refused", "reason": reason, "version-id": oi.VersionID}, err)
 		return true
 	}
 	traceFn(ILMExpiry, map[string]string{"result": "failed", "reason": reason, "version-id": oi.VersionID}, err)

cmd/bucket-lifecycle-incomplete_test.go

@@ -35,6 +35,7 @@ import (
 	loghttp "github.com/minio/minio/internal/logger/target/http"
 	"github.com/minio/minio/internal/pubsub"
 	xnet "github.com/pgsty/silo-pkg/v3/net"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 func TestILMExpiryNotDoneClassification(t *testing.T) {
@@ -85,7 +86,12 @@ func ilmTestAudit(t *testing.T, bucket string) <-chan map[string]any {
 		var entry struct {
 			Error string         `json:"error"`
 			Tags  map[string]any `json:"tags"`
-			API   struct{ Name, Bucket string }
+			API   struct {
+				Name, Bucket string
+				Objects      []struct {
+					ObjectName string `json:"objectName"`
+				} `json:"objects"`
+			}
 		}
 		if err := json.NewDecoder(r.Body).Decode(&entry); err == nil && entry.API.Name == "ILMExpiry" && entry.API.Bucket == bucket {
 			tags := entry.Tags
@@ -93,6 +99,9 @@ func ilmTestAudit(t *testing.T, bucket string) <-chan map[string]any {
 				tags = map[string]any{}
 			}
 			tags["error"] = entry.Error
+			if len(entry.API.Objects) > 0 {
+				tags["object"] = entry.API.Objects[0].ObjectName
+			}
 			entries <- tags
 		}
 		w.WriteHeader(http.StatusOK)
@@ -159,6 +168,9 @@ func TestDeleteObjectVersionsReportsRefusalAndFailure(t *testing.T) {
 		case tr := <-traces:
 			if r := tr.Custom["result"]; r != "" {
 				results[tr.Custom["version-id"]] = r
+				if tr.Error == "" {
+					t.Errorf("%s expiry of %s traced without its error", r, tr.Custom["version-id"])
+				}
 			}
 		default:
 			drained = true
@@ -170,20 +182,40 @@ func TestDeleteObjectVersionsReportsRefusalAndFailure(t *testing.T) {
 	if results[toDel[0].VersionID] != "refused" || results[toDel[1].VersionID] != "failed" {
 		t.Errorf("traces %v, want refused and failed", results)
 	}
-	select {
+	// Drain every audit entry of the bucket: until the removal of ncv-c,
-	case tags := <-audits:
+	// audited after the refusal and the failure were reported, has arrived,
-		if tags["ilm-incomplete-reason"] != ilmIncompleteHistoryUnavailable || tags["error"] == "" {
+	// and then while the target empties its queue.
-			t.Errorf("audit of the failure: %v", tags)
+	var got []map[string]any
+	deadline := time.After(10 * time.Second)
+	for removal := false; !removal; {
+		select {
+		case tags := <-audits:
+			got = append(got, tags)
+			removal = removal || tags["object"] == "ncv-c"
+		case <-deadline:
+			t.Fatalf("audits %v: the removal of ncv-c was not audited", got)
 		}
-	case <-time.After(5 * time.Second):
-		t.Fatal("the failure was not audited")
 	}
-	select {
+	for idle := time.After(time.Second); idle != nil; {
-	case tags := <-audits:
+		select {
-		if _, failed := tags["ilm-incomplete-reason"]; failed {
+		case tags := <-audits:
-			t.Errorf("a second incomplete expiry was audited: %v", tags)
+			got = append(got, tags)
+		case <-idle:
+			idle = nil
 		}
-	case <-time.After(500 * time.Millisecond):
+	}
+	perObject := map[any][]map[string]any{}
+	for _, tags := range got {
+		perObject[tags["object"]] = append(perObject[tags["object"]], tags)
+	}
+	if a := perObject["ncv-a"]; len(a) != 0 {
+		t.Errorf("the refusal was audited: %v", a)
+	}
+	if b := perObject["ncv-b"]; len(b) != 1 || b[0]["ilm-incomplete-reason"] != ilmIncompleteHistoryUnavailable || b[0]["error"] == "" {
+		t.Errorf("audits of the failure: %v, want one with its reason and error", b)
+	}
+	if c := perObject["ncv-c"]; len(c) != 1 || c[0]["ilm-incomplete-reason"] != nil {
+		t.Errorf("audits of the removal: %v, want one", c)
 	}
 }
 
@@ -288,24 +320,31 @@ func TestILMExpiryIncompleteMetrics(t *testing.T) {
 	for _, r := range ilmIncompleteReasons {
 		globalILMIncomplete.add(r)
 	}
-	if md := ilmExpiryIncompleteTotalMD; md.Type != CounterMT || !slices.Equal(md.VariableLabels, []string{"reason"}) {
+	// v3, through the collector the metrics handler registers.
-		t.Fatalf("v3 descriptor %+v", md)
+	families, err := newMetricGroups(prometheus.NewRegistry()).mgGatherers[ilmCollectorPath].Gather()
-	}
+	if err != nil {
-	d := map[MetricName]MetricDescriptor{}
-	for _, md := range []MetricDescriptor{ilmExpiryPendingTasksMD, ilmTransitionActiveTasksMD, ilmTransitionPendingTasksMD, ilmTransitionMissedImmediateTasksMD, ilmVersionsScannedMD, ilmExpiryIncompleteTotalMD} {
-		d[md.Name] = md
-	}
-	m := newMetricValues(d)
-	if err := loadILMMetrics(t.Context(), m, nil); err != nil {
 		t.Fatal(err)
 	}
+	want := ilmCollectorPath.metricPrefix() + "_" + string(expiryIncompleteTotal)
 	reasons := map[string]bool{}
-	for _, v := range m.values[expiryIncompleteTotal] {
+	for _, family := range families {
-		reasons[v.Labels["reason"]] = true
+		if family.GetName() != want {
+			continue
+		}
+		for _, m := range family.Metric {
+			if m.Counter == nil || m.GetCounter().GetValue() <= 0 {
+				t.Errorf("v3 %s: %v, want a counter above zero", want, m)
+			}
+			for _, l := range m.Label {
+				if l.GetName() == "reason" {
+					reasons[l.GetValue()] = true
+				}
+			}
+		}
 	}
 	for _, r := range ilmIncompleteReasons {
 		if !reasons[r] {
-			t.Errorf("v3: reason %q not exported", r)
+			t.Errorf("v3: %s{reason=%q} not exported", want, r)
 		}
 	}
 	reasons = map[string]bool{}

cmd/global-heal-lifecycle_test.go

@@ -30,15 +30,29 @@ import (
 // is decided again under the object lock and can be refused, and a version
 // left unhealed would stay without the replaced drive for as long as it is.
 
+// setTestStandardParity sets the standard storage class parity for the test
+// and restores the whole previous configuration after it, also when it was
+// never initialized and so meant the default parity.
+func setTestStandardParity(t *testing.T, parity int) {
+	t.Helper()
+	storageclass.ConfigLock.RLock()
+	saved := globalStorageClass
+	storageclass.ConfigLock.RUnlock()
+	globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: parity}})
+	t.Cleanup(func() {
+		storageclass.ConfigLock.Lock()
+		globalStorageClass = saved
+		storageclass.ConfigLock.Unlock()
+	})
+}
+
 // startTestBackgroundHeal registers the background heal sequence and a heal
 // worker, which drive healing needs for its bucket heals.
 func startTestBackgroundHeal(t *testing.T, z *erasureServerPools) {
 	t.Helper()
 	// The healed drive is rebuilt from parity. Earlier tests can leave a
 	// storage class of parity 0 behind (newTestConfig on a single drive).
-	previous := globalStorageClass
+	setTestStandardParity(t, z.serverPools[0].setDriveCount/2)
-	globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: z.serverPools[0].setDriveCount / 2}})
-	t.Cleanup(func() { globalStorageClass.Update(previous) })
 	if globalBackgroundHealState == nil {
 		globalBackgroundHealState = newHealState(GlobalContext, false)
 	}

cmd/mrf.go

@@ -74,11 +74,20 @@ func newMRFState() mrfState {
 	}
 }
 
+// mrfPartialOpHook, when set, receives each partial operation in place of
+// the queue, so that a test observes what is queued without racing the heal
+// routines that consume it. Production never sets it.
+var mrfPartialOpHook atomic.Pointer[func(PartialOperation)]
+
 // Add a partial S3 operation (put/delete) when one or more disks are offline.
 func (m *mrfState) addPartialOp(op PartialOperation) {
 	if m == nil {
 		return
 	}
+	if f := mrfPartialOpHook.Load(); f != nil {
+		(*f)(op)
+		return
+	}
 
 	if atomic.LoadInt32(&m.closed) == 1 {
 		return

cmd/restore-two-phase_test.go

@@ -27,6 +27,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -862,9 +863,7 @@ func TestRestorePublicationBelowWriteQuorumOnWideSet(t *testing.T) {
 // the convergence path queues too. That heal repairs the null version on the
 // drive that missed the commit, also where the null version is noncurrent.
 func TestRestorePartialCommitQueuesNullVersionHeal(t *testing.T) {
-	saved := globalStorageClass
+	setTestStandardParity(t, 2)
-	globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 2}})
-	t.Cleanup(func() { globalStorageClass.Update(saved) })
 	r := newRestoreSafetyTest(t, 1, versioning.Suspended)
 	oi, body := nativeRestoreSeed(t, r, 0, true, false)
 	// A version written while versioning was enabled makes the null version
@@ -878,22 +877,26 @@ func TestRestorePartialCommitQueuesNullVersionHeal(t *testing.T) {
 	failing := &restoreTwoPhaseFaults{}
 	failing.rename.Store("fail")
 	disks := restoreTwoPhaseFaulty(t, r, nil, nil, nil, failing)
-	// The test owns the MRF queue while the commit runs: the heal routines of
+	// Observe the MRF entries in place of the queue, which the heal routines
-	// the test servers would otherwise take the entry first.
+	// of the test servers consume.
-	queue := make(chan PartialOperation, mrfOpsQueueSize)
+	var mu sync.Mutex
-	previous := globalMRFState.opCh
+	var queued []PartialOperation
-	globalMRFState.opCh = queue
+	hook := func(op PartialOperation) {
+		if op.Bucket == r.bucket && op.Object == r.object {
+			mu.Lock()
+			queued = append(queued, op)
+			mu.Unlock()
+		}
+	}
+	mrfPartialOpHook.Store(&hook)
+	t.Cleanup(func() { mrfPartialOpHook.Store(nil) })
 	err := r.z.RestoreTransitionedObject(t.Context(), r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID))
-	globalMRFState.opCh = previous
+	mrfPartialOpHook.Store(nil)
 	if err != nil {
 		t.Fatalf("restore with one drive failing its commit: %v", err)
 	}
-	var queued []PartialOperation
+	mu.Lock()
-	for len(queue) > 0 {
+	defer mu.Unlock()
-		if op := <-queue; op.Bucket == r.bucket && op.Object == r.object {
-			queued = append(queued, op)
-		}
-	}
 	if len(queued) != 1 || queued[0].VersionID != nullVersionID {
 		t.Fatalf("queued heals %+v, want one of version %q", queued, nullVersionID)
 	}