fix(iam): persist and propagate scoped STS revocation

Enumerate durable STS identities for parent/type revocation, retain the
operation cutoff, and notify sibling caches for each committed deletion.
Repair missed notifications during periodic object/etcd IAM refresh.

Carry immutable STS tombstones and expiry through the existing IAM revision
protocol and deletion healing. Delayed creation cannot undo revocation,
while unrelated and newly issued keys stay valid. Keep partial failures
visible while propagating the successfully committed deletions.

Cover uncached credentials, warm sibling caches, dropped notifications,
object/etcd backends, reordered replay, concurrent issuance and partial
persistence failure. No new dependencies or authentication-path I/O.

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

.github/workflows/go.yml

@@ -93,8 +93,10 @@ jobs:
       - name: Run conditional PUT tests under race detector
         run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP)' -count=1 -timeout=5m
 
-      - name: Run multipart listing and cancellation tests under race detector
-        run: go test -race ./cmd -run '^Test(MultipartListing|MultipartAbort|PaginateMultipartUploads|ListMultipartUploads)' -count=1 -timeout=5m
+      - name: Run multipart listing, cancellation and cleanup tests under race detector
+        env:
+          MINIO_API_REQUESTS_MAX: "10000"
+        run: go test -race ./cmd -run '^Test(Multipart|PaginateMultipartUploads|ListMultipartUploads|CleanupStaleUploads|ExtractModTime)' -count=1 -timeout=5m
 
       - name: Run listing metadata recovery, pagination and cancellation tests under race detector
         run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|ListingRecovery|ListingPending|ListingEntry|ListingSecondLock|StreamMetadataParts|LocalLockerListing|DistributedListing|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m

CHANGELOG.md

@@ -11,9 +11,12 @@ See the [component matrix](https://silo.pgsty.com/compatibility/versions/) and
 
 ### October 18 dependency update
 
-- Select silo-pkg v3.14.2, upstream minio-go v7.4.1, published mcli 20261018
+- Select silo-pkg v3.14.3, upstream minio-go v7.4.1, published mcli 20261018
   and Console v2.4.2 source (`ef2dd2c41d21`). The SDK includes cancellation cleanup, UTC retention
   timestamps and preserved STS duration; parallel GET/Compose remain opt-in.
+- Adopt silo-pkg's faster admin-action classification during policy evaluation,
+  preserving authorization behavior and compatibility with existing clients.
+  Update the standalone `reorder-disks` debugging module to the same package version.
 - Build with Go 1.27.2 and refresh Go x/* modules. Apply go-jose v4.1.6,
   LDAP v3.4.15, compress v1.20.3, msgp v1.6.5, etcd v3.7.2, fwd v1.2.1 and
   cpuid v2.4.1 bug/security fixes. OIDC JWK RSA keys must not exceed 4096 bits.
@@ -73,19 +76,16 @@ candidate's exact delta.
 
 ### Object storage and replication
 
-- Make durable `ListMultipartUploads` scanning the only listing path (#79):
-  apply prefix, delimiter, markers, ordering and the 1,000-entry page limit
-  globally across pools and sets, independently of node-local caches. Omit
-  readable historical uploads only when both bucket/key identity fields are
-  absent; malformed identities, uncertain coverage and resource limits remain
-  errors. Old uploads still support part upload, completion and cancellation
-  with a known bucket/key/upload ID. No backfill or new storage layout is added.
-  `MINIO_API_MULTIPART_LISTING` is retired and ignored with a startup warning;
-  historical shared `api multipart_listing` keys remain tolerated. The read-only
-  admin `multipart-preflight` reports `mode=persistent`; `ready` diagnoses the
-  absence of old records and no longer gates listing. Each page still scans
-  durable state within the existing 100,000-directory-entry, 30-second and
-  per-process concurrency budgets. See the
+- Withdraw default persistent `ListMultipartUploads` scanning (#79). Restore
+  node-local cache listing for an empty prefix and exact-key lookup otherwise;
+  restart omissions, true-prefix and cache-pagination limitations remain.
+  Preserve bucket/key identity metadata, path safety and stale-upload cleanup.
+  `MINIO_API_MULTIPART_LISTING` stays retired and ignored with a startup warning;
+  obsolete shared `api multipart_listing` keys remain tolerated. **Existing
+  opt-in strict users lose durable listing and strict cancellation.** The
+  read-only admin `multipart-preflight` reports `mode=legacy`; its scan budgets
+  and readiness do not gate public listing or startup. Durable, complete,
+  scalable enumeration remains deferred. See the
   [design and limits](https://silo.pgsty.com/blog/design/list-multipart-uploads/#implementation).
   Thanks to mr javad seydi (@mrjavadseydi) for the original metadata-and-scan
   implementation (#198).

buildscripts/verify-rebrand.sh

@@ -11,6 +11,33 @@ fail() {
 	exit 1
 }
 
+# Git is already a build prerequisite. --no-index also scans untracked files
+# and exported source trees. As before, honor .gitignore only in a worktree.
+search_ignore=--no-exclude-standard
+if git rev-parse --show-toplevel >/dev/null 2>&1; then
+	search_ignore=--exclude-standard
+fi
+search_errors="$(mktemp)"
+trap 'rm -f "${search_errors}"' EXIT
+
+search_text() {
+	local status=0
+	git --no-pager grep --no-index "${search_ignore}" --no-color -nIHE "$@" 2>"${search_errors}" || status=$?
+	# Git can return 1 for both no matches and a file read error. Check stderr
+	# as well, before callers filter matches through their allowlists.
+	if [ "${status}" -gt 1 ] || [ -s "${search_errors}" ]; then
+		cat "${search_errors}" >&2
+		fail "could not scan source files"
+	fi
+}
+
+# Keep hidden files and build output out of the repository-wide scans.
+source_excludes=(
+	':(exclude,glob)**/.*'
+	':(exclude,glob)**/.*/**'
+	':(exclude)dist/**'
+)
+
 require_file() {
 	[ -f "$1" ] || fail "missing required file: $1"
 }
@@ -152,7 +179,11 @@ if grep -Ev '^[[:space:]]*(#|$)' silo.env | grep -q '='; then
 	fail "silo.env must not contain active assignments that shadow /etc/default/minio"
 fi
 
-if rg -n 'pgsty/minio:' .github/workflows Dockerfile.goreleaser helm/silo; then
+delivery_hits="$(search_text 'pgsty/minio:' -- .github/workflows Dockerfile.goreleaser helm/silo \
+	':(exclude,glob).github/workflows/**/.*' ':(exclude,glob).github/workflows/**/.*/**' \
+	':(exclude,glob)helm/silo/**/.*' ':(exclude,glob)helm/silo/**/.*/**')"
+if [ -n "${delivery_hits}" ]; then
+	printf '%s\n' "${delivery_hits}" >&2
 	fail "an active delivery surface still publishes the frozen pgsty/minio image"
 fi
 
@@ -165,10 +196,10 @@ fi
 # So two rules. First, no live URL may resolve to the old repository anywhere,
 # READMEs and CONTRIBUTORS.md included.
 old_repo_pattern='pgsty/minio(\.git)?([^[:alnum:]_.-]|$)'
-stale_repo_url="$(rg -n -e "github\.com/${old_repo_pattern}" -e "hub\.docker\.com/r/${old_repo_pattern}" \
-	--glob '!.git/**' --glob '!dist/**' \
-	--glob '!SILO_REBRANDING_MIGRATION.md' \
-	--glob '!buildscripts/rebrand-guard/compat-baseline.json' . |
+stale_repo_url="$(search_text -e "github\.com/${old_repo_pattern}" -e "hub\.docker\.com/r/${old_repo_pattern}" -- . \
+	"${source_excludes[@]}" ':(exclude)SILO_REBRANDING_MIGRATION.md' \
+	':(exclude)buildscripts/rebrand-guard/compat-baseline.json')"
+stale_repo_url="$(printf '%s\n' "${stale_repo_url}" |
 	sed 's#^\./##' | grep -v '^buildscripts/verify-rebrand\.sh:' || true)"
 if [ -n "${stale_repo_url}" ]; then
 	printf '%s\n' "${stale_repo_url}" >&2
@@ -180,32 +211,32 @@ fi
 # legacy image, the two READMEs that document the rename and the archived
 # minio branch, and historical issue titles in CONTRIBUTORS.md.
 repo_guard_allowlist='^(buildscripts/minio-upgrade\.sh|buildscripts/verify-rebrand\.sh|buildscripts/helm-migration-guard/main\.go|README\.md|README_ZH\.md|CONTRIBUTORS\.md):'
-stale_repo="$(rg -n "${old_repo_pattern}" --glob '!.git/**' --glob '!dist/**' \
-	--glob '!SILO_REBRANDING_MIGRATION.md' \
-	--glob '!buildscripts/rebrand-guard/compat-baseline.json' . |
+stale_repo="$(search_text "${old_repo_pattern}" -- . "${source_excludes[@]}" \
+	':(exclude)SILO_REBRANDING_MIGRATION.md' ':(exclude)buildscripts/rebrand-guard/compat-baseline.json')"
+stale_repo="$(printf '%s\n' "${stale_repo}" |
 	sed 's#^\./##' | grep -Ev "${repo_guard_allowlist}" || true)"
 if [ -n "${stale_repo}" ]; then
 	printf '%s\n' "${stale_repo}" >&2
 	fail "a source reference still names the pre-rename pgsty/minio repository"
 fi
 
-stale_branch="$(rg -n 'pgsty/silo/(blob/|tree/|raw/)?master' \
-	--glob '!.git/**' --glob '!dist/**' . || true)"
+stale_branch="$(search_text 'pgsty/silo/(blob/|tree/|raw/)?master' -- . "${source_excludes[@]}")"
 if [ -n "${stale_branch}" ]; then
 	printf '%s\n' "${stale_branch}" >&2
 	fail "a link still targets the retired master branch; raw and Actions URLs do not follow a branch rename"
 fi
 
 for workflow in .github/workflows/go.yml .github/workflows/vulncheck.yml; do
-	if rg -q '^\s+- master$' "${workflow}"; then
+	stale_filter="$(search_text '^[[:space:]]+- master$' -- "${workflow}")"
+	if [ -n "${stale_filter}" ]; then
 		fail "${workflow} still filters on master and would go silently dormant on main"
 	fi
 	require_text "${workflow}" "      - main"
 done
 
-network_hits="$(rg -n --glob '*.go' --glob '!**/*_test.go' \
-	'https?://[^"`[:space:]]*(dl\.min\.io|subnet\.min\.io|api\.min\.io|slack\.min\.io|play\.min\.io)' \
-	cmd internal || true)"
+network_hits="$(search_text \
+	'https?://[^"`[:space:]]*(dl\.min\.io|subnet\.min\.io|api\.min\.io|slack\.min\.io|play\.min\.io)' -- \
+	':(glob)cmd/**/*.go' ':(glob)internal/**/*.go' ':(exclude,glob)**/*_test.go' "${source_excludes[@]}")"
 network_hits="$(printf '%s\n' "${network_hits}" | grep -Ev '^[^:]+:[0-9]+:[[:space:]]*//' || true)"
 if [ -n "${network_hits}" ]; then
 	printf '%s\n' "${network_hits}" >&2

cmd/admin-handlers-users-authz_test.go

@@ -11,6 +11,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -248,8 +249,20 @@ func TestIAMAdminPolicylessSelf(t *testing.T) {
 	ctx, sys := iamAdminFixture(t)
 	caller := iamAdminUser(ctx, t, sys, "review-policyless", "")
 	svc := iamAdminServiceAccount(ctx, t, sys, caller.AccessKey, "rv-policyless-child", "")
+	other := iamAdminUser(ctx, t, sys, "bulk-other", "readwrite")
+	_ = iamAdminServiceAccount(ctx, t, sys, other.AccessKey, "bulk-other-child", "")
 	api := adminAPIHandlers{}
 	assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 200)
+	rec := iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-access-keys-bulk?listType=all", nil, nil, api.ListAccessKeysBulk)
+	assertIAMAdminStatus(t, "list bulk self", rec, 200)
+	data, err := madmin.DecryptData(caller.SecretKey, rec.Body)
+	mustIAM(t, err)
+	var listed map[string]madmin.ListAccessKeysResp
+	mustIAM(t, json.Unmarshal(data, &listed))
+	entry, ok := listed[caller.AccessKey]
+	if len(listed) != 1 || !ok || len(entry.ServiceAccounts) != 1 || entry.ServiceAccounts[0].AccessKey != svc.AccessKey || len(entry.STSKeys) != 0 {
+		t.Fatalf("policyless bulk self returned unexpected identities: %+v", listed)
+	}
 	assertIAMAdminStatus(t, "info service", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.InfoServiceAccount), 200)
 	assertIAMAdminStatus(t, "info access key", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-access-key?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.InfoAccessKey), 200)
 	assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.DeleteServiceAccount), 204)
@@ -384,6 +397,7 @@ func TestIAMAdminSelfDenials(t *testing.T) {
 			}
 			api := adminAPIHandlers{}
 			assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 403)
+			assertIAMAdminStatus(t, "list bulk self", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-access-keys-bulk?listType=all", nil, nil, api.ListAccessKeysBulk), 403)
 			assertIAMAdminStatus(t, "info service", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.InfoServiceAccount), 403)
 			assertIAMAdminStatus(t, "info access key", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-access-key?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.InfoAccessKey), 403)
 			assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 403)
@@ -477,7 +491,10 @@ func TestIAMAdminListSelfAliases(t *testing.T) {
 			ctx, sys := iamAdminFixture(t)
 			parent := iamAdminUser(ctx, t, sys, "alias-parent", "readwrite")
 			target := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-target", "")
-			caller := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-caller", "")
+			serviceCaller := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-caller", "")
+			caller := serviceCaller
+			other := iamAdminUser(ctx, t, sys, "alias-other", "readwrite")
+			_ = iamAdminServiceAccount(ctx, t, sys, other.AccessKey, "alias-other-child", "")
 			if kind == "sts" {
 				var err error
 				caller, err = auth.GetNewCredentialsWithMetadata(map[string]any{parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix()}, globalActiveCred.SecretKey)
@@ -501,6 +518,38 @@ func TestIAMAdminListSelfAliases(t *testing.T) {
 				if !found {
 					t.Fatalf("self alias %q did not list the parent's service accounts", alias)
 				}
+
+				query := url.Values{"listType": {madmin.AccessKeyListAll}}
+				if alias != "" {
+					query.Set("users", alias)
+				}
+				bulk := iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-access-keys-bulk?"+query.Encode(), nil, nil, api.ListAccessKeysBulk)
+				assertIAMAdminStatus(t, "list bulk self alias", bulk, 200)
+				data, err = madmin.DecryptData(caller.SecretKey, bulk.Body)
+				mustIAM(t, err)
+				var listed map[string]madmin.ListAccessKeysResp
+				mustIAM(t, json.Unmarshal(data, &listed))
+				entry, ok := listed[parent.AccessKey]
+				if len(listed) != 1 || !ok || len(entry.ServiceAccounts) != 2 {
+					t.Fatalf("bulk self alias %q returned unexpected identities: %+v", alias, listed)
+				}
+				want := map[string]bool{target.AccessKey: true, serviceCaller.AccessKey: true}
+				for _, account := range entry.ServiceAccounts {
+					if !want[account.AccessKey] {
+						t.Fatalf("bulk self alias %q returned unexpected or duplicate service account %s", alias, account.AccessKey)
+					}
+					delete(want, account.AccessKey)
+				}
+				if kind == "sts" {
+					if len(entry.STSKeys) != 1 || entry.STSKeys[0].AccessKey != caller.AccessKey {
+						t.Fatalf("bulk self alias %q returned unexpected STS accounts: %+v", alias, entry.STSKeys)
+					}
+				} else if len(entry.STSKeys) != 0 {
+					t.Fatalf("service caller returned unexpected STS accounts: %+v", entry.STSKeys)
+				}
+				if bytes.Contains(data, []byte(serviceCaller.SecretKey)) || bytes.Contains(data, []byte(`"secretKey"`)) || bytes.Contains(data, []byte(`"sessionToken"`)) {
+					t.Fatal("bulk listing returned credential secrets")
+				}
 			}
 		})
 	}

cmd/admin-handlers-users.go

@@ -1237,7 +1237,7 @@ func (a adminAPIHandlers) ListAccessKeysBulk(w http.ResponseWriter, r *http.Requ
 		}
 	}
 
