fix(batch): authorize the versions that a purge of all versions deletes

A batch expiration that keeps no versions authorized the versions its
walk saw, and the purge then deleted the versions it read under its
lock. A version written after the walk, or tags changed on one since,
was deleted without being authorized for the submitter.

The purge takes an authorization that a batch purge must set and no
other delete may. Under the object lock, once it has approved the
history and checked every drive, and before its first delete, it
hands the writes it is about to delete to the job: every copy of every
approved version and every revision of one that a drive holds, one per
distinct version ID and tags, as they are then. The job evaluates the
expire actions on each for the submitter. A denial deletes nothing and
ends the job as refused; the time it takes counts against the purge's
budget. A locked version still refuses the whole key, counted as a
refused key. The checks on the walk's snapshot are removed.

Reviewed adversarially; the review's regressions are included.

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

cmd/batch-expire-purge-authz_test.go

@@ -0,0 +1,475 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/auth"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+// A batch expiration that keeps no versions purges each key it matched under
+// the purge's lock. What it authorizes for the submitter, and checks for
+// Object Lock, are the versions the purge deletes then, not the versions the
+// walk saw before.
+
+// batchPurgeHook runs change once, after the walk of a job and before its
+// first purge of all versions.
+type batchPurgeHook struct {
+	ObjectLayer
+	once   sync.Once
+	change func()
+}
+
+func (h *batchPurgeHook) DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
+	if opts.DeleteAllVersions {
+		h.once.Do(h.change)
+	}
+	return h.ObjectLayer.DeleteObject(ctx, bucket, object, opts)
+}
+
+func (f *batchAuthzFixture) purgeYAML(prefix string) string {
+	return fmt.Sprintf(`expire:
+  apiVersion: v1
+  bucket: %s
+  prefix: %q
+  rules:
+    - type: object
+      purge:
+        retainVersions: 0
+`, f.bucket, prefix)
+}
+
+// runChanged submits a purge of prefix as cred and runs it, with change made
+// between its walk and its purge, and returns the job's final state.
+func (f *batchAuthzFixture) runChanged(t *testing.T, cred auth.Credentials, prefix string, change func()) *batchJobInfo {
+	t.Helper()
+	rec, job := f.submit(t, cred, f.purgeYAML(prefix))
+	if job == nil {
+		t.Fatalf("submission status %d: %s", rec.Code, rec.Body.String())
+	}
+	hook := &batchPurgeHook{ObjectLayer: f.obj, change: change}
+	if err := job.Expire.Start(f.ctx, hook, *job); err != nil {
+		t.Logf("start: %v", err)
+	}
+	return f.jobInfo(t, job)
+}
+
+// putOlder writes a version of key older than every version it has, as a
+// replicated or pulled version can be, with metadata meta.
+func (f *batchAuthzFixture) putOlder(t *testing.T, key, versionID string, meta map[string]string) {
+	t.Helper()
+	data := []byte("older " + key)
+	_, err := f.obj.PutObject(f.ctx, f.bucket, key, mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""),
+		ObjectOptions{Versioned: true, VersionID: versionID, MTime: UTCNow().Add(-time.Hour), UserDefined: meta})
+	mustIAM(t, err)
+}
+
+func batchPurgeVersionIDs(t *testing.T, f *batchAuthzFixture, key string) []string {
+	t.Helper()
+	var ids []string
+	marker, vmarker := "", ""
+	for {
+		res, err := f.obj.ListObjectVersions(f.ctx, f.bucket, key, marker, vmarker, "", 1000)
+		mustIAM(t, err)
+		for _, o := range res.Objects {
+			if o.Name == key {
+				ids = append(ids, o.VersionID)
+			}
+		}
+		if !res.IsTruncated {
+			return ids
+		}
+		marker, vmarker = res.NextMarker, res.NextVersionIDMarker
+	}
+}
+
+// A version that the submitter may not expire, because of its tags or its
+// version ID, is refused although the walk did not see it so: the job fails
+// and the key keeps all its versions.
+func TestBatchExpirePurgeAuthorizesCurrentVersions(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		action string // the action a conditional Deny denies
+		deny   string // its condition
+		// change returns the version IDs that change made.
+		change func(t *testing.T, f *batchAuthzFixture, key, older string) []string
+	}{
+		{
+			name:   "tagged-after-walk",
+			action: "s3:GetObjectVersion",
+			deny:   `{"StringEquals":{"s3:ExistingObjectTag/keep":["yes"]}}`,
+			change: func(t *testing.T, f *batchAuthzFixture, key, older string) []string {
+				_, err := f.obj.PutObjectTags(f.ctx, f.bucket, key, "keep=yes", ObjectOptions{VersionID: older})
+				mustIAM(t, err)
+				return nil
+			},
+		},
+		{
+			name:   "version-after-walk",
+			action: "s3:DeleteObjectVersion",
+			deny:   `{"StringEquals":{"s3:versionid":["%s"]}}`,
+			change: func(t *testing.T, f *batchAuthzFixture, key, older string) []string {
+				f.putOlder(t, key, batchPurgeDeniedVersion, nil)
+				return []string{batchPurgeDeniedVersion}
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f := newBatchAuthzFixture(t)
+			deny := tc.deny
+			if strings.Contains(deny, "%s") {
+				deny = fmt.Sprintf(deny, batchPurgeDeniedVersion)
+			}
+			batchAuthzPolicy(f.ctx, t, f.sys, "batch-purge", batchAuthzFullAccess+`,
+{"Effect":"Deny","Action":["`+tc.action+`"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":`+deny+`}`)
+			cred := iamAdminUser(f.ctx, t, f.sys, "batch-purge-user", "batch-purge")
+			const key = "purge/key"
+			older := f.putVersions(t, key)
+			want := batchPurgeVersionIDs(t, f, key)
+			ri := f.runChanged(t, cred, "purge/", func() {
+				want = append(want, tc.change(t, f, key, older)...)
+			})
+			for _, id := range want {
+				if !f.versionExists(t, key, id) {
+					t.Errorf("version %s was purged although the submitter may not delete it", id)
+				}
+			}
+			if !ri.Failed || ri.Complete || ri.FailureReason == "" {
+				t.Errorf("job complete=%v failed=%v reason=%q, want a refused job", ri.Complete, ri.Failed, ri.FailureReason)
+			}
+			if ri.Objects != 0 || ri.KeysRefused != 0 {
+				t.Errorf("job counts objects=%d refused=%d", ri.Objects, ri.KeysRefused)
+			}
+		})
+	}
+}
+
+const batchPurgeDeniedVersion = "6c4b2f8e-1d3a-4e5b-9c7d-0a1b2c3d4e5f"
+
+// A version that Object Lock protects refuses the whole key, also when it
+// was added after the walk. The job completes with the key counted as
+// refused, not failed.
+func TestBatchExpirePurgeRefusesLockedAfterWalk(t *testing.T) {
+	for _, kind := range batchLockKinds {
+		t.Run(kind, func(t *testing.T) {
+			f := newBatchAuthzFixture(t)
+			enableBucketObjectLock(t, f.bucket)
+			batchAuthzPolicy(f.ctx, t, f.sys, "batch-full", batchAuthzFullAccess)
+			cred := iamAdminUser(f.ctx, t, f.sys, "batch-user", "batch-full")
+			const key = "purge/locked"
+			f.putVersions(t, key)
+			want := batchPurgeVersionIDs(t, f, key)
+			locked := mustGetUUID()
+			ri := f.runChanged(t, cred, "purge/", func() {
+				f.putOlder(t, key, locked, batchLockMeta(kind))
+				want = append(want, locked)
+			})
+			for _, id := range want {
+				if !f.versionExists(t, key, id) {
+					t.Errorf("version %s of a key with a locked version was purged", id)
+				}
+			}
+			if !ri.Complete || ri.Failed || ri.KeysRefused != 1 || ri.ObjectsFailed != 0 || ri.Objects != 0 {
+				t.Errorf("job complete=%v failed=%v refused=%d failed objects=%d objects=%d, want complete with 1 refused key",
+					ri.Complete, ri.Failed, ri.KeysRefused, ri.ObjectsFailed, ri.Objects)
+			}
+		})
+	}
+}
+
+// Controls: a purge by root and by an authorized user removes every version,
+// including one added after the walk.
+func TestBatchExpirePurgeAuthorizedControls(t *testing.T) {
+	for _, root := range []bool{true, false} {
+		t.Run(fmt.Sprintf("root=%v", root), func(t *testing.T) {
+			f := newBatchAuthzFixture(t)
+			cred := globalActiveCred
+			if !root {
+				batchAuthzPolicy(f.ctx, t, f.sys, "batch-full", batchAuthzFullAccess)
+				cred = iamAdminUser(f.ctx, t, f.sys, "batch-user", "batch-full")
+			}
+			const key = "purge/key"
+			f.putVersions(t, key)
+			added := mustGetUUID()
+			ri := f.runChanged(t, cred, "purge/", func() {
+				f.putOlder(t, key, added, map[string]string{xhttp.AmzObjectTagging: "keep=yes"})
+			})
+			if ids := batchPurgeVersionIDs(t, f, key); len(ids) != 0 {
+				t.Errorf("an authorized purge kept %v", ids)
+			}
+			if !ri.Complete || ri.Failed || ri.Objects != 3 || ri.KeysRefused != 0 {
+				t.Errorf("job complete=%v failed=%v objects=%d refused=%d, want complete with 3 objects",
+					ri.Complete, ri.Failed, ri.Objects, ri.KeysRefused)
+			}
+		})
+	}
+}
+
+// Cases found in the sixth review round.
+
+// batchPurgeBudgetChange runs change once the first purge stopped at its
+// budget after removing versions.
+type batchPurgeBudgetChange struct {
+	ObjectLayer
+	change func()
+	calls  int
+}
+
+func (h *batchPurgeBudgetChange) DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
+	oi, err := h.ObjectLayer.DeleteObject(ctx, bucket, object, opts)
+	if opts.DeleteAllVersions {
+		h.calls++
+		if h.calls == 1 && errors.Is(err, errDeleteAllBudget) && opts.DeleteAllResult.Removed > 0 {
+			h.change()
+		}
+	}
+	return oi, err
+}
+
+// A purge that continues after its budget authorizes the versions again, as
+// they are then; the refusal and the versions already removed persist
+// across another start.
+func TestBatchExpirePurgeBudgetReauthorizes(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	const key = "purge/budget"
+	oldest := batchPutVersion(t, f.obj, f.bucket, key, nil)
+	middle := batchPutVersion(t, f.obj, f.bucket, key, nil)
+	latest := batchPutVersion(t, f.obj, f.bucket, key, nil)
+	batchAuthzPolicy(f.ctx, t, f.sys, "review6-budget-deny", batchAuthzFullAccess+`,
+{"Effect":"Deny","Action":["s3:GetObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/keep":"yes"}}}`)
+	cred := iamAdminUser(f.ctx, t, f.sys, "review6-budget-user", "review6-budget-deny")
+	rec, job := f.submit(t, cred, f.purgeYAML("purge/"))
+	if job == nil {
+		t.Fatalf("submit: %d %s", rec.Code, rec.Body.String())
+	}
+	oldBudget := deleteAllBudget
+	deleteAllBudget.versions = 1
+	t.Cleanup(func() { deleteAllBudget = oldBudget })
+	hook := &batchPurgeBudgetChange{ObjectLayer: f.obj, change: func() {
+		_, err := f.obj.PutObjectTags(f.ctx, f.bucket, key, "keep=yes", ObjectOptions{VersionID: middle.VersionID})
+		mustIAM(t, err)
+	}}
+	mustIAM(t, job.Expire.Start(f.ctx, hook, *job))
+	ri := f.jobInfo(t, job)
+	if hook.calls != 2 || !ri.Failed || ri.Complete || ri.FailureReason == "" || ri.Objects != 1 || ri.KeysRefused != 0 {
+		t.Fatalf("continuation: calls=%d complete=%v failed=%v reason=%q objects=%d refused=%d", hook.calls, ri.Complete, ri.Failed, ri.FailureReason, ri.Objects, ri.KeysRefused)
+	}
+	if f.versionExists(t, key, oldest.VersionID) || !f.versionExists(t, key, middle.VersionID) || !f.versionExists(t, key, latest.VersionID) {
+		t.Fatal("only the first authorized version may be removed")
+	}
+	_, err := f.obj.DeleteObjectTags(f.ctx, f.bucket, key, ObjectOptions{VersionID: middle.VersionID})
+	mustIAM(t, err)
+	mustIAM(t, job.Expire.Start(f.ctx, hook, *job))
+	after := f.jobInfo(t, job)
+	if hook.calls != 2 || after.FailureReason != ri.FailureReason || after.Objects != 1 || after.Complete {
+		t.Fatal("persisted refusal must prevent a fresh purge after the tags change")
+	}
+}
+
+// The purge's lock excludes tag writers from its authorization through its
+// last delete, for a directory key in a lock domain of its own on one pool
+// and for a key in a second pool.
+func TestBatchExpirePurgeExcludesTagWriter(t *testing.T) {
+	for _, mode := range []string{ldDsync, "two-pools"} {
+		t.Run(mode, func(t *testing.T) {
+			var z *erasureServerPools
+			var bucket string
+			pool, key := 0, "purge/tags/"
+			if mode == "two-pools" {
+				z, bucket = consistencyPools(t)
+				pool = 1
+			} else {
+				z, bucket = ldPool(t, mode)
+				for i := 0; ; i++ {
+					key = fmt.Sprintf("purge/tags-%d/", i)
+					if z.serverPools[0].getHashedSetIndex(encodeDirObject(key)) == 1 {
+						break
+					}
+				}
+			}
+			before := putConsistencyObject(t, z, bucket, encodeDirObject(key), pool, "", ObjectOptions{Versioned: true})
+			ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+			defer cancel()
+			entered, release := make(chan struct{}), make(chan struct{})
+			var released atomic.Bool
+			unblock := func() {
+				if released.CompareAndSwap(false, true) {
+					close(release)
+				}
+			}
+			defer unblock()
+			purged := make(chan error, 1)
+			go func() {
+				_, err := z.DeleteObject(ctx, bucket, key, ObjectOptions{
+					DeleteAllVersions: true,
+					DeleteAllLatest:   &deleteAllMatch{VersionID: before.VersionID, ModTime: before.ModTime},
+					deleteAllAuthorize: func(versions []ObjectInfo) error {
+						close(entered)
+						select {
+						case <-release:
+							return nil
+						case <-ctx.Done():
+							return ctx.Err()
+						}
+					},
+				})
+				purged <- err
+			}()
+			select {
+			case <-entered:
+			case <-ctx.Done():
+				t.Fatal("purge never reached authorization")
+			}
+			written := make(chan error, 1)
+			go func() {
+				_, err := z.PutObjectTags(ctx, bucket, key, "keep=yes", ObjectOptions{VersionID: before.VersionID})
+				written <- err
+			}()
+			select {
+			case err := <-written:
+				t.Errorf("tag writer crossed the purge's authorization lock: %v", err)
+			case <-time.After(ldWait):
+				unblock()
+				err := <-written
+				if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+					t.Errorf("writer must observe the completed purge: %v", err)
+				}
+			}
+			unblock()
+			mustIAM(t, <-purged)
+		})
+	}
+}
+
+// batchTagWriteFault fails metadata updates of one key, so that a tag write
+// reaches only the drives it does not wrap.
+type batchTagWriteFault struct {
+	StorageAPI
+	bucket, object string
+}
+
+func (d batchTagWriteFault) UpdateMetadata(ctx context.Context, bucket, object string, fi FileInfo, opts UpdateMetadataOpts) error {
+	if bucket == d.bucket && object == d.object {
+		return errDiskNotFound
+	}
+	return d.StorageAPI.UpdateMetadata(ctx, bucket, object, fi, opts)
+}
+
+// A failed tag write can leave a revision that the submitter may not expire
+// on a minority of drives. The purge removes it with the version, so it is
+// authorized too.
+func TestBatchExpirePurgeAuthorizesDriveRevisions(t *testing.T) {
+	resetTestGlobals()
+	t.Cleanup(resetTestGlobals)
+	z := purgeSinglePool(t, 16)
+	globalIAMSys.initStore(z, nil)
+	mustIAM(t, globalIAMSys.Load(t.Context(), true))
+	f := &batchAuthzFixture{ctx: t.Context(), sys: globalIAMSys, obj: z, bucket: "review6-batch-authz"}
+	mustIAM(t, z.MakeBucket(f.ctx, f.bucket, MakeBucketOptions{VersioningEnabled: true}))
+	batchTestJobPool(f.ctx, t, z)
+	const key = "purge/minority-tags"
+	older := f.putVersions(t, key)
+	batchAuthzPolicy(f.ctx, t, f.sys, "review6-minority-deny", strings.ReplaceAll(batchAuthzFullAccess+`,
+{"Effect":"Deny","Action":["s3:GetObjectVersion"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/keep":"yes"}}}`, "batch-authz", f.bucket))
+	cred := iamAdminUser(f.ctx, t, f.sys, "review6-minority-user", "review6-minority-deny")
+	sub, err := newBatchJobSubmitter(cred)
+	mustIAM(t, err)
+	if err := sub.authorize(f.ctx, batchAuthzRequest{action: policy.GetObjectVersionAction, bucket: f.bucket, object: key, versionID: older, tags: "keep=yes"}); !isBatchJobAuthzError(err) {
+		t.Fatal("control: this tag revision must deny the read needed for expiration")
+	}
+	set := z.serverPools[0].getHashedSet(key)
+	original := set.getDisks
+	disks := append([]StorageAPI(nil), original()...)
+	wrapped := append([]StorageAPI(nil), disks...)
+	for i := 1; i < len(wrapped); i++ {
+		wrapped[i] = batchTagWriteFault{StorageAPI: wrapped[i], bucket: f.bucket, object: key}
+	}
+	set.getDisks = func() []StorageAPI { return wrapped }
+	_, err = f.obj.PutObjectTags(f.ctx, f.bucket, key, "keep=yes", ObjectOptions{VersionID: older, UserDefined: map[string]string{
+		ReservedMetadataPrefixLower + TaggingTimestamp: UTCNow().Format(time.RFC3339Nano),
+	}})
+	set.getDisks = original
+	if err == nil {
+		t.Fatal("control: the tag write must fail its quorum")
+	}
+	fi, err := disks[0].ReadVersion(f.ctx, "", f.bucket, key, older, ReadOptions{})
+	mustIAM(t, err)
+	if fi.Metadata[xhttp.AmzObjectTagging] != "keep=yes" {
+		t.Fatal("control: the successful drive must hold the denied tag revision")
+	}
+	history, err := z.deleteAllHistory(f.ctx, f.bucket, key)
+	mustIAM(t, err)
+	for _, v := range history {
+		for _, c := range v.copies {
+			if c.fi.Metadata[xhttp.AmzObjectTagging] != "" {
+				t.Fatal("control: the quorum history must still have the original tags")
+			}
+		}
+	}
+	ri := f.runChanged(t, cred, "purge/", func() {})
+	if !ri.Failed || ri.Complete || !strings.Contains(ri.FailureReason, "is not allowed") {
+		t.Errorf("denied physical revision must refuse the job: complete=%v failed=%v reason=%q", ri.Complete, ri.Failed, ri.FailureReason)
+	}
+	if _, err := disks[0].ReadVersion(f.ctx, "", f.bucket, key, older, ReadOptions{}); err != nil {
+		t.Errorf("denied physical revision was removed: %v", err)
+	}
+	if got := batchPurgeVersionIDs(t, f, key); len(got) != 2 {
+		t.Errorf("denial must keep the whole key; remaining versions=%v", got)
+	}
+}
+
+// Cases found in the seventh review round.
+
+// The time the authorization takes counts against the purge's budget: a
+// late allow and a deny both delete nothing.
+func TestBatchExpirePurgeBudgetIncludesAuthorization(t *testing.T) {
+	z := purgeSinglePool(t, 16)
+	p := newPurgeTest(t, z, "review7-purge-auth-time", MakeBucketOptions{VersioningEnabled: true}, "")
+	const key = "purge/key"
+	p.version(0, key, "older", 2)
+	p.version(0, key, "latest", 1)
+	exp := p.matched(key)
+	oldBudget := deleteAllBudget
+	deleteAllBudget.duration = 10 * time.Millisecond
+	t.Cleanup(func() { deleteAllBudget = oldBudget })
+	denied := errors.New("review7: revision denied")
+	for _, decision := range []error{nil, denied} {
+		called := false
+		var res deleteAllResult
+		_, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{
+			DeleteAllVersions: true,
+			DeleteAllLatest:   &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime},
+			DeleteAllResult:   &res,
+			deleteAllAuthorize: func(versions []ObjectInfo) error {
+				called = true
+				if len(versions) < 2 {
+					t.Error("authorization did not receive both approved versions")
+				}
+				time.Sleep(20 * time.Millisecond)
+				return decision
+			},
+		})
+		want := decision
+		if want == nil {
+			want = errDeleteAllBudget
+		}
+		if !called || !errors.Is(err, want) || res.Started || res.Complete || res.Removed != 0 {
+			t.Errorf("late authorization: called=%v err=%v result=%+v", called, err, res)
+		}
+		if got := p.held(0, key); len(got) != 2 {
+			t.Fatalf("purge removed versions after late authorization: %v", got)
+		}
+	}
+}

