fix(batch): authorize batch jobs as their submitter

A batch job ran with the server's object access, whoever submitted it,
and acted on versions that Object Lock protects.

Submitter: StartBatchJob binds the authenticated credential to the job
(type, access key, parent, identity revision and a keyed fingerprint).
Submission, every start and every object resolve the binding against
IAM and evaluate the action with the submitter's policies and session
policy. A credential that is gone, disabled, expired or changed fails
the job. A policy statement that would need request context the job
does not have refuses the action. With an authorization plugin, the
plugin and the session policy must both allow. A refusal is persisted
as the job's failure reason and ends the job. A job stored without a
binding fails and has to be submitted again; a definition holds one
job type.

Expire deletes each older version on its own, checking retention and
legal hold and authorizing it under the delete's lock; a locked version
is left and counted. Until the exact-key purge is integrated, a
retainVersions: 0 key is refused when a version the walk saw is locked.
Pull writes refuse to replace a locked destination version, use the
destination's versioning, validate copied retention and apply default
retention. Rotation leaves locked versions as they are.

The status reports versionsSkippedLocked, keysRefused and the failure
reason. Expire and rotation persist their cursor together with these
counts, so a resumed job counts each version once. Describe returns a
redacted copy of the job.

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

cmd/batch-authz.go

@@ -0,0 +1,742 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"crypto/subtle"
+	"encoding/base64"
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/minio/minio-go/v7/pkg/tags"
+	"github.com/minio/minio/internal/arn"
+	"github.com/minio/minio/internal/auth"
+	"github.com/pgsty/silo-pkg/v3/policy"
+	"github.com/pgsty/silo-pkg/v3/policy/condition"
+)
+
+// A batch job acts on objects long after the request that submitted it has
+// gone. It therefore records who submitted it and acts only as that
+// principal:
+//
+//   - StartBatchJob binds the authenticated credential to the job
+//     (batchJobSubmitter). Nothing in the submitted definition names it.
+//   - Every start of the job, and every object it touches, resolves the
+//     binding against IAM again. A credential that is gone, disabled,
+//     expired, revoked or edited since submission fails the job; a reused
+//     access key does not match the binding.
+//   - Each action is evaluated with the submitter's policies and session
+//     policy but without a request. A policy statement that would need
+//     request context the job does not have (source address, transport,
+//     headers and so on) refuses the action instead of being skipped.
+//
+// A job without a binding was queued by an earlier release. It is never
+// authorized, whoever submitted it, and has to be submitted again.
+
+//go:generate msgp -file $GOFILE -unexported
+
+// batchJobSubmitter is the server-assigned identity of the credential that
+// submitted a batch job.
+type batchJobSubmitter struct {
+	Version   int    `msg:"v"`
+	Type      string `msg:"t"`
+	AccessKey string `msg:"ak"`
+	// ParentUser is the parent of a derived credential.
+	ParentUser string `msg:"pu"`
+	// UpdatedAt and RevokedBefore are the revision of the identity record;
+	// any edit of the identity changes them. They are unset for root.
+	UpdatedAt     time.Time `msg:"ua"`
+	RevokedBefore time.Time `msg:"rb"`
+	// Binding is a keyed fingerprint of the credential itself, see
+	// batchSubmitterBinding.
+	Binding []byte `msg:"bd"`
+}
+
+const batchJobSubmitterVersion = 1
+
+// Submitter types. A credential is classified in the order IsAllowed
+// evaluates it.
+const (
+	batchSubmitterRoot      = "root"
+	batchSubmitterUser      = "user"
+	batchSubmitterTemporary = "temporary"
+	batchSubmitterService   = "service-account"
+)
+
+//msgp:ignore batchJobAuthzError
+
+// batchJobAuthzError refuses a batch job, or one of its actions, for its
+// submitter. It ends the job: it is persisted as the job's failure and is
+// never retried.
+type batchJobAuthzError struct {
+	reason string
+}
+
+func (e batchJobAuthzError) Error() string {
+	return e.reason
+}
+
+func isBatchJobAuthzError(err error) bool {
+	var e batchJobAuthzError
+	return errors.As(err, &e)
+}
+
+var (
+	errBatchJobResubmit = batchJobAuthzError{reason: "the batch job does not record who submitted it; submit it again"}
+	errBatchJobChanged  = batchJobAuthzError{reason: "the credential that submitted the batch job has changed since submission; submit it again"}
+	errBatchJobInvalid  = batchJobAuthzError{reason: "the credential that submitted the batch job is no longer valid"}
+)
+
+func batchSubmitterType(cred auth.Credentials) string {
+	switch {
+	case cred.IsTemp():
+		return batchSubmitterTemporary
+	case cred.IsServiceAccount():
+		return batchSubmitterService
+	default:
+		return batchSubmitterUser
+	}
+}
+
+// batchSubmitterBinding is a fingerprint of one credential keyed with its own
+// secret. It changes when the secret, the session token or the parent
+// changes, and a new credential that reuses the access key does not have it.
+func batchSubmitterBinding(typ string, cred auth.Credentials) []byte {
+	mac := hmac.New(sha256.New, []byte(cred.SecretKey))
+	for _, s := range []string{"silo batch-job submitter v1", typ, cred.AccessKey, cred.ParentUser, cred.SessionToken} {
+		var n [8]byte
+		binary.BigEndian.PutUint64(n[:], uint64(len(s)))
+		mac.Write(n[:])
+		mac.Write([]byte(s))
+	}
+	return mac.Sum(nil)
+}
+
+// identityRecord returns the stored identity of a non-root access key.
+func (sys *IAMSys) identityRecord(accessKey string) (UserIdentity, bool, error) {
+	if !sys.Initialized() {
+		return UserIdentity{}, false, errServerNotInitialized
+	}
+	u, ok := sys.store.GetUser(accessKey)
+	return u, ok, nil
+}
+
+// newBatchJobSubmitter binds the credential that authenticated a
+// StartBatchJob request.
+func newBatchJobSubmitter(cred auth.Credentials) (*batchJobSubmitter, error) {
+	root := globalActiveCred
+	if cred.AccessKey == "" {
+		return nil, errBatchJobInvalid
+	}
+	if cred.AccessKey == root.AccessKey {
+		if subtle.ConstantTimeCompare([]byte(cred.SecretKey), []byte(root.SecretKey)) != 1 {
+			return nil, errBatchJobChanged
+		}
+		return &batchJobSubmitter{
+			Version:   batchJobSubmitterVersion,
+			Type:      batchSubmitterRoot,
+			AccessKey: root.AccessKey,
+			Binding:   batchSubmitterBinding(batchSubmitterRoot, root),
+		}, nil
+	}
+	u, ok, err := globalIAMSys.identityRecord(cred.AccessKey)
+	if err != nil {
+		return nil, err
+	}
+	// The record must be the one the request was authenticated with.
+	if !ok || u.Credentials.AccessKey != cred.AccessKey || u.Credentials.ParentUser != cred.ParentUser ||
+		subtle.ConstantTimeCompare([]byte(u.Credentials.SecretKey), []byte(cred.SecretKey)) != 1 ||
+		subtle.ConstantTimeCompare([]byte(u.Credentials.SessionToken), []byte(cred.SessionToken)) != 1 {
+		return nil, errBatchJobChanged
+	}
+	typ := batchSubmitterType(u.Credentials)
+	return &batchJobSubmitter{
+		Version:       batchJobSubmitterVersion,
+		Type:          typ,
+		AccessKey:     u.Credentials.AccessKey,
+		ParentUser:    u.Credentials.ParentUser,
+		UpdatedAt:     u.UpdatedAt,
+		RevokedBefore: u.RevokedBefore,
+		Binding:       batchSubmitterBinding(typ, u.Credentials),
+	}, nil
+}
+
+//msgp:ignore batchJobPrincipal batchAuthzRequest
+
+// batchJobPrincipal is a submitter resolved against current IAM state.
+type batchJobPrincipal struct {
+	typ        string
+	accessKey  string
+	parentUser string
+	groups     []string
+	claims     map[string]any
+	owner      bool
+	temp       bool // evaluated as a temporary credential
+	service    bool // evaluated as a service account
+}
+
+// principal resolves the binding against current IAM state. It fails if the
+// credential no longer exists, is no longer valid, or is no longer the one
+// that was bound.
+func (s *batchJobSubmitter) principal(ctx context.Context) (batchJobPrincipal, error) {
+	if s == nil || s.AccessKey == "" {
+		return batchJobPrincipal{}, errBatchJobResubmit
+	}
+	if s.Version != batchJobSubmitterVersion {
+		return batchJobPrincipal{}, batchJobAuthzError{reason: fmt.Sprintf("unsupported batch job submitter version %d; submit the job again", s.Version)}
+	}
+	root := globalActiveCred
+	if s.Type == batchSubmitterRoot {
+		if s.AccessKey != root.AccessKey || !hmac.Equal(s.Binding, batchSubmitterBinding(batchSubmitterRoot, root)) {
+			return batchJobPrincipal{}, errBatchJobChanged
+		}
+		if !globalAPIConfig.permitRootAccess() {
+			return batchJobPrincipal{}, errBatchJobInvalid
+		}
+		return batchJobPrincipal{typ: batchSubmitterRoot, accessKey: root.AccessKey, owner: true}, nil
+	}
+	switch s.Type {
+	case batchSubmitterUser, batchSubmitterTemporary, batchSubmitterService:
+	default:
+		return batchJobPrincipal{}, errBatchJobResubmit
+	}
+	if s.AccessKey == root.AccessKey {
+		return batchJobPrincipal{}, errBatchJobChanged
+	}
+	u, ok, err := globalIAMSys.CheckKey(ctx, s.AccessKey)
+	if err != nil {
+		return batchJobPrincipal{}, err
+	}
+	if !ok {
+		// Deleted, disabled, expired or revoked.
+		return batchJobPrincipal{}, errBatchJobInvalid
+	}
+	cred := u.Credentials
+	typ := batchSubmitterType(cred)
+	if typ != s.Type || cred.AccessKey != s.AccessKey || cred.ParentUser != s.ParentUser ||
+		!u.UpdatedAt.Equal(s.UpdatedAt) || !u.RevokedBefore.Equal(s.RevokedBefore) ||
+		!hmac.Equal(s.Binding, batchSubmitterBinding(typ, cred)) {
+		return batchJobPrincipal{}, errBatchJobChanged
+	}
+	if cred.IsExpired() || cred.Status == auth.AccountOff {
+		return batchJobPrincipal{}, errBatchJobInvalid
+	}
+	p := batchJobPrincipal{
+		typ:        typ,
+		accessKey:  cred.AccessKey,
+		parentUser: cred.ParentUser,
+		groups:     cred.Groups,
+		claims:     map[string]any{},
+		temp:       cred.IsTemp(),
+		service:    !cred.IsTemp() && cred.IsServiceAccount(),
+	}
+	if typ != batchSubmitterUser {
+		// Verify the token as a request would, with the current signing keys.
+		jwtClaims, err := extractJWTClaims(u)
+		if err != nil {
+			return batchJobPrincipal{}, errBatchJobInvalid
+		}
+		p.claims = jwtClaims.Map()
+		if err := decodeBatchSessionPolicy(p.claims); err != nil {
+			return batchJobPrincipal{}, errBatchJobInvalid
+		}
+	}
+	p.owner = cred.ParentUser == root.AccessKey && cred.AccessKey != siteReplicatorSvcAcc
+	if p.owner && !globalAPIConfig.permitRootAccess() {
+		return batchJobPrincipal{}, errBatchJobInvalid
+	}
+	if _, ok := p.claims[policy.SessionPolicyName]; ok {
+		p.owner = false
+	}
+	return p, nil
+}
+
+// decodeBatchSessionPolicy decodes a session policy claim for evaluation.
+// A request decodes it while verifying its token, except when an
+// authorization plugin is configured; a batch job decodes it always, so the
+// session policy also limits what the plugin allows.
+func decodeBatchSessionPolicy(claims map[string]any) error {
+	sp, ok := claims[policy.SessionPolicyName]
+	if !ok {
+		return nil
+	}
+	s, ok := sp.(string)
+	if !ok {
+		return errAuthentication
+	}
+	b, err := base64.StdEncoding.DecodeString(s)
+	if err != nil {
+		return errAuthentication
+	}
+	claims[sessionPolicyNameExtracted] = string(b)
+	return nil
+}
+
+// batchAuthzRequest is one action a batch job is about to take.
+type batchAuthzRequest struct {
+	action policy.Action
+	bucket string
+	object string // empty for a bucket action
+	// versionID is the version an object version action addresses.
+	versionID string
+	// tags are the current tags of the object version acted on.
+	tags string
+	// tagsUnknown is set when the job does not read the version it acts
+	// on, so s3:ExistingObjectTag is context it does not have.
+	tagsUnknown bool
+	// values are request values the job supplies itself, such as the
+	// listing prefix or the encryption and retention it writes, keyed by
+	// condition key name.
+	values map[string][]string
+}
+
+// batchAuthzMarker marks the condition values of a batch job for an
+// authorization plugin. Requests cannot set s3:authType.
+const batchAuthzMarker = "batch-job"
+
+// Condition keys a batch job always provides, as a request would: identity,
+// time, the addressed version and the stored tags. Keys a request takes from
+// itself are provided only when the job sets them (see batchAuthzRequest
+// values); every other key is request context the job does not have.
+var batchServerConditionKeys = map[string]bool{
+	condition.AWSCurrentTime.ToKey().String():    true,
+	condition.AWSEpochTime.ToKey().String():      true,
+	condition.AWSPrincipalType.ToKey().String():  true,
+	condition.AWSUserID.ToKey().String():         true,
+	condition.AWSUsername.ToKey().String():       true,
+	condition.AWSGroups.ToKey().String():         true,
+	condition.S3VersionID.ToKey().String():       true,
+	condition.ExistingObjectTag.ToKey().String(): true,
+}
+
+// Request keys a job may supply, by the name they have in condition values.
+var batchRequestConditionKeys = map[string]condition.KeyName{
+	condition.S3Prefix.Name():                              condition.S3Prefix,
+	condition.S3XAmzServerSideEncryption.Name():            condition.S3XAmzServerSideEncryption,
+	condition.S3XAmzServerSideEncryptionAwsKmsKeyID.Name(): condition.S3XAmzServerSideEncryptionAwsKmsKeyID,
+	condition.S3ObjectLockMode.Name():                      condition.S3ObjectLockMode,
+	condition.S3ObjectLockRetainUntilDate.Name():           condition.S3ObjectLockRetainUntilDate,
+	condition.S3ObjectLockLegalHold.Name():                 condition.S3ObjectLockLegalHold,
+	condition.RequestObjectTagKeys.Name():                  condition.RequestObjectTagKeys,
+}
+
+// suppliedConditionKeys returns the condition keys whose values req provides.
+func (req batchAuthzRequest) suppliedConditionKeys() map[string]bool {
+	supplied := make(map[string]bool, len(batchServerConditionKeys)+len(req.values))
+	for k := range batchServerConditionKeys {
+		supplied[k] = true
+	}
+	if req.tagsUnknown {
+		delete(supplied, condition.ExistingObjectTag.ToKey().String())
+	}
+	for name := range req.values {
+		if key, ok := batchRequestConditionKeys[name]; ok {
+			supplied[key.ToKey().String()] = true
+		}
+		if strings.HasPrefix(name, condition.RequestObjectTag.Name()+"/") {
+			supplied[condition.RequestObjectTag.ToKey().String()] = true
+		}
+	}
+	return supplied
+}
+
+func batchKeySupplied(supplied map[string]bool, name string) bool {
+	if i := strings.IndexByte(name, '/'); i >= 0 {
+		name = name[:i]
+	}
+	if strings.HasPrefix(name, "jwt:") || strings.HasPrefix(name, "ldap:") {
+		// Claims of the verified token, as a request has them.
+		return true
+	}
+	return supplied[name]
+}
+
+// conditionValues are the values a request with the same identity would
+// carry, without any value taken from the request itself.
+func (p batchJobPrincipal) conditionValues(req batchAuthzRequest) map[string][]string {
+	now := UTCNow()
+	username := p.accessKey
+	if p.temp || p.service {
+		username = p.parentUser
+	}
+	principalType := "User"
+	if len(p.claims) > 0 {
+		principalType = "AssumedRole"
+	}
+	if username == globalActiveCred.AccessKey {
+		principalType = "Account"
+	}
+	// Identity and time, then the claims over them, as a request has them
+	// (an LDAP session's short username replaces its parent DN).
+	values := map[string][]string{
+		"CurrentTime":   {now.Format(time.RFC3339)},
+		"EpochTime":     {strconv.FormatInt(now.Unix(), 10)},
+		"principaltype": {principalType},
+		"userid":        {username},
+		"username":      {username},
+	}
+	for k, v := range p.claims {
+		if s, ok := v.(string); ok {
+			values[strings.ToLower(strings.TrimPrefix(k, "ldap"))] = []string{s}
+		}
+	}
+	// No claim stands for a value of the object or of the job, which
+	// follow and always win.
+	for k := range values {
+		if k == "versionid" || k == "prefix" || k == "authtype" ||
+			strings.HasPrefix(k, "existingobjecttag/") || strings.HasPrefix(k, "requestobjecttag") ||
+			strings.HasPrefix(k, "x-amz-") || strings.HasPrefix(k, "object-lock-") {
+			delete(values, k)
+		}
+	}
+	// For an authorization plugin: this is a batch job, not a request.
+	values["authType"] = []string{batchAuthzMarker}
+	if req.versionID != "" {
+		values["versionid"] = []string{req.versionID}
+	}
+	if req.tags != "" {
+		if t, err := tags.ParseObjectTags(req.tags); err == nil {
+			for k, v := range t.ToMap() {
+				values[pathJoin("ExistingObjectTag", k)] = []string{v}
+			}
+		}
+	}
+	for k, v := range req.values {
+		values[k] = v
+	}
+	if grpsVal, ok := p.claims["groups"]; ok {
+		if grpsIs, ok := grpsVal.([]any); ok {
+			grps := []string{}
+			for _, gI := range grpsIs {
+				if g, ok := gI.(string); ok {
+					grps = append(grps, g)
+				}
+			}
+			if len(grps) > 0 {
+				values["groups"] = grps
+			}
+		}
+	}
+	if _, ok := values["groups"]; !ok && len(p.groups) > 0 {
+		values["groups"] = p.groups
+	}
+	return values
+}
+
+// authorize evaluates one action of the job as its submitter.
+func (p batchJobPrincipal) authorize(req batchAuthzRequest) error {
+	args := policy.Args{
+		AccountName:     p.accessKey,
+		Groups:          p.groups,
+		Action:          req.action,
+		BucketName:      req.bucket,
+		ObjectName:      req.object,
+		ConditionValues: p.conditionValues(req),
+		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
+	}
+	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)
+		if err != nil {
+			authZLogIf(GlobalContext, err)
+		}
+		if !ok {
+			return denied
+		}
+		// The plugin does not see a session policy; it only limits further.
+		if has, allowed := p.sessionPolicyDecision(args); has && !allowed {
+			return denied
+		}
+		return nil
+	}
+	if !globalIAMSys.isAllowedLocally(args) {
+		return denied
+	}
+	return nil
+}
+
+func batchDeniedReason(req batchAuthzRequest) string {
+	resource := req.bucket
+	if req.object != "" {
+		resource = pathJoin(req.bucket, req.object)
+	}
+	if req.versionID != "" {
+		resource += " (versionId " + req.versionID + ")"
+	}
+	return fmt.Sprintf("the batch job submitter is not allowed %s on %s", req.action, resource)
+}
+
+// sessionPolicyDecision evaluates the session policy the way IsAllowed would
+// for this credential.
+func (p batchJobPrincipal) sessionPolicyDecision(args policy.Args) (bool, bool) {
+	switch {
+	case p.temp:
+		return isAllowedBySessionPolicy(args)
+	case p.service:
+		if p.claims[iamPolicyClaimNameSA()] == inheritedPolicyType {
+			return false, false
+		}
+		return isAllowedBySessionPolicyForServiceAccount(args)
+	}
+	return false, false
+}
+
+// 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) {
+	var out []policy.Policy
+	if withMapped && args.IsOwner {
+		// IsAllowed does not evaluate any policy for the owner.
+		return nil, nil
+	}
+	if withMapped {
+		var names []string
+		ownerDerived := (p.temp || p.service) && p.parentUser == globalActiveCred.AccessKey
+		switch {
+		case p.typ == batchSubmitterRoot || ownerDerived:
+		case p.temp || p.service:
+			if roleArn := args.GetRoleArn(); roleArn != "" {
+				a, err := arn.Parse(roleArn)
+				if err != nil {
+					return nil, err
+				}
+				names = append(names, newMappedPolicy(sys.rolesMap[a]).toSlice()...)
+			} else {
+				mapped, err := sys.PolicyDBGet(p.parentUser, args.Groups...)
+				if err != nil {
+					return nil, err
+				}
+				names = append(names, mapped...)
+				// The claimed policies apply only when nothing is mapped.
+				if len(mapped) == 0 {
+					if claimed, ok := policy.GetPoliciesFromClaims(args.Claims, iamPolicyClaimNameOpenID()); ok {
+						names = append(names, claimed.ToSlice()...)
+					}
+				}
+			}
+		default:
+			mapped, err := sys.PolicyDBGet(p.accessKey, args.Groups...)
+			if err != nil {
+				return nil, err
+			}
+			names = append(names, mapped...)
+		}
+		if len(names) > 0 {
+			out = append(out, sys.GetCombinedPolicy(names...))
+		}
+	}
+	if p.temp || p.service {
+		if p.service && p.claims[iamPolicyClaimNameSA()] == inheritedPolicyType {
+			return out, nil
+		}
+		if sp, ok := p.claims[sessionPolicyNameExtracted]; ok {
+			s, ok := sp.(string)
+			if !ok {
+				return nil, errAuthentication
+			}
+			parsed, err := policy.ParseConfig(strings.NewReader(s))
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, *parsed)
+		}
+	}
+	return out, nil
+}
+
+// batchStatementNeedsRequest reports whether a statement that applies to
+// args depends on a condition key that the job does not supply. Applying
+// such a statement without its value could grant (Allow) or drop (Deny)
+// what a request with that context would not, so the job refuses the action.
+// Resources are matched without conditions; a statement is taken to apply
+// whenever it might.
+func batchStatementNeedsRequest(st policy.Statement, args policy.Args, supplied map[string]bool) bool {
+	if (!st.Actions.Match(args.Action) && !st.Actions.IsEmpty()) || st.NotActions.Match(args.Action) {
+		return false
+	}
+	// Policy variables in resources or condition values.
+	if data, err := json.Marshal(st); err == nil {
+		s := string(data)
+		for {
+			i := strings.Index(s, "${")
+			if i < 0 {
+				break
+			}
+			s = s[i+2:]
+			j := strings.IndexByte(s, '}')
+			if j < 0 {
+				break
+			}
+			if !batchKeySupplied(supplied, s[:j]) {
+				return true
+			}
+			s = s[j+1:]
+		}
+	}
+	if !batchStatementMayApply(st, args) {
+		return false
+	}
+	for key := range st.Conditions.Keys() {
+		if !batchKeySupplied(supplied, key.String()) {
+			return true
+		}
+	}
+	return false
+}
+
+// batchStatementMayApply matches the resources of a statement, in both forms
+// a bucket request can take. A statement whose resources the evaluator
+// translates (S3 Tables and vectors actions cover S3 data actions) is taken
+// to apply.
+func batchStatementMayApply(st policy.Statement, args policy.Args) bool {
+	for _, actions := range []policy.ActionSet{st.Actions, st.NotActions} {
+		for action := range actions {
+			if policy.TableAction(action).IsValid() || policy.VectorsAction(action).IsValid() {
+				return true
+			}
+		}
+	}
+	for _, resources := range []policy.ResourceSet{st.Resources, st.NotResources} {
+		for r := range resources {
+			if r.Type != policy.ResourceARNS3 && r.Type != policy.ResourceARNAll {
+				return true
+			}
+		}
+	}
+	resources := []string{args.BucketName + "/" + strings.TrimPrefix(args.ObjectName, "/")}
+	if args.ObjectName == "" {
+		resources = []string{args.BucketName, args.BucketName + "/"}
+	}
+	for _, resource := range resources {
+		if len(st.Resources) > 0 && !st.Resources.Match(resource, args.ConditionValues) {
+			continue
+		}
+		if len(st.NotResources) > 0 && st.NotResources.Match(resource, args.ConditionValues) {
+			continue
+		}
+		return true
+	}
+	return false
+}
+
+// authorize resolves the submitter and evaluates each request.
+func (s *batchJobSubmitter) authorize(ctx context.Context, reqs ...batchAuthzRequest) error {
+	p, err := s.principal(ctx)
+	if err != nil {
+		return err
+	}
+	for _, req := range reqs {
+		if err := p.authorize(req); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// startAuthzRequests are the bucket actions a job needs before it acts on any
+// object: the listings it walks. Object actions are authorized per object.
+func (j BatchJobRequest) startAuthzRequests() []batchAuthzRequest {
+	list := func(bucket string, prefixes []string) []batchAuthzRequest {
+		if len(prefixes) == 0 {
+			prefixes = []string{""}
+		}
+		var reqs []batchAuthzRequest
+		for _, prefix := range prefixes {
+			values := map[string][]string{condition.S3Prefix.Name(): {prefix}}
+			reqs = append(reqs,
+				batchAuthzRequest{action: policy.ListBucketAction, bucket: bucket, values: values},
+				batchAuthzRequest{action: policy.ListBucketVersionsAction, bucket: bucket, values: values},
+			)
+		}
+		return reqs
+	}
+	// The same order as Type() and the job pool.
+	switch {
+	case j.Replicate != nil:
+		if !j.Replicate.RemoteToLocal() {
+			return list(j.Replicate.Source.Bucket, j.Replicate.Source.Prefix.F())
+		}
+	case j.KeyRotate != nil:
+		return list(j.KeyRotate.Bucket, []string{j.KeyRotate.Prefix})
+	case j.Expire != nil:
+		return list(j.Expire.Bucket, batchCanonicalPrefixes(j.Expire.Prefix.F()))
+	}
+	return nil
+}
+
+// jobSections counts the job types a definition holds. A job is one of them.
+func (j BatchJobRequest) jobSections() int {
+	n := 0
+	for _, set := range []bool{j.Replicate != nil, j.KeyRotate != nil, j.Expire != nil} {
+		if set {
+			n++
+		}
+	}
+	return n
+}
+
+var errBatchJobSections = batchJobAuthzError{reason: "a batch job must define exactly one job type"}
+
+// refuse ends the job for its submitter.
+func (ri *batchJobInfo) refuse(err error) {
+	ri.mu.Lock()
+	defer ri.mu.Unlock()
+	if ri.FailureReason == "" {
+		ri.FailureReason = err.Error()
+	}
+	ri.Complete = false
+	ri.Failed = true
+}
+
+func (ri *batchJobInfo) refused() bool {
+	ri.mu.RLock()
+	defer ri.mu.RUnlock()
+	return ri.FailureReason != ""
+}
+
+// startAuthorized checks the submitter of a job that is about to run. It
+// returns done for a job that must not run now: one refused before, which
+// stays refused, or one refused now, which is persisted as failed first.
+// An error that is not a decision leaves the job to a later start.
+func (ri *batchJobInfo) startAuthorized(ctx context.Context, api ObjectLayer, job BatchJobRequest) (done bool, err error) {
+	if !ri.refused() {
+		err := job.Submitter.authorize(ctx, job.startAuthzRequests()...)
+		if err == nil && job.jobSections() != 1 {
+			err = errBatchJobSections
+		}
+		if err == nil {
+			return false, nil
+		}
+		if !isBatchJobAuthzError(err) {
+			return true, err
+		}
+		ri.refuse(err)
+		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)
+}

cmd/batch-authz_gen.go

@@ -0,0 +1,260 @@
+// Code generated by github.com/tinylib/msgp DO NOT EDIT.
+
+package cmd
+
+import (
+	"github.com/tinylib/msgp/msgp"
+)
+
+// DecodeMsg implements msgp.Decodable
+func (z *batchJobSubmitter) DecodeMsg(dc *msgp.Reader) (err error) {
+	var field []byte
+	_ = field
+	var zb0001 uint32
+	zb0001, err = dc.ReadMapHeader()
+	if err != nil {
+		err = msgp.WrapError(err)
+		return
+	}
+	for zb0001 > 0 {
+		zb0001--
+		field, err = dc.ReadMapKeyPtr()
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		switch msgp.UnsafeString(field) {
+		case "v":
+			z.Version, err = dc.ReadInt()
+			if err != nil {
+				err = msgp.WrapError(err, "Version")
+				return
+			}
+		case "t":
+			z.Type, err = dc.ReadString()
+			if err != nil {
+				err = msgp.WrapError(err, "Type")
+				return
+			}
+		case "ak":
+			z.AccessKey, err = dc.ReadString()
+			if err != nil {
+				err = msgp.WrapError(err, "AccessKey")
+				return
+			}
+		case "pu":
+			z.ParentUser, err = dc.ReadString()
+			if err != nil {
+				err = msgp.WrapError(err, "ParentUser")
+				return
+			}
+		case "ua":
+			z.UpdatedAt, err = dc.ReadTime()
+			if err != nil {
+				err = msgp.WrapError(err, "UpdatedAt")
+				return
+			}
+		case "rb":
+			z.RevokedBefore, err = dc.ReadTime()
+			if err != nil {
+				err = msgp.WrapError(err, "RevokedBefore")
+				return
+			}
+		case "bd":
+			z.Binding, err = dc.ReadBytes(z.Binding)
+			if err != nil {
+				err = msgp.WrapError(err, "Binding")
+				return
+			}
+		default:
+			err = dc.Skip()
+			if err != nil {
+				err = msgp.WrapError(err)
+				return
+			}
+		}
+	}
+	return
+}
+
+// EncodeMsg implements msgp.Encodable
+func (z *batchJobSubmitter) EncodeMsg(en *msgp.Writer) (err error) {
+	// map header, size 7
+	// write "v"
+	err = en.Append(0x87, 0xa1, 0x76)
+	if err != nil {
+		return
+	}
+	err = en.WriteInt(z.Version)
+	if err != nil {
+		err = msgp.WrapError(err, "Version")
+		return
+	}
+	// write "t"
+	err = en.Append(0xa1, 0x74)
+	if err != nil {
+		return
+	}
+	err = en.WriteString(z.Type)
+	if err != nil {
+		err = msgp.WrapError(err, "Type")
+		return
+	}
+	// write "ak"
+	err = en.Append(0xa2, 0x61, 0x6b)
+	if err != nil {
+		return
+	}
+	err = en.WriteString(z.AccessKey)
+	if err != nil {
+		err = msgp.WrapError(err, "AccessKey")
+		return
+	}
+	// write "pu"
+	err = en.Append(0xa2, 0x70, 0x75)
+	if err != nil {
+		return
+	}
+	err = en.WriteString(z.ParentUser)
+	if err != nil {
+		err = msgp.WrapError(err, "ParentUser")
+		return
+	}
+	// write "ua"
+	err = en.Append(0xa2, 0x75, 0x61)
+	if err != nil {
+		return
+	}
+	err = en.WriteTime(z.UpdatedAt)
+	if err != nil {
+		err = msgp.WrapError(err, "UpdatedAt")
+		return
+	}
+	// write "rb"
+	err = en.Append(0xa2, 0x72, 0x62)
+	if err != nil {
+		return
+	}
+	err = en.WriteTime(z.RevokedBefore)
+	if err != nil {
+		err = msgp.WrapError(err, "RevokedBefore")
+		return
+	}
+	// write "bd"
+	err = en.Append(0xa2, 0x62, 0x64)
+	if err != nil {
+		return
+	}
+	err = en.WriteBytes(z.Binding)
+	if err != nil {
+		err = msgp.WrapError(err, "Binding")
+		return
+	}
+	return
+}
+
+// MarshalMsg implements msgp.Marshaler
+func (z *batchJobSubmitter) MarshalMsg(b []byte) (o []byte, err error) {
+	o = msgp.Require(b, z.Msgsize())
+	// map header, size 7
+	// string "v"
+	o = append(o, 0x87, 0xa1, 0x76)
+	o = msgp.AppendInt(o, z.Version)
+	// string "t"
+	o = append(o, 0xa1, 0x74)
+	o = msgp.AppendString(o, z.Type)
+	// string "ak"
+	o = append(o, 0xa2, 0x61, 0x6b)
+	o = msgp.AppendString(o, z.AccessKey)
+	// string "pu"
+	o = append(o, 0xa2, 0x70, 0x75)
+	o = msgp.AppendString(o, z.ParentUser)
+	// string "ua"
+	o = append(o, 0xa2, 0x75, 0x61)
+	o = msgp.AppendTime(o, z.UpdatedAt)
+	// string "rb"
+	o = append(o, 0xa2, 0x72, 0x62)
+	o = msgp.AppendTime(o, z.RevokedBefore)
+	// string "bd"
+	o = append(o, 0xa2, 0x62, 0x64)
+	o = msgp.AppendBytes(o, z.Binding)
+	return
+}
+
+// UnmarshalMsg implements msgp.Unmarshaler
+func (z *batchJobSubmitter) UnmarshalMsg(bts []byte) (o []byte, err error) {
+	var field []byte
+	_ = field
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadMapHeaderBytes(bts)
+	if err != nil {
+		err = msgp.WrapError(err)
+		return
+	}
+	for zb0001 > 0 {
+		zb0001--
+		field, bts, err = msgp.ReadMapKeyZC(bts)
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		switch msgp.UnsafeString(field) {
+		case "v":
+			z.Version, bts, err = msgp.ReadIntBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Version")
+				return
+			}
+		case "t":
+			z.Type, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Type")
+				return
+			}
+		case "ak":
+			z.AccessKey, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "AccessKey")
+				return
+			}
+		case "pu":
+			z.ParentUser, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "ParentUser")
+				return
+			}
+		case "ua":
+			z.UpdatedAt, bts, err = msgp.ReadTimeBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "UpdatedAt")
+				return
+			}
+		case "rb":
+			z.RevokedBefore, bts, err = msgp.ReadTimeBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "RevokedBefore")
+				return
+			}
+		case "bd":
+			z.Binding, bts, err = msgp.ReadBytesBytes(bts, z.Binding)
+			if err != nil {
+				err = msgp.WrapError(err, "Binding")
+				return
+			}
+		default:
+			bts, err = msgp.Skip(bts)
+			if err != nil {
+				err = msgp.WrapError(err)
+				return
+			}
+		}
+	}
+	o = bts
+	return
+}
+
+// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
+func (z *batchJobSubmitter) Msgsize() (s int) {
+	s = 1 + 2 + msgp.IntSize + 2 + msgp.StringPrefixSize + len(z.Type) + 3 + msgp.StringPrefixSize + len(z.AccessKey) + 3 + msgp.StringPrefixSize + len(z.ParentUser) + 3 + msgp.TimeSize + 3 + msgp.TimeSize + 3 + msgp.BytesPrefixSize + len(z.Binding)
+	return
+}

