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()