cmd/batch-expire.go

@@ -347,22 +347,10 @@ const (
 	batchExpireJobDefaultRetryDelay = 250 * time.Millisecond
 )
 
-// expireVersionRef is a version of a key whose versions are all purged, as
-// the walk saw it.
-type expireVersionRef struct {
-	VersionID    string
-	UserTags     string
-	DeleteMarker bool
-	Locked       bool
-}
-
 type expireObjInfo struct {
 	ObjectInfo
 	ExpireAll         bool
 	DeleteMarkerCount int64
-	// versions are the versions of an ExpireAll key the walk saw, the
-	// latest first.
-	versions []expireVersionRef
 }
 
 // isErrObjectLocked reports a version that Object Lock protects.
@@ -485,35 +473,20 @@ func (r *BatchJobExpire) expireVersions(ctx context.Context, api ObjectLayer, vc
 
 // 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.
+// while the version the walk matched is still its latest. Under its lock,
+// before it deletes anything, the purge refuses the key when Object Lock
+// protects a version and has every write it would delete, each copy and
+// each revision a drive holds, authorized for the submitter, with the
+// version ID and tags it has then.
 //
-// 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.
+// What the purge removed is counted at once; a refused key 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: an authorization failure or the end of ctx.
 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++ {
@@ -523,11 +496,15 @@ func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer,
 		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,
-		})
+		p, err := job.Submitter.principal(ctx)
+		if err == nil {
+			_, err = api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{
+				DeleteAllVersions:  true,
+				DeleteAllLatest:    &latest,
+				DeleteAllResult:    &purge,
+				deleteAllAuthorize: r.expireAllAuthorize(p, exp.Name),
+			})
+		}
 		// 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))
