fix(batch): fail an expiration run whose listing fails at once

A Walk that returned an error before any result cancelled the run like
an ordinary cancellation: the job was saved as neither complete nor
failed. The listing failure is now recorded first, and the run is saved
as failed with its cursor where it was. Start still returns the error,
so the job is left to a later run, which lists again.

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

cmd/batch-authz-review6_test.go

@@ -20,7 +20,7 @@ import (
 	"github.com/minio/minio/internal/kms"
 )
 
-// Rotation cases found in review.
+// Rotation and listing cases found in review.
 
 // Rotation rewrites the copy GetObjectInfo returns, which can carry another
 // tag revision than the merged version of all pools: both are authorized.
@@ -65,6 +65,33 @@ func TestBatchReviewRotationAuthorizesWrittenCopy(t *testing.T) {
 	}
 }
 
+// batchWalkError fails every listing before it returns a result.
+type batchWalkError struct {
+	ObjectLayer
+	err error
+}
+
+func (w batchWalkError) Walk(context.Context, string, string, chan<- itemOrErr[ObjectInfo], WalkOptions) error {
+	return w.err
+}
+
+// A listing that fails before it returns anything fails the run, as a
+// listing error in the results does, and the job is left to a later run.
+func TestBatchReviewExpireWalkErrorFailsRun(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	rec, job := f.submit(t, globalActiveCred, f.expireYAML("data/"))
+	if job == nil {
+		t.Fatalf("submission status %d: %s", rec.Code, rec.Body.String())
+	}
+	listErr := errors.New("review6: listing failed")
+	if err := job.Expire.Start(f.ctx, batchWalkError{ObjectLayer: f.obj, err: listErr}, *job); !errors.Is(err, listErr) {
+		t.Fatalf("start returned %v, want %v", err, listErr)
+	}
+	if ri := f.jobInfo(t, job); !ri.Failed || ri.Complete {
+		t.Fatalf("failed listing: complete=%v failed=%v", ri.Complete, ri.Failed)
+	}
+}
+
 // Two pools hold different writes of the null version, and a newer version
 // of the key is in the second one. Rotation rewrites the copy it read and
 // authorized, in the first pool, not the copy the latest version of the key

cmd/batch-expire-purge-authz_test.go

@@ -473,3 +473,45 @@ func TestBatchExpirePurgeBudgetIncludesAuthorization(t *testing.T) {
 		}
 	}
 }
+
+// A listing that failed before any result fails the run without moving the
+// cursor or refusing the job; a later run completes it, in both modes.
+func TestBatchExpireWalkFailureResumes(t *testing.T) {
+	for _, purgeAll := range []bool{false, true} {
+		name := "retain-one"
+		if purgeAll {
+			name = "retain-zero"
+		}
+		t.Run(name, func(t *testing.T) {
+			f := newBatchAuthzFixture(t)
+			const key = "data/key"
+			older := f.putVersions(t, key)
+			definition := f.expireYAML("data/")
+			if purgeAll {
+				definition = f.purgeYAML("data/")
+			}
+			rec, job := f.submit(t, globalActiveCred, definition)
+			if job == nil {
+				t.Fatalf("submit: %d %s", rec.Code, rec.Body.String())
+			}
+			listErr := errors.New("review7: transient listing failure")
+			err := job.Expire.Start(f.ctx, batchWalkError{ObjectLayer: f.obj, err: listErr}, *job)
+			if !errors.Is(err, listErr) {
+				t.Fatalf("listing error: %v", err)
+			}
+			ri := f.jobInfo(t, job)
+			if !ri.Failed || ri.Complete || ri.FailureReason != "" || ri.Object != "" || !f.versionExists(t, key, older) {
+				t.Fatalf("listing failure changed progress or became terminal: failed=%v complete=%v reason=%q cursor=%q", ri.Failed, ri.Complete, ri.FailureReason, ri.Object)
+			}
+			mustIAM(t, job.Expire.Start(f.ctx, f.obj, *job))
+			ri = f.jobInfo(t, job)
+			wantRemoved := int64(1)
+			if purgeAll {
+				wantRemoved = 2
+			}
+			if ri.Failed || !ri.Complete || ri.FailureReason != "" || ri.Objects != wantRemoved || f.versionExists(t, key, older) {
+				t.Fatalf("resume did not complete: failed=%v complete=%v reason=%q removed=%d", ri.Failed, ri.Complete, ri.FailureReason, ri.Objects)
+			}
+		})
+	}
+}

cmd/batch-expire.go

@@ -673,6 +673,8 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 	// run, so the cursor cannot move past what was not listed.
 	walkCtx, stopWalk := context.WithCancel(ctx)
 	defer stopWalk()
+	// walkFailed: a listing failed before it returned any result.
+	var walkFailed atomic.Bool
 	results := make(chan itemOrErr[ObjectInfo], workerSize)
 	go func() {
 		// One ascending walk over disjoint prefixes, which the cursor
@@ -689,6 +691,7 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 			})
 			if err != nil {
 				if walkCtx.Err() == nil {
+					walkFailed.Store(true)
 					cancelCause(err)
 				}
 				xioutil.SafeClose(results)
@@ -881,6 +884,13 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 	refused := ri.refused()
 	cause := context.Cause(ctx)
 	if cause != nil && !refused {
+		if walkFailed.Load() {
+			// This run failed; a later run lists again from the cursor.
+			ri.mu.Lock()
+			ri.Complete = false
+			ri.Failed = true
+			ri.mu.Unlock()
+		}
 		saveCtx, cancel := context.WithTimeout(GlobalContext, 30*time.Second)
 		batchLogIf(saveCtx, ri.updateAfter(saveCtx, api, 0, job))
 		cancel()