diff --git a/cmd/batch-authz-review6_test.go b/cmd/batch-authz-review6_test.go
new file mode 100644
index 000000000..447183e92
--- /dev/null
+++ b/cmd/batch-authz-review6_test.go
@@ -0,0 +1,493 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"maps"
+	"net/http"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/crypto"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/minio/internal/kms"
+)
+
+// Rotation cases found in review.
+
+// Rotation rewrites the copy GetObjectInfo returns, which can carry another
+// tag revision than the merged version of all pools: both are authorized.
+func TestBatchReviewRotationAuthorizesWrittenCopy(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	f := newBatchAuthzFixture(t)
+	previousKMS := GlobalKMS
+	GlobalKMS = kms.NewStub("review6-rotation-copy-key")
+	t.Cleanup(func() { GlobalKMS = previousKMS })
+	batchAuthzPolicy(f.ctx, t, f.sys, "review6-rotation-copy-deny", fmt.Sprintf(`
+{"Effect":"Allow","Action":["s3:GetObject","s3:GetObjectVersion","s3:PutObject"],"Resource":["arn:aws:s3:::%s/*"]},
+{"Effect":"Deny","Action":["s3:GetObject","s3:GetObjectVersion"],"Resource":["arn:aws:s3:::%s/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/classification":"private"}}}`, bucket, bucket))
+	cred := iamAdminUser(f.ctx, t, f.sys, "review6-rotation-copy-user", "review6-rotation-copy-deny")
+	sub, err := newBatchJobSubmitter(cred)
+	mustIAM(t, err)
+	const name = "rotation/written-copy"
+	first := make(map[string]string)
+	key, err := newEncryptMetadata(f.ctx, crypto.S3, "", nil, bucket, name, first, nil)
+	mustIAM(t, err)
+	first[xhttp.AmzObjectTagging] = "classification=private"
+	first[ReservedMetadataPrefixLower+TaggingTimestamp] = UTCNow().Add(-time.Hour).Format(time.RFC3339Nano)
+	second := maps.Clone(first)
+	second[xhttp.AmzObjectTagging] = "classification=public"
+	second[ReservedMetadataPrefixLower+TaggingTimestamp] = UTCNow().Format(time.RFC3339Nano)
+	ciphertext, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader("review6 rotation payload"), key))
+	mustIAM(t, err)
+	before := putConsistencyObject(t, z, bucket, name, 0, string(ciphertext), ObjectOptions{Versioned: true, UserDefined: first})
+	putConsistencyObject(t, z, bucket, name, 1, string(ciphertext), ObjectOptions{Versioned: true, VersionID: before.VersionID, MTime: before.ModTime, UserDefined: second})
+	logical, err := z.replicaObjectInfo(f.ctx, bucket, name, ObjectOptions{VersionID: before.VersionID})
+	mustIAM(t, err)
+	if logical.UserTags != "classification=public" {
+		t.Fatalf("control: merged tags=%q, want public", logical.UserTags)
+	}
+	job := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: sses3}}
+	if err = job.KeyRotate(f.ctx, z, before, sub); !isBatchJobAuthzError(err) {
+		t.Fatalf("rotation of a copy the submitter may not read returned %v", err)
+	}
+	after, err := z.serverPools[0].GetObjectInfo(f.ctx, bucket, name, ObjectOptions{VersionID: before.VersionID})
+	mustIAM(t, err)
+	if after.UserDefined[crypto.MetaSealedKeyS3] != before.UserDefined[crypto.MetaSealedKeyS3] {
+		t.Fatal("a refused rotation rewrote the copy")
+	}
+}
+
+// Two pools hold different writes of the null version, and a newer version
+// of the key is in the second one. Rotation rewrites the copy it read and
+// authorized, in the first pool, not the copy the latest version of the key
+// would route a write to; the other write keeps its key and stays readable.
+func TestBatchReviewRotationRewritesTheCopyItRead(t *testing.T) {
+	for _, otherLatest := range []bool{false, true} {
+		name := "addressed-version-is-latest"
+		if otherLatest {
+			name = "another-version-is-latest"
+		}
+		t.Run(name, func(t *testing.T) {
+			resetTestGlobals()
+			t.Cleanup(resetTestGlobals)
+			z, _ := consistencyPools(t)
+			globalIAMSys.initStore(z, nil)
+			mustIAM(t, globalIAMSys.Load(t.Context(), true))
+			const bucket = "review7-rotation-copies"
+			mustIAM(t, z.MakeBucket(t.Context(), bucket, MakeBucketOptions{VersioningEnabled: true}))
+			_, versionErr := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, enabledBucketVersioningConfig)
+			mustIAM(t, versionErr)
+			oldKMS := GlobalKMS
+			GlobalKMS = kms.NewStub("review7-rotation-key")
+			t.Cleanup(func() { GlobalKMS = oldKMS })
+			const object = "rotation/null-copy"
+			const payload = "rotation must preserve this data"
+			readCopy := func(pool int) ([]byte, error) {
+				gr, err := z.serverPools[pool].GetObjectNInfo(t.Context(), bucket, object, nil, nil, ObjectOptions{VersionID: nullVersionID})
+				if err != nil {
+					return nil, err
+				}
+				defer gr.Close()
+				return io.ReadAll(gr)
+			}
+			var before [2]ObjectInfo
+			var keys [2]crypto.ObjectKey
+			kinds := []crypto.Type{crypto.S3, crypto.S3KMS}
+			for pool, kind := range kinds {
+				meta := make(map[string]string)
+				key, err := newEncryptMetadata(t.Context(), kind, "review7-rotation-key", nil, bucket, object, meta, nil)
+				mustIAM(t, err)
+				keys[pool] = key
+				meta[xhttp.AmzObjectTagging] = "classification=public"
+				if pool == 1 {
+					meta[xhttp.AmzObjectTagging] = "classification=private"
+				}
+				meta[ReservedMetadataPrefixLower+TaggingTimestamp] = UTCNow().Add(-time.Duration(pool+2) * time.Hour).Format(time.RFC3339Nano)
+				body, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader(payload), key))
+				mustIAM(t, err)
+				putConsistencyObject(t, z, bucket, object, pool, string(body), ObjectOptions{
+					Versioned: true, VersionID: nullVersionID, UserDefined: meta,
+					MTime: UTCNow().Add(-time.Duration(pool+2) * time.Hour),
+				})
+				// As a read returns it, without the write's internal fields.
+				before[pool], err = z.serverPools[pool].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: nullVersionID})
+				mustIAM(t, err)
+				data, err := readCopy(pool)
+				if err != nil || string(data) != payload {
+					t.Fatalf("control: pool %d was not readable before rotation: %v", pool, err)
+				}
+			}
+			if otherLatest {
+				putConsistencyObject(t, z, bucket, object, 1, "later version", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour)})
+			}
+			selected, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: nullVersionID, Versioned: true})
+			mustIAM(t, err)
+			if selected.VersionID != nullVersionID || !selected.ModTime.Equal(before[0].ModTime) || !crypto.S3.IsEncrypted(selected.UserDefined) {
+				t.Fatal("control: the addressed null version must select the SSE-S3 write in pool zero")
+			}
+			logical, err := batchAuthoritativeVersion(t.Context(), z, bucket, object, selected)
+			mustIAM(t, err)
+			if logical.UserTags != "classification=public" || selected.UserTags != logical.UserTags {
+				t.Fatal("control: both current authorization inputs must have public tags")
+			}
+			r := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: ssekms, Key: "review7-rotation-key"}}
+			sub := batchRootSubmitter(t)
+			if otherLatest {
+				batchAuthzPolicy(t.Context(), t, globalIAMSys, "review7-rotate-public", fmt.Sprintf(`
+{"Effect":"Allow","Action":["s3:GetObject","s3:GetObjectVersion","s3:PutObject"],"Resource":["arn:aws:s3:::%s/*"]},
+{"Effect":"Deny","Action":["s3:GetObject","s3:GetObjectVersion"],"Resource":["arn:aws:s3:::%s/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/classification":"private"}}}`, bucket, bucket))
+				cred := iamAdminUser(t.Context(), t, globalIAMSys, "review7-rotate-user", "review7-rotate-public")
+				sub, err = newBatchJobSubmitter(cred)
+				mustIAM(t, err)
+				if err := sub.authorize(t.Context(), r.rotateAuthzRequests(before[1])...); !isBatchJobAuthzError(err) {
+					t.Fatal("control: the older write in pool one must be denied")
+				}
+			}
+			err = r.KeyRotate(t.Context(), z, selected, sub)
+			if err != nil {
+				t.Logf("rotation refused: %v", err)
+			}
+			for pool, want := range keys {
+				after, readErr := z.serverPools[pool].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: nullVersionID})
+				mustIAM(t, readErr)
+				if otherLatest && pool == 1 && (!maps.Equal(after.UserDefined, before[pool].UserDefined) || !after.ModTime.Equal(before[pool].ModTime) || after.UserTags != before[pool].UserTags) {
+					t.Error("rotation rewrote a copy denied by the submitter's tag-conditioned policy")
+				}
+				kind, _ := crypto.IsEncrypted(after.UserDefined)
+				var got crypto.ObjectKey
+				switch kind {
+				case crypto.S3:
+					got, readErr = crypto.S3.UnsealObjectKey(GlobalKMS, after.UserDefined, bucket, object)
+				case crypto.S3KMS:
+					got, readErr = crypto.S3KMS.UnsealObjectKey(GlobalKMS, after.UserDefined, bucket, object)
+				default:
+					t.Fatalf("pool %d: rotation lost encryption metadata", pool)
+				}
+				mustIAM(t, readErr)
+				if got != want {
+					t.Errorf("pool %d: rotation changed the data key without reencrypting its ciphertext (rotation error: %v)", pool, err)
+				}
+				data, readErr := readCopy(pool)
+				if readErr != nil || string(data) != payload {
+					t.Errorf("pool %d: rotation left unreadable ciphertext: %v", pool, readErr)
+				}
+			}
+		})
+	}
+}
+
+// A version in a pool that a decommission or a rebalance is moving is not
+// rewritten there: only the source holds it, so the rotation is left to a
+// later run.
+func TestBatchReviewRotationLeavesMovingSource(t *testing.T) {
+	resetTestGlobals()
+	t.Cleanup(resetTestGlobals)
+	z, _ := consistencyPools(t)
+	globalIAMSys.initStore(z, nil)
+	mustIAM(t, globalIAMSys.Load(t.Context(), true))
+	const bucket = "review8-rotation-draining"
+	mustIAM(t, z.MakeBucket(t.Context(), bucket, MakeBucketOptions{VersioningEnabled: true}))
+	_, err := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, enabledBucketVersioningConfig)
+	mustIAM(t, err)
+	oldKMS := GlobalKMS
+	GlobalKMS = kms.NewStub("review8-rotation-key")
+	t.Cleanup(func() { GlobalKMS = oldKMS })
+	const object = "rotation/draining"
+	meta := make(map[string]string)
+	key, err := newEncryptMetadata(t.Context(), crypto.S3, "review8-rotation-key", nil, bucket, object, meta, nil)
+	mustIAM(t, err)
+	body, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader("draining rotation"), key))
+	mustIAM(t, err)
+	before := putConsistencyObject(t, z, bucket, object, 0, string(body), ObjectOptions{Versioned: true, UserDefined: meta, MTime: UTCNow().Add(-time.Hour)})
+	beforeInfo, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+	mustIAM(t, err)
+
+	z.poolMetaMutex.Lock()
+	z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{}
+	z.poolMetaMutex.Unlock()
+
+	r := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: ssekms, Key: "review8-rotation-key"}}
+	err = r.KeyRotate(t.Context(), z, beforeInfo, batchRootSubmitter(t))
+	if !errors.Is(err, errBatchRotateMoving) {
+		t.Errorf("rotation of a version only a draining pool holds returned %v", err)
+	}
+	after, readErr := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+	mustIAM(t, readErr)
+	if after.UserDefined[crypto.MetaSealedKeyS3] != beforeInfo.UserDefined[crypto.MetaSealedKeyS3] {
+		t.Fatal("rotation rewrote a draining pool")
+	}
+}
+
+// Once a move copied a version, the rotation leaves it to a later run, so the
+// source stays as the move's cleanup expects it.
+func TestBatchReviewRotationKeepsMigrationCleanup(t *testing.T) {
+	for _, tc := range []struct {
+		kind   string
+		rotate bool
+	}{{"decommission", false}, {"decommission", true}, {"rebalance", false}, {"rebalance", true}} {
+		name := tc.kind + "/without-rotation"
+		if tc.rotate {
+			name = tc.kind + "/with-rotation"
+		}
+		t.Run(name, func(t *testing.T) {
+			resetTestGlobals()
+			t.Cleanup(resetTestGlobals)
+			z, _ := migrationTestPools(t)
+			globalIAMSys.initStore(z, nil)
+			mustIAM(t, globalIAMSys.Load(t.Context(), true))
+			const bucket = "review8-rotation-migration"
+			mustIAM(t, z.MakeBucket(t.Context(), bucket, MakeBucketOptions{VersioningEnabled: true}))
+			_, err := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, enabledBucketVersioningConfig)
+			mustIAM(t, err)
+			oldKMS := GlobalKMS
+			GlobalKMS = kms.NewStub("review8-rotation-key")
+			t.Cleanup(func() { GlobalKMS = oldKMS })
+			const object = "rotation/migration"
+			meta := make(map[string]string)
+			key, err := newEncryptMetadata(t.Context(), crypto.S3, "", nil, bucket, object, meta, nil)
+			mustIAM(t, err)
+			body, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader("rotation and migration"), key))
+			mustIAM(t, err)
+			before := putConsistencyObject(t, z, bucket, object, 0, string(body), ObjectOptions{Versioned: true, UserDefined: meta, MTime: UTCNow().Add(-time.Hour)})
+			beforeInfo, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+			mustIAM(t, err)
+			move := migrationTestMover(t, z, 0, tc.kind)
+			mustIAM(t, move(t.Context(), 0, bucket, migrationTestReader(t, z, 0, beforeInfo)))
+			if _, err := z.serverPools[1].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID}); err != nil {
+				t.Fatalf("control: migration did not copy the version: %v", err)
+			}
+			r := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: ssekms, Key: "review8-rotation-key"}}
+			var rotationErr error
+			if tc.rotate {
+				moved, err := z.serverPools[1].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+				mustIAM(t, err)
+				rotationErr = r.KeyRotate(t.Context(), z, beforeInfo, batchRootSubmitter(t))
+				if !errors.Is(rotationErr, errBatchRotateMoving) {
+					t.Errorf("rotation during a move returned %v", rotationErr)
+				}
+				after, err := z.serverPools[1].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+				mustIAM(t, err)
+				if !maps.Equal(after.UserDefined, moved.UserDefined) || after.UserTags != moved.UserTags {
+					t.Error("rotation during a move rewrote the moved copy")
+				}
+			}
+			after, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+			mustIAM(t, err)
+			if !after.ModTime.Equal(beforeInfo.ModTime) || !maps.Equal(after.UserDefined, beforeInfo.UserDefined) {
+				t.Errorf("rotation changed a draining source after its version was copied (rotation error: %v)", rotationErr)
+			}
+			snapshot := []FileInfo{{VersionID: before.VersionID, ModTime: beforeInfo.ModTime}}
+			if err := z.removeMovedVersions(t.Context(), bucket, object, 0, snapshot, nil); err != nil {
+				t.Fatalf("rotation prevented migration cleanup: %v", err)
+			}
+			if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID}); !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+				t.Fatalf("migration left a source version behind: %v", err)
+			}
+		})
+	}
+}
+
+// The rewrite keeps the tags and the expiry of the version, which a read
+// returns apart from its metadata.
+func TestBatchReviewRotationKeepsTagsAndExpiry(t *testing.T) {
+	for _, mode := range []string{"single-pool", "multiple-pools"} {
+		t.Run(mode, func(t *testing.T) {
+			resetTestGlobals()
+			t.Cleanup(resetTestGlobals)
+			var z *erasureServerPools
+			if mode == "single-pool" {
+				z = purgeSinglePool(t, 4)
+			} else {
+				z, _ = consistencyPools(t)
+			}
+			globalIAMSys.initStore(z, nil)
+			mustIAM(t, globalIAMSys.Load(t.Context(), true))
+			const bucket = "review8-rotation-metadata"
+			mustIAM(t, z.MakeBucket(t.Context(), bucket, MakeBucketOptions{VersioningEnabled: true}))
+			_, err := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, enabledBucketVersioningConfig)
+			mustIAM(t, err)
+			oldKMS := GlobalKMS
+			GlobalKMS = kms.NewStub("review8-rotation-metadata-key")
+			t.Cleanup(func() { GlobalKMS = oldKMS })
+			const object = "rotation/tagged"
+			meta := make(map[string]string)
+			key, err := newEncryptMetadata(t.Context(), crypto.S3, "", nil, bucket, object, meta, nil)
+			mustIAM(t, err)
+			meta[xhttp.AmzObjectTagging] = "classification=private"
+			meta[ReservedMetadataPrefixLower+TaggingTimestamp] = UTCNow().Add(-time.Minute).Format(time.RFC3339Nano)
+			meta["expires"] = UTCNow().Add(time.Hour).Format(http.TimeFormat)
+			body, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader("metadata preservation"), key))
+			mustIAM(t, err)
+			written := putConsistencyObject(t, z, bucket, object, 0, string(body), ObjectOptions{Versioned: true, UserDefined: meta})
+			before, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: written.VersionID})
+			mustIAM(t, err)
+			if before.UserTags == "" || before.Expires.IsZero() {
+				t.Fatal("control: stored version is missing the metadata to preserve")
+			}
+			r := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: sses3}}
+			mustIAM(t, r.KeyRotate(t.Context(), z, before, batchRootSubmitter(t)))
+			after, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+			mustIAM(t, err)
+			if after.UserDefined[crypto.MetaSealedKeyS3] == before.UserDefined[crypto.MetaSealedKeyS3] {
+				t.Error("control: rotation did not rewrap the key")
+			}
+			if after.UserTags != before.UserTags {
+				t.Errorf("rotation changed stored tags: got %q, want %q", after.UserTags, before.UserTags)
+			}
+			if !after.Expires.Equal(before.Expires) {
+				t.Errorf("rotation changed Expires: got %v, want %v", after.Expires, before.Expires)
+			}
+		})
+	}
+}
+
+// Two pools hold the same write with different tags, with or without
+// tagging timestamps. A rotation, also during a move, leaves the tags that
+// the merged version has.
+func TestBatchReviewRotationKeepsMergedTags(t *testing.T) {
+	for _, tc := range []struct {
+		ordered, moving bool
+	}{{false, false}, {false, true}, {true, false}, {true, true}} {
+		t.Run(fmt.Sprintf("timestamp=%t/moving=%t", tc.ordered, tc.moving), func(t *testing.T) {
+			resetTestGlobals()
+			t.Cleanup(resetTestGlobals)
+			z, _ := consistencyPools(t)
+			globalIAMSys.initStore(z, nil)
+			mustIAM(t, globalIAMSys.Load(t.Context(), true))
+			const bucket = "review9-rotation-tag-tie"
+			const object = "rotation/tag-tie"
+			mustIAM(t, z.MakeBucket(t.Context(), bucket, MakeBucketOptions{VersioningEnabled: true}))
+			_, err := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, enabledBucketVersioningConfig)
+			mustIAM(t, err)
+			previousKMS := GlobalKMS
+			GlobalKMS = kms.NewStub("review9-tag-tie-key")
+			t.Cleanup(func() { GlobalKMS = previousKMS })
+			meta := make(map[string]string)
+			key, err := newEncryptMetadata(t.Context(), crypto.S3, "", nil, bucket, object, meta, nil)
+			mustIAM(t, err)
+			meta[xhttp.AmzObjectTagging] = "classification=private"
+			if tc.ordered {
+				meta[ReservedMetadataPrefixLower+TaggingTimestamp] = UTCNow().Add(-time.Minute).Format(time.RFC3339Nano)
+			}
+			other := maps.Clone(meta)
+			other[xhttp.AmzObjectTagging] = "classification=public"
+			body, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader("same write"), key))
+			mustIAM(t, err)
+			before := putConsistencyObject(t, z, bucket, object, 0, string(body), ObjectOptions{Versioned: true, UserDefined: meta, MTime: UTCNow().Add(-time.Hour)})
+			putConsistencyObject(t, z, bucket, object, 1, string(body), ObjectOptions{Versioned: true, VersionID: before.VersionID, MTime: before.ModTime, UserDefined: other})
+			prior, err := z.replicaObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+			mustIAM(t, err)
+			if prior.UserTags != "classification=private" {
+				t.Fatalf("control: logical tags=%q", prior.UserTags)
+			}
+			if tc.moving {
+				migrationTestMover(t, z, 0, "decommission")
+			}
+			r := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: sses3}}
+			rotationErr := r.KeyRotate(t.Context(), z, before, batchRootSubmitter(t))
+			after, err := z.replicaObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+			mustIAM(t, err)
+			if after.UserTags != prior.UserTags {
+				t.Fatalf("rotation changed logical tags from %q to %q (rotation error: %v)", prior.UserTags, after.UserTags, rotationErr)
+			}
+		})
+	}
+}
+
+// batchRotateFlipDisk starts a decommission of pool 0 at the flipAt-th
+// version read, after the rotation's first read of the version.
+type batchRotateFlipDisk struct {
+	StorageAPI
+	z       *erasureServerPools
+	reads   *atomic.Int32
+	flipAt  int32
+	flipped *atomic.Bool
+}
+
+func (d batchRotateFlipDisk) ReadVersion(ctx context.Context, origvolume, volume, path, versionID string, opts ReadOptions) (FileInfo, error) {
+	n := d.reads.Add(1)
+	if n == d.flipAt && d.flipped.CompareAndSwap(false, true) {
+		d.z.poolMetaMutex.Lock()
+		d.z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{}
+		d.z.poolMetaMutex.Unlock()
+	}
+	return d.StorageAPI.ReadVersion(ctx, origvolume, volume, path, versionID, opts)
+}
+
+// A move that starts while the rotation authorizes and rewraps the version
+// defers it: the state is checked again right before the write, where the
+// pools layer checks it.
+func TestBatchReviewRotationRechecksPoolBeforeWrite(t *testing.T) {
+	resetTestGlobals()
+	t.Cleanup(resetTestGlobals)
+	z, _ := consistencyPools(t)
+	globalIAMSys.initStore(z, nil)
+	mustIAM(t, globalIAMSys.Load(t.Context(), true))
+	const bucket = "review11-rotation-window"
+	const object = "rotation/window"
+	mustIAM(t, z.MakeBucket(t.Context(), bucket, MakeBucketOptions{VersioningEnabled: true}))
+	_, err := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, enabledBucketVersioningConfig)
+	mustIAM(t, err)
+	oldKMS := GlobalKMS
+	GlobalKMS = kms.NewStub("review11-rotation-key")
+	t.Cleanup(func() { GlobalKMS = oldKMS })
+
+	meta := make(map[string]string)
+	key, err := newEncryptMetadata(t.Context(), crypto.S3, "", nil, bucket, object, meta, nil)
+	mustIAM(t, err)
+	body, err := io.ReadAll(crypto.EncryptSinglePart(strings.NewReader("rotation window"), key))
+	mustIAM(t, err)
+	before := putConsistencyObject(t, z, bucket, object, 0, string(body), ObjectOptions{
+		Versioned:   true,
+		UserDefined: meta,
+		MTime:       UTCNow().Add(-time.Hour),
+	})
+	beforeInfo, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+	mustIAM(t, err)
+
+	var reads atomic.Int32
+	var flipped atomic.Bool
+	firstPass := int32(0)
+	for _, pool := range z.serverPools {
+		firstPass += int32(pool.SetDriveCount())
+	}
+	var restore []func()
+	for _, pool := range z.serverPools {
+		restore = append(restore, conditionalPutSwapDisks(pool, object, func(d StorageAPI) StorageAPI {
+			return batchRotateFlipDisk{
+				StorageAPI: d,
+				z:          z,
+				reads:      &reads,
+				flipAt:     firstPass + 1,
+				flipped:    &flipped,
+			}
+		}))
+	}
+	t.Cleanup(func() {
+		for _, fn := range restore {
+			fn()
+		}
+	})
+
+	r := BatchJobKeyRotateV1{Bucket: bucket, Encryption: BatchJobKeyRotateEncryption{Type: sses3}}
+	gotErr := r.KeyRotate(t.Context(), z, beforeInfo, batchRootSubmitter(t))
+	if !flipped.Load() {
+		t.Fatal("fault injection did not occur after the initial rotation read")
+	}
+	if !errors.Is(gotErr, errBatchRotateMoving) {
+		t.Errorf("rotation did not defer after the pool became moving during pre-work: %v", gotErr)
+	}
+	after, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: before.VersionID})
+	mustIAM(t, err)
+	if after.UserDefined[crypto.MetaSealedKeyS3] != beforeInfo.UserDefined[crypto.MetaSealedKeyS3] {
+		t.Error("rotation rewrote the pool after it became moving during pre-work")
+	}
+}
diff --git a/cmd/batch-rotate.go b/cmd/batch-rotate.go
index 1f7306ff1..b8599fd1d 100644
--- a/cmd/batch-rotate.go
+++ b/cmd/batch-rotate.go
@@ -20,6 +20,7 @@ package cmd
 import (
 	"context"
 	"encoding/base64"
+	"errors"
 	"fmt"
 	"maps"
 	"math/rand"
@@ -256,7 +257,7 @@ func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, ob
 		VersionSuspended: versionSuspended,
 		NoLock:           true,
 	}