@@ -538,6 +515,8 @@ func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer,
 		switch {
 		case err == nil:
 			return nil
+		case isBatchJobAuthzError(err):
+			return err
 		case errors.Is(err, errDeleteAllChanged):
 			// Changed since the walk matched it: not expired.
 			return nil
@@ -562,6 +541,22 @@ func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer,
 	return nil
 }
 
+// expireAllAuthorize returns the authorization of a purge of all versions of
+// name for p: each version the purge would delete, as it holds it, needs the
+// actions of expiring it (expireAuthzRequests).
+func (r *BatchJobExpire) expireAllAuthorize(p batchJobPrincipal, name string) func([]ObjectInfo) error {
+	return func(versions []ObjectInfo) error {
+		for _, v := range versions {
+			for _, req := range r.expireAuthzRequests(name, v.VersionID, v.UserTags) {
+				if err := p.authorize(req); err != nil {
+					return err
+				}
+			}
+		}
+		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.
@@ -833,17 +828,12 @@ 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 {
-					// 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
+					continue // including latest version in toDel suffices, skipping other versions
 				}
 				versionsCount++
 			} else {
@@ -917,15 +907,6 @@ 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/erasure-server-pool-delete-all-options_test.go

@@ -25,6 +25,8 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+
+	xhttp "github.com/minio/minio/internal/http"
 )
 
 // The parts of the purge of all versions that need its own options.
@@ -65,16 +67,22 @@ func TestPurgeAllVersionsOptions(t *testing.T) {
 	p.version(0, "ea/key", "v1", 5)
 	match := &deleteAllMatch{}
 	expire := ExpirationOptions{Expire: true}
+	allow := func([]ObjectInfo) error { return nil }
 	for name, opts := range map[string]ObjectOptions{
 		"no approval":   {DeleteAllVersions: true},
-		"two approvals": {DeleteAllVersions: true, Expiration: expire, DeleteAllLatest: match},
-		"prefix":        {DeleteAllVersions: true, Expiration: expire, DeletePrefix: true},
-		"prefix object": {DeleteAllVersions: true, Expiration: expire, DeletePrefixObject: true},
-		"version":       {DeleteAllVersions: true, Expiration: expire, VersionID: mustGetUUID()},
-		"movement":      {DeleteAllVersions: true, Expiration: expire, DataMovement: true},
-		"marker":        {DeleteAllVersions: true, Expiration: expire, DeleteMarker: true},
-		"versioned":     {DeleteAllVersions: true, Expiration: expire, Versioned: true},
-		"suspended":     {DeleteAllVersions: true, Expiration: expire, VersionSuspended: true},
+		"two approvals": {DeleteAllVersions: true, Expiration: expire, DeleteAllLatest: match, deleteAllAuthorize: allow},
+		// A batch purge deletes as the submitter of its job, an ILM purge
+		// as the server, and no other delete takes an authorization.
+		"batch unauthorized": {DeleteAllVersions: true, DeleteAllLatest: match},
+		"ilm authorized":     {DeleteAllVersions: true, Expiration: expire, deleteAllAuthorize: allow},
+		"authorized delete":  {deleteAllAuthorize: allow},
+		"prefix":             {DeleteAllVersions: true, Expiration: expire, DeletePrefix: true},
+		"prefix object":      {DeleteAllVersions: true, Expiration: expire, DeletePrefixObject: true},
+		"version":            {DeleteAllVersions: true, Expiration: expire, VersionID: mustGetUUID()},
+		"movement":           {DeleteAllVersions: true, Expiration: expire, DataMovement: true},
+		"marker":             {DeleteAllVersions: true, Expiration: expire, DeleteMarker: true},
+		"versioned":          {DeleteAllVersions: true, Expiration: expire, Versioned: true},
+		"suspended":          {DeleteAllVersions: true, Expiration: expire, VersionSuspended: true},
 	} {
 		if _, err := z.DeleteObject(t.Context(), p.bucket, "ea/key", opts); !errors.As(err, new(InvalidArgument)) {
 			t.Errorf("%s: %v", name, err)
@@ -134,9 +142,10 @@ func TestPurgeAllVersionsBudgetCoversReads(t *testing.T) {
 	t.Cleanup(func() { pool.erasureDisksMu.Lock(); pool.erasureDisks = prior; pool.erasureDisksMu.Unlock() })
 	var res deleteAllResult
 	_, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{
-		DeleteAllVersions: true,
-		DeleteAllLatest:   &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime},
-		DeleteAllResult:   &res,
+		DeleteAllVersions:  true,
+		DeleteAllLatest:    &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime},
+		DeleteAllResult:    &res,
+		deleteAllAuthorize: func([]ObjectInfo) error { return nil },
 	})
 	if !errors.Is(err, errDeleteAllBudget) || res.Started || res.Removed != 0 {
 		t.Errorf("purge after its budget ran out on reads: err=%v %+v", err, res)
@@ -213,9 +222,10 @@ func TestPurgeAllVersionsDeletesOnCheckedDrives(t *testing.T) {
 	})
 	var res deleteAllResult
 	if _, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{
-		DeleteAllVersions: true,
-		DeleteAllLatest:   &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime},
-		DeleteAllResult:   &res,
+		DeleteAllVersions:  true,
+		DeleteAllLatest:    &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime},
+		DeleteAllResult:    &res,
+		deleteAllAuthorize: func([]ObjectInfo) error { return nil },
 	}); err != nil || !res.Complete {
 		t.Fatalf("purge: %v %+v", err, res)
 	}
