From bdd17b0f474debe2802d943d007cb93d4c0d76ec Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 7 Nov 2026 00:42:02 +0800 Subject: [PATCH] fix(batch): pause transient authorization failures with owned retries Signed-off-by: Feng Ruohang --- cmd/batch-authz-review6_test.go | 27 +- cmd/batch-authz.go | 136 ++++++-- cmd/batch-expire-purge-authz_test.go | 6 +- cmd/batch-expire.go | 31 +- cmd/batch-handlers.go | 174 +++++++---- cmd/batch-handlers_gen.go | 35 ++- cmd/batch-progress.go | 2 + cmd/batch-retry.go | 194 ++++++++++++ cmd/batch-retry_test.go | 312 +++++++++++++++++++ cmd/batch-rotate.go | 26 +- cmd/batch-submitter-authz_test.go | 8 +- cmd/iam-store.go | 4 +- internal/config/policy/plugin/config.go | 61 ++-- internal/config/policy/plugin/config_test.go | 48 +++ 14 files changed, 911 insertions(+), 153 deletions(-) create mode 100644 cmd/batch-retry.go create mode 100644 cmd/batch-retry_test.go create mode 100644 internal/config/policy/plugin/config_test.go diff --git a/cmd/batch-authz-review6_test.go b/cmd/batch-authz-review6_test.go index 63303b733..90259f3cb 100644 --- a/cmd/batch-authz-review6_test.go +++ b/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 diff --git a/cmd/batch-authz.go b/cmd/batch-authz.go index 192fbb473..e5e2ccc30 100644 --- a/cmd/batch-authz.go +++ b/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 } diff --git a/cmd/batch-expire-purge-authz_test.go b/cmd/batch-expire-purge-authz_test.go index 6c8b14dfd..6e3dfdedd 100644 --- a/cmd/batch-expire-purge-authz_test.go +++ b/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)) diff --git a/cmd/batch-expire.go b/cmd/batch-expire.go index 847d7154f..542bd39ad 100644 --- a/cmd/batch-expire.go +++ b/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() } diff --git a/cmd/batch-handlers.go b/cmd/batch-handlers.go index 28985f728..0e0150b10 100644 --- a/cmd/batch-handlers.go +++ b/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 } diff --git a/cmd/batch-handlers_gen.go b/cmd/batch-handlers_gen.go index d0e1d7086..f1eed6519 100644 --- a/cmd/batch-handlers_gen.go +++ b/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 } diff --git a/cmd/batch-progress.go b/cmd/batch-progress.go index 968e8e0ef..a8c4eb824 100644 --- a/cmd/batch-progress.go +++ b/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, } } diff --git a/cmd/batch-retry.go b/cmd/batch-retry.go new file mode 100644 index 000000000..97bc71a5b --- /dev/null +++ b/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<= 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") } diff --git a/internal/config/policy/plugin/config_test.go b/internal/config/policy/plugin/config_test.go new file mode 100644 index 000000000..a32b9a883 --- /dev/null +++ b/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") + } + }) + } +}