cmd/batch-authz_gen_test.go

@@ -0,0 +1,123 @@
+// Code generated by github.com/tinylib/msgp DO NOT EDIT.
+
+package cmd
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/tinylib/msgp/msgp"
+)
+
+func TestMarshalUnmarshalbatchJobSubmitter(t *testing.T) {
+	v := batchJobSubmitter{}
+	bts, err := v.MarshalMsg(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	left, err := v.UnmarshalMsg(bts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(left) > 0 {
+		t.Errorf("%d bytes left over after UnmarshalMsg(): %q", len(left), left)
+	}
+
+	left, err = msgp.Skip(bts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(left) > 0 {
+		t.Errorf("%d bytes left over after Skip(): %q", len(left), left)
+	}
+}
+
+func BenchmarkMarshalMsgbatchJobSubmitter(b *testing.B) {
+	v := batchJobSubmitter{}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		v.MarshalMsg(nil)
+	}
+}
+
+func BenchmarkAppendMsgbatchJobSubmitter(b *testing.B) {
+	v := batchJobSubmitter{}
+	bts := make([]byte, 0, v.Msgsize())
+	bts, _ = v.MarshalMsg(bts[0:0])
+	b.SetBytes(int64(len(bts)))
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		bts, _ = v.MarshalMsg(bts[0:0])
+	}
+}
+
+func BenchmarkUnmarshalbatchJobSubmitter(b *testing.B) {
+	v := batchJobSubmitter{}
+	bts, _ := v.MarshalMsg(nil)
+	b.ReportAllocs()
+	b.SetBytes(int64(len(bts)))
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, err := v.UnmarshalMsg(bts)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func TestEncodeDecodebatchJobSubmitter(t *testing.T) {
+	v := batchJobSubmitter{}
+	var buf bytes.Buffer
+	msgp.Encode(&buf, &v)
+
+	m := v.Msgsize()
+	if buf.Len() > m {
+		t.Log("WARNING: TestEncodeDecodebatchJobSubmitter Msgsize() is inaccurate")
+	}
+
+	vn := batchJobSubmitter{}
+	err := msgp.Decode(&buf, &vn)
+	if err != nil {
+		t.Error(err)
+	}
+
+	buf.Reset()
+	msgp.Encode(&buf, &v)
+	err = msgp.NewReader(&buf).Skip()
+	if err != nil {
+		t.Error(err)
+	}
+}
+
+func BenchmarkEncodebatchJobSubmitter(b *testing.B) {
+	v := batchJobSubmitter{}
+	var buf bytes.Buffer
+	msgp.Encode(&buf, &v)
+	b.SetBytes(int64(buf.Len()))
+	en := msgp.NewWriter(msgp.Nowhere)
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		v.EncodeMsg(en)
+	}
+	en.Flush()
+}
+
+func BenchmarkDecodebatchJobSubmitter(b *testing.B) {
+	v := batchJobSubmitter{}
+	var buf bytes.Buffer
+	msgp.Encode(&buf, &v)
+	b.SetBytes(int64(buf.Len()))
+	rd := msgp.NewEndlessReader(buf.Bytes(), b)
+	dc := msgp.NewReader(rd)
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		err := v.DecodeMsg(dc)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}

cmd/batch-expire.go

@@ -27,6 +27,7 @@ import (
 	"net/http"
 	"runtime"
 	"strconv"
+	"sync/atomic"
 	"time"
 
 	"github.com/minio/minio-go/v7/pkg/tags"
@@ -34,6 +35,7 @@ import (
 	xhttp "github.com/minio/minio/internal/http"
 	xioutil "github.com/minio/minio/internal/ioutil"
 	"github.com/pgsty/silo-pkg/v3/env"
+	"github.com/pgsty/silo-pkg/v3/policy"
 	"github.com/pgsty/silo-pkg/v3/wildcard"
 	"github.com/pgsty/silo-pkg/v3/workers"
 	"github.com/pgsty/silo-pkg/v3/xtime"
@@ -289,16 +291,6 @@ type BatchJobExpire struct {
 
 var _ yaml.Unmarshaler = &BatchJobExpire{}
 
-// RedactSensitive will redact any sensitive information in b.
-func (r *BatchJobExpire) RedactSensitive() {
-	if r == nil {
-		return
-	}
-	if r.NotificationCfg.Token != "" {
-		r.NotificationCfg.Token = redactedText
-	}
-}
-
 // UnmarshalYAML - BatchJobExpire extends default unmarshal to extract line, col information.
 func (r *BatchJobExpire) UnmarshalYAML(val *yaml.Node) error {
 	type expireJob BatchJobExpire
@@ -345,32 +337,6 @@ func (r BatchJobExpire) Notify(ctx context.Context, body io.Reader) error {
 	return nil
 }
 
-// Expire expires object versions which have already matched supplied filter conditions
-func (r *BatchJobExpire) Expire(ctx context.Context, api ObjectLayer, vc *versioning.Versioning, objsToDel []ObjectToDelete) []error {
-	opts := ObjectOptions{
-		PrefixEnabledFn:  vc.PrefixEnabled,
-		VersionSuspended: vc.Suspended(),
-	}
-
-	allErrs := make([]error, 0, len(objsToDel))
-
-	for {
-		count := len(objsToDel)
-		if count == 0 {
-			break
-		}
-		if count > maxDeleteList {
-			count = maxDeleteList
-		}
-		_, errs := api.DeleteObjects(ctx, r.Bucket, objsToDel[:count], opts)
-		allErrs = append(allErrs, errs...)
-		// Next batch of deletion
-		objsToDel = objsToDel[count:]
-	}
-
-	return allErrs
-}
-
 const (
 	batchExpireName                 = "batch-expire.bin"
 	batchExpireFormat               = 1
@@ -381,22 +347,238 @@ const (
 	batchExpireJobDefaultRetryDelay = 250 * time.Millisecond
 )
 
-type objInfoCache map[string]*ObjectInfo
+// expireVersionRef is a version of a key whose versions are all purged, as
-
+// the walk saw it.
-func newObjInfoCache() objInfoCache {
+type expireVersionRef struct {
-	return objInfoCache(make(map[string]*ObjectInfo))
+	VersionID    string
+	UserTags     string
+	DeleteMarker bool
+	Locked       bool
 }
 
-func (oiCache objInfoCache) Add(toDel ObjectToDelete, oi *ObjectInfo) {
+type expireObjInfo struct {
-	oiCache[fmt.Sprintf("%s-%s", toDel.ObjectName, toDel.VersionID)] = oi
+	ObjectInfo
+	ExpireAll         bool
+	DeleteMarkerCount int64
+	// versions are the versions of an ExpireAll key the walk saw, the
+	// latest first.
+	versions []expireVersionRef
 }
 
-func (oiCache objInfoCache) Get(toDel ObjectToDelete) (*ObjectInfo, bool) {
+// isErrObjectLocked reports a version that Object Lock protects.
-	oi, ok := oiCache[fmt.Sprintf("%s-%s", toDel.ObjectName, toDel.VersionID)]
+func isErrObjectLocked(err error) bool {
-	return oi, ok
+	var locked ObjectLocked
+	return errors.As(err, &locked)
 }
 
-func batchObjsForDelete(ctx context.Context, r *BatchJobExpire, ri *batchJobInfo, job BatchJobRequest, api ObjectLayer, wk *workers.Workers, expireCh <-chan []expireObjInfo) {
+func (r *BatchJobExpire) tagsFiltered() bool {
+	for _, rule := range r.Rules {
+		if len(rule.Tags) > 0 {
+			return true
+		}
+	}
+	return false
+}
+
+// expireAuthzRequests are the actions the job takes on one version: it reads
+// the version, and its tags when a rule filters by them, and deletes it.
+func (r *BatchJobExpire) expireAuthzRequests(name, versionID, userTags string) []batchAuthzRequest {
+	if versionID == "" {
+		versionID = nullVersionID
+	}
+	reqs := []batchAuthzRequest{
+		{action: policy.GetObjectAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags},
+		{action: policy.GetObjectVersionAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags},
+		{action: policy.DeleteObjectAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags},
+		{action: policy.DeleteObjectVersionAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags},
+	}
+	if r.tagsFiltered() {
+		reqs = append(reqs,
+			batchAuthzRequest{action: policy.GetObjectTaggingAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags},
+			batchAuthzRequest{action: policy.GetObjectVersionTaggingAction, bucket: r.Bucket, object: name, versionID: versionID, tags: userTags},
+		)
+	}
+	return reqs
+}
+
+// expireVersion deletes one version that is older than the versions the rule
+// retains. Under the object's write lock, with the version as it is then, it
+// leaves a version that Object Lock protects (ObjectLocked) and refuses one
+// that the submitter may not delete (batchJobAuthzError). An unreadable
+// version is not deleted.
+func (r *BatchJobExpire) expireVersion(ctx context.Context, api ObjectLayer, vc *versioning.Versioning, job BatchJobRequest, exp expireObjInfo) error {
+	p, err := job.Submitter.principal(ctx)
+	if err != nil {
+		return err
+	}
+	versionID := exp.VersionID
+	if versionID == "" {
+		// An empty version ID would add a delete marker instead.
+		versionID = nullVersionID
+	}
+	opts := ObjectOptions{
+		VersionID:        versionID,
+		Versioned:        vc.PrefixEnabled(exp.Name),
+		VersionSuspended: vc.Suspended(),
+	}
+	opts.SetEvalRetentionBypassFn(func(goi ObjectInfo, gerr error) error {
+		// A delete marker reads as MethodNotAllowed; any other error,
+		// including a missing version, stops the delete.
+		marker := isErrMethodNotAllowed(gerr) && (goi.DeleteMarker || !goi.VersionPurgeStatus.Empty())
+		if gerr != nil && !marker {
+			return gerr
+		}
+		if enforceRetentionForDeletion(ctx, goi) {
+			return ObjectLocked{Bucket: r.Bucket, Object: exp.Name, VersionID: versionID}
+		}
+		for _, req := range r.expireAuthzRequests(exp.Name, versionID, goi.UserTags) {
+			if err := p.authorize(req); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+	_, err = api.DeleteObject(ctx, r.Bucket, exp.Name, opts)
+	return err
+}
+
+// expireVersions deletes the versions of one batch entry and records the
+// outcome in res. It returns an error that ends the job.
+func (r *BatchJobExpire) expireVersions(ctx context.Context, api ObjectLayer, vc *versioning.Versioning, job BatchJobRequest, ri *batchJobInfo, exp expireObjInfo, retryAttempts int, delay time.Duration, res *batchCounts) error {
+	for attempts := 1; attempts <= retryAttempts; attempts++ {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempts)
+		err := r.expireVersion(ctx, api, vc, job, exp)
+		stopFn(exp, err)
+		switch {
+		case err == nil:
+			if exp.DeleteMarker {
+				ri.countRemoved(r.Bucket, 0, 1)
+			} else {
+				ri.countRemoved(r.Bucket, 1, 0)
+			}
+			return nil
+		case isBatchJobAuthzError(err):
+			return err
+		case isErrObjectLocked(err):
+			// Left in place, not retried and not a failure.
+			res.versionsSkipped++
+			return nil
+		case isErrVersionNotFound(err) || isErrObjectNotFound(err):
+			// Removed concurrently.
+			return nil
+		}
+		batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s versionID=%s due to %v (attempts=%d)", r.Bucket, exp.Name, exp.VersionID, err, attempts))
+		if attempts < retryAttempts {
+			time.Sleep(delay)
+		}
+	}
+	if exp.DeleteMarker {
+		res.deleteMarkersFailed++
+	} else {
+		res.objectsFailed++
+	}
+	return nil
+}
+
+// expireAllVersions purges every version of a key whose rule retains none,
+// with the purge of exactly that key (erasureServerPools.deleteAllVersions),
+// while the version the walk matched is still its latest. What the purge
+// removed is counted at once; a key it refused because Object Lock protects
+// a version is counted in res as refused, and the versions a failed purge
+// left in res as failures. A failed purge is retried up to retryAttempts
+// times, each time with the version the walk matched; a purge that stopped
+// at its budget after removing versions continues without spending an
+// attempt. A key whose latest version changed since the walk is not
+// expired. It returns an error that ends the job.
+//
+// TODO(rebase onto the delete-all purge, claude/fix-delete-all): before the
+// purge removes anything, authorize every version of the history it
+// approved under its lock, with version-specific conditions
+// (expireAuthzRequests). Until then the checks here use the versions the
+// walk saw, and do not cover a version added after the walk; the purge
+// itself refuses a key that holds a locked version.
+func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer, job BatchJobRequest, ri *batchJobInfo, exp expireObjInfo, retryAttempts int, res *batchCounts) error {
+	for _, v := range exp.versions {
+		if v.Locked {
+			res.keysRefused++
+			return nil
+		}
+	}
+	var reqs []batchAuthzRequest
+	for _, v := range exp.versions {
+		reqs = append(reqs, r.expireAuthzRequests(exp.Name, v.VersionID, v.UserTags)...)
+	}
+	if err := job.Submitter.authorize(ctx, reqs...); err != nil {
+		return err
+	}
+	var objects, markers int64
+	matched := deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime}
+	for attempts := 1; attempts <= retryAttempts; attempts++ {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempts)
+		latest := matched
+		var purge deleteAllResult
+		_, err := api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{
+			DeleteAllVersions: true,
+			DeleteAllLatest:   &latest,
+			DeleteAllResult:   &purge,
+		})
+		// Removed versions are gone for any later run: count them now.
+		if purge.Removed > 0 {
+			ri.countRemoved(r.Bucket, int64(purge.Removed-purge.Markers), int64(purge.Markers))
+			objects += int64(purge.Removed - purge.Markers)
+			markers += int64(purge.Markers)
+		}
+		stopFn(exp, err)
+		switch {
+		case err == nil:
+			return nil
+		case errors.Is(err, errDeleteAllChanged):
+			// Changed since the walk matched it: not expired.
+			return nil
+		case isErrObjectLocked(err):
+			// Refused as a whole: nothing removed, not retried and not a
+			// failure.
+			res.keysRefused++
+			return nil
+		case errors.Is(err, errDeleteAllBudget) && purge.Removed > 0:
+			attempts-- // progress
+			continue
+		}
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s due to %v (attempts=%d)", exp.Bucket, exp.Name, err, attempts))
+		if errors.Is(err, errDeleteAllTiered) {
+			break // refused; a later run of the job checks again
+		}
+	}
+	res.expireAllFailed(exp, objects, markers)
+	return nil
+}
+
+// expireAllFailed counts the versions of info, as the walk saw them, that a
+// failed purge of all its versions did not remove: those beyond the objects
+// and markers it removed, at least one.
+func (c *batchCounts) expireAllFailed(info expireObjInfo, objects, markers int64) {
+	failedObjects := max(int64(info.NumVersions)-info.DeleteMarkerCount-objects, 0)
+	failedMarkers := max(info.DeleteMarkerCount-markers, 0)
+	if failedObjects+failedMarkers == 0 {
+		failedObjects = 1
+	}
+	c.objectsFailed += failedObjects
+	c.deleteMarkersFailed += failedMarkers
+}
+
+// batchObjsForDelete expires the batches of expireCh. walkDone reports, once
+// expireCh is closed, whether the walk reached its end: only then is its last
+// key known to be complete.
+func batchObjsForDelete(ctx context.Context, refuse func(error), walkDone func() bool, r *BatchJobExpire, ri *batchJobInfo, job BatchJobRequest, api ObjectLayer, wk *workers.Workers, expireCh <-chan []expireObjInfo) {
 	vc, _ := globalBucketVersioningSys.Get(r.Bucket)
 	retryAttempts := job.Expire.Retry.Attempts
 	if retryAttempts <= 0 {
@@ -407,162 +589,53 @@ func batchObjsForDelete(ctx context.Context, r *BatchJobExpire, ri *batchJobInfo
 		delay = batchExpireJobDefaultRetryDelay
 	}
 
-	var i int
+	progress := newBatchProgress(ri, r.Bucket)
+	defer func() {
+		wk.Wait()
+		if walkDone() {
+			progress.finish()
+		}
+	}()
 	for toExpire := range expireCh {
 		select {
 		case <-ctx.Done():
 			return
 		default:
 		}
-		if i > 0 {
+		seq := progress.begin()
+		if seq > 0 {
 			if wait := globalBatchConfig.ExpirationWait(); wait > 0 {
 				time.Sleep(wait)
 			}
 		}
-		i++
 		wk.Take()
-		go func(toExpire []expireObjInfo) {
+		go func(seq uint64, toExpire []expireObjInfo) {
 			defer wk.Give()
-
+			var res batchCounts
-			toExpireAll := make([]expireObjInfo, 0, len(toExpire))
-			toDel := make([]ObjectToDelete, 0, len(toExpire))
-			oiCache := newObjInfoCache()
 			for _, exp := range toExpire {
+				var err error
 				if exp.ExpireAll {
-					toExpireAll = append(toExpireAll, exp)
+					err = r.expireAllVersions(ctx, api, job, ri, exp, retryAttempts, &res)
-					continue
+				} else {
+					err = r.expireVersions(ctx, api, vc, job, ri, exp, retryAttempts, delay, &res)
 				}
-				// Cache ObjectInfo value via pointers for
+				if err != nil {
-				// subsequent use to track objects which
+					// The batch is not done; a later run does it again.
-				// couldn't be deleted.
+					if isBatchJobAuthzError(err) {
-				od := ObjectToDelete{
+						refuse(err)
-					ObjectV: ObjectV{
-						ObjectName: exp.Name,
-						VersionID:  exp.VersionID,
-					},
-				}
-				toDel = append(toDel, od)
-				oiCache.Add(od, &exp.ObjectInfo)
-			}
-
-			// Purge all versions of exactly each key while the version the
-			// walk matched is still its latest; see
-			// erasureServerPools.deleteAllVersions.
-			for _, exp := range toExpireAll {
-				if !expireAllVersions(ctx, ri, api, exp, retryAttempts) {
-					return
-				}
-			}
-
-			// DeleteMultiple objects
-			toDelCopy := make([]ObjectToDelete, len(toDel))
-			for attempts := 1; attempts <= retryAttempts; attempts++ {
-				select {
-				case <-ctx.Done():
-					return
-				default:
-				}
-
-				stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempts)
-				// Copying toDel to select from objects whose
-				// deletion failed
-				copy(toDelCopy, toDel)
-				var failed int
-				errs := r.Expire(ctx, api, vc, toDel)
-				// reslice toDel in preparation for next retry attempt
-				toDel = toDel[:0]
-				for i, err := range errs {
-					if err != nil {
-						stopFn(toDelCopy[i], err)
-						batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s versionID=%s due to %v (attempts=%d)", ri.Bucket, toDelCopy[i].ObjectName, toDelCopy[i].VersionID,
-							err, attempts))
-						failed++
-						if oi, ok := oiCache.Get(toDelCopy[i]); ok {
-							ri.trackCurrentBucketObject(r.Bucket, *oi, false, attempts)
-						}
-						if attempts != retryAttempts {
-							// retry
-							toDel = append(toDel, toDelCopy[i])
-						}
-					} else {
-						stopFn(toDelCopy[i], nil)
-						if oi, ok := oiCache.Get(toDelCopy[i]); ok {
-							ri.trackCurrentBucketObject(r.Bucket, *oi, true, attempts)
-						}
 					}
-				}
+					globalBatchJobsMetrics.save(ri.JobID, ri)
-
+					return
-				globalBatchJobsMetrics.save(ri.JobID, ri)
-
-				if failed == 0 {
-					break
-				}
-
-				// Add a delay between retry attempts
-				if attempts < retryAttempts {
-					time.Sleep(delay)
 				}
 			}
-		}(toExpire)
+			if n := len(toExpire); n > 0 {
+				progress.done(seq, toExpire[n-1].Name, res)
+			}
+			globalBatchJobsMetrics.save(ri.JobID, ri)
+		}(seq, toExpire)
 	}
 }
 
-// expireAllVersions purges all versions of the key exp, which the walk
-// matched by its latest version, and counts what the purge removed. It
-// retries a failed purge up to attempts times, each time with the version
-// the walk matched; a purge that stopped at its budget continues without
-// spending an attempt. It does not retry once the key no longer has that
-// latest version, which is not an expiry, or once the purge refused the
-// key. It returns false once ctx ends.
-func expireAllVersions(ctx context.Context, ri *batchJobInfo, api ObjectLayer, exp expireObjInfo, attempts int) bool {
-	var objects, markers int64
-	matched := deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime}
-	for attempt := 1; attempt <= attempts; attempt++ {
-		select {
-		case <-ctx.Done():
-			ri.trackExpireAll(exp, objects, markers, true)
-			return false
-		default:
-		}
-		stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempt)
-		latest := matched
-		var purge deleteAllResult
-		_, err := api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{
-			DeleteAllVersions: true,
-			DeleteAllLatest:   &latest,
-			DeleteAllResult:   &purge,
-		})
-		objects += int64(purge.Removed - purge.Markers)
-		markers += int64(purge.Markers)
-		stopFn(exp, err)
-		switch {
-		case err == nil:
-			ri.trackExpireAll(exp, objects, markers, false)
-			return true
-		case errors.Is(err, errDeleteAllChanged):
-			// Changed since the walk matched it: not expired.
-			ri.trackExpireAll(exp, objects, markers, false)
-			return true
-		case errors.Is(err, errDeleteAllBudget) && purge.Removed > 0:
-			attempt-- // progress
-			continue
-		}
-		batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s due to %v (attempts=%d)", exp.Bucket, exp.Name, err, attempt))
-		var locked ObjectLocked
-		if errors.Is(err, errDeleteAllTiered) || errors.As(err, &locked) {
-			break // refused; a later run of the job checks again
-		}
-	}
-	ri.trackExpireAll(exp, objects, markers, true)
-	return true
-}
-
-type expireObjInfo struct {
-	ObjectInfo
-	ExpireAll         bool
-	DeleteMarkerCount int64
-}
-
 // Start the batch expiration job, resumes if there was a pending job via "job.ID"
 func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {
 	ri := &batchJobInfo{
@@ -573,6 +646,9 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 	if err := ri.loadOrInit(ctx, api, job); err != nil {
 		return err
 	}
+	if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
+		return err
+	}
 
 	globalBatchJobsMetrics.save(job.ID, ri)
 	lastObject := ri.Object
@@ -592,27 +668,43 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 
 	ctx, cancelCause := context.WithCancelCause(ctx)
 	defer cancelCause(nil)
+	// refuse ends the job for its submitter: record why before stopping.
+	refuse := func(err error) {
+		ri.refuse(err)
+		cancelCause(err)
+	}
 
+	// The walk stops at a listing failure: work after it is left to a later
+	// run, so the cursor cannot move past what was not listed.
+	walkCtx, stopWalk := context.WithCancel(ctx)
+	defer stopWalk()
 	results := make(chan itemOrErr[ObjectInfo], workerSize)
 	go func() {
-		prefixes := r.Prefix.F()
+		// One ascending walk over disjoint prefixes, which the cursor
-		if len(prefixes) == 0 {
+		// describes.
-			prefixes = []string{""}
+		for _, prefix := range batchCanonicalPrefixes(r.Prefix.F()) {
-		}
+			if walkCtx.Err() != nil {
-		for _, prefix := range prefixes {
+				break
+			}
 			prefixResultCh := make(chan itemOrErr[ObjectInfo], workerSize)
-			err := api.Walk(ctx, r.Bucket, prefix, prefixResultCh, WalkOptions{
+			err := api.Walk(walkCtx, r.Bucket, prefix, prefixResultCh, WalkOptions{
 				Marker:       lastObject,
 				LatestOnly:   false, // we need to visit all versions of the object to implement purge: retainVersions
 				VersionsSort: WalkVersionsSortDesc,
 			})
 			if err != nil {
-				cancelCause(err)
+				if walkCtx.Err() == nil {
+					cancelCause(err)
+				}
 				xioutil.SafeClose(results)
 				return
 			}
 			for result := range prefixResultCh {
-				results <- result
+				select {
+				case results <- result:
+				case <-walkCtx.Done():
+					// Stopped: let the walk end without a reader.
+				}
 			}
 		}
 		xioutil.SafeClose(results)
@@ -620,7 +712,9 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 
 	// Goroutine to periodically save batch-expire job's in-memory state
 	saverQuitCh := make(chan struct{})
+	saverDoneCh := make(chan struct{})
 	go func() {
+		defer close(saverDoneCh)
 		saveTicker := time.NewTicker(10 * time.Second)
 		defer saveTicker.Stop()
 		quit := false
@@ -647,9 +741,10 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 
 	expireCh := make(chan []expireObjInfo, workerSize)
 	expireDoneCh := make(chan struct{})
+	var walkDone atomic.Bool
 	go func() {
 		defer close(expireDoneCh)
-		batchObjsForDelete(ctx, r, ri, job, api, wk, expireCh)
+		batchObjsForDelete(ctx, refuse, walkDone.Load, r, ri, job, api, wk, expireCh)
 	}()
 
 	var (
@@ -661,6 +756,8 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 		done          bool
 	)
 	deleteMarkerCountMap := map[string]int64{}
+	// pushToExpire is called between keys only, so that a batch holds all
+	// the versions of its keys and its last key is done with it.
 	pushToExpire := func() {
 		// set preObject deleteMarkerCount
 		if len(toDel) > 0 {
@@ -695,6 +792,12 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 			if result.Err != nil {
 				failed = true
 				batchLogIf(ctx, result.Err)
+				stopWalk()
+				done = true
+				break
+			}
+			if !batchResumeAfter(lastObject, result.Item.Name) {
+				// Completed by the run that set the cursor.
 				continue
 			}
 			if result.Item.DeleteMarker {
@@ -730,12 +833,17 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 					toDel = append(toDel, expireObjInfo{
 						ObjectInfo: result.Item,
 						ExpireAll:  true,
+						versions:   []expireVersionRef{newExpireVersionRef(ctx, result.Item)},
 					})
 					continue
 				}
 			} else if prevObj.Name == result.Item.Name {
 				if matchedFilter.Purge.RetainVersions == 0 {
-					continue // including latest version in toDel suffices, skipping other versions
+					// The latest version in toDel purges them all.
+					if n := len(toDel); n > 0 && toDel[n-1].Name == result.Item.Name {
+						toDel[n-1].versions = append(toDel[n-1].versions, newExpireVersionRef(ctx, result.Item))
+					}
+					continue
 				}
 				versionsCount++
 			} else {
@@ -752,7 +860,6 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 			toDel = append(toDel, expireObjInfo{
 				ObjectInfo: result.Item,
 			})
-			pushToExpire()
 		case <-ctx.Done():
 			done = true
 		}
@@ -761,16 +868,15 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 		}
 	}
 
-	if context.Cause(ctx) != nil {
+	if context.Cause(ctx) == nil {
-		xioutil.SafeClose(expireCh)
+		walkDone.Store(!failed)
-		return context.Cause(ctx)
+		pushToExpire()
-	}
+		// Send any remaining objects downstream
-	pushToExpire()
+		if len(toDel) > 0 {
-	// Send any remaining objects downstream
+			select {
-	if len(toDel) > 0 {
+			case <-ctx.Done():
-		select {
+			case expireCh <- toDel:
-		case <-ctx.Done():
+			}
-		case expireCh <- toDel:
 		}
 	}
 	xioutil.SafeClose(expireCh)
@@ -778,13 +884,29 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 	<-expireDoneCh // waits for the expire goroutine to complete
 	wk.Wait()      // waits for all expire workers to retire
 
-	ri.Complete = !failed && !ri.expireFailures()
+	// Persist the final state once the saver has stopped, so that its last
-	ri.Failed = failed || ri.expireFailures()
+	// save cannot overwrite it.
-	globalBatchJobsMetrics.save(job.ID, ri)
-
-	// Close the saverQuitCh - this also triggers saving in-memory state
-	// immediately one last time before we exit this method.
 	xioutil.SafeClose(saverQuitCh)
+	<-saverDoneCh
+	refused := ri.refused()
+	cause := context.Cause(ctx)
+	if cause != nil && !refused {
+		saveCtx, cancel := context.WithTimeout(GlobalContext, 30*time.Second)
+		batchLogIf(saveCtx, ri.updateAfter(saveCtx, api, 0, job))
+		cancel()
+		globalBatchJobsMetrics.save(job.ID, ri)
+		return cause
+	}
+
+	ri.mu.Lock()
+	itemsFailed := ri.expireFailures()
+	ri.Complete = !refused && !failed && !itemsFailed
+	ri.Failed = refused || failed || itemsFailed
+	ri.mu.Unlock()
+	globalBatchJobsMetrics.save(job.ID, ri)
+	saveCtx, cancel := context.WithTimeout(GlobalContext, 30*time.Second)
+	batchLogIf(saveCtx, ri.updateAfter(saveCtx, api, 0, job))
+	cancel()
 
 	// Notify expire jobs final status to the configured endpoint
 	buf, _ := json.Marshal(ri)
@@ -795,6 +917,15 @@ func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJo
 	return nil
 }
 
+func newExpireVersionRef(ctx context.Context, oi ObjectInfo) expireVersionRef {
+	return expireVersionRef{
+		VersionID:    oi.VersionID,
+		UserTags:     oi.UserTags,
+		DeleteMarker: oi.DeleteMarker,
+		Locked:       enforceRetentionForDeletion(ctx, oi),
+	}
+}
+
 //msgp:ignore batchExpireJobError
 type batchExpireJobError struct {
 	Code           string

cmd/batch-handlers.go

@@ -43,6 +43,8 @@ import (
 	"github.com/minio/minio-go/v7/pkg/credentials"
 	"github.com/minio/minio-go/v7/pkg/encrypt"
 	"github.com/minio/minio-go/v7/pkg/tags"
+	"github.com/minio/minio/internal/amztime"
+	objectlock "github.com/minio/minio/internal/bucket/object/lock"
 	"github.com/minio/minio/internal/config/batch"
 	"github.com/minio/minio/internal/crypto"
 	"github.com/minio/minio/internal/hash"
@@ -72,32 +74,47 @@ type BatchJobRequest struct {
 	Replicate *BatchJobReplicateV1 `yaml:"replicate" json:"replicate"`
 	KeyRotate *BatchJobKeyRotateV1 `yaml:"keyrotate" json:"keyrotate"`
 	Expire    *BatchJobExpire      `yaml:"expire" json:"expire"`
-	ctx       context.Context      `msg:"-"`
+	// Submitter is the credential the job acts as. The server assigns it
+	// at submission; a job definition cannot name it.
+	Submitter *batchJobSubmitter `yaml:"-" json:"-"`
+	ctx       context.Context    `msg:"-"`
 }
 
-// RedactSensitive will redact any sensitive information in b.
+// redacted returns a copy of the job for output outside the server: the
-func (j *BatchJobRequest) RedactSensitive() {
+// describe response, logs and errors. The copy has no secret keys, session
-	j.Replicate.RedactSensitive()
+// tokens or notification tokens, and no submitter binding; the job itself,
-	j.Expire.RedactSensitive()
+// which a run needs as it is, is not changed.
-	j.KeyRotate.RedactSensitive()
+func (j BatchJobRequest) redacted() (BatchJobRequest, error) {
+	buf, err := j.MarshalMsg(nil)
+	if err != nil {
+		return BatchJobRequest{}, err
+	}
+	var c BatchJobRequest
+	if _, err = c.UnmarshalMsg(buf); err != nil {
+		return BatchJobRequest{}, err
+	}
+	redact := func(s *string) {
+		if *s != "" {
+			*s = redactedText
+		}
+	}
+	c.Submitter = nil
+	if r := c.Replicate; r != nil {
+		redact(&r.Source.Creds.SecretKey)
+		redact(&r.Source.Creds.SessionToken)
+		redact(&r.Target.Creds.SecretKey)
+		redact(&r.Target.Creds.SessionToken)
+		redact(&r.Flags.Notify.Token)
+	}
+	if r := c.KeyRotate; r != nil {
+		redact(&r.Flags.Notify.Token)
+	}
+	if r := c.Expire; r != nil {
+		redact(&r.NotificationCfg.Token)
+	}
+	return c, nil
 }
 
-// RedactSensitive will redact any sensitive information in b.
-func (r *BatchJobReplicateV1) RedactSensitive() {
-	if r == nil {
-		return
-	}
-	if r.Target.Creds.SecretKey != "" {
-		r.Target.Creds.SecretKey = redactedText
-	}
-	if r.Target.Creds.SessionToken != "" {
-		r.Target.Creds.SessionToken = redactedText
-	}
-}
-
-// RedactSensitive will redact any sensitive information in b.
-func (r *BatchJobKeyRotateV1) RedactSensitive() {}
-
 func notifyEndpoint(ctx context.Context, ri *batchJobInfo, endpoint, token string) error {
 	if endpoint == "" {
 		return nil
@@ -140,8 +157,114 @@ func (r BatchJobReplicateV1) Notify(ctx context.Context, ri *batchJobInfo) error
 	return notifyEndpoint(ctx, ri, r.Flags.Notify.Endpoint, r.Flags.Notify.Token)
 }
 
-// ReplicateFromSource - this is not implemented yet where source is 'remote' and target is local.
+// batchPullObjectLock prepares the Object Lock metadata that a pull writes
-func (r *BatchJobReplicateV1) ReplicateFromSource(ctx context.Context, api ObjectLayer, core *minio.Core, srcObjInfo ObjectInfo, retry bool) error {
+// with a version, as a PUT would on the destination: Object Lock metadata
+// copied from the source must be valid, a version without retention takes the
+// destination bucket's default retention, and both are dated with the
+// destination's clock. Outside a lock-enabled bucket the metadata has no
+// meaning and is not written. It reports the retention and legal hold that
+// the pull copies from the source.
+func batchPullObjectLock(bucket string, meta map[string]string) (retention, legalHold bool, err error) {
+	cfg, err := globalBucketObjectLockSys.Get(bucket)
+	if err != nil {
+		return false, false, err
+	}
+	h := make(http.Header)
+	for k, v := range meta {
+		for _, name := range []string{xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold} {
+			if strings.EqualFold(k, name) {
+				h.Set(name, v)
+				delete(meta, k)
+			}
+		}
+	}
+	if !cfg.LockEnabled {
+		return false, false, nil
+	}
+	if h.Get(xhttp.AmzObjectLockMode) != "" || h.Get(xhttp.AmzObjectLockRetainUntilDate) != "" {
+		// A copied retention may have expired already.
+		mode, until, err := objectlock.ParseObjectLockRetentionHeaders(h, true)
+		if err != nil {
+			return false, false, err
+		}
+		meta[strings.ToLower(xhttp.AmzObjectLockMode)] = string(mode)
+		meta[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(until.UTC())
+		retention = true
+	} else if cfg.Validity > 0 {
+		now, err := objectlock.UTCNowNTP()
+		if err != nil {
+			return false, false, err
+		}
+		meta[strings.ToLower(xhttp.AmzObjectLockMode)] = string(cfg.Mode)
+		meta[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(now.Add(cfg.Validity).UTC())
+	}
+	if _, ok := h[xhttp.AmzObjectLockLegalHold]; ok {
+		hold, err := objectlock.ParseObjectLockLegalHoldHeaders(h)
+		if err != nil {
+			return false, false, err
+		}
+		meta[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(hold.Status)
+		legalHold = true
+	}
+	return retention, legalHold, nil
+}
+
+// pullAuthzRequests are the writes of one version that a pull makes to the
+// destination key: the version, with the tags, retention and legal hold it
+// copies, and the replication permission for a version ID it keeps.
+//
+// The pull does not encrypt what it writes, so it supplies no encryption
+// values, whatever the source reports; and it does not read the destination
+// version it may replace, so the tags of that version are unknown.
+func pullAuthzRequests(bucket, object, versionID string, keepsVersion bool, opts ObjectOptions, retention, legalHold bool) []batchAuthzRequest {
+	values := map[string][]string{}
+	meta := caseInsensitiveMap(opts.UserDefined)
+	if retention {
+		mode, _ := meta.Lookup(xhttp.AmzObjectLockMode)
+		until, _ := meta.Lookup(xhttp.AmzObjectLockRetainUntilDate)
+		values[strings.TrimPrefix(strings.ToLower(xhttp.AmzObjectLockMode), "x-amz-")] = []string{mode}
+		values[strings.TrimPrefix(strings.ToLower(xhttp.AmzObjectLockRetainUntilDate), "x-amz-")] = []string{until}
+	}
+	if legalHold {
+		hold, _ := meta.Lookup(xhttp.AmzObjectLockLegalHold)
+		values[strings.TrimPrefix(strings.ToLower(xhttp.AmzObjectLockLegalHold), "x-amz-")] = []string{hold}
+	}
+	tagStr, hasTags := meta.Lookup(xhttp.AmzObjectTagging)
+	if hasTags && tagStr != "" {
+		if t, err := tags.ParseObjectTags(tagStr); err == nil {
+			m := t.ToMap()
+			keys := make([]string, 0, len(m))
+			for k, v := range m {
+				values[pathJoin("RequestObjectTag", k)] = []string{v}
+				keys = append(keys, k)
+			}
+			values["RequestObjectTagKeys"] = keys
+		}
+	}
+	actions := []policy.Action{policy.PutObjectAction}
+	if keepsVersion {
+		actions = append(actions, policy.ReplicateObjectAction)
+	}
+	if hasTags && tagStr != "" {
+		actions = append(actions, policy.PutObjectTaggingAction)
+	}
+	if retention {
+		actions = append(actions, policy.PutObjectRetentionAction)
+	}
+	if legalHold {
+		actions = append(actions, policy.PutObjectLegalHoldAction)
+	}
+	reqs := make([]batchAuthzRequest, 0, len(actions))
+	for _, action := range actions {
+		reqs = append(reqs, batchAuthzRequest{action: action, bucket: bucket, object: object, versionID: versionID, tagsUnknown: true, values: values})
+	}
+	return reqs
+}
+
+// ReplicateFromSource copies one version from the remote source to the local
+// target. A write that would replace a version protected by Object Lock
+// returns ObjectLocked and leaves it.
+func (r *BatchJobReplicateV1) ReplicateFromSource(ctx context.Context, api ObjectLayer, core *minio.Core, srcObjInfo ObjectInfo, retry bool, sub *batchJobSubmitter) error {
 	srcBucket := r.Source.Bucket
 	tgtBucket := r.Target.Bucket
 	srcObject := srcObjInfo.Name
@@ -150,11 +273,28 @@ func (r *BatchJobReplicateV1) ReplicateFromSource(ctx context.Context, api Objec
 		tgtObject = pathJoin(r.Target.Prefix, srcObjInfo.Name)
 	}
 
+	p, err := sub.principal(ctx)
+	if err != nil {
+		return err
+	}
 	versionID := srcObjInfo.VersionID
 	if r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3 {
 		versionID = ""
 	}
 	if srcObjInfo.DeleteMarker {
+		markerVersion := versionID
+		if markerVersion == "" {
+			markerVersion = nullVersionID
+		}
+		for _, req := range []batchAuthzRequest{
+			{action: policy.DeleteObjectAction, bucket: tgtBucket, object: tgtObject, versionID: markerVersion, tagsUnknown: true},
+			{action: policy.DeleteObjectVersionAction, bucket: tgtBucket, object: tgtObject, versionID: markerVersion, tagsUnknown: true},
+			{action: policy.ReplicateDeleteAction, bucket: tgtBucket, object: tgtObject, versionID: markerVersion, tagsUnknown: true},
+		} {
+			if err := p.authorize(req); err != nil {
+				return err
+			}
+		}
 		_, err := api.DeleteObject(ctx, tgtBucket, tgtObject, ObjectOptions{
 			VersionID: versionID,
 			// Since we are preserving a delete marker, we have to make sure this is always true.
@@ -164,22 +304,43 @@ func (r *BatchJobReplicateV1) ReplicateFromSource(ctx context.Context, api Objec
 			MTime:              srcObjInfo.ModTime,
 			DeleteMarker:       srcObjInfo.DeleteMarker,
 			ReplicationRequest: true,
+			// A new marker may hide a locked version; one that would replace
+			// it is refused.
+			RefuseLockedOverwrite: true,
 		})
 		return err
 	}
 
-	opts := ObjectOptions{
+	userDefined := cloneMSS(srcObjInfo.UserDefined)
-		VersionID:    srcObjInfo.VersionID,
+	retention, legalHold, err := batchPullObjectLock(tgtBucket, userDefined)
-		MTime:        srcObjInfo.ModTime,
+	if err != nil {
-		PreserveETag: srcObjInfo.ETag,
+		return err
-		UserDefined:  srcObjInfo.UserDefined,
 	}
-	if r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3 {
+	opts := ObjectOptions{
-		opts.VersionID = ""
+		VersionID:        versionID,
+		Versioned:        globalBucketVersioningSys.PrefixEnabled(tgtBucket, tgtObject),
+		VersionSuspended: globalBucketVersioningSys.PrefixSuspended(tgtBucket, tgtObject),
+		MTime:            srcObjInfo.ModTime,
+		PreserveETag:     srcObjInfo.ETag,
+		UserDefined:      userDefined,
+		// Checked with the commit, under the destination's write lock.
+		RefuseLockedOverwrite: true,
 	}
 	if crypto.S3.IsEncrypted(srcObjInfo.UserDefined) {
 		opts.ServerSideEncryption = encrypt.NewSSE()
 	}
+	// The version the write replaces or creates: the source's version ID it
+	// keeps, or the null version of a destination that does not version
+	// the key. A versioned destination gives a write without an ID a new one.
+	writeVersion := versionID
+	if writeVersion == "" && !opts.Versioned {
+		writeVersion = nullVersionID
+	}
+	for _, req := range pullAuthzRequests(tgtBucket, tgtObject, writeVersion, versionID != "", opts, retention, legalHold) {
+		if err := p.authorize(req); err != nil {
+			return err
+		}
+	}
 	slc := strings.Split(srcObjInfo.ETag, "-")
 	if len(slc) == 2 {
 		partsCount, err := strconv.Atoi(slc[1])
@@ -298,6 +459,9 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 	if ri.Complete {
 		return nil
 	}
+	if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
+		return err
+	}
 	globalBatchJobsMetrics.save(job.ID, ri)
 
 	retryAttempts := job.Replicate.Flags.Retry.Attempts
@@ -411,7 +575,17 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 		// one of source/target is s3, skip delete marker and all versions under the same object name.
 		s3Type := r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3
 		minioSrc := r.Source.Type == BatchJobReplicateResourceMinIO
-		ctx, cancel := context.WithCancel(ctx)
+		ctx, cancelCause := context.WithCancelCause(ctx)
+		// refuse ends the job for its submitter: record why before stopping.
+		refuse := func(err error) {
+			ri.refuse(err)
+			cancelCause(err)
+		}
+		// Each run lists the source from the start and counts every locked
+		// collision again.
+		ri.mu.Lock()
+		ri.VersionsSkippedLocked = 0
+		ri.mu.Unlock()
 
 		objInfoCh := make(chan minio.ObjectInfo, 1)
 		go func() {
@@ -437,6 +611,10 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 		skipReplicate := false
 
 		for obj := range objInfoCh {
+			if context.Cause(ctx) != nil {
+				// Refused or canceled: schedule nothing more.
+				continue
+			}
 			oi := toObjectInfo(r.Source.Bucket, obj.Key, obj)
 			if !minioSrc {
 				// Check if metadata filter was requested and it is expected to have
@@ -484,7 +662,19 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 				defer wk.Give()
 				stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts)
 				success := true
-				if err := r.ReplicateFromSource(ctx, api, core, oi, retry); err != nil {
+				if err := r.ReplicateFromSource(ctx, api, core, oi, retry, job.Submitter); err != nil {
+					if isBatchJobAuthzError(err) {
+						refuse(err)
+						return
+					}
+					// A locked destination version is left as it is: the
+					// job completes with it skipped, not replicated.
+					if isErrObjectLocked(err) {
+						stopFn(oi, err)
+						ri.countSkipped()
+						globalBatchJobsMetrics.save(job.ID, ri)
+						return
+					}
 					// object must be deleted concurrently, allow these failures but do not count them
 					if isErrVersionNotFound(err) || isErrObjectNotFound(err) {
 						return
@@ -495,7 +685,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 				} else {
 					stopFn(oi, nil)
 				}
-				ri.trackCurrentBucketObject(r.Target.Bucket, oi, success, attempts)
+				ri.trackCurrentBucketObject(r.Target.Bucket, oi, success)
 				globalBatchJobsMetrics.save(job.ID, ri)
 				// persist in-memory state to disk after every 10secs.
 				batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job))
@@ -507,9 +697,19 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 		}
 		wk.Wait()
 
+		if ri.refused() {
+			cancelCause(nil)
+			globalBatchJobsMetrics.save(job.ID, ri)
+			batchLogIf(ctx, ri.updateAfter(GlobalContext, api, 0, job))
+			if err := r.Notify(GlobalContext, ri); err != nil {
+				batchLogIf(ctx, fmt.Errorf("unable to notify %v", err))
+			}
+			return nil
+		}
+
 		ri.RetryAttempts = attempts
-		ri.Complete = ri.ObjectsFailed == 0
+		ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0
-		ri.Failed = ri.ObjectsFailed > 0
+		ri.Failed = !ri.Complete
 
 		globalBatchJobsMetrics.save(job.ID, ri)
 		// persist in-memory state to disk.
@@ -519,12 +719,14 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 			batchLogIf(ctx, fmt.Errorf("unable to notify %v", err))
 		}
 
-		cancel()
+		cancelCause(nil)
 		if ri.Failed {
 			ri.ObjectsFailed = 0
+			ri.DeleteMarkersFailed = 0
 			ri.Bucket = ""
 			ri.Object = ""
 			ri.Objects = 0
+			ri.DeleteMarkers = 0
 			ri.BytesFailed = 0
 			ri.BytesTransferred = 0
 			retry = true // indicate we are retrying..
@@ -580,7 +782,10 @@ func toObjectInfo(bucket, object string, objInfo minio.ObjectInfo) ObjectInfo {
 	return oi
 }
 
-func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLayer, remoteClnt *minio.Client, entries []ObjectInfo, prefix string) error {
+func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLayer, remoteClnt *minio.Client, entries []ObjectInfo, prefix string, sub *batchJobSubmitter) error {
+	// An entry the submitter may not read ends the archive and the job; an
+	// entry whose metadata cannot be read ends the archive.
+	var refused, readErr error
 	input := make(chan minio.SnowballObject, 1)
 	opts := minio.SnowballOptions{
 		Opts: minio.PutObjectOptions{
@@ -593,17 +798,41 @@ func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLa
 		SkipErrs: *r.Source.Snowball.SkipErrs,
 	}
 
+	uploadCtx, cancelUpload := context.WithCancel(ctx)
+	defer cancelUpload()
+	producerDone := make(chan struct{})
 	go func() {
+		defer close(producerDone)
 		defer xioutil.SafeClose(input)
 
 		for _, entry := range entries {
-			gr, err := objAPI.GetObjectNInfo(ctx, r.Source.Bucket,
+			gr, err := objAPI.GetObjectNInfo(uploadCtx, r.Source.Bucket,
 				entry.Name, nil, nil, ObjectOptions{
 					VersionID: entry.VersionID,
 				})
 			if err != nil {
 				batchLogIf(ctx, err)
-				continue
+				if isErrObjectNotFound(err) || isErrVersionNotFound(err) || *r.Source.Snowball.SkipErrs {
+					continue
+				}
+				// Not sent and not skippable: the batch goes to the
+				// per-object path.
+				readErr = err
+				return
+			}
+			// Each entry is a decision of its own, for the submitter as it is,
+			// on the metadata of all copies of the version.
+			logical, lerr := batchAuthoritativeRead(uploadCtx, objAPI, r.Source.Bucket, entry.Name, gr.ObjInfo)
+			if lerr != nil {
+				// Not sent: the batch goes to the per-object path.
+				gr.Close()
+				readErr = lerr
+				return
+			}
+			refused = sub.authorize(uploadCtx, append(r.pushAuthzRequests(gr.ObjInfo), r.pushAuthzRequests(logical)...)...)
+			if refused != nil {
+				gr.Close()
+				return
 			}
 
 			if prefix != "" {
@@ -625,6 +854,7 @@ func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLa
 
 			opts, _, err := batchReplicationOpts(ctx, "", gr.ObjInfo)
 			if err != nil {
+				gr.Close()
 				batchLogIf(ctx, err)
 				continue
 			}
@@ -635,23 +865,75 @@ func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLa
 				}
 			}
 
-			input <- snowballObj
+			select {
+			case input <- snowballObj:
+			case <-uploadCtx.Done():
+				gr.Close()
+				return
+			}
 		}
 	}()
 
 	// Collect and upload all entries.
-	return remoteClnt.PutObjectsSnowball(ctx, r.Target.Bucket, opts, input)
+	err := remoteClnt.PutObjectsSnowball(uploadCtx, r.Target.Bucket, opts, input)
+	cancelUpload()
+	<-producerDone
+	if refused != nil {
+		return refused
+	}
+	if readErr != nil {
+		return readErr
+	}
+	return err
+}
+
+// pushAuthzRequests are the reads of one source version that a push copies to
+// the target: the version itself, and its tags, retention and legal hold.
+func (r *BatchJobReplicateV1) pushAuthzRequests(oi ObjectInfo) []batchAuthzRequest {
+	bucket, object, versionID := r.Source.Bucket, oi.Name, oi.VersionID
+	if versionID == "" {
+		versionID = nullVersionID
+	}
+	reqs := []batchAuthzRequest{
+		{action: policy.GetObjectAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags},
+		{action: policy.GetObjectVersionAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags},
+	}
+	if oi.UserTags != "" {
+		reqs = append(reqs,
+			batchAuthzRequest{action: policy.GetObjectTaggingAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags},
+			batchAuthzRequest{action: policy.GetObjectVersionTaggingAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags},
+		)
+	}
+	meta := caseInsensitiveMap(oi.UserDefined)
+	_, mode := meta.Lookup(xhttp.AmzObjectLockMode)
+	_, until := meta.Lookup(xhttp.AmzObjectLockRetainUntilDate)
+	if mode || until {
+		reqs = append(reqs, batchAuthzRequest{action: policy.GetObjectRetentionAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags})
+	}
+	if _, hold := meta.Lookup(xhttp.AmzObjectLockLegalHold); hold {
+		reqs = append(reqs, batchAuthzRequest{action: policy.GetObjectLegalHoldAction, bucket: bucket, object: object, versionID: versionID, tags: oi.UserTags})
+	}
+	return reqs
 }
 
 // ReplicateToTarget read from source and replicate to configured target
-func (r *BatchJobReplicateV1) ReplicateToTarget(ctx context.Context, api ObjectLayer, c *minio.Core, srcObjInfo ObjectInfo, retry bool) error {
+func (r *BatchJobReplicateV1) ReplicateToTarget(ctx context.Context, api ObjectLayer, c *minio.Core, srcObjInfo ObjectInfo, retry bool, sub *batchJobSubmitter) error {
 	srcBucket := r.Source.Bucket
 	tgtBucket := r.Target.Bucket
 	tgtPrefix := r.Target.Prefix
 	srcObject := srcObjInfo.Name
 	s3Type := r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3
 
+	p, err := sub.principal(ctx)
+	if err != nil {
+		return err
+	}
 	if srcObjInfo.DeleteMarker || !srcObjInfo.VersionPurgeStatus.Empty() {
+		for _, req := range r.pushAuthzRequests(srcObjInfo) {
+			if err := p.authorize(req); err != nil {
+				return err
+			}
+		}
 		if retry && !s3Type {
 			if _, err := c.StatObject(ctx, tgtBucket, pathJoin(tgtPrefix, srcObject), minio.StatObjectOptions{
 				VersionID: srcObjInfo.VersionID,
@@ -707,6 +989,19 @@ func (r *BatchJobReplicateV1) ReplicateToTarget(ctx context.Context, api ObjectL
 	}
 	defer rd.Close()
 	objInfo := rd.ObjInfo
+	// Authorize the version as it is read, with the metadata of all its
+	// copies.
+	logical, err := batchAuthoritativeRead(ctx, api, srcBucket, srcObject, objInfo)
+	if err != nil {
+		return err
+	}
+	// What is sent comes from the opened copy; both it and the version as
+	// all copies describe it must be authorized.
+	for _, req := range append(r.pushAuthzRequests(objInfo), r.pushAuthzRequests(logical)...) {
+		if err := p.authorize(req); err != nil {
+			return err
+		}
+	}
 
 	size, err := objInfo.GetActualSize()
 	if err != nil {
@@ -760,6 +1055,16 @@ type batchJobInfo struct {
 	DeleteMarkersFailed int64 `json:"deleteMarkersFailed" msg:"dmf"`
 	BytesTransferred    int64 `json:"bytesTransferred" msg:"bt"`
 	BytesFailed         int64 `json:"bytesFailed" msg:"bf"`
+
+	// Versions the job left in place because Object Lock protects them,
+	// and keys it refused as a whole for the same reason. A job that
+	// completes with them has not acted on those versions.
+	VersionsSkippedLocked int64 `json:"versionsSkippedLocked" msg:"vsl"`
+	KeysRefused           int64 `json:"keysRefused" msg:"kr"`
+
+	// 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"`
 }
 
 const (
@@ -901,40 +1206,31 @@ func (ri *batchJobInfo) clone() *batchJobInfo {
 		BytesTransferred:    ri.BytesTransferred,
 		BytesFailed:         ri.BytesFailed,
 		Attempts:            ri.Attempts,
+
+		VersionsSkippedLocked: ri.VersionsSkippedLocked,
+		KeysRefused:           ri.KeysRefused,
+		FailureReason:         ri.FailureReason,
 	}
 }
 
-func (ri *batchJobInfo) countItem(size int64, dmarker, success bool, attempt int) {
+// countItem counts the outcome of one item in the current attempt of a
+// replicate job, which resets its counters before it retries.
+func (ri *batchJobInfo) countItem(size int64, dmarker, success bool) {
 	if ri == nil {
 		return
 	}
 	ri.Attempts++
-	if success {
+	switch {
-		if dmarker {
+	case success && dmarker:
-			ri.DeleteMarkers++
+		ri.DeleteMarkers++
-		} else {
+	case success:
-			ri.Objects++
+		ri.Objects++
-			ri.BytesTransferred += size
+		ri.BytesTransferred += size
-		}
+	case dmarker:
-		if attempt > 1 {
+		ri.DeleteMarkersFailed++
-			if dmarker {
+	default:
-				ri.DeleteMarkersFailed--
+		ri.ObjectsFailed++
-			} else {
+		ri.BytesFailed += size
-				ri.ObjectsFailed--
-				ri.BytesFailed += size
-			}
-		}
-	} else {
-		if attempt > 1 {
-			// Only count first attempt
-			return
-		}
-		if dmarker {
-			ri.DeleteMarkersFailed++
-		} else {
-			ri.ObjectsFailed++
-			ri.BytesFailed += size
-		}
 	}
 }
 
@@ -1002,35 +1298,7 @@ func (ri *batchJobInfo) expireFailures() bool {
 	return ri.ObjectsFailed > 0 || ri.DeleteMarkersFailed > 0
 }
 
-// trackExpireAll counts the objects and delete markers that an expiry of all
+func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo, success bool) {
-// versions of info removed and, when it failed, those the walk saw that it
-// did not remove, at least one. Only batch-expire uses it, for a rule that
-// expires all versions of an object.
-func (ri *batchJobInfo) trackExpireAll(info expireObjInfo, objects, markers int64, failed bool) {
-	if ri == nil {
-		return
-	}
-
-	ri.mu.Lock()
-	defer ri.mu.Unlock()
-
-	ri.Objects += objects
-	ri.DeleteMarkers += markers
-	if !failed {
-		ri.Bucket = info.Bucket
-		ri.Object = info.Name
-		return
-	}
-	failedObjects := max(int64(info.NumVersions)-info.DeleteMarkerCount-objects, 0)
-	failedMarkers := max(info.DeleteMarkerCount-markers, 0)
-	if failedObjects+failedMarkers == 0 {
-		failedObjects = 1
-	}
-	ri.ObjectsFailed += failedObjects
-	ri.DeleteMarkersFailed += failedMarkers
-}
-
-func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo, success bool, attempt int) {
 	if ri == nil {
 		return
 	}
@@ -1042,7 +1310,7 @@ func (ri *batchJobInfo) trackCurrentBucketObject(bucket string, info ObjectInfo,
 		ri.Bucket = bucket
 		ri.Object = info.Name
 	}
-	ri.countItem(info.Size, info.DeleteMarker, success, attempt)
+	ri.countItem(info.Size, info.DeleteMarker, success)
 }
 
 func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInfo) {
@@ -1056,7 +1324,7 @@ func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInf
 	ri.Bucket = bucket
 	for i := range batch {
 		ri.Object = batch[i].Name
-		ri.countItem(batch[i].Size, batch[i].DeleteMarker, true, 1)
+		ri.countItem(batch[i].Size, batch[i].DeleteMarker, true)
 	}
 }
 
@@ -1073,6 +1341,9 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 	if ri.Complete {
 		return nil
 	}
+	if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
+		return err
+	}
 	globalBatchJobsMetrics.save(job.ID, ri)
 	lastObject := ri.Object
 
@@ -1180,6 +1451,12 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 			walkCh = make(chan itemOrErr[ObjectInfo], 100)
 			slowCh = make(chan itemOrErr[ObjectInfo], 100)
 		)
+		ctx, cancelCause := context.WithCancelCause(ctx)
+		// refuse ends the job for its submitter: record why before stopping.
+		refuse := func(err error) {
+			ri.refuse(err)
+			cancelCause(err)
+		}
 
 		if r.Source.Snowball.Disable != nil && !*r.Source.Snowball.Disable && r.Source.Type.isMinio() && r.Target.Type.isMinio() {
 			go func() {
@@ -1201,7 +1478,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 				batch := make([]ObjectInfo, 0, *r.Source.Snowball.Batch)
 				writeFn := func(batch []ObjectInfo) {
 					if len(batch) > 0 {
-						if err := r.writeAsArchive(ctx, api, cl, batch, r.Target.Prefix); err != nil {
+						if err := r.writeAsArchive(ctx, api, cl, batch, r.Target.Prefix, job.Submitter); err != nil {
+							if isBatchJobAuthzError(err) {
+								refuse(err)
+								return
+							}
 							batchLogOnceIf(ctx, err, job.ID+"writeAsArchive")
 							for _, b := range batch {
 								slowCh <- itemOrErr[ObjectInfo]{Item: b}
@@ -1250,7 +1531,6 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		if walkQuorum == "" {
 			walkQuorum = "strict"
 		}
-		ctx, cancelCause := context.WithCancelCause(ctx)
 		// one of source/target is s3, skip delete marker and all versions under the same object name.
 		s3Type := r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3
 
@@ -1281,6 +1561,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 
 		skipReplicate := false
 		for res := range slowCh {
+			if context.Cause(ctx) != nil {
+				// Refused or canceled: schedule nothing more.
+				continue
+			}
 			if res.Err != nil {
 				ri.Failed = true
 				batchLogOnceIf(ctx, res.Err, job.ID+"res.Err")
@@ -1300,7 +1584,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 
 				stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts)
 				success := true
-				if err := r.ReplicateToTarget(ctx, api, c, result, retry); err != nil {
+				if err := r.ReplicateToTarget(ctx, api, c, result, retry, job.Submitter); err != nil {
+					if isBatchJobAuthzError(err) {
+						refuse(err)
+						return
+					}
 					if minio.ToErrorResponse(err).Code == "PreconditionFailed" {
 						// pre-condition failed means we already have the object copied over.
 						return
@@ -1315,7 +1603,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 				} else {
 					stopFn(result, nil)
 				}
-				ri.trackCurrentBucketObject(r.Source.Bucket, result, success, attempts)
+				ri.trackCurrentBucketObject(r.Source.Bucket, result, success)
 				globalBatchJobsMetrics.save(job.ID, ri)
 				// persist in-memory state to disk after every 10secs.
 				batchLogOnceIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job), job.ID+"updateAfter2")
@@ -1326,13 +1614,22 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 			}()
 		}
 		wk.Wait()
+		if ri.refused() {
+			cancelCause(nil)
+			globalBatchJobsMetrics.save(job.ID, ri)
+			batchLogOnceIf(ctx, ri.updateAfter(GlobalContext, api, 0, job), job.ID+"updateAfter3")
+			if err := r.Notify(GlobalContext, ri); err != nil {
+				batchLogOnceIf(ctx, fmt.Errorf("unable to notify %v", err), job.ID+"notify")
+			}
+			return nil
+		}
 		// Do not need to retry if we can't list objects on source.
 		if context.Cause(ctx) != nil {
 			return context.Cause(ctx)
 		}
 		ri.RetryAttempts = attempts
-		ri.Complete = ri.ObjectsFailed == 0
+		ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0
-		ri.Failed = ri.ObjectsFailed > 0
+		ri.Failed = !ri.Complete
 
 		globalBatchJobsMetrics.save(job.ID, ri)
 		// persist in-memory state to disk.
@@ -1345,9 +1642,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		cancelCause(nil)
 		if ri.Failed {
 			ri.ObjectsFailed = 0
+			ri.DeleteMarkersFailed = 0
 			ri.Bucket = ""
 			ri.Object = ""
 			ri.Objects = 0
+			ri.DeleteMarkers = 0
 			ri.BytesFailed = 0
 			ri.BytesTransferred = 0
 			retry = true // indicate we are retrying..
@@ -1549,8 +1848,12 @@ func (j BatchJobRequest) Validate(ctx context.Context, o ObjectLayer) error {
 	return errInvalidArgument
 }
 
-func (j BatchJobRequest) delete(ctx context.Context, api ObjectLayer) {
+func (j BatchJobRequest) delete(ctx context.Context, api ObjectLayer) error {
-	deleteConfig(ctx, api, getJobPath(j))
+	err := deleteConfig(ctx, api, getJobPath(j))
+	if errors.Is(err, errConfigNotFound) {
+		return nil
+	}
+	return err
 }
 
 func (j BatchJobRequest) getJobReportPath() (string, error) {
@@ -1709,7 +2012,7 @@ func (a adminAPIHandlers) BatchJobStatus(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
-	buf, err := json.Marshal(madmin.BatchJobStatus{LastMetric: ri.metric()})
+	buf, err := json.Marshal(batchJobStatus{LastMetric: ri.statusMetric()})
 	if err != nil {
 		batchLogIf(ctx, err)
 		writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
@@ -1747,8 +2050,13 @@ func (a adminAPIHandlers) DescribeBatchJob(w http.ResponseWriter, r *http.Reques
 	}
 
 	// Remove sensitive fields.
-	req.RedactSensitive()
+	out, err := req.redacted()
-	buf, err := yaml.Marshal(req)
+	if err != nil {
+		batchLogIf(ctx, err)
+		writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+	buf, err := yaml.Marshal(out)
 	if err != nil {
 		batchLogIf(ctx, err)
 		writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
@@ -1806,6 +2114,27 @@ func (a adminAPIHandlers) StartBatchJob(w http.ResponseWriter, r *http.Request)
 		}
 	}
 
+	if job.jobSections() != 1 {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrInvalidRequest, errBatchJobSections), r.URL)
+		return
+	}
+
+	// The job acts as the credential that submits it, and only as far as
+	// that credential may act on the job's buckets.
+	submitter, err := newBatchJobSubmitter(creds)
+	if err == nil {
+		job.Submitter = submitter
+		err = submitter.authorize(ctx, job.startAuthzRequests()...)
+	}
+	if err != nil {
+		apiErr := toAPIError(ctx, err)
+		if isBatchJobAuthzError(err) {
+			apiErr = errorCodes.ToAPIErrWithErr(ErrAccessDenied, err)
+		}
+		writeErrorResponseJSON(ctx, w, apiErr, r.URL)
+		return
+	}
+
 	//  Validate the incoming job request
 	if err := job.Validate(ctx, objectAPI); err != nil {
 		writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
@@ -1942,6 +2271,14 @@ func (j *BatchJobPool) cleanupReports(randomWait func() time.Duration) {
 					continue
 				}
 				if (ri.Complete || ri.Failed) && time.Since(ri.LastUpdate) > oldJobsExpiration {
+					// Remove the job before its report, so that a finished
+					// job is never resumed without its outcome.
+					if ri.JobID != "" {
+						if err := (BatchJobRequest{ID: ri.JobID}).delete(ctx, j.objLayer); err != nil {
+							batchLogIf(ctx, err)
+							continue
+						}
+					}
 					deleteConfig(ctx, j.objLayer, result.Item.Name)
 				}
 			}
@@ -2082,7 +2419,7 @@ func (j *BatchJobPool) queueJob(req *BatchJobRequest) error {
 		})
 	case j.jobCh <- req:
 	default:
-		return fmt.Errorf("batch job queue is currently full please try again later %#v", req)
+		return fmt.Errorf("batch job queue is currently full please try again later (job %s)", req.ID)
 	}
 	return nil
 }

cmd/batch-handlers_gen.go

@@ -179,6 +179,24 @@ func (z *BatchJobRequest) DecodeMsg(dc *msgp.Reader) (err error) {
 					return
 				}
 			}
+		case "Submitter":
+			if dc.IsNil() {
+				err = dc.ReadNil()
+				if err != nil {
+					err = msgp.WrapError(err, "Submitter")
+					return
+				}
+				z.Submitter = nil
+			} else {
+				if z.Submitter == nil {
+					z.Submitter = new(batchJobSubmitter)
+				}
+				err = z.Submitter.DecodeMsg(dc)
+				if err != nil {
+					err = msgp.WrapError(err, "Submitter")
+					return
+				}
+			}
 		default:
 			err = dc.Skip()
 			if err != nil {
@@ -192,9 +210,9 @@ func (z *BatchJobRequest) DecodeMsg(dc *msgp.Reader) (err error) {
 
 // EncodeMsg implements msgp.Encodable
 func (z *BatchJobRequest) EncodeMsg(en *msgp.Writer) (err error) {
-	// map header, size 6
+	// map header, size 7
 	// write "ID"
-	err = en.Append(0x86, 0xa2, 0x49, 0x44)
+	err = en.Append(0x87, 0xa2, 0x49, 0x44)
 	if err != nil {
 		return
 	}
@@ -274,15 +292,32 @@ func (z *BatchJobRequest) EncodeMsg(en *msgp.Writer) (err error) {
 			return
 		}
 	}
+	// write "Submitter"
+	err = en.Append(0xa9, 0x53, 0x75, 0x62, 0x6d, 0x69, 0x74, 0x74, 0x65, 0x72)
+	if err != nil {
+		return
+	}
+	if z.Submitter == nil {
+		err = en.WriteNil()
+		if err != nil {
+			return
+		}
+	} else {
+		err = z.Submitter.EncodeMsg(en)
+		if err != nil {
+			err = msgp.WrapError(err, "Submitter")
+			return
+		}
+	}
 	return
 }
 
 // MarshalMsg implements msgp.Marshaler
 func (z *BatchJobRequest) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 6
+	// map header, size 7
 	// string "ID"
-	o = append(o, 0x86, 0xa2, 0x49, 0x44)
+	o = append(o, 0x87, 0xa2, 0x49, 0x44)
 	o = msgp.AppendString(o, z.ID)
 	// string "User"
 	o = append(o, 0xa4, 0x55, 0x73, 0x65, 0x72)
@@ -323,6 +358,17 @@ func (z *BatchJobRequest) MarshalMsg(b []byte) (o []byte, err error) {
 			return
 		}
 	}
+	// string "Submitter"
+	o = append(o, 0xa9, 0x53, 0x75, 0x62, 0x6d, 0x69, 0x74, 0x74, 0x65, 0x72)
+	if z.Submitter == nil {
+		o = msgp.AppendNil(o)
+	} else {
+		o, err = z.Submitter.MarshalMsg(o)
+		if err != nil {
+			err = msgp.WrapError(err, "Submitter")
+			return
+		}
+	}
 	return
 }
 
@@ -413,6 +459,23 @@ func (z *BatchJobRequest) UnmarshalMsg(bts []byte) (o []byte, err error) {
 					return
 				}
 			}
+		case "Submitter":
+			if msgp.IsNil(bts) {
+				bts, err = msgp.ReadNilBytes(bts)
+				if err != nil {
+					return
+				}
+				z.Submitter = nil
+			} else {
+				if z.Submitter == nil {
+					z.Submitter = new(batchJobSubmitter)
+				}
+				bts, err = z.Submitter.UnmarshalMsg(bts)
+				if err != nil {
+					err = msgp.WrapError(err, "Submitter")
+					return
+				}
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -445,6 +508,12 @@ func (z *BatchJobRequest) Msgsize() (s int) {
 	} else {
 		s += z.Expire.Msgsize()
 	}
+	s += 10
+	if z.Submitter == nil {
+		s += msgp.NilSize
+	} else {
+		s += z.Submitter.Msgsize()
+	}
 	return
 }
 
@@ -568,6 +637,24 @@ func (z *batchJobInfo) DecodeMsg(dc *msgp.Reader) (err error) {
 				err = msgp.WrapError(err, "BytesFailed")
 				return
 			}
+		case "vsl":
+			z.VersionsSkippedLocked, err = dc.ReadInt64()
+			if err != nil {
+				err = msgp.WrapError(err, "VersionsSkippedLocked")
+				return
+			}
+		case "kr":
+			z.KeysRefused, err = dc.ReadInt64()
+			if err != nil {
+				err = msgp.WrapError(err, "KeysRefused")
+				return
+			}
+		case "fr":
+			z.FailureReason, err = dc.ReadString()
+			if err != nil {
+				err = msgp.WrapError(err, "FailureReason")
+				return
+			}
 		default:
 			err = dc.Skip()
 			if err != nil {
@@ -581,9 +668,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 17
+	// map header, size 20
 	// write "v"
-	err = en.Append(0xde, 0x0, 0x11, 0xa1, 0x76)
+	err = en.Append(0xde, 0x0, 0x14, 0xa1, 0x76)
 	if err != nil {
 		return
 	}
@@ -752,15 +839,45 @@ func (z *batchJobInfo) EncodeMsg(en *msgp.Writer) (err error) {
 		err = msgp.WrapError(err, "BytesFailed")
 		return
 	}
+	// write "vsl"
+	err = en.Append(0xa3, 0x76, 0x73, 0x6c)
+	if err != nil {
+		return
+	}
+	err = en.WriteInt64(z.VersionsSkippedLocked)
+	if err != nil {
+		err = msgp.WrapError(err, "VersionsSkippedLocked")
+		return
+	}
+	// write "kr"
+	err = en.Append(0xa2, 0x6b, 0x72)
+	if err != nil {
+		return
+	}
+	err = en.WriteInt64(z.KeysRefused)
+	if err != nil {
+		err = msgp.WrapError(err, "KeysRefused")
+		return
+	}
+	// write "fr"
+	err = en.Append(0xa2, 0x66, 0x72)
+	if err != nil {
+		return
+	}
+	err = en.WriteString(z.FailureReason)
+	if err != nil {
+		err = msgp.WrapError(err, "FailureReason")
+		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 17
+	// map header, size 20
 	// string "v"
-	o = append(o, 0xde, 0x0, 0x11, 0xa1, 0x76)
+	o = append(o, 0xde, 0x0, 0x14, 0xa1, 0x76)
 	o = msgp.AppendInt(o, z.Version)
 	// string "jid"
 	o = append(o, 0xa3, 0x6a, 0x69, 0x64)
@@ -810,6 +927,15 @@ func (z *batchJobInfo) MarshalMsg(b []byte) (o []byte, err error) {
 	// string "bf"
 	o = append(o, 0xa2, 0x62, 0x66)
 	o = msgp.AppendInt64(o, z.BytesFailed)
+	// string "vsl"
+	o = append(o, 0xa3, 0x76, 0x73, 0x6c)
+	o = msgp.AppendInt64(o, z.VersionsSkippedLocked)
+	// string "kr"
+	o = append(o, 0xa2, 0x6b, 0x72)
+	o = msgp.AppendInt64(o, z.KeysRefused)
+	// string "fr"
+	o = append(o, 0xa2, 0x66, 0x72)
+	o = msgp.AppendString(o, z.FailureReason)
 	return
 }
 
@@ -933,6 +1059,24 @@ func (z *batchJobInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "BytesFailed")
 				return
 			}
+		case "vsl":
+			z.VersionsSkippedLocked, bts, err = msgp.ReadInt64Bytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "VersionsSkippedLocked")
+				return
+			}
+		case "kr":
+			z.KeysRefused, bts, err = msgp.ReadInt64Bytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "KeysRefused")
+				return
+			}
+		case "fr":
+			z.FailureReason, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "FailureReason")
+				return
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -947,6 +1091,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
+	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)
 	return
 }

cmd/batch-progress.go

@@ -0,0 +1,187 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"slices"
+	"strings"
+	"sync"
+
+	"github.com/minio/madmin-go/v3"
+)
+
+// batchCounts are outcomes of a batch job that a later run of the job would
+// see again, and so must be persisted together with the cursor of the walk.
+type batchCounts struct {
+	objects             int64
+	objectsFailed       int64
+	deleteMarkersFailed int64
+	versionsSkipped     int64
+	keysRefused         int64
+}
+
+func (c *batchCounts) add(o batchCounts) {
+	c.objects += o.objects
+	c.objectsFailed += o.objectsFailed
+	c.deleteMarkersFailed += o.deleteMarkersFailed
+	c.versionsSkipped += o.versionsSkipped
+	c.keysRefused += o.keysRefused
+}
+
+//msgp:ignore batchProgress batchProgressItem batchCounts
+
+// batchProgress is the cursor of a walk whose items are done concurrently.
+//
+// The items of a walk are numbered in walk order. A key is done when every
+// item of it, and every item before it, is done; only then does the cursor
+// move past it and its counts become part of the job's counts. The cursor and
+// the counts are persisted together, so a job resumed after the cursor
+// neither skips an item that was not done nor counts an item twice.
+type batchProgress struct {
+	mu      sync.Mutex
+	ri      *batchJobInfo
+	bucket  string
+	issued  uint64
+	next    uint64
+	pending map[uint64]batchProgressItem
+	// key is the key of the item last committed, and held are the counts of
+	// its items so far.
+	key  string
+	held batchCounts
+}
+
+type batchProgressItem struct {
+	key    string
+	counts batchCounts
+}
+
+func newBatchProgress(ri *batchJobInfo, bucket string) *batchProgress {
+	return &batchProgress{ri: ri, bucket: bucket, pending: make(map[uint64]batchProgressItem)}
+}
+
+// begin numbers the next item of the walk.
+func (p *batchProgress) begin() uint64 {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	seq := p.issued
+	p.issued++
+	return seq
+}
+
+// done records item seq, of key, as done with counts.
+func (p *batchProgress) done(seq uint64, key string, counts batchCounts) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.pending[seq] = batchProgressItem{key: key, counts: counts}
+	for {
+		item, ok := p.pending[p.next]
+		if !ok {
+			return
+		}
+		delete(p.pending, p.next)
+		p.next++
+		if item.key != p.key {
+			p.commitLocked()
+			p.key = item.key
+		}
+		p.held.add(item.counts)
+	}
+}
+
+// finish commits the last key if every item of the walk is done. Call it
+// once no item is in progress.
+func (p *batchProgress) finish() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.next == p.issued {
+		p.commitLocked()
+	}
+}
+
+func (p *batchProgress) commitLocked() {
+	if p.key == "" {
+		return
+	}
+	ri := p.ri
+	ri.mu.Lock()
+	ri.Bucket = p.bucket
+	ri.Object = p.key
+	ri.Objects += p.held.objects
+	ri.ObjectsFailed += p.held.objectsFailed
+	ri.DeleteMarkersFailed += p.held.deleteMarkersFailed
+	ri.VersionsSkippedLocked += p.held.versionsSkipped
+	ri.KeysRefused += p.held.keysRefused
+	ri.mu.Unlock()
+	p.held = batchCounts{}
+}
+
+// countRemoved counts versions a job removed. A removed version is gone for
+// any later run, so it is counted at once, not at the cursor.
+func (ri *batchJobInfo) countRemoved(bucket string, objects, markers int64) {
+	ri.mu.Lock()
+	defer ri.mu.Unlock()
+	ri.Bucket = bucket
+	ri.Objects += objects
+	ri.DeleteMarkers += markers
+}
+
+// countSkipped counts a version a job left in place because Object Lock
+// protects it, for a job that is counted per run rather than at a cursor.
+func (ri *batchJobInfo) countSkipped() {
+	ri.mu.Lock()
+	defer ri.mu.Unlock()
+	ri.VersionsSkippedLocked++
+}
+
+// batchCanonicalPrefixes orders the prefixes of a walk and drops a prefix
+// that another one covers. The keys of the walk are then visited once each,
+// in one ascending order, which a single cursor can describe.
+func batchCanonicalPrefixes(prefixes []string) []string {
+	if len(prefixes) == 0 {
+		return []string{""}
+	}
+	sorted := slices.Clone(prefixes)
+	slices.Sort(sorted)
+	out := sorted[:0]
+	for _, p := range sorted {
+		if len(out) > 0 && strings.HasPrefix(p, out[len(out)-1]) {
+			continue
+		}
+		out = append(out, p)
+	}
+	return out
+}
+
+// batchResumeAfter reports whether a resumed walk returned a key after the
+// cursor. The cursor key is done; the walk forwards conservatively and
+// returns it again. Keys compare as the walk orders them, by name, so a
+// directory object comes before the keys under it.
+func batchResumeAfter(cursor, name string) bool {
+	return cursor == "" || name > cursor
+}
+
+// batchJobStatus is the status response: madmin.BatchJobStatus with the
+// fields this server adds. The added fields are additive; a client that does
+// not know them ignores them.
+type batchJobStatus struct {
+	LastMetric batchJobStatusMetric
+}
+
+type batchJobStatusMetric struct {
+	madmin.JobMetric
+	VersionsSkippedLocked int64  `json:"versionsSkippedLocked,omitempty"`
+	KeysRefused           int64  `json:"keysRefused,omitempty"`
+	FailureReason         string `json:"failureReason,omitempty"`
+}
+
+func (ri *batchJobInfo) statusMetric() batchJobStatusMetric {
+	ri.mu.RLock()
+	defer ri.mu.RUnlock()
+	return batchJobStatusMetric{
+		JobMetric:             ri.metric(),
+		VersionsSkippedLocked: ri.VersionsSkippedLocked,
+		KeysRefused:           ri.KeysRefused,
+		FailureReason:         ri.FailureReason,
+	}
+}

cmd/batch-rotate.go

@@ -35,6 +35,7 @@ import (
 	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/kms"
 	"github.com/pgsty/silo-pkg/v3/env"
+	"github.com/pgsty/silo-pkg/v3/policy"
 	"github.com/pgsty/silo-pkg/v3/workers"
 )
 
@@ -161,8 +162,68 @@ func (r BatchJobKeyRotateV1) Notify(ctx context.Context, ri *batchJobInfo) error
 	return notifyEndpoint(ctx, ri, r.Flags.Notify.Endpoint, r.Flags.Notify.Token)
 }
 
-// KeyRotate rotates encryption key of an object
+// batchAuthoritativeVersion returns the version obj as all its copies
-func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, objInfo ObjectInfo) error {
+// describe it. With several pools a version can have copies in more than one
+// pool, and a read returns one of them; the merged copy carries the newest
+// Object Lock and tag state. The caller holds the object's write lock.
+func batchAuthoritativeVersion(ctx context.Context, api ObjectLayer, bucket, object string, obj ObjectInfo) (ObjectInfo, error) {
+	z, ok := api.(*erasureServerPools)
+	if !ok || z.SinglePool() {
+		return obj, nil
+	}
+	versionID := obj.VersionID
+	if versionID == "" {
+		versionID = nullVersionID
+	}
+	return z.replicaObjectInfo(ctx, bucket, encodeDirObject(object), ObjectOptions{VersionID: versionID, NoLock: true, NoAuditLog: true})
+}
+
+// batchAuthoritativeRead is batchAuthoritativeVersion for a version that has
+// been opened for reading. It takes the object's read lock itself, since an
+// opened reader may hold none.
+func batchAuthoritativeRead(ctx context.Context, api ObjectLayer, bucket, object string, obj ObjectInfo) (ObjectInfo, error) {
+	z, ok := api.(*erasureServerPools)
+	if !ok || z.SinglePool() {
+		return obj, nil
+	}
+	lk := z.NewNSLock(bucket, encodeDirObject(object))
+	lkctx, err := lk.GetRLock(ctx, globalOperationTimeout)
+	if err != nil {
+		return ObjectInfo{}, err
+	}
+	defer lk.RUnlock(lkctx)
+	return batchAuthoritativeVersion(lkctx.Context(), api, bucket, object, obj)
+}
+
+// rotateAuthzRequests are the actions of rotating one version: it is read and
+// written again in place with the new key.
+func (r *BatchJobKeyRotateV1) rotateAuthzRequests(oi ObjectInfo) []batchAuthzRequest {
+	versionID := oi.VersionID
+	if versionID == "" {
+		versionID = nullVersionID
+	}
+	// The encryption the rewrite leaves: rotation keeps the kind of the
+	// version (rotateKey), and SSE-KMS takes the requested key.
+	values := map[string][]string{}
+	switch {
+	case crypto.S3KMS.IsEncrypted(oi.UserDefined):
+		values[strings.ToLower(xhttp.AmzServerSideEncryption)] = []string{xhttp.AmzEncryptionKMS}
+		if r.Encryption.Type == ssekms && r.Encryption.Key != "" {
+			values[strings.ToLower(xhttp.AmzServerSideEncryptionKmsID)] = []string{r.Encryption.Key}
+		}
+	case crypto.S3.IsEncrypted(oi.UserDefined):
+		values[strings.ToLower(xhttp.AmzServerSideEncryption)] = []string{xhttp.AmzEncryptionAES}
+	}
+	return []batchAuthzRequest{
+		{action: policy.GetObjectAction, bucket: r.Bucket, object: oi.Name, versionID: versionID, tags: oi.UserTags},
+		{action: policy.GetObjectVersionAction, bucket: r.Bucket, object: oi.Name, versionID: versionID, tags: oi.UserTags},
+		{action: policy.PutObjectAction, bucket: r.Bucket, object: oi.Name, versionID: versionID, tags: oi.UserTags, values: values},
+	}
+}
+
+// KeyRotate rotates encryption key of an object. A version that Object Lock
+// protects is not rewritten: it returns ObjectLocked.
+func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, objInfo ObjectInfo, sub *batchJobSubmitter) error {
 	srcBucket := r.Bucket
 	srcObject := objInfo.Name
 
@@ -180,7 +241,8 @@ func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, ob
 	versioned := globalBucketVersioningSys.PrefixEnabled(srcBucket, srcObject)
 	versionSuspended := globalBucketVersioningSys.PrefixSuspended(srcBucket, srcObject)
 
-	lock := api.NewObjectNSLock(r.Bucket, objInfo.Name)
+	// Lock the name writers lock: the stored name of a directory object.
+	lock := api.NewObjectNSLock(r.Bucket, encodeDirObject(objInfo.Name))
 	lkctx, err := lock.GetLock(ctx, globalOperationTimeout)
 	if err != nil {
 		return err
@@ -198,6 +260,24 @@ func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, ob
 	if err != nil {
 		return err
 	}
+	// Decide on the version as it is under the lock that the rewrite holds,
+	// before any key is requested, with the metadata of all its copies.
+	logical, err := batchAuthoritativeVersion(ctx, api, r.Bucket, objInfo.Name, obj)
+	if err != nil {
+		return err
+	}
+	if enforceRetentionForDeletion(ctx, logical) {
+		return ObjectLocked{Bucket: r.Bucket, Object: obj.Name, VersionID: obj.VersionID}
+	}
+	p, err := sub.principal(ctx)
+	if err != nil {
+		return err
+	}
+	for _, req := range r.rotateAuthzRequests(logical) {
+		if err := p.authorize(req); err != nil {
+			return err
+		}
+	}
 	oi := obj.Clone()
 	var (
 		newKeyID      string
@@ -260,6 +340,9 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 	if ri.Complete {
 		return nil
 	}
+	if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
+		return err
+	}
 
 	globalBatchJobsMetrics.save(job.ID, ri)
 	lastObject := ri.Object
@@ -273,8 +356,6 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		delay = batchKeyRotateJobDefaultRetryDelay
 	}
 
-	rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
-
 	selectObj := func(info FileInfo) (ok bool) {
 		if r.Flags.Filter.OlderThan > 0 && time.Since(info.ModTime) < r.Flags.Filter.OlderThan {
 			// skip all objects that are newer than specified older duration
@@ -355,65 +436,92 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		return err
 	}
 
-	ctx, cancel := context.WithCancel(ctx)
+	ctx, cancelCause := context.WithCancelCause(ctx)
+	defer cancelCause(nil)
+	// refuse ends the job for its submitter: record why before stopping.
+	refuse := func(err error) {
+		ri.refuse(err)
+		cancelCause(err)
+	}
+	progress := newBatchProgress(ri, r.Bucket)
 
 	results := make(chan itemOrErr[ObjectInfo], 100)
 	if err := api.Walk(ctx, r.Bucket, r.Prefix, results, WalkOptions{
 		Marker: lastObject,
 		Filter: selectObj,
 	}); err != nil {
-		cancel()
 		// Do not need to retry if we can't list objects on source.
 		return err
 	}
 	failed := false
 	for res := range results {
+		if context.Cause(ctx) != nil {
+			// Refused or canceled: schedule nothing more.
+			continue
+		}
 		if res.Err != nil {
 			failed = true
 			batchLogIf(ctx, res.Err)
 			break
 		}
 		result := res.Item
+		if !batchResumeAfter(lastObject, result.Name) {
+			// Completed by the run that set the cursor.
+			continue
+		}
 		sseKMS := crypto.S3KMS.IsEncrypted(result.UserDefined)
 		sseS3 := crypto.S3.IsEncrypted(result.UserDefined)
 		if !sseKMS && !sseS3 { // neither sse-s3 nor sse-kms disallowed
 			continue
 		}
+		seq := progress.begin()
 		wk.Take()
 		go func() {
 			defer wk.Give()
+			var counts batchCounts
 			for attempts := 1; attempts <= retryAttempts; attempts++ {
 				stopFn := globalBatchJobsMetrics.trace(batchJobMetricKeyRotation, job.ID, attempts)
-				success := true
+				err := r.KeyRotate(ctx, api, result, job.Submitter)
-				if err := r.KeyRotate(ctx, api, result); err != nil {
+				stopFn(result, err)
-					stopFn(result, err)
+				if err == nil {
-					batchLogIf(ctx, err)
+					counts.objects = 1
-					success = false
+					break
-					if attempts >= retryAttempts {
-						auditOptions := AuditLogOptions{
-							Event:     "KeyRotate",
-							APIName:   "StartBatchJob",
-							Bucket:    result.Bucket,
-							Object:    result.Name,
-							VersionID: result.VersionID,
-							Error:     err.Error(),
-						}
-						auditLogInternal(ctx, auditOptions)
-					}
-				} else {
-					stopFn(result, nil)
 				}
-				ri.trackCurrentBucketObject(r.Bucket, result, success, attempts)
+				if isBatchJobAuthzError(err) {
-				globalBatchJobsMetrics.save(job.ID, ri)
+					refuse(err)
-				// persist in-memory state to disk after every 10secs.
+					return
-				batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job))
+				}
-				if success {
+				if isErrObjectLocked(err) {
+					// Left as it is, not retried and not a failure.
+					counts.versionsSkipped = 1
+					break
+				}
+				if ctx.Err() != nil {
+					// Not done; a later run rotates it.
+					return
+				}
+				batchLogIf(ctx, err)
+				if attempts >= retryAttempts {
+					auditOptions := AuditLogOptions{
+						Event:     "KeyRotate",
+						APIName:   "StartBatchJob",
+						Bucket:    result.Bucket,
+						Object:    result.Name,
+						VersionID: result.VersionID,
+						Error:     err.Error(),
+					}
+					auditLogInternal(ctx, auditOptions)
+					counts.objectsFailed = 1
 					break
 				}
 				if delay > 0 {
-					time.Sleep(delay + time.Duration(rnd.Float64()*float64(delay)))
+					time.Sleep(delay + time.Duration(rand.Float64()*float64(delay)))
 				}
 			}
+			progress.done(seq, result.Name, counts)
+			globalBatchJobsMetrics.save(job.ID, ri)
+			// persist in-memory state to disk after every 10secs.
+			batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job))
 
 			if wait := globalBatchConfig.KeyRotationWait(); wait > 0 {
 				time.Sleep(wait)
@@ -421,18 +529,31 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 		}()
 	}
 	wk.Wait()
-
+	cause := context.Cause(ctx)
-	ri.Complete = !failed && ri.ObjectsFailed == 0
+	if !failed && cause == nil {
-	ri.Failed = failed || ri.ObjectsFailed > 0
+		// The walk ended: its last key is done.
-	globalBatchJobsMetrics.save(job.ID, ri)
+		progress.finish()
-	// persist in-memory state to disk.
+	}
-	batchLogIf(ctx, ri.updateAfter(ctx, api, 0, job))
+	if cause != nil && !ri.refused() {
-
+		// Stopped before the end: keep the work after the cursor pending.
-	if err := r.Notify(ctx, ri); err != nil {
+		globalBatchJobsMetrics.save(job.ID, ri)
-		batchLogIf(ctx, fmt.Errorf("unable to notify %v", err))
+		batchLogIf(ctx, ri.updateAfter(GlobalContext, api, 0, job))
+		return cause
 	}
 
-	cancel()
+	if !ri.refused() {
+		ri.mu.Lock()
+		ri.Complete = !failed && ri.ObjectsFailed == 0
+		ri.Failed = failed || ri.ObjectsFailed > 0
+		ri.mu.Unlock()
+	}
+	globalBatchJobsMetrics.save(job.ID, ri)
+	// persist in-memory state to disk.
+	batchLogIf(ctx, ri.updateAfter(GlobalContext, api, 0, job))
+
+	if err := r.Notify(GlobalContext, ri); err != nil {
+		batchLogIf(ctx, fmt.Errorf("unable to notify %v", err))
+	}
 	return nil
 }
 

cmd/erasure-server-pool-delete-all-options_test.go

@@ -149,7 +149,11 @@ func TestPurgeAllVersionsBudgetCoversReads(t *testing.T) {
 // A marker that an expiration job failed to remove fails the job.
 func TestPurgeAllVersionsMarkerFailureFailsJob(t *testing.T) {
 	ri := &batchJobInfo{}
-	ri.trackExpireAll(expireObjInfo{ObjectInfo: ObjectInfo{NumVersions: 2}, DeleteMarkerCount: 1}, 1, 0, true)
+	var res batchCounts
+	res.expireAllFailed(expireObjInfo{ObjectInfo: ObjectInfo{Name: "key", NumVersions: 2}, DeleteMarkerCount: 1}, 1, 0)
+	progress := newBatchProgress(ri, "bucket")
+	progress.done(progress.begin(), "key", res)
+	progress.finish()
 	if ri.ObjectsFailed != 0 || ri.DeleteMarkersFailed != 1 || !ri.expireFailures() {
 		t.Errorf("failed=%d/%d failures=%t", ri.ObjectsFailed, ri.DeleteMarkersFailed, ri.expireFailures())
 	}

cmd/erasure-server-pool-delete-all_test.go

@@ -332,8 +332,9 @@ func (p *purgeTest) matched(name string) expireObjInfo {
 	return expireObjInfo{ObjectInfo: oi, ExpireAll: true, DeleteMarkerCount: markers}
 }
 
-// batch runs the deletes of a batch expiration job for walked, the keys its
+// batch runs the deletes of a batch expiration job of the root user for
-// walk matched, and returns the counts of the job.
+// walked, the keys its walk matched, as one batch of a walk that reached its
+// end, and returns the counts of the job.
 func (p *purgeTest) batch(walked ...expireObjInfo) *batchJobInfo {
 	p.t.Helper()
 	globalBatchJobsMetrics.Lock()
@@ -341,7 +342,11 @@ func (p *purgeTest) batch(walked ...expireObjInfo) *batchJobInfo {
 		globalBatchJobsMetrics.metrics = make(map[string]*batchJobInfo)
 	}
 	globalBatchJobsMetrics.Unlock()
-	job := BatchJobRequest{ID: "purge-" + mustGetUUID(), Started: time.Now(), Expire: &BatchJobExpire{Bucket: p.bucket}}
+	sub, err := newBatchJobSubmitter(globalActiveCred)
+	if err != nil {
+		p.t.Fatal(err)
+	}
+	job := BatchJobRequest{ID: "purge-" + mustGetUUID(), Started: time.Now(), Submitter: sub, Expire: &BatchJobExpire{Bucket: p.bucket}}
 	ri := &batchJobInfo{JobID: job.ID}
 	wk, err := workers.New(1)
 	if err != nil {
@@ -350,7 +355,8 @@ func (p *purgeTest) batch(walked ...expireObjInfo) *batchJobInfo {
 	ch := make(chan []expireObjInfo, 1)
 	ch <- walked
 	close(ch)
-	batchObjsForDelete(p.t.Context(), job.Expire, ri, job, p.z, wk, ch)
+	refuse := func(err error) { p.t.Errorf("batch job refused: %v", err) }
+	batchObjsForDelete(p.t.Context(), refuse, func() bool { return true }, job.Expire, ri, job, p.z, wk, ch)
 	wk.Wait()
 	return ri
 }
@@ -837,8 +843,8 @@ func TestPurgeAllVersionsKeepsProtectedWrites(t *testing.T) {
 			t.Errorf("%s holds %v, want both versions", name, got)
 		}
 	}
-	if ri.Objects != 2 || ri.ObjectsFailed == 0 {
+	if ri.Objects != 2 || ri.KeysRefused != 3 || ri.ObjectsFailed != 0 {
-		t.Errorf("job counts objects=%d failed=%d, want 2 and the refusals", ri.Objects, ri.ObjectsFailed)
+		t.Errorf("job counts objects=%d refused=%d failed=%d, want 2, 3 and 0", ri.Objects, ri.KeysRefused, ri.ObjectsFailed)
 	}
 }
 

cmd/erasure-server-pool-lock-domain_test.go

@@ -1013,9 +1013,13 @@ func TestLockDomainKeyRotationWaitsForWriter(t *testing.T) {
 			// The rotation's listing found an SSE-S3 object; it reads it again
 			// under the lock.
 			listed := ObjectInfo{Bucket: bucket, Name: key, UserDefined: map[string]string{crypto.MetaSealedKeyS3: "listed"}}
+			sub, err := newBatchJobSubmitter(globalActiveCred)
+			if err != nil {
+				t.Fatal(err)
+			}
 			waited, _, _ := ldOverlapRun(t,
 				func(ctx context.Context) error { return ldPut(ctx, z, bucket, key, "written") },
-				func(ctx context.Context) error { return job.KeyRotate(ctx, z, listed) })
+				func(ctx context.Context) error { return job.KeyRotate(ctx, z, listed, sub) })
 			if !waited {
 				t.Errorf("key rotation ran while a PUT held its commit")
 			}