-	obj, err := api.GetObjectInfo(ctx, r.Bucket, objInfo.Name, opts)
+	obj, pool, err := batchRotateRead(ctx, api, r.Bucket, objInfo.Name, opts)
 	if err != nil {
 		return err
 	}
@@ -273,9 +274,13 @@ func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, ob
 	if err != nil {
 		return err
 	}
-	for _, req := range r.rotateAuthzRequests(logical) {
-		if err := p.authorize(req); err != nil {
-			return err
+	// Authorize the copy that is read and rewritten, and the version as all
+	// its copies describe it.
+	for _, v := range []ObjectInfo{obj, logical} {
+		for _, req := range r.rotateAuthzRequests(v) {
+			if err := p.authorize(req); err != nil {
+				return err
+			}
 		}
 	}
 	oi := obj.Clone()
@@ -305,16 +310,61 @@ func (r *BatchJobKeyRotateV1) KeyRotate(ctx context.Context, api ObjectLayer, ob
 	oi.metadataOnly = true
 	oi.keyRotation = true
 	maps.Copy(oi.UserDefined, encMetadata)
-	if _, err := api.CopyObject(ctx, r.Bucket, oi.Name, r.Bucket, oi.Name, oi, ObjectOptions{
-		VersionID: oi.VersionID,
-	}, ObjectOptions{
-		VersionID: oi.VersionID,
-		NoLock:    true,
-	}); err != nil {
-		return err
+	// The rewrite replaces the stored metadata with UserDefined, which a
+	// read returns without the tags and the expiry: keep them.
+	if oi.UserTags != "" {
+		oi.UserDefined[xhttp.AmzObjectTagging] = oi.UserTags
 	}
+	if !oi.Expires.IsZero() {
+		oi.UserDefined["expires"] = oi.ExpiresStr()
+	}
+	srcOpts := ObjectOptions{VersionID: oi.VersionID}
+	dstOpts := ObjectOptions{VersionID: oi.VersionID, NoLock: true}
+	if z, ok := api.(*erasureServerPools); ok && pool >= 0 {
+		// Rewrite the copy that was read and authorized. The pools layer
+		// would route the rewrite by the latest version of the key, which
+		// can be in a pool that holds another write of this version ID.
+		//
+		// Check the pool's state again where the pools layer checks it,
+		// right before the write: a move can have started since the read.
+		if z.IsSuspended(pool) || z.IsPoolRebalancing(pool) {
+			return errBatchRotateMoving
+		}
+		dstOpts.replicaObjectInfo = z.replicaObjectInfo
+		name := encodeDirObject(oi.Name)
+		_, err = z.serverPools[pool].CopyObject(ctx, r.Bucket, name, r.Bucket, name, oi, srcOpts, dstOpts)
+	} else {
+		_, err = api.CopyObject(ctx, r.Bucket, oi.Name, r.Bucket, oi.Name, oi, srcOpts, dstOpts)
+	}
+	return err
+}
 
-	return nil
+// errBatchRotateMoving: the version is read from a pool that a decommission
+// or a rebalance is moving.
+var errBatchRotateMoving = errors.New("the version is being moved to another pool; rotate it once the move is done")
+
+// batchRotateRead reads the version that a rotation rewrites, as
+// GetObjectInfo does. With several pools it also returns the pool of the
+// copy it read, otherwise -1. The caller holds the object lock.
+//
+// A copy in a pool that a decommission or a rebalance is moving is the
+// source of that move, which must not change under it, and rewriting a copy
+// in another pool would change which copy the version's metadata comes
+// from. Such a version is left to a later run (errBatchRotateMoving).
+func batchRotateRead(ctx context.Context, api ObjectLayer, bucket, object string, opts ObjectOptions) (ObjectInfo, int, error) {
+	z, ok := api.(*erasureServerPools)
+	if !ok || z.SinglePool() {
+		oi, err := api.GetObjectInfo(ctx, bucket, object, opts)
+		return oi, -1, err
+	}
+	if err := checkGetObjArgs(ctx, bucket, object); err != nil {
+		return ObjectInfo{}, -1, err
+	}
+	oi, idx, err := z.getLatestObjectInfoWithIdx(ctx, bucket, object, opts)
+	if err == nil && (z.IsSuspended(idx) || z.IsPoolRebalancing(idx)) {
+		return ObjectInfo{}, -1, errBatchRotateMoving
+	}
+	return oi, idx, err
 }
 
 const (