From adc824087b6e94dc7d980ed8de260d29ce387f55 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Thu, 5 Nov 2026 09:04:39 +0800 Subject: [PATCH] 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 Signed-off-by: Feng Ruohang --- cmd/batch-authz-review6_test.go | 29 ++++++++++++++++++- cmd/batch-expire-purge-authz_test.go | 42 ++++++++++++++++++++++++++++ cmd/batch-expire.go | 10 +++++++ 3 files changed, 80 insertions(+), 1 deletion(-) diff --git a/cmd/batch-authz-review6_test.go b/cmd/batch-authz-review6_test.go index 447183e92..63303b733 100644 --- a/cmd/batch-authz-review6_test.go +++ b/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 diff --git a/cmd/batch-expire-purge-authz_test.go b/cmd/batch-expire-purge-authz_test.go index 20f6815af..259f725d9 100644 --- a/cmd/batch-expire-purge-authz_test.go +++ b/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) + } + }) + } +} diff --git a/cmd/batch-expire.go b/cmd/batch-expire.go index 3ddf86450..77c444dfc 100644 --- a/cmd/batch-expire.go +++ b/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()