-	if !globalIAMSys.IsAllowed(policy.Args{
+	if !isAllowedServiceAccountAction(cred, policy.Args{
 		AccountName:     cred.AccessKey,
 		Groups:          cred.Groups,
 		Action:          policy.ListServiceAccountsAdminAction,
@@ -1250,12 +1250,12 @@ func (a adminAPIHandlers) ListAccessKeysBulk(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
-	if selfOnly && len(users) == 0 {
+	if selfOnly {
 		selfUser := cred.AccessKey
 		if cred.ParentUser != "" {
 			selfUser = cred.ParentUser
 		}
-		users = append(users, selfUser)
+		users = []string{selfUser}
 	}
 
 	var checkedUserList []string

cmd/common-main.go

@@ -883,7 +883,7 @@ func serverHandleEnvVars() {
 	}
 	globalInplaceUpdateDisabled = true
 	if env.IsSet(api.EnvAPIMultipartListing) {
-		logger.Warning("%s is ignored: multipart listing always uses persistent metadata; uploads without identity fields are omitted", api.EnvAPIMultipartListing)
+		logger.Warning("%s is ignored: multipart listing uses the legacy cache/exact-key behavior; persistent listing is deferred", api.EnvAPIMultipartListing)
 	}
 
 	// Check if the supported credential env vars,

cmd/erasure-multipart-listing-benchmark_test.go

@@ -14,7 +14,7 @@ import (
 // Run explicitly with -benchtime=1x. Fixture creation is excluded from timing;
 // each request still discovers real on-disk metadata. The 16-drive/10k case
 // deliberately exceeds the existing aggregate directory-entry budget.
-func BenchmarkMultipartListingCapacity(b *testing.B) {
+func BenchmarkMultipartDiagnosticScanCapacity(b *testing.B) {
 	for _, tc := range []struct {
 		drives, uploads int
 		mixed, limited  bool
@@ -67,7 +67,7 @@ func BenchmarkMultipartListingCapacity(b *testing.B) {
 			}
 			b.ResetTimer()
 			for range b.N {
-				got, err := z.ListMultipartUploads(b.Context(), bucket, "dir/", "", "", "", 1000)
+				got, err := z.scanMultipartUploadsForTest(b.Context(), bucket, "dir/", "", "", "", 1000)
 				if tc.limited {
 					var limited SlowDown
 					if !errors.As(err, &limited) || len(got.Uploads) != 0 {
@@ -83,3 +83,88 @@ func BenchmarkMultipartListingCapacity(b *testing.B) {
 		})
 	}
 }
+
+// The public path is measured with a populated cache as well as after cache
+// loss. An empty post-restart response alone is not capacity acceptance.
+func BenchmarkMultipartPublicListingCapacity(b *testing.B) {
+	for _, tc := range []struct {
+		drives, uploads int
+		sameKey         bool
+	}{
+		{4, 10_000, false},
+		{16, 3_126, false},
+		{16, 10_000, true},
+	} {
+		b.Run(fmt.Sprintf("drives=%d/uploads=%d/same-key=%t", tc.drives, tc.uploads, tc.sameKey), func(b *testing.B) {
+			obj, dirs, err := prepareErasure(b.Context(), tc.drives)
+			if err != nil {
+				b.Fatal(err)
+			}
+			z := obj.(*erasureServerPools)
+			b.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
+			const bucket, other = "multipart-public-capacity", "multipart-public-other"
+			for _, name := range []string{bucket, other} {
+				if err := z.MakeBucket(b.Context(), name, MakeBucketOptions{}); err != nil {
+					b.Fatal(err)
+				}
+			}
+			wantCached, wantExact := 0, 1
+			for n := range tc.uploads {
+				target, key := bucket, fmt.Sprintf("dir/%05d", n)
+				if tc.sameKey {
+					key = "dir/00000"
+				} else if n%2 != 0 {
+					target = other
+				}
+				if target == bucket {
+					wantCached++
+				}
+				if _, err := z.NewMultipartUpload(b.Context(), target, key, ObjectOptions{}); err != nil {
+					b.Fatal(err)
+				}
+			}
+			if tc.sameKey {
+				wantExact = 1000
+			}
+			if tc.drives == 16 && !tc.sameKey {
+				_, err := z.scanMultipartUploadsForTest(b.Context(), bucket, "", "", "", "", 1000)
+				var limited SlowDown
+				if !errors.As(err, &limited) {
+					b.Fatalf("fixture did not reproduce the previous global budget failure: %v", err)
+				}
+			}
+			var directoryCalls, metadataCalls atomic.Int64
+			for _, set := range z.serverPools[0].sets {
+				original := set.getDisks
+				set.getDisks = func() []StorageAPI {
+					disks := append([]StorageAPI(nil), original()...)
+					for i, disk := range disks {
+						disks[i] = multipartListingCountingDisk{disk, &directoryCalls, &metadataCalls}
+					}
+					return disks
+				}
+				b.Cleanup(func() { set.getDisks = original })
+			}
+			measure := func(name, prefix string, want int) {
+				b.Run(name, func(b *testing.B) {
+					directoryCalls.Store(0)
+					metadataCalls.Store(0)
+					b.ResetTimer()
+					for range b.N {
+						got, err := z.ListMultipartUploads(b.Context(), bucket, prefix, "", "", "", 1000)
+						if err != nil || len(got.Uploads) != want {
+							b.Fatalf("public listing: count=%d want=%d err=%v", len(got.Uploads), want, err)
+						}
+					}
+					b.StopTimer()
+					b.ReportMetric(float64(directoryCalls.Load())/float64(b.N), "ListDir/op")
+					b.ReportMetric(float64(metadataCalls.Load())/float64(b.N), "ReadVersion/op")
+				})
+			}
+			measure("populated-cache", "", wantCached)
+			z.mpCache.Clear()
+			measure("lost-cache", "", 0)
+			measure("known-key", "dir/00000", wantExact)
+		})
+	}
+}

cmd/erasure-multipart-listing.go

@@ -359,7 +359,7 @@ func (z *erasureServerPools) multipartPreflight(ctx context.Context) (multipartP
 		return multipartPreflightReport{}, err
 	}
 	defer s.close()
-	report := multipartPreflightReport{Complete: true, Mode: "persistent"}
+	report := multipartPreflightReport{Complete: true, Mode: "legacy"}
 	for p, pool := range z.serverPools {
 		for i, set := range pool.sets {
 			_, _ = set.scanMultipartUploads(s, "", p, i)

cmd/erasure-multipart-listing_test.go

@@ -22,10 +22,53 @@ import (
 	"github.com/minio/minio/internal/config/storageclass"
 )
 
-// Check the real handler and storage path, including a successful abort between
-// pages. Choose IDs increasing in both initiation time and lexical order so the
-// failure does not depend on differing interpretations of S3 marker ordering.
-func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
+// The persistent scanner is retained for diagnostics. This test-only wrapper
+// preserves its pagination coverage without exposing it through public listing.
+func (z *erasureServerPools) scanMultipartUploadsForTest(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (ListMultipartsInfo, error) {
+	if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
+		return ListMultipartsInfo{}, err
+	}
+	if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
+		return ListMultipartsInfo{}, toObjectErr(err, bucket)
+	}
+	scan, err := startMultipartScan(ctx, false)
+	if err != nil {
+		return ListMultipartsInfo{}, err
+	}
+	defer scan.close()
+
+	var uploads []MultipartInfo
+	for idx, pool := range z.serverPools {
+		if z.IsSuspended(idx) {
+			continue
+		}
+		poolUploads, err := pool.scanMultipartUploadsForTest(scan, bucket, idx)
+		if err != nil {
+			return ListMultipartsInfo{}, err
+		}
+		uploads = append(uploads, poolUploads...)
+	}
+
+	// Confirmed historical records lack a recoverable bucket/key and are
+	// omitted by the scan. They do not block listing identifiable uploads.
+	return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
+}
+
+func (s *erasureSets) scanMultipartUploadsForTest(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, error) {
+	var uploads []MultipartInfo
+	for i, set := range s.sets {
+		setUploads, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
+		if err != nil {
+			return nil, err
+		}
+		uploads = append(uploads, setUploads...)
+	}
+	return uploads, nil
+}
+
+// Check that a successful HTTP abort removes the cached upload. The restored
+// cache path still ignores page limits; this is a documented legacy limitation.
+func TestMultipartListingAbortHTTPUpdatesCache(t *testing.T) {
 	z, _ := consistencyPools(t)
 	bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads", "AbortMultipart"}, MakeBucketOptions{})
 	if err != nil {
@@ -81,8 +124,8 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
 		return result
 	}
 	first := list("", "", 1)
-	if len(first.Uploads) != 1 || first.Uploads[0].UploadID != firstID || !first.IsTruncated {
-		t.Fatalf("unexpected first page: %+v", first)
+	if len(first.Uploads) != 3 || first.Uploads[0].UploadID != firstID || first.IsTruncated {
+		t.Fatalf("unexpected legacy cache view: %+v", first)
 	}
 	rec := request(http.MethodDelete, getAbortMultipartUploadURL("", bucket, "a", firstID))
 	if rec.Code != http.StatusNoContent {
@@ -93,12 +136,12 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
 	for _, upload := range rest.Uploads {
 		keys = append(keys, upload.Key)
 	}
-	t.Logf("GET first page=200; DELETE marker=204; GET next page=200 keys=%v truncated=%v", keys, rest.IsTruncated)
+	t.Logf("GET cache=200; DELETE upload=204; GET cache=200 keys=%v truncated=%v", keys, rest.IsTruncated)
 	if _, err := z.GetMultipartInfo(t.Context(), bucket, "a", secondID, ObjectOptions{}); err != nil {
 		t.Fatalf("remaining upload is not valid: %v", err)
 	}
 	if len(rest.Uploads) != 2 || rest.Uploads[0].UploadID != secondID {
-		t.Fatalf("valid remaining upload for key a is missing from continuation: %+v", rest.Uploads)
+		t.Fatalf("valid remaining upload for key a is missing from cache: %+v", rest.Uploads)
 	}
 }
 
@@ -149,7 +192,7 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
 	}
 	z.mpCache.Clear()
 	for _, target := range []string{bucket, other} {
-		got, err := z.ListMultipartUploads(t.Context(), target, "", "", "", "", 10)
+		got, err := z.scanMultipartUploadsForTest(t.Context(), target, "", "", "", "", 10)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -159,14 +202,14 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
 			requireMultipartUploadKeys(t, got)
 		}
 	}
-	// Exact-looking prefixes now have the same durable semantics.
-	exact, err := z.ListMultipartUploads(t.Context(), other, "old", "", "", "", 10)
+	// Diagnostic discovery cannot recover an unidentified historical key.
+	exact, err := z.scanMultipartUploadsForTest(t.Context(), other, "old", "", "", "", 10)
 	if err != nil {
 		t.Fatal(err)
 	}
 	requireMultipartUploadKeys(t, exact)
 	report, err := z.multipartPreflight(t.Context())
-	if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
+	if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "legacy" {
 		t.Fatalf("legacy preflight: %+v %v", report, err)
 	}
 	if err = z.AbortMultipartUpload(t.Context(), other, "old", old.UploadID, ObjectOptions{}); err != nil {
@@ -176,14 +219,14 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
 	if err != nil || !report.Ready || report.LegacyUploads != 0 {
 		t.Fatalf("drained preflight: %+v %v", report, err)
 	}
-	got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+	got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
 	if err != nil {
 		t.Fatal(err)
 	}
 	requireMultipartUploadKeys(t, got, "a")
 }
 
-func TestMultipartListingIdentityFallback(t *testing.T) {
+func TestMultipartDiagnosticIdentityFallback(t *testing.T) {
 	for _, kind := range []string{"old", "partial", "corrupt", "read-failure", "wrong-bucket"} {
 		t.Run(kind, func(t *testing.T) {
 			z, set, bucket := multipartListingFixture(t)
@@ -220,7 +263,7 @@ func TestMultipartListingIdentityFallback(t *testing.T) {
 				}
 				return disks
 			}
-			got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+			got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -469,7 +512,7 @@ func (d multipartListingCountingDisk) ReadVersion(ctx context.Context, original,
 }
 
 // Counts storage API calls; this is not a deployment throughput benchmark.
-func TestMultipartListingScanCosts(t *testing.T) {
+func TestMultipartDiagnosticScanCosts(t *testing.T) {
 	obj, dirs, err := prepareErasure(t.Context(), 4)
 	if err != nil {
 		t.Fatal(err)
@@ -507,7 +550,7 @@ func TestMultipartListingScanCosts(t *testing.T) {
 		t.Helper()
 		directoryCalls.Store(0)
 		metadataCalls.Store(0)
-		result, err := z.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 1)
+		result, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "dir/", "", "", "", 1)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -589,7 +632,7 @@ func TestMultipartListingMissingMarkers(t *testing.T) {
 	}
 }
 
-func TestMultipartListingCoverage(t *testing.T) {
+func TestMultipartDiagnosticCoverage(t *testing.T) {
 	z, set, bucket := multipartListingFixture(t)
 	mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
 	if err != nil {
@@ -605,12 +648,12 @@ func TestMultipartListingCoverage(t *testing.T) {
 		}
 	}
 	set.getDisks = func() []StorageAPI { return []StorageAPI{disks[0], disks[1], nil, nil} }
-	_, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+	_, err = z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
 	if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
 		t.Fatalf("incomplete discovery returned success: %v", err)
 	}
 	set.getDisks = func() []StorageAPI { return []StorageAPI{nil, disks[1], disks[2], disks[3]} }
-	got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+	got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -621,7 +664,7 @@ func TestMultipartListingCoverage(t *testing.T) {
 	}
 }
 
-func TestMultipartListingBudgetAndAdmission(t *testing.T) {
+func TestMultipartDiagnosticBudgetAndAdmission(t *testing.T) {
 	_, set, bucket := multipartListingFixture(t)
 	if _, err := set.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil {
 		t.Fatal(err)
@@ -651,7 +694,7 @@ func TestMultipartListingBudgetAndAdmission(t *testing.T) {
 	}
 }
 
-func TestMultipartListingAdmissionHTTP(t *testing.T) {
+func TestMultipartPublicListingAdmissionHTTP(t *testing.T) {
 	z, _, _ := multipartListingFixture(t)
 	bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
 	if err != nil {
@@ -673,12 +716,12 @@ func TestMultipartListingAdmissionHTTP(t *testing.T) {
 	}
 	rec := httptest.NewRecorder()
 	router.ServeHTTP(rec, req)
-	var response APIErrorResponse
+	var response ListMultipartUploadsResponse
 	if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
 		t.Fatal(err)
 	}
-	if rec.Code != http.StatusServiceUnavailable || response.Code != "SlowDown" {
-		t.Fatalf("admission returned %d %s", rec.Code, rec.Body.String())
+	if rec.Code != http.StatusOK || len(response.Uploads) != 0 {
+		t.Fatalf("diagnostic admission blocked public listing: %d %s", rec.Code, rec.Body.String())
 	}
 }
 
@@ -705,12 +748,12 @@ func TestMultipartListingPreflightMinorityLegacy(t *testing.T) {
 		}
 	}
 	report, err := z.multipartPreflight(t.Context())
-	if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
+	if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "legacy" {
 		t.Fatalf("minority legacy copy must prevent readiness: %+v %v", report, err)
 	}
 }
 
-func TestMultipartListingCancellationRetainsAdmission(t *testing.T) {
+func TestMultipartDiagnosticCancellationRetainsAdmission(t *testing.T) {
 	z, set, bucket := multipartListingFixture(t)
 	for i := range 40 {
 		if _, err := z.NewMultipartUpload(t.Context(), bucket, fmt.Sprintf("key-%02d", i), ObjectOptions{}); err != nil {
@@ -749,7 +792,7 @@ func TestMultipartListingCancellationRetainsAdmission(t *testing.T) {
 	defer cancel()
 	done := make(chan error, 1)
 	go func() {
-		_, err := z.ListMultipartUploads(ctx, bucket, "", "", "", "", 10)
+		_, err := z.scanMultipartUploadsForTest(ctx, bucket, "", "", "", "", 10)
 		done <- err
 	}()
 	for range 16 {

cmd/erasure-multipart-public_test.go

@@ -0,0 +1,106 @@
+// Copyright (c) 2026 Ruohang Feng
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"encoding/xml"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+)
+
+type multipartRootScanGuard struct {
+	StorageAPI
+	roots *atomic.Int64
+}
+
+func (d multipartRootScanGuard) ListDir(ctx context.Context, original, volume, dir string, count int) ([]string, error) {
+	if volume == minioMetaMultipartBucket && dir == "" {
+		d.roots.Add(1)
+		return nil, errFaultyDisk
+	}
+	return d.StorageAPI.ListDir(ctx, original, volume, dir, count)
+}
+
+// A busy or unreadable diagnostic namespace must not become a prerequisite for
+// ordinary listing, including the lower-level exact-object entry points.
+func TestMultipartPublicListingIndependentOfDiagnostics(t *testing.T) {
+	for _, busy := range []bool{false, true} {
+		t.Run(map[bool]string{false: "unreadable-root", true: "busy-diagnostics"}[busy], func(t *testing.T) {
+			z, _ := consistencyPools(t)
+			bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			const key = "dir/known"
+			mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			var roots atomic.Int64
+			for _, pool := range z.serverPools {
+				for _, set := range pool.sets {
+					original := set.getDisks
+					set.getDisks = func() []StorageAPI {
+						disks := append([]StorageAPI(nil), original()...)
+						for i, disk := range disks {
+							if disk != nil {
+								disks[i] = multipartRootScanGuard{disk, &roots}
+							}
+						}
+						return disks
+					}
+					t.Cleanup(func() { set.getDisks = original })
+				}
+			}
+			if busy {
+				for range cap(multipartScanSlots) {
+					scan, err := startMultipartScan(t.Context(), true)
+					if err != nil {
+						t.Fatal(err)
+					}
+					defer scan.close()
+				}
+			}
+			for _, prefix := range []string{"", key, "dir/"} {
+				req, err := newTestSignedRequestV4(http.MethodGet,
+					getListMultipartUploadsURLWithParams("", bucket, prefix, "", "", "", "1"),
+					0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+				if err != nil {
+					t.Fatal(err)
+				}
+				rec := httptest.NewRecorder()
+				router.ServeHTTP(rec, req)
+				if rec.Code != http.StatusOK {
+					t.Fatalf("prefix %q: HTTP %d %s", prefix, rec.Code, rec.Body.String())
+				}
+				var got ListMultipartUploadsResponse
+				if err := xml.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+					t.Fatal(err)
+				}
+				if prefix == "dir/" {
+					if len(got.Uploads) != 0 {
+						t.Fatalf("partial prefix must retain the documented exact-key limitation: %+v", got)
+					}
+				} else if len(got.Uploads) != 1 || got.Uploads[0].UploadID != mp.UploadID {
+					t.Fatalf("known upload missing: %+v", got)
+				}
+			}
+			for _, pool := range z.serverPools {
+				for _, list := range []func(context.Context, string, string, string, string, string, int) (ListMultipartsInfo, error){
+					pool.ListMultipartUploads, pool.getHashedSet(key).ListMultipartUploads,
+				} {
+					if _, err := list(t.Context(), bucket, key, "", "", "", 10); err != nil {
+						t.Fatalf("lower-level exact listing: %v", err)
+					}
+				}
+			}
+			if roots.Load() != 0 {
+				t.Fatalf("public listing attempted %d global multipart scans", roots.Load())
+			}
+		})
+	}
+}

cmd/erasure-multipart-unified_test.go

@@ -4,11 +4,10 @@
 package cmd
 
 import (
+	"bytes"
 	"context"
-	"encoding/xml"
 	"errors"
 	"net/http"
-	"net/http/httptest"
 	"slices"
 	"testing"
 )
@@ -33,14 +32,10 @@ func removeMultipartIdentity(metadata map[string]string) {
 	delete(metadata, multipartMetaObject)
 }
 
-// The actual S3 handler must use the same semantics for every prefix and page,
-// even with old records and a stale, misleading node-local cache.
-func TestMultipartListingMixedHTTP(t *testing.T) {
-	z, set, _ := multipartListingFixture(t)
-	bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
-	if err != nil {
-		t.Fatal(err)
-	}
+// The retained diagnostic scanner handles mixed records independently of a
+// stale cache. Public HTTP behavior is tested separately.
+func TestMultipartDiagnosticMixed(t *testing.T) {
+	z, set, bucket := multipartListingFixture(t)
 	for _, key := range []string{"dir/old", "dir/a", "dir/sub/b", "outside"} {
 		mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
 		if err != nil {
@@ -54,22 +49,11 @@ func TestMultipartListingMixedHTTP(t *testing.T) {
 	z.mpCache.Store("stale", MultipartInfo{Bucket: bucket, Object: "ghost", UploadID: "stale"})
 	list := func(prefix, keyMarker, uploadMarker, delimiter string) ListMultipartUploadsResponse {
 		t.Helper()
-		req, err := newTestSignedRequestV4(http.MethodGet,
-			getListMultipartUploadsURLWithParams("", bucket, prefix, keyMarker, uploadMarker, delimiter, "1"),
-			0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+		got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, prefix, keyMarker, uploadMarker, delimiter, 1)
 		if err != nil {
 			t.Fatal(err)
 		}
-		rec := httptest.NewRecorder()
-		router.ServeHTTP(rec, req)
-		var got ListMultipartUploadsResponse
-		if rec.Code != http.StatusOK {
-			t.Fatalf("list %q: HTTP %d %s", prefix, rec.Code, rec.Body.String())
-		}
-		if err := xml.Unmarshal(rec.Body.Bytes(), &got); err != nil {
-			t.Fatal(err)
-		}
-		return got
+		return generateListMultipartUploadsResponse(bucket, got, "")
 	}
 	var keys []string
 	keyMarker, uploadMarker := "", ""
@@ -103,7 +87,7 @@ func TestMultipartListingMixedHTTP(t *testing.T) {
 	}
 }
 
-func TestMultipartListingInvalidIdentity(t *testing.T) {
+func TestMultipartDiagnosticInvalidIdentity(t *testing.T) {
 	for _, kind := range []string{"missing-bucket", "missing-object", "empty-bucket", "empty-object", "empty-both", "wrong-hash", "read-failure"} {
 		t.Run(kind, func(t *testing.T) {
 			z, set, bucket := multipartListingFixture(t)
@@ -142,7 +126,7 @@ func TestMultipartListingInvalidIdentity(t *testing.T) {
 					return disks
 				}
 			}
-			got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+			got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
 			if err == nil || len(got.Uploads) != 0 || toAPIError(t.Context(), err).HTTPStatusCode != http.StatusServiceUnavailable {
 				t.Fatalf("invalid/unreadable record returned a successful page: %+v %v", got, err)
 			}
@@ -185,7 +169,14 @@ func TestMultipartListingRestart(t *testing.T) {
 	if _, cached := obj.(*erasureServerPools).mpCache.Load(mp.UploadID); cached {
 		t.Fatal("restart unexpectedly preserved the upload cache")
 	}
-	got, err := obj.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 10)
+	for _, prefix := range []string{"", "dir/"} {
+		got, err := obj.ListMultipartUploads(t.Context(), bucket, prefix, "", "", "", 10)
+		if err != nil {
+			t.Fatal(err)
+		}
+		requireMultipartUploadKeys(t, got)
+	}
+	got, err := obj.ListMultipartUploads(t.Context(), bucket, key, "", "", "", 10)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -193,9 +184,18 @@ func TestMultipartListingRestart(t *testing.T) {
 	if got.Uploads[0].UploadID != mp.UploadID {
 		t.Fatal("restart changed upload identity")
 	}
+	part, err := obj.PutObjectPart(t.Context(), bucket, key, mp.UploadID, 1,
+		mustGetPutObjReader(t, bytes.NewBufferString("part"), 4, "", ""), ObjectOptions{})
+	if err != nil {
+		t.Fatalf("continue known upload after restart: %v", err)
+	}
+	if _, err := obj.CompleteMultipartUpload(t.Context(), bucket, key, mp.UploadID,
+		[]CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{}); err != nil {
+		t.Fatalf("complete known upload after restart: %v", err)
+	}
 }
 
-func TestMultipartListingConflictingIdentityCopies(t *testing.T) {
+func TestMultipartDiagnosticConflictingIdentityCopies(t *testing.T) {
 	for _, oldCopies := range []int{1, 3} {
 		z, set, bucket := multipartListingFixture(t)
 		mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
@@ -213,14 +213,14 @@ func TestMultipartListingConflictingIdentityCopies(t *testing.T) {
 				t.Fatal(err)
 			}
 		}
-		got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+		got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
 		if !errors.Is(err, errMultipartListingIdentity) || len(got.Uploads) != 0 {
 			t.Fatalf("%d contradictory copies returned a successful page: %+v %v", oldCopies, got, err)
 		}
 	}
 }
 
-func TestMultipartListingUnsupportedStoredID(t *testing.T) {
+func TestMultipartDiagnosticUnsupportedStoredID(t *testing.T) {
 	for _, historical := range []bool{false, true} {
 		z, set, bucket := multipartListingFixture(t)
 		mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
@@ -244,7 +244,7 @@ func TestMultipartListingUnsupportedStoredID(t *testing.T) {
 				t.Fatal(err)
 			}
 		}
-		got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
+		got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
 		if historical {
 			if err != nil || len(got.Uploads) != 0 {
 				t.Fatalf("unidentified historical ID: %+v %v", got, err)

cmd/erasure-multipart.go

@@ -408,7 +408,7 @@ func paginateMultipartUploads(uploads []MultipartInfo, prefix, keyMarker, upload
 }
 
 // listMultipartUploadsExact preserves the hashed exact-object lookup used by
-// multipart write placement. Public listing always uses the durable scan.
+// multipart write placement and nonempty public listing prefixes.
 func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket, object, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
 	auditObjectErasureSet(ctx, "ListMultipartUploads", object, &er)
 
@@ -516,16 +516,7 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, prefi
 	if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
 		return ListMultipartsInfo{}, err
 	}
-	scan, err := startMultipartScan(ctx, false)
-	if err != nil {
-		return ListMultipartsInfo{}, err
-	}
-	defer scan.close()
-	uploads, err := er.scanMultipartUploads(scan, bucket, 0, 0)
-	if err != nil {
-		return ListMultipartsInfo{}, err
-	}
-	return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
+	return er.listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
 }
 
 // newMultipartUpload - wrapper for initializing a new multipart

cmd/erasure-object.go

@@ -2195,6 +2195,13 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
 		}
 	}
 
+	if opts.DataMovement && opts.DeleteMarker {
+		// Moving a null marker replaces any older null data/marker. Moving
+		// a UUID marker preserves it on retries instead of purging it.
+		deleteMarker = true
+		markDelete = opts.VersionID != nullVersionID
+	}
+
 	purge := opts.isVersionPurge()
 	if purge {
 		// A delete marker describes the version being removed; it is not an

cmd/erasure-server-pool-consistency.go

@@ -62,6 +62,125 @@ func (z *erasureServerPools) objectPoolInfos(ctx context.Context, bucket, object
 	return copies, nil
 }
 
+// checkDataMovementCopies rejects moving a version when another pool holds a
+// newer copy of it, which reads return instead. A cross-pool overwrite can
+// leave such a copy, and a client write can add one while the move copies.
+// Equal timestamps normally identify an interrupted move. Conflicting ties
+// retain the same pool-order winner as reads. While the caller holds the
+// pools-layer object lock, the returned resolver serves the set layer's
+// reconciliation of this version from the same reads. Without opts.NoLock
+// it takes the read lock itself.
+func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket, object string, opts ObjectOptions) (GetObjectInfoFn, error) {
+	if opts.MTime.IsZero() {
+		// The set layer would stamp the moved copy as the newest one.
+		return nil, errInvalidArgument
+	}
+	if !opts.NoLock {
+		lk := z.NewNSLock(bucket, object)
+		lkctx, err := lk.GetRLock(ctx, globalOperationTimeout)
+		if err != nil {
+			return nil, err
+		}
+		ctx = lkctx.Context()
+		defer lk.RUnlock(lkctx)
+	}
+	versionID := opts.VersionID
+	if versionID == "" {
+		versionID = nullVersionID
+	}
+	copies, err := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{VersionID: versionID, NoAuditLog: true})
+	if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+		return nil, err
+	}
+	// The snapshot may have been purged or replaced after the reader opened.
+	// In that case the walker must leave the current source state alone.
+	sourcePresent := false
+	for _, copy := range copies {
+		if copy.Index == opts.SrcPoolIdx && copy.ObjInfo.ModTime.Equal(opts.MTime) && copy.ObjInfo.DeleteMarker == opts.DeleteMarker {
+			sourcePresent = true
+		}
+	}
+	if !sourcePresent {
+		return nil, VersionNotFound{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID}
+	}
+	for _, copy := range copies {
+		if copy.Index != opts.SrcPoolIdx && copy.ObjInfo.ModTime.After(opts.MTime) {
+			return nil, DataMovementOverwriteErr{Bucket: bucket, Object: object, VersionID: opts.VersionID, Err: errDataMovementSuperseded}
+		}
+	}
+	if winner := copies[0]; winner.Index != opts.SrcPoolIdx && winner.ObjInfo.ModTime.Equal(opts.MTime) {
+		for _, copy := range copies {
+			if copy.Index == opts.SrcPoolIdx && (copy.ObjInfo.ETag != winner.ObjInfo.ETag || copy.ObjInfo.DeleteMarker != winner.ObjInfo.DeleteMarker || copy.ObjInfo.Size != winner.ObjInfo.Size) {
+				return nil, DataMovementOverwriteErr{Bucket: bucket, Object: object, VersionID: opts.VersionID, Err: errDataMovementSuperseded}
+			}
+		}
+	}
+	return func(ctx context.Context, b, o string, ropts ObjectOptions) (ObjectInfo, error) {
+		if ropts.VersionID == "" {
+			ropts.VersionID = nullVersionID
+		}
+		if b != bucket || o != object || ropts.VersionID != versionID {
+			return z.replicaObjectInfo(ctx, b, o, ropts)
+		}
+		return replicaCopiesInfo(copies, err, bucket, object, versionID)
+	}, nil
+}
+
+// removeMovedVersions retires only the source versions in the walker's
+// snapshot. A concurrent delete can create a new marker in the source pool;
+// deleting the whole prefix would remove that marker and expose moved data.
+func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo) error {
+	object = encodeDirObject(object)
+	lk := z.NewNSLock(bucket, object)
+	lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
+	if err != nil {
+		return err
+	}
+	ctx = lkctx.Context()
+	defer lk.Unlock(lkctx)
+	for _, version := range versions {
+		vid := version.VersionID
+		if vid == "" {
+			vid = nullVersionID
+		}
+		copies, err := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{VersionID: vid, NoAuditLog: true})
+		if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
+			continue
+		}
+		if err != nil {
+			return err
+		}
+		var src *ObjectInfo
+		var retained []PoolObjInfo
+		for _, copy := range copies {
+			if copy.Index == source {
+				src = &copy.ObjInfo
+			} else {
+				retained = append(retained, copy)
+			}
+		}
+		if src == nil {
+			continue
+		}
+		if !src.ModTime.Equal(version.ModTime) || src.DeleteMarker != version.Deleted {
+			return PreConditionFailed{}
+		}
+		// Successful movement is not enough if the destination changed again.
+		// Keep the source unless another copy still preserves its read result.
+		if len(retained) == 0 || retained[0].ObjInfo.ModTime.Before(src.ModTime) {
+			return PreConditionFailed{}
+		}
+		_, err = z.serverPools[source].DeleteObject(ctx, bucket, object, ObjectOptions{
+			VersionID: vid, NoLock: true, NoAuditLog: true,
+			SkipFreeVersion: true, // preserve the old prefix cleanup's tier-GC behavior
+		})
+		if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+			return err
+		}
+	}
+	return nil
+}
+
 // Healing also commits object metadata. Both queued healing and drive healing
 // must use the same namespace as pooled writes, rather than racing them under
 // a destination set's independent namespace.
@@ -137,11 +256,15 @@ func (z *erasureServerPools) replicaObjectInfo(ctx context.Context, bucket, obje
 		opts.VersionID = nullVersionID
 	}
 	copies, err := z.objectPoolInfos(ctx, bucket, object, opts)
+	return replicaCopiesInfo(copies, err, bucket, object, opts.VersionID)
+}
+
+func replicaCopiesInfo(copies []PoolObjInfo, err error, bucket, object, versionID string) (ObjectInfo, error) {
 	if err != nil {
 		return ObjectInfo{}, err
 	}
 	if copies[0].ObjInfo.DeleteMarker {
-		return copies[0].ObjInfo, MethodNotAllowed{Bucket: bucket, Object: decodeDirObject(object), VersionID: opts.VersionID}
+		return copies[0].ObjInfo, MethodNotAllowed{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID}
 	}
 	return mergedPoolObjectInfo(copies), nil
 }

cmd/erasure-server-pool-decom.go

@@ -616,6 +616,12 @@ func (z *erasureServerPools) decommissionObject(ctx context.Context, idx int, bu
 	}
 
 	if objInfo.isMultipart() {
+		// Completion repeats this check under the object write lock.
+		if _, err := z.checkDataMovementCopies(ctx, bucket, encodeDirObject(objInfo.Name), ObjectOptions{
+			VersionID: objInfo.VersionID, MTime: objInfo.ModTime, SrcPoolIdx: idx,
+		}); err != nil {
+			return err
+		}
 		res, err := z.NewMultipartUpload(ctx, bucket, objInfo.Name, ObjectOptions{
 			VersionID:    objInfo.VersionID,
 			UserDefined:  migrationObjectMetadata(objInfo),
@@ -656,6 +662,7 @@ func (z *erasureServerPools) decommissionObject(ctx context.Context, idx int, bu
 			}
 		}
 		_, err = z.CompleteMultipartUpload(ctx, bucket, objInfo.Name, res.UploadID, parts, ObjectOptions{
+			VersionID:    objInfo.VersionID,
 			SrcPoolIdx:   idx,
 			DataMovement: true,
 			MTime:        objInfo.ModTime,
@@ -825,27 +832,17 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 			// to create the appropriate stack.
 			versionsSorter(fivs.Versions).reverse()
 
-			var decommissioned, expired int
+			var decommissioned int
 			for _, version := range fivs.Versions {
 				stopFn := globalDecommissionMetrics.log(decomMetricDecommissionObject, idx, bi.Name, version.Name, version.VersionID)
 				// Apply lifecycle rules on the objects that are expired.
 				if filterLifecycle(bi.Name, version.Name, version) {
-					expired++
 					decommissioned++
 					stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
 					continue
 				}
 
-				// any object with only single DEL marker we don't need
-				// to decommission, just skip it, this also includes
-				// any other versions that have already expired.
-				remainingVersions := len(fivs.Versions) - expired
-				if version.Deleted && remainingVersions == 1 && rcfg == nil {
-					decommissioned++
-					stopFn(version.Size, errors.New("DELETE marked object with no other non-current versions will be skipped"))
-					continue
-				}
-
+				// Preserve every marker: older versions may live in another pool.
 				versionID := version.VersionID
 				if versionID == "" {
 					versionID = nullVersionID
@@ -870,6 +867,9 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 							SkipDecommissioned: true, // make sure we skip the decommissioned pool
 							NoAuditLog:         true,
 						})
+					if isDataMovementSuperseded(err) {
+						err = nil
+					}
 					if err != nil {
 						// This can happen when rebalance stop races with ongoing rebalance workers.
 						// These rebalance failures can be ignored.
@@ -898,18 +898,20 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 				// gr.Close() is ensured by decommissionObject().
 				for range 3 {
 					if version.IsRemote() {
-						if err := z.DecomTieredObject(ctx, bi.Name, version.Name, version, ObjectOptions{
+						err := z.DecomTieredObject(ctx, bi.Name, version.Name, version, ObjectOptions{
 							VersionID:    versionID,
 							MTime:        version.ModTime,
 							UserDefined:  version.Metadata,
 							SrcPoolIdx:   idx,
 							DataMovement: true,
-						}); err != nil {
+						})
+						if isDataMovementSuperseded(err) {
+							err = nil
+						}
 						if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) {
 							ignore = true
 							stopFn(0, nil)
 						}
-						}
 						if !ignore {
 							stopFn(version.Size, err)
 							failure = err != nil
@@ -949,7 +951,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 						stopFn(version.Size, err)
 						continue
 					}
-					if err = z.decommissionObject(ctx, idx, bi.Name, gr); err != nil {
+					if err = z.decommissionObject(ctx, idx, bi.Name, gr); err != nil && !isDataMovementSuperseded(err) {
 						if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) {
 							ignore = true
 							stopFn(0, nil)
@@ -979,15 +981,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 			// if all versions were decommissioned, then we can delete the object versions.
 			if decommissioned == len(fivs.Versions) {
 				stopFn := globalDecommissionMetrics.log(decomMetricDecommissionRemoveObject, idx, bi.Name, entry.name)
-				_, err := set.DeleteObject(ctx,
-					bi.Name,
-					encodeDirObject(entry.name),
-					ObjectOptions{
-						DeletePrefix:       true, // use prefix delete to delete all versions at once.
-						DeletePrefixObject: true, // use prefix delete on exact object (this is an optimization to avoid fan-out calls)
-						NoAuditLog:         true,
-					},
-				)
+				err := z.removeMovedVersions(ctx, bi.Name, entry.name, idx, fivs.Versions)
 				stopFn(0, err)
 				auditLogDecom(ctx, "DecomDeleteObject", bi.Name, entry.name, "", err)
 				if err != nil {
@@ -1196,10 +1190,6 @@ func (z *erasureServerPools) checkAfterDecom(ctx context.Context, idx int) error
 						ignored++
 						continue
 					}
-					if version.Deleted {
-						ignored++
-						continue
-					}
 				}
 
 				versionsFound += len(fivs.Versions) - ignored

cmd/erasure-server-pool-migration-delete_test.go

@@ -0,0 +1,169 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// This file is part of Silo Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+// Both walkers must preserve markers even when all older versions reside in
+// another pool. Local version counts cannot prove that a marker is redundant.
+func TestPoolsMigrationDeleteMarkers(t *testing.T) {
+	for _, kind := range []string{"rebalance", "decommission"} {
+		for _, pools := range []int{2, 3} {
+			for _, null := range []bool{false, true} {
+				for _, history := range []string{"none", "source", "destination"} {
+					t.Run(fmt.Sprintf("%s/pools=%d/null=%t/history=%s", kind, pools, null, history), func(t *testing.T) {
+						var z *erasureServerPools
+						var bucket string
+						if pools == 2 {
+							z, bucket = migrationTestPools(t)
+						} else {
+							z, bucket = migrationTestThreePools(t)
+						}
+						const object = "deleted"
+						old, stamp := UTCNow().Add(-time.Hour), UTCNow().Add(-time.Minute)
+						var prior ObjectInfo
+						if history != "none" {
+							pool := 0
+							if history == "destination" {
+								pool = 1
+							}
+							prior = putConsistencyObject(t, z, bucket, object, pool, "history", ObjectOptions{Versioned: true, MTime: old})
+						}
+						// A stale null copy is a normal consequence of cross-pool overwrites.
+						putConsistencyObject(t, z, bucket, object, 1, "stale", ObjectOptions{MTime: old})
+						vid := mustGetUUID()
+						if null {
+							vid = nullVersionID
+						}
+						opts := ObjectOptions{Versioned: true, VersionID: vid, DeleteMarker: true, MTime: stamp}
+						marker, err := z.serverPools[0].DeleteObject(t.Context(), bucket, object, opts)
+						mustIAM(t, err)
+						if !marker.DeleteMarker {
+							t.Fatal("fixture did not create marker")
+						}
+						if oi, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); !isErrObjectNotFound(err) || !oi.DeleteMarker {
+							t.Fatalf("fixture visible: %+v %v", oi, err)
+						}
+						if failed := migrationTestWalk(t, z, bucket, 0, kind); failed != 0 {
+							t.Fatalf("failed moves: %d", failed)
+						}
+						for attempt := 0; attempt < 2; attempt++ {
+							if oi, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); !isErrObjectNotFound(err) || !oi.DeleteMarker || !oi.ModTime.Equal(stamp) {
+								t.Fatalf("marker lost: %+v %v", oi, err)
+							}
+							if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: migrationTestVersion(marker)}); !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+								t.Fatalf("source marker retained: %v", err)
+							}
+							if prior.VersionID != "" && migrationTestBody(t, z.GetObjectNInfo, bucket, object, prior.VersionID) != "history" {
+								t.Fatal("lost historical version")
+							}
+							// Repeating a completed walk must preserve the same state.
+							migrationTestWalk(t, z, bucket, 0, kind)
+						}
+					})
+				}
+			}
+		}
+	}
+}
+
+// A reader opened before a public purge must not recreate the purged version.
+func TestPoolsMigrationDoesNotResurrectPurgedSource(t *testing.T) {
+	for _, kind := range []string{"rebalance", "decommission"} {
+		for _, multipart := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/multipart=%t", kind, multipart), func(t *testing.T) {
+				z, bucket := migrationTestPools(t)
+				move := migrationTestMover(t, z, 0, kind)
+				oi := migrationTestObject(t, z, bucket, "purged", 0, multipart, []string{"payload"}, ObjectOptions{Versioned: true})
+				gr := migrationTestReader(t, z, 0, oi)
+				_, err := z.DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionID: oi.VersionID, Versioned: true})
+				mustIAM(t, err)
+				err = move(t.Context(), 0, bucket, gr)
+				if err == nil {
+					t.Error("move accepted a purged source")
+				}
+				if oi, err := z.GetObjectInfo(t.Context(), bucket, oi.Name, ObjectOptions{}); !isErrObjectNotFound(err) {
+					t.Fatalf("purged version resurrected: %+v %v", oi, err)
+				}
+			})
+		}
+	}
+}
+
+// Cleanup must not erase a marker created after the data was copied. It must
+// also fail closed if the destination becomes unreadable before source removal.
+func TestPoolsMigrationCleanupRechecksSource(t *testing.T) {
+	for _, change := range []string{"delete-null", "delete-uuid", "unreadable-destination", "missing-destination"} {
+		t.Run(change, func(t *testing.T) {
+			z, bucket := migrationTestPools(t)
+			move := migrationTestMover(t, z, 0, "decommission")
+			oi := migrationTestObject(t, z, bucket, "cleanup", 0, false, []string{"payload"}, ObjectOptions{})
+			snapshot := []FileInfo{{VersionID: oi.VersionID, ModTime: oi.ModTime}}
+			mustIAM(t, move(t.Context(), 0, bucket, migrationTestReader(t, z, 0, oi)))
+			switch change {
+			case "delete-null", "delete-uuid":
+				_, err := z.DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionSuspended: change == "delete-null", Versioned: change == "delete-uuid"})
+				mustIAM(t, err)
+			case "unreadable-destination":
+				restore := conditionalPutSwapDisks(z.serverPools[1], oi.Name, func(d StorageAPI) StorageAPI {
+					return consistencyReadFaultDisk{StorageAPI: d, bucket: bucket, object: oi.Name}
+				})
+				defer restore()
+			case "missing-destination":
+				_, err := z.serverPools[1].DeleteObject(t.Context(), bucket, oi.Name, ObjectOptions{VersionID: nullVersionID})
+				mustIAM(t, err)
+			}
+			err := z.removeMovedVersions(t.Context(), bucket, oi.Name, 0, snapshot)
+			if change != "delete-uuid" && err == nil {
+				t.Fatal("cleanup accepted changed or uncertain state")
+			}
+			if change == "delete-null" || change == "delete-uuid" {
+				if got, err := z.GetObjectInfo(t.Context(), bucket, oi.Name, ObjectOptions{}); !isErrObjectNotFound(err) || !got.DeleteMarker {
+					t.Fatalf("cleanup resurrected data: %v", err)
+				}
+			} else if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, oi.Name, ObjectOptions{}); err != nil {
+				t.Fatalf("cleanup lost source: %v", err)
+			}
+		})
+	}
+}
+
+func TestPoolsMigrationConflictingTimestampTie(t *testing.T) {
+	for _, kind := range []string{"rebalance", "decommission"} {
+		for source := range 2 {
+			t.Run(fmt.Sprintf("%s/source=%d", kind, source), func(t *testing.T) {
+				z, bucket := migrationTestPools(t)
+				stamp := UTCNow().Add(-time.Minute)
+				for i := range 2 {
+					putConsistencyObject(t, z, bucket, "tie", i, fmt.Sprintf("pool-%d", i), ObjectOptions{MTime: stamp})
+				}
+				if got := migrationTestBody(t, z.GetObjectNInfo, bucket, "tie", ""); got != "pool-0" {
+					t.Fatalf("fixture winner: %s", got)
+				}
+				migrationTestWalk(t, z, bucket, source, kind)
+				if got := migrationTestBody(t, z.GetObjectNInfo, bucket, "tie", ""); got != "pool-0" {
+					t.Fatalf("changed tie winner: %s", got)
+				}
+			})
+		}
+	}
+}

cmd/erasure-server-pool-migration_test.go

@@ -24,11 +24,13 @@ import (
 	"fmt"
 	"io"
 	"maps"
+	"net/http"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/minio/minio/internal/bucket/lifecycle"
 	xhttp "github.com/minio/minio/internal/http"
 )
 
@@ -114,15 +116,18 @@ func migrationTestPools(t *testing.T) (*erasureServerPools, string) {
 	t.Helper()
 	z, bucket := consistencyPools(t)
 	for _, pool := range z.serverPools {
-		for _, set := range pool.sets {
-			original := set.getDisks
-			disks := append([]StorageAPI(nil), original()...)
-			for i := range disks {
-				disks[i] = tagTestCapacityDisk{StorageAPI: disks[i]}
+		pool.erasureDisksMu.Lock()
+		previous := pool.erasureDisks
+		wrapped := make([][]StorageAPI, len(previous))
+		for i, disks := range previous {
+			wrapped[i] = append([]StorageAPI(nil), disks...)
+			for j := range wrapped[i] {
+				wrapped[i][j] = tagTestCapacityDisk{StorageAPI: wrapped[i][j]}
 			}
-			set.getDisks = func() []StorageAPI { return disks }
-			t.Cleanup(func() { set.getDisks = original })
 		}
+		pool.erasureDisks = wrapped
+		pool.erasureDisksMu.Unlock()
+		t.Cleanup(func() { pool.erasureDisksMu.Lock(); pool.erasureDisks = previous; pool.erasureDisksMu.Unlock() })
 	}
 	return z, bucket
 }
@@ -131,7 +136,10 @@ func migrationTestMover(t *testing.T, z *erasureServerPools, source int, kind st
 	t.Helper()
 	if kind == "rebalance" {
 		z.rebalMu.Lock()
-		z.rebalMeta = &rebalanceMeta{PoolStats: []*rebalanceStats{{}, {}}}
+		z.rebalMeta = &rebalanceMeta{PoolStats: make([]*rebalanceStats, len(z.serverPools))}
+		for i := range z.rebalMeta.PoolStats {
+			z.rebalMeta.PoolStats[i] = &rebalanceStats{}
+		}
 		z.rebalMeta.PoolStats[source] = &rebalanceStats{Participating: true, Info: rebalanceInfo{Status: rebalStarted}}
 		z.rebalMu.Unlock()
 		t.Cleanup(func() { z.rebalMu.Lock(); z.rebalMeta = nil; z.rebalMu.Unlock() })
@@ -590,3 +598,331 @@ func TestPoolsMigrationUnreadablePool(t *testing.T) {
 		}
 	}
 }
+
+// Three pools give a move two candidate destinations.
+func migrationTestThreePools(t *testing.T) (*erasureServerPools, string) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(t.Context())
+	dirs, err := getRandomDisks(12)
+	if err != nil {
+		cancel()
+		t.Fatal(err)
+	}
+	var endpoints EndpointServerPools
+	for i := range 3 {
+		endpoints = append(endpoints, mustGetPoolEndpoints(i, dirs[4*i:4*i+4]...)...)
+	}
+	obj, _, err := initObjectLayer(ctx, endpoints)
+	if err != nil {
+		cancel()
+		removeRoots(dirs)
+		t.Fatal(err)
+	}
+	z := obj.(*erasureServerPools)
+	previous := newObjectLayerFn()
+	setObjectLayer(z)
+	t.Cleanup(func() { cancel(); z.Shutdown(context.Background()); removeRoots(dirs); setObjectLayer(previous) })
+	bucket := "pool-migration"
+	if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	return z, bucket
+}
+
+type migrationTestGetter func(context.Context, string, string, *HTTPRangeSpec, http.Header, ObjectOptions) (*GetObjectReader, error)
+
+// migrationTestBody returns the content read for a version, or the read error.
+func migrationTestBody(t *testing.T, get migrationTestGetter, bucket, object, versionID string) string {
+	t.Helper()
+	gr, err := get(t.Context(), bucket, object, nil, nil, ObjectOptions{VersionID: versionID})
+	if err != nil {
+		return fmt.Sprintf("<%v>", err)
+	}
+	defer gr.Close()
+	data, err := io.ReadAll(gr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return string(data)
+}
+
+// migrationTestWalk runs one bucket of a decommission or rebalance of source,
+// including the walker's accounting and source cleanup. It returns the number
+// of versions decommission counted as failed.
+func migrationTestWalk(t *testing.T, z *erasureServerPools, bucket string, source int, kind string) int64 {
+	t.Helper()
+	migrationTestMover(t, z, source, kind)
+	if kind == "rebalance" {
+		if err := z.rebalanceBucket(t.Context(), bucket, source); err != nil {
+			t.Fatal(err)
+		}
+		return 0
+	}
+	z.poolMetaMutex.Lock()
+	z.poolMeta.Pools[source].LastUpdate = UTCNow() // keep pool.bin and peers out of the walk
+	z.poolMetaMutex.Unlock()
+	if err := z.decommissionPool(t.Context(), source, z.serverPools[source], decomBucketInfo{Name: bucket}); err != nil {
+		t.Fatal(err)
+	}
+	z.poolMetaMutex.RLock()
+	defer z.poolMetaMutex.RUnlock()
+	return z.poolMeta.Pools[source].Decommission.ItemsDecommissionFailed
+}
+
+// migrationTestTiered stores a null version whose data is in a remote tier, in
+// the form decommission moves it.
+func migrationTestTiered(t *testing.T, z *erasureServerPools, bucket, object string, pool int, mtime time.Time) ObjectInfo {
+	t.Helper()
+	oi := putConsistencyObject(t, z, bucket, object, pool, "old", ObjectOptions{MTime: mtime})
+	infos, errs := readAllFileInfo(t.Context(), z.serverPools[pool].getHashedSet(oi.Name).getDisks(), "", bucket, oi.Name, nullVersionID, false, false)
+	if errs[0] != nil {
+		t.Fatal(errs[0])
+	}
+	fi := infos[0]
+	fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", "remote"
+	if err := z.serverPools[pool].DecomTieredObject(t.Context(), bucket, oi.Name, fi, ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	return oi
+}
+
+// A cross-pool overwrite can leave copies of one version in several pools.
+// Reads return the newest copy, and moving an older one must not replace it.
+// An equal ModTime marks the same write, as an interrupted move leaves it.
+func TestPoolsMigrationKeepsCurrentCopy(t *testing.T) {
+	z, bucket := migrationTestPools(t)
+	base := time.Date(2026, 9, 16, 8, 0, 0, 0, time.UTC)
+	for _, kind := range []string{"rebalance", "decommission"} {
+		for _, method := range []string{"put", "multipart"} {
+			for _, versioned := range []bool{false, true} {
+				for source := range 2 {
+					for _, offset := range []time.Duration{-time.Minute, 0, time.Minute} {
+						t.Run(fmt.Sprintf("%s/%s/versioned=%t/source=%d/offset=%s", kind, method, versioned, source, offset), func(t *testing.T) {
+							move := migrationTestMover(t, z, source, kind)
+							opts := ObjectOptions{Versioned: versioned, MTime: base.Add(offset)}
+							if versioned {
+								opts.VersionID = mustGetUUID()
+							}
+							oi := migrationTestObject(t, z, bucket, t.Name(), source, method == "multipart", []string{"source"}, opts)
+							opts.MTime = base
+							want, other := "source", "other"
+							if offset == 0 {
+								other = "source"
+								opts.PreserveETag = oi.ETag // interrupted copies preserve the original multipart ETag
+							} else if offset < 0 {
+								want = "other"
+							}
+							putConsistencyObject(t, z, bucket, t.Name(), 1-source, other, opts)
+							version := migrationTestVersion(oi)
+							if got := migrationTestBody(t, z.GetObjectNInfo, bucket, t.Name(), version); got != want {
+								t.Fatalf("fixture reads %q, want %q", got, want)
+							}
+							err := move(t.Context(), source, bucket, migrationTestReader(t, z, source, oi))
+							if want == "other" && !isDataMovementSuperseded(err) || want == "source" && err != nil {
+								t.Fatalf("move: %v", err)
+							}
+							if err != nil && strings.Contains(err.Error(), "CompleteMultipartUpload") {
+								t.Error("copied the parts of a superseded version")
+							}
+							if got := migrationTestBody(t, z.serverPools[1-source].GetObjectNInfo, bucket, t.Name(), version); got != want {
+								t.Errorf("destination holds %q, want %q", got, want)
+							}
+							if got := migrationTestBody(t, z.GetObjectNInfo, bucket, t.Name(), version); got != want {
+								t.Errorf("move changed the read result to %q, want %q", got, want)
+							}
+						})
+					}
+				}
+			}
+		}
+	}
+}
+
+// A client overwrite, or a replica of the same version, committed while a move
+// copies the old version is newer than the moved copy. The move must yield to
+// it, also at multipart completion.
+func TestPoolsMigrationYieldsToNewerWrite(t *testing.T) {
+	z, bucket := migrationTestPools(t)
+	for _, kind := range []string{"rebalance", "decommission"} {
+		for _, method := range []string{"put", "multipart"} {
+			for _, versioned := range []bool{false, true} {
+				for source := range 2 {
+					t.Run(fmt.Sprintf("%s/%s/versioned=%t/source=%d", kind, method, versioned, source), func(t *testing.T) {
+						migrationTestNewerWrite(t, z, bucket, kind, method, versioned, source)
+					})
+				}
+			}
+		}
+	}
+}
+
+func migrationTestNewerWrite(t *testing.T, z *erasureServerPools, bucket, kind, method string, versioned bool, source int) {
+	move := migrationTestMover(t, z, source, kind)
+	oi := migrationTestObject(t, z, bucket, t.Name(), source, method == "multipart", []string{"moved"}, ObjectOptions{Versioned: versioned, MTime: UTCNow().Add(-time.Hour)})
+	gr := migrationTestReader(t, z, source, oi)
+	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
+	defer cancel()
+	gate := newMigrationTestGate()
+	done, finished := make(chan error, 1), make(chan struct{})
+	if method == "multipart" {
+		// Hold the move after its initial check, while it copies parts.
+		gr.Reader = migrationTestGateReader{Reader: gr.Reader, ctx: ctx, gate: gate}
+		go func() { defer close(finished); done <- move(ctx, source, bucket, gr) }()
+		defer func() { gate.unblock(); cancel(); <-finished }()
+		select {
+		case <-gate.entered:
+		case err := <-done:
+			t.Fatalf("migration finished before the part barrier: %v", err)
+		case <-ctx.Done():
+			t.Fatal(ctx.Err())
+		}
+	}
+	// An ordinary PUT copies under the object lock, so this write
+	// lands between the mover's source read and its write.
+	if versioned {
+		putConsistencyObject(t, z, bucket, t.Name(), 1-source, "newer", ObjectOptions{Versioned: true, VersionID: oi.VersionID, MTime: UTCNow()})
+	} else if _, err := z.PutObject(ctx, bucket, t.Name(), mustGetPutObjReader(t, strings.NewReader("newer"), 5, "", ""), ObjectOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	var err error
+	if method == "multipart" {
+		gate.unblock()
+		err = <-done
+	} else {
+		err = move(ctx, source, bucket, gr)
+	}
+	if !isDataMovementSuperseded(err) {
+		t.Fatalf("move over a newer write: %v", err)
+	}
+	if got := migrationTestBody(t, z.GetObjectNInfo, bucket, t.Name(), migrationTestVersion(oi)); got != "newer" {
+		t.Fatalf("newer write replaced by %q", got)
+	}
+}
+
+// With two candidate destinations a superseded copy must neither replace the
+// newer copy nor become a new duplicate in the third pool.
+func TestPoolsMigrationThreePools(t *testing.T) {
+	z, bucket := migrationTestThreePools(t)
+	stale, recent := UTCNow().Add(-time.Hour), UTCNow().Add(-time.Minute)
+	for _, kind := range []string{"rebalance", "decommission"} {
+		for _, newest := range []int{0, 1} {
+			t.Run(fmt.Sprintf("%s/newest=%d", kind, newest), func(t *testing.T) {
+				move := migrationTestMover(t, z, 0, kind)
+				mtime := func(pool int) time.Time {
+					if pool == newest {
+						return recent
+					}
+					return stale
+				}
+				oi := migrationTestObject(t, z, bucket, t.Name(), 0, false, []string{"pool-0"}, ObjectOptions{MTime: mtime(0)})
+				putConsistencyObject(t, z, bucket, t.Name(), 1, "pool-1", ObjectOptions{MTime: mtime(1)})
+				if newest == 0 {
+					putConsistencyObject(t, z, bucket, t.Name(), 2, "pool-2", ObjectOptions{MTime: stale})
+				}
+				want := fmt.Sprintf("pool-%d", newest)
+				err := move(t.Context(), 0, bucket, migrationTestReader(t, z, 0, oi))
+				if newest == 1 && !isDataMovementSuperseded(err) || newest == 0 && err != nil {
+					t.Fatalf("move: %v", err)
+				}
+				if got := migrationTestBody(t, z.GetObjectNInfo, bucket, t.Name(), ""); got != want {
+					t.Errorf("reads return %q, want %q", got, want)
+				}
+				if newest == 1 {
+					if _, err := z.serverPools[2].GetObjectInfo(t.Context(), bucket, t.Name(), ObjectOptions{}); !isErrObjectNotFound(err) {
+						t.Errorf("superseded copy duplicated into pool 2: %v", err)
+					}
+				}
+			})
+		}
+	}
+}
+
+// The walkers count a superseded version as moved, so its stale source copy is
+// removed with the object and the source pool drains.
+func TestPoolsMigrationRetiresSupersededCopy(t *testing.T) {
+	for _, kind := range []string{"rebalance", "decommission"} {
+		t.Run(kind, func(t *testing.T) {
+			z, bucket := migrationTestPools(t)
+			ctx := t.Context()
+			stale, recent := UTCNow().Add(-time.Hour), UTCNow().Add(-time.Minute)
+			put := func(object string, pool int, body string, multipart bool, opts ObjectOptions) ObjectInfo {
+				return migrationTestObject(t, z, bucket, object, pool, multipart, []string{body}, opts)
+			}
+			put("stale", 0, "old", false, ObjectOptions{MTime: stale})
+			put("stale", 1, "new", false, ObjectOptions{MTime: recent})
+			put("stale-multipart", 0, "old", true, ObjectOptions{MTime: stale})
+			put("stale-multipart", 1, "new", false, ObjectOptions{MTime: recent})
+			put("newer", 0, "new", false, ObjectOptions{MTime: recent})
+			put("newer", 1, "old", false, ObjectOptions{MTime: stale})
+			oldest := put("history", 0, "old", false, ObjectOptions{Versioned: true, MTime: stale})
+			put("history", 1, "new", false, ObjectOptions{Versioned: true, MTime: recent})
+			// A suspended bucket's older null delete marker must not hide a newer
+			// null version.
+			hidden := put("stale-marker", 0, "old", false, ObjectOptions{Versioned: true, MTime: stale.Add(-time.Minute)})
+			if _, err := z.serverPools[0].DeleteObject(ctx, bucket, "stale-marker", ObjectOptions{VersionSuspended: true, MTime: stale}); err != nil {
+				t.Fatal(err)
+			}
+			put("stale-marker", 1, "new", false, ObjectOptions{VersionSuspended: true, MTime: recent})
+			objects := []string{"stale", "stale-multipart", "newer", "history", "stale-marker"}
+			if kind == "decommission" {
+				// Rebalance leaves tiered versions in place; decommission moves
+				// their metadata.
+				migrationTestTiered(t, z, bucket, "stale-tiered", 0, stale)
+				put("stale-tiered", 1, "new", false, ObjectOptions{MTime: recent})
+				objects = append(objects, "stale-tiered")
+			}
+			want := map[string]string{}
+			for _, object := range objects {
+				want[object] = migrationTestBody(t, z.GetObjectNInfo, bucket, object, "")
+			}
+			if want["stale"] != "new" || want["stale-marker"] != "new" {
+				t.Fatalf("fixture reads %v", want)
+			}
+			if failed := migrationTestWalk(t, z, bucket, 0, kind); failed != 0 {
+				t.Fatalf("%d versions failed to move", failed)
+			}
+			for _, object := range objects {
+				if got := migrationTestBody(t, z.GetObjectNInfo, bucket, object, ""); got != want[object] {
+					t.Errorf("%s: reads %q after the move, want %q", object, got, want[object])
+				}
+			}
+			var retained []string
+			for _, set := range z.serverPools[0].sets {
+				if err := set.listObjectsToDecommission(ctx, decomBucketInfo{Name: bucket}, func(entry metaCacheEntry) {
+					retained = append(retained, entry.name)
+				}); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if len(retained) > 0 {
+				t.Errorf("source pool retained %v", retained)
+			}
+			for _, oi := range []ObjectInfo{oldest, hidden} {
+				if got := migrationTestBody(t, z.GetObjectNInfo, bucket, oi.Name, oi.VersionID); got != "old" {
+					t.Errorf("%s: lost version %s: %q", oi.Name, oi.VersionID, got)
+				}
+			}
+		})
+	}
+}
+
+// A tiered move that fails, here because another pool cannot be read, must keep
+// its source like any other failed version.
+func TestPoolsMigrationTieredFailureKeepsSource(t *testing.T) {
+	z, bucket := migrationTestPools(t)
+	tiered := migrationTestTiered(t, z, bucket, t.Name(), 0, UTCNow().Add(-time.Hour))
+	set := z.serverPools[1].getHashedSet(tiered.Name)
+	original := set.getDisks
+	disks := append([]StorageAPI(nil), original()...)
+	for i := range disks {
+		disks[i] = consistencyReadFaultDisk{StorageAPI: disks[i], bucket: bucket, object: tiered.Name}
+	}
+	set.getDisks = func() []StorageAPI { return disks }
+	defer func() { set.getDisks = original }()
+	if failed := migrationTestWalk(t, z, bucket, 0, "decommission"); failed != 1 {
+		t.Fatalf("%d failed versions, want 1", failed)
+	}
+	if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, tiered.Name, ObjectOptions{VersionID: nullVersionID}); err != nil {
+		t.Fatalf("failed move removed its source: %v", err)
+	}
+}

cmd/erasure-server-pool-put-conditional_test.go

@@ -355,7 +355,7 @@ func TestPoolsConditionalPutVersionSelection(t *testing.T) {
 				opts.ReplicaLockReconcile, opts.ReplicationRequest = true, true
 				want = addressed
 			case "movement":
-				opts.DataMovement, opts.SrcPoolIdx = true, 1
+				opts.DataMovement, opts.SrcPoolIdx, opts.MTime = true, 1, addressed.ModTime
 				want = addressed
 			case "tie":
 				want = putConsistencyObject(t, z, bucket, object, 0, "tie-winner", ObjectOptions{Versioned: true, MTime: current.ModTime})

cmd/erasure-server-pool-rebalance.go

@@ -623,7 +623,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 			// to create the appropriate stack.
 			versionsSorter(fivs.Versions).reverse()
 
-			var rebalanced, expired int
+			var rebalanced int
 			for _, version := range fivs.Versions {
 				stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceObject, poolIdx, bucket, version.Name, version.VersionID)
 
@@ -635,21 +635,11 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 
 				// Apply lifecycle rules on the objects that are expired.
 				if filterLifecycle(bucket, version.Name, version) {
-					expired++
 					stopFn(version.Size, errors.New("ILM expired object/version will be skipped"))
 					continue
 				}
 
-				// any object with only single DEL marker we don't need
-				// to rebalance, just skip it, this also includes
-				// any other versions that have already expired.
-				remainingVersions := len(fivs.Versions) - expired
-				if version.Deleted && remainingVersions == 1 {
-					rebalanced++
-					stopFn(version.Size, errors.New("DELETE marked object with no other non-current versions will be skipped"))
-					continue
-				}
-
+				// Preserve every marker: older versions may live in another pool.
 				versionID := version.VersionID
 				if versionID == "" {
 					versionID = nullVersionID
@@ -671,6 +661,9 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 							SkipRebalancing:   true, // make sure we skip the decommissioned pool
 							NoAuditLog:        true,
 						})
+					if isDataMovementSuperseded(err) {
+						err = nil
+					}
 					// This can happen when rebalance stop races with ongoing rebalance workers.
 					// These rebalance failures can be ignored.
 					if err != nil {
@@ -719,7 +712,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 						continue
 					}
 
-					if err = z.rebalanceObject(ctx, poolIdx, bucket, gr); err != nil {
+					if err = z.rebalanceObject(ctx, poolIdx, bucket, gr); err != nil && !isDataMovementSuperseded(err) {
 						// This can happen when rebalance stop races with ongoing rebalance workers.
 						// These rebalance failures can be ignored.
 						if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) {
@@ -750,15 +743,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
 			// if all versions were rebalanced, we can delete the object versions.
 			if rebalanced == len(fivs.Versions) {
 				stopFn := globalRebalanceMetrics.log(rebalanceMetricRebalanceRemoveObject, poolIdx, bucket, entry.name)
-				_, err := set.DeleteObject(ctx,
-					bucket,
-					encodeDirObject(entry.name),
-					ObjectOptions{
-						DeletePrefix:       true, // use prefix delete to delete all versions at once.
-						DeletePrefixObject: true, // use prefix delete on exact object (this is an optimization to avoid fan-out calls)
-						NoAuditLog:         true,
-					},
-				)
+				err := z.removeMovedVersions(ctx, bucket, entry.name, poolIdx, fivs.Versions)
 				stopFn(0, err)
 				auditLogRebalance(ctx, "Rebalance:DeleteObject", bucket, entry.name, "", err)
 				if err != nil {
@@ -863,6 +848,12 @@ func (z *erasureServerPools) rebalanceObject(ctx context.Context, poolIdx int, b
 	}
 
 	if oi.isMultipart() {
+		// Completion repeats this check under the object write lock.
+		if _, err := z.checkDataMovementCopies(ctx, bucket, encodeDirObject(oi.Name), ObjectOptions{
+			VersionID: oi.VersionID, MTime: oi.ModTime, SrcPoolIdx: poolIdx,
+		}); err != nil {
+			return err
+		}
 		res, err := z.NewMultipartUpload(ctx, bucket, oi.Name, ObjectOptions{
 			VersionID:    oi.VersionID,
 			UserDefined:  migrationObjectMetadata(oi),
@@ -900,6 +891,8 @@ func (z *erasureServerPools) rebalanceObject(ctx context.Context, poolIdx int, b
 			}
 		}
 		_, err = z.CompleteMultipartUpload(ctx, bucket, oi.Name, res.UploadID, parts, ObjectOptions{
+			VersionID:    oi.VersionID,
+			SrcPoolIdx:   poolIdx,
 			DataMovement: true,
 			MTime:        oi.ModTime,
 			NoAuditLog:   true,
@@ -1010,7 +1003,10 @@ const (
 	rebalanceMetricSaveMetadata
 )
 
-var errDataMovementSrcDstPoolSame = errors.New("source and destination pool are the same")
+var (
+	errDataMovementSrcDstPoolSame = errors.New("source and destination pool are the same")
+	errDataMovementSuperseded     = errors.New("another pool holds a newer copy")
+)
 
 func rebalanceTrace(r rebalanceMetric, poolIdx int, startTime time.Time, duration time.Duration, err error, path string, sz int64) madmin.TraceInfo {
 	var errStr string

cmd/erasure-server-pool.go

@@ -1183,6 +1183,14 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec
 		opts.CheckPrecondFn = nil
 	}
 
+	if opts.DataMovement {
+		resolve, err := z.checkDataMovementCopies(ctx, bucket, object, opts)
+		if err != nil {
+			return ObjectInfo{}, err
+		}
+		opts.replicaObjectInfo = resolve
+	}
+
 	idx, err := z.getWritePoolIdx(ctx, bucket, object, data.Size(), true)
 	if err != nil {
 		return ObjectInfo{}, err
@@ -1202,7 +1210,7 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec
 	}
 	// Public overwrites need all pools for Object Lock. Unaddressed replica
 	// callbacks retain their set-local reads unless reconciliation is needed.
-	if opts.ReplicaLockReconcile || !opts.ReplicationRequest {
+	if opts.replicaObjectInfo == nil && (opts.ReplicaLockReconcile || !opts.ReplicationRequest) {
 		opts.replicaObjectInfo = z.replicaObjectInfo
 	}
 	oi, err := z.serverPools[idx].PutObject(ctx, bucket, object, data, opts)
@@ -1243,6 +1251,25 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
 		return ObjectInfo{}, z.deletePrefix(ctx, bucket, object)
 	}
 
+	if opts.DataMovement {
+		opts.NoLock = true
+		if _, err := z.checkDataMovementCopies(ctx, bucket, object, opts); err != nil {
+			return objInfo, err
+		}
+		// A marker still needs a destination when it is the only version of
+		// this key outside the draining pool.
+		idx, err := z.getPoolIdxNoLock(ctx, bucket, object, 0)
+		if err != nil {
+			return objInfo, err
+		}
+		if idx == opts.SrcPoolIdx {
+			return objInfo, DataMovementOverwriteErr{Bucket: bucket, Object: object, VersionID: opts.VersionID, Err: errDataMovementSrcDstPoolSame}
+		}
+		objInfo, err = z.serverPools[idx].DeleteObject(ctx, bucket, object, opts)
+		objInfo.Name = decodeDirObject(object)
+		return objInfo, err
+	}
+
 	// Resolve a physical purge by its addressed version even on a replica
 	// receiver: latest-key routing can select a different pool or miss copies.
 	// Marker creation and specialized movement/scanner operations retain their
@@ -1328,22 +1355,6 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
 		return pinfo.ObjInfo, nil
 	}
 
-	// Datamovement must never be allowed on the same pool.
-	if opts.DataMovement && opts.SrcPoolIdx == pinfo.Index {
-		return pinfo.ObjInfo, DataMovementOverwriteErr{
-			Bucket:    bucket,
-			Object:    object,
-			VersionID: opts.VersionID,
-			Err:       errDataMovementSrcDstPoolSame,
-		}
-	}
-
-	if opts.DataMovement {
-		objInfo, err = z.serverPools[pinfo.Index].DeleteObject(ctx, bucket, object, opts)
-		objInfo.Name = decodeDirObject(object)
-		return objInfo, err
-	}
-
 	// Delete concurrently in all server pools with read quorum error for unversioned objects.
 	if len(noReadQuorumPools) > 0 && !opts.Versioned && !opts.VersionSuspended {
 		return z.deleteObjectFromAllPools(ctx, bucket, object, opts, noReadQuorumPools)
@@ -1916,27 +1927,45 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
 	if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
 		return ListMultipartsInfo{}, toObjectErr(err, bucket)
 	}
-	scan, err := startMultipartScan(ctx, false)
-	if err != nil {
-		return ListMultipartsInfo{}, err
-	}
-	defer scan.close()
 
-	var uploads []MultipartInfo
+	poolResult := ListMultipartsInfo{}
+	poolResult.MaxUploads = maxUploads
+	poolResult.KeyMarker = keyMarker
+	poolResult.Prefix = prefix
+	poolResult.Delimiter = delimiter
+
+	// Preserve the legacy cache view, including its incomplete pagination.
+	// Persistent discovery remains an explicit admin diagnostic, not a
+	// prerequisite for public listing or known-upload operations.
+	if prefix == "" {
+		z.mpCache.Range(func(_ string, mp MultipartInfo) bool {
+			if mp.Bucket == bucket {
+				poolResult.Uploads = append(poolResult.Uploads, mp)
+			}
+			return true
+		})
+		sort.Slice(poolResult.Uploads, func(i int, j int) bool {
+			return poolResult.Uploads[i].Initiated.Before(poolResult.Uploads[j].Initiated)
+		})
+		return poolResult, nil
+	}
+
+	if z.SinglePool() {
+		return z.serverPools[0].getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
+	}
+
 	for idx, pool := range z.serverPools {
 		if z.IsSuspended(idx) {
 			continue
 		}
-		poolUploads, err := pool.scanMultipartUploads(scan, bucket, idx)
+		result, err := pool.getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
 		if err != nil {
-			return ListMultipartsInfo{}, err
+			return result, err
 		}
-		uploads = append(uploads, poolUploads...)
+		poolResult.Uploads = append(poolResult.Uploads, result.Uploads...)
 	}
 
-	// Confirmed historical records lack a recoverable bucket/key and are
-	// omitted by the scan. They do not block listing identifiable uploads.
-	return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
+	return poolResult, nil
 }
 
 // Initiate a new multipart upload on a hashedSet based on object name.
@@ -2230,6 +2259,13 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
 			}
 			opts.CheckPrecondFn = nil
 		}
+		if opts.DataMovement {
+			resolve, err := z.checkDataMovementCopies(ctx, bucket, encodeDirObject(object), opts)
+			if err != nil {
+				return ObjectInfo{}, err
+			}
+			opts.replicaObjectInfo = resolve
+		}
 	}
 
 	// Hold write locks to verify uploaded parts, also disallows any
@@ -2249,7 +2285,7 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
 	if opts.ReplicaLockReconcile || opts.DataMovement {
 		opts.ReplicaLockReconcile = true
 	}
-	if opts.ReplicaLockReconcile || !opts.ReplicationRequest {
+	if opts.replicaObjectInfo == nil && (opts.ReplicaLockReconcile || !opts.ReplicationRequest) {
 		opts.replicaObjectInfo = z.replicaObjectInfo
 	}
 
@@ -3240,6 +3276,11 @@ func (z *erasureServerPools) DecomTieredObject(ctx context.Context, bucket, obje
 		defer ns.Unlock(lkctx)
 		opts.NoLock = true
 	}
+	if opts.DataMovement {
+		if _, err := z.checkDataMovementCopies(ctx, bucket, object, opts); err != nil {
+			return err
+		}
+	}
 	idx, err := z.getPoolIdxNoLock(ctx, bucket, object, fi.Size)
 	if err != nil {
 		return err

cmd/erasure-sets.go

@@ -886,31 +886,7 @@ func (s *erasureSets) CopyObject(ctx context.Context, srcBucket, srcObject, dstB
 }
 
 func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
-	if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
-		return ListMultipartsInfo{}, err
-	}
-	scan, err := startMultipartScan(ctx, false)
-	if err != nil {
-		return ListMultipartsInfo{}, err
-	}
-	defer scan.close()
-	uploads, err := s.scanMultipartUploads(scan, bucket, 0)
-	if err != nil {
-		return ListMultipartsInfo{}, err
-	}
-	return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
-}
-
-func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, error) {
-	var uploads []MultipartInfo
-	for i, set := range s.sets {
-		setUploads, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
-		if err != nil {
-			return nil, err
-		}
-		uploads = append(uploads, setUploads...)
-	}
-	return uploads, nil
+	return s.getHashedSet(prefix).ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
 }
 
 func (s *erasureSets) listMultipartUploadsExact(ctx context.Context, bucket, object string) (ListMultipartsInfo, error) {

cmd/iam-replication-protocol.go

@@ -29,6 +29,7 @@ const (
 	iamRevisionPeerPath  = "/v3/site-replication/peer/iam-revisions"
 	iamUserBoundaryType  = "silo-user-revocation"
 	iamGroupBoundaryType = "silo-group-revocation"
+	iamSTSDeleteType     = "silo-sts-revocation"
 	maxIAMRevisionBatch  = 128
 )
 
@@ -49,6 +50,7 @@ type iamGroupBoundary struct {
 // before applying any change that would lose revocation or member metadata.
 type iamReplicationItem struct {
 	madmin.SRIAMItem
+	STSExpiresAt    time.Time            `json:"stsExpiresAt,omitempty"`
 	GroupGrants     map[string]time.Time `json:"groupGrants,omitempty"`
 	GroupSnapshot   bool                 `json:"groupSnapshot,omitempty"`
 	UserRevocation  *iamUserBoundary     `json:"userRevocation,omitempty"`
@@ -155,6 +157,13 @@ type (
 
 func (c *SiteReplicationSys) replicationItem(ctx context.Context, item madmin.SRIAMItem) (iamReplicationItem, error) {
 	out := iamReplicationItem{SRIAMItem: item}
+	if item.Type == iamSTSDeleteType {
+		r, err := loadIAMRevision(ctx, globalIAMSys.store, getUserIdentityPath(item.Name, stsUser))
+		if err != nil {
+			return out, err
+		}
+		out.STSExpiresAt = r.ExpiresAt
+	}
 	if item.Type == madmin.SRIAMItemSvcAcc && item.SvcAccChange != nil {
 		var key string
 		if item.SvcAccChange.Create != nil {
@@ -242,6 +251,15 @@ func applyIAMReplicationItem(ctx context.Context, item iamReplicationItem) error
 		ctx = context.WithValue(ctx, iamRecordBoundaryKey{}, item.RevokedBefore)
 	}
 	switch item.Type {
+	case iamSTSDeleteType:
+		if item.Name == "" || strings.ContainsAny(item.Name, "/\\") || item.UpdatedAt.IsZero() {
+			return errSRInvalidRequest(errInvalidArgument)
+		}
+		if err := globalIAMSys.store.revokeTempUser(withIAMReplicationTime(ctx, item.UpdatedAt), item.Name, item.STSExpiresAt); err != nil {
+			return iamReplicationError(err)
+		}
+		globalIAMSys.notifyForUser(ctx, item.Name, true)
+		return nil
 	case iamUserBoundaryType:
 		if item.UserRevocation == nil || item.UserRevocation.User == "" || item.UserRevocation.Before.IsZero() {
 			return errSRInvalidRequest(errInvalidArgument)

cmd/iam-revision.go

@@ -348,6 +348,13 @@ func saveIAMRevision(ctx context.Context, store IAMStorageAPI, path string, item
 		// STS access keys identify immutable tokens, not reusable user names.
 		return errIAMStaleUpdate
 	}
+	if strings.HasPrefix(path, iamConfigSTSPrefix) && deleted {
+		// STS keys are immutable. A delayed revocation must also defeat a
+		// replayed creation that was recorded with a newer event timestamp.
+		if origin, replicated := iamReplicationTime(ctx); replicated && previous.timestamp().After(origin) {
+			ctx = withIAMReplicationTime(ctx, previous.timestamp())
+		}
+	}
 	if origin, replicated := iamReplicationTime(ctx); replicated {
 		*at = origin
 		if previous.timestamp().After(origin) || (previous.Deleted && !deleted && !origin.After(previous.timestamp())) {
@@ -394,12 +401,19 @@ func saveIAMRevision(ctx context.Context, store IAMStorageAPI, path string, item
 			// Retain only the parent name for root-account exclusion during heal.
 			v.Credentials = auth.Credentials{ParentUser: previous.Credentials.ParentUser}
 			v.RevokedBefore = *at
-			if strings.HasPrefix(path, iamConfigSTSPrefix) && !previous.Credentials.Expiration.IsZero() && !previous.Credentials.Expiration.Equal(timeSentinel) {
-				// The signed STS token cannot authorize beyond this time, even
-				// if an offline site replays it with a newer event timestamp.
-				v.ExpiresAt = previous.Credentials.Expiration.Add(globalMaxSkewTime)
+			if strings.HasPrefix(path, iamConfigSTSPrefix) {
+				// Carry expiry when a revocation arrives before the token itself.
+				// It is safe to forget this key only after its signed lifetime.
+				if expiration := previous.Credentials.Expiration; !expiration.IsZero() && !expiration.Equal(timeSentinel) && expiration.Add(globalMaxSkewTime).After(v.ExpiresAt) {
+					v.ExpiresAt = expiration.Add(globalMaxSkewTime)
+				}
+				if previous.ExpiresAt.After(v.ExpiresAt) {
+					v.ExpiresAt = previous.ExpiresAt
+				}
+				if !v.ExpiresAt.IsZero() {
 					opts = []options{{ttl: max(1, int64(time.Until(v.ExpiresAt).Seconds())+1)}}
 				}
+			}
 		} else {
 			if v.Credentials.SessionToken != "" && v.Credentials.Claims == nil {
 				claims, err := extractJWTClaims(*v)
@@ -489,6 +503,12 @@ func iamDeletionItem(path string, r iamRevision) (item madmin.SRIAMItem, ok bool
 		name := strings.TrimSuffix(strings.TrimPrefix(path, iamConfigUsersPrefix), "/"+iamIdentityFile)
 		item.Type = madmin.SRIAMItemIAMUser
 		item.IAMUser = &madmin.SRIAMUser{AccessKey: name, IsDeleteReq: true}
+	case strings.HasPrefix(path, iamConfigSTSPrefix):
+		if !r.ExpiresAt.IsZero() && UTCNow().After(r.ExpiresAt) {
+			return item, false
+		}
+		item.Type = iamSTSDeleteType
+		item.Name = strings.TrimSuffix(strings.TrimPrefix(path, iamConfigSTSPrefix), "/"+iamIdentityFile)
 	case strings.HasPrefix(path, iamConfigServiceAccountsPrefix):
 		name := strings.TrimSuffix(strings.TrimPrefix(path, iamConfigServiceAccountsPrefix), "/"+iamIdentityFile)
 		if name == siteReplicatorSvcAcc || r.Credentials.ParentUser == globalActiveCred.AccessKey {
@@ -518,8 +538,7 @@ func iamDeletionItem(path string, r iamRevision) (item madmin.SRIAMItem, ok bool
 		item.Type = madmin.SRIAMItemPolicyMapping
 		item.PolicyMapping = &madmin.SRPolicyMapping{UserOrGroup: strings.TrimSuffix(name, ".json"), UserType: int(typ), IsGroup: prefix == "groups"}
 	default:
-		// Expired STS credentials are not replayed. Parent revocations and
-		// their retained timestamp reject delayed copies of derived tokens.
+		// Other config paths do not represent replicated IAM deletions.
 		return item, false
 	}
 	return item, true
@@ -527,6 +546,8 @@ func iamDeletionItem(path string, r iamRevision) (item madmin.SRIAMItem, ok bool
 
 func iamDeletionPath(item madmin.SRIAMItem) string {
 	switch item.Type {
+	case iamSTSDeleteType:
+		return getUserIdentityPath(item.Name, stsUser)
 	case madmin.SRIAMItemIAMUser:
 		if item.IAMUser != nil && item.IAMUser.IsDeleteReq {
 			return getUserIdentityPath(item.IAMUser.AccessKey, regUser)
@@ -582,6 +603,9 @@ func (c *SiteReplicationSys) healIAMDeletions(ctx context.Context) (err error) {
 			continue
 		}
 		out := iamReplicationItem{SRIAMItem: item}
+		if item.Type == iamSTSDeleteType {
+			out.STSExpiresAt = r.ExpiresAt
+		}
 		if item.Type == madmin.SRIAMItemIAMUser && !r.Deleted {
 			out.Type, out.IAMUser = iamUserBoundaryType, nil
 			out.UserRevocation = &iamUserBoundary{User: item.IAMUser.AccessKey, Before: r.RevokedBefore}
@@ -593,7 +617,7 @@ func (c *SiteReplicationSys) healIAMDeletions(ctx context.Context) (err error) {
 		byType[item.Type] = append(byType[item.Type], out)
 	}
 	var items []iamReplicationItem
-	for _, typ := range []string{madmin.SRIAMItemPolicyMapping, madmin.SRIAMItemIAMUser, madmin.SRIAMItemSvcAcc, madmin.SRIAMItemGroupInfo, madmin.SRIAMItemPolicy} {
+	for _, typ := range []string{madmin.SRIAMItemPolicyMapping, madmin.SRIAMItemIAMUser, madmin.SRIAMItemSvcAcc, iamSTSDeleteType, madmin.SRIAMItemGroupInfo, madmin.SRIAMItemPolicy} {
 		items = append(items, byType[typ]...)
 	}
 	if len(items) == 0 {

cmd/iam-store.go

@@ -711,6 +711,11 @@ func (store *IAMStoreSys) LoadIAMCache(ctx context.Context, firstTime bool) erro
 		if err := store.loadUsers(ctx, svcUser, newCache.iamUsersMap); err != nil {
 			return err
 		}
+		// Rebuild STS revisions as well: a missed etcd watch event must not
+		// leave a cached token valid or hide its tombstone after a restart.
+		if err := store.loadUsers(ctx, stsUser, newCache.iamSTSAccountsMap); err != nil {
+			return err
+		}
 
 		newCache.buildUserGroupMemberships()
 	}
@@ -754,6 +759,14 @@ func (store *IAMStoreSys) LoadIAMCache(ctx context.Context, firstTime bool) erro
 		cache.updatedAt = time.Now()
 	}
 
+	// Revision loading is independent of cache replacement. Even if ongoing
+	// IAM writes defer the full swap, a missed STS notification must converge.
+	for key := range cache.iamSTSAccountsMap {
+		if store.revisionIndex().get(getUserIdentityPath(key, stsUser)).Deleted {
+			delete(cache.iamSTSAccountsMap, key)
+		}
+	}
+
 	return nil
 }
 
@@ -2184,45 +2197,68 @@ func (store *IAMStoreSys) SetTempUser(ctx context.Context, accessKey string, cre
 
 // RevokeTokens - revokes all temporary credentials, or those with matching type,
 // associated with the parent user.
-func (store *IAMStoreSys) RevokeTokens(ctx context.Context, parentUser string, tokenRevokeType string) error {
+func (store *IAMStoreSys) RevokeTokens(ctx context.Context, parentUser string, tokenRevokeType string) ([]string, error) {
 	if parentUser == "" {
-		return errInvalidArgument
+		return nil, errInvalidArgument
 	}
-
+	before := UTCNow()
 	cache := store.lock()
 	defer store.unlock()
-
 	secret, err := getTokenSigningKey()
 	if err != nil {
-		return err
+		return nil, err
 	}
-
-	var revoked bool
-	for _, ui := range cache.iamSTSAccountsMap {
-		if ui.Credentials.ParentUser != parentUser {
+	// Another node may have issued a token whose notification was lost.
+	// Enumerate durable identities, retaining cached entries for retrying a
+	// partially propagated revocation. Do not hold up cached authentication.
+	accounts := maps.Clone(cache.iamSTSAccountsMap)
+	if err := store.withIAMStorage(ctx, func(ctx context.Context) error {
+		return store.loadUsers(ctx, stsUser, accounts)
+	}); err != nil {
+		return nil, err
+	}
+	var revoked []string
+	var failures []error
+	for key, ui := range accounts {
+		if ui.Credentials.ParentUser != parentUser || ui.UpdatedAt.After(before) || ui.Credentials.IsExpired() {
 			continue
 		}
 		if tokenRevokeType != "" {
 			claims, err := getClaimsFromTokenWithSecret(ui.Credentials.SessionToken, secret)
 			if err != nil {
-				continue // skip if token is invalid
+				continue
 			}
-			// skip if token type is given and does not match
 			if v, _ := claims.Lookup(tokenRevokeTypeClaim); v != tokenRevokeType {
 				continue
 			}
 		}
-		if err := store.deleteUserIdentity(ctx, ui.Credentials.AccessKey, stsUser); err != nil {
-			return err
+		if err := store.deleteUserIdentity(ctx, key, stsUser); err != nil {
+			failures = append(failures, err)
+			continue
 		}
-		delete(cache.iamSTSAccountsMap, ui.Credentials.AccessKey)
-		revoked = true
+		delete(cache.iamSTSAccountsMap, key)
+		revoked = append(revoked, key)
 	}
-
-	if revoked {
+	if len(revoked) != 0 {
 		cache.updatedAt = time.Now()
 	}
+	return revoked, errors.Join(failures...)
+}
 
+// revokeTempUser applies an immutable token's replicated tombstone without
+// touching the parent, its policies, or other (including newer) tokens.
+func (store *IAMStoreSys) revokeTempUser(ctx context.Context, key string, expiresAt time.Time) error {
+	cache := store.lock()
+	defer store.unlock()
+	u := UserIdentity{Version: 1, Deleted: true, ExpiresAt: expiresAt}
+	if err := store.saveUserIdentity(ctx, key, stsUser, &u); err != nil {
+		return err
+	}
+	delete(cache.iamSTSAccountsMap, key)
+	cache.updatedAt = time.Now()
+	if store.group != nil {
+		store.group.Forget(key)
+	}
 	return nil
 }
 

cmd/iam-sts-revocation_test.go

@@ -0,0 +1,254 @@
+// Copyright (c) 2026 Feng Ruohang
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/auth"
+	"github.com/minio/minio/internal/grid"
+	xnet "github.com/pgsty/silo-pkg/v3/net"
+)
+
+func TestIAMSTSRevocationDurableScope(t *testing.T) {
+	for _, backend := range []string{"object", "etcd"} {
+		for _, notify := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/notify=%t", backend, notify), func(t *testing.T) {
+				ctx, sys, obj := prepareIAMRevisionFixture(t, backend)
+				for _, parent := range []string{"sts-parent", "other-parent"} {
+					_, err := sys.CreateUser(ctx, parent, madmin.AddOrUpdateUserReq{SecretKey: "synthetic-parent-password", Status: madmin.AccountEnabled})
+					mustIAM(t, err)
+				}
+				sibling := &IAMStoreSys{IAMStorageAPI: newIAMObjectStore(obj, MinIOUsersSysType)}
+				if backend == "etcd" {
+					sibling.IAMStorageAPI = newIAMEtcdStore(sys.store.IAMStorageAPI.(*IAMEtcdStore).client, MinIOUsersSysType)
+				}
+				mustIAM(t, sibling.LoadIAMCache(ctx, false))
+				secret, err := getTokenSigningKey()
+				mustIAM(t, err)
+				issue := func(store *IAMStoreSys, parent, typ string) auth.Credentials {
+					cred, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: parent, tokenRevokeTypeClaim: typ}, secret)
+					mustIAM(t, err)
+					cred.ParentUser = parent
+					_, err = store.SetTempUser(ctx, cred.AccessKey, cred, "")
+					mustIAM(t, err)
+					return cred
+				}
+				first := issue(sys.store, "sts-parent", "selected")
+				uncached := issue(sibling, "sts-parent", "selected")
+				otherType := issue(sys.store, "sts-parent", "other-type")
+				otherUser := issue(sys.store, "other-parent", "selected")
+				if _, ok := sys.store.GetUser(uncached.AccessKey); ok {
+					t.Fatal("fixture cached the remote token")
+				}
+				for _, c := range []auth.Credentials{first, otherType, otherUser} {
+					mustIAM(t, sibling.UserNotificationHandler(ctx, c.AccessKey, stsUser))
+				}
+				prior := globalNotificationSys
+				globalNotificationSys = &NotificationSys{}
+				t.Cleanup(func() { globalNotificationSys = prior })
+				if notify {
+					tg, err := grid.SetupTestGrid(2)
+					mustIAM(t, err)
+					t.Cleanup(tg.Cleanup)
+					mustIAM(t, loadUserRPC.Register(tg.Managers[1], func(r *grid.MSS) (grid.NoPayload, *grid.RemoteErr) {
+						if err := sibling.UserNotificationHandler(ctx, r.Get(peerRESTUser), stsUser); err != nil {
+							return grid.NoPayload{}, grid.NewRemoteErr(err)
+						}
+						return grid.NoPayload{}, nil
+					}))
+					host, err := xnet.ParseHost(strings.TrimPrefix(tg.Hosts[1], "http://"))
+					mustIAM(t, err)
+					globalNotificationSys = &NotificationSys{peerClients: []*peerRESTClient{{host: host, gridConn: func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) }}}}
+				}
+				mustIAM(t, sys.RevokeTokens(ctx, "sts-parent", "selected"))
+				for _, c := range []auth.Credentials{first, uncached} {
+					r, err := loadIAMRevision(ctx, sys.store, getUserIdentityPath(c.AccessKey, stsUser))
+					mustIAM(t, err)
+					if !r.Deleted {
+						t.Errorf("durable token was omitted: %s", c.AccessKey)
+					}
+					if _, ok := iamDeletionItem(getUserIdentityPath(c.AccessKey, stsUser), r); !ok {
+						t.Error("STS revocation cannot be replayed")
+					}
+					if notify && backend == "object" {
+						if _, ok := sibling.GetUser(c.AccessKey); ok {
+							t.Error("notified sibling accepted revoked token")
+						}
+					}
+				}
+				fresh := issue(sibling, "sts-parent", "selected")
+				// A missed notification must converge through the normal cache refresh.
+				mustIAM(t, sibling.LoadIAMCache(ctx, false))
+				for _, c := range []auth.Credentials{first, uncached} {
+					if _, ok := sibling.GetUser(c.AccessKey); ok {
+						t.Error("refresh retained revoked token")
+					}
+				}
+				for _, c := range []auth.Credentials{otherType, otherUser, fresh} {
+					mustIAM(t, sibling.UserNotificationHandler(ctx, c.AccessKey, stsUser))
+					if _, ok := sibling.GetUser(c.AccessKey); !ok {
+						t.Error("revocation removed an out-of-scope or new token")
+					}
+				}
+				if _, ok := sibling.GetUser("sts-parent"); !ok {
+					t.Error("revocation removed parent")
+				}
+			})
+		}
+	}
+}
+
+// Deletion is keyed by the immutable STS access key, so delayed delivery cannot
+// affect a newly issued token and a delayed creation cannot undo revocation.
+func TestIAMSTSRevocationReorderedReplay(t *testing.T) {
+	for _, backend := range []string{"object", "etcd"} {
+		for _, missing := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/missing=%t", backend, missing), func(t *testing.T) {
+				ctx, sys, _ := prepareIAMRevisionFixture(t, backend)
+				const parent = "sts-replay-parent"
+				_, err := sys.CreateUser(ctx, parent, madmin.AddOrUpdateUserReq{SecretKey: "synthetic-parent-password", Status: madmin.AccountEnabled})
+				mustIAM(t, err)
+				secret, err := getTokenSigningKey()
+				mustIAM(t, err)
+				cred, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: parent}, secret)
+				mustIAM(t, err)
+				cred.ParentUser = parent
+				created := UTCNow().Add(-time.Minute)
+				peer := madmin.SRSTSCredential{AccessKey: cred.AccessKey, SecretKey: cred.SecretKey, SessionToken: cred.SessionToken, ParentUser: parent}
+				if !missing {
+					mustIAM(t, globalSiteReplicationSys.PeerSTSAccHandler(ctx, &peer, created))
+				}
+				fresh, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: parent}, secret)
+				mustIAM(t, err)
+				fresh.ParentUser = parent
+				_, err = sys.SetTempUser(ctx, fresh.AccessKey, fresh, "")
+				mustIAM(t, err)
+				// Older than the live record on this peer: deletion must still win.
+				item := iamReplicationItem{SRIAMItem: madmin.SRIAMItem{Type: iamSTSDeleteType, Name: cred.AccessKey, UpdatedAt: created.Add(-time.Second)}, STSExpiresAt: cred.Expiration.Add(globalMaxSkewTime)}
+				for range 2 {
+					mustIAM(t, applyIAMReplicationItem(ctx, item))
+					mustIAM(t, globalSiteReplicationSys.PeerSTSAccHandler(ctx, &peer, UTCNow()))
+					mustIAM(t, sys.store.LoadIAMCache(ctx, false))
+					if _, ok := sys.GetUser(ctx, cred.AccessKey); ok {
+						t.Fatal("delayed STS creation undid revocation")
+					}
+					if _, ok := sys.GetUser(ctx, fresh.AccessKey); !ok {
+						t.Fatal("delayed revocation removed new token")
+					}
+				}
+				r, err := loadIAMRevision(ctx, sys.store, getUserIdentityPath(cred.AccessKey, stsUser))
+				mustIAM(t, err)
+				if !r.Deleted || !r.ExpiresAt.Equal(item.STSExpiresAt) || r.Credentials.SessionToken != "" || r.Credentials.SecretKey != "" {
+					t.Fatal("revocation lost expiry or retained secret material")
+				}
+				for _, status := range []madmin.AccountStatus{madmin.AccountDisabled, madmin.AccountEnabled} {
+					_, err = sys.SetUserStatus(ctx, parent, status)
+					mustIAM(t, err)
+				}
+				if _, ok := sys.GetUser(ctx, cred.AccessKey); ok {
+					t.Fatal("parent re-enable revived revoked token")
+				}
+			})
+		}
+	}
+}
+
+// Pause the durable scan while another node persists a new credential. The
+// revocation's original cutoff must survive the scan and leave that token live.
+type iamSTSScanGateStore struct {
+	IAMStorageAPI
+	entered chan struct{}
+	resume  chan struct{}
+}
+
+func (s *iamSTSScanGateStore) loadUsers(ctx context.Context, typ IAMUserType, users map[string]UserIdentity) error {
+	close(s.entered)
+	select {
+	case <-s.resume:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+	return s.IAMStorageAPI.loadUsers(ctx, typ, users)
+}
+
+func TestIAMSTSRevocationConcurrentIssuance(t *testing.T) {
+	ctx, sys, obj := prepareIAMRevisionFixture(t)
+	_, err := sys.CreateUser(ctx, "sts-race-parent", madmin.AddOrUpdateUserReq{SecretKey: "synthetic-parent-password", Status: madmin.AccountEnabled})
+	mustIAM(t, err)
+	sibling := &IAMStoreSys{IAMStorageAPI: newIAMObjectStore(obj, MinIOUsersSysType)}
+	secret, err := getTokenSigningKey()
+	mustIAM(t, err)
+	issue := func() auth.Credentials {
+		c, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: "sts-race-parent"}, secret)
+		mustIAM(t, err)
+		c.ParentUser = "sts-race-parent"
+		_, err = sibling.SetTempUser(ctx, c.AccessKey, c, "")
+		mustIAM(t, err)
+		return c
+	}
+	old := issue()
+	gate := &iamSTSScanGateStore{IAMStorageAPI: sys.store.IAMStorageAPI, entered: make(chan struct{}), resume: make(chan struct{})}
+	sys.store.IAMStorageAPI = gate
+	done := make(chan error, 1)
+	go func() { done <- sys.RevokeTokens(ctx, "sts-race-parent", "") }()
+	<-gate.entered
+	fresh := issue()
+	close(gate.resume)
+	mustIAM(t, <-done)
+	mustIAM(t, sibling.UserNotificationHandler(ctx, old.AccessKey, stsUser))
+	if _, ok := sibling.GetUser(old.AccessKey); ok {
+		t.Fatal("old uncached token survived")
+	}
+	mustIAM(t, sibling.UserNotificationHandler(ctx, fresh.AccessKey, stsUser))
+	if _, ok := sibling.GetUser(fresh.AccessKey); !ok {
+		t.Fatal("new token was included in earlier revocation")
+	}
+}
+
+type iamSTSFailureStore struct {
+	IAMStorageAPI
+	failedPath string
+}
+
+func (s *iamSTSFailureStore) saveIAMConfig(ctx context.Context, item any, path string, opts ...options) error {
+	if path == s.failedPath {
+		return errIAMInjectedWrite
+	}
+	return s.IAMStorageAPI.saveIAMConfig(ctx, item, path, opts...)
+}
+
+func TestIAMSTSRevocationPartialFailure(t *testing.T) {
+	ctx, sys, _ := prepareIAMRevisionFixture(t)
+	_, err := sys.CreateUser(ctx, "sts-partial-parent", madmin.AddOrUpdateUserReq{SecretKey: "synthetic-parent-password", Status: madmin.AccountEnabled})
+	mustIAM(t, err)
+	secret, err := getTokenSigningKey()
+	mustIAM(t, err)
+	var keys []string
+	for range 2 {
+		c, err := auth.GetNewCredentialsWithMetadata(map[string]any{"exp": UTCNow().Add(time.Hour).Unix(), parentClaim: "sts-partial-parent"}, secret)
+		mustIAM(t, err)
+		c.ParentUser = "sts-partial-parent"
+		_, err = sys.SetTempUser(ctx, c.AccessKey, c, "")
+		mustIAM(t, err)
+		keys = append(keys, c.AccessKey)
+	}
+	sys.store.IAMStorageAPI = &iamSTSFailureStore{IAMStorageAPI: sys.store.IAMStorageAPI, failedPath: getUserIdentityPath(keys[0], stsUser)}
+	revoked, err := sys.store.RevokeTokens(ctx, "sts-partial-parent", "")
+	if !errors.Is(err, errIAMInjectedWrite) || len(revoked) != 1 || revoked[0] != keys[1] {
+		t.Fatalf("committed keys or failure lost: %v %v", revoked, err)
+	}
+	if _, ok := sys.store.GetUser(keys[1]); ok {
+		t.Fatal("committed revocation retained cache entry")
+	}
+	if _, ok := sys.store.GetUser(keys[0]); !ok {
+		t.Fatal("failed revocation claimed cache success")
+	}
+}

cmd/iam.go

@@ -717,7 +717,14 @@ func (sys *IAMSys) RevokeTokens(ctx context.Context, accessKey, tokenRevokeType
 		return errServerNotInitialized
 	}
 
-	return sys.store.RevokeTokens(ctx, accessKey, tokenRevokeType)
+	keys, err := sys.store.RevokeTokens(ctx, accessKey, tokenRevokeType)
+	// Publish every committed tombstone even if another token failed to save.
+	for _, key := range keys {
+		sys.notifyForUser(ctx, key, true)
+		// Offline peers recover this same record through IAM deletion healing.
+		iamLogIf(ctx, globalSiteReplicationSys.IAMChangeHook(ctx, madmin.SRIAMItem{Type: iamSTSDeleteType, Name: key}))
+	}
+	return err
 }
 
 // DeleteUser - delete user (only for long-term users not STS users).

cmd/list-multipart-uploads-compat_test.go

@@ -40,7 +40,7 @@ func requireMultipartUploadKeys(t *testing.T, got ListMultipartsInfo, want ...st
 	}
 }
 
-func TestListMultipartUploadsS3Compatibility(t *testing.T) {
+func TestMultipartDiagnosticScanPagination(t *testing.T) {
 	obj, dirs, err := prepareErasureSets32(t.Context())
 	if err != nil {
 		t.Fatal(err)
@@ -81,14 +81,14 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
 		uploadIDs[object] = mp.UploadID
 	}
 
-	// Durable multipart metadata, rather than this node-local cache, must be
-	// authoritative after a restart or when another node handles the request.
+	// The diagnostic scanner must discover metadata independently of the cache.
+	// These checks do not claim that public listing offers durable pagination.
 	z.mpCache.Range(func(uploadID string, _ MultipartInfo) bool {
 		z.mpCache.Delete(uploadID)
 		return true
 	})
 
-	all, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
+	all, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 100)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -97,7 +97,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
 		t.Fatalf("complete listing has truncation state: %+v", all)
 	}
 
-	first, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 1)
+	first, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 1)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -108,31 +108,31 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
 			objects[0], uploadIDs[objects[0]])
 	}
 
-	rest, err := z.ListMultipartUploads(t.Context(), bucket, "", first.NextKeyMarker, first.NextUploadIDMarker, "", 100)
+	rest, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", first.NextKeyMarker, first.NextUploadIDMarker, "", 100)
 	if err != nil {
 		t.Fatal(err)
 	}
 	requireMultipartUploadKeys(t, rest, objects[1:]...)
 
-	afterKey, err := z.ListMultipartUploads(t.Context(), bucket, "", objects[1], "", "", 100)
+	afterKey, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", objects[1], "", "", 100)
 	if err != nil {
 		t.Fatal(err)
 	}
 	requireMultipartUploadKeys(t, afterKey, objects[2:]...)
 
-	prefixed, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", "", 100)
+	prefixed, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", "", "", "", 100)
 	if err != nil {
 		t.Fatal(err)
 	}
 	requireMultipartUploadKeys(t, prefixed, objects[:3]...)
 
-	nested, err := z.ListMultipartUploads(t.Context(), bucket, "t/a_b/", "", "", "", 100)
+	nested, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/a_b/", "", "", "", 100)
 	if err != nil {
 		t.Fatal(err)
 	}
 	requireMultipartUploadKeys(t, nested, objects[:2]...)
 
-	grouped, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", SlashSeparator, 100)
+	grouped, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", "", "", SlashSeparator, 100)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -141,7 +141,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
 		t.Fatalf("common prefixes = %v, want %v", grouped.CommonPrefixes, want)
 	}
 
-	groupPage, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", SlashSeparator, 1)
+	groupPage, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", "", "", SlashSeparator, 1)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -153,7 +153,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
 			groupPage.NextKeyMarker, groupPage.NextUploadIDMarker, groupPage.IsTruncated)
 	}
 
-	groupRest, err := z.ListMultipartUploads(t.Context(), bucket, "t/", groupPage.NextKeyMarker, "", SlashSeparator, 1)
+	groupRest, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", groupPage.NextKeyMarker, "", SlashSeparator, 1)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -197,12 +197,12 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
 		er.getUploadIDDir(bucket, legacyObject, uploadIDs[legacyObject]), metadata, fi.WriteQuorum(er.defaultWQuorum())); err != nil {
 		t.Fatal(err)
 	}
-	legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100)
+	legacy, err := z.scanMultipartUploadsForTest(t.Context(), bucket, legacyObject, "", "", "", 100)
 	if err != nil {
 		t.Fatal(err)
 	}
 	requireMultipartUploadKeys(t, legacy)
-	remaining, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
+	remaining, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 100)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -280,7 +280,7 @@ func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
 		return true
 	})
 
-	page, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 2)
+	page, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 2)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -289,7 +289,7 @@ func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
 		t.Fatalf("first global page = %+v", page)
 	}
 
-	rest, err := z.ListMultipartUploads(t.Context(), bucket, "", page.NextKeyMarker, page.NextUploadIDMarker, "", 2)
+	rest, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", page.NextKeyMarker, page.NextUploadIDMarker, "", 2)
 	if err != nil {
 		t.Fatal(err)
 	}

cmd/metacache-entries.go

@@ -781,6 +781,15 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
 	}
 	last := ""
 	var toMerge []int
+	var directoryCollisions []int
+	discardDirectories := func() error {
+		for _, idx := range directoryCollisions {
+			if err := selectFrom(idx); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
 
 	// Choose the best to return.
 	for {
@@ -790,6 +799,7 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
 		best := top[0]
 		bestIdx := 0
 		toMerge = toMerge[:0]
+		directoryCollisions = directoryCollisions[:0]
 		for i, other := range top[1:] {
 			otherIdx := i + 1
 			if other == nil {
@@ -819,10 +829,9 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
 						if serverDebugLog {
 							console.Debugln("mergeEntryChannels: discarding directory", other.name, "for object", best.name)
 						}
-						// Discard the directory.
-						if err := selectFrom(otherIdx); err != nil {
-							return err
-						}
+						// A pending object can resolve to absence. Keep the
+						// directory until the whole object group is confirmed.
+						directoryCollisions = append(directoryCollisions, otherIdx)
 						continue
 					}
 					// Replace directory with object.
@@ -838,6 +847,7 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
 			}
 			if best.name > other.name {
 				toMerge = toMerge[:0]
+				directoryCollisions = directoryCollisions[:0]
 				best = other
 				bestIdx = otherIdx
 			}
@@ -866,6 +876,11 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
 			if err != nil {
 				return err
 			}
+			if resolved != nil {
+				if err := discardDirectories(); err != nil {
+					return err
+				}
+			}
 			if resolved != nil && resolved.name > last {
 				select {
 				case <-ctxDone:
@@ -876,6 +891,9 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
 			}
 			continue
 		}
+		if err := discardDirectories(); err != nil {
+			return err
+		}
 
 		// Merge any unmerged
 		if len(toMerge) > 0 {

cmd/metacache-listing-directory_test.go

@@ -0,0 +1,243 @@
+// Copyright (c) 2026 Ruohang Feng
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"encoding/xml"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"testing"
+	"time"
+)
+
+type listingDirectoryWalkSnapshot struct {
+	StorageAPI
+	bucket   string
+	snapshot []byte
+}
+
+func (d listingDirectoryWalkSnapshot) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) error {
+	if opts.Bucket != d.bucket {
+		return d.StorageAPI.WalkDir(ctx, opts, out)
+	}
+	_, err := out.Write(d.snapshot)
+	return err
+}
+
+// Capture two actual directory-walk generations for a directory object, then
+// acknowledge its deletion. A different pool still contains dir/child. The
+// stale source walkers disagree, while every current source drive is absent.
+func TestListingRecoveryDeletedDirectoryKeepsCommonPrefix(t *testing.T) {
+	for _, marker := range []bool{false, true} {
+		for _, sourcePool := range []int{0, 1} {
+			t.Run(fmt.Sprintf("pool=%d/marker=%t", sourcePool, marker), func(t *testing.T) {
+				z, bucket, router := listingPoolsBackend(t)
+				ctx := t.Context()
+				source := z.serverPools[sourcePool].sets[0]
+				target := z.serverPools[1-sourcePool].sets[0]
+				childBody := []byte("surviving child")
+				put := func(set *erasureObjects, key string, tm time.Time) {
+					var body []byte
+					if key == "dir/child" {
+						body = childBody
+					}
+					_, err := set.PutObject(ctx, bucket, encodeDirObject(key), mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm, VersionSuspended: marker})
+					if err != nil {
+						t.Fatal(err)
+					}
+				}
+				tm := time.Now().UTC().Add(-time.Hour)
+				put(target, "dir/child", tm)
+				disks, switchDisks := installListingDiskSwitch(source)
+				snapshots := make([][]byte, len(disks))
+				for gen := range 3 {
+					if gen < 2 {
+						put(source, "dir/", tm.Add(time.Duration(gen)*time.Minute))
+					} else {
+						if _, err := source.DeleteObject(ctx, bucket, encodeDirObject("dir/"), ObjectOptions{VersionSuspended: marker, DeleteMarker: marker, MTime: tm.Add(2 * time.Minute)}); err != nil {
+							t.Fatal(err)
+						}
+					}
+					for i, disk := range disks {
+						wanted := 0
+						if i >= 6 {
+							wanted = 1
+						}
+						if i >= 12 {
+							wanted = 2
+						}
+						if wanted != gen {
+							continue
+						}
+						var b bytes.Buffer
+						if err := disk.WalkDir(ctx, WalkDirOptions{Bucket: bucket, Recursive: false}, &b); err != nil {
+							t.Fatal(err)
+						}
+						snapshots[i] = append([]byte(nil), b.Bytes()...)
+					}
+				}
+				for _, key := range []string{"dir/child", "dir/"} {
+					request, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucket, key), 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+					if err != nil {
+						t.Fatal(err)
+					}
+					response := httptest.NewRecorder()
+					router.ServeHTTP(response, request.WithContext(ctx))
+					t.Logf("GET %s HTTP %d", key, response.Code)
+					want := http.StatusOK
+					if key == "dir/" {
+						want = http.StatusNotFound
+					}
+					if response.Code != want {
+						t.Fatalf("GET %s: got %d want %d: %s", key, response.Code, want, response.Body.String())
+					}
+					if key == "dir/child" && !bytes.Equal(response.Body.Bytes(), childBody) {
+						t.Fatal("surviving child content changed")
+					}
+				}
+				wrapped := make([]StorageAPI, len(disks))
+				for i, disk := range disks {
+					wrapped[i] = listingDirectoryWalkSnapshot{disk, bucket, snapshots[i]}
+				}
+				switchDisks(wrapped)
+				t.Cleanup(func() { switchDisks(disks) })
+				req, err := newTestSignedRequestV4(http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", "/"), 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+				if err != nil {
+					t.Fatal(err)
+				}
+				rec := httptest.NewRecorder()
+				router.ServeHTTP(rec, req.WithContext(ctx))
+				t.Logf("LIST HTTP %d body=%s", rec.Code, rec.Body.String())
+				var page ListObjectsV2Response
+				if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil {
+					t.Fatalf("LIST failed: %d %s", rec.Code, rec.Body.String())
+				}
+				if len(page.CommonPrefixes) != 1 || page.CommonPrefixes[0].Prefix != "dir/" {
+					t.Fatalf("acknowledged child lost its CommonPrefix: %+v", page)
+				}
+			})
+		}
+	}
+}
+
+// The directory must remain in the sorted input when an earlier object name
+// disappears. Emitting it immediately would skip an intervening name, and
+// keeping only one colliding directory would leave input order significant.
+func TestListingRecoveryDirectoryCollisionOrdering(t *testing.T) {
+	for _, object := range []string{"dir", "dir/"} {
+		for _, present := range []bool{false, true} {
+			if present && object == "dir" {
+				continue
+			}
+			for source := range 3 {
+				t.Run(fmt.Sprintf("object=%s/present=%t/source=%d", object, present, source), func(t *testing.T) {
+					in := make([]chan metaCacheEntry, 6)
+					for i := range in {
+						in[i] = make(chan metaCacheEntry, 1)
+						entry := metaCacheEntry{name: "dir/"}
+						if i == source {
+							entry = metaCacheEntry{name: object, listingSet: &erasureObjects{}}
+						}
+						if i >= 3 {
+							entry = metaCacheEntry{name: []string{"dir-", "dir.ext", "dir0"}[i-3], metadata: []byte{1}}
+						}
+						in[i] <- entry
+						close(in[i])
+					}
+					out := make(chan metaCacheEntry, len(in))
+					err := mergeEntryChannelsWithResolver(t.Context(), in, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
+						if len(group) != 1 || group[0].listingSet == nil {
+							t.Fatalf("directory entered the object group: %+v", group)
+						}
+						if !present {
+							return nil, nil
+						}
+						return &metaCacheEntry{name: object, metadata: []byte{1}}, nil
+					})
+					if err != nil {
+						t.Fatal(err)
+					}
+					var names []string
+					for entry := range out {
+						names = append(names, entry.name)
+						if entry.name == "dir/" && entry.isObject() != present {
+							t.Fatalf("directory collision returned the wrong entry type: %+v", entry)
+						}
+					}
+					want := []string{"dir-", "dir.ext", "dir/", "dir0"}
+					if !slices.Equal(names, want) {
+						t.Fatalf("names=%v, want %v", names, want)
+					}
+				})
+			}
+		}
+	}
+}
+
+// A regular walk result may be stale when a pending peer forces current
+// recovery. Keep sibling prefixes until the complete object group is resolved.
+func TestListingRecoveryMixedDirectoryCollision(t *testing.T) {
+	orders := [][3]int{{0, 1, 2}, {0, 2, 1}, {1, 0, 2}, {1, 2, 0}, {2, 0, 1}, {2, 1, 0}}
+	for _, order := range orders {
+		for _, present := range []bool{false, true} {
+			t.Run(fmt.Sprintf("order=%v/present=%t", order, present), func(t *testing.T) {
+				entries := []metaCacheEntry{
+					{name: "dir/", metadata: []byte{1}},
+					{name: "dir/"},
+					{name: "dir/", listingSet: &erasureObjects{}},
+				}
+				in := make([]chan metaCacheEntry, len(order))
+				for i, index := range order {
+					in[i] = make(chan metaCacheEntry, 2)
+					in[i] <- entries[index]
+					in[i] <- metaCacheEntry{name: fmt.Sprintf("tail-%d", i), metadata: []byte{1}}
+					close(in[i])
+				}
+				out := make(chan metaCacheEntry, 6)
+				calls := 0
+				err := mergeEntryChannelsWithResolver(t.Context(), in, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
+					calls++
+					if len(group) != 2 {
+						t.Fatalf("incomplete object group: %+v", group)
+					}
+					pending := 0
+					for _, entry := range group {
+						if entry.name != "dir/" || entry.isDir() {
+							t.Fatalf("unexpected recovery entry: %+v", entry)
+						}
+						if entry.listingSet != nil {
+							pending++
+						}
+					}
+					if pending != 1 {
+						t.Fatalf("pending entries=%d, want 1", pending)
+					}
+					if !present {
+						return nil, nil
+					}
+					return &metaCacheEntry{name: "dir/", metadata: []byte{1}}, nil
+				})
+				if err != nil || calls != 1 {
+					t.Fatalf("err=%v recovery calls=%d", err, calls)
+				}
+				var names []string
+				for entry := range out {
+					names = append(names, entry.name)
+					if entry.listingSet != nil || entry.name == "dir/" && entry.isObject() != present {
+						t.Fatalf("unexpected merged entry: %+v", entry)
+					}
+				}
+				want := []string{"dir/", "tail-0", "tail-1", "tail-2"}
+				if !slices.Equal(names, want) {
+					t.Fatalf("names=%v, want %v", names, want)
+				}
+			})
+		}
+	}
+}

cmd/metacache-listing-recovery_test.go

@@ -221,35 +221,52 @@ func TestListingRecoveryNormalPathDoesNotReadOrLock(t *testing.T) {
 
 func TestListingPendingDirectoryCollision(t *testing.T) {
 	for _, objectFirst := range []bool{false, true} {
-		t.Run(fmt.Sprint(objectFirst), func(t *testing.T) {
-			in := []chan metaCacheEntry{make(chan metaCacheEntry, 1), make(chan metaCacheEntry, 1)}
+		for _, present := range []bool{true, false} {
+			t.Run(fmt.Sprintf("objectFirst=%v/present=%v", objectFirst, present), func(t *testing.T) {
+				in := []chan metaCacheEntry{make(chan metaCacheEntry, 2), make(chan metaCacheEntry, 2), make(chan metaCacheEntry, 2)}
 				objectIndex := 1
 				if objectFirst {
 					objectIndex = 0
 				}
 				in[objectIndex] <- metaCacheEntry{name: "dir/", listingSet: &erasureObjects{}}
 				in[1-objectIndex] <- metaCacheEntry{name: "dir/"}
+				in[2] <- metaCacheEntry{name: "a", metadata: []byte{1}}
 				for _, c := range in {
 					close(c)
 				}
-			out := make(chan metaCacheEntry, 2)
+				out := make(chan metaCacheEntry, 4)
 				calls := 0
 				err := mergeEntryChannelsWithResolver(t.Context(), in, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
 					calls++
 					if len(group) != 1 || group[0].listingSet == nil {
 						t.Fatalf("wrong pending group: %+v", group)
 					}
+					if !present {
+						return nil, nil
+					}
 					return &metaCacheEntry{name: "dir/", metadata: []byte{1}}, nil
 				})
-			if err != nil || calls != 1 || len(out) != 1 {
+				if err != nil || calls != 1 || len(out) != 2 {
 					t.Fatalf("err=%v recoveries=%d outputs=%d", err, calls, len(out))
 				}
-			if entry := <-out; entry.isDir() || !entry.isObjectDir() {
+				first := <-out
+				if first.name != "a" {
+					t.Fatalf("expected smaller key first, got %+v", first)
+				}
+				entry := <-out
+				if entry.name != "dir/" || entry.listingSet != nil {
+					t.Fatalf("unexpected recovered entry: %+v", entry)
+				}
+				if present && (entry.isDir() || !entry.isObjectDir()) {
 					t.Fatalf("object became directory: %+v", entry)
 				}
+				if !present && (!entry.isDir() || entry.name != "dir/") {
+					t.Fatalf("absent pending object dropped live directory prefix: %+v", entry)
+				}
 			})
 		}
 	}
+}
 
 func TestListingRecoveryRejectsUnexpectedSinglePoolSet(t *testing.T) {
 	sets := []*erasureObjects{{}, {}}

cmd/object-api-errors.go

@@ -834,6 +834,9 @@ func (de DataMovementOverwriteErr) Error() string {
 	if de.VersionID != "" {
 		objInfoStr = fmt.Sprintf("%s version-id=%s", objInfoStr, de.VersionID)
 	}
+	if de.Err == errDataMovementSuperseded {
+		return fmt.Sprintf("data movement skipped, another pool holds a newer copy of %s", objInfoStr)
+	}
 	return fmt.Sprintf("invalid data movement operation, source and destination pool are the same for %s", objInfoStr)
 }
 
@@ -841,3 +844,11 @@ func isDataMovementOverWriteErr(err error) bool {
 	var de DataMovementOverwriteErr
 	return errors.As(err, &de)
 }
+
+// isDataMovementSuperseded reports a version that was not moved because
+// another pool already holds a newer copy, which reads return. The source copy
+// is a stale duplicate: the move is complete, and the source may be removed.
+func isDataMovementSuperseded(err error) bool {
+	var de DataMovementOverwriteErr
+	return errors.As(err, &de) && de.Err == errDataMovementSuperseded
+}

docs/debugging/reorder-disks/go.mod

@@ -2,4 +2,4 @@ module github.com/minio/minio/docs/debugging/reorder-disks
 
 go 1.26.0
 
-require github.com/pgsty/silo-pkg/v3 v3.13.2
+require github.com/pgsty/silo-pkg/v3 v3.14.3

docs/debugging/reorder-disks/go.sum

@@ -1,2 +1,2 @@
-github.com/pgsty/silo-pkg/v3 v3.13.2 h1:Clw11c/J54Tx6pijNCWtXiC7e0fwP/f5Tgeb6fsXg2w=
-github.com/pgsty/silo-pkg/v3 v3.13.2/go.mod h1:0GmaDA0ArQ8bkAI/obiSNTBQzdgBu6W0e0olDCbyXXo=
+github.com/pgsty/silo-pkg/v3 v3.14.3 h1:hMDxOo36F1whGRkYScrUGGwn98FvsmJ8hVM6pE4i8Bs=
+github.com/pgsty/silo-pkg/v3 v3.14.3/go.mod h1:zqIm3FMETmwPHRKuUExCDGfYkCNXt+vpfBnPM821RdI=

go.mod

@@ -81,7 +81,7 @@ require (
 	github.com/nats-io/stan.go v0.10.4
 	github.com/ncw/directio v1.0.5
 	github.com/nsqio/go-nsq v1.1.0
-	github.com/pgsty/silo-pkg/v3 v3.14.2
+	github.com/pgsty/silo-pkg/v3 v3.14.3
 	github.com/philhofer/fwd v1.2.1
 	github.com/pierrec/lz4/v4 v4.1.30
 	github.com/pkg/errors v0.9.1

go.sum

@@ -549,8 +549,8 @@ github.com/pgsty/mc v0.0.0-20261017101407-fa3b8bf72458 h1:f3kbkjtjj6D7PXmsIeKrv8
 github.com/pgsty/mc v0.0.0-20261017101407-fa3b8bf72458/go.mod h1:n8NDi0J3hPUGCIM9tOxlhYXkYS9xiTVIxlxa3Rwx1cw=
 github.com/pgsty/silo-console v0.0.0-20261017112417-ef2dd2c41d21 h1:PHxPswGiEwU9MVisu0/Ab7AcNX4IveiX1thQzUXOsM4=
 github.com/pgsty/silo-console v0.0.0-20261017112417-ef2dd2c41d21/go.mod h1:HFNaafu5iCO9bWS86PIyTbgF+yPD4phFt+gxML/1jJU=
-github.com/pgsty/silo-pkg/v3 v3.14.2 h1:nU9gQ9LuDhN/Z6gZNvFj94Vb9xrjEMKAegX16v55s/I=
-github.com/pgsty/silo-pkg/v3 v3.14.2/go.mod h1:zqIm3FMETmwPHRKuUExCDGfYkCNXt+vpfBnPM821RdI=
+github.com/pgsty/silo-pkg/v3 v3.14.3 h1:hMDxOo36F1whGRkYScrUGGwn98FvsmJ8hVM6pE4i8Bs=
+github.com/pgsty/silo-pkg/v3 v3.14.3/go.mod h1:zqIm3FMETmwPHRKuUExCDGfYkCNXt+vpfBnPM821RdI=
 github.com/philhofer/fwd v1.2.1 h1:UtpS73q8bFgy/yH3gCq8LSNvj5gv/FowNOiWZa7uKwQ=
 github.com/philhofer/fwd v1.2.1/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
 github.com/pierrec/lz4/v4 v4.1.30 h1:cchX8N2DVP668WkElI9QMwVyoNabLkq1LofDHFeIrdg=