@@ -223,3 +233,62 @@ func TestPurgeAllVersionsDeletesOnCheckedDrives(t *testing.T) {
 		t.Errorf("reads=%d deletes on the replacement handle=%d", reads.Load(), deletes.Load())
 	}
 }
+
+// A batch purge has every copy of every version it approved authorized, as
+// the copy is now, before it deletes any; an authorization error deletes
+// nothing and is returned as it is.
+func TestPurgeAllVersionsAuthorizesEveryCopy(t *testing.T) {
+	z, _ := consistencyPools(t)
+	p := newPurgeTest(t, z, "purge-authorize", MakeBucketOptions{VersioningEnabled: true}, "")
+	const key = "bx/authorize"
+	p.put(0, key, "old", daysAgo(6), ObjectOptions{})
+	p.put(1, key, "new", daysAgo(5), ObjectOptions{UserDefined: map[string]string{xhttp.AmzObjectTagging: "copy=new"}})
+	tagged := p.put(0, key, "v1", daysAgo(4), ObjectOptions{Versioned: true, UserDefined: map[string]string{xhttp.AmzObjectTagging: "keep=yes"}})
+	exp := p.matched(key)
+	denied := errors.New("denied")
+	var seen []string // the distinct versions and tags authorized
+	purge := func(authorize func([]ObjectInfo) error) (deleteAllResult, error) {
+		var res deleteAllResult
+		_, err := z.DeleteObject(t.Context(), p.bucket, key, ObjectOptions{
+			DeleteAllVersions:  true,
+			DeleteAllLatest:    &deleteAllMatch{VersionID: exp.VersionID, ModTime: exp.ModTime},
+			DeleteAllResult:    &res,
+			deleteAllAuthorize: authorize,
+		})
+		return res, err
+	}
+	res, err := purge(func(versions []ObjectInfo) error {
+		for _, v := range versions {
+			if got := v.VersionID + " " + v.UserTags; !slices.Contains(seen, got) {
+				seen = append(seen, got)
+			}
+		}
+		return denied
+	})
+	if !errors.Is(err, denied) || res.Started || res.Removed != 0 {
+		t.Errorf("denied purge: err=%v %+v", err, res)
+	}
+	want := []string{"null ", "null copy=new", tagged.VersionID + " keep=yes"}
+	slices.Sort(seen)
+	slices.Sort(want)
+	if !slices.Equal(seen, want) {
+		t.Errorf("authorized %q, want every copy %q", seen, want)
+	}
+	for pool := range z.serverPools {
+		if got := p.held(pool, key); len(got) == 0 {
+			t.Errorf("pool %d lost its copies to a denied purge", pool)
+		}
+	}
+	if got := p.get(key); got != "v1" {
+		t.Errorf("reads %s after a denied purge, want v1", got)
+	}
+	// Control: an allowed purge deletes them all.
+	if res, err = purge(func([]ObjectInfo) error { return nil }); err != nil || !res.Complete {
+		t.Fatalf("allowed purge: %v %+v", err, res)
+	}
+	for pool := range z.serverPools {
+		if got := p.held(pool, key); got != nil {
+			t.Errorf("pool %d keeps %v", pool, got)
+		}
+	}
+}

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

