fix(batch): pause transient authorization failures with owned retries

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/batch-authz-review6_test.go

@@ -75,9 +75,8 @@ func (w batchWalkError) Walk(context.Context, string, string, chan<- itemOrErr[O
 	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) {
+// A listing outage pauses the run, as a streamed listing error does.
+func TestBatchReviewExpireWalkErrorPausesRun(t *testing.T) {
 	f := newBatchAuthzFixture(t)
 	rec, job := f.submit(t, globalActiveCred, f.expireYAML("data/"))
 	if job == nil {
@@ -87,7 +86,7 @@ func TestBatchReviewExpireWalkErrorFailsRun(t *testing.T) {
 	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 {
+	if ri := f.jobInfo(t, job); ri.Failed || ri.Complete || ri.PausedReason == "" {
 		t.Fatalf("failed listing: complete=%v failed=%v", ri.Complete, ri.Failed)
 	}
 }
@@ -251,6 +250,26 @@ func TestBatchReviewRotationLeavesMovingSource(t *testing.T) {
 	if after.UserDefined[crypto.MetaSealedKeyS3] != beforeInfo.UserDefined[crypto.MetaSealedKeyS3] {
 		t.Fatal("rotation rewrote a draining pool")
 	}
+	// The job pauses and resumes in-process once the selected pool stops moving.
+	job := BatchJobRequest{ID: "rotation-moving-resume", Started: UTCNow(), Submitter: batchRootSubmitter(t), KeyRotate: &r}
+	batchTestJobPool(t.Context(), t, z)
+	if err := r.Start(t.Context(), z, job); !isBatchJobUnavailable(err) {
+		t.Fatalf("moving job did not pause: %v", err)
+	}
+	ri := &batchJobInfo{}
+	mustIAM(t, ri.load(t.Context(), z, job))
+	if ri.PausedReason == "" || ri.Failed || ri.Complete || ri.Object != "" || ri.ObjectsFailed != 0 {
+		t.Fatalf("moving job report: %+v", ri)
+	}
+	z.poolMetaMutex.Lock()
+	z.poolMeta.Pools[0].Decommission = nil
+	z.poolMetaMutex.Unlock()
+	mustIAM(t, r.Start(t.Context(), z, job))
+	mustIAM(t, ri.load(t.Context(), z, job))
+	if !ri.Complete || ri.Failed || ri.PausedReason != "" || ri.Objects != 1 {
+		t.Fatalf("resumed rotation: %+v", ri)
+	}
+
 }
 
 // Once a move copied a version, the rotation leaves it to a later run, so the

cmd/batch-authz.go

@@ -150,7 +150,7 @@ func newBatchJobSubmitter(cred auth.Credentials) (*batchJobSubmitter, error) {
 	}
 	u, ok, err := globalIAMSys.identityRecord(cred.AccessKey)
 	if err != nil {
-		return nil, err
+		return nil, batchUnavailable("IAM identity lookup", err)
 	}
 	// The record must be the one the request was authenticated with.
 	if !ok || u.Credentials.AccessKey != cred.AccessKey || u.Credentials.ParentUser != cred.ParentUser ||
@@ -174,6 +174,7 @@ func newBatchJobSubmitter(cred auth.Credentials) (*batchJobSubmitter, error) {
 
 // batchJobPrincipal is a submitter resolved against current IAM state.
 type batchJobPrincipal struct {
+	ctx        context.Context
 	typ        string
 	accessKey  string
 	parentUser string
@@ -194,6 +195,9 @@ func (s *batchJobSubmitter) principal(ctx context.Context) (batchJobPrincipal, e
 	if s.Version != batchJobSubmitterVersion {
 		return batchJobPrincipal{}, batchJobAuthzError{reason: fmt.Sprintf("unsupported batch job submitter version %d; submit the job again", s.Version)}
 	}
+	if !batchIAMReady() {
+		return batchJobPrincipal{}, batchUnavailable("IAM is not initialized", errServerNotInitialized)
+	}
 	root := globalActiveCred
 	if s.Type == batchSubmitterRoot {
 		if s.AccessKey != root.AccessKey || !hmac.Equal(s.Binding, batchSubmitterBinding(batchSubmitterRoot, root)) {
@@ -202,7 +206,7 @@ func (s *batchJobSubmitter) principal(ctx context.Context) (batchJobPrincipal, e
 		if !globalAPIConfig.permitRootAccess() {
 			return batchJobPrincipal{}, errBatchJobInvalid
 		}
-		return batchJobPrincipal{typ: batchSubmitterRoot, accessKey: root.AccessKey, owner: true}, nil
+		return batchJobPrincipal{ctx: ctx, typ: batchSubmitterRoot, accessKey: root.AccessKey, owner: true}, nil
 	}
 	switch s.Type {
 	case batchSubmitterUser, batchSubmitterTemporary, batchSubmitterService:
@@ -214,7 +218,7 @@ func (s *batchJobSubmitter) principal(ctx context.Context) (batchJobPrincipal, e
 	}
 	u, ok, err := globalIAMSys.CheckKey(ctx, s.AccessKey)
 	if err != nil {
-		return batchJobPrincipal{}, err
+		return batchJobPrincipal{}, batchUnavailable("IAM credential lookup", err)
 	}
 	if !ok {
 		// Deleted, disabled, expired or revoked.
@@ -231,6 +235,7 @@ func (s *batchJobSubmitter) principal(ctx context.Context) (batchJobPrincipal, e
 		return batchJobPrincipal{}, errBatchJobInvalid
 	}
 	p := batchJobPrincipal{
+		ctx:        ctx,
 		typ:        typ,
 		accessKey:  cred.AccessKey,
 		parentUser: cred.ParentUser,
@@ -243,6 +248,9 @@ func (s *batchJobSubmitter) principal(ctx context.Context) (batchJobPrincipal, e
 		// Verify the token as a request would, with the current signing keys.
 		jwtClaims, err := extractJWTClaims(u)
 		if err != nil {
+			if errors.Is(err, errTokenSigningKeyUnavailable) {
+				return batchJobPrincipal{}, batchUnavailable("token signing key", err)
+			}
 			return batchJobPrincipal{}, errBatchJobInvalid
 		}
 		p.claims = jwtClaims.Map()
@@ -444,24 +452,47 @@ func (p batchJobPrincipal) authorize(req batchAuthzRequest) error {
 		IsOwner:         p.owner,
 		Claims:          p.claims,
 	}
-	denied := batchJobAuthzError{reason: batchDeniedReason(req)}
-	plugin := newGlobalAuthZPluginFn()
-	policies, err := globalIAMSys.batchLocalPolicies(p, args, plugin == nil)
-	if err != nil {
-		return denied
+	return p.authorizeArgs(args, &req)
+}
+
+// authorizeArgs uses the same policy snapshot for request-context checks and
+// the decision. No boolean IAM evaluator re-reads and silently drops policies.
+func (p batchJobPrincipal) authorizeArgs(args policy.Args, req *batchAuthzRequest) error {
+	denied := batchJobAuthzError{reason: "the batch job submitter is not allowed " + string(args.Action)}
+	if req != nil {
+		denied.reason = batchDeniedReason(*req)
 	}
-	supplied := req.suppliedConditionKeys()
-	for _, pol := range policies {
-		for _, st := range pol.Statements {
-			if batchStatementNeedsRequest(st, args, supplied) {
-				return batchJobAuthzError{reason: denied.reason + ": a policy statement depends on request context that a batch job does not have"}
+	ctx := p.ctx
+	if ctx == nil {
+		ctx = GlobalContext
+	}
+	plugin := newGlobalAuthZPluginFn()
+	var cache *iamCache
+	if plugin == nil && !args.IsOwner {
+		cache = globalIAMSys.store.lock()
+		defer globalIAMSys.store.unlock()
+	}
+	policies, err := globalIAMSys.batchLocalPolicies(ctx, p, args, plugin == nil, cache)
+	if err != nil {
+		if errors.Is(err, errNoSuchPolicy) || errors.Is(err, errAuthentication) || errors.Is(err, errNoSuchUser) || errors.Is(err, errNoSuchGroup) {
+			return denied
+		}
+		return batchUnavailable("IAM policy lookup", err)
+	}
+	if req != nil {
+		supplied := req.suppliedConditionKeys()
+		for _, pol := range policies {
+			for _, st := range pol.Statements {
+				if batchStatementNeedsRequest(st, args, supplied) {
+					return batchJobAuthzError{reason: denied.reason + ": a policy statement depends on request context that a batch job does not have"}
+				}
 			}
 		}
 	}
 	if plugin != nil {
-		ok, err := plugin.IsAllowed(args)
+		ok, err := plugin.IsAllowedContext(ctx, args)
 		if err != nil {
-			authZLogIf(GlobalContext, err)
+			return batchUnavailable("authorization plugin", err)
 		}
 		if !ok {
 			return denied
@@ -472,9 +503,29 @@ func (p batchJobPrincipal) authorize(req batchAuthzRequest) error {
 		}
 		return nil
 	}
-	if !globalIAMSys.isAllowedLocally(args) {
+	if args.IsOwner {
+		return nil
+	}
+	if p.service {
+		if p.claims[parentClaim] != p.parentUser {
+			return denied
+		}
+		if _, ok := p.claims[iamPolicyClaimNameSA()].(string); !ok {
+			return denied
+		}
+	}
+	if has, allowed := p.sessionPolicyDecision(args); has && !allowed {
 		return denied
 	}
+	ownerDerived := (p.temp || p.service) && p.parentUser == globalActiveCred.AccessKey
+	if !ownerDerived {
+		if p.service {
+			args.AccountName = p.parentUser
+		}
+		if len(policies) == 0 || !policies[0].IsAllowed(args) {
+			return denied
+		}
+	}
 	return nil
 }
 
@@ -507,13 +558,23 @@ func (p batchJobPrincipal) sessionPolicyDecision(args policy.Args) (bool, bool)
 // batchLocalPolicies returns every local policy that can take part in the
 // decision on args: the mapped policies that IsAllowed would merge when
 // withMapped is set (no authorization plugin), and the session policy.
-func (sys *IAMSys) batchLocalPolicies(p batchJobPrincipal, args policy.Args, withMapped bool) ([]policy.Policy, error) {
+func (sys *IAMSys) batchLocalPolicies(ctx context.Context, p batchJobPrincipal, args policy.Args, withMapped bool, cache *iamCache) ([]policy.Policy, error) {
 	var out []policy.Policy
 	if withMapped && args.IsOwner {
 		// IsAllowed does not evaluate any policy for the owner.
 		return nil, nil
 	}
 	if withMapped {
+		// The caller holds the IAM writer lock through evaluation. Lazy mapping
+		// loads and selected documents therefore belong to one cache revision.
+		mappedPolicies := func(name string) ([]string, error) {
+			names, _, err := cache.policyDBGet(sys.store, name, false, false)
+			if err != nil {
+				return nil, err
+			}
+			groups, err := cache.policyDBGetGroups(sys.store, len(names) > 0, args.Groups...)
+			return append(names, groups...), err
+		}
 		var names []string
 		ownerDerived := (p.temp || p.service) && p.parentUser == globalActiveCred.AccessKey
 		switch {
@@ -526,7 +587,7 @@ func (sys *IAMSys) batchLocalPolicies(p batchJobPrincipal, args policy.Args, wit
 				}
 				names = append(names, newMappedPolicy(sys.rolesMap[a]).toSlice()...)
 			} else {
-				mapped, err := sys.PolicyDBGet(p.parentUser, args.Groups...)
+				mapped, err := mappedPolicies(p.parentUser)
 				if err != nil {
 					return nil, err
 				}
@@ -539,15 +600,31 @@ func (sys *IAMSys) batchLocalPolicies(p batchJobPrincipal, args policy.Args, wit
 				}
 			}
 		default:
-			mapped, err := sys.PolicyDBGet(p.accessKey, args.Groups...)
+			mapped, err := mappedPolicies(p.accessKey)
 			if err != nil {
 				return nil, err
 			}
 			names = append(names, mapped...)
 		}
-		if len(names) > 0 {
-			out = append(out, sys.GetCombinedPolicy(names...))
+		var selected []policy.Policy
+		for _, name := range names {
+			doc, ok := cache.iamPolicyDocsMap[name]
+			if !ok {
+				loaded := make(map[string]PolicyDoc)
+				loadCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+				err := sys.store.loadPolicyDoc(loadCtx, name, loaded)
+				cancel()
+				if err != nil {
+					return nil, err
+				}
+				doc, ok = loaded[name]
+				if !ok {
+					return nil, errNoSuchPolicy
+				}
+			}
+			selected = append(selected, doc.Policy)
 		}
+		out = append(out, policy.MergePolicies(selected...))
 	}
 	if p.temp || p.service {
 		if p.service && p.claims[iamPolicyClaimNameSA()] == inheritedPolicyType {
@@ -710,6 +787,7 @@ func (ri *batchJobInfo) refuse(err error) {
 	}
 	ri.Complete = false
 	ri.Failed = true
+	ri.PausedReason = ""
 }
 
 func (ri *batchJobInfo) refused() bool {
@@ -729,8 +807,19 @@ func (ri *batchJobInfo) startAuthorized(ctx context.Context, api ObjectLayer, jo
 			err = errBatchJobSections
 		}
 		if err == nil {
+			ri.mu.Lock()
+			ri.PausedReason = ""
+			ri.mu.Unlock()
+			// Persist the report version before acting. Older binaries must not
+			// resume an authorization-bearing job with server authority.
+			if err := ri.updateAfter(ctx, api, 0, job); err != nil {
+				return true, batchUnavailable("saving job progress", err)
+			}
 			return false, nil
 		}
+		if isBatchJobUnavailable(err) {
+			return true, ri.savePaused(ctx, api, job, err)
+		}
 		if !isBatchJobAuthzError(err) {
 			return true, err
 		}
@@ -738,5 +827,8 @@ func (ri *batchJobInfo) startAuthorized(ctx context.Context, api ObjectLayer, jo
 		batchLogIf(ctx, fmt.Errorf("batch job %s refused: %w", job.ID, err))
 	}
 	globalBatchJobsMetrics.save(job.ID, ri)
-	return true, ri.updateAfter(ctx, api, 0, job)
+	if err := ri.updateAfter(ctx, api, 0, job); err != nil {
+		return true, batchUnavailable("saving job refusal", err)
+	}
+	return true, nil
 }

cmd/batch-expire-purge-authz_test.go

@@ -481,8 +481,8 @@ 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.
+// A listing outage pauses 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"
@@ -507,7 +507,7 @@ func TestBatchExpireWalkFailureResumes(t *testing.T) {
 				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) {
+			if ri.Failed || ri.Complete || ri.PausedReason == "" || 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))

cmd/batch-expire.go

@@ -448,7 +448,7 @@ func (r *BatchJobExpire) expireVersions(ctx context.Context, api ObjectLayer, vc
 				ri.countRemoved(r.Bucket, 1, 0)
 			}
 			return nil
-		case isBatchJobAuthzError(err):
+		case isBatchJobStopped(err):
 			return err
 		case isErrObjectLocked(err):
 			// Left in place, not retried and not a failure.
@@ -515,7 +515,7 @@ func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer,
 		switch {
 		case err == nil:
 			return nil
-		case isBatchJobAuthzError(err):
+		case isBatchJobStopped(err):
 			return err
 		case errors.Is(err, errDeleteAllChanged):
 			// Changed since the walk matched it: not expired.
@@ -616,7 +616,7 @@ func batchObjsForDelete(ctx context.Context, refuse func(error), walkDone func()
 				}
 				if err != nil {
 					// The batch is not done; a later run does it again.
-					if isBatchJobAuthzError(err) {
+					if isBatchJobStopped(err) {
 						refuse(err)
 					}
 					globalBatchJobsMetrics.save(ri.JobID, ri)
@@ -639,7 +639,7 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 		StartTime: job.Started,
 	}
 	if err := ri.loadOrInit(ctx, api, job); err != nil {
-		return err
+		return batchUnavailable("loading job progress", err)
 	}
 	// A completed job is not run again, as replication and key rotation
 	// jobs are not: its submitter is not checked again either, so a
@@ -671,7 +671,11 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 	defer cancelCause(nil)
 	// refuse ends the job for its submitter: record why before stopping.
 	refuse := func(err error) {
-		ri.refuse(err)
+		if isBatchJobUnavailable(err) {
+			ri.pause(err)
+		} else {
+			ri.refuse(err)
+		}
 		cancelCause(err)
 	}
 
@@ -679,8 +683,6 @@ 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
@@ -697,8 +699,7 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 			})
 			if err != nil {
 				if walkCtx.Err() == nil {
-					walkFailed.Store(true)
-					cancelCause(err)
+					cancelCause(batchListingError(err))
 				}
 				xioutil.SafeClose(results)
 				return
@@ -786,6 +787,7 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 			}
 		}
 	}
+	var listingErr error
 	for {
 		select {
 		case result, ok := <-results:
@@ -795,6 +797,7 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 			}
 			if result.Err != nil {
 				failed = true
+				listingErr = batchListingError(result.Err)
 				batchLogIf(ctx, result.Err)
 				stopWalk()
 				done = true
@@ -889,11 +892,15 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 	<-saverDoneCh
 	refused := ri.refused()
 	cause := context.Cause(ctx)
+	if cause == nil {
+		cause = listingErr
+	}
 	if cause != nil && !refused {
-		if walkFailed.Load() {
-			// This run failed; a later run lists again from the cursor.
+		if isBatchJobUnavailable(cause) {
+			return ri.savePaused(ctx, api, job, cause)
+		}
+		if isErrBucketNotFound(cause) {
 			ri.mu.Lock()
-			ri.Complete = false
 			ri.Failed = true
 			ri.mu.Unlock()
 		}

cmd/batch-handlers.go

@@ -454,7 +454,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 		StartTime: job.Started,
 	}
 	if err := ri.loadOrInit(ctx, api, job); err != nil {
-		return err
+		return batchUnavailable("loading job progress", err)
 	}
 	if ri.Complete {
 		return nil
@@ -1065,16 +1065,20 @@ type batchJobInfo struct {
 	// FailureReason is set when the job was refused for its submitter. It
 	// ends the job: a job that has one does not run again.
 	FailureReason string `json:"failureReason,omitempty" msg:"fr"`
+	PausedReason  string `json:"pausedReason,omitempty" msg:"pr"`
 }
 
 const (
-	batchReplName         = "batch-replicate.bin"
-	batchReplFormat       = 1
-	batchReplVersionV1    = 1
-	batchReplVersion      = batchReplVersionV1
-	batchJobName          = "job.bin"
-	batchJobPrefix        = "batch-jobs"
-	batchJobReportsPrefix = batchJobPrefix + "/reports"
+	batchReplName      = "batch-replicate.bin"
+	batchReplFormat    = 1
+	batchReplVersionV1 = 1
+	batchReplVersion   = batchReplVersionV1
+	batchJobName       = "job.bin"
+	// The envelope rejects old decoders before they can ignore Submitter.
+	batchJobEnvelope             = "SILO-BATCH\x02"
+	batchAuthorizedReportVersion = 2
+	batchJobPrefix               = "batch-jobs"
+	batchJobReportsPrefix        = batchJobPrefix + "/reports"
 
 	batchReplJobAPIVersion        = "v1"
 	batchReplJobDefaultRetries    = 3
@@ -1162,7 +1166,7 @@ func (ri *batchJobInfo) loadByPath(ctx context.Context, api ObjectLayer, path st
 		return fmt.Errorf("%s: unknown format: %d", ri.JobType, binary.LittleEndian.Uint16(data[0:2]))
 	}
 	switch binary.LittleEndian.Uint16(data[2:4]) {
-	case version:
+	case version, batchAuthorizedReportVersion:
 	default:
 		return fmt.Errorf("%s: unknown version: %d", ri.JobType, binary.LittleEndian.Uint16(data[2:4]))
 	}
@@ -1210,6 +1214,7 @@ func (ri *batchJobInfo) clone() *batchJobInfo {
 		VersionsSkippedLocked: ri.VersionsSkippedLocked,
 		KeysRefused:           ri.KeysRefused,
 		FailureReason:         ri.FailureReason,
+		PausedReason:          ri.PausedReason,
 	}
 }
 
@@ -1263,6 +1268,7 @@ func (ri *batchJobInfo) updateAfter(ctx context.Context, api ObjectLayer, durati
 			jobTyp = string(job.Type())
 			ri.Version = batchExpireVersionV1
 		default:
+			ri.mu.Unlock()
 			return errInvalidArgument
 		}
 		if serverDebugLog {
@@ -1270,6 +1276,9 @@ func (ri *batchJobInfo) updateAfter(ctx context.Context, api ObjectLayer, durati
 		}
 		ri.LastUpdate = now
 
+		if job.Submitter != nil {
+			version = batchAuthorizedReportVersion
+		}
 		data := make([]byte, 4, ri.Msgsize()+4)
 
 		// Initialize the header.
@@ -1336,7 +1345,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		StartTime: job.Started,
 	}
 	if err := ri.loadOrInit(ctx, api, job); err != nil {
-		return err
+		return batchUnavailable("loading job progress", err)
 	}
 	if ri.Complete {
 		return nil
@@ -1880,7 +1889,11 @@ func (j *BatchJobRequest) save(ctx context.Context, api ObjectLayer) error {
 		return err
 	}
 
-	job, err := j.MarshalMsg(nil)
+	var prefix []byte
+	if j.Submitter != nil {
+		prefix = []byte(batchJobEnvelope)
+	}
+	job, err := j.MarshalMsg(prefix)
 	if err != nil {
 		return err
 	}
@@ -1901,6 +1914,12 @@ func (j *BatchJobRequest) load(ctx context.Context, api ObjectLayer, name string
 		return err
 	}
 
+	if bytes.HasPrefix(job, []byte("SILO-BATCH")) {
+		if !bytes.HasPrefix(job, []byte(batchJobEnvelope)) {
+			return errors.New("unsupported batch job envelope version")
+		}
+		job = job[len(batchJobEnvelope):]
+	}
 	_, err = j.UnmarshalMsg(job)
 	return err
 }
@@ -2070,7 +2089,7 @@ func (a adminAPIHandlers) DescribeBatchJob(w http.ResponseWriter, r *http.Reques
 func (a adminAPIHandlers) StartBatchJob(w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 
-	objectAPI, creds := validateAdminReq(ctx, w, r, policy.StartBatchJobAction)
+	objectAPI, creds := validateBatchJobAdminReq(ctx, w, r)
 	if objectAPI == nil {
 		return
 	}
@@ -2130,6 +2149,8 @@ func (a adminAPIHandlers) StartBatchJob(w http.ResponseWriter, r *http.Request)
 		apiErr := toAPIError(ctx, err)
 		if isBatchJobAuthzError(err) {
 			apiErr = errorCodes.ToAPIErrWithErr(ErrAccessDenied, err)
+		} else if isBatchJobUnavailable(err) {
+			apiErr = batchServiceUnavailable(err)
 		}
 		writeErrorResponseJSON(ctx, w, apiErr, r.URL)
 		return
@@ -2151,6 +2172,11 @@ func (a adminAPIHandlers) StartBatchJob(w http.ResponseWriter, r *http.Request)
 	}
 
 	if err = globalBatchJobPool.queueJob(job); err != nil {
+		// Admission failed (normally shutdown). Do not leave an unacknowledged
+		// manifest that could start after restart.
+		if deleteErr := job.delete(ctx, objectAPI); deleteErr != nil {
+			batchLogIf(ctx, deleteErr)
+		}
 		writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
 		return
 	}
@@ -2197,8 +2223,10 @@ func (a adminAPIHandlers) CancelBatchJob(w http.ResponseWriter, r *http.Request)
 		ID: jobID,
 	}
 
-	j.delete(ctx, objectAPI)
-
+	if err := j.delete(ctx, objectAPI); err != nil {
+		writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
 	writeSuccessNoContent(w)
 }
 
@@ -2208,11 +2236,10 @@ func (a adminAPIHandlers) CancelBatchJob(w http.ResponseWriter, r *http.Request)
 type BatchJobPool struct {
 	ctx          context.Context
 	objLayer     ObjectLayer
-	once         sync.Once
 	mu           sync.Mutex
 	jobCh        chan *BatchJobRequest
-	jmu          sync.Mutex // protects jobCancelers
-	jobCancelers map[string]context.CancelFunc
+	jmu          sync.Mutex // protects jobOwners
+	jobOwners    map[string]*batchJobOwner
 	workerKillCh chan struct{}
 	workerSize   int
 }
@@ -2226,9 +2253,10 @@ func newBatchJobPool(ctx context.Context, o ObjectLayer, workers int) *BatchJobP
 		objLayer:     o,
 		jobCh:        make(chan *BatchJobRequest, 10000),
 		workerKillCh: make(chan struct{}, workers),
-		jobCancelers: make(map[string]context.CancelFunc),
+		jobOwners:    make(map[string]*batchJobOwner),
 	}
 	jpool.ResizeWorkers(workers)
+	go jpool.retryLoop()
 
 	randomWait := func() time.Duration {
 		// randomWait depends on the number of nodes to avoid triggering resume and cleanups at the same time.
@@ -2249,14 +2277,14 @@ func (j *BatchJobPool) cleanupReports(randomWait func() time.Duration) {
 
 	for {
 		select {
-		case <-GlobalContext.Done():
+		case <-j.ctx.Done():
 			return
 		case <-t.C:
 			results := make(chan itemOrErr[ObjectInfo], 100)
 			ctx, cancel := context.WithCancel(j.ctx)
-			defer cancel()
 			if err := j.objLayer.Walk(ctx, minioMetaBucket, batchJobReportsPrefix, results, WalkOptions{}); err != nil {
 				batchLogIf(j.ctx, err)
+				cancel()
 				t.Reset(randomWait())
 				continue
 			}
@@ -2283,13 +2311,20 @@ func (j *BatchJobPool) cleanupReports(randomWait func() time.Duration) {
 				}
 			}
 
+			cancel()
 			t.Reset(randomWait())
 		}
 	}
 }
 
 func (j *BatchJobPool) resume(randomWait func() time.Duration) {
-	time.Sleep(randomWait())
+	timer := time.NewTimer(randomWait())
+	defer timer.Stop()
+	select {
+	case <-j.ctx.Done():
+		return
+	case <-timer.C:
+	}
 
 	results := make(chan itemOrErr[ObjectInfo], 100)
 	ctx, cancel := context.WithCancel(j.ctx)
@@ -2340,43 +2375,26 @@ func (j *BatchJobPool) AddWorker() {
 			if !ok {
 				return
 			}
+			if !j.beginJob(job) {
+				continue
+			}
+			var err error
 			switch {
 			case job.Replicate != nil:
 				if job.Replicate.RemoteToLocal() {
-					if err := job.Replicate.StartFromSource(job.ctx, j.objLayer, *job); err != nil {
-						if !isErrBucketNotFound(err) {
-							batchLogIf(j.ctx, err)
-							j.canceler(job.ID, false)
-							continue
-						}
-						// Bucket not found proceed to delete such a job.
-					}
+					err = job.Replicate.StartFromSource(job.ctx, j.objLayer, *job)
 				} else {
-					if err := job.Replicate.Start(job.ctx, j.objLayer, *job); err != nil {
-						if !isErrBucketNotFound(err) {
-							batchLogIf(j.ctx, err)
-							j.canceler(job.ID, false)
-							continue
-						}
-						// Bucket not found proceed to delete such a job.
-					}
+					err = job.Replicate.Start(job.ctx, j.objLayer, *job)
 				}
 			case job.KeyRotate != nil:
-				if err := job.KeyRotate.Start(job.ctx, j.objLayer, *job); err != nil {
-					if !isErrBucketNotFound(err) {
-						batchLogIf(j.ctx, err)
-						continue
-					}
-				}
+				err = job.KeyRotate.Start(job.ctx, j.objLayer, *job)
 			case job.Expire != nil:
-				if err := job.Expire.Start(job.ctx, j.objLayer, *job); err != nil {
-					if !isErrBucketNotFound(err) {
-						batchLogIf(j.ctx, err)
-						continue
-					}
-				}
+				err = job.Expire.Start(job.ctx, j.objLayer, *job)
 			}
-			j.canceler(job.ID, false)
+			if err != nil {
+				batchLogIf(j.ctx, err)
+			}
+			j.finishJob(job, err)
 		case <-j.workerKillCh:
 			return
 		}
@@ -2398,46 +2416,66 @@ func (j *BatchJobPool) ResizeWorkers(n int) {
 	}
 	for j.workerSize > n {
 		j.workerSize--
-		go func() { j.workerKillCh <- struct{}{} }()
+		go func() {
+			select {
+			case j.workerKillCh <- struct{}{}:
+			case <-j.ctx.Done():
+			}
+		}()
 	}
 }
 
 func (j *BatchJobPool) queueJob(req *BatchJobRequest) error {
-	if j == nil {
+	if j == nil || req == nil || req.ID == "" {
 		return errInvalidArgument
 	}
-	jctx, jcancel := context.WithCancel(j.ctx)
 	j.jmu.Lock()
-	j.jobCancelers[req.ID] = jcancel
-	j.jmu.Unlock()
+	defer j.jmu.Unlock()
+	if err := j.ctx.Err(); err != nil {
+		return err
+	}
+	if owner := j.jobOwners[req.ID]; owner != nil {
+		if owner.state == "canceled" {
+			return context.Canceled
+		}
+		return nil
+	}
+	if j.jobOwners == nil {
+		j.jobOwners = make(map[string]*batchJobOwner)
+	}
+	jctx, jcancel := context.WithCancel(j.ctx)
 	req.ctx = jctx
-
+	owner := &batchJobOwner{req: req, cancel: jcancel, state: "queued"}
+	j.jobOwners[req.ID] = owner
+	// A full channel retains explicit pending work owned by the scheduler.
+	// The durable submission still succeeds and cannot become ownerless.
 	select {
-	case <-j.ctx.Done():
-		j.once.Do(func() {
-			xioutil.SafeClose(j.jobCh)
-		})
 	case j.jobCh <- req:
 	default:
-		return fmt.Errorf("batch job queue is currently full please try again later (job %s)", req.ID)
+		owner.state = "paused"
 	}
 	return nil
 }
 
-// delete canceler from the map, cancel job if requested
 func (j *BatchJobPool) canceler(jobID string, cancel bool) error {
 	if j == nil {
 		return errInvalidArgument
 	}
 	j.jmu.Lock()
 	defer j.jmu.Unlock()
-	if canceler, ok := j.jobCancelers[jobID]; ok {
-		if cancel {
-			canceler()
-		}
+	if !cancel {
+		return nil
+	} // Worker completion uses its generation token.
+	if j.jobOwners == nil {
+		j.jobOwners = make(map[string]*batchJobOwner)
 	}
-	if cancel {
-		delete(j.jobCancelers, jobID)
+	if owner := j.jobOwners[jobID]; owner != nil {
+		if owner.cancel != nil {
+			owner.cancel()
+		}
+		j.jobOwners[jobID] = &batchJobOwner{state: "canceled"}
+	} else {
+		j.jobOwners[jobID] = &batchJobOwner{state: "canceled"}
 	}
 	return nil
 }

cmd/batch-handlers_gen.go

@@ -655,6 +655,12 @@ func (z *batchJobInfo) DecodeMsg(dc *msgp.Reader) (err error) {
 				err = msgp.WrapError(err, "FailureReason")
 				return
 			}
+		case "pr":
+			z.PausedReason, err = dc.ReadString()
+			if err != nil {
+				err = msgp.WrapError(err, "PausedReason")
+				return
+			}
 		default:
 			err = dc.Skip()
 			if err != nil {
@@ -668,9 +674,9 @@ func (z *batchJobInfo) DecodeMsg(dc *msgp.Reader) (err error) {
 
 // EncodeMsg implements msgp.Encodable
 func (z *batchJobInfo) EncodeMsg(en *msgp.Writer) (err error) {
-	// map header, size 20
+	// map header, size 21
 	// write "v"
-	err = en.Append(0xde, 0x0, 0x14, 0xa1, 0x76)
+	err = en.Append(0xde, 0x0, 0x15, 0xa1, 0x76)
 	if err != nil {
 		return
 	}
@@ -869,15 +875,25 @@ func (z *batchJobInfo) EncodeMsg(en *msgp.Writer) (err error) {
 		err = msgp.WrapError(err, "FailureReason")
 		return
 	}
+	// write "pr"
+	err = en.Append(0xa2, 0x70, 0x72)
+	if err != nil {
+		return
+	}
+	err = en.WriteString(z.PausedReason)
+	if err != nil {
+		err = msgp.WrapError(err, "PausedReason")
+		return
+	}
 	return
 }
 
 // MarshalMsg implements msgp.Marshaler
 func (z *batchJobInfo) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 20
+	// map header, size 21
 	// string "v"
-	o = append(o, 0xde, 0x0, 0x14, 0xa1, 0x76)
+	o = append(o, 0xde, 0x0, 0x15, 0xa1, 0x76)
 	o = msgp.AppendInt(o, z.Version)
 	// string "jid"
 	o = append(o, 0xa3, 0x6a, 0x69, 0x64)
@@ -936,6 +952,9 @@ func (z *batchJobInfo) MarshalMsg(b []byte) (o []byte, err error) {
 	// string "fr"
 	o = append(o, 0xa2, 0x66, 0x72)
 	o = msgp.AppendString(o, z.FailureReason)
+	// string "pr"
+	o = append(o, 0xa2, 0x70, 0x72)
+	o = msgp.AppendString(o, z.PausedReason)
 	return
 }
 
@@ -1077,6 +1096,12 @@ func (z *batchJobInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "FailureReason")
 				return
 			}
+		case "pr":
+			z.PausedReason, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "PausedReason")
+				return
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -1091,6 +1116,6 @@ func (z *batchJobInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z *batchJobInfo) Msgsize() (s int) {
-	s = 3 + 2 + msgp.IntSize + 4 + msgp.StringPrefixSize + len(z.JobID) + 3 + msgp.StringPrefixSize + len(z.JobType) + 3 + msgp.TimeSize + 3 + msgp.TimeSize + 3 + msgp.IntSize + 3 + msgp.IntSize + 4 + msgp.BoolSize + 4 + msgp.BoolSize + 5 + msgp.StringPrefixSize + len(z.Bucket) + 5 + msgp.StringPrefixSize + len(z.Object) + 3 + msgp.Int64Size + 3 + msgp.Int64Size + 4 + msgp.Int64Size + 4 + msgp.Int64Size + 3 + msgp.Int64Size + 3 + msgp.Int64Size + 4 + msgp.Int64Size + 3 + msgp.Int64Size + 3 + msgp.StringPrefixSize + len(z.FailureReason)
+	s = 3 + 2 + msgp.IntSize + 4 + msgp.StringPrefixSize + len(z.JobID) + 3 + msgp.StringPrefixSize + len(z.JobType) + 3 + msgp.TimeSize + 3 + msgp.TimeSize + 3 + msgp.IntSize + 3 + msgp.IntSize + 4 + msgp.BoolSize + 4 + msgp.BoolSize + 5 + msgp.StringPrefixSize + len(z.Bucket) + 5 + msgp.StringPrefixSize + len(z.Object) + 3 + msgp.Int64Size + 3 + msgp.Int64Size + 4 + msgp.Int64Size + 4 + msgp.Int64Size + 3 + msgp.Int64Size + 3 + msgp.Int64Size + 4 + msgp.Int64Size + 3 + msgp.Int64Size + 3 + msgp.StringPrefixSize + len(z.FailureReason) + 3 + msgp.StringPrefixSize + len(z.PausedReason)
 	return
 }

cmd/batch-progress.go

@@ -173,6 +173,7 @@ type batchJobStatusMetric struct {
 	VersionsSkippedLocked int64  `json:"versionsSkippedLocked,omitempty"`
 	KeysRefused           int64  `json:"keysRefused,omitempty"`
 	FailureReason         string `json:"failureReason,omitempty"`
+	PausedReason          string `json:"pausedReason,omitempty"`
 }
 
 func (ri *batchJobInfo) statusMetric() batchJobStatusMetric {
@@ -183,5 +184,6 @@ func (ri *batchJobInfo) statusMetric() batchJobStatusMetric {
 		VersionsSkippedLocked: ri.VersionsSkippedLocked,
 		KeysRefused:           ri.KeysRefused,
 		FailureReason:         ri.FailureReason,
+		PausedReason:          ri.PausedReason,
 	}
 }

cmd/batch-retry.go

@@ -0,0 +1,194 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/rand"
+	"net/http"
+	"time"
+
+	"github.com/minio/minio/internal/auth"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+// An unavailable dependency is not an authorization decision. It stops the
+// current attempt without turning an outage into either an allow or a denial.
+type batchJobUnavailableError struct {
+	reason string
+	err    error
+}
+
+func (e batchJobUnavailableError) Error() string { return fmt.Sprintf("%s: %v", e.reason, e.err) }
+func (e batchJobUnavailableError) Unwrap() error { return e.err }
+func batchUnavailable(reason string, err error) error {
+	if err == nil {
+		return nil
+	}
+	if isBatchJobUnavailable(err) {
+		return err
+	}
+	return batchJobUnavailableError{reason: reason, err: err}
+}
+func isBatchJobUnavailable(err error) bool { var e batchJobUnavailableError; return errors.As(err, &e) }
+func isBatchJobStopped(err error) bool {
+	return isBatchJobAuthzError(err) || isBatchJobUnavailable(err)
+}
+func batchListingError(err error) error {
+	if isErrBucketNotFound(err) {
+		return err
+	}
+	return batchUnavailable("object listing unavailable", err)
+}
+
+func (ri *batchJobInfo) pause(err error) {
+	ri.mu.Lock()
+	defer ri.mu.Unlock()
+	// A concurrent explicit denial wins over a transient worker error.
+	if ri.FailureReason != "" {
+		return
+	}
+	ri.Complete, ri.Failed = false, false
+	ri.PausedReason = err.Error()
+}
+func (ri *batchJobInfo) savePaused(ctx context.Context, api ObjectLayer, job BatchJobRequest, err error) error {
+	ri.pause(err)
+	globalBatchJobsMetrics.save(job.ID, ri)
+	saveCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+	defer cancel()
+	if saveErr := ri.updateAfter(saveCtx, api, 0, job); saveErr != nil {
+		return batchUnavailable("saving paused job", errors.Join(err, saveErr))
+	}
+	return err
+}
+
+// Authenticate the HTTP request normally, but keep dependency errors through
+// the authorization step so submission returns 503, not a permanent 403.
+func validateBatchJobAdminReq(ctx context.Context, w http.ResponseWriter, r *http.Request) (ObjectLayer, auth.Credentials) {
+	api := newObjectLayerFn()
+	if api == nil || globalNotificationSys == nil || !batchIAMReady() {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
+		return nil, auth.Credentials{}
+	}
+	cred, owner, code := validateAdminSignature(ctx, r, "")
+	if code != ErrNone {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(code), r.URL)
+		return nil, cred
+	}
+	submitter, err := newBatchJobSubmitter(cred)
+	if err == nil {
+		var principal batchJobPrincipal
+		principal, err = submitter.principal(ctx)
+		if err == nil {
+			err = principal.authorizeArgs(policy.Args{AccountName: cred.AccessKey, Groups: cred.Groups,
+				Action: policy.Action(policy.StartBatchJobAction), ConditionValues: getConditionValues(r, "", cred), IsOwner: owner, Claims: principal.claims}, nil)
+		}
+	}
+	if err != nil {
+		apiErr := errorCodes.ToAPIErrWithErr(ErrAccessDenied, err)
+		if isBatchJobUnavailable(err) {
+			apiErr = batchServiceUnavailable(err)
+		}
+		writeErrorResponseJSON(ctx, w, apiErr, r.URL)
+		return nil, cred
+	}
+	return api, cred
+}
+
+// One owner record spans queued, running and paused attempts. The request
+// pointer is the generation token. Canceled records are tombstones until
+// process shutdown, so a manifest loaded before cancellation cannot requeue.
+type batchJobOwner struct {
+	req     *BatchJobRequest
+	cancel  context.CancelFunc
+	state   string
+	retries uint
+	due     time.Time
+}
+
+func (j *BatchJobPool) beginJob(job *BatchJobRequest) bool {
+	j.jmu.Lock()
+	defer j.jmu.Unlock()
+	o := j.jobOwners[job.ID]
+	if o == nil || o.req != job || o.state != "queued" || job.ctx.Err() != nil || j.ctx.Err() != nil {
+		return false
+	}
+	o.state = "running"
+	return true
+}
+func (j *BatchJobPool) finishJob(job *BatchJobRequest, err error) {
+	j.jmu.Lock()
+	defer j.jmu.Unlock()
+	o := j.jobOwners[job.ID]
+	if o == nil || o.req != job || o.state != "running" {
+		return
+	}
+	if isBatchJobUnavailable(err) && job.ctx.Err() == nil && j.ctx.Err() == nil {
+		delay := 30 * time.Second * time.Duration(1<<min(o.retries, 4))
+		o.retries++
+		o.due = time.Now().Add(delay/2 + time.Duration(rand.Float64()*float64(delay)))
+		o.state = "paused"
+		return
+	}
+	o.cancel()
+	delete(j.jobOwners, job.ID)
+}
+
+// retryDue holds the ownership lock across the nonblocking handoff. A fast
+// worker cannot replace a pause while the scheduler removes the old pause.
+func (j *BatchJobPool) retryDue(now time.Time) {
+	j.jmu.Lock()
+	defer j.jmu.Unlock()
+	if j.ctx.Err() != nil {
+		return
+	}
+	for _, o := range j.jobOwners {
+		if o.state != "paused" || now.Before(o.due) || o.req.ctx.Err() != nil {
+			continue
+		}
+		select {
+		case j.jobCh <- o.req:
+			o.state = "queued"
+		default:
+		}
+	}
+}
+func (j *BatchJobPool) retryLoop() {
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-j.ctx.Done():
+			return
+		case now := <-ticker.C:
+			j.retryDue(now)
+		}
+	}
+}
+
+func batchServiceUnavailable(err error) APIError {
+	return APIError{Code: "ServiceUnavailable", Description: err.Error(), HTTPStatusCode: http.StatusServiceUnavailable}
+}
+
+// initStore precedes the first successful cache load. During that interval a
+// missing identity or policy is not a revocation decision, including for root.
+func batchIAMReady() bool {
+	sys := globalIAMSys
+	if sys == nil {
+		return false
+	}
+	sys.Lock()
+	defer sys.Unlock()
+	if sys.store == nil {
+		return false
+	}
+	select {
+	case <-sys.configLoaded:
+		return true
+	default:
+		return false
+	}
+}

cmd/batch-retry_test.go

@@ -0,0 +1,312 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	polplugin "github.com/minio/minio/internal/config/policy/plugin"
+	xnet "github.com/pgsty/silo-pkg/v3/net"
+	"github.com/pgsty/silo-pkg/v3/policy"
+	"github.com/tinylib/msgp/msgp"
+)
+
+func waitBatchState(t *testing.T, pool *BatchJobPool, id, state string) {
+	t.Helper()
+	timeout := time.NewTimer(10 * time.Second)
+	defer timeout.Stop()
+	tick := time.NewTicker(time.Millisecond)
+	defer tick.Stop()
+	for {
+		pool.jmu.Lock()
+		owner := pool.jobOwners[id]
+		ready := owner == nil && state == ""
+		if owner != nil {
+			ready = owner.state == state
+		}
+		pool.jmu.Unlock()
+		if ready {
+			return
+		}
+		select {
+		case <-timeout.C:
+			t.Fatalf("job did not reach %q", state)
+		case <-tick.C:
+		}
+	}
+}
+
+func TestBatchPluginPauseRecoversInProcess(t *testing.T) {
+	for _, midrun := range []bool{false, true} {
+		t.Run(map[bool]string{false: "before-start", true: "during-delete"}[midrun], func(t *testing.T) {
+			f := newBatchAuthzFixture(t)
+			older := f.putVersions(t, "data/key")
+			var unavailable atomic.Bool
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				var request struct {
+					Input policy.Args `json:"input"`
+				}
+				_ = json.NewDecoder(r.Body).Decode(&request)
+				if unavailable.Load() && (!midrun || request.Input.Action == policy.DeleteObjectVersionAction) {
+					w.WriteHeader(503)
+					io.WriteString(w, `{"result":true}`)
+					return
+				}
+				io.WriteString(w, `{"result":true}`)
+			}))
+			defer server.Close()
+			url, err := xnet.ParseHTTPURL(server.URL)
+			mustIAM(t, err)
+			previous := newGlobalAuthZPluginFn()
+			setGlobalAuthZPlugin(polplugin.New(polplugin.Args{URL: url, CloseRespFn: func(r io.ReadCloser) { r.Close() }}))
+			defer setGlobalAuthZPlugin(previous)
+			rec, job := f.submit(t, globalActiveCred, f.expireYAML("data/"))
+			if job == nil {
+				t.Fatalf("submit %d: %s", rec.Code, rec.Body.String())
+			}
+			pool := globalBatchJobPool
+			ctx, cancel := context.WithCancel(f.ctx)
+			pool.ctx = ctx
+			// Re-admit the fixture's consumed request with a context owned by this worker.
+			pool.jmu.Lock()
+			pool.jobOwners[job.ID].cancel()
+			delete(pool.jobOwners, job.ID)
+			pool.jmu.Unlock()
+			mustIAM(t, pool.queueJob(job))
+			unavailable.Store(true)
+			done := make(chan struct{})
+			go func() { defer close(done); pool.AddWorker() }()
+			defer func() { cancel(); <-done }()
+			waitBatchState(t, pool, job.ID, "paused")
+			ri := f.jobInfo(t, job)
+			if ri.Failed || ri.Complete || ri.FailureReason != "" || ri.PausedReason == "" || ri.Object != "" {
+				t.Fatalf("paused report: %+v", ri)
+			}
+			if !f.versionExists(t, "data/key", older) {
+				t.Fatal("outage authorized a deletion")
+			}
+			unavailable.Store(false)
+			pool.retryDue(time.Now().Add(time.Hour))
+			waitBatchState(t, pool, job.ID, "")
+			ri = f.jobInfo(t, job)
+			if !ri.Complete || ri.Failed || ri.PausedReason != "" || ri.Objects != 1 || f.versionExists(t, "data/key", older) {
+				t.Fatalf("resumed report: %+v", ri)
+			}
+		})
+	}
+}
+
+func TestBatchSubmissionUnavailableIs503(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(503); io.WriteString(w, `{"result":true}`) }))
+	defer server.Close()
+	url, err := xnet.ParseHTTPURL(server.URL)
+	mustIAM(t, err)
+	previous := newGlobalAuthZPluginFn()
+	setGlobalAuthZPlugin(polplugin.New(polplugin.Args{URL: url, CloseRespFn: func(r io.ReadCloser) { r.Close() }}))
+	defer setGlobalAuthZPlugin(previous)
+	rec, job := f.submit(t, globalActiveCred, f.expireYAML("data/"))
+	if rec.Code != 503 || job != nil {
+		t.Fatalf("outage submission: %d %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestBatchPoolOwnershipAndCancellation(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	pool := &BatchJobPool{ctx: ctx, jobCh: make(chan *BatchJobRequest, 1)}
+	first, second := &BatchJobRequest{ID: "first"}, &BatchJobRequest{ID: "second"}
+	mustIAM(t, pool.queueJob(first))
+	mustIAM(t, pool.queueJob(second))
+	if pool.jobOwners[second.ID].state != "paused" {
+		t.Fatal("full queue lost pending owner")
+	}
+	original := second.ctx
+	var wg sync.WaitGroup
+	for range 8 {
+		wg.Go(func() {
+			if err := pool.queueJob(&BatchJobRequest{ID: "second"}); err != nil {
+				t.Error(err)
+			}
+		})
+	}
+	wg.Wait()
+	if second.ctx != original || len(pool.jobCh) != 1 {
+		t.Fatal("duplicate changed ownership")
+	}
+	if !pool.beginJob(<-pool.jobCh) {
+		t.Fatal("first not owned")
+	}
+	pool.finishJob(first, nil)
+	pool.retryDue(time.Now())
+	next := <-pool.jobCh
+	if next != second || !pool.beginJob(next) {
+		t.Fatal("pending handoff lost ownership")
+	}
+	for range 2 {
+		pool.finishJob(second, batchUnavailable("test", errors.New("outage")))
+		pool.retryDue(time.Now().Add(time.Hour))
+		if !pool.beginJob(<-pool.jobCh) {
+			t.Fatal("retry lost owner")
+		}
+	}
+	if pool.jobOwners[second.ID].retries != 2 {
+		t.Fatal("retry count reset during handoff")
+	}
+	mustIAM(t, pool.canceler(second.ID, true))
+	mustIAM(t, pool.canceler(second.ID, true))
+	pool.finishJob(second, batchUnavailable("late", errors.New("outage")))
+	pool.retryDue(time.Now().Add(time.Hour))
+	if len(pool.jobCh) != 0 || second.ctx.Err() == nil {
+		t.Fatal("cancel lost to retry")
+	}
+	if err := pool.queueJob(&BatchJobRequest{ID: second.ID}); !errors.Is(err, context.Canceled) {
+		t.Fatalf("stale admission: %v", err)
+	}
+	// A cancellation before this node admitted a manifest is a tombstone too.
+	mustIAM(t, pool.canceler("not-loaded", true))
+	mustIAM(t, pool.canceler("not-loaded", true))
+	if err := pool.queueJob(&BatchJobRequest{ID: "not-loaded"}); !errors.Is(err, context.Canceled) {
+		t.Fatalf("stale resume: %v", err)
+	}
+	cancel()
+	if err := pool.queueJob(&BatchJobRequest{ID: "shutdown"}); !errors.Is(err, context.Canceled) {
+		t.Fatalf("shutdown admission: %v", err)
+	}
+	pool.retryDue(time.Now().Add(time.Hour)) // channel remains open; no send/close race.
+}
+
+type batchPolicyOutage struct {
+	IAMStorageAPI
+	failed string
+	err    error
+}
+
+func (s *batchPolicyOutage) loadPolicyDoc(ctx context.Context, name string, out map[string]PolicyDoc) error {
+	if name == s.failed {
+		return s.err
+	}
+	return s.IAMStorageAPI.loadPolicyDoc(ctx, name, out)
+}
+
+func TestBatchPolicySnapshotKeepsUnavailableDeny(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	batchAuthzPolicy(f.ctx, t, f.sys, "retry-allow", batchAuthzFullAccess)
+	batchAuthzPolicy(f.ctx, t, f.sys, "retry-deny", `{"Effect":"Deny","Action":["s3:DeleteObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"]}`)
+	cred := iamAdminUser(f.ctx, t, f.sys, "retry-user", "retry-allow,retry-deny")
+	sub, err := newBatchJobSubmitter(cred)
+	mustIAM(t, err)
+	store := f.sys.store
+	original := store.IAMStorageAPI
+	defer func() { store.IAMStorageAPI = original }()
+	cache := store.lock()
+	delete(cache.iamPolicyDocsMap, "retry-deny")
+	store.unlock()
+	store.IAMStorageAPI = &batchPolicyOutage{IAMStorageAPI: original, failed: "retry-deny", err: errErasureReadQuorum}
+	request := batchAuthzRequest{action: policy.DeleteObjectVersionAction, bucket: f.bucket, object: "key", versionID: mustGetUUID()}
+	if err := sub.authorize(f.ctx, request); !isBatchJobUnavailable(err) {
+		t.Fatalf("missing Deny must pause, got %v", err)
+	}
+	store.IAMStorageAPI = original
+	if err := sub.authorize(f.ctx, request); !isBatchJobAuthzError(err) {
+		t.Fatalf("recovered Deny was not enforced: %v", err)
+	}
+	// Root also waits for IAM readiness; startup must not bypass the pause gate.
+	root, err := newBatchJobSubmitter(globalActiveCred)
+	mustIAM(t, err)
+	f.sys.Lock()
+	f.sys.store = nil
+	f.sys.Unlock()
+	_, err = root.principal(f.ctx)
+	f.sys.Lock()
+	f.sys.store = store
+	f.sys.Unlock()
+	if !isBatchJobUnavailable(err) {
+		t.Fatalf("root before IAM readiness: %v", err)
+	}
+}
+
+func TestBatchPausedReportCompatibility(t *testing.T) {
+	before := &batchJobInfo{JobID: "paused", Object: "cursor", PausedReason: "IAM unavailable"}
+	data, err := before.MarshalMsg(nil)
+	mustIAM(t, err)
+	var after batchJobInfo
+	_, err = after.UnmarshalMsg(data)
+	mustIAM(t, err)
+	if after.PausedReason != before.PausedReason || after.clone().PausedReason != before.PausedReason || after.statusMetric().PausedReason != before.PausedReason {
+		t.Fatal("paused reason lost")
+	}
+	old := msgp.AppendMapHeader(nil, 2)
+	old = msgp.AppendString(old, "jid")
+	old = msgp.AppendString(old, "legacy")
+	old = msgp.AppendString(old, "cmp")
+	old = msgp.AppendBool(old, true)
+	var legacy batchJobInfo
+	_, err = legacy.UnmarshalMsg(old)
+	mustIAM(t, err)
+	if legacy.PausedReason != "" || !legacy.Complete {
+		t.Fatal("old report decode changed")
+	}
+	after.refuse(errBatchJobInvalid)
+	if after.PausedReason != "" || !after.Failed {
+		t.Fatal("terminal denial retained pause")
+	}
+}
+
+func TestBatchPersistenceFencesOlderExecutors(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	rec, job := f.submit(t, globalActiveCred, f.expireYAML("data/"))
+	if job == nil {
+		t.Fatalf("submit: %d %s", rec.Code, rec.Body.String())
+	}
+	raw, err := readConfig(f.ctx, f.obj, getJobPath(*job))
+	mustIAM(t, err)
+	if !bytes.HasPrefix(raw, []byte(batchJobEnvelope)) {
+		t.Fatal("new manifest has no executor version fence")
+	}
+	var oldDecoder BatchJobRequest
+	if _, err := oldDecoder.UnmarshalMsg(raw); err == nil {
+		t.Fatal("old bare-msgpack executor could decode new manifest")
+	}
+	var loaded BatchJobRequest
+	mustIAM(t, loaded.load(f.ctx, f.obj, getJobPath(*job)))
+	if loaded.Submitter == nil {
+		t.Fatal("new executor lost submitter")
+	}
+	ri := &batchJobInfo{JobID: job.ID, JobType: string(job.Type()), StartTime: job.Started}
+	outage := batchUnavailable("test", errErasureReadQuorum)
+	if err := ri.savePaused(f.ctx, f.obj, *job, outage); !errors.Is(err, outage) {
+		t.Fatal(err)
+	}
+	reportPath, err := job.getJobReportPath()
+	mustIAM(t, err)
+	report, err := readConfig(f.ctx, f.obj, reportPath)
+	mustIAM(t, err)
+	if binary.LittleEndian.Uint16(report[2:4]) == batchExpireVersionV1 {
+		t.Fatal("old executor would accept paused report")
+	}
+	mustIAM(t, ri.load(f.ctx, f.obj, *job))
+	if ri.PausedReason == "" || ri.Failed {
+		t.Fatal("restart lost pause")
+	}
+	// A stale resume loaded the manifest before cancellation.
+	mustIAM(t, globalBatchJobPool.canceler(job.ID, true))
+	mustIAM(t, job.delete(f.ctx, f.obj))
+	if err := globalBatchJobPool.queueJob(&loaded); !errors.Is(err, context.Canceled) {
+		t.Fatalf("canceled job re-admitted: %v", err)
+	}
+	if err := loaded.load(f.ctx, f.obj, getJobPath(*job)); !errors.Is(err, errNoSuchJob) {
+		t.Fatalf("canceled manifest recreated: %v", err)
+	}
+}

cmd/batch-rotate.go

@@ -385,7 +385,7 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		StartTime: job.Started,
 	}
 	if err := ri.loadOrInit(ctx, api, job); err != nil {
-		return err
+		return batchUnavailable("loading job progress", err)
 	}
 	if ri.Complete {
 		return nil
@@ -490,7 +490,11 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 	defer cancelCause(nil)
 	// refuse ends the job for its submitter: record why before stopping.
 	refuse := func(err error) {
-		ri.refuse(err)
+		if isBatchJobUnavailable(err) {
+			ri.pause(err)
+		} else {
+			ri.refuse(err)
+		}
 		cancelCause(err)
 	}
 	progress := newBatchProgress(ri, r.Bucket)
@@ -500,8 +504,11 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		Marker: lastObject,
 		Filter: selectObj,
 	}); err != nil {
-		// Do not need to retry if we can't list objects on source.
-		return err
+		cause := batchListingError(err)
+		if isBatchJobUnavailable(cause) {
+			return ri.savePaused(ctx, api, job, cause)
+		}
+		return cause
 	}
 	failed := false
 	for res := range results {
@@ -511,8 +518,9 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		}
 		if res.Err != nil {
 			failed = true
+			cancelCause(batchListingError(res.Err))
 			batchLogIf(ctx, res.Err)
-			break
+			continue
 		}
 		result := res.Item
 		if !batchResumeAfter(lastObject, result.Name) {
@@ -532,12 +540,15 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 			for attempts := 1; attempts <= retryAttempts; attempts++ {
 				stopFn := globalBatchJobsMetrics.trace(batchJobMetricKeyRotation, job.ID, attempts)
 				err := r.KeyRotate(ctx, api, result, job.Submitter)
+				if errors.Is(err, errBatchRotateMoving) {
+					err = batchUnavailable("object is being moved", err)
+				}
 				stopFn(result, err)
 				if err == nil {
 					counts.objects = 1
 					break
 				}
-				if isBatchJobAuthzError(err) {
+				if isBatchJobStopped(err) {
 					refuse(err)
 					return
 				}
@@ -585,6 +596,9 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		progress.finish()
 	}
 	if cause != nil && !ri.refused() {
+		if isBatchJobUnavailable(cause) {
+			return ri.savePaused(ctx, api, job, cause)
+		}
 		// Stopped before the end: keep the work after the cursor pending.
 		globalBatchJobsMetrics.save(job.ID, ri)
 		batchLogIf(ctx, ri.updateAfter(GlobalContext, api, 0, job))

cmd/batch-submitter-authz_test.go

@@ -46,10 +46,10 @@ func batchTestJobPool(ctx context.Context, t *testing.T, obj ObjectLayer) {
 	t.Helper()
 	prevPool := globalBatchJobPool
 	globalBatchJobPool = &BatchJobPool{
-		ctx:          ctx,
-		objLayer:     obj,
-		jobCh:        make(chan *BatchJobRequest, 10),
-		jobCancelers: make(map[string]context.CancelFunc),
+		ctx:       ctx,
+		objLayer:  obj,
+		jobCh:     make(chan *BatchJobRequest, 10),
+		jobOwners: make(map[string]*batchJobOwner),
 	}
 	globalBatchJobsMetrics.Lock()
 	globalBatchJobsMetrics.metrics = make(map[string]*batchJobInfo)

cmd/iam-store.go

@@ -3262,6 +3262,8 @@ func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error
 	return nil
 }
 
+var errTokenSigningKeyUnavailable = errors.New("token signing key unavailable")
+
 func extractJWTClaims(u UserIdentity) (jwtClaims *jwt.MapClaims, err error) {
 	keys := make([]string, 0, 3)
 
@@ -3272,7 +3274,7 @@ func extractJWTClaims(u UserIdentity) (jwtClaims *jwt.MapClaims, err error) {
 	if globalSiteReplicationSys.isEnabled() {
 		secretKey, err := getTokenSigningKey()
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("%w: %w", errTokenSigningKeyUnavailable, err)
 		}
 		keys = append(keys, secretKey)
 	}

internal/config/policy/plugin/config.go

@@ -19,7 +19,9 @@ package plugin
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"net/http"
 	"time"
@@ -180,6 +182,14 @@ func New(args Args) *AuthZPlugin {
 
 // IsAllowed - checks given policy args is allowed to continue the REST API.
 func (o *AuthZPlugin) IsAllowed(args policy.Args) (bool, error) {
+	return o.IsAllowedContext(context.Background(), args)
+}
+
+// IsAllowedContext requires an explicit decision from a successful response.
+// The timeout also bounds a server that accepts a connection but never replies.
+func (o *AuthZPlugin) IsAllowedContext(ctx context.Context, args policy.Args) (bool, error) {
+	ctx, cancel := context.WithTimeout(ctx, time.Minute)
+	defer cancel()
 	if o == nil {
 		return false, nil
 	}
@@ -193,7 +203,7 @@ func (o *AuthZPlugin) IsAllowed(args policy.Args) (bool, error) {
 		return false, err
 	}
 
-	req, err := http.NewRequest(http.MethodPost, o.args.URL.String(), bytes.NewReader(inputBytes))
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, o.args.URL.String(), bytes.NewReader(inputBytes))
 	if err != nil {
 		return false, err
 	}
@@ -209,37 +219,32 @@ func (o *AuthZPlugin) IsAllowed(args policy.Args) (bool, error) {
 	}
 	defer o.args.CloseRespFn(resp.Body)
 
-	// Read the body to be saved later.
-	opaRespBytes, err := io.ReadAll(resp.Body)
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return false, fmt.Errorf("authorization plugin returned HTTP %d", resp.StatusCode)
+	}
+	// Bound input and reject trailing data as well as absent/null decisions.
+	data, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20+1))
 	if err != nil {
 		return false, err
 	}
-
-	// Handle large OPA responses when OPA URL is of
-	// form http://localhost:8181/v1/data/httpapi/authz
-	type opaResultAllow struct {
-		Result struct {
-			Allow bool `json:"allow"`
-		} `json:"result"`
+	if len(data) > 1<<20 {
+		return false, fmt.Errorf("authorization plugin response is too large")
 	}
-
-	// Handle simpler OPA responses when OPA URL is of
-	// form http://localhost:8181/v1/data/httpapi/authz/allow
-	type opaResult struct {
-		Result bool `json:"result"`
+	var result struct {
+		Result json.RawMessage `json:"result"`
 	}
-
-	respBody := bytes.NewReader(opaRespBytes)
-
-	var result opaResult
-	if err = json.NewDecoder(respBody).Decode(&result); err != nil {
-		respBody.Seek(0, 0)
-		var resultAllow opaResultAllow
-		if err = json.NewDecoder(respBody).Decode(&resultAllow); err != nil {
-			return false, err
-		}
-		return resultAllow.Result.Allow, nil
+	if err := json.Unmarshal(data, &result); err != nil {
+		return false, err
 	}
-
-	return result.Result, nil
+	var allowed *bool
+	if err := json.Unmarshal(result.Result, &allowed); err == nil && allowed != nil {
+		return *allowed, nil
+	}
+	var nested struct {
+		Allow *bool `json:"allow"`
+	}
+	if err := json.Unmarshal(result.Result, &nested); err == nil && nested.Allow != nil {
+		return *nested.Allow, nil
+	}
+	return false, fmt.Errorf("authorization plugin response has no boolean decision")
 }

internal/config/policy/plugin/config_test.go

@@ -0,0 +1,48 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+package plugin
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/pgsty/silo-pkg/v3/net"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+func TestAuthZPluginExplicitDecision(t *testing.T) {
+	for _, tc := range []struct {
+		body        string
+		status      int
+		allow, fail bool
+	}{
+		{`{"result":true}`, 200, true, false}, {`{"result":false}`, 200, false, false},
+		{`{"result":{"allow":true}}`, 200, true, false}, {`{"result":{"allow":false}}`, 200, false, false},
+		{`{"result":true}`, 503, false, true}, {`{"result":true}`, 403, false, true},
+		{`{}`, 200, false, true}, {`{"error":"unavailable"}`, 200, false, true}, {`{"result":null}`, 200, false, true},
+		{`{"result":{}}`, 200, false, true}, {`{"result":{"allow":null}}`, 200, false, true}, {`{"result":"true"}`, 200, false, true},
+		{`{"result":true} garbage`, 200, false, true},
+	} {
+		t.Run(tc.body+http.StatusText(tc.status), func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(tc.status); io.WriteString(w, tc.body) }))
+			defer server.Close()
+			u, err := net.ParseHTTPURL(server.URL)
+			if err != nil {
+				t.Fatal(err)
+			}
+			plugin := New(Args{URL: u, CloseRespFn: func(r io.ReadCloser) { r.Close() }})
+			allowed, err := plugin.IsAllowedContext(t.Context(), policy.Args{})
+			if allowed != tc.allow || (err != nil) != tc.fail {
+				t.Fatalf("allow=%v err=%v", allowed, err)
+			}
+			ctx, cancel := context.WithCancel(t.Context())
+			cancel()
+			if allowed, err := plugin.IsAllowedContext(ctx, policy.Args{}); allowed || err == nil {
+				t.Fatal("canceled evaluation did not stop")
+			}
+		})
+	}
+}