@@ -26,6 +26,7 @@ import (
 	"sync"
 	"time"
 
+	xhttp "github.com/minio/minio/internal/http"
 	"github.com/pgsty/silo-pkg/v3/sync/errgroup"
 )
 
@@ -49,6 +50,9 @@ import (
 //     when any revision of an approved version there is transitioned or,
 //     when the bucket has Object Lock, protected by retention or a legal
 //     hold. It deletes on exactly the drives it read;
+//   - a batch expiry then authorizes every copy of every approved version,
+//     and every revision of one that a drive holds, for the submitter of the
+//     job, and the purge deletes nothing unless all of them are allowed;
 //   - it deletes the oldest version first and the latest version last and
 //     alone, once every older version is gone from every pool. A purge that
 //     stops early keeps the latest version, so it never exposes an older
@@ -118,11 +122,14 @@ func deleteAllVersionID(versionID string) string {
 }
 
 // deleteAllInvalid returns why opts cannot ask for a purge of all versions.
-// The purge is approved either as an ILM expiry or by a batch match, and it
-// only deletes: it never creates a marker, addresses a single version,
-// moves data or deletes a prefix.
+// The purge is approved either as an ILM expiry or by a batch match, which
+// also authorizes the versions it deletes, and it only deletes: it never
+// creates a marker, addresses a single version, moves data or deletes a
+// prefix.
 func (opts ObjectOptions) deleteAllInvalid() error {
 	switch {
+	case !opts.DeleteAllVersions:
+		return errors.New("only a purge of all versions authorizes the versions it deletes")
 	case opts.DeletePrefix || opts.DeletePrefixObject:
 		return errors.New("a purge of all versions is not a prefix delete")
 	case opts.VersionID != "":
@@ -137,6 +144,8 @@ func (opts ObjectOptions) deleteAllInvalid() error {
 		return errors.New("a purge of all versions takes no options of other deletes")
 	case opts.Expiration.Expire == (opts.DeleteAllLatest != nil):
 		return errors.New("a purge of all versions needs exactly one approval")
+	case (opts.DeleteAllLatest != nil) != (opts.deleteAllAuthorize != nil):
+		return errors.New("a batch purge of all versions, and only it, authorizes the versions it deletes")
 	}
 	return nil
 }
@@ -297,12 +306,29 @@ func (z *erasureServerPools) deleteAllVersions(ctx context.Context, bucket, obje
 			}
 		}
 	}
+	// A batch expiry deletes as its submitter: every write it is about to
+	// delete, the accepted copies and every revision a drive holds, must be
+	// allowed with the version ID and tags it has now.
+	var writes *deleteAllWrites
+	if opts.deleteAllAuthorize != nil {
+		writes = &deleteAllWrites{bucket: bucket, name: name, seen: make(map[[2]string]bool, len(history))}
+		for _, v := range history {
+			for _, c := range v.copies {
+				writes.add(c.fi)
+			}
+		}
+	}
 	drives := make([][]StorageAPI, len(z.serverPools))
 	for pool := range z.serverPools {
 		if !held[pool] {
 			continue
 		}
-		if drives[pool], err = z.serverPools[pool].getHashedSet(object).deleteAllDrives(ctx, bucket, object, ids, lr.LockEnabled); err != nil {
+		if drives[pool], err = z.serverPools[pool].getHashedSet(object).deleteAllDrives(ctx, bucket, object, ids, lr.LockEnabled, writes); err != nil {
+			return fail(err)
+		}
+	}
+	if opts.deleteAllAuthorize != nil {
+		if err := opts.deleteAllAuthorize(writes.versions); err != nil {
 			return fail(err)
 		}
 	}
@@ -329,6 +355,28 @@ func (z *erasureServerPools) deleteAllVersions(ctx context.Context, bucket, obje
 	return res.Latest, nil
 }
 
+// deleteAllWrites collects, for the authorization of a batch purge, the
+// writes it deletes: one of each distinct version ID and tags.
+type deleteAllWrites struct {
+	bucket, name string
+	seen         map[[2]string]bool
+	versions     []ObjectInfo
+}
+
+// add adds the write fi unless w is nil or holds one with its version ID and
+// tags.
+func (w *deleteAllWrites) add(fi FileInfo) {
+	if w == nil {
+		return
+	}
+	key := [2]string{deleteAllVersionID(fi.VersionID), fi.Metadata[xhttp.AmzObjectTagging]}
+	if w.seen[key] {
+		return
+	}
+	w.seen[key] = true
+	w.versions = append(w.versions, fi.ToObjectInfo(w.bucket, w.name, true))
+}
+
 // deleteAllRefuses returns why the purge must keep the key that holds the
 // revision fi of an approved version: it is transitioned, whose remote
 // object a later release hands to the tier cleanup, or, with Object Lock,
@@ -344,7 +392,8 @@ func deleteAllRefuses(ctx context.Context, bucket, object string, fi FileInfo, l
 }
 
 // deleteAllDrives reads object on every drive of the set and checks each
-// revision of the versions ids that a drive holds with deleteAllRefuses.
+// revision of the versions ids that a drive holds with deleteAllRefuses, and
+// adds it to writes.
 // A minority of drives can hold a revision other than the accepted one, and
 // heal would later remove what the purge leaves on a drive, so a drive that
 // cannot be read or decoded refuses the key. A drive without the object
@@ -353,7 +402,7 @@ func deleteAllRefuses(ctx context.Context, bucket, object string, fi FileInfo, l
 // handle names a drive, not a backend: a backend that replaces the drive
 // behind it starts empty and receives this object only from a write or a
 // heal of it, which take the object lock the purge holds.
-func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object string, ids map[string]bool, lockEnabled bool) ([]StorageAPI, error) {
+func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object string, ids map[string]bool, lockEnabled bool, writes *deleteAllWrites) ([]StorageAPI, error) {
 	disks := slices.Clone(er.getDisks())
 	raws, errs := readAllRawFileInfo(ctx, disks, bucket, object, false)
 	if err := ctx.Err(); err != nil {
@@ -381,6 +430,7 @@ func (er erasureObjects) deleteAllDrives(ctx context.Context, bucket, object str
 			if err := deleteAllRefuses(ctx, bucket, object, fi, lockEnabled); err != nil {
 				return nil, err
 			}
+			writes.add(fi)
 		}
 	}
 	return disks, nil

cmd/erasure-server-pool.go

@@ -1234,7 +1234,7 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
 	if err = checkDelObjArgs(ctx, bucket, object); err != nil {
 		return objInfo, err
 	}
-	if opts.DeleteAllVersions {
+	if opts.DeleteAllVersions || opts.deleteAllAuthorize != nil {
 		if err := opts.deleteAllInvalid(); err != nil {
 			return objInfo, InvalidArgument{Bucket: bucket, Object: object, Err: err}
 		}

cmd/object-api-interface.go

@@ -113,6 +113,12 @@ type ObjectOptions struct {
 	// DeleteAllResult, when set, receives what a purge of all versions
 	// confirmed, also when it fails part way.
 	DeleteAllResult *deleteAllResult
+	// deleteAllAuthorize authorizes a batch purge of all versions, which
+	// needs it, on every copy of every version the purge approved and every
+	// revision of one that a drive holds, one of each distinct version ID
+	// and tags, before it deletes any; its error refuses the key and is
+	// returned. Set by batch expiration only, never from a request.
+	deleteAllAuthorize func(versions []ObjectInfo) error
 
 	// Pools-layer version resolver; the caller holds the shared object lock.
 	replicaObjectInfo GetObjectInfoFn