Enforce Object Lock during S3 bulk version deletion

Signed-off-by: Feng Ruohang <rh@vonng.com>
docs(healthcheck): say what ready does not wait for
2026-11-10 01:31:52 +03:00 · 2026-11-09 23:53:43 +08:00 · 2026-11-09 22:07:59 +08:00 · 2026-11-09 22:07:59 +08:00 · 2026-11-09 21:26:02 +08:00 · 2026-11-09 21:05:22 +08:00
49 changed files with 3251 additions and 401 deletions
@@ -78,7 +78,7 @@ jobs:
  race-s3select:
    name: Race, S3 Select
    runs-on: ubuntu-latest
-    timeout-minutes: 15
+    timeout-minutes: 25
    steps:
      - uses: actions/checkout@v7

@@ -105,7 +105,7 @@ jobs:
      - name: Run lifecycle expiry and native restore tests under race detector
        env:
          MINIO_API_REQUESTS_MAX: "10000"
-        run: go test -race -tags kqueue,dev ./cmd -run '^Test(PoolsLifecycle|PoolsMigrationLifecycle|ExpireRestored|NativeRestore)' -count=1 -timeout=5m
+        run: go test -race -tags kqueue,dev ./cmd -run '^Test(PoolsLifecycle|PoolsMigrationLifecycle|ExpireRestored|NativeRestore)' -count=1 -timeout=10m

      - name: Run multipart listing, cancellation and cleanup tests under race detector
        env:
@@ -127,7 +127,7 @@ jobs:
      - name: Run Object Lock authorization and overwrite tests under race detector
        env:
          MINIO_API_REQUESTS_MAX: "10000"
-        run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m
+        run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|ReplicaDeleteObjectLock|PoolsReplicaDeleteObjectLock|BulkDeleteDirectoryObjectLock|BulkDeleteObjectLock|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m

      - name: Run maintenance storage timeout and compatibility tests under race detector
        env:
@@ -74,7 +74,7 @@ jobs:
      - name: Gate release on Object Lock authorization and overwrite tests under race detector
        env:
          MINIO_API_REQUESTS_MAX: "10000"
-        run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m
+        run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|ReplicaDeleteObjectLock|PoolsReplicaDeleteObjectLock|BulkDeleteDirectoryObjectLock|BulkDeleteObjectLock|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m

      - name: Compute release variables
        env:
@@ -1,5 +1,174 @@
 # Changelog

+## RELEASE.2026-11-09T00-00-00Z
+
+Package version: `20261109000000.0.0`. Publication status and artifacts are
+listed in [GitHub Releases](https://github.com/pgsty/silo/releases). This
+section covers the changes after the published 20261026 source `c3ad836aed0d`.
+Most fixes come from reports and patches by
+[@botdeniska](https://github.com/botdeniska); see [CONTRIBUTORS.md](CONTRIBUTORS.md).
+
+Lifecycle, multi-pool and tiering:
+
+- Decide an ILM expiry under the object lock against the versions of the key
+  in all pools, and refuse one that does not hold for the whole object or whose
+  history or lock protection lacks an authoritative quorum. Decommission and
+  rebalance move such versions instead of dropping them, and an approved
+  version expiry deletes every copy ([#278](https://github.com/pgsty/silo/issues/278)).
+- Expire the restored copy of a current version instead of the object
+  ([#281](https://github.com/pgsty/silo/issues/281)). Keep lifecycle delete-all
+  (`ExpiredObjectAllVersions`, `DelMarkerExpiration`, batch expiration with
+  `retainVersions: 0`) on the expired key instead of the keys below it; a key
+  with a tiered or Object Lock protected version is refused
+  ([#297](https://github.com/pgsty/silo/issues/297)).
+- Stop taking the free version of a deleted tiered version for a version
+  ([#263](https://github.com/pgsty/silo/issues/263)), apply rules whose only
+  action is `DelMarkerExpiration` ([#268](https://github.com/pgsty/silo/issues/268)),
+  and report and account only the removals that happened
+  ([#305](https://github.com/pgsty/silo/issues/305)).
+- Keep remote tier data that another pool still references after a version
+  expiry or tier cleanup ([#306](https://github.com/pgsty/silo/issues/306),
+  [#308](https://github.com/pgsty/silo/issues/308)). Heal versions whose queued
+  expiry is refused ([#303](https://github.com/pgsty/silo/issues/303)), and read
+  a key's history once in migration prechecks ([#304](https://github.com/pgsty/silo/issues/304)).
+- Decommission and rebalance: move objects whose copies have two erasure
+  layouts ([#261](https://github.com/pgsty/silo/issues/261)) and ILM-expired
+  current versions ([#265](https://github.com/pgsty/silo/issues/265)), continue
+  with later buckets when one cannot finish and keep cancellation
+  ([#307](https://github.com/pgsty/silo/issues/307)), and do not panic when a
+  rebalance stop finds no metadata ([#270](https://github.com/pgsty/silo/issues/270)).
+- Do not replicate the delete markers and version purges that lifecycle
+  creates, as in S3. [PR #240](https://github.com/pgsty/silo/pull/240) remains
+  open as a proposal.
+
+Native restore:
+
+- Restore in two phases, without holding the object write lock during the tier
+  download ([#284](https://github.com/pgsty/silo/issues/284)). Restores with more
+  than one pool no longer hang, stay in progress or change the restored version
+  ([#289](https://github.com/pgsty/silo/issues/289)), and a failed ordinary
+  multipart completion stays usable ([#327](https://github.com/pgsty/silo/issues/327)).
+  A failed or canceled restore returns at once and clears its marker in the
+  background, instead of waiting up to a minute for the object lock.
+
+Locking, conditional writes and storage:
+
+- Lock every write and delete of a single pool in the object's own erasure set
+  ([#298](https://github.com/pgsty/silo/issues/298)). Fix CopyObject waiting on
+  its own lock for normalized aliases ([#299](https://github.com/pgsty/silo/issues/299))
+  and starving its destination while waiting for the source
+  ([#300](https://github.com/pgsty/silo/issues/300)); delete the set groups of a
+  bulk delete concurrently ([#301](https://github.com/pgsty/silo/issues/301)).
+- Evaluate `If-Match` and `If-None-Match` of a multi-pool CreateMultipartUpload
+  against the current object ([#291](https://github.com/pgsty/silo/issues/291),
+  [PR #292](https://github.com/pgsty/silo/pull/292)).
+- Answer a retryable `503` instead of `NoSuchKey` when a node sees only some
+  drives of an existing object ([#302](https://github.com/pgsty/silo/issues/302)), remove the empty object
+  directory of a failed write ([#285](https://github.com/pgsty/silo/issues/285),
+  [PR #286](https://github.com/pgsty/silo/pull/286)), classify `xl.meta`
+  checksum failures for healing ([#309](https://github.com/pgsty/silo/issues/309)),
+  and keep a heal dry run from deleting data
+  ([PR #280](https://github.com/pgsty/silo/pull/280), thanks to
+  [@dihmandrake](https://github.com/dihmandrake)).
+
+Object Lock, IAM and batch jobs:
+
+- Keep the target's legal hold and retention on a replicated delete; a source
+  purge that bypassed GOVERNANCE retention is refused by a target copy that
+  still carries it, and the source retries. Check a directory object's lock
+  before a bulk delete removes its null version. S3 `DeleteObjects` checks
+  physical version removals under the mutation lock and merges protection
+  across pools, including concurrent legal-hold and retention updates. These
+  removals in Object Lock buckets use the single-object path sequentially;
+  large version-purge batches can take longer. Marker creation and ordinary
+  buckets retain the bulk path.
+- Batch jobs enforce the submitter's data permissions throughout execution and
+  leave Object Lock protected versions alone ([#310](https://github.com/pgsty/silo/issues/310),
+  [#311](https://github.com/pgsty/silo/issues/311)). Key rotation keeps other
+  pool copies readable ([#312](https://github.com/pgsty/silo/issues/312)).
+  Transient failures pause work instead of ending it, saved jobs are found
+  after a failed startup scan, and running replication resumes
+  ([#313](https://github.com/pgsty/silo/issues/313), [#314](https://github.com/pgsty/silo/issues/314),
+  [#324](https://github.com/pgsty/silo/issues/324)). A completed expiration job
+  stays completed after a restart ([#315](https://github.com/pgsty/silo/issues/315)).
+  Authorization skips deleted policies a mapping still names, refuses a job
+  only for a Deny that needs request context, and survives metadata edits and
+  LDAP refreshes that keep the credential
+  ([#318](https://github.com/pgsty/silo/issues/318)–[#320](https://github.com/pgsty/silo/issues/320)).
+  Live metrics keep pause reasons and skipped or refused counts
+  ([#325](https://github.com/pgsty/silo/issues/325)), and `make check-gen`
+  leaves the runtime policy snapshot alone ([#326](https://github.com/pgsty/silo/issues/326)).
+- IAM refuses incomplete policy loads, keeps cached authentication available
+  during lazy loads, and applies edits, LDAP group refreshes and service-account
+  creation to the stored record, so a stale node cannot restore old credentials
+  or resurrect deleted identities ([#316](https://github.com/pgsty/silo/issues/316),
+  [#317](https://github.com/pgsty/silo/issues/317), [#321](https://github.com/pgsty/silo/issues/321)–[#323](https://github.com/pgsty/silo/issues/323)).
+- Log in LDAP users with long DNs on erasure sets, and root to the embedded
+  Console with LDAP enabled ([#271](https://github.com/pgsty/silo/issues/271),
+  [#272](https://github.com/pgsty/silo/issues/272)). Keep audit and logger
+  webhook targets running on every server after a configuration reload
+  ([#259](https://github.com/pgsty/silo/issues/259), reported by
+  [@haf](https://github.com/haf)).
+
+Components and packaging:
+
+- Embed the published Console v2.7.0 source `4e91ae506e70`, including its
+  ten-language UI, rebuilt frontend assets, Swagger array-validation fixes,
+  Redoc routing, plain-HTTP folder-size and key-copy actions, usage-worker TLS
+  roots and scan budgets, and malformed-multipart rejection. Select the
+  published mcli 20261109 source `e1d4cc5abfe0` and silo-pkg v3.16.0 directly
+  in the maintained dependency graph.
+- Adopt go-jose/v4 v4.1.9, cpuid/v2 v2.4.2, YAML v3.0.6 and gjson v1.20.1,
+  plus Console's runtime/server-middleware v0.33.4. Preserve Go 1.27.3, the
+  verified upstream minio-go commit, networking/crypto security floors and
+  terminal-width/NetBSD compatibility selections.
+- Update NATS credentials to jwt/v2 v2.8.3 and nkeys v0.4.17, including
+  rejection of identity-point and small-order public keys. Use nats-server
+  v2.15.2 for notification integration tests.
+- Update Google Cloud auth to v0.24.3 and oauth2adapt to v0.3.1 for REST
+  request validation and AWS FIPS endpoint signing fixes.
+- Bundle verified mcli 20261109 archives in source container builds and use
+  that release by default in the integration-test client installer.
+- Prevent replication worker metrics from panicking when the replication
+  pool is absent, while retaining historical averages and peak counts.
+- Give the distroless image the same CA roots as the standard image, and ship
+  mcli's license files in both ([#295](https://github.com/pgsty/silo/pull/295)).
+- Deploy SILO and mcli `RELEASE.2026-11-09T00-00-00Z` by default from Helm
+  chart 7.0.4; the server image tag was left at 20260916 before
+  ([#296](https://github.com/pgsty/silo/issues/296)).
+
+Upgrade notes:
+
+- A single pool whose erasure sets do not all use the same hosts (with the
+  automatic layout, 18 or more hosts) needs a drained restart, not a rolling
+  upgrade: old and new nodes lock deletes and copies in different sets and do
+  not exclude each other. Stop every old process before a new one serves.
+- During a rolling upgrade, upgraded nodes answer `POST ?restore` with `503`
+  while any peer is old or unreachable. Pause new restores until every node is
+  upgraded. Multi-pool lifecycle protection is complete only after a full
+  upgrade.
+- Stop batch submissions and drain or cancel unfinished jobs before changing
+  versions. Unfinished legacy jobs require resubmission after all nodes upgrade;
+  completed jobs stay completed. Submitter-bound manifests and progress reports
+  use a new format that 20261026 cannot execute or resume. Do not downgrade
+  with pending or paused new jobs. An old node can still execute old-format
+  jobs, so mixed-version execution is not protected by the new authorization
+  rules. Preserve job definitions and reconcile completed work before resubmitting.
+- `healthcheck ready` remains a process/dependency probe and can succeed before
+  storage or IAM load. `cluster` reports quorum, which can precede a restarted
+  node reconnecting its remote drives. Before resuming applications, run bounded
+  signed write/read checks through every node; neither HTTP probe alone proves
+  data-path readiness. Preserve peer discovery during bootstrap.
+- `helm upgrade --reuse-values` keeps the previously stored `image.tag`; set it
+  explicitly when upgrading such a release.
+
+Known issues: `ListMultipartUploads` prefix and paging
+([#79](https://github.com/pgsty/silo/issues/79)), multi-site delete-marker purge
+([#217](https://github.com/pgsty/silo/issues/217)), replayed bucket creation
+under site replication ([PR #241](https://github.com/pgsty/silo/pull/241)), and
+`403` instead of the server error for reads allowed only by object tags while
+the object is unreadable ([#293](https://github.com/pgsty/silo/issues/293)).
+
 ## RELEASE.2026-10-26T00-00-00Z

 Package version: `20261026000000.0.0`; published from source `c3ad836aed0d`.
@@ -17,9 +17,9 @@ ENV GOPATH=/go
 ENV CGO_ENABLED=0

 ARG MC_REPO=pgsty/mc
-ARG MC_VERSION=RELEASE.2026-10-26T00-00-00Z
-ARG MC_AMD64_SHA256=e86295bb11ed59369c0560f030d59abd42fb64a376b1d7e9799ec2e8596ebf64
-ARG MC_ARM64_SHA256=47b1a558f766f26bef5beeedd5300e082bcdec07a5e7d4a16f5f9d168f5275d1
+ARG MC_VERSION=RELEASE.2026-11-09T00-00-00Z
+ARG MC_AMD64_SHA256=79e6cf4ad00a9e5eb2e7ceccc6e6496e4af4ec096e59695515ea8f223c287e4a
+ARG MC_ARM64_SHA256=5473c4b39a31dd66478f5c44686a09f3e9e37861a30f1faf18564fbe66118387

 RUN apk add -U --no-cache \
    ca-certificates \
@@ -32,29 +32,26 @@
 > [!IMPORTANT]
 > **PGSTY Silo** (hereinafter “Silo”) is an independent, community-maintained fork of the open-source MinIO server, published by [Pigsty](https://pigsty.io) from [`pgsty/silo`](https://github.com/pgsty/silo). It is not affiliated with, endorsed by, or sponsored by MinIO, Inc. “MinIO” is used only to identify the upstream project and compatibility lineage.

-> [!NOTE]
-> Renamed from `pgsty/minio` to `pgsty/silo`, default branch `master` → `main`, on 2026-08-06. Artifacts under the original MinIO identity stay published on the archived [`minio`](https://github.com/pgsty/silo/tree/minio) branch and in releases up to [`RELEASE.2026-08-04T00-00-00Z`](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-08-04T00-00-00Z).
-
-## SILO 20261026000000
-
-[SILO 20261026000000](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-26T00-00-00Z) is available as platform archives, Linux packages, and standard or distroless Docker images for amd64 and arm64. The release tag is `RELEASE.2026-10-26T00-00-00Z`; the package version is `20261026000000.0.0`, built from source `c3ad836aed0d`.
-
-This release fixes LDAP long-DN login, IAM policy refresh, multi-pool deletion safety, tiered-object expiry, multipart replication checksums, and remote rollback-option propagation. It embeds Console v2.5.1 (`fff1618d9abb`) and uses silo-pkg v3.15.2, Go 1.27.3, and upstream minio-go `v7.4.2-0.20261025165926-320af050c445`. Use mcli 20261026 for administration; the container images bundle it as `mcli`.
-
-The Server IAM cache repair enables Console's **Reset to default** action. Existing storage timeouts and cancellation behavior remain: a timeout does not prove that a disk operation stopped or a write did not commit. Read the [coordinated release notes](https://silo.pgsty.com/blog/release/silo-20261026/) and [remaining storage limits](CHANGELOG.md#release2026-10-26t00-00-00z) before upgrading. See [CONTRIBUTORS.md](CONTRIBUTORS.md) for the people behind the code, reports, and documentation.
-
 ## Overview

 PGSTY SILO keeps one maintained release line of the open-source MinIO server alive after upstream ended community distribution: builds, packages, multi-arch images, security fixes, and the full web console. Pigsty runs it in production as its PostgreSQL backup repository.

 It follows one rule — **the product and its delivery surfaces are renamed; the protocol and your data are not.** Everything else lives on [silo.pgsty.com](https://silo.pgsty.com/).

-**Related:** [`pgsty/mc`](https://github.com/pgsty/mc) client (shipped as `mcli`) · [`pgsty/silo-console`](https://github.com/pgsty/silo-console) · [`pgsty/silo-pkg`](https://github.com/pgsty/silo-pkg) · [`pgsty/pigsty`](https://github.com/pgsty/pigsty)
+**Related:** [`pgsty/mc`](https://github.com/pgsty/mc) · [`pgsty/silo-console`](https://github.com/pgsty/silo-console) · [`pgsty/silo-pkg`](https://github.com/pgsty/silo-pkg) · [`pgsty/kes`](https://github.com/pgsty/kes) · [`pgsty/pigsty`](https://github.com/pgsty/pigsty)

 <p align="center">
  <img src="https://silo.pgsty.com/images/silo-console/console-metrics-simple.webp" alt="Silo Console">
 </p>

+#### SILO 20261109
+
+> [!WARNING]
+> [SILO 20261109](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-11-09T00-00-00Z) fixes bugs and security vulnerabilities. We strongly recommend scheduling an upgrade to reduce security risks.
+
+> [!NOTE]
+> On 2026-08-06, `pgsty/minio` became `pgsty/silo`, and the default branch changed from `master` to `main`. The last MinIO-branded release is [`20260804`](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-08-04T00-00-00Z) on the [`minio`](https://github.com/pgsty/silo/tree/minio) branch.
+
 ## Quick Start

 ```bash
@@ -62,7 +59,7 @@ docker run -d --name silo -p 9000:9000 -p 9001:9001 \
  -e MINIO_ROOT_USER=minioadmin \
  -e MINIO_ROOT_PASSWORD=change-me-long-password \
  -v "$PWD/data:/data" \
-  docker.io/pgsty/silo:RELEASE.2026-10-26T00-00-00Z server /data --console-address ":9001"
+  docker.io/pgsty/silo:RELEASE.2026-11-09T00-00-00Z server /data --console-address ":9001"
 ```

 <p align="center">
@@ -32,29 +32,26 @@
 > [!IMPORTANT]
 > **PGSTY Silo**（以下简称 Silo）是由 [Pigsty](https://pigsty.cc) 独立维护、从 [`pgsty/silo`](https://github.com/pgsty/silo) 发布的开源 MinIO 社区分支。本项目与 MinIO, Inc. 不存在隶属、背书或赞助关系；文中使用 “MinIO” 仅用于说明上游项目及兼容谱系。

-> [!NOTE]
-> 2026-08-06，本仓库由 `pgsty/minio` 更名为 `pgsty/silo`，默认分支由 `master` 更名为 `main`。以原 MinIO 形态维持的归档构件仍位于归档的 [`minio`](https://github.com/pgsty/silo/tree/minio) 分支，以及截止 [`RELEASE.2026-08-04T00-00-00Z`](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-08-04T00-00-00Z) 的历次发布中。
-
-## SILO 20261026000000
-
-[SILO 20261026000000](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-26T00-00-00Z) 已正式发布，提供各平台归档包、Linux 软件包，以及 amd64／arm64 标准版和 distroless Docker 镜像。版本标签为 `RELEASE.2026-10-26T00-00-00Z`，包版本为 `20261026000000.0.0`，构建源码为 `c3ad836aed0d`。
-
-本版修复 LDAP 长 DN 登录、IAM 策略刷新、多池删除安全、分层对象过期、分片复制校验和及远端回滚参数透传。内置 Console v2.5.1（`fff1618d9abb`），使用 silo-pkg v3.15.2、Go 1.27.3 与上游 minio-go `v7.4.2-0.20261025165926-320af050c445`。配套管理客户端为 mcli 20261026，容器镜像内置该版本的 `mcli`。
-
-Server IAM 缓存修复使 Console **Reset to default（恢复默认）** 操作即时生效。存储超时与取消行为保持既有语义：超时不代表底层磁盘操作已停止，也不代表写入未提交。升级前请阅读[联合发布说明](https://silo.pgsty.com/zh/blog/release/silo-20261026/)与[仍存的存储限制](CHANGELOG.md#release2026-10-26t00-00-00z)。代码、问题报告和文档贡献者见 [CONTRIBUTORS.md](CONTRIBUTORS.md)。
-
 ## 概述

 上游停止社区发行后，Silo 为开源 MinIO 服务端维护一条持续可用的版本线：构建、软件包、多架构镜像、安全修复与完整 Web 控制台。Pigsty 在生产环境中用它承载 PostgreSQL 备份存储。

 它只遵循一条原则：**改名的是产品与交付物，不是协议与你的数据。** 其余内容都在 [silo.pgsty.com](https://silo.pgsty.com/zh/)。

-**相关项目：**[`pgsty/mc`](https://github.com/pgsty/mc) 客户端（以 `mcli` 发行） · [`pgsty/silo-console`](https://github.com/pgsty/silo-console) · [`pgsty/silo-pkg`](https://github.com/pgsty/silo-pkg) · [`pgsty/pigsty`](https://github.com/pgsty/pigsty)
+**相关项目：**[`pgsty/mc`](https://github.com/pgsty/mc) · [`pgsty/silo-console`](https://github.com/pgsty/silo-console) · [`pgsty/silo-pkg`](https://github.com/pgsty/silo-pkg) · [`pgsty/kes`](https://github.com/pgsty/kes) · [`pgsty/pigsty`](https://github.com/pgsty/pigsty)

 <p align="center">
  <img src="https://silo.pgsty.com/images/silo-console/console-metrics-simple.webp" alt="Silo 控制台">
 </p>

+#### SILO 20261109
+
+> [!WARNING]
+> [SILO 20261109](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-11-09T00-00-00Z) 修复了缺陷与安全漏洞，强烈建议尽快安排升级，以降低安全风险。
+
+> [!NOTE]
+> 2026-08-06，`pgsty/minio` 更名为 `pgsty/silo`，默认分支由 `master` 改为 `main`。最后一个 MinIO 品牌发布是 [`minio`](https://github.com/pgsty/silo/tree/minio) 分支上的 [`20260804`](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-08-04T00-00-00Z)。
+
 ## 快速上手

 ```bash
@@ -62,7 +59,7 @@ docker run -d --name silo -p 9000:9000 -p 9001:9001 \
  -e MINIO_ROOT_USER=minioadmin \
  -e MINIO_ROOT_PASSWORD=change-me-long-password \
  -v "$PWD/data:/data" \
-  docker.io/pgsty/silo:RELEASE.2026-10-26T00-00-00Z server /data --console-address ":9001"
+  docker.io/pgsty/silo:RELEASE.2026-11-09T00-00-00Z server /data --console-address ":9001"
 ```

 <p align="center">
@@ -38,7 +38,7 @@ For vulnerabilities in this fork:

 1. Follow the fork-specific expectations in [VULNERABILITY_REPORT.md](VULNERABILITY_REPORT.md).
 2. Prefer this repository's [private GitHub security advisory](https://github.com/pgsty/silo/security/advisories/new) workflow.
-3. If you cannot use GitHub private reporting, email [ron@pgsty.com](mailto:ron@pgsty.com)
+3. If you cannot use GitHub private reporting, email [ron@pgsty.com](mailto:ron@pgsty.com) or rh@vonng.com
   directly. Do not publish vulnerability details or exploit code in public
   issues or pull requests.
 4. If you confirm the issue also affects upstream `minio/minio`, report it upstream separately.
@@ -40,7 +40,7 @@ if [ -n "${MCLI_BIN:-}" ]; then
 	exit 0
 fi

-release=${MCLI_RELEASE:-RELEASE.2026-10-26T00-00-00Z}
+release=${MCLI_RELEASE:-RELEASE.2026-11-09T00-00-00Z}
 version_hyphen=${release#RELEASE.}
 package_version=$(printf '%s\n' "${version_hyphen}" | sed -E 's/^([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2})-([0-9]{2})-([0-9]{2})Z$/\1\2\3\4\5\6.0.0/')
 if [ "${package_version}" = "${version_hyphen}" ]; then
@@ -112,8 +112,17 @@ helm_run template my-release "${new_chart}" \

 go run ./buildscripts/helm-migration-guard "${old_render}" "${new_render}"

+# The templates deploy values.image.tag, not appVersion; a release must move both.
+chart_version="$(sed -n 's/^version: *//p' helm/silo/Chart.yaml)"
+app_version="$(sed -n 's/^appVersion: *//p' helm/silo/Chart.yaml)"
+image_tag="$(sed -n '/^image:/,/^[^ ]/s/^  tag: *//p' helm/silo/values.yaml)"
+if [ -z "${app_version}" ] || [ "${image_tag}" != "${app_version}" ]; then
+	echo "helm/silo: image.tag '${image_tag}' must equal appVersion '${app_version}'" >&2
+	exit 1
+fi
+
 helm_run package "${new_chart}" --destination "${output_dir}" >/dev/null
-test -s "${work_dir}/silo-7.0.3.tgz"
+test -s "${work_dir}/silo-${chart_version}.tgz"
 if find "${work_dir}" -maxdepth 1 -type f -name 'minio-*.tgz' | grep -q .; then
 	echo "Helm packaging emitted a legacy MinIO chart name" >&2
 	exit 1
@@ -903,6 +903,9 @@ func (a adminAPIHandlers) UpdateServiceAccount(w http.ResponseWriter, r *http.Re
 		description:   updateReq.NewDescription,
 		expiration:    updateReq.NewExpiration,
 		sessionPolicy: sp,
+		// The edit applies to the stored account only while it still has
+		// the parent the checks above authorized.
+		parentUser: svcAccount.ParentUser,
 	}
 	updatedAt, err := globalIAMSys.UpdateServiceAccount(ctx, accessKey, opts)
 	if err != nil {
@@ -791,7 +791,7 @@ func (a adminAPIHandlers) MetricsHandler(w http.ResponseWriter, r *http.Request)

 	enc := json.NewEncoder(w)
 	for n > 0 {
-		var m madmin.RealtimeMetrics
+		var m realtimeMetrics
 		mLocal := collectLocalMetrics(types, collectMetricsOpts{
 			hosts: hostMap,
 			disks: diskMap,
@@ -2675,7 +2675,7 @@ func fetchHealthInfo(healthCtx context.Context, objectAPI ObjectLayer, query *ur
 	// collect all realtime metrics except disk
 	// disk metrics are already included under drive info of each server
 	getRealtimeMetrics := func() *madmin.RealtimeMetrics {
-		var m madmin.RealtimeMetrics
+		var m realtimeMetrics
 		types := madmin.MetricsAll &^ madmin.MetricsDisk
 		mLocal := collectLocalMetrics(types, collectMetricsOpts{})
 		m.Merge(&mLocal)
@@ -2690,7 +2690,8 @@ func fetchHealthInfo(healthCtx context.Context, objectAPI ObjectLayer, query *ur
 			m.ByHost[anonAddr(host)] = metrics
 			delete(m.ByHost, host)
 		}
-		return &m
+		legacy := m.legacy()
+		return &legacy
 	}

 	anonymizeCmdLine := func(cmdLine string) string {
@@ -0,0 +1,118 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/minio/minio/internal/auth"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+// batchContextRead is the object read the tests below authorize.
+func batchContextRead(f *batchAuthzFixture) batchAuthzRequest {
+	return batchAuthzRequest{action: policy.GetObjectAction, bucket: f.bucket, object: "data/a"}
+}
+
+// batchContextUser creates a user with statements as its only policy and
+// returns its submitter binding.
+func batchContextUser(t *testing.T, f *batchAuthzFixture, name, statements string) (auth.Credentials, *batchJobSubmitter) {
+	t.Helper()
+	batchAuthzPolicy(f.ctx, t, f.sys, name, statements)
+	cred := iamAdminUser(f.ctx, t, f.sys, name+"-user", name)
+	sub, err := newBatchJobSubmitter(cred)
+	mustIAM(t, err)
+	return cred, sub
+}
+
+func isBatchContextRefusal(err error) bool {
+	return isBatchJobAuthzError(err) && strings.Contains(err.Error(), "request context")
+}
+
+// A statement whose resources rule the object out does not apply, whatever
+// its conditions need, policy variables in condition values included. One
+// whose resources match still refuses.
+func TestBatchAuthzContextResourceFirst(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	const unrelated = `{"Effect":"Deny","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::other-bucket/*"],"Condition":{"StringEquals":{"s3:ExistingObjectTag/owner":"${aws:UserAgent}"}}}`
+	_, sub := batchContextUser(t, f, "context-unrelated", batchAuthzFullAccess+",\n"+unrelated)
+	if err := sub.authorize(f.ctx, batchContextRead(f)); err != nil {
+		t.Errorf("a Deny on another bucket refused an allowed read: %v", err)
+	}
+
+	related := strings.Replace(unrelated, "other-bucket", f.bucket, 1)
+	_, sub = batchContextUser(t, f, "context-related", batchAuthzFullAccess+",\n"+related)
+	if err := sub.authorize(f.ctx, batchContextRead(f)); !isBatchContextRefusal(err) {
+		t.Errorf("a Deny on the object with an unsupplied variable must refuse, got %v", err)
+	}
+
+	// A variable in the resource leaves open whether it matches.
+	const resourceVar = `{"Effect":"Deny","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::batch-authz/${aws:UserAgent}/*"]}`
+	_, sub = batchContextUser(t, f, "context-resource-var", batchAuthzFullAccess+",\n"+resourceVar)
+	if err := sub.authorize(f.ctx, batchContextRead(f)); !isBatchContextRefusal(err) {
+		t.Errorf("a Deny whose resource holds an unsupplied variable must refuse, got %v", err)
+	}
+}
+
+// An Allow that needs request context the job does not have grants nothing
+// to the job; it neither refuses the job nor grants by an absent key.
+func TestBatchAuthzContextAllowDropped(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	const ipAllow = `{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::*"],"Condition":{"IpAddress":{"aws:SourceIp":["10.0.0.0/8"]}}}`
+	_, sub := batchContextUser(t, f, "context-ip-allow", batchAuthzFullAccess+",\n"+ipAllow)
+	if err := sub.authorize(f.ctx, batchContextRead(f)); err != nil {
+		t.Errorf("a conditional Allow refused a read another Allow grants: %v", err)
+	}
+
+	// A Null condition holds for an absent key: evaluated without the
+	// context, this Allow would grant what no request with an address gets.
+	const nullAllow = `{"Effect":"Allow","Action":["admin:StartBatchJob"]},
+{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"Null":{"aws:SourceIp":"true"}}}`
+	cred, sub := batchContextUser(t, f, "context-null-allow", nullAllow)
+	if !f.sys.isAllowedLocally(policy.Args{AccountName: cred.AccessKey, Action: policy.GetObjectAction, BucketName: f.bucket, ObjectName: "data/a", ConditionValues: map[string][]string{}}) {
+		t.Fatal("control: without the address the Null Allow grants")
+	}
+	err := sub.authorize(f.ctx, batchContextRead(f))
+	if !isBatchJobAuthzError(err) || isBatchContextRefusal(err) {
+		t.Errorf("an Allow that needs the address must grant nothing, and deny rather than refuse; got %v", err)
+	}
+}
+
+// The same holds for the Allows of a session policy, which only narrows.
+func TestBatchAuthzContextSessionAllow(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	batchAuthzPolicy(f.ctx, t, f.sys, "context-session-parent", batchAuthzFullAccess)
+	parent := iamAdminUser(f.ctx, t, f.sys, "context-session-parent", "context-session-parent")
+	for _, tc := range []struct {
+		name, statements string
+		allowed          bool
+	}{
+		{"null-only", `{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"Null":{"aws:SourceIp":"true"}}}`, false},
+		{"with-plain", `{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::batch-authz/*"],"Condition":{"Null":{"aws:SourceIp":"true"}}},
+{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::batch-authz/data/*"]}`, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			sp, err := policy.ParseConfig(strings.NewReader(`{"Version":"2012-10-17","Statement":[` + tc.statements + `]}`))
+			mustIAM(t, err)
+			created, _, err := f.sys.NewServiceAccount(f.ctx, parent.AccessKey, nil, newServiceAccountOpts{sessionPolicy: sp})
+			mustIAM(t, err)
+			stored, ok := f.sys.GetUser(f.ctx, created.AccessKey)
+			if !ok {
+				t.Fatal("service account missing")
+			}
+			sub, err := newBatchJobSubmitter(stored.Credentials)
+			mustIAM(t, err)
+			err = sub.authorize(f.ctx, batchContextRead(f))
+			switch {
+			case isBatchContextRefusal(err):
+				t.Errorf("a session-policy Allow refused the job: %v", err)
+			case tc.allowed && err != nil:
+				t.Errorf("the plain session Allow must still grant: %v", err)
+			case !tc.allowed && !isBatchJobAuthzError(err):
+				t.Errorf("a session policy whose only Allow needs the address must deny, got %v", err)
+			}
+		})
+	}
+}
@@ -34,12 +34,17 @@ import (
 //     (batchJobSubmitter). Nothing in the submitted definition names it.
 //   - Every start of the job, and every object it touches, resolves the
 //     binding against IAM again. A credential that is gone, disabled,
-//     expired, revoked or edited since submission fails the job; a reused
-//     access key does not match the binding.
+//     expired or revoked fails the job, and so does one whose secret,
+//     session token or parent changed since submission; a reused access key
+//     does not match the binding. Edits that leave the credential itself as
+//     it was (a name, a description, a later expiry of an expiring
+//     credential, the LDAP groups) keep the job: its policies, groups,
+//     status and expiry are read again each time.
 //   - Each action is evaluated with the submitter's policies and session
-//     policy but without a request. A policy statement that would need
+//     policy but without a request. A Deny that may apply and would need
 //     request context the job does not have (source address, transport,
-//     headers and so on) refuses the action instead of being skipped.
+//     headers and so on) refuses the action instead of being skipped; an
+//     Allow that would need it grants nothing to the job.
 //
 // A job without a binding was queued by an earlier release. It is never
 // authorized, whoever submitted it, and has to be submitted again.
@@ -54,8 +59,13 @@ type batchJobSubmitter struct {
 	AccessKey string `msg:"ak"`
 	// ParentUser is the parent of a derived credential.
 	ParentUser string `msg:"pu"`
-	// UpdatedAt and RevokedBefore are the revision of the identity record;
-	// any edit of the identity changes them. They are unset for root.
+	// UpdatedAt is the revision of the identity record at submission. It is
+	// recorded for diagnosis only: an edit that keeps the credential, such as
+	// a description, a later expiry or an LDAP group refresh, changes it and
+	// must not end the job. RevokedBefore is the revocation boundary of the
+	// record, which edits keep and a deletion advances, so a credential
+	// created again under the access key does not match. Both are unset for
+	// root.
 	UpdatedAt     time.Time `msg:"ua"`
 	RevokedBefore time.Time `msg:"rb"`
 	// Binding is a keyed fingerprint of the credential itself, see
@@ -228,8 +238,13 @@ func (s *batchJobSubmitter) principal(ctx context.Context) (batchJobPrincipal, e
 	}
 	cred := u.Credentials
 	typ := batchSubmitterType(cred)
+	// The binding covers the secret, the session token and the parent, so a
+	// rotated or reissued credential does not match it. The rest of the
+	// identity record (UpdatedAt) changes with edits that keep the
+	// credential and is not compared; what those edits decide, status,
+	// expiry, groups and policies, is evaluated below and per action.
 	if typ != s.Type || cred.AccessKey != s.AccessKey || cred.ParentUser != s.ParentUser ||
-		!u.UpdatedAt.Equal(s.UpdatedAt) || !u.RevokedBefore.Equal(s.RevokedBefore) ||
+		!u.RevokedBefore.Equal(s.RevokedBefore) ||
 		!hmac.Equal(s.Binding, batchSubmitterBinding(typ, cred)) {
 		return batchJobPrincipal{}, errBatchJobChanged
 	}
@@ -474,22 +489,36 @@ func (p batchJobPrincipal) authorizeArgs(args policy.Args, req *batchAuthzReques
 		snap = globalIAMSys.store.batchSnapshot(ctx)
 		defer snap.release()
 	}
-	policies, err := globalIAMSys.batchLocalPolicies(p, args, plugin == nil, snap)
+	mapped, session, err := globalIAMSys.batchLocalPolicies(p, args, plugin == nil, snap)
 	if err != nil {
 		if errors.Is(err, errNoSuchPolicy) || errors.Is(err, errAuthentication) || errors.Is(err, errNoSuchUser) || errors.Is(err, errNoSuchGroup) {
 			return denied
 		}
 		return batchUnavailable("IAM policy lookup", err)
 	}
+	// sessionNarrowed: the session policy lost an Allow below, so the
+	// narrowed policy decides in its place.
+	sessionNarrowed := false
 	if req != nil {
 		supplied := req.suppliedConditionKeys()
-		for _, pol := range policies {
-			for _, st := range pol.Statements {
-				if batchStatementNeedsRequest(st, args, supplied) {
-					return batchJobAuthzError{reason: denied.reason + ": a policy statement depends on request context that a batch job does not have"}
-				}
-			}
+		var refuse bool
+		if mapped, _, refuse = batchContextFree(mapped, args, supplied); !refuse {
+			session, sessionNarrowed, refuse = batchContextFree(session, args, supplied)
 		}
+		if refuse {
+			return batchJobAuthzError{reason: denied.reason + ": a policy statement depends on request context that a batch job does not have"}
+		}
+	}
+	sessionDenies := func() bool {
+		has, allowed := p.sessionPolicyDecision(args)
+		if has && allowed && sessionNarrowed {
+			// As sessionPolicyDecision evaluates it, without its dropped
+			// Allows. This can only turn an allow into a deny.
+			spArgs := args
+			spArgs.IsOwner, spArgs.DenyOnly = false, false
+			allowed = session.IsAllowed(spArgs)
+		}
+		return has && !allowed
 	}
 	if plugin != nil {
 		ok, err := plugin.IsAllowedContext(ctx, args)
@@ -500,7 +529,7 @@ func (p batchJobPrincipal) authorizeArgs(args policy.Args, req *batchAuthzReques
 			return denied
 		}
 		// The plugin does not see a session policy; it only limits further.
-		if has, allowed := p.sessionPolicyDecision(args); has && !allowed {
+		if sessionDenies() {
 			return denied
 		}
 		return nil
@@ -516,7 +545,7 @@ func (p batchJobPrincipal) authorizeArgs(args policy.Args, req *batchAuthzReques
 			return denied
 		}
 	}
-	if has, allowed := p.sessionPolicyDecision(args); has && !allowed {
+	if sessionDenies() {
 		return denied
 	}
 	ownerDerived := (p.temp || p.service) && p.parentUser == globalActiveCred.AccessKey
@@ -524,13 +553,50 @@ func (p batchJobPrincipal) authorizeArgs(args policy.Args, req *batchAuthzReques
 		if p.service {
 			args.AccountName = p.parentUser
 		}
-		if len(policies) == 0 || !policies[0].IsAllowed(args) {
+		if mapped == nil || !mapped.IsAllowed(args) {
 			return denied
 		}
 	}
 	return nil
 }

+// batchContextFree returns pol without the Allow statements that may apply
+// to args but need request context the job does not supply (see
+// batchStatementNeedsRequest), and whether it dropped any. Such an Allow
+// grants nothing to the job. It is dropped rather than evaluated, because
+// evaluated without the context it could grant what a request would not:
+// IfExists and Null conditions hold for an absent key. Dropping an Allow can
+// only narrow the decision. Dropping a Deny that needs the context could
+// allow what a request with that context would not, so refuse reports one,
+// and the job refuses the action.
+func batchContextFree(pol *policy.Policy, args policy.Args, supplied map[string]bool) (out *policy.Policy, dropped, refuse bool) {
+	if pol == nil {
+		return nil, false, false
+	}
+	var kept []policy.Statement
+	for i, st := range pol.Statements {
+		if !batchStatementNeedsRequest(st, args, supplied) {
+			if dropped {
+				kept = append(kept, st)
+			}
+			continue
+		}
+		if st.Effect != policy.Allow {
+			return pol, false, true
+		}
+		if !dropped {
+			kept = append(make([]policy.Statement, 0, len(pol.Statements)-1), pol.Statements[:i]...)
+			dropped = true
+		}
+	}
+	if !dropped {
+		return pol, false, false
+	}
+	// A new policy: pol's statement index refers to pol's own statements, and
+	// pol may be a cached policy document.
+	return &policy.Policy{ID: pol.ID, Version: pol.Version, Statements: kept}, true, false
+}
+
 func batchDeniedReason(req batchAuthzRequest) string {
 	resource := req.bucket
 	if req.object != "" {
@@ -636,13 +702,13 @@ func (s *batchPolicySnapshot) loadPolicyDoc(name string, out map[string]PolicyDo
 }

 // batchLocalPolicies returns every local policy that can take part in the
-// decision on args: the mapped policies that IsAllowed would merge when
-// withMapped is set (no authorization plugin), and the session policy.
-func (sys *IAMSys) batchLocalPolicies(p batchJobPrincipal, args policy.Args, withMapped bool, snap *batchPolicySnapshot) ([]policy.Policy, error) {
-	var out []policy.Policy
+// decision on args: mapped, the mapped policies merged as IsAllowed would
+// merge them, when withMapped is set (no authorization plugin), and session,
+// the session policy. Each is nil when it takes no part.
+func (sys *IAMSys) batchLocalPolicies(p batchJobPrincipal, args policy.Args, withMapped bool, snap *batchPolicySnapshot) (mapped, session *policy.Policy, err error) {
 	if withMapped && args.IsOwner {
 		// IsAllowed does not evaluate any policy for the owner.
-		return nil, nil
+		return nil, nil, nil
 	}
 	if withMapped {
 		// The caller holds snap through evaluation. Lazy mapping loads and
@@ -664,28 +730,28 @@ func (sys *IAMSys) batchLocalPolicies(p batchJobPrincipal, args policy.Args, wit
 			if roleArn := args.GetRoleArn(); roleArn != "" {
 				a, err := arn.Parse(roleArn)
 				if err != nil {
-					return nil, err
+					return nil, nil, err
 				}
 				names = append(names, newMappedPolicy(sys.rolesMap[a]).toSlice()...)
 			} else {
-				mapped, err := mappedPolicies(p.parentUser)
+				parent, err := mappedPolicies(p.parentUser)
 				if err != nil {
-					return nil, err
+					return nil, nil, err
 				}
-				names = append(names, mapped...)
+				names = append(names, parent...)
 				// The claimed policies apply only when nothing is mapped.
-				if len(mapped) == 0 {
+				if len(parent) == 0 {
 					if claimed, ok := policy.GetPoliciesFromClaims(args.Claims, iamPolicyClaimNameOpenID()); ok {
 						names = append(names, claimed.ToSlice()...)
 					}
 				}
 			}
 		default:
-			mapped, err := mappedPolicies(p.accessKey)
+			own, err := mappedPolicies(p.accessKey)
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
-			names = append(names, mapped...)
+			names = append(names, own...)
 		}
 		var selected []policy.Policy
 		for _, name := range names {
@@ -700,63 +766,56 @@ func (sys *IAMSys) batchLocalPolicies(p batchJobPrincipal, args policy.Args, wit
 					continue
 				}
 				if err != nil {
-					return nil, err
+					return nil, nil, err
 				}
 				doc, ok = loaded[name]
 				if !ok {
-					return nil, fmt.Errorf("load IAM policy %q returned no document", name)
+					return nil, nil, fmt.Errorf("load IAM policy %q returned no document", name)
 				}
 			}
 			selected = append(selected, doc.Policy)
 		}
-		out = append(out, policy.MergePolicies(selected...))
+		merged := policy.MergePolicies(selected...)
+		mapped = &merged
 	}
 	if p.temp || p.service {
 		if p.service && p.claims[iamPolicyClaimNameSA()] == inheritedPolicyType {
-			return out, nil
+			return mapped, nil, nil
 		}
 		if sp, ok := p.claims[sessionPolicyNameExtracted]; ok {
 			s, ok := sp.(string)
 			if !ok {
-				return nil, errAuthentication
+				return nil, nil, errAuthentication
 			}
 			parsed, err := policy.ParseConfig(strings.NewReader(s))
 			if err != nil {
-				return nil, err
+				return nil, nil, err
 			}
-			out = append(out, *parsed)
+			session = parsed
 		}
 	}
-	return out, nil
+	return mapped, session, nil
 }

-// batchStatementNeedsRequest reports whether a statement that applies to
-// args depends on a condition key that the job does not supply. Applying
+// batchStatementNeedsRequest reports whether a statement may apply to args
+// and depends on request context that the job does not supply: a condition
+// key, or a policy variable in a resource or a condition value. Applying
 // such a statement without its value could grant (Allow) or drop (Deny)
-// what a request with that context would not, so the job refuses the action.
-// Resources are matched without conditions; a statement is taken to apply
-// whenever it might.
+// what a request with that context would not (see batchContextFree).
+//
+// Resources are matched first and without conditions; a statement is taken
+// to apply whenever it might. A statement whose resources rule args out
+// does not apply, whatever its conditions need. One whose resources hold a
+// variable the job does not supply might match, so it may apply.
 func batchStatementNeedsRequest(st policy.Statement, args policy.Args, supplied map[string]bool) bool {
 	if (!st.Actions.Match(args.Action) && !st.Actions.IsEmpty()) || st.NotActions.Match(args.Action) {
 		return false
 	}
-	// Policy variables in resources or condition values.
-	if data, err := json.Marshal(st); err == nil {
-		s := string(data)
-		for {
-			i := strings.Index(s, "${")
-			if i < 0 {
-				break
-			}
-			s = s[i+2:]
-			j := strings.IndexByte(s, '}')
-			if j < 0 {
-				break
-			}
-			if !batchKeySupplied(supplied, s[:j]) {
+	for _, resources := range []policy.ResourceSet{st.Resources, st.NotResources} {
+		for r := range resources {
+			if batchUnsuppliedVariable(r.Pattern, supplied) {
 				return true
 			}
-			s = s[j+1:]
 		}
 	}
 	if !batchStatementMayApply(st, args) {
@@ -767,9 +826,36 @@ func batchStatementNeedsRequest(st policy.Statement, args policy.Args, supplied
 			return true
 		}
 	}
+	// Policy variables in condition values.
+	if len(st.Conditions) > 0 {
+		data, err := json.Marshal(st.Conditions)
+		if err != nil || batchUnsuppliedVariable(string(data), supplied) {
+			return true
+		}
+	}
 	return false
 }

+// batchUnsuppliedVariable reports whether s holds a policy variable, ${key},
+// whose value the job does not supply.
+func batchUnsuppliedVariable(s string, supplied map[string]bool) bool {
+	for {
+		i := strings.Index(s, "${")
+		if i < 0 {
+			return false
+		}
+		s = s[i+2:]
+		j := strings.IndexByte(s, '}')
+		if j < 0 {
+			return false
+		}
+		if !batchKeySupplied(supplied, s[:j]) {
+			return true
+		}
+		s = s[j+1:]
+	}
+}
+
 // batchStatementMayApply matches the resources of a statement, in both forms
 // a bucket request can take. A statement whose resources the evaluator
 // translates (S3 Tables and vectors actions cover S3 data actions) is taken
@@ -196,11 +196,23 @@ func TestBatchExpirePurgeRefusesLockedAfterWalk(t *testing.T) {
 func (f *batchAuthzFixture) transition(t *testing.T, key, versionID string) {
 	t.Helper()
 	set := f.obj.(*erasureServerPools).serverPools[0]
-	infos, errs := readAllFileInfo(f.ctx, set.getHashedSet(key).getDisks(), "", f.bucket, key, versionID, false, false)
+	er := set.getHashedSet(key)
+	// DecomTieredObject takes its parity from the global storage class, which
+	// earlier tests can leave at a parity this set cannot have; it then
+	// writes on no drive and still succeeds. Use the set's own parity; the
+	// whole configuration is restored when the test ends.
+	setTestStandardParity(t, er.defaultParityCount)
+	infos, errs := readAllFileInfo(f.ctx, er.getDisks(), "", f.bucket, key, versionID, false, false)
 	mustIAM(t, errs[0])
 	fi := infos[0]
 	fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", "remote/"+key+"/"+versionID
 	mustIAM(t, set.DecomTieredObject(f.ctx, f.bucket, key, fi, ObjectOptions{}))
+	infos, errs = readAllFileInfo(f.ctx, er.getDisks(), "", f.bucket, key, versionID, false, false)
+	for i := range infos {
+		if errs[i] != nil || infos[i].TransitionStatus != lifecycle.TransitionComplete {
+			t.Fatalf("drive %d does not hold the transition: %v", i, errs[i])
+		}
+	}
 }

 // A key with a transitioned version is refused as a whole, as a key with a
@@ -462,6 +462,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 	if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
 		return err
 	}
+	ri.resetReplicationAttempt()
 	globalBatchJobsMetrics.save(job.ID, ri)

 	retryAttempts := job.Replicate.Flags.Retry.Attempts
@@ -576,9 +577,10 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 		s3Type := r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3
 		minioSrc := r.Source.Type == BatchJobReplicateResourceMinIO
 		ctx, cancelCause := context.WithCancelCause(ctx)
-		// refuse ends the job for its submitter: record why before stopping.
-		refuse := func(err error) {
-			ri.refuse(err)
+		stop := func(err error) {
+			if isBatchJobAuthzError(err) {
+				ri.refuse(err)
+			}
 			cancelCause(err)
 		}
 		// Each run lists the source from the start and counts every locked
@@ -615,6 +617,15 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 				// Refused or canceled: schedule nothing more.
 				continue
 			}
+			if obj.Err != nil {
+				switch minio.ToErrorResponse(obj.Err).StatusCode {
+				case http.StatusUnauthorized, http.StatusForbidden:
+					stop(batchJobAuthzError{reason: "source listing denied: " + obj.Err.Error()})
+				default:
+					stop(batchListingError(obj.Err))
+				}
+				continue
+			}
 			oi := toObjectInfo(r.Source.Bucket, obj.Key, obj)
 			if !minioSrc {
 				// Check if metadata filter was requested and it is expected to have
@@ -663,8 +674,11 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 				stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts)
 				success := true
 				if err := r.ReplicateFromSource(ctx, api, core, oi, retry, job.Submitter); err != nil {
-					if isBatchJobAuthzError(err) {
-						refuse(err)
+					if isBatchJobStopped(err) {
+						stop(err)
+						return
+					}
+					if ctx.Err() != nil {
 						return
 					}
 					// A locked destination version is left as it is: the
@@ -706,6 +720,14 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 			}
 			return nil
 		}
+		if cause := context.Cause(ctx); cause != nil {
+			cancelCause(nil)
+			if isBatchJobUnavailable(cause) {
+				ri.resetReplicationAttempt()
+				return ri.savePaused(ctx, api, job, cause)
+			}
+			return cause
+		}

 		ri.RetryAttempts = attempts
 		ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0
@@ -721,14 +743,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay

 		cancelCause(nil)
 		if ri.Failed {
-			ri.ObjectsFailed = 0
-			ri.DeleteMarkersFailed = 0
-			ri.Bucket = ""
-			ri.Object = ""
-			ri.Objects = 0
-			ri.DeleteMarkers = 0
-			ri.BytesFailed = 0
-			ri.BytesTransferred = 0
+			ri.resetReplicationAttempt()
 			retry = true // indicate we are retrying..
 			time.Sleep(delay + time.Duration(rnd.Float64()*float64(delay)))
 			continue
@@ -1337,6 +1352,21 @@ func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInf
 	}
 }

+func (ri *batchJobInfo) resetReplicationAttempt() {
+	ri.mu.Lock()
+	defer ri.mu.Unlock()
+	ri.Complete, ri.Failed = false, false
+	ri.ObjectsFailed = 0
+	ri.DeleteMarkersFailed = 0
+	ri.Bucket = ""
+	ri.Object = ""
+	ri.Objects = 0
+	ri.DeleteMarkers = 0
+	ri.BytesFailed = 0
+	ri.BytesTransferred = 0
+	ri.VersionsSkippedLocked = 0
+}
+
 // Start start the batch replication job, resumes if there was a pending job via "job.ID"
 func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {
 	ri := &batchJobInfo{
@@ -1353,8 +1383,8 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 	if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
 		return err
 	}
+	ri.resetReplicationAttempt()
 	globalBatchJobsMetrics.save(job.ID, ri)
-	lastObject := ri.Object

 	retryAttempts := job.Replicate.Flags.Retry.Attempts
 	if retryAttempts <= 0 {
@@ -1461,9 +1491,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 			slowCh = make(chan itemOrErr[ObjectInfo], 100)
 		)
 		ctx, cancelCause := context.WithCancelCause(ctx)
-		// refuse ends the job for its submitter: record why before stopping.
-		refuse := func(err error) {
-			ri.refuse(err)
+		stop := func(err error) {
+			if isBatchJobAuthzError(err) {
+				ri.refuse(err)
+			}
 			cancelCause(err)
 		}

@@ -1488,8 +1519,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 				writeFn := func(batch []ObjectInfo) {
 					if len(batch) > 0 {
 						if err := r.writeAsArchive(ctx, api, cl, batch, r.Target.Prefix, job.Submitter); err != nil {
-							if isBatchJobAuthzError(err) {
-								refuse(err)
+							if isBatchJobStopped(err) {
+								stop(err)
+								return
+							}
+							if ctx.Err() != nil {
 								return
 							}
 							batchLogOnceIf(ctx, err, job.ID+"writeAsArchive")
@@ -1505,7 +1539,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 					}
 				}
 				for obj := range walkCh {
-					if obj.Item.DeleteMarker || !obj.Item.VersionPurgeStatus.Empty() || obj.Item.Size >= int64(smallerThan) {
+					if context.Cause(ctx) != nil {
+						continue
+					}
+					if obj.Err != nil || obj.Item.DeleteMarker || !obj.Item.VersionPurgeStatus.Empty() || obj.Item.Size >= int64(smallerThan) {
 						slowCh <- obj
 						continue
 					}
@@ -1518,7 +1555,9 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 					writeFn(batch)
 					batch = batch[:0]
 				}
-				writeFn(batch)
+				if context.Cause(ctx) == nil {
+					writeFn(batch)
+				}
 				xioutil.SafeClose(slowCh)
 			}()
 		} else {
@@ -1551,11 +1590,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 			for _, prefix := range prefixes {
 				prefixWalkCh := make(chan itemOrErr[ObjectInfo], 100)
 				if err := api.Walk(ctx, r.Source.Bucket, prefix, prefixWalkCh, WalkOptions{
-					Marker:   lastObject,
 					Filter:   selectObj,
 					AskDisks: walkQuorum,
 				}); err != nil {
-					cancelCause(err)
+					cancelCause(batchListingError(err))
 					xioutil.SafeClose(walkCh)
 					return
 				}
@@ -1575,7 +1613,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 				continue
 			}
 			if res.Err != nil {
-				ri.Failed = true
+				cancelCause(batchListingError(res.Err))
 				batchLogOnceIf(ctx, res.Err, job.ID+"res.Err")
 				continue
 			}
@@ -1594,8 +1632,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 				stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts)
 				success := true
 				if err := r.ReplicateToTarget(ctx, api, c, result, retry, job.Submitter); err != nil {
-					if isBatchJobAuthzError(err) {
-						refuse(err)
+					if isBatchJobStopped(err) {
+						stop(err)
+						return
+					}
+					if ctx.Err() != nil {
 						return
 					}
 					if minio.ToErrorResponse(err).Code == "PreconditionFailed" {
@@ -1632,9 +1673,13 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 			}
 			return nil
 		}
-		// Do not need to retry if we can't list objects on source.
-		if context.Cause(ctx) != nil {
-			return context.Cause(ctx)
+		if cause := context.Cause(ctx); cause != nil {
+			cancelCause(nil)
+			if isBatchJobUnavailable(cause) {
+				ri.resetReplicationAttempt()
+				return ri.savePaused(ctx, api, job, cause)
+			}
+			return cause
 		}
 		ri.RetryAttempts = attempts
 		ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0
@@ -1650,14 +1695,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba

 		cancelCause(nil)
 		if ri.Failed {
-			ri.ObjectsFailed = 0
-			ri.DeleteMarkersFailed = 0
-			ri.Bucket = ""
-			ri.Object = ""
-			ri.Objects = 0
-			ri.DeleteMarkers = 0
-			ri.BytesFailed = 0
-			ri.BytesTransferred = 0
+			ri.resetReplicationAttempt()
 			retry = true // indicate we are retrying..
 			time.Sleep(delay + time.Duration(rnd.Float64()*float64(delay)))
 			continue
@@ -2627,20 +2665,20 @@ func (ri *batchJobInfo) metric() madmin.JobMetric {
 	return m
 }

-func (m *batchJobMetrics) report(jobID string) (metrics *madmin.BatchJobMetrics) {
-	metrics = &madmin.BatchJobMetrics{CollectedAt: time.Now(), Jobs: make(map[string]madmin.JobMetric)}
+func (m *batchJobMetrics) report(jobID string) (metrics *batchJobMetricsReport) {
+	metrics = &batchJobMetricsReport{CollectedAt: time.Now(), Jobs: make(map[string]batchJobStatusMetric)}
 	m.RLock()
 	defer m.RUnlock()

 	if jobID != "" {
 		if job, ok := m.metrics[jobID]; ok {
-			metrics.Jobs[jobID] = job.metric()
+			metrics.Jobs[jobID] = job.statusMetric()
 		}
 		return metrics
 	}

 	for id, job := range m.metrics {
-		metrics.Jobs[id] = job.metric()
+		metrics.Jobs[id] = job.statusMetric()
 	}
 	return metrics
 }
@@ -170,7 +170,7 @@ type batchJobStatus struct {

 type batchJobStatusMetric struct {
 	madmin.JobMetric
-	VersionsSkippedLocked int64  `json:"versionsSkippedLocked,omitempty"`
+	VersionsSkippedLocked *int64 `json:"versionsSkippedLocked,omitempty"`
 	KeysRefused           int64  `json:"keysRefused,omitempty"`
 	FailureReason         string `json:"failureReason,omitempty"`
 	PausedReason          string `json:"pausedReason,omitempty"`
@@ -179,9 +179,10 @@ type batchJobStatusMetric struct {
 func (ri *batchJobInfo) statusMetric() batchJobStatusMetric {
 	ri.mu.RLock()
 	defer ri.mu.RUnlock()
+	versionsSkipped := ri.VersionsSkippedLocked
 	return batchJobStatusMetric{
 		JobMetric:             ri.metric(),
-		VersionsSkippedLocked: ri.VersionsSkippedLocked,
+		VersionsSkippedLocked: &versionsSkipped,
 		KeysRefused:           ri.KeysRefused,
 		FailureReason:         ri.FailureReason,
 		PausedReason:          ri.PausedReason,
@@ -0,0 +1,343 @@
+package cmd
+
+import (
+	"archive/tar"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	polplugin "github.com/minio/minio/internal/config/policy/plugin"
+	xnet "github.com/pgsty/silo-pkg/v3/net"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+type batchReplicationWriteObserver struct {
+	ObjectLayer
+	written func(string, string)
+}
+
+type batchReplicationWalkFailure struct {
+	ObjectLayer
+	streamed bool
+}
+
+func (failure batchReplicationWalkFailure) Walk(ctx context.Context, _, _ string, results chan<- itemOrErr[ObjectInfo], _ WalkOptions) error {
+	if !failure.streamed {
+		return errErasureReadQuorum
+	}
+	go func() {
+		defer close(results)
+		select {
+		case results <- itemOrErr[ObjectInfo]{Err: errErasureReadQuorum}:
+		case <-ctx.Done():
+		}
+	}()
+	return nil
+}
+
+func TestBatchReplicationListingUnavailable(t *testing.T) {
+	for _, snowball := range []bool{false, true} {
+		for _, streamed := range []bool{false, true} {
+			t.Run(fmt.Sprintf("snowball=%t/streamed=%t", snowball, streamed), func(t *testing.T) {
+				fixture := newBatchAuthzFixture(t)
+				job := BatchJobRequest{
+					ID: "replicate-listing" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+					Replicate: &BatchJobReplicateV1{
+						Source: BatchJobReplicateSource{
+							Type: BatchJobReplicateResourceMinIO, Bucket: fixture.bucket,
+							Snowball: BatchJobSnowball{
+								Disable: ptr(!snowball), InMemory: ptr(true), Compress: ptr(false), SkipErrs: ptr(false),
+								Batch: ptr(1), SmallerThan: ptr("1MiB"),
+							},
+						},
+						Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://target.invalid"},
+					},
+				}
+				api := batchReplicationWalkFailure{ObjectLayer: fixture.obj, streamed: streamed}
+				if err := job.Replicate.Start(fixture.ctx, api, job); !isBatchJobUnavailable(err) {
+					t.Fatalf("listing failure did not pause: %v", err)
+				}
+				report := fixture.jobInfo(t, &job)
+				if report.PausedReason == "" || report.Failed || report.Complete || report.Object != "" || report.ObjectsFailed != 0 {
+					t.Fatalf("listing failure saved an invalid outcome: %+v", report)
+				}
+			})
+		}
+	}
+}
+
+func (observer batchReplicationWriteObserver) PutObject(ctx context.Context, bucket, object string, reader *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
+	info, err := observer.ObjectLayer.PutObject(ctx, bucket, object, reader, opts)
+	if err == nil {
+		observer.written(bucket, object)
+	}
+	return info, err
+}
+
+func TestBatchReplicationRemoteListingDenied(t *testing.T) {
+	fixture := newBatchAuthzFixture(t)
+	previousTransport := remoteInstanceTransport
+	remoteInstanceTransport = atomic.Value{}
+	setRemoteInstanceTransport(batchLockTransport(func(request *http.Request) (*http.Response, error) {
+		status := http.StatusForbidden
+		body := `<Error><Code>AccessDenied</Code><Message>source listing denied</Message></Error>`
+		if request.URL.Query().Has("location") {
+			status = http.StatusOK
+			body = `<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/">us-east-1</LocationConstraint>`
+		}
+		return &http.Response{
+			StatusCode: status, Header: make(http.Header),
+			Body: io.NopCloser(strings.NewReader(body)), ContentLength: int64(len(body)), Request: request,
+		}, nil
+	}))
+	defer func() { remoteInstanceTransport = previousTransport }()
+	job := BatchJobRequest{
+		ID: "replicate-listing-denied" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+		Replicate: &BatchJobReplicateV1{
+			Source: BatchJobReplicateSource{
+				Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://source.invalid",
+				Creds: BatchJobReplicateCredentials{AccessKey: "test-access", SecretKey: "test-secret"},
+			},
+			Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: fixture.bucket},
+		},
+	}
+	if err := job.Replicate.StartFromSource(fixture.ctx, fixture.obj, job); err != nil {
+		t.Fatalf("explicit remote denial was not terminal: %v", err)
+	}
+	report := fixture.jobInfo(t, &job)
+	if !report.Failed || report.Complete || report.FailureReason == "" || report.PausedReason != "" {
+		t.Fatalf("remote denial saved an invalid outcome: %+v", report)
+	}
+}
+
+func TestBatchReplicationDependencyRecovery(t *testing.T) {
+	for _, mode := range []string{"push", "pull", "snowball"} {
+		for _, denied := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/denied=%t", mode, denied), func(t *testing.T) {
+				t.Setenv("_MINIO_BATCH_REPLICATION_WORKERS", "2")
+				fixture := newBatchAuthzFixture(t)
+				keys := []string{"a-done", "b-pending", "c-later"}
+				versions := make(map[string]string)
+				for _, key := range keys {
+					versions[key] = mustGetUUID()
+					if mode != "pull" {
+						versions[key] = batchPutVersion(t, fixture.obj, fixture.bucket, key, nil).VersionID
+					}
+				}
+				laterDone := make(chan struct{})
+				signalLater := sync.OnceFunc(func() { close(laterDone) })
+				var writtenMu sync.Mutex
+				written := make(map[string]bool)
+				recordWrite := func(bucket, key string) {
+					if bucket != fixture.bucket && bucket != "remote" {
+						return
+					}
+					writtenMu.Lock()
+					written[key] = true
+					writtenMu.Unlock()
+					if key == "c-later" {
+						signalLater()
+					}
+				}
+				var outage atomic.Bool
+				outage.Store(true)
+				var decisions atomic.Int32
+				pluginServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+					var input struct {
+						Input policy.Args `json:"input"`
+					}
+					if err := json.NewDecoder(request.Body).Decode(&input); err != nil {
+						t.Error(err)
+						writer.WriteHeader(http.StatusBadRequest)
+						return
+					}
+					action := policy.Action(policy.GetObjectAction)
+					if mode == "pull" {
+						action = policy.PutObjectAction
+					}
+					if outage.Load() && input.Input.Action == action && input.Input.ObjectName == "b-pending" {
+						decisions.Add(1)
+						if mode != "snowball" {
+							select {
+							case <-laterDone:
+							case <-request.Context().Done():
+								return
+							}
+						}
+						if denied {
+							io.WriteString(writer, `{"result":false}`)
+							return
+						}
+						writer.WriteHeader(http.StatusServiceUnavailable)
+					}
+					io.WriteString(writer, `{"result":true}`)
+				}))
+				defer pluginServer.Close()
+				pluginURL, err := xnet.ParseHTTPURL(pluginServer.URL)
+				mustIAM(t, err)
+				previousPlugin := newGlobalAuthZPluginFn()
+				setGlobalAuthZPlugin(polplugin.New(polplugin.Args{
+					URL: pluginURL, CloseRespFn: func(reader io.ReadCloser) { reader.Close() },
+				}))
+				defer setGlobalAuthZPlugin(previousPlugin)
+				previousTransport := remoteInstanceTransport
+				remoteInstanceTransport = atomic.Value{}
+				setRemoteInstanceTransport(batchLockTransport(func(request *http.Request) (*http.Response, error) {
+					if err := request.Context().Err(); err != nil {
+						return nil, err
+					}
+					body := ""
+					header := make(http.Header)
+					query := request.URL.Query()
+					switch {
+					case query.Has("location"):
+						body = `<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/">us-east-1</LocationConstraint>`
+					case query.Has("versions"):
+						body = `<ListVersionsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>remote</Name><IsTruncated>false</IsTruncated>`
+						for _, key := range keys {
+							body += fmt.Sprintf(`<Version><Key>%s</Key><VersionId>%s</VersionId><IsLatest>true</IsLatest><LastModified>2026-11-01T00:00:00.000Z</LastModified><ETag>"etag"</ETag><Size>7</Size><StorageClass>STANDARD</StorageClass></Version>`, key, versions[key])
+						}
+						body += `</ListVersionsResult>`
+					case request.Method == http.MethodGet:
+						body = "payload"
+						header.Set("Etag", `"etag"`)
+						header.Set("Last-Modified", UTCNow().Format(http.TimeFormat))
+						header.Set("X-Amz-Version-Id", query.Get("versionId"))
+					case request.Method == http.MethodPut:
+						if mode == "snowball" && request.Body != nil {
+							archive := tar.NewReader(request.Body)
+							for {
+								entry, readErr := archive.Next()
+								if readErr == io.EOF {
+									break
+								}
+								if readErr != nil {
+									return nil, readErr
+								}
+								recordWrite("remote", entry.Name)
+							}
+						} else if mode != "snowball" {
+							if _, err := io.Copy(io.Discard, request.Body); err != nil {
+								return nil, err
+							}
+							recordWrite("remote", strings.TrimPrefix(request.URL.Path, "/remote/"))
+						}
+						header.Set("Etag", `"etag"`)
+					default:
+						return nil, fmt.Errorf("unexpected replication request: %s %s", request.Method, request.URL)
+					}
+					header.Set("Content-Length", strconv.Itoa(len(body)))
+					return &http.Response{
+						StatusCode: http.StatusOK, Header: header,
+						Body: io.NopCloser(strings.NewReader(body)), ContentLength: int64(len(body)), Request: request,
+					}, nil
+				}))
+				defer func() { remoteInstanceTransport = previousTransport }()
+				job := &BatchJobRequest{
+					ID: "replicate-retry" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+					Replicate: &BatchJobReplicateV1{
+						Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: fixture.bucket},
+						Target: BatchJobReplicateTarget{
+							Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://target.invalid",
+							Creds: BatchJobReplicateCredentials{AccessKey: "test-access", SecretKey: "test-secret"},
+						},
+					},
+				}
+				if mode == "pull" {
+					job.Replicate.Source.Endpoint = job.Replicate.Target.Endpoint
+					job.Replicate.Source.Creds = job.Replicate.Target.Creds
+					job.Replicate.Source.Bucket = "remote"
+					job.Replicate.Target.Endpoint = ""
+					job.Replicate.Target.Bucket = fixture.bucket
+				}
+				if mode == "snowball" {
+					job.Replicate.Source.Snowball = BatchJobSnowball{
+						Disable: ptr(false), InMemory: ptr(true), Compress: ptr(false), SkipErrs: ptr(false),
+						Batch: ptr(1), SmallerThan: ptr("1MiB"),
+					}
+				}
+				job.Replicate.Flags.Retry.Attempts = 1
+				job.Replicate.Flags.Retry.Delay = time.Millisecond
+				interrupted := &batchJobInfo{
+					JobID: job.ID, JobType: string(job.Type()), StartTime: job.Started,
+					Object: "c-later", Objects: 2,
+				}
+				mustIAM(t, interrupted.updateAfter(fixture.ctx, fixture.obj, 0, *job))
+				pool := globalBatchJobPool
+				pool.objLayer = batchReplicationWriteObserver{ObjectLayer: fixture.obj, written: recordWrite}
+				ctx, cancel := context.WithCancel(fixture.ctx)
+				pool.ctx = ctx
+				mustIAM(t, pool.queueJob(job))
+				done := make(chan struct{})
+				go func() { defer close(done); pool.AddWorker() }()
+				defer func() {
+					cancel()
+					<-done
+					if t.Failed() {
+						t.Logf("last report: %+v", fixture.jobInfo(t, job))
+					}
+				}()
+				state := "paused"
+				if denied {
+					state = "terminal"
+				}
+				waitBatchState(t, pool, job.ID, state)
+				report := fixture.jobInfo(t, job)
+				if decisions.Load() == 0 {
+					t.Fatal("test did not reach per-object authorization")
+				}
+				if denied {
+					if !report.Failed || report.Complete || report.FailureReason == "" || report.PausedReason != "" {
+						t.Fatalf("denial was not terminal: %+v", report)
+					}
+					pool.retryDue(time.Now().Add(time.Hour))
+					if len(pool.jobCh) != 0 {
+						t.Fatal("denied job was retried")
+					}
+					return
+				}
+				if report.Failed || report.Complete || report.FailureReason != "" || report.PausedReason == "" ||
+					report.Object != "" || report.Objects != 0 || report.ObjectsFailed != 0 {
+					t.Fatalf("outage was not saved with a safe restart point: %+v", report)
+				}
+				pool.jmu.Lock()
+				owner := pool.jobOwners[job.ID]
+				sameOwner := owner.req == job && owner.retries == 1
+				pool.jmu.Unlock()
+				if !sameOwner {
+					t.Fatal("pause lost the job owner")
+				}
+				outage.Store(false)
+				pool.retryDue(time.Now().Add(time.Hour))
+				waitBatchState(t, pool, job.ID, "terminal")
+				report = fixture.jobInfo(t, job)
+				if !report.Complete || report.Failed || report.PausedReason != "" || report.Objects != int64(len(keys)) || report.ObjectsFailed != 0 {
+					t.Fatalf("retry did not finish with consistent counts: %+v", report)
+				}
+				writtenMu.Lock()
+				for _, key := range keys {
+					if !written[key] {
+						t.Errorf("retry skipped %s", key)
+					}
+				}
+				writtenMu.Unlock()
+				if mode == "pull" {
+					for _, key := range keys {
+						if !batchVersionExists(t, fixture.obj, fixture.bucket, key, versions[key]) {
+							t.Errorf("retry lost remote version of %s", key)
+						}
+					}
+				}
+			})
+		}
+	}
+}
@@ -0,0 +1,124 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/auth"
+)
+
+// Edits that keep the credential keep a job running; edits of the
+// credential itself, and its status, still end it.
+func TestBatchSubmitterCredentialEdits(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	batchAuthzPolicy(f.ctx, t, f.sys, "edit-parent", batchAuthzFullAccess)
+	parent := iamAdminUser(f.ctx, t, f.sys, "edit-parent", "edit-parent")
+	soon := UTCNow().Add(time.Hour)
+	later := UTCNow().Add(24 * time.Hour)
+	newService := func(t *testing.T, expiring bool) (auth.Credentials, *batchJobSubmitter) {
+		t.Helper()
+		var opts newServiceAccountOpts
+		if expiring {
+			opts.expiration = &soon
+		}
+		created, _, err := f.sys.NewServiceAccount(f.ctx, parent.AccessKey, nil, opts)
+		mustIAM(t, err)
+		stored, ok := f.sys.GetUser(f.ctx, created.AccessKey)
+		if !ok {
+			t.Fatal("service account missing")
+		}
+		sub, err := newBatchJobSubmitter(stored.Credentials)
+		mustIAM(t, err)
+		mustIAM(t, sub.authorize(f.ctx, batchContextRead(f)))
+		return stored.Credentials, sub
+	}
+	for _, tc := range []struct {
+		name     string
+		expiring bool
+		edit     func(t *testing.T, cred auth.Credentials)
+		keep     bool
+	}{
+		{"description", false, func(t *testing.T, cred auth.Credentials) {
+			_, err := f.sys.UpdateServiceAccount(f.ctx, cred.AccessKey, updateServiceAccountOpts{description: "cosmetic"})
+			mustIAM(t, err)
+		}, true},
+		{"name", false, func(t *testing.T, cred auth.Credentials) {
+			_, err := f.sys.UpdateServiceAccount(f.ctx, cred.AccessKey, updateServiceAccountOpts{name: "renamed"})
+			mustIAM(t, err)
+		}, true},
+		{"expiry-extended", true, func(t *testing.T, cred auth.Credentials) {
+			_, err := f.sys.UpdateServiceAccount(f.ctx, cred.AccessKey, updateServiceAccountOpts{expiration: &later})
+			mustIAM(t, err)
+		}, true},
+		// An expiry makes a service account a temporary credential, which
+		// IsAllowed evaluates differently.
+		{"expiry-added", false, func(t *testing.T, cred auth.Credentials) {
+			_, err := f.sys.UpdateServiceAccount(f.ctx, cred.AccessKey, updateServiceAccountOpts{expiration: &later})
+			mustIAM(t, err)
+		}, false},
+		{"groups-refreshed", false, func(t *testing.T, cred auth.Credentials) {
+			// As the LDAP group refresh rewrites a derived credential.
+			u, ok := f.sys.GetUser(f.ctx, cred.AccessKey)
+			if !ok {
+				t.Fatal("service account missing")
+			}
+			wrote, err := f.sys.store.RefreshUserGroups(f.ctx, u.Credentials, append(u.Credentials.Groups, "refreshed-group"))
+			mustIAM(t, err)
+			if !wrote {
+				t.Fatal("control: the refresh did not write")
+			}
+		}, true},
+		{"secret-rotated", false, func(t *testing.T, cred auth.Credentials) {
+			_, err := f.sys.UpdateServiceAccount(f.ctx, cred.AccessKey, updateServiceAccountOpts{secretKey: "edit-rotated-secret-key"})
+			mustIAM(t, err)
+		}, false},
+		{"disabled", false, func(t *testing.T, cred auth.Credentials) {
+			_, err := f.sys.UpdateServiceAccount(f.ctx, cred.AccessKey, updateServiceAccountOpts{status: string(madmin.AccountDisabled)})
+			mustIAM(t, err)
+		}, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			cred, sub := newService(t, tc.expiring)
+			before, _ := f.sys.GetUser(f.ctx, cred.AccessKey)
+			time.Sleep(2 * time.Millisecond) // a distinct UpdatedAt
+			tc.edit(t, cred)
+			after, _ := f.sys.GetUser(f.ctx, cred.AccessKey)
+			if after.UpdatedAt.Equal(before.UpdatedAt) {
+				t.Fatal("control: the edit did not change the identity record")
+			}
+			err := sub.authorize(f.ctx, batchContextRead(f))
+			if tc.keep && err != nil {
+				t.Errorf("an edit that keeps the credential ended the job: %v", err)
+			}
+			if !tc.keep && !isBatchJobAuthzError(err) {
+				t.Errorf("an edit of the credential must end the job, got %v", err)
+			}
+		})
+	}
+}
+
+// A user deleted and created again under its access key, with the same
+// secret, is not the credential that submitted the job.
+func TestBatchSubmitterRecreatedSameSecret(t *testing.T) {
+	f := newBatchAuthzFixture(t)
+	batchAuthzPolicy(f.ctx, t, f.sys, "recreate-full", batchAuthzFullAccess)
+	cred := iamAdminUser(f.ctx, t, f.sys, "recreate-user", "recreate-full")
+	sub, err := newBatchJobSubmitter(cred)
+	mustIAM(t, err)
+	mustIAM(t, sub.authorize(f.ctx, batchContextRead(f)))
+	mustIAM(t, f.sys.DeleteUser(f.ctx, cred.AccessKey, false))
+	time.Sleep(10 * time.Millisecond)
+	again := iamAdminUser(f.ctx, t, f.sys, cred.AccessKey, "recreate-full")
+	if again.SecretKey != cred.SecretKey {
+		t.Fatal("control: the user must be created again with the same secret")
+	}
+	err = sub.authorize(f.ctx, batchContextRead(f))
+	t.Logf("recreated: %v", err)
+	if !isBatchJobAuthzError(err) {
+		t.Errorf("a recreated user must not run the job of the deleted one: %v", err)
+	}
+}
@@ -494,6 +494,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 	}

 	deleteResults := make([]deleteResult, len(deleteObjectsReq.Objects))
+	duplicateOf := make(map[int]int)

 	vc, _ := globalBucketVersioningSys.Get(bucket)
 	oss := make([]*objSweeper, len(deleteObjectsReq.Objects))
@@ -523,13 +524,32 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 			}
 		}

+		// All deletes on directory objects needs to be for `nullVersionID`.
+		// Decide this before the lookup below, so that the retention check
+		// reads the version the delete removes.
+		if isDirObject(object.ObjectName) && object.VersionID == "" {
+			object.VersionID = nullVersionID
+		}
+
 		opts := ObjectOptions{
 			VersionID:        object.VersionID,
 			Versioned:        vc.PrefixEnabled(object.ObjectName),
 			VersionSuspended: vc.Suspended(),
 		}

-		if replicateDeletes || object.VersionID != "" && hasLockEnabled || !globalTierConfigMgr.Empty() {
+		// Physical version deletes in a locked bucket are checked by the
+		// single-object path below, while its mutation lock is held. A lookup
+		// here could miss a later protection update or a protected pool copy.
+		if hasLockEnabled && object.VersionID != "" {
+			if first, ok := objectsToDelete[object]; ok {
+				duplicateOf[index] = first
+			} else {
+				objectsToDelete[object] = index
+			}
+			continue
+		}
+
+		if replicateDeletes || !globalTierConfigMgr.Empty() {
 			if !globalTierConfigMgr.Empty() && object.VersionID == "" && opts.VersionSuspended {
 				opts.VersionID = nullVersionID
 			}
@@ -541,11 +561,6 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 			oss[index].SetTransitionState(goi.TransitionedObject)
 		}

-		// All deletes on directory objects needs to be for `nullVersionID`
-		if isDirObject(object.ObjectName) && object.VersionID == "" {
-			object.VersionID = nullVersionID
-		}
-
 		if replicateDeletes {
 			dsc = checkReplicateDelete(ctx, bucket, ObjectToDelete{
 				ObjectV: ObjectV{
@@ -563,21 +578,10 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 				object.ReplicateDecisionStr = dsc.String()
 			}
 		}
-		if object.VersionID != "" && hasLockEnabled {
-			if err := enforceRetentionBypassForDelete(ctx, r, bucket, object, goi, gerr); err != nil {
-				apiErr := toAPIError(ctx, err)
-				deleteResults[index].errInfo = DeleteError{
-					Code:      apiErr.Code,
-					Message:   apiErr.Description,
-					Key:       object.ObjectName,
-					VersionID: object.VersionID,
-				}
-				continue
-			}
-		}
-
 		// Avoid duplicate objects, we use map to filter them out.
-		if _, ok := objectsToDelete[object]; !ok {
+		if first, ok := objectsToDelete[object]; ok {
+			duplicateOf[index] = first
+		} else {
 			objectsToDelete[object] = index
 		}
 	}
@@ -596,10 +600,57 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 	ctx = bgContext(ctx)

 	deleteList := toNames(objectsToDelete)
-	dObjects, errs := deleteObjectsFn(ctx, bucket, deleteList, ObjectOptions{
-		PrefixEnabledFn:  vc.PrefixEnabled,
-		VersionSuspended: vc.Suspended(),
-	})
+	dObjects := make([]DeletedObject, len(deleteList))
+	errs := make([]error, len(deleteList))
+	bulk := make([]ObjectToDelete, 0, len(deleteList))
+	bulkIndex := make([]int, 0, len(deleteList))
+	for i, object := range deleteList {
+		if !hasLockEnabled || object.VersionID == "" {
+			bulk = append(bulk, object)
+			bulkIndex = append(bulkIndex, i)
+			continue
+		}
+		reqInfo.ObjectName, reqInfo.VersionID = object.ObjectName, object.VersionID
+		opts := ObjectOptions{
+			VersionID:        object.VersionID,
+			Versioned:        vc.PrefixEnabled(object.ObjectName),
+			VersionSuspended: vc.Suspended(),
+		}
+		index := objectsToDelete[object]
+		if !globalTierConfigMgr.Empty() {
+			oss[index] = newObjSweeper(bucket, object.ObjectName).WithVersion(object.VersionID).WithVersioning(opts.Versioned, opts.VersionSuspended)
+		}
+		opts.SetEvalRetentionBypassFn(func(oi ObjectInfo, err error) error {
+			return enforceRetentionBypassForDelete(ctx, r, bucket, object, oi, err)
+		})
+		opts.SetEvalMetadataFn(func(oi *ObjectInfo, err error) (ReplicateDecision, error) {
+			if err == nil && oss[index] != nil {
+				oss[index].SetTransitionState(oi.TransitionedObject)
+			}
+			if replicateDeletes {
+				return checkReplicateDelete(ctx, bucket, object, *oi, opts, err), nil
+			}
+			return ReplicateDecision{}, nil
+		})
+		// DeleteObject merges protection across pools before evaluating these
+		// callbacks and holds the same lock as retention and legal-hold writes.
+		oi, err := objectAPI.DeleteObject(ctx, bucket, object.ObjectName, opts)
+		dObjects[i] = DeletedObject{
+			ObjectName:       object.ObjectName,
+			VersionID:        object.VersionID,
+			ReplicationState: oi.ReplicationState(),
+		}
+		errs[i] = err
+	}
+	if len(bulk) > 0 {
+		deleted, bulkErrs := deleteObjectsFn(ctx, bucket, bulk, ObjectOptions{
+			PrefixEnabledFn:  vc.PrefixEnabled,
+			VersionSuspended: vc.Suspended(),
+		})
+		for i, index := range bulkIndex {
+			dObjects[index], errs[index] = deleted[i], bulkErrs[i]
+		}
+	}

 	// Are all objects saying bucket not found?
 	if isAllBucketsNotFound(errs) {
@@ -608,23 +659,11 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 	}

 	for i := range errs {
-		// DeleteMarkerVersionID is not used specifically to avoid
-		// lookup errors, since DeleteMarkerVersionID is only
-		// created during DeleteMarker creation when client didn't
-		// specify a versionID.
-		objToDel := ObjectToDelete{
-			ObjectV: ObjectV{
-				ObjectName: decodeDirObject(dObjects[i].ObjectName),
-				VersionID:  dObjects[i].VersionID,
-			},
-			VersionPurgeStatus:            dObjects[i].VersionPurgeStatus(),
-			VersionPurgeStatuses:          dObjects[i].ReplicationState.VersionPurgeStatusInternal,
-			DeleteMarkerReplicationStatus: dObjects[i].ReplicationState.ReplicationStatusInternal,
-			ReplicateDecisionStr:          dObjects[i].ReplicationState.ReplicateDecisionStr,
-		}
-		dindex := objectsToDelete[objToDel]
+		// Match by the input identity: replication decisions may have changed
+		// under the mutation lock, and a missing version has no stored identity.
+		dindex := objectsToDelete[deleteList[i]]
 		if errs[i] == nil || isErrObjectNotFound(errs[i]) || isErrVersionNotFound(errs[i]) {
-			if replicateDeletes {
+			if replicateDeletes && (!hasLockEnabled || deleteList[i].VersionID == "") {
 				dObjects[i].ReplicationState = deleteList[i].ReplicationState()
 			}
 			deleteResults[dindex].delInfo = dObjects[i]
@@ -638,20 +677,28 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 			VersionID: deleteList[i].VersionID,
 		}
 	}
+	for index, first := range duplicateOf {
+		deleteResults[index] = deleteResults[first]
+	}

 	// Generate response
 	deleteErrors := make([]DeleteError, 0, len(deleteObjectsReq.Objects))
 	deletedObjects := make([]DeletedObject, 0, len(deleteObjectsReq.Objects))
-	for _, deleteResult := range deleteResults {
+	eventObjects := make([]DeletedObject, 0, len(deleteObjectsReq.Objects))
+	for index, deleteResult := range deleteResults {
+		// Storage normalizes a directory to the null version. The response
+		// retains the caller's version identity, including an explicit null.
+		if isDirObject(deleteObjectsReq.Objects[index].ObjectName) {
+			deleteResult.delInfo.VersionID = deleteObjectsReq.Objects[index].VersionID
+			deleteResult.errInfo.VersionID = deleteObjectsReq.Objects[index].VersionID
+		}
 		if deleteResult.errInfo.Code != "" {
 			deleteErrors = append(deleteErrors, deleteResult.errInfo)
 		} else {
-			// All deletes on directory objects was with `nullVersionID`.
-			// Remove it from response.
-			if isDirObject(deleteResult.delInfo.ObjectName) && deleteResult.delInfo.VersionID == nullVersionID {
-				deleteResult.delInfo.VersionID = ""
-			}
 			deletedObjects = append(deletedObjects, deleteResult.delInfo)
+			if _, duplicate := duplicateOf[index]; !duplicate {
+				eventObjects = append(eventObjects, deleteResult.delInfo)
+			}
 		}
 	}

@@ -660,7 +707,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,

 	// Write success response.
 	writeSuccessResponseXML(w, encodedSuccessResponse)
-	for _, dobj := range deletedObjects {
+	for _, dobj := range eventObjects {
 		if dobj.ObjectName == "" {
 			continue
 		}
@@ -703,8 +750,8 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 	}

 	// Clean up transitioned objects from remote tier
-	for _, os := range oss {
-		if os == nil { // skip objects that weren't deleted due to invalid versionID etc.
+	for index, os := range oss {
+		if os == nil || deleteResults[index].errInfo.Code != "" { // Sweep only successful deletions.
 			continue
 		}
 		os.Sweep()
@@ -119,7 +119,10 @@ func (a *ActiveWorkerStat) update() {
 	if a == nil {
 		return
 	}
-	a.Curr = globalReplicationPool.Get().ActiveWorkers()
+	a.Curr = 0
+	if pool := globalReplicationPool.Get(); pool != nil {
+		a.Curr = pool.ActiveWorkers()
+	}
 	a.hist.Update(int64(a.Curr))
 	a.Avg = float32(a.hist.Mean())
 	a.Max = int(a.hist.Max())
@@ -0,0 +1,32 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package cmd
+
+import (
+	"testing"
+
+	"github.com/minio/minio/internal/once"
+	"github.com/rcrowley/go-metrics"
+)
+
+func TestReplicationWorkerMetricsWithoutPool(t *testing.T) {
+	previous := globalReplicationPool
+	t.Cleanup(func() { globalReplicationPool = previous })
+
+	globalReplicationPool = once.NewSingleton[ReplicationPool]()
+	globalReplicationPool.Set(&ReplicationPool{activeWorkers: 8})
+	stat := newActiveWorkerStat(metrics.NewRegistry())
+	stat.update()
+	if got := stat.get(); got.Curr != 8 || got.Avg != 8 || got.Max != 8 {
+		t.Fatalf("active pool metrics = %+v, want current/average/maximum 8", got)
+	}
+
+	// A collector may still run after replication has been disabled.
+	globalReplicationPool = once.NewSingleton[ReplicationPool]()
+	globalReplicationPool.Set(nil)
+	stat.update()
+	if got := stat.get(); got.Curr != 0 || got.Avg != 4 || got.Max != 8 {
+		t.Fatalf("disabled pool metrics = %+v, want current 0, average 4, maximum 8", got)
+	}
+}
@@ -0,0 +1,346 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"encoding/xml"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	minio "github.com/minio/minio-go/v7"
+	"github.com/minio/minio/internal/bucket/replication"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/mux"
+)
+
+// Schedule a protection update immediately before the storage mutation. Both
+// public delete entry points are gated so the test is independent of whether
+// the handler batches a version or uses the single-object deletion path.
+type bulkDeleteLockBarrier struct {
+	ObjectLayer
+	entered chan struct{}
+	release chan struct{}
+	once    sync.Once
+}
+
+func (g *bulkDeleteLockBarrier) DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) {
+	g.wait()
+	return g.ObjectLayer.DeleteObjects(ctx, bucket, objects, opts)
+}
+
+func (g *bulkDeleteLockBarrier) wait() {
+	g.once.Do(func() { close(g.entered); <-g.release })
+}
+
+func (g *bulkDeleteLockBarrier) DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
+	g.wait()
+	return g.ObjectLayer.DeleteObject(ctx, bucket, object, opts)
+}
+
+func bulkDeleteLockRequest(t *testing.T, bucket, key, version string) *http.Request {
+	t.Helper()
+	body := encodeResponse(DeleteObjectsRequest{Objects: []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key, VersionID: version}}}})
+	req, err := newTestSignedRequestV4(http.MethodPost, getDeleteMultipleObjectsURL("", bucket), int64(len(body)), bytes.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return mux.SetURLVars(req, map[string]string{"bucket": bucket})
+}
+
+func TestBulkDeleteObjectLockInterleaving(t *testing.T) {
+	for _, pools := range []int{1, 2} {
+		for _, directory := range []bool{false, true} {
+			for _, kind := range []string{"legal-hold", "retention"} {
+				t.Run(fmt.Sprintf("pools=%d/directory=%v/%s", pools, directory, kind), func(t *testing.T) {
+					z, _ := bulkDeletePools(t, pools)
+					bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true})
+					if err != nil {
+						t.Fatal(err)
+					}
+					key := "protection-update"
+					if directory {
+						key += "/"
+					}
+					var oi ObjectInfo
+					if directory {
+						oi, err = z.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewReader(nil), 0, "", ""), ObjectOptions{Versioned: true, VersionID: nullVersionID})
+						if err != nil {
+							t.Fatal(err)
+						}
+					} else {
+						oi = batchPutVersion(t, z, bucket, key, nil)
+					}
+					gate := &bulkDeleteLockBarrier{ObjectLayer: z, entered: make(chan struct{}), release: make(chan struct{})}
+					api := objectAPIHandlers{ObjectAPI: func() ObjectLayer { return gate }}
+					requestVersion := oi.VersionID
+					if directory {
+						requestVersion = ""
+					}
+					req := bulkDeleteLockRequest(t, bucket, key, requestVersion)
+					rec := httptest.NewRecorder()
+					done := make(chan struct{})
+					go func() { defer close(done); api.DeleteMultipleObjectsHandler(rec, req) }()
+					var release sync.Once
+					unblock := func() { release.Do(func() { close(gate.release) }) }
+					defer func() {
+						unblock()
+						select {
+						case <-done:
+						case <-time.After(10 * time.Second):
+							t.Error("delete did not finish")
+						}
+					}()
+					select {
+					case <-gate.entered:
+					case <-time.After(10 * time.Second):
+						t.Fatal("bulk handler did not reach deletion after validation")
+					}
+					body := []byte(`<LegalHold xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>ON</Status></LegalHold>`)
+					if kind == "retention" {
+						body = retentionBody("COMPLIANCE", UTCNow().Add(24*time.Hour))
+					}
+					holdReq, err := newTestSignedRequestV4(http.MethodPut, "/"+bucket+"/"+key+"?"+kind+"&versionId="+url.QueryEscape(oi.VersionID), int64(len(body)), bytes.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+					if err != nil {
+						t.Fatal(err)
+					}
+					holdRec := httptest.NewRecorder()
+					router.ServeHTTP(holdRec, holdReq)
+					protected, getErr := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: oi.VersionID})
+					unblock()
+					select {
+					case <-done:
+					case <-time.After(10 * time.Second):
+						t.Fatal("delete did not finish")
+					}
+					if holdRec.Code != http.StatusOK || getErr != nil {
+						t.Fatalf("protection setup failed: status=%d body=%s err=%v", holdRec.Code, holdRec.Body.String(), getErr)
+					}
+					if kind == "legal-hold" && protected.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] != "ON" {
+						t.Fatal("legal hold was not stored")
+					}
+					if kind == "retention" && protected.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] != "COMPLIANCE" {
+						t.Fatal("retention was not stored")
+					}
+					var response DeleteObjectsResponse
+					if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
+						t.Fatal(err)
+					}
+					exists := batchVersionExists(t, z, bucket, key, oi.VersionID)
+					t.Logf("protection PUT=%d; delete POST=%d response=%s; protected_version_exists=%v", holdRec.Code, rec.Code, rec.Body.String(), exists)
+					if rec.Code != http.StatusOK || !exists || len(response.Errors) != 1 || len(response.DeletedObjects) != 0 {
+						t.Errorf("bulk deletion removed a version after its protection update succeeded")
+					}
+				})
+			}
+		}
+	}
+}
+
+func TestBulkDeleteObjectLockPoolCopies(t *testing.T) {
+	for _, kind := range []string{"COMPLIANCE", "LEGALHOLD"} {
+		t.Run(kind, func(t *testing.T) {
+			z, _ := bulkDeletePools(t, 2)
+			bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true})
+			if err != nil {
+				t.Fatal(err)
+			}
+			for _, bulk := range []bool{false, true} {
+				key := "single-control"
+				if bulk {
+					key = "bulk"
+				}
+				id := mustGetUUID()
+				mtime := UTCNow().Add(-time.Hour)
+				putConsistencyObject(t, z, bucket, key, 0, "same-data", ObjectOptions{Versioned: true, VersionID: id, MTime: mtime, UserDefined: nil})
+				putConsistencyObject(t, z, bucket, key, 1, "same-data", ObjectOptions{Versioned: true, VersionID: id, MTime: mtime, UserDefined: batchLockMeta(kind)})
+				selected, err := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: id})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if selected.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] != "" || selected.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] != "" {
+					t.Fatal("fixture did not select the unprotected pool copy")
+				}
+				req := bulkDeleteLockRequest(t, bucket, key, id)
+				if !bulk {
+					req, err = newTestSignedRequestV4(http.MethodDelete, getDeleteObjectURL("", bucket, key)+"?versionId="+id, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+					if err != nil {
+						t.Fatal(err)
+					}
+				}
+				rec := httptest.NewRecorder()
+				router.ServeHTTP(rec, req)
+				if !strings.Contains(rec.Body.String(), "Object is WORM protected") {
+					t.Fatalf("expected Object Lock refusal: HTTP %d: %s", rec.Code, rec.Body.String())
+				}
+				remaining := 0
+				for _, pool := range z.serverPools {
+					_, err := pool.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: id})
+					if err == nil {
+						remaining++
+					} else if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+						t.Fatal(err)
+					}
+				}
+				t.Logf("bulk=%v HTTP=%d response=%s remaining_pool_copies=%d", bulk, rec.Code, rec.Body.String(), remaining)
+				if remaining != 2 {
+					t.Errorf("protected version copies removed by bulk=%v", bulk)
+				}
+			}
+		})
+	}
+}
+
+// One response must retain the identity and outcome of every requested item,
+// even when physical version removals and marker creation take different paths.
+func TestBulkDeleteObjectLockMixedResponse(t *testing.T) {
+	for _, pools := range []int{1, 2} {
+		for _, quiet := range []bool{false, true} {
+			t.Run(fmt.Sprintf("pools=%d/quiet=%v", pools, quiet), func(t *testing.T) {
+				z, _ := bulkDeletePools(t, pools)
+				bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+				locked := batchPutVersion(t, z, bucket, "locked", batchLockMeta("COMPLIANCE"))
+				free := batchPutVersion(t, z, bucket, "free", nil)
+				marker := batchPutVersion(t, z, bucket, "marker", batchLockMeta("LEGALHOLD"))
+				if _, err := z.PutObject(t.Context(), bucket, "directory/", mustGetPutObjReader(t, bytes.NewReader(nil), 0, "", ""), ObjectOptions{Versioned: true, VersionID: nullVersionID}); err != nil {
+					t.Fatal(err)
+				}
+				missing := mustGetUUID()
+				objects := []ObjectToDelete{
+					{ObjectV: ObjectV{ObjectName: "locked", VersionID: locked.VersionID}},
+					{ObjectV: ObjectV{ObjectName: "free", VersionID: free.VersionID}},
+					{ObjectV: ObjectV{ObjectName: "missing", VersionID: missing}},
+					{ObjectV: ObjectV{ObjectName: "marker"}},
+					{ObjectV: ObjectV{ObjectName: "directory/"}},
+					{ObjectV: ObjectV{ObjectName: "directory/", VersionID: nullVersionID}},
+					{ObjectV: ObjectV{ObjectName: "invalid", VersionID: "invalid"}},
+				}
+				objects = append(objects, objects[0], objects[1])
+				body := encodeResponse(DeleteObjectsRequest{Quiet: quiet, Objects: objects})
+				rec := r5Request(t, router, globalActiveCred, http.MethodPost, getDeleteMultipleObjectsURL("", bucket), string(body), nil)
+				var response DeleteObjectsResponse
+				if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &response) != nil {
+					t.Fatalf("HTTP %d: %s", rec.Code, rec.Body.String())
+				}
+				wantErrors := []ObjectV{objects[0].ObjectV, objects[6].ObjectV, objects[0].ObjectV}
+				if len(response.Errors) != len(wantErrors) {
+					t.Fatalf("errors: %+v", response.Errors)
+				}
+				for i, e := range response.Errors {
+					if e.Key != wantErrors[i].ObjectName || e.VersionID != wantErrors[i].VersionID {
+						t.Errorf("error identity %d: %+v", i, e)
+					}
+				}
+				if quiet {
+					if len(response.DeletedObjects) != 0 {
+						t.Errorf("quiet successes: %+v", response.DeletedObjects)
+					}
+				} else {
+					want := []ObjectV{objects[1].ObjectV, objects[2].ObjectV, objects[3].ObjectV, objects[4].ObjectV, objects[5].ObjectV, objects[1].ObjectV}
+					if len(response.DeletedObjects) != len(want) {
+						t.Fatalf("successes: %+v", response.DeletedObjects)
+					}
+					for i, d := range response.DeletedObjects {
+						if d.ObjectName != want[i].ObjectName || d.VersionID != want[i].VersionID {
+							t.Errorf("success identity %d: %+v", i, d)
+						}
+						if i == 2 && (!d.DeleteMarker || d.DeleteMarkerVersionID == "") {
+							t.Errorf("missing delete marker: %+v", d)
+						}
+					}
+				}
+				if !batchVersionExists(t, z, bucket, "locked", locked.VersionID) || !batchVersionExists(t, z, bucket, "marker", marker.VersionID) || batchVersionExists(t, z, bucket, "free", free.VersionID) || batchVersionExists(t, z, bucket, "directory/", nullVersionID) {
+					t.Fatal("unexpected stored versions after mixed deletion")
+				}
+			})
+		}
+	}
+}
+
+func TestBulkDeleteObjectLockGovernance(t *testing.T) {
+	for _, pools := range []int{1, 2} {
+		t.Run(fmt.Sprintf("pools=%d", pools), func(t *testing.T) {
+			z, _ := bulkDeletePools(t, pools)
+			bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true})
+			if err != nil {
+				t.Fatal(err)
+			}
+			deleteOnly := newObjectAttributesAuthzUser(t, "bulk-governance", bucket, `"s3:DeleteObjectVersion"`)
+			for _, allowed := range []bool{false, true} {
+				for _, bypass := range []bool{false, true} {
+					key := fmt.Sprintf("allowed-%v-bypass-%v", allowed, bypass)
+					oi := batchPutVersion(t, z, bucket, key, batchLockMeta("GOVERNANCE"))
+					cred := deleteOnly
+					if allowed {
+						cred = globalActiveCred
+					}
+					body := encodeResponse(DeleteObjectsRequest{Objects: []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key, VersionID: oi.VersionID}}}})
+					header := map[string]string{}
+					if bypass {
+						header[xhttp.AmzObjectLockBypassGovernance] = "true"
+					}
+					rec := r5Request(t, router, cred, http.MethodPost, getDeleteMultipleObjectsURL("", bucket), string(body), header)
+					var response DeleteObjectsResponse
+					if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &response) != nil {
+						t.Fatalf("HTTP %d: %s", rec.Code, rec.Body.String())
+					}
+					deleted := allowed && bypass
+					if batchVersionExists(t, z, bucket, key, oi.VersionID) == deleted || (len(response.Errors) == 0) != deleted {
+						t.Fatalf("%s: %s", key, rec.Body.String())
+					}
+				}
+			}
+		})
+	}
+}
+
+// The locked path computes replication decisions while holding the mutation
+// lock. Keep pending purges and their queue entries, including duplicate input.
+func TestBulkDeleteObjectLockReplication(t *testing.T) {
+	for _, pools := range []int{1, 2} {
+		t.Run(fmt.Sprintf("pools=%d", pools), func(t *testing.T) {
+			z, _ := bulkDeletePools(t, pools)
+			bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true})
+			if err != nil {
+				t.Fatal(err)
+			}
+			client, err := minio.New("localhost:1", &minio.Options{Region: "us-east-1", MaxRetries: 1})
+			if err != nil {
+				t.Fatal(err)
+			}
+			worker, cleanup := r5ReplicationFixture(t, z, bucket, client)
+			defer cleanup()
+			free := batchPutVersion(t, z, bucket, "free", nil)
+			locked := batchPutVersion(t, z, bucket, "locked", batchLockMeta("LEGALHOLD"))
+			objects := []ObjectToDelete{{ObjectV: ObjectV{ObjectName: "free", VersionID: free.VersionID}}, {ObjectV: ObjectV{ObjectName: "locked", VersionID: locked.VersionID}}}
+			objects = append(objects, objects[0])
+			body := encodeResponse(DeleteObjectsRequest{Objects: objects})
+			rec := r5Request(t, router, globalActiveCred, http.MethodPost, getDeleteMultipleObjectsURL("", bucket), string(body), nil)
+			var response DeleteObjectsResponse
+			if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &response) != nil || len(response.Errors) != 1 || len(response.DeletedObjects) != 2 {
+				t.Fatalf("HTTP %d: %s", rec.Code, rec.Body.String())
+			}
+			if len(worker) != 1 {
+				t.Fatalf("want one purge queued, got %d", len(worker))
+			}
+			op, ok := (<-worker).(DeletedObjectReplicationInfo)
+			if !ok || op.ObjectName != "free" || op.VersionID != free.VersionID || op.VersionPurgeStatus() != replication.VersionPurgePending || op.ReplicationState.ReplicateDecisionStr == "" {
+				t.Fatalf("lost purge identity/state: %+v", op)
+			}
+			if !batchVersionExists(t, z, bucket, "locked", locked.VersionID) {
+				t.Fatal("protected version was deleted")
+			}
+		})
+	}
+}
@@ -0,0 +1,57 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"bytes"
+	"encoding/xml"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/minio/minio/internal/auth"
+)
+
+// A directory object always has the null version, and a bulk delete without a
+// version ID removes it. The retention check must read that version's lock.
+func TestBulkDeleteDirectoryObjectLock(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t: t, endpoints: []string{"DeleteMultipleObjects"},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+			enableBucketObjectLock(t, bucket)
+			for _, kind := range append([]string{"unlocked"}, batchLockKinds...) {
+				key := "dir-" + kind + "/"
+				if _, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewReader(nil), 0, "", ""),
+					ObjectOptions{Versioned: true, VersionID: nullVersionID, UserDefined: batchLockMeta(kind)}); err != nil {
+					t.Fatal(err)
+				}
+				body := encodeResponse(DeleteObjectsRequest{Objects: []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key}}}})
+				req, err := newTestSignedRequestV4(http.MethodPost, getDeleteMultipleObjectsURL("", bucket), int64(len(body)), bytes.NewReader(body),
+					cred.AccessKey, cred.SecretKey, nil)
+				if err != nil {
+					t.Fatal(err)
+				}
+				rec := httptest.NewRecorder()
+				router.ServeHTTP(rec, req)
+				var resp DeleteObjectsResponse
+				if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &resp) != nil {
+					t.Fatalf("%s: %s: status %d: %s", instance, key, rec.Code, rec.Body.String())
+				}
+				exists := batchVersionExists(t, obj, bucket, key, nullVersionID)
+				if kind == "unlocked" {
+					if exists || len(resp.DeletedObjects) != 1 {
+						t.Errorf("%s: %s: unlocked directory object was not deleted: %s", instance, key, rec.Body.String())
+					}
+					continue
+				}
+				if !exists || len(resp.Errors) != 1 || len(resp.DeletedObjects) != 0 ||
+					resp.Errors[0].Key != key || !strings.Contains(resp.Errors[0].Message, "WORM") {
+					t.Errorf("%s: %s: want the WORM error and the object kept (exists=%v): %s", instance, key, exists, rec.Body.String())
+				}
+			}
+		},
+	})
+}
@@ -0,0 +1,226 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"slices"
+	"testing"
+
+	"github.com/minio/minio/internal/auth"
+)
+
+// The LDAP group refresh computes groups from a snapshot of a server's cache
+// taken before a slow directory lookup, then writes them. These tests take
+// such a snapshot, change the credential the way an administrator would on
+// this or another server, and run the write: it must change only the
+// groups of what is stored now, and never undo the change.
+
+type groupRefreshFixture struct {
+	ctx    context.Context
+	sys    *IAMSys
+	obj    ObjectLayer
+	parent auth.Credentials
+}
+
+func newGroupRefreshFixture(t *testing.T) *groupRefreshFixture {
+	t.Helper()
+	ctx, sys, obj := prepareIAMRevisionFixture(t)
+	batchAuthzPolicy(ctx, t, sys, "refresh-parent", batchAuthzFullAccess)
+	parent := iamAdminUser(ctx, t, sys, "refresh-parent", "refresh-parent")
+	return &groupRefreshFixture{ctx: ctx, sys: sys, obj: obj, parent: parent}
+}
+
+// service creates a service account of the fixture's parent and returns
+// its stored credential, the snapshot a refresh would take.
+func (f *groupRefreshFixture) service(t *testing.T, accessKey string) auth.Credentials {
+	t.Helper()
+	_, _, err := f.sys.NewServiceAccount(f.ctx, f.parent.AccessKey, nil, newServiceAccountOpts{accessKey: accessKey, secretKey: "refresh-original-secret"})
+	mustIAM(t, err)
+	u, ok := f.sys.store.GetUser(accessKey)
+	if !ok {
+		t.Fatal("service account missing")
+	}
+	return u.Credentials
+}
+
+// sibling is another server: the same storage, its own cache.
+func (f *groupRefreshFixture) sibling(t *testing.T) *IAMStoreSys {
+	t.Helper()
+	s := &IAMStoreSys{IAMStorageAPI: newIAMObjectStore(f.obj, f.sys.usersSysType)}
+	mustIAM(t, s.LoadIAMCache(f.ctx, false))
+	return s
+}
+
+// stored is the record as a server that starts now loads it.
+func (f *groupRefreshFixture) stored(t *testing.T, accessKey string) (UserIdentity, bool) {
+	t.Helper()
+	return f.sibling(t).GetUser(accessKey)
+}
+
+func TestIAMGroupRefreshKeepsRotation(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "refresh-rotate")
+	_, err := f.sys.UpdateServiceAccount(f.ctx, snapshot.AccessKey, updateServiceAccountOpts{secretKey: "refresh-rotated-secret", description: "rotated"})
+	mustIAM(t, err)
+
+	wrote, err := f.sys.store.RefreshUserGroups(f.ctx, snapshot, []string{"g1"})
+	mustIAM(t, err)
+	if !wrote {
+		t.Fatal("a refresh of a live credential must write its new groups")
+	}
+	check := func(where string, u UserIdentity, ok bool) {
+		t.Helper()
+		switch {
+		case !ok:
+			t.Fatalf("%s: credential missing", where)
+		case u.Credentials.SecretKey != "refresh-rotated-secret":
+			t.Errorf("%s: the refresh undid the rotation", where)
+		case u.Credentials.Description != "rotated":
+			t.Errorf("%s: the refresh undid an edit", where)
+		case !slices.Equal(u.Credentials.Groups, []string{"g1"}):
+			t.Errorf("%s: groups %v, want [g1]", where, u.Credentials.Groups)
+		}
+	}
+	u, ok := f.sys.store.GetUser(snapshot.AccessKey)
+	check("cache", u, ok)
+	u, ok = f.stored(t, snapshot.AccessKey)
+	check("storage", u, ok)
+	if _, valid, _ := f.sys.CheckKey(f.ctx, snapshot.AccessKey); !valid {
+		t.Error("the rotated credential must stay valid")
+	}
+}
+
+func TestIAMGroupRefreshKeepsDeletion(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "refresh-delete")
+	mustIAM(t, f.sys.DeleteServiceAccount(f.ctx, snapshot.AccessKey, false))
+
+	wrote, err := f.sys.store.RefreshUserGroups(f.ctx, snapshot, []string{"g1"})
+	mustIAM(t, err)
+	if wrote {
+		t.Error("a refresh wrote a deleted credential")
+	}
+	if _, ok := f.sys.store.GetUser(snapshot.AccessKey); ok {
+		t.Error("a refresh brought a deleted credential back into the cache")
+	}
+	if _, ok := f.stored(t, snapshot.AccessKey); ok {
+		t.Error("a refresh brought a deleted credential back in storage")
+	}
+}
+
+// Another server changed the credential and this server's cache has not
+// heard of it: its snapshot and its cache are both stale.
+func TestIAMGroupRefreshStaleSiblingCache(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		change func(t *testing.T, f *groupRefreshFixture, accessKey string)
+		check  func(t *testing.T, u UserIdentity, ok bool)
+	}{
+		{"rotated", func(t *testing.T, f *groupRefreshFixture, accessKey string) {
+			_, err := f.sys.UpdateServiceAccount(f.ctx, accessKey, updateServiceAccountOpts{secretKey: "refresh-rotated-secret"})
+			mustIAM(t, err)
+		}, func(t *testing.T, u UserIdentity, ok bool) {
+			if !ok || u.Credentials.SecretKey != "refresh-rotated-secret" {
+				t.Errorf("a stale server undid the rotation of another (present=%v)", ok)
+			}
+		}},
+		{"deleted", func(t *testing.T, f *groupRefreshFixture, accessKey string) {
+			mustIAM(t, f.sys.DeleteServiceAccount(f.ctx, accessKey, false))
+		}, func(t *testing.T, u UserIdentity, ok bool) {
+			if ok {
+				t.Error("a stale server brought back a credential another deleted")
+			}
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f := newGroupRefreshFixture(t)
+			snapshot := f.service(t, "refresh-sib-"+tc.name)
+			other := f.sibling(t)
+			stale, ok := other.GetUser(snapshot.AccessKey)
+			if !ok {
+				t.Fatal("sibling fixture did not load the credential")
+			}
+			tc.change(t, f, snapshot.AccessKey)
+			if cached, _ := other.GetUser(snapshot.AccessKey); cached.Credentials.SecretKey != stale.Credentials.SecretKey {
+				t.Fatal("control: the sibling cache must not have heard of the change")
+			}
+			_, err := other.RefreshUserGroups(f.ctx, stale.Credentials, []string{"g1"})
+			mustIAM(t, err)
+			u, ok := f.stored(t, snapshot.AccessKey)
+			tc.check(t, u, ok)
+			// The refreshing server's own cache follows storage too: it holds
+			// the stored record, or nothing, so a lookup loads it.
+			u, ok = other.GetUser(snapshot.AccessKey)
+			tc.check(t, u, ok)
+		})
+	}
+}
+
+// A stored record may lack its access key; the loaders take it from the
+// path, and so does the refresh. Such a credential still gets its groups,
+// including a removal.
+func TestIAMGroupRefreshRecordWithoutAccessKey(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "refresh-nokey")
+	path := getUserIdentityPath(snapshot.AccessKey, svcUser)
+	var u UserIdentity
+	mustIAM(t, f.sys.store.loadIAMConfig(f.ctx, &u, path))
+	u.Credentials.AccessKey = ""
+	u.Credentials.Groups = []string{"admins"}
+	mustIAM(t, f.sys.store.saveIAMConfig(f.ctx, &u, path))
+	loaded, ok := f.stored(t, snapshot.AccessKey)
+	if !ok || loaded.Credentials.AccessKey != snapshot.AccessKey {
+		t.Fatal("control: the loaders must take the access key from the path")
+	}
+	wrote, err := f.sys.store.RefreshUserGroups(f.ctx, loaded.Credentials, []string{"users"})
+	mustIAM(t, err)
+	after, ok := f.stored(t, snapshot.AccessKey)
+	if !wrote || !ok || !slices.Equal(after.Credentials.Groups, []string{"users"}) {
+		t.Errorf("wrote=%v present=%v groups=%v: a record without its access key must still lose a group", wrote, ok, after.Credentials.Groups)
+	}
+}
+
+// The access key was deleted and created again. Under the same parent the
+// groups the lookup computed are that parent's, and the new record gets
+// them with everything else it has; under another parent it is left alone.
+func TestIAMGroupRefreshRecreated(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "refresh-recreate")
+	mustIAM(t, f.sys.DeleteServiceAccount(f.ctx, snapshot.AccessKey, false))
+	_, _, err := f.sys.NewServiceAccount(f.ctx, f.parent.AccessKey, nil, newServiceAccountOpts{accessKey: snapshot.AccessKey, secretKey: "refresh-original-secret", description: "recreated"})
+	mustIAM(t, err)
+	_, err = f.sys.store.RefreshUserGroups(f.ctx, snapshot, []string{"g1"})
+	mustIAM(t, err)
+	u, ok := f.stored(t, snapshot.AccessKey)
+	if !ok || u.Credentials.Description != "recreated" || !slices.Equal(u.Credentials.Groups, []string{"g1"}) {
+		t.Errorf("same parent: present=%v description=%q groups=%v, want the recreated record with [g1]", ok, u.Credentials.Description, u.Credentials.Groups)
+	}
+
+	other := iamAdminUser(f.ctx, t, f.sys, "refresh-other-parent", "refresh-parent")
+	mustIAM(t, f.sys.DeleteServiceAccount(f.ctx, snapshot.AccessKey, false))
+	_, _, err = f.sys.NewServiceAccount(f.ctx, other.AccessKey, nil, newServiceAccountOpts{accessKey: snapshot.AccessKey, secretKey: "refresh-other-secret"})
+	mustIAM(t, err)
+	wrote, err := f.sys.store.RefreshUserGroups(f.ctx, snapshot, []string{"g2"})
+	mustIAM(t, err)
+	u, ok = f.stored(t, snapshot.AccessKey)
+	if wrote || !ok || u.Credentials.ParentUser != other.AccessKey || slices.Contains(u.Credentials.Groups, "g2") {
+		t.Errorf("another parent: wrote=%v present=%v parent=%q groups=%v, want it left alone", wrote, ok, u.Credentials.ParentUser, u.Credentials.Groups)
+	}
+}
+
+// Groups the record already has are not written again.
+func TestIAMGroupRefreshUnchanged(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "refresh-unchanged")
+	_, err := f.sys.store.RefreshUserGroups(f.ctx, snapshot, []string{"g1"})
+	mustIAM(t, err)
+	before, _ := f.stored(t, snapshot.AccessKey)
+	wrote, err := f.sys.store.RefreshUserGroups(f.ctx, snapshot, []string{"g1"})
+	mustIAM(t, err)
+	after, _ := f.stored(t, snapshot.AccessKey)
+	if wrote || !after.UpdatedAt.Equal(before.UpdatedAt) {
+		t.Errorf("wrote=%v, revision %s -> %s: unchanged groups must not be written", wrote, before.UpdatedAt, after.UpdatedAt)
+	}
+}
@@ -0,0 +1,325 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"slices"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+// Hold a lazy user load after it has read storage. This makes the ordering
+// between an in-flight read and a completed credential edit deterministic.
+type iamLoadUserBlocker struct {
+	IAMStorageAPI
+	key          string
+	waiting      chan struct{}
+	release      chan struct{}
+	once         sync.Once
+	loads        atomic.Int32
+	retryWaiting chan struct{}
+	retryRelease chan struct{}
+	retryErr     error
+}
+
+func (s *iamLoadUserBlocker) loadUser(ctx context.Context, key string, userType IAMUserType, out map[string]UserIdentity) error {
+	if key == s.key && userType == regUser && s.loads.Add(1) == 2 {
+		if s.retryWaiting != nil {
+			close(s.retryWaiting)
+			select {
+			case <-s.retryRelease:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+		if s.retryErr != nil {
+			return s.retryErr
+		}
+	}
+	err := s.IAMStorageAPI.loadUser(ctx, key, userType, out)
+	if err == nil && key == s.key && userType == regUser {
+		s.once.Do(func() { close(s.waiting) })
+		select {
+		case <-s.release:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	return err
+}
+
+func TestIAMLoadUserConcurrentUpdates(t *testing.T) {
+	for _, change := range []string{"unrelated-edit", "unrelated-load", "delete", "mapping", "policy"} {
+		t.Run(change, func(t *testing.T) {
+			ctx, sys, _ := prepareIAMRevisionFixture(t)
+			const policyName = "lazy-policy"
+			batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+			cold := iamAdminUser(ctx, t, sys, "lazy-cold", policyName)
+			other := iamAdminUser(ctx, t, sys, "lazy-other", "readwrite")
+			cache := sys.store.lock()
+			delete(cache.iamUsersMap, cold.AccessKey)
+			sys.store.unlock()
+			original := sys.store.IAMStorageAPI
+			blocker := &iamLoadUserBlocker{
+				IAMStorageAPI: original, key: cold.AccessKey,
+				waiting: make(chan struct{}), release: make(chan struct{}),
+			}
+			sys.store.IAMStorageAPI = blocker
+			unblock := sync.OnceFunc(func() { close(blocker.release) })
+			done := make(chan struct{})
+			t.Cleanup(func() {
+				unblock()
+				<-done
+				sys.store.IAMStorageAPI = original
+			})
+			var valid bool
+			var loadErr error
+			go func() {
+				defer close(done)
+				_, valid, loadErr = sys.CheckKey(ctx, cold.AccessKey)
+			}()
+			select {
+			case <-blocker.waiting:
+			case <-time.After(5 * time.Second):
+				t.Fatal("cold load did not read the identity")
+			}
+			switch change {
+			case "unrelated-edit":
+				mustIAM(t, sys.SetUserSecretKey(ctx, other.AccessKey, "unrelated-rotated-secret"))
+			case "unrelated-load":
+				mustIAM(t, sys.store.LoadUser(ctx, other.AccessKey))
+			case "delete":
+				mustIAM(t, sys.DeleteUser(ctx, cold.AccessKey, false))
+			case "mapping":
+				_, err := sys.PolicyDBSet(ctx, cold.AccessKey, "readonly", regUser, false)
+				mustIAM(t, err)
+			case "policy":
+				batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+			}
+			unblock()
+			select {
+			case <-done:
+			case <-time.After(5 * time.Second):
+				t.Fatal("cold load did not finish")
+			}
+			if loadErr != nil || valid != (change != "delete") {
+				t.Fatalf("valid=%v err=%v after %s", valid, loadErr, change)
+			}
+			if blocker.loads.Load() != 2 {
+				t.Fatalf("conflicting load read the identity %d times, want 2", blocker.loads.Load())
+			}
+			if change == "mapping" {
+				names, err := sys.store.PolicyDBGet(cold.AccessKey)
+				mustIAM(t, err)
+				if !slices.Equal(names, []string{"readonly"}) {
+					t.Fatalf("restored stale mapping: %v", names)
+				}
+			}
+			if change == "policy" {
+				current, err := sys.store.GetPolicy(policyName)
+				mustIAM(t, err)
+				if current.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) {
+					t.Fatal("restored stale policy")
+				}
+			}
+		})
+	}
+}
+
+func TestIAMLoadUserRetryKeepsCachedReadsAvailable(t *testing.T) {
+	for _, unavailable := range []bool{false, true} {
+		t.Run(map[bool]string{false: "loaded", true: "unavailable"}[unavailable], func(t *testing.T) {
+			ctx, sys, _ := prepareIAMRevisionFixture(t)
+			cold := iamAdminUser(ctx, t, sys, "retry-cold", "readwrite")
+			other := iamAdminUser(ctx, t, sys, "retry-cached", "readwrite")
+			cache := sys.store.lock()
+			delete(cache.iamUsersMap, cold.AccessKey)
+			sys.store.unlock()
+			original := sys.store.IAMStorageAPI
+			blocker := &iamLoadUserBlocker{
+				IAMStorageAPI: original, key: cold.AccessKey,
+				waiting: make(chan struct{}), release: make(chan struct{}),
+				retryWaiting: make(chan struct{}), retryRelease: make(chan struct{}),
+			}
+			if unavailable {
+				blocker.retryErr = errErasureReadQuorum
+			}
+			sys.store.IAMStorageAPI = blocker
+			unblock := sync.OnceFunc(func() { close(blocker.release) })
+			unblockRetry := sync.OnceFunc(func() { close(blocker.retryRelease) })
+			done := make(chan struct{})
+			t.Cleanup(func() {
+				unblock()
+				unblockRetry()
+				<-done
+				sys.store.IAMStorageAPI = original
+			})
+			var valid bool
+			var loadErr error
+			go func() {
+				defer close(done)
+				_, valid, loadErr = sys.CheckKey(ctx, cold.AccessKey)
+			}()
+			awaitIAMPolicyReload(t, blocker.waiting)
+			mustIAM(t, sys.SetUserSecretKey(ctx, other.AccessKey, "retry-rotated-secret"))
+			unblock()
+			awaitIAMPolicyReload(t, blocker.retryWaiting)
+			readDone := make(chan struct{})
+			go func() {
+				defer close(readDone)
+				user, ok, err := sys.CheckKey(ctx, other.AccessKey)
+				if err != nil || !ok || user.Credentials.SecretKey != "retry-rotated-secret" {
+					t.Errorf("cached read failed: valid=%v err=%v", ok, err)
+				}
+			}()
+			select {
+			case <-readDone:
+			case <-time.After(time.Second):
+				unblockRetry()
+				<-readDone
+				t.Fatal("cold-load retry blocked cached authentication")
+			}
+			unblockRetry()
+			<-done
+			if unavailable {
+				if valid || !errors.Is(loadErr, errIAMNotInitialized) {
+					t.Fatalf("retry outage was not retryable: valid=%v err=%v", valid, loadErr)
+				}
+			} else if !valid || loadErr != nil {
+				t.Fatalf("retry lost the cold identity: valid=%v err=%v", valid, loadErr)
+			}
+		})
+	}
+}
+
+func TestIAMLoadUserRefreshesAssociatedRecords(t *testing.T) {
+	t.Setenv("_MINIO_IAM_GROUP_REFRESH", "on")
+	for _, change := range []string{"policy", "mapping", "group", "parent"} {
+		t.Run(change, func(t *testing.T) {
+			ctx, sys, obj := prepareIAMRevisionFixture(t)
+			const policyName = "associated-policy"
+			const groupName = "associated-group"
+			batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+			parent := iamAdminUser(ctx, t, sys, "associated-parent", policyName)
+			_, err := sys.AddUsersToGroup(ctx, groupName, []string{parent.AccessKey})
+			mustIAM(t, err)
+			key := parent.AccessKey
+			if change == "parent" {
+				key = iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "associated-service", "").AccessKey
+			}
+			gate := &iamPolicyRefreshReadGate{
+				ObjectLayer: obj, userPath: getGroupInfoPath(groupName),
+				refreshEntered: make(chan struct{}), refreshRelease: make(chan struct{}),
+			}
+			sys.store.IAMStorageAPI.(*IAMObjectStore).objAPI = gate
+			unblock := sync.OnceFunc(func() { close(gate.refreshRelease) })
+			done := make(chan struct{})
+			var loadErr error
+			t.Cleanup(func() {
+				unblock()
+				<-done
+			})
+			go func() {
+				defer close(done)
+				loadErr = sys.store.LoadUser(context.WithValue(ctx, iamPolicyRefreshContext{}, "refresh"), key)
+			}()
+			awaitIAMPolicyReload(t, gate.refreshEntered)
+			switch change {
+			case "policy":
+				batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+			case "mapping":
+				_, err = sys.PolicyDBSet(ctx, parent.AccessKey, "readonly", regUser, false)
+				mustIAM(t, err)
+			case "group":
+				_, err = sys.RemoveUsersFromGroup(ctx, groupName, []string{parent.AccessKey})
+				mustIAM(t, err)
+			case "parent":
+				mustIAM(t, sys.SetUserSecretKey(ctx, parent.AccessKey, "associated-rotated-secret"))
+			}
+			unblock()
+			select {
+			case <-done:
+				mustIAM(t, loadErr)
+			case <-time.After(5 * time.Second):
+				t.Fatal("associated-record reload did not finish")
+			}
+			switch change {
+			case "policy":
+				current, err := sys.store.GetPolicy(policyName)
+				mustIAM(t, err)
+				if current.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) {
+					t.Fatal("partial snapshot restored old policy")
+				}
+			case "mapping":
+				mapping, ok := sys.store.GetMappedPolicy(parent.AccessKey, false)
+				if !ok || mapping.Policies != "readonly" {
+					t.Fatalf("partial snapshot restored old mapping: %+v", mapping)
+				}
+			case "group":
+				cache := sys.store.rlock()
+				member := cache.iamUserGroupMemberships[parent.AccessKey].Contains(groupName)
+				sys.store.runlock()
+				if member {
+					t.Fatal("partial snapshot restored old group membership")
+				}
+			case "parent":
+				current, valid, err := sys.CheckKey(ctx, parent.AccessKey)
+				if err != nil || !valid || current.Credentials.SecretKey != "associated-rotated-secret" {
+					t.Fatal("child reload restored old parent credential")
+				}
+			}
+		})
+	}
+}
+
+func TestIAMLoadUserDoesNotRestoreRotatedCredential(t *testing.T) {
+	ctx, sys, _ := prepareIAMRevisionFixture(t)
+	user := iamAdminUser(ctx, t, sys, "load-user-race", "readwrite")
+	original := sys.store.IAMStorageAPI
+	blocker := &iamLoadUserBlocker{
+		IAMStorageAPI: original,
+		key:           user.AccessKey,
+		waiting:       make(chan struct{}),
+		release:       make(chan struct{}),
+	}
+	sys.store.IAMStorageAPI = blocker
+	t.Cleanup(func() { sys.store.IAMStorageAPI = original })
+
+	loadDone := make(chan error, 1)
+	go func() { loadDone <- sys.store.LoadUser(ctx, user.AccessKey) }()
+	select {
+	case <-blocker.waiting:
+	case <-time.After(5 * time.Second):
+		t.Fatal("LoadUser did not reach the controlled read barrier")
+	}
+
+	const rotatedSecret = "load-user-rotated-secret"
+	if err := sys.SetUserSecretKey(ctx, user.AccessKey, rotatedSecret); err != nil {
+		t.Fatal(err)
+	}
+	current, valid, err := sys.CheckKey(ctx, user.AccessKey)
+	if err != nil || !valid || current.Credentials.SecretKey != rotatedSecret {
+		t.Fatalf("rotation did not update the serving cache: valid=%v secret=%q err=%v", valid, current.Credentials.SecretKey, err)
+	}
+
+	close(blocker.release)
+	select {
+	case err := <-loadDone:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("LoadUser did not finish after the read barrier was released")
+	}
+	loaded, valid, err := sys.CheckKey(ctx, user.AccessKey)
+	if err != nil || !valid || loaded.Credentials.SecretKey != rotatedSecret {
+		t.Fatalf("an in-flight LoadUser restored the old credential: valid=%v secret=%q err=%v", valid, loaded.Credentials.SecretKey, err)
+	}
+}
@@ -45,6 +45,35 @@ type iamGroupRevocation struct {
 // its revision so an older non-expiring credential cannot return.
 type iamExpireIdentity struct{}

+// An edit of an identity applies to the identity as it is stored now, under
+// its revision lock, not to a server's cached copy. A copy written back
+// whole would undo a rotation or an edit made meanwhile through this or
+// another server, whose notification this one may have missed, and bring
+// back an identity deleted meanwhile. edit receives the stored record and
+// changes it in place; it returns errIAMIdentityUnchanged to write nothing,
+// or any other error to refuse. An identity that is absent or deleted is
+// refused with errIAMIdentityGone.
+type iamIdentityEdit struct {
+	UserIdentity // the record written, or found unchanged
+	accessKey    string
+	edit         func(u *UserIdentity) error
+	unchanged    bool
+}
+
+// An identity create is checked against the record under the same revision
+// lock as the write. Local callers may recreate a deleted access key, but may
+// never replace a live stored service account that their cache did not contain.
+// Replicated creates retain the normal revision ordering below and may replace
+// an older live snapshot when their source revision wins.
+type iamIdentityCreate struct {
+	UserIdentity
+}
+
+var (
+	errIAMIdentityGone      = errors.New("IAM identity is absent or deleted in storage")
+	errIAMIdentityUnchanged = errors.New("IAM identity edit changes nothing")
+)
+
 // The authoritative revocation is durable even if dependent cleanup fails.
 // Callers must publish it to sibling caches before returning the error.
 type iamCommittedCleanupError struct {
@@ -252,6 +281,69 @@ func saveIAMRevision(ctx context.Context, store IAMStorageAPI, path string, item
 	if err != nil {
 		return err
 	}
+	if op, ok := item.(*iamIdentityCreate); ok {
+		if _, replicated := iamReplicationTime(ctx); !replicated {
+			var current UserIdentity
+			err := store.loadIAMConfig(ctx, &current, path)
+			switch {
+			case err == nil && !current.Deleted:
+				return fmt.Errorf("%w: access key already exists in storage", errIAMServiceAccountNotAllowed)
+			case err == nil, errors.Is(err, errConfigNotFound):
+				// A tombstone or an absent record can be recreated. The
+				// existing revision path below retains its revocation boundary.
+			default:
+				return err
+			}
+		}
+		item = &op.UserIdentity
+	}
+	if op, ok := item.(*iamIdentityEdit); ok {
+		var current UserIdentity
+		err := store.loadIAMConfig(ctx, &current, path)
+		if errors.Is(err, errConfigNotFound) {
+			return errIAMIdentityGone
+		}
+		if err != nil {
+			return err
+		}
+		if current.Deleted {
+			if origin, replicated := iamReplicationTime(ctx); replicated && !origin.After(previous.timestamp()) {
+				// A peer's edit from before this deletion: acknowledged.
+				return errIAMStaleUpdate
+			}
+			return errIAMIdentityGone
+		}
+		// Normalize the record as the identity loaders do: the path names the
+		// access key, the claims are the token's, and a deprecated comment
+		// stands in for an empty description.
+		if current.Credentials.AccessKey == "" {
+			current.Credentials.AccessKey = op.accessKey
+		}
+		if current.Credentials.SessionToken != "" {
+			claims, err := extractJWTClaims(current)
+			if err != nil {
+				return err
+			}
+			current.Credentials.Claims = claims.Map()
+		}
+		if current.Credentials.Description == "" {
+			current.Credentials.Description = current.Credentials.Comment
+		}
+		if err := op.edit(&current); err != nil {
+			if errors.Is(err, errIAMIdentityUnchanged) {
+				op.UserIdentity, op.unchanged = current, true
+				return nil
+			}
+			return err
+		}
+		current.UpdatedAt = time.Time{} // a new revision, as for any local edit
+		op.UserIdentity = current
+		item = &op.UserIdentity
+		if strings.HasPrefix(path, iamConfigSTSPrefix) {
+			// Keep the record expiring with its token, as SetTempUser stores it.
+			opts = []options{{ttl: max(1, int64(time.Until(current.Credentials.Expiration).Seconds()))}}
+		}
+	}
 	if _, expiring := item.(*iamExpireIdentity); expiring {
 		sts := strings.HasPrefix(path, iamConfigSTSPrefix)
 		if previous.Deleted {
@@ -2778,37 +2778,26 @@ func (store *IAMStoreSys) SetUserStatus(ctx context.Context, accessKey string, s
 	cache := store.lock()
 	defer store.unlock()

-	ui, ok := cache.iamUsersMap[accessKey]
-	if !ok {
-		return updatedAt, errNoSuchUser
-	}
-	cred := ui.Credentials
-
-	if cred.IsTemp() || cred.IsServiceAccount() {
+	if ui, ok := cache.iamUsersMap[accessKey]; ok && (ui.Credentials.IsTemp() || ui.Credentials.IsServiceAccount()) {
 		return updatedAt, errIAMActionNotAllowed
 	}
-
-	uinfo := newUserIdentity(auth.Credentials{
-		AccessKey: accessKey,
-		SecretKey: cred.SecretKey,
-		Status: func() string {
-			switch string(status) {
-			case string(madmin.AccountEnabled), string(auth.AccountOn):
-				return auth.AccountOn
-			}
-			return auth.AccountOff
-		}(),
+	u, _, err := store.editStoredIdentity(ctx, cache, accessKey, regUser, func(stored *UserIdentity) error {
+		if stored.Credentials.IsTemp() || stored.Credentials.IsServiceAccount() {
+			return errIAMActionNotAllowed
+		}
+		stored.Credentials.Status = auth.AccountOff
+		if status == madmin.AccountEnabled {
+			stored.Credentials.Status = auth.AccountOn
+		}
+		return nil
 	})
-
-	if err := store.saveUserIdentity(ctx, accessKey, regUser, &uinfo); err != nil {
+	if errors.Is(err, errIAMIdentityGone) {
+		return updatedAt, errNoSuchUser
+	}
+	if err != nil {
 		return updatedAt, err
 	}
-
-	if err := cache.updateUserWithClaims(accessKey, uinfo); err != nil {
-		return updatedAt, err
-	}
-
-	return uinfo.UpdatedAt, nil
+	return u.UpdatedAt, nil
 }

 // AddServiceAccount - add a new service account
@@ -2842,10 +2831,12 @@ func (store *IAMStoreSys) AddServiceAccount(ctx context.Context, cred auth.Crede
 	}

 	u := newUserIdentity(cred)
-	err = store.saveUserIdentity(ctx, u.Credentials.AccessKey, svcUser, &u)
+	create := &iamIdentityCreate{UserIdentity: u}
+	err = store.saveIAMRevision(ctx, getUserIdentityPath(u.Credentials.AccessKey, svcUser), create)
 	if err != nil {
 		return updatedAt, err
 	}
+	u = create.UserIdentity

 	if err := cache.updateUserWithClaims(u.Credentials.AccessKey, u); err != nil {
 		// Persistence already replaced the credential. A failed refresh must
@@ -2863,15 +2854,39 @@ func (store *IAMStoreSys) UpdateServiceAccount(ctx context.Context, accessKey st
 	cache := store.lock()
 	defer store.unlock()

-	ui, ok := cache.iamUsersMap[accessKey]
-	if !ok || !ui.Credentials.IsServiceAccount() {
+	u, _, err := store.editStoredIdentity(ctx, cache, accessKey, svcUser, func(stored *UserIdentity) error {
+		if !stored.Credentials.IsServiceAccount() {
+			return errIAMIdentityGone
+		}
+		if opts.parentUser != "" && stored.Credentials.ParentUser != opts.parentUser {
+			// Not the account the caller was authorized for: it was
+			// deleted and created again for another parent meanwhile.
+			return errIAMIdentityGone
+		}
+		cr, err := updatedServiceAccountCredentials(ctx, accessKey, stored.Credentials, opts)
+		if err != nil {
+			return err
+		}
+		stored.Credentials = cr
+		return nil
+	})
+	if errors.Is(err, errIAMIdentityGone) {
 		return updatedAt, errNoSuchServiceAccount
 	}
-	cr := ui.Credentials
+	if err != nil {
+		return updatedAt, err
+	}
+	return u.UpdatedAt, nil
+}
+
+// updatedServiceAccountCredentials returns cr, the stored credential of
+// service account accessKey, with the edits of opts and a session token
+// signed again for them.
+func updatedServiceAccountCredentials(ctx context.Context, accessKey string, cr auth.Credentials, opts updateServiceAccountOpts) (auth.Credentials, error) {
 	currentSecretKey := cr.SecretKey
 	if opts.secretKey != "" {
 		if !auth.IsSecretKeyValid(opts.secretKey) {
-			return updatedAt, auth.ErrInvalidSecretKeyLength
+			return cr, auth.ErrInvalidSecretKeyLength
 		}
 		cr.SecretKey = opts.secretKey
 	}
@@ -2887,7 +2902,7 @@ func (store *IAMStoreSys) UpdateServiceAccount(ctx context.Context, accessKey st
 	if opts.expiration != nil {
 		expirationInUTC := opts.expiration.UTC()
 		if err := validateSvcExpirationInUTC(ctx, expirationInUTC); err != nil {
-			return updatedAt, err
+			return cr, err
 		}
 		cr.Expiration = expirationInUTC
 	}
@@ -2903,12 +2918,12 @@ func (store *IAMStoreSys) UpdateServiceAccount(ctx context.Context, accessKey st
 	case auth.AccountOn, auth.AccountOff:
 		cr.Status = opts.status
 	default:
-		return updatedAt, errors.New("unknown account status value")
+		return cr, errors.New("unknown account status value")
 	}

 	m, err := getClaimsFromTokenWithSecret(cr.SessionToken, currentSecretKey)
 	if err != nil {
-		return updatedAt, fmt.Errorf("unable to get svc acc claims: %v", err)
+		return cr, fmt.Errorf("unable to get svc acc claims: %v", err)
 	}

 	// Extracted session policy name string can be removed as its not useful
@@ -2926,17 +2941,17 @@ func (store *IAMStoreSys) UpdateServiceAccount(ctx context.Context, accessKey st

 	if opts.sessionPolicy != nil { // session policies is being updated
 		if err := opts.sessionPolicy.Validate(); err != nil {
-			return updatedAt, err
+			return cr, err
 		}

 		if opts.sessionPolicy.Version != "" && len(opts.sessionPolicy.Statements) > 0 {
 			policyBuf, err := json.Marshal(opts.sessionPolicy)
 			if err != nil {
-				return updatedAt, err
+				return cr, err
 			}

 			if len(policyBuf) > maxSVCSessionPolicySize {
-				return updatedAt, errSessionPolicyTooLarge
+				return cr, errSessionPolicyTooLarge
 			}

 			// Overwrite session policy claims.
@@ -2947,21 +2962,11 @@ func (store *IAMStoreSys) UpdateServiceAccount(ctx context.Context, accessKey st

 	cr.SessionToken, err = auth.JWTSignWithAccessKey(accessKey, m.Map(), cr.SecretKey)
 	if err != nil {
-		return updatedAt, err
+		return cr, err
 	}
-
-	u := newUserIdentity(cr)
-	if err := store.saveUserIdentity(ctx, u.Credentials.AccessKey, svcUser, &u); err != nil {
-		return updatedAt, err
-	}
-
-	if err := cache.updateUserWithClaims(u.Credentials.AccessKey, u); err != nil {
-		delete(cache.iamUsersMap, accessKey)
-		cache.updatedAt = UTCNow()
-		return updatedAt, err
-	}
-
-	return u.UpdatedAt, nil
+	// The stored claims are those of the new token; the save extracts them.
+	cr.Claims = nil
+	return cr, nil
 }

 // ListTempAccounts - lists only temporary accounts from the cache.
@@ -3105,20 +3110,14 @@ func (store *IAMStoreSys) UpdateUserSecretKey(ctx context.Context, accessKey, se
 	cache := store.lock()
 	defer store.unlock()

-	cache.updatedAt = time.Now()
-
-	ui, ok := cache.iamUsersMap[accessKey]
-	if !ok {
+	_, _, err := store.editStoredIdentity(ctx, cache, accessKey, regUser, func(stored *UserIdentity) error {
+		stored.Credentials.SecretKey = secretKey
+		return nil
+	})
+	if errors.Is(err, errIAMIdentityGone) {
 		return errNoSuchUser
 	}
-	cred := ui.Credentials
-	cred.SecretKey = secretKey
-	u := newUserIdentity(cred)
-	if err := store.saveUserIdentity(ctx, accessKey, regUser, &u); err != nil {
-		return err
-	}
-
-	return cache.updateUserWithClaims(accessKey, u)
+	return err
 }

 // GetSTSAndServiceAccounts - returns all STS and Service account credentials.
@@ -3144,39 +3143,99 @@ func (store *IAMStoreSys) getSTSAndServiceAccounts(cache *iamCache) []auth.Crede
 	return res
 }

-// UpdateUserIdentity - updates a user credential.
-func (store *IAMStoreSys) UpdateUserIdentity(ctx context.Context, cred auth.Credentials) error {
+// RefreshUserGroups sets the groups of the STS or service-account
+// credential that cred was read from to groups, on the record stored now
+// (see editStoredIdentity). Nothing else of the record changes. A
+// credential that is gone, expired or now has another parent, whose groups
+// the directory lookup did not compute, is left alone in storage and
+// dropped from the cache. It reports whether it wrote.
+func (store *IAMStoreSys) RefreshUserGroups(ctx context.Context, cred auth.Credentials, groups []string) (bool, error) {
 	cache := store.lock()
 	defer store.unlock()

-	cache.updatedAt = time.Now()
-
-	userType := regUser
-	if cred.IsServiceAccount() {
-		userType = svcUser
-	} else if cred.IsTemp() {
+	userType := svcUser
+	if !cred.IsServiceAccount() {
+		if !cred.IsTemp() {
+			return false, errInvalidArgument
+		}
 		userType = stsUser
 	}
-	ui := newUserIdentity(cred)
-	// Overwrite the user identity here. As store should be
-	// atomic, it shouldn't cause any corruption.
-	if err := store.saveUserIdentity(ctx, cred.AccessKey, userType, &ui); err != nil {
-		return err
+	_, unchanged, err := store.editStoredIdentity(ctx, cache, cred.AccessKey, userType, func(stored *UserIdentity) error {
+		if stored.Credentials.ParentUser != cred.ParentUser || stored.Credentials.IsExpired() {
+			return errIAMIdentityGone
+		}
+		if set.CreateStringSet(stored.Credentials.Groups...).Equals(set.CreateStringSet(groups...)) {
+			return errIAMIdentityUnchanged
+		}
+		stored.Credentials.Groups = groups
+		return nil
+	})
+	if errors.Is(err, errIAMIdentityGone) || errors.Is(err, errIAMStaleUpdate) {
+		return false, nil
 	}
+	if err != nil {
+		return false, err
+	}
+	return !unchanged, nil
+}

-	return cache.updateUserWithClaims(cred.AccessKey, ui)
+// editStoredIdentity applies edit to identity accessKey of userType as it
+// is stored now, under its revision lock (see iamIdentityEdit), and caches
+// the result. The caller holds the cache lock. When the identity is absent
+// or deleted in storage, or edit returns errIAMIdentityGone because it is
+// no longer the identity the edit addressed, errIAMIdentityGone is
+// returned; a replicated edit older than what is stored returns
+// errIAMStaleUpdate. Either way storage holds something else than the
+// cached copy, which is dropped, so a lookup loads what is stored.
+func (store *IAMStoreSys) editStoredIdentity(ctx context.Context, cache *iamCache, accessKey string, userType IAMUserType, edit func(u *UserIdentity) error) (UserIdentity, bool, error) {
+	op := &iamIdentityEdit{accessKey: accessKey, edit: edit}
+	err := store.saveIAMRevision(ctx, getUserIdentityPath(accessKey, userType), op)
+	if errors.Is(err, errIAMIdentityGone) || errors.Is(err, errIAMStaleUpdate) {
+		// Storage holds something else than the cached copy: drop it.
+		store.forgetCachedIdentity(cache, accessKey, userType)
+	}
+	if err != nil {
+		return UserIdentity{}, false, err
+	}
+	if err := cache.updateUserWithClaims(accessKey, op.UserIdentity); err != nil {
+		// Stored but not cached: a lookup loads it.
+		store.forgetCachedIdentity(cache, accessKey, userType)
+		return UserIdentity{}, false, err
+	}
+	return op.UserIdentity, op.unchanged, nil
+}
+
+// forgetCachedIdentity drops identity accessKey of userType from the cache;
+// a lookup then loads it from storage. Users and service accounts share a
+// map, so an entry of the other kind under the same key is kept.
+func (store *IAMStoreSys) forgetCachedIdentity(cache *iamCache, accessKey string, userType IAMUserType) {
+	switch userType {
+	case stsUser:
+		delete(cache.iamSTSAccountsMap, accessKey)
+	default:
+		if ui, ok := cache.iamUsersMap[accessKey]; ok && ui.Credentials.IsServiceAccount() == (userType == svcUser) {
+			delete(cache.iamUsersMap, accessKey)
+		}
+	}
+	cache.updatedAt = time.Now()
+	if store.group != nil {
+		store.group.Forget(accessKey)
+	}
 }

 // LoadUser - attempts to load user info from storage and updates cache.
 func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error {
 	groupLoad := env.Get("_MINIO_IAM_GROUP_REFRESH", config.EnableOff) == config.EnableOn

-	newCachePopulate := func() (val any, err error) {
-		newCache := newIamCache()
+	newCachePopulate := func(ctx context.Context) (newCache *iamCache, err error) {
+		newCache = newIamCache()
 		newCache.revisions = store.revisionIndex()

 		// Check for service account first
-		store.loadUser(ctx, accessKey, svcUser, newCache.iamUsersMap)
+		err = store.loadUser(ctx, accessKey, svcUser, newCache.iamUsersMap)
+		if err != nil && !errors.Is(err, errNoSuchUser) && !errors.Is(err, errConfigNotFound) {
+			return nil, err
+		}

 		svc, found := newCache.iamUsersMap[accessKey]
 		if found {
@@ -3194,6 +3253,9 @@ func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error

 		if !found {
 			err = store.loadUser(ctx, accessKey, regUser, newCache.iamUsersMap)
+			if err != nil && !errors.Is(err, errNoSuchUser) && !errors.Is(err, errConfigNotFound) {
+				return nil, err
+			}
 			if _, found = newCache.iamUsersMap[accessKey]; found {
 				// NOTE: we are not worried about loading errors from policies.
 				store.loadMappedPolicyWithRetry(ctx, accessKey, regUser, false, newCache.iamUserPolicyMap, 3)
@@ -3237,62 +3299,67 @@ func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error
 		return newCache, err
 	}

-	var (
-		val any
-		err error
-	)
-	if store.group != nil {
-		val, err, _ = store.group.Do(accessKey, newCachePopulate)
-	} else {
-		val, err = newCachePopulate()
-	}
-
-	// Return error right away if any.
-	if err != nil {
-		if errors.Is(err, errNoSuchUser) || errors.Is(err, errConfigNotFound) {
-			return nil
+	load := func() (any, error) {
+		loadedAt := time.Now()
+		newCache, err := newCachePopulate(ctx)
+		if err != nil {
+			return nil, err
 		}
-		return err
+
+		cache := store.lock()
+		defer store.unlock()
+
+		if cache.updatedAt.After(loadedAt) {
+			err = store.withIAMStorage(ctx, func(ctx context.Context) error {
+				newCache, err = newCachePopulate(ctx)
+				return err
+			})
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		// We need to merge the new cache with the existing cache because the
+		// periodic IAM reload is partial. The periodic load here is to account.
+		newCache.iamGroupPolicyMap.Range(func(k string, v MappedPolicy) bool {
+			cache.iamGroupPolicyMap.Store(k, v)
+			return true
+		})
+
+		maps.Copy(cache.iamGroupsMap, newCache.iamGroupsMap)
+
+		maps.Copy(cache.iamPolicyDocsMap, newCache.iamPolicyDocsMap)
+
+		maps.Copy(cache.iamUserGroupMemberships, newCache.iamUserGroupMemberships)
+
+		newCache.iamUserPolicyMap.Range(func(k string, v MappedPolicy) bool {
+			cache.iamUserPolicyMap.Store(k, v)
+			return true
+		})
+
+		maps.Copy(cache.iamUsersMap, newCache.iamUsersMap)
+
+		maps.Copy(cache.iamSTSAccountsMap, newCache.iamSTSAccountsMap)
+
+		newCache.iamSTSPolicyMap.Range(func(k string, v MappedPolicy) bool {
+			cache.iamSTSPolicyMap.Store(k, v)
+			return true
+		})
+
+		cache.updatedAt = time.Now()
+		return nil, nil
 	}

-	newCache, ok := val.(*iamCache)
-	if !ok {
+	var err error
+	if store.group != nil {
+		_, err, _ = store.group.Do(accessKey, load)
+	} else {
+		_, err = load()
+	}
+	if errors.Is(err, errNoSuchUser) || errors.Is(err, errConfigNotFound) {
 		return nil
 	}
-
-	cache := store.lock()
-	defer store.unlock()
-
-	// We need to merge the new cache with the existing cache because the
-	// periodic IAM reload is partial. The periodic load here is to account.
-	newCache.iamGroupPolicyMap.Range(func(k string, v MappedPolicy) bool {
-		cache.iamGroupPolicyMap.Store(k, v)
-		return true
-	})
-
-	maps.Copy(cache.iamGroupsMap, newCache.iamGroupsMap)
-
-	maps.Copy(cache.iamPolicyDocsMap, newCache.iamPolicyDocsMap)
-
-	maps.Copy(cache.iamUserGroupMemberships, newCache.iamUserGroupMemberships)
-
-	newCache.iamUserPolicyMap.Range(func(k string, v MappedPolicy) bool {
-		cache.iamUserPolicyMap.Store(k, v)
-		return true
-	})
-
-	maps.Copy(cache.iamUsersMap, newCache.iamUsersMap)
-
-	maps.Copy(cache.iamSTSAccountsMap, newCache.iamSTSAccountsMap)
-
-	newCache.iamSTSPolicyMap.Range(func(k string, v MappedPolicy) bool {
-		cache.iamSTSPolicyMap.Store(k, v)
-		return true
-	})
-
-	cache.updatedAt = time.Now()
-
-	return nil
+	return err
 }

 var errTokenSigningKeyUnavailable = errors.New("token signing key unavailable")
@@ -0,0 +1,301 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/auth"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+// An administrator edits an identity through a server whose cache missed
+// a rotation or a deletion made through another. The edit applies to what
+// is stored and must not write the stale copy back.
+
+func TestIAMStaleServiceAccountEdit(t *testing.T) {
+	t.Run("rotated", func(t *testing.T) {
+		f := newGroupRefreshFixture(t)
+		snapshot := f.service(t, "edit-rotated")
+		other := f.sibling(t)
+		_, err := f.sys.UpdateServiceAccount(f.ctx, snapshot.AccessKey, updateServiceAccountOpts{secretKey: "edit-rotated-secret"})
+		mustIAM(t, err)
+		if cached, _ := other.GetUser(snapshot.AccessKey); cached.Credentials.SecretKey != snapshot.SecretKey {
+			t.Fatal("control: the stale server must not have heard of the rotation")
+		}
+		_, err = other.UpdateServiceAccount(f.ctx, snapshot.AccessKey, updateServiceAccountOpts{description: "edited on a stale server"})
+		mustIAM(t, err)
+		for where, get := range map[string]func() (UserIdentity, bool){
+			"storage":      func() (UserIdentity, bool) { return f.stored(t, snapshot.AccessKey) },
+			"stale server": func() (UserIdentity, bool) { return other.GetUser(snapshot.AccessKey) },
+		} {
+			u, ok := get()
+			if !ok || u.Credentials.SecretKey != "edit-rotated-secret" || u.Credentials.Description != "edited on a stale server" {
+				t.Errorf("%s: present=%v rotated=%v description=%q: the edit must apply to the rotated credential",
+					where, ok, u.Credentials.SecretKey == "edit-rotated-secret", u.Credentials.Description)
+			}
+		}
+	})
+	t.Run("deleted", func(t *testing.T) {
+		f := newGroupRefreshFixture(t)
+		snapshot := f.service(t, "edit-deleted")
+		other := f.sibling(t)
+		mustIAM(t, f.sys.DeleteServiceAccount(f.ctx, snapshot.AccessKey, false))
+		_, err := other.UpdateServiceAccount(f.ctx, snapshot.AccessKey, updateServiceAccountOpts{description: "edited on a stale server"})
+		if !errors.Is(err, errNoSuchServiceAccount) {
+			t.Errorf("an edit of a deleted account must fail with no such service account, got %v", err)
+		}
+		if _, ok := f.stored(t, snapshot.AccessKey); ok {
+			t.Error("an edit on a stale server brought a deleted account back")
+		}
+		if _, ok := other.GetUser(snapshot.AccessKey); ok {
+			t.Error("the stale server still caches the deleted account")
+		}
+	})
+}
+
+// A server that missed the create notification must not replace a live
+// service account merely because its local cache has no entry for the key.
+func TestIAMStaleServiceAccountCreate(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	other := f.sibling(t)
+	original := f.service(t, "create-stale")
+	if _, ok := other.GetUser(original.AccessKey); ok {
+		t.Fatal("control: the stale server must not have loaded the new account")
+	}
+
+	claims := map[string]any{
+		parentClaim:            f.parent.AccessKey,
+		iamPolicyClaimNameSA(): inheritedPolicyType,
+	}
+	mustIAM(t, setIAMParentRevocationClaim(f.ctx, other, f.parent.AccessKey, claims))
+	replacement, err := auth.CreateNewCredentialsWithMetadata(original.AccessKey, "create-replacement-secret", claims, "create-replacement-secret")
+	mustIAM(t, err)
+	replacement.ParentUser = f.parent.AccessKey
+	replacement.Status = auth.AccountOn
+	if _, err := other.AddServiceAccount(f.ctx, replacement); !errors.Is(err, errIAMServiceAccountNotAllowed) {
+		t.Fatalf("stale create should be rejected, got %v", err)
+	}
+
+	stored, ok := f.stored(t, original.AccessKey)
+	if !ok || stored.Credentials.SecretKey != original.SecretKey || stored.Credentials.ParentUser != original.ParentUser {
+		t.Fatalf("stale create changed the live account: present=%v parent=%q secret=%q", ok, stored.Credentials.ParentUser, stored.Credentials.SecretKey)
+	}
+}
+
+// A session policy edit stores the claims of the token it signs.
+func TestIAMServiceAccountEditStoresNewClaims(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "edit-claims")
+	sp, err := policy.ParseConfig(strings.NewReader(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::edit/*"]}]}`))
+	mustIAM(t, err)
+	_, err = f.sys.UpdateServiceAccount(f.ctx, snapshot.AccessKey, updateServiceAccountOpts{sessionPolicy: sp})
+	mustIAM(t, err)
+	var raw UserIdentity
+	mustIAM(t, f.sys.store.loadIAMConfig(f.ctx, &raw, getUserIdentityPath(snapshot.AccessKey, svcUser)))
+	claims, err := getClaimsFromTokenWithSecret(raw.Credentials.SessionToken, raw.Credentials.SecretKey)
+	mustIAM(t, err)
+	if raw.Credentials.Claims[iamPolicyClaimNameSA()] != embeddedPolicyType || claims.MapClaims[iamPolicyClaimNameSA()] != embeddedPolicyType {
+		t.Errorf("stored claims %v, token claims %v: both must name the embedded policy", raw.Credentials.Claims[iamPolicyClaimNameSA()], claims.MapClaims[iamPolicyClaimNameSA()])
+	}
+}
+
+func TestIAMStaleUserEdit(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		change func(t *testing.T, f *groupRefreshFixture, user auth.Credentials)
+		edit   func(s *IAMStoreSys, f *groupRefreshFixture, user auth.Credentials) error
+		check  func(t *testing.T, u UserIdentity, ok bool, err error)
+	}{
+		{"status-after-delete", func(t *testing.T, f *groupRefreshFixture, user auth.Credentials) {
+			mustIAM(t, f.sys.DeleteUser(f.ctx, user.AccessKey, false))
+		}, func(s *IAMStoreSys, f *groupRefreshFixture, user auth.Credentials) error {
+			_, err := s.SetUserStatus(f.ctx, user.AccessKey, madmin.AccountEnabled)
+			return err
+		}, func(t *testing.T, u UserIdentity, ok bool, err error) {
+			if !errors.Is(err, errNoSuchUser) || ok {
+				t.Errorf("err=%v present=%v: a status change on a stale server must not bring a deleted user back", err, ok)
+			}
+		}},
+		{"status-after-rotation", func(t *testing.T, f *groupRefreshFixture, user auth.Credentials) {
+			mustIAM(t, f.sys.SetUserSecretKey(f.ctx, user.AccessKey, "edit-user-rotated"))
+		}, func(s *IAMStoreSys, f *groupRefreshFixture, user auth.Credentials) error {
+			_, err := s.SetUserStatus(f.ctx, user.AccessKey, madmin.AccountDisabled)
+			return err
+		}, func(t *testing.T, u UserIdentity, ok bool, err error) {
+			if err != nil || !ok || u.Credentials.SecretKey != "edit-user-rotated" || u.Credentials.Status != auth.AccountOff {
+				t.Errorf("err=%v present=%v rotated=%v status=%q: the status change must apply to the rotated user",
+					err, ok, u.Credentials.SecretKey == "edit-user-rotated", u.Credentials.Status)
+			}
+		}},
+		{"secret-after-delete", func(t *testing.T, f *groupRefreshFixture, user auth.Credentials) {
+			mustIAM(t, f.sys.DeleteUser(f.ctx, user.AccessKey, false))
+		}, func(s *IAMStoreSys, f *groupRefreshFixture, user auth.Credentials) error {
+			return s.UpdateUserSecretKey(f.ctx, user.AccessKey, "edit-user-new-secret")
+		}, func(t *testing.T, u UserIdentity, ok bool, err error) {
+			if !errors.Is(err, errNoSuchUser) || ok {
+				t.Errorf("err=%v present=%v: a secret change on a stale server must not bring a deleted user back", err, ok)
+			}
+		}},
+		{"secret-after-disable", func(t *testing.T, f *groupRefreshFixture, user auth.Credentials) {
+			_, err := f.sys.SetUserStatus(f.ctx, user.AccessKey, madmin.AccountDisabled)
+			mustIAM(t, err)
+		}, func(s *IAMStoreSys, f *groupRefreshFixture, user auth.Credentials) error {
+			return s.UpdateUserSecretKey(f.ctx, user.AccessKey, "edit-user-new-secret")
+		}, func(t *testing.T, u UserIdentity, ok bool, err error) {
+			if err != nil || !ok || u.Credentials.SecretKey != "edit-user-new-secret" || u.Credentials.Status != auth.AccountOff {
+				t.Errorf("err=%v present=%v status=%q: a secret change on a stale server must keep the user disabled", err, ok, u.Credentials.Status)
+			}
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			f := newGroupRefreshFixture(t)
+			user := iamAdminUser(f.ctx, t, f.sys, "edit-user", "refresh-parent")
+			other := f.sibling(t)
+			tc.change(t, f, user)
+			if cached, ok := other.GetUser(user.AccessKey); !ok || cached.Credentials.SecretKey != user.SecretKey {
+				t.Fatal("control: the stale server must not have heard of the change")
+			}
+			err := tc.edit(other, f, user)
+			u, ok := f.stored(t, user.AccessKey)
+			tc.check(t, u, ok, err)
+		})
+	}
+}
+
+// An edit that the caller was authorized for under one parent does not
+// apply to an account stored under another parent: the account was deleted
+// and created again for it meanwhile.
+func TestIAMServiceAccountEditKeepsAuthorizedParent(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "edit-parent")
+	other := iamAdminUser(f.ctx, t, f.sys, "edit-other-parent", "refresh-parent")
+	mustIAM(t, f.sys.DeleteServiceAccount(f.ctx, snapshot.AccessKey, false))
+	_, _, err := f.sys.NewServiceAccount(f.ctx, other.AccessKey, nil, newServiceAccountOpts{accessKey: snapshot.AccessKey, secretKey: "edit-other-secret"})
+	mustIAM(t, err)
+	_, err = f.sys.store.UpdateServiceAccount(f.ctx, snapshot.AccessKey, updateServiceAccountOpts{secretKey: "edit-chosen-secret", parentUser: snapshot.ParentUser})
+	if !errors.Is(err, errNoSuchServiceAccount) {
+		t.Errorf("an edit authorized for another parent must fail, got %v", err)
+	}
+	if u, ok := f.stored(t, snapshot.AccessKey); !ok || u.Credentials.SecretKey != "edit-other-secret" {
+		t.Errorf("present=%v: the account of the other parent must keep its secret", ok)
+	}
+	// Without a bound parent (replication), the edit applies.
+	_, err = f.sys.store.UpdateServiceAccount(f.ctx, snapshot.AccessKey, updateServiceAccountOpts{description: "replicated"})
+	mustIAM(t, err)
+}
+
+// A record stored without claims, or with only a deprecated comment, is
+// edited as the loaders present it.
+func TestIAMServiceAccountEditLegacyRecord(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "edit-legacy")
+	path := getUserIdentityPath(snapshot.AccessKey, svcUser)
+	var raw UserIdentity
+	mustIAM(t, f.sys.store.loadIAMConfig(f.ctx, &raw, path))
+	raw.Credentials.Claims = nil
+	raw.Credentials.Description, raw.Credentials.Comment = "", "legacy comment"
+	mustIAM(t, f.sys.store.saveIAMConfig(f.ctx, &raw, path))
+	_, err := f.sys.store.UpdateServiceAccount(f.ctx, snapshot.AccessKey, updateServiceAccountOpts{status: string(madmin.AccountDisabled)})
+	mustIAM(t, err)
+	for where, get := range map[string]func() (UserIdentity, bool){
+		"storage": func() (UserIdentity, bool) { return f.stored(t, snapshot.AccessKey) },
+		"cache":   func() (UserIdentity, bool) { return f.sys.store.GetUser(snapshot.AccessKey) },
+	} {
+		u, ok := get()
+		if !ok || u.Credentials.Status != auth.AccountOff || u.Credentials.Description != "legacy comment" {
+			t.Errorf("%s: present=%v status=%q description=%q", where, ok, u.Credentials.Status, u.Credentials.Description)
+		}
+	}
+}
+
+// A peer's edit from before a deletion this site applied is acknowledged as
+// stale; one from after it is not found.
+func TestIAMReplicatedEditAfterDeletion(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "edit-repl")
+	user := iamAdminUser(f.ctx, t, f.sys, "edit-repl-user", "refresh-parent")
+	before := UTCNow()
+	mustIAM(t, f.sys.DeleteServiceAccount(f.ctx, snapshot.AccessKey, false))
+	mustIAM(t, f.sys.DeleteUser(f.ctx, user.AccessKey, false))
+	after := UTCNow().Add(time.Second)
+
+	_, err := f.sys.store.UpdateServiceAccount(withIAMReplicationTime(f.ctx, before), snapshot.AccessKey, updateServiceAccountOpts{description: "late"})
+	if iamReplicationError(err) != nil {
+		t.Errorf("an older service-account edit must be acknowledged, got %v", err)
+	}
+	_, err = f.sys.store.SetUserStatus(withIAMReplicationTime(f.ctx, before), user.AccessKey, madmin.AccountDisabled)
+	if iamReplicationError(err) != nil {
+		t.Errorf("an older status edit must be acknowledged, got %v", err)
+	}
+	if _, err = f.sys.store.UpdateServiceAccount(withIAMReplicationTime(f.ctx, after), snapshot.AccessKey, updateServiceAccountOpts{description: "late"}); !errors.Is(err, errNoSuchServiceAccount) {
+		t.Errorf("a newer edit of a deleted account must not be acknowledged as stale, got %v", err)
+	}
+	if _, ok := f.stored(t, snapshot.AccessKey); ok {
+		t.Error("a replicated edit brought a deleted account back")
+	}
+}
+
+// A status change addressed to a service account is refused as before, and
+// the account stays cached.
+func TestIAMUserStatusOnServiceAccount(t *testing.T) {
+	f := newGroupRefreshFixture(t)
+	snapshot := f.service(t, "edit-sa-status")
+	_, err := f.sys.store.SetUserStatus(f.ctx, snapshot.AccessKey, madmin.AccountDisabled)
+	if !errors.Is(err, errIAMActionNotAllowed) {
+		t.Errorf("got %v, want errIAMActionNotAllowed", err)
+	}
+	if _, ok := f.sys.store.GetUser(snapshot.AccessKey); !ok {
+		t.Error("a refused status change dropped the service account from the cache")
+	}
+	if err := f.sys.store.UpdateUserSecretKey(f.ctx, snapshot.AccessKey, "edit-not-a-user"); !errors.Is(err, errNoSuchUser) {
+		t.Errorf("a user secret change of a service account must find no user, got %v", err)
+	}
+	if u, ok := f.sys.store.GetUser(snapshot.AccessKey); !ok || u.Credentials.SecretKey != snapshot.SecretKey {
+		t.Error("a user secret change of a service account touched the account")
+	}
+}
+
+// A server whose cache still holds the account of the first parent drops
+// it when its edit is refused, whether the edit was bound to that parent or
+// replicated from before the deletion: storage says the cached copy is gone.
+func TestIAMRefusedEditDropsStaleCache(t *testing.T) {
+	t.Run("rebound-parent", func(t *testing.T) {
+		f := newGroupRefreshFixture(t)
+		snapshot := f.service(t, "edit-rebind")
+		other := f.sibling(t)
+		next := iamAdminUser(f.ctx, t, f.sys, "edit-rebind-next", "refresh-parent")
+		mustIAM(t, f.sys.DeleteServiceAccount(f.ctx, snapshot.AccessKey, false))
+		_, _, err := f.sys.NewServiceAccount(f.ctx, next.AccessKey, nil, newServiceAccountOpts{accessKey: snapshot.AccessKey, secretKey: "edit-rebind-secret"})
+		mustIAM(t, err)
+		if cached, ok := other.GetUser(snapshot.AccessKey); !ok || cached.Credentials.ParentUser != snapshot.ParentUser {
+			t.Fatal("control: the stale server must still cache the first account")
+		}
+		_, err = other.UpdateServiceAccount(f.ctx, snapshot.AccessKey, updateServiceAccountOpts{description: "stale", parentUser: snapshot.ParentUser})
+		if !errors.Is(err, errNoSuchServiceAccount) {
+			t.Errorf("got %v, want errNoSuchServiceAccount", err)
+		}
+		if _, ok := other.GetUser(snapshot.AccessKey); ok {
+			t.Error("the stale server still caches the account of the first parent")
+		}
+	})
+	t.Run("replicated-stale", func(t *testing.T) {
+		f := newGroupRefreshFixture(t)
+		snapshot := f.service(t, "edit-repl-stale")
+		other := f.sibling(t)
+		before := UTCNow()
+		mustIAM(t, f.sys.DeleteServiceAccount(f.ctx, snapshot.AccessKey, false))
+		_, err := other.UpdateServiceAccount(withIAMReplicationTime(f.ctx, before), snapshot.AccessKey, updateServiceAccountOpts{description: "late"})
+		if iamReplicationError(err) != nil {
+			t.Errorf("an older replicated edit must be acknowledged, got %v", err)
+		}
+		if _, ok := other.GetUser(snapshot.AccessKey); ok {
+			t.Error("the server that acknowledged a stale edit still caches the deleted account")
+		}
+	})
+}
@@ -1211,6 +1211,9 @@ type updateServiceAccountOpts struct {
 	status            string
 	name, description string
 	expiration        *time.Time
+	// parentUser, when set, is the parent the caller was authorized for; an
+	// account stored under another parent is refused as absent.
+	parentUser string
 }

 // UpdateServiceAccount - edit a service account
@@ -1658,8 +1661,10 @@ func (sys *IAMSys) updateGroupMembershipsForLDAP(ctx context.Context) {
 				continue
 			}

-			cred.Groups = currGroups
-			if err := sys.store.UpdateUserIdentity(ctx, cred); err != nil {
+			// Only the groups change, on the credential as it is stored now:
+			// cred is a snapshot from before the lookup, and a rotation, an
+			// edit or a deletion since then must stand.
+			if _, err := sys.store.RefreshUserGroups(ctx, cred, currGroups); err != nil {
 				// Log and continue error - perhaps it'll work the next time.
 				iamLogIf(GlobalContext, err)
 			}
@@ -0,0 +1,78 @@
+package cmd
+
+import (
+	"maps"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+)
+
+type realtimeMetrics struct {
+	madmin.RealtimeMetrics
+	Aggregated realtimeMetricGroups            `json:"aggregated"`
+	ByHost     map[string]realtimeMetricGroups `json:"by_host,omitempty"`
+}
+
+type realtimeMetricGroups struct {
+	madmin.Metrics
+	BatchJobs *batchJobMetricsReport `json:"batchJobs,omitempty"`
+}
+
+type batchJobMetricsReport struct {
+	CollectedAt time.Time `json:"collected"`
+	Jobs        map[string]batchJobStatusMetric
+}
+
+func (metrics *realtimeMetrics) Merge(other *realtimeMetrics) {
+	if other == nil {
+		return
+	}
+	metrics.RealtimeMetrics.Merge(&other.RealtimeMetrics)
+	metrics.Aggregated.Merge(&other.Aggregated.Metrics)
+	if other.Aggregated.BatchJobs != nil {
+		if metrics.Aggregated.BatchJobs == nil {
+			metrics.Aggregated.BatchJobs = &batchJobMetricsReport{Jobs: make(map[string]batchJobStatusMetric)}
+		}
+		jobs := metrics.Aggregated.BatchJobs
+		if other.Aggregated.BatchJobs.CollectedAt.After(jobs.CollectedAt) {
+			jobs.CollectedAt = other.Aggregated.BatchJobs.CollectedAt
+		}
+		for id, job := range other.Aggregated.BatchJobs.Jobs {
+			if current, ok := jobs.Jobs[id]; !ok || !job.LastUpdate.Before(current.LastUpdate) {
+				jobs.Jobs[id] = job
+			}
+		}
+	}
+	if len(other.ByHost) != 0 {
+		if metrics.ByHost == nil {
+			metrics.ByHost = make(map[string]realtimeMetricGroups, len(other.ByHost))
+		}
+		maps.Copy(metrics.ByHost, other.ByHost)
+	}
+}
+
+func (metrics realtimeMetricGroups) legacy() madmin.Metrics {
+	out := metrics.Metrics
+	if metrics.BatchJobs != nil {
+		out.BatchJobs = &madmin.BatchJobMetrics{
+			CollectedAt: metrics.BatchJobs.CollectedAt,
+			Jobs:        make(map[string]madmin.JobMetric, len(metrics.BatchJobs.Jobs)),
+		}
+		for id, job := range metrics.BatchJobs.Jobs {
+			out.BatchJobs.Jobs[id] = job.JobMetric
+		}
+	}
+	return out
+}
+
+func (metrics realtimeMetrics) legacy() madmin.RealtimeMetrics {
+	out := metrics.RealtimeMetrics
+	out.Aggregated = metrics.Aggregated.legacy()
+	if metrics.ByHost != nil {
+		out.ByHost = make(map[string]madmin.Metrics, len(metrics.ByHost))
+		for host, groups := range metrics.ByHost {
+			out.ByHost[host] = groups.legacy()
+		}
+	}
+	return out
+}
@@ -0,0 +1,153 @@
+package cmd
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/grid"
+)
+
+func TestBatchMetricsStatusSnapshots(t *testing.T) {
+	fixture := newBatchAuthzFixture(t)
+	job := BatchJobRequest{
+		ID: "expire-metrics-snapshot" + getKeySeparator() + "0", Started: UTCNow(),
+		Submitter: batchRootSubmitter(t), Expire: &BatchJobExpire{},
+	}
+	for _, state := range []string{"skipped", "paused", "failed", "running"} {
+		t.Run(state, func(t *testing.T) {
+			report := &batchJobInfo{JobID: job.ID, JobType: string(job.Type()), StartTime: job.Started}
+			switch state {
+			case "skipped":
+				report.Complete = true
+				report.VersionsSkippedLocked, report.KeysRefused = 2, 1
+			case "paused":
+				report.PausedReason = "authorization plugin unavailable"
+			case "failed":
+				report.Failed = true
+				report.FailureReason = "authorization denied"
+			}
+			mustIAM(t, report.updateAfter(fixture.ctx, fixture.obj, 0, job))
+			globalBatchJobsMetrics.save(job.ID, report)
+			saved := batchJobStatusJSON(t, job.ID)
+			for _, field := range []string{"startTime", "lastUpdate"} {
+				stamp, err := time.Parse(time.RFC3339Nano, saved[field].(string))
+				mustIAM(t, err)
+				saved[field] = stamp.UTC().Format(time.RFC3339Nano)
+			}
+			target := fmt.Sprintf("http://localhost/minio/admin/v3/metrics?types=%d&by-jobID=%s&n=1&by-host=true", madmin.MetricsBatchJobs, job.ID)
+			request, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+			mustIAM(t, err)
+			mustIAM(t, request.ParseForm())
+			response := httptest.NewRecorder()
+			request = request.WithContext(newContext(request, response, "Metrics"))
+			adminAPIHandlers{}.MetricsHandler(response, request)
+			if response.Code != http.StatusOK {
+				t.Fatalf("metrics returned %d: %s", response.Code, response.Body.String())
+			}
+			var wire struct {
+				Aggregated struct {
+					BatchJobs struct{ Jobs map[string]map[string]any }
+				}
+				ByHost map[string]struct {
+					BatchJobs struct{ Jobs map[string]map[string]any }
+				} `json:"by_host"`
+			}
+			mustIAM(t, json.Unmarshal(response.Body.Bytes(), &wire))
+			if !reflect.DeepEqual(wire.Aggregated.BatchJobs.Jobs[job.ID], saved) {
+				t.Fatalf("live snapshot differs from saved status: live=%v saved=%v", wire.Aggregated.BatchJobs.Jobs[job.ID], saved)
+			}
+			if len(wire.ByHost) != 1 {
+				t.Fatalf("missing per-host snapshot: %s", response.Body.String())
+			}
+			for _, host := range wire.ByHost {
+				if !reflect.DeepEqual(host.BatchJobs.Jobs[job.ID], saved) {
+					t.Fatal("per-host metrics lost details")
+				}
+			}
+			var legacy madmin.RealtimeMetrics
+			mustIAM(t, json.Unmarshal(response.Body.Bytes(), &legacy))
+			if legacy.Aggregated.BatchJobs.Jobs[job.ID].JobID != job.ID || legacy.Aggregated.BatchJobs.Jobs[job.ID].Complete != report.Complete {
+				t.Fatal("additive details changed legacy metrics")
+			}
+		})
+	}
+}
+
+func TestBatchMetricsPeerRoundTrip(t *testing.T) {
+	report := &batchJobInfo{JobID: "remote-job", LastUpdate: UTCNow(), PausedReason: "IAM unavailable", VersionsSkippedLocked: 3, KeysRefused: 2}
+	remote := realtimeMetrics{}
+	remote.Hosts = []string{"peer"}
+	remote.Aggregated.BatchJobs = &batchJobMetricsReport{
+		CollectedAt: UTCNow(), Jobs: map[string]batchJobStatusMetric{report.JobID: report.statusMetric()},
+	}
+	remote.ByHost = map[string]realtimeMetricGroups{"peer": remote.Aggregated}
+	encoded := madminRealtimeMetrics.NewJSONWith(&remote)
+	data, err := encoded.MarshalMsg(nil)
+	mustIAM(t, err)
+	decoded := madminRealtimeMetrics.NewJSON()
+	defer decoded.Recycle()
+	_, err = decoded.UnmarshalMsg(data)
+	mustIAM(t, err)
+	var merged realtimeMetrics
+	received := decoded.ValueOrZero()
+	merged.Merge(&received)
+	got := merged.Aggregated.BatchJobs.Jobs[report.JobID]
+	if got.PausedReason != report.PausedReason || got.KeysRefused != 2 || got.VersionsSkippedLocked == nil || *got.VersionsSkippedLocked != 3 {
+		t.Fatalf("peer aggregation lost details: %+v", got)
+	}
+	older := received
+	older.Aggregated.BatchJobs = &batchJobMetricsReport{Jobs: map[string]batchJobStatusMetric{
+		report.JobID: {JobMetric: madmin.JobMetric{JobID: report.JobID, LastUpdate: report.LastUpdate.Add(-time.Second), Complete: true}},
+	}}
+	merged.Merge(&older)
+	if got := merged.Aggregated.BatchJobs.Jobs[report.JobID]; got.Complete || got.PausedReason != report.PausedReason {
+		t.Fatal("older peer snapshot replaced current progress")
+	}
+	legacyPool := grid.NewJSONPool[madmin.RealtimeMetrics]()
+	legacy := legacyPool.NewJSON()
+	defer legacy.Recycle()
+	_, err = legacy.UnmarshalMsg(data)
+	mustIAM(t, err)
+	if got := legacy.ValueOrZero().Aggregated.BatchJobs.Jobs[report.JobID]; got.JobID != report.JobID || !got.LastUpdate.Equal(report.LastUpdate) {
+		t.Fatal("legacy peer could not read base metrics")
+	}
+	legacyData, err := legacy.MarshalMsg(nil)
+	mustIAM(t, err)
+	_, err = decoded.UnmarshalMsg(legacyData)
+	mustIAM(t, err)
+	received = decoded.ValueOrZero()
+	if received.Aggregated.BatchJobs.Jobs[report.JobID].VersionsSkippedLocked != nil {
+		t.Fatal("legacy peer falsely advertised complete status details")
+	}
+}
+
+func TestBatchMetricsPreservesOtherMetrics(t *testing.T) {
+	var legacy madmin.RealtimeMetrics
+	var extended realtimeMetrics
+	for _, host := range []string{"first", "second"} {
+		groups := madmin.Metrics{
+			Disk: &madmin.DiskMetric{CollectedAt: time.Unix(1, 0).UTC(), NDisks: 2, Offline: 1},
+			CPU:  &madmin.CPUMetrics{CPUCount: 4},
+		}
+		sample := madmin.RealtimeMetrics{
+			Hosts: []string{host}, Errors: []string{host + ": test"},
+			Aggregated: groups, ByHost: map[string]madmin.Metrics{host: groups},
+			ByDisk: map[string]madmin.DiskMetric{host: *groups.Disk},
+		}
+		data, err := json.Marshal(sample)
+		mustIAM(t, err)
+		var decoded realtimeMetrics
+		mustIAM(t, json.Unmarshal(data, &decoded))
+		extended.Merge(&decoded)
+		legacy.Merge(&sample)
+	}
+	if got := extended.legacy(); !reflect.DeepEqual(got, legacy) {
+		t.Fatalf("batch extension changed other metrics: got=%+v want=%+v", got, legacy)
+	}
+}
@@ -38,7 +38,7 @@ type collectMetricsOpts struct {
 	depID string
 }

-func collectLocalMetrics(types madmin.MetricType, opts collectMetricsOpts) (m madmin.RealtimeMetrics) {
+func collectLocalMetrics(types madmin.MetricType, opts collectMetricsOpts) (m realtimeMetrics) {
 	if types == madmin.MetricsNone {
 		return m
 	}
@@ -144,7 +144,7 @@ func collectLocalMetrics(types madmin.MetricType, opts collectMetricsOpts) (m ma
 	// Add types...

 	// ByHost is a shallow reference, so careful about sharing.
-	m.ByHost = map[string]madmin.Metrics{byHostName: m.Aggregated}
+	m.ByHost = map[string]realtimeMetricGroups{byHostName: m.Aggregated}
 	m.Hosts = append(m.Hosts, byHostName)

 	return m
@@ -221,11 +221,11 @@ func collectLocalDisksMetrics(disks map[string]struct{}) map[string]madmin.DiskM
 	return metrics
 }

-func collectRemoteMetrics(ctx context.Context, types madmin.MetricType, opts collectMetricsOpts) (m madmin.RealtimeMetrics) {
+func collectRemoteMetrics(ctx context.Context, types madmin.MetricType, opts collectMetricsOpts) (m realtimeMetrics) {
 	if !globalIsDistErasure {
 		return m
 	}
-	all := globalNotificationSys.GetMetrics(ctx, types, opts)
+	all := globalNotificationSys.getMetrics(ctx, types, opts)
 	for _, remote := range all {
 		m.Merge(&remote)
 	}
@@ -246,7 +246,7 @@ func updateDriveIOStats(currentStats madmin.DiskIOStats, latestStats madmin.Disk
 	updateResourceMetrics(driveSubsystem, percUtil, float64(diffStats.TotalTicks)/(diffInSeconds*10), labels, false)
 }

-func collectDriveMetrics(m madmin.RealtimeMetrics) {
+func collectDriveMetrics(m realtimeMetrics) {
 	latestDriveStatsMu.Lock()
 	for d, dm := range m.ByDisk {
 		labels := map[string]string{"drive": d}
@@ -3804,7 +3804,7 @@ func getBatchJobsMetrics(opts MetricsGroupOpts) *MetricsGroupV2 {
 	}

 	mg.RegisterRead(func(ctx context.Context) (metrics []MetricV2) {
-		var m madmin.RealtimeMetrics
+		var m realtimeMetrics
 		mLocal := collectLocalMetrics(madmin.MetricsBatchJobs, collectMetricsOpts{})
 		m.Merge(&mLocal)

@@ -873,9 +873,9 @@ func (sys *NotificationSys) GetOSInfo(ctx context.Context) []madmin.OSInfo {
 	return reply
 }

-// GetMetrics - Get metrics from all peers.
-func (sys *NotificationSys) GetMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) []madmin.RealtimeMetrics {
-	reply := make([]madmin.RealtimeMetrics, len(sys.peerClients))
+// getMetrics - Get metrics from all peers.
+func (sys *NotificationSys) getMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) []realtimeMetrics {
+	reply := make([]realtimeMetrics, len(sys.peerClients))

 	g := errgroup.WithNErrs(len(sys.peerClients))
 	for index, client := range sys.peerClients {
@@ -3193,9 +3193,12 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 	vID := opts.VersionID
 	if replica {
 		opts.SetReplicaStatus(replication.Replica)
-		if opts.VersionPurgeStatus().Empty() {
-			// opts.VersionID holds delete marker version ID to replicate and not yet present on disk
+		if opts.DeleteMarker && opts.VersionPurgeStatus().Empty() {
+			// opts.VersionID names the delete marker to replicate, not yet
+			// present here. It must not replace a stored version this site
+			// protects; a replicated purge keeps the retention check below.
 			vID = ""
+			opts.RefuseLockedOverwrite = true
 		}
 	}
 	opts.SetEvalRetentionBypassFn(func(goi ObjectInfo, gerr error) (err error) {
@@ -228,7 +228,7 @@ func (client *peerRESTClient) GetMemInfo(ctx context.Context) (info madmin.MemIn
 }

 // GetMetrics - fetch metrics from a remote node.
-func (client *peerRESTClient) GetMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) (info madmin.RealtimeMetrics, err error) {
+func (client *peerRESTClient) GetMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) (info realtimeMetrics, err error) {
 	values := make(url.Values)
 	values.Set(peerRESTMetricsTypes, strconv.FormatUint(uint64(t), 10))
 	for disk := range opts.disks {
@@ -62,7 +62,7 @@ var (
 	madminOSInfo           = grid.NewJSONPool[madmin.OSInfo]()
 	madminPartitions       = grid.NewJSONPool[madmin.Partitions]()
 	madminProcInfo         = grid.NewJSONPool[madmin.ProcInfo]()
-	madminRealtimeMetrics  = grid.NewJSONPool[madmin.RealtimeMetrics]()
+	madminRealtimeMetrics  = grid.NewJSONPool[realtimeMetrics]()
 	madminServerProperties = grid.NewJSONPool[madmin.ServerProperties]()
 	madminStorageInfo      = grid.NewJSONPool[madmin.StorageInfo]()
 	madminSysConfig        = grid.NewJSONPool[madmin.SysConfig]()
@@ -84,7 +84,7 @@ var (
 	getLocksRPC                    = grid.NewSingleHandler[*grid.MSS, *localLockMap](grid.HandlerGetLocks, grid.NewMSS, func() *localLockMap { return &localLockMap{} })
 	getMemInfoRPC                  = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.MemInfo]](grid.HandlerGetMemInfo, grid.NewMSS, madminMemInfo.NewJSON)
 	getMetacacheListingRPC         = grid.NewSingleHandler[*listPathOptions, *metacache](grid.HandlerGetMetacacheListing, func() *listPathOptions { return &listPathOptions{} }, func() *metacache { return &metacache{} })
-	getMetricsRPC                  = grid.NewSingleHandler[*grid.URLValues, *grid.JSON[madmin.RealtimeMetrics]](grid.HandlerGetMetrics, grid.NewURLValues, madminRealtimeMetrics.NewJSON)
+	getMetricsRPC                  = grid.NewSingleHandler[*grid.URLValues, *grid.JSON[realtimeMetrics]](grid.HandlerGetMetrics, grid.NewURLValues, madminRealtimeMetrics.NewJSON)
 	getNetInfoRPC                  = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.NetInfo]](grid.HandlerGetNetInfo, grid.NewMSS, madminNetInfo.NewJSON)
 	getOSInfoRPC                   = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.OSInfo]](grid.HandlerGetOSInfo, grid.NewMSS, madminOSInfo.NewJSON)
 	getPartitionsRPC               = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.Partitions]](grid.HandlerGetPartitions, grid.NewMSS, madminPartitions.NewJSON)
@@ -423,7 +423,7 @@ func (s *peerRESTServer) GetMemInfoHandler(_ *grid.MSS) (*grid.JSON[madmin.MemIn
 }

 // GetMetricsHandler - returns server metrics.
-func (s *peerRESTServer) GetMetricsHandler(v *grid.URLValues) (*grid.JSON[madmin.RealtimeMetrics], *grid.RemoteErr) {
+func (s *peerRESTServer) GetMetricsHandler(v *grid.URLValues) (*grid.JSON[realtimeMetrics], *grid.RemoteErr) {
 	values := v.Values()
 	var types madmin.MetricType
 	if t, _ := strconv.ParseUint(values.Get(peerRESTMetricsTypes), 10, 64); t != 0 {
@@ -0,0 +1,125 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/amztime"
+	"github.com/minio/minio/internal/auth"
+	xhttp "github.com/minio/minio/internal/http"
+)
+
+// replicaDeleteRequest sends DELETE ?versionId= as site replication does.
+func replicaDeleteRequest(t *testing.T, router http.Handler, cred auth.Credentials, bucket, key, versionID string, marker bool) *httptest.ResponseRecorder {
+	t.Helper()
+	target := getDeleteObjectURL("", bucket, key) + "?" + url.Values{xhttp.VersionID: {versionID}}.Encode()
+	req, err := newTestSignedRequestV4(http.MethodDelete, target, 0, nil, cred.AccessKey, cred.SecretKey, map[string]string{
+		xhttp.MinIOSourceReplicationRequest: "true",
+		xhttp.AmzBucketReplicationStatus:    "REPLICA",
+		xhttp.MinIOSourceDeleteMarker:       strconv.FormatBool(marker),
+		xhttp.MinIOSourceMTime:              UTCNow().Format(time.RFC3339Nano),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, req)
+	return rec
+}
+
+func replicaDeleteMarkerExists(t *testing.T, obj ObjectLayer, bucket, key, versionID string) bool {
+	t.Helper()
+	oi, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: versionID})
+	return oi.DeleteMarker && (err == nil || isErrMethodNotAllowed(err))
+}
+
+// A replicated delete leaves a destination version under Object Lock as it
+// is, whether it purges that version or names it in a delete marker. The
+// replication credential is not a retention bypass on the destination.
+func TestReplicaDeleteObjectLock(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t: t, endpoints: []string{"DeleteObject"},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, cred auth.Credentials, t *testing.T) {
+			enableBucketObjectLock(t, bucket)
+			for _, kind := range batchLockKinds {
+				for _, marker := range []bool{false, true} {
+					key := "locked/" + kind + "/marker=" + strconv.FormatBool(marker)
+					before := batchPutVersion(t, obj, bucket, key, batchLockMeta(kind))
+					rec := replicaDeleteRequest(t, router, cred, bucket, key, before.VersionID, marker)
+					if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "WORM") {
+						t.Errorf("%s: %s: status %d, want 400 WORM: %s", instance, key, rec.Code, rec.Body.String())
+					}
+					after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: before.VersionID})
+					if err != nil || after.DeleteMarker || after.ETag != before.ETag {
+						t.Errorf("%s: %s: locked version was removed or replaced: %v", instance, key, err)
+					}
+				}
+			}
+
+			// Controls: replication still purges unprotected versions and
+			// writes new delete markers, also beside a locked version.
+			expired := batchLockMeta("GOVERNANCE")
+			expired[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(UTCNow().Add(-time.Hour))
+			for name, meta := range map[string]map[string]string{"unlocked": nil, "expired": expired} {
+				key := "purge/" + name
+				v := batchPutVersion(t, obj, bucket, key, meta)
+				if rec := replicaDeleteRequest(t, router, cred, bucket, key, v.VersionID, false); rec.Code != http.StatusNoContent {
+					t.Errorf("%s: %s: status %d, want 204: %s", instance, key, rec.Code, rec.Body.String())
+				}
+				if batchVersionExists(t, obj, bucket, key, v.VersionID) {
+					t.Errorf("%s: %s: version survived the replicated purge", instance, key)
+				}
+			}
+
+			const key = "marker/new"
+			locked := batchPutVersion(t, obj, bucket, key, batchLockMeta("COMPLIANCE"))
+			marker := mustGetUUID()
+			if rec := replicaDeleteRequest(t, router, cred, bucket, key, marker, true); rec.Code != http.StatusNoContent {
+				t.Errorf("%s: new marker: status %d, want 204: %s", instance, rec.Code, rec.Body.String())
+			}
+			if !replicaDeleteMarkerExists(t, obj, bucket, key, marker) {
+				t.Errorf("%s: the replicated delete marker was not written", instance)
+			}
+			if !batchVersionExists(t, obj, bucket, key, locked.VersionID) {
+				t.Errorf("%s: a new marker removed a locked version", instance)
+			}
+		},
+	})
+}
+
+// With more than one pool, the protected version may be in a pool other than
+// the one that holds the latest version and takes the delete marker.
+func TestPoolsReplicaDeleteObjectLock(t *testing.T) {
+	z, _ := consistencyPools(t)
+	bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for pool := range z.serverPools {
+		for _, marker := range []bool{false, true} {
+			key := "pool-" + strconv.Itoa(pool) + "/marker=" + strconv.FormatBool(marker)
+			before := putConsistencyObject(t, z, bucket, key, pool, "locked",
+				ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Hour), UserDefined: batchLockMeta("COMPLIANCE")})
+			putConsistencyObject(t, z, bucket, key, 1-pool, "newer", ObjectOptions{Versioned: true, MTime: UTCNow()})
+			rec := replicaDeleteRequest(t, router, globalActiveCred, bucket, key, before.VersionID, marker)
+			if rec.Code != http.StatusBadRequest {
+				t.Errorf("%s: status %d, want 400: %s", key, rec.Code, rec.Body.String())
+			}
+			// The logical lookup also finds a marker written with this ID in
+			// the other pool, which would shadow the protected version.
+			after, gerr := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: before.VersionID})
+			if gerr != nil || after.DeleteMarker || after.ETag != before.ETag {
+				t.Errorf("%s: locked version was removed, replaced or shadowed: %v", key, gerr)
+			}
+		}
+	}
+}
@@ -43,6 +43,11 @@ const (
 	// restoreTakeoverWait bounds how long a POST waits for the restore lock
 	// of its version. A running restore holds it for its whole run.
 	restoreTakeoverWait = 2 * time.Second
+	// restoreResetAttemptTimeout bounds the synchronous cleanup attempt made
+	// after a restore fails. A caller whose context has ended must not wait for
+	// another object operation to release the restore lock; the recovery loop
+	// below retries the cleanup after this attempt gives up.
+	restoreResetAttemptTimeout = 100 * time.Millisecond
 	// restoreCleanupTimeout bounds each status reset and convergence pass.
 	restoreCleanupTimeout = time.Minute
 )
@@ -639,6 +644,10 @@ func (z *erasureServerPools) recoverRestoreStatus(bucket, object string, opts Ob
 func (z *erasureServerPools) resetRestoreStatus(bucket, object string, opts ObjectOptions) error {
 	ctx, cancel := context.WithTimeout(GlobalContext, restoreCleanupTimeout)
 	defer cancel()
+	return z.resetRestoreStatusWithContext(ctx, bucket, object, opts)
+}
+
+func (z *erasureServerPools) resetRestoreStatusWithContext(ctx context.Context, bucket, object string, opts ObjectOptions) error {
 	lk := z.restoreLock(bucket, object)
 	lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
 	if err != nil {
@@ -455,7 +455,13 @@ func TestRestoreRecoversStrandedMarker(t *testing.T) {
 						t.Fatal(lerr)
 					}
 					ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+					start := time.Now()
 					err = r.z.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID))
+					// The canceled caller must not wait for the minute-long
+					// background cleanup deadline before releasing its lock.
+					if elapsed := time.Since(start); elapsed > 5*time.Second {
+						t.Errorf("canceled restore blocked on marker cleanup for %s", elapsed)
+					}
 					cancel()
 					lk.Unlock(lkctx)
 				} else {
@@ -263,7 +263,13 @@ func TestNativeRestoreWithoutOuterLock(t *testing.T) {
 			before := nativeRestoreSnapshot(t, r)
 			ctx, cancel := context.WithTimeout(t.Context(), 200*time.Millisecond)
 			defer cancel()
+			start := time.Now()
 			err = er.RestoreTransitionedObject(ctx, r.bucket, r.object, nativeRestoreOptions(r, oi.VersionID))
+			// Failure cleanup must return while this caller still holds the
+			// object lock, leaving marker recovery to the background retry.
+			if elapsed := time.Since(start); elapsed > 5*time.Second {
+				t.Errorf("restore blocked on failure cleanup for %s", elapsed)
+			}
 			if err == nil {
 				t.Error("restore wrote through another caller's lock")
 			}
@@ -793,22 +793,47 @@ func (er erasureObjects) RestoreTransitionedObject(ctx context.Context, bucket,
 		return lkctx, func() { lk.Unlock(lkctx) }, nil
 	}
 	clearMarker := func() {
-		ctx, cancel := context.WithTimeout(GlobalContext, restoreCleanupTimeout)
-		defer cancel()
-		lkctx, unlock, err := lock(ctx)
-		if err != nil {
-			storageLogIf(ctx, fmt.Errorf("Unable to clear the restore status of %s/%s(%s): %w", bucket, object, opts.VersionID, err))
+		tryClear := func(ctx context.Context) error {
+			lkctx, unlock, err := lock(ctx)
+			if err != nil {
+				return err
+			}
+			defer unlock()
+			versionID := opts.VersionID
+			if a := opts.Transition.attempt; a != nil {
+				versionID = a.VersionID
+			}
+			err = er.clearRestoreStatus(lkctx.Context(), bucket, object, versionID, opts.Transition.attempt)
+			if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isErrMethodNotAllowed(err) {
+				return nil
+			}
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(GlobalContext, restoreResetAttemptTimeout)
+		err := tryClear(ctx)
+		cancel()
+		if err == nil {
 			return
 		}
-		defer unlock()
-		versionID := opts.VersionID
-		if a := opts.Transition.attempt; a != nil {
-			versionID = a.VersionID
-		}
-		if err := er.clearRestoreStatus(lkctx.Context(), bucket, object, versionID, opts.Transition.attempt); err != nil &&
-			!isErrObjectNotFound(err) && !isErrVersionNotFound(err) && !isErrMethodNotAllowed(err) {
-			storageLogIf(ctx, fmt.Errorf("Unable to clear the restore status of %s/%s(%s): %w", bucket, object, versionID, err))
-		}
+
+		go func() {
+			var lastErr error
+			for _, delay := range restoreRecoveryDelays {
+				select {
+				case <-GlobalContext.Done():
+					return
+				case <-time.After(delay):
+				}
+				ctx, cancel := context.WithTimeout(GlobalContext, restoreCleanupTimeout)
+				lastErr = tryClear(ctx)
+				cancel()
+				if lastErr == nil {
+					return
+				}
+			}
+			storageLogIf(GlobalContext, fmt.Errorf("Unable to clear the restore status of %s/%s(%s): %w", bucket, object, opts.VersionID, lastErr))
+		}()
 	}
 	return er.restoreVersion(ctx, bucket, object, opts, restoreHooks{lock: lock, clear: clearMarker})
 }
@@ -835,7 +860,10 @@ func (z *erasureServerPools) RestoreTransitionedObject(ctx context.Context, buck
 	opts.SkipDecommissioned = true
 	opts.NoLock = true
 	clearMarker := func() {
-		if err := z.resetRestoreStatus(bucket, object, opts); err != nil {
+		ctx, cancel := context.WithTimeout(GlobalContext, restoreResetAttemptTimeout)
+		err := z.resetRestoreStatusWithContext(ctx, bucket, object, opts)
+		cancel()
+		if err != nil {
 			z.recoverRestoreStatus(bucket, object, opts)
 		}
 	}
@@ -29,11 +29,15 @@ healthcheck:
  retries: 5
 ```

+`ready` can return success before storage and IAM finish loading. It checks the process and, when configured, KMS and etcd; it does not certify S3 data access. Keep this bootstrap behavior so distributed peers can discover and connect to one another. Kubernetes headless peer services must retain `publishNotReadyAddresses: true`.
+
+`cluster` checks initialization and reported write quorum, but a restarted node can still be reconnecting its remote drives after that check succeeds. Neither endpoint is a sufficient gate for an application that immediately reads or writes S3. Compose `depends_on: condition: service_healthy` inherits the image probe's limited meaning. Keep the node probe, and gate the dependent application with a bounded signed S3 check: write a uniquely named small object through each node, then read every acknowledged object through every node and verify its bytes and version. Retry within a fixed overall deadline with per-request timeouts, and clean up the probe objects. Failure should stop the rollout or dependent startup. This samples the tested keys and erasure sets; it does not prove spare failure capacity or that every drive is online. See the [startup window](#startup-readiness-window) below.
+
 `silo healthcheck --maintenance cluster` answers the pre-drain question documented below: exit `0` when the node can be taken down safely, exit `1` (HTTP 412) when doing so would lose HA. Keep the `cluster` checks out of per-container liveness probes — they reflect cluster-wide quorum, not this process.

 ## Liveness probe

-This probe always responds with '200 OK'. Only fails if 'etcd' is configured and unreachable. When liveness probe fails, Kubernetes like platforms restart the container.
+This probe returns `200 OK` while the process accepts requests, including before the object layer initializes. It can return `503` under request-queue saturation and does not check KMS or etcd. Kubernetes uses repeated liveness failures to restart the container, so configure its startup budget and failure threshold accordingly.

 ```
 livenessProbe:
@@ -50,7 +54,7 @@ livenessProbe:

 ## Readiness probe

-This probe responds with '200 OK' once the server process is up: it fails only while the request queue is overloaded, or when the health operation against the configured KMS or 'etcd' fails. When readiness probe fails, Kubernetes like platforms turn-off routing to the container.
+This probe responds with '200 OK' once the server process is up: it fails only while the request queue is overloaded, or when the health operation against the configured KMS or 'etcd' fails. It does not wait for storage or IAM to load, because a distributed node must be reachable by its peers before it can load them; the headless service of a StatefulSet should still publish not-ready addresses. When readiness probe fails, Kubernetes like platforms turn-off routing to the container.

 ```
 readinessProbe:
@@ -76,7 +76,7 @@ require (
 	github.com/minio/xxml v0.0.3
 	github.com/minio/zipindex v0.5.0
 	github.com/mitchellh/go-homedir v1.1.0
-	github.com/nats-io/nats-server/v2 v2.15.1
+	github.com/nats-io/nats-server/v2 v2.15.2
 	github.com/nats-io/nats.go v1.55.0
 	github.com/nats-io/stan.go v0.10.4
 	github.com/ncw/directio v1.0.5
@@ -122,8 +122,8 @@ require (
 	aead.dev/minisign v0.3.0 // indirect
 	cel.dev/expr v0.25.3 // indirect
 	cloud.google.com/go v0.123.0 // indirect
-	cloud.google.com/go/auth v0.24.1 // indirect
-	cloud.google.com/go/auth/oauth2adapt v0.3.0 // indirect
+	cloud.google.com/go/auth v0.24.3 // indirect
+	cloud.google.com/go/auth/oauth2adapt v0.3.1 // indirect
 	cloud.google.com/go/compute/metadata v0.10.0 // indirect
 	cloud.google.com/go/iam v1.15.0 // indirect
 	cloud.google.com/go/monitoring v1.32.0 // indirect
@@ -255,9 +255,9 @@ require (
 	github.com/muesli/reflow v0.3.0 // indirect
 	github.com/muesli/termenv v0.16.0 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	github.com/nats-io/jwt/v2 v2.8.2 // indirect
+	github.com/nats-io/jwt/v2 v2.8.3 // indirect
 	github.com/nats-io/nats-streaming-server v0.25.6 // indirect
-	github.com/nats-io/nkeys v0.4.16 // indirect
+	github.com/nats-io/nkeys v0.4.17 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/oklog/ulid/v2 v2.1.2 // indirect
 	github.com/olekukonko/tablewriter v0.0.5 // indirect
@@ -9,10 +9,10 @@ cel.dev/expr v0.25.3 h1:A2jO8jwOugrrovveCWfj0KEZOfqiLgAcwjpHPhzIGw0=
 cel.dev/expr v0.25.3/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4=
 cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE=
 cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU=
-cloud.google.com/go/auth v0.24.1 h1:NTmd2iJb4DsGIav0KiqFMYI88nMshA2wnFMBatRmoCs=
-cloud.google.com/go/auth v0.24.1/go.mod h1:1dmBXa9xfBvBr39ZnDUDkaaA91ccq9CcqENDgJA4d7c=
-cloud.google.com/go/auth/oauth2adapt v0.3.0 h1:FY8oSZpCYoUNv6QxVODuMjQz4IlSOVeiQtZ08vLPz88=
-cloud.google.com/go/auth/oauth2adapt v0.3.0/go.mod h1:7+2uCm7++XFO+/lN06c2HXpDXb/NMNn2/UwyBPbTnkk=
+cloud.google.com/go/auth v0.24.3 h1:oVmRvR/pGSCNXjviwXCisJjRSEdQ0BlcOpW0uhuHLm8=
+cloud.google.com/go/auth v0.24.3/go.mod h1:p4UXqfSXnMVUDutk9xePzI8lj8yyoJ3BLsKWv0NSvQk=
+cloud.google.com/go/auth/oauth2adapt v0.3.1 h1:BKLDdbWS7pafd820dsyFDF6fPEzIGqr7ny6hbTzCCgE=
+cloud.google.com/go/auth/oauth2adapt v0.3.1/go.mod h1:B1nM4FJDxoDqAt3PXFG4IflUoQ5kvWAOOM3R+5wAszg=
 cloud.google.com/go/compute/metadata v0.10.0 h1:pyKMUQSwchgkIBBJGdILqQbs/BNJXqwSA7Ej6LAvvtY=
 cloud.google.com/go/compute/metadata v0.10.0/go.mod h1:rGFHRrIif570kSibjFTMbt6/4/tzgJWFGI/HVol4GIk=
 cloud.google.com/go/iam v1.15.0 h1:U/wqjfrM0ceBFdPlOI2+cu68yPAxh72M8fe9uLG0+ZE=
@@ -487,18 +487,18 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc
 github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
-github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc=
-github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc=
-github.com/nats-io/nats-server/v2 v2.15.1 h1:fK3AeYFIji1Smng60jB0hVoAPZezA/UzsaI3BqXFNEI=
-github.com/nats-io/nats-server/v2 v2.15.1/go.mod h1:yeuYRtKVKIFtmMz2/YPk+6hvmNBnk+WwTUzErBdH6IE=
+github.com/nats-io/jwt/v2 v2.8.3 h1:pqImTmlqH2zPHKNdRXO0YRyWif2diGRlrtLFJ2RxM2I=
+github.com/nats-io/jwt/v2 v2.8.3/go.mod h1:OOY2rP/UU8tcfoqhbHmBKcNdpFMKvmxIkmFFa16M4wI=
+github.com/nats-io/nats-server/v2 v2.15.2 h1:yLsgt2kslgymvxvjFq1WGWck5hkat2YYfbXvQ145tzU=
+github.com/nats-io/nats-server/v2 v2.15.2/go.mod h1:VXHCZgDglV5wO2B8ZE58yG2HzqxQxYUbQqibUh5Xcr8=
 github.com/nats-io/nats-streaming-server v0.25.6 h1:8OBRaIl64u+DFvZYpF50RRzwG/yLcJZL0R7VMc7tp4Y=
 github.com/nats-io/nats-streaming-server v0.25.6/go.mod h1:LEcu6uGSDBB4O/IBUsDBHYk/O0K7XZza8nMjCIXicLk=
 github.com/nats-io/nats.go v1.22.1/go.mod h1:tLqubohF7t4z3du1QDPYJIQQyhb4wl6DhjxEajSI7UA=
 github.com/nats-io/nats.go v1.55.0 h1:Ng8QKHdvdHIWoaiq0+d/aH1BWkxvcJxWJ0Y/9BPKx3M=
 github.com/nats-io/nats.go v1.55.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
-github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
-github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
+github.com/nats-io/nkeys v0.4.17 h1:JAJrlDhOyQqOt0a1eg4cbIZ1Kc7+vmkCigAltpwQpVA=
+github.com/nats-io/nkeys v0.4.17/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/nats-io/stan.go v0.10.4 h1:19GS/eD1SeQJaVkeM9EkvEYattnvnWrZ3wkSWSw4uXw=
@@ -1,8 +1,8 @@
 apiVersion: v1
 description: S3-Interface Libre Object Storage
 name: silo
-version: 7.0.3
-appVersion: RELEASE.2026-09-16T00-00-00Z
+version: 7.0.4
+appVersion: RELEASE.2026-11-09T00-00-00Z
 keywords:
  - silo
  - storage
@@ -15,7 +15,7 @@ clusterDomain: cluster.local
 ##
 image:
  repository: pgsty/silo
-  tag: RELEASE.2026-09-16T00-00-00Z
+  tag: RELEASE.2026-11-09T00-00-00Z
  pullPolicy: IfNotPresent

 imagePullSecrets: []
@@ -25,7 +25,7 @@ imagePullSecrets: []
 ##
 mcImage:
  repository: pgsty/mc
-  tag: RELEASE.2026-10-26T00-00-00Z
+  tag: RELEASE.2026-11-09T00-00-00Z
  pullPolicy: IfNotPresent

 ## Silo mode, i.e. standalone or distributed.
Author	SHA1	Message	Date
Feng Ruohang	6ef3a52fc3	Enforce Object Lock during S3 bulk version deletion Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 23:53:43 +08:00
Feng Ruohang	e21b1c3a71	docs(healthcheck): say what ready does not wait for The ready probe turns healthy before storage and IAM load, by design, so that distributed nodes can reach each other during startup. Point startup ordering such as Compose service_healthy at the cluster check, and keep publishing not-ready addresses for StatefulSet peers. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 22:07:59 +08:00
Feng Ruohang	86dc6ee2e2	fix(restore): return from a failed restore without waiting for its cleanup A failed or canceled native restore cleared its ongoing marker under a fresh one-minute deadline, while the restore itself could still hold the object lock that the cleanup waits for. The request then stalled for about a minute (60 s in TestNativeRestoreWithoutOuterLock and TestRestoreRecoversStrandedMarker). Try the cleanup for 100 ms and leave the rest to the bounded background retries, which already exist for the pools path and now also serve the single-set path. Each retry still takes the object lock afresh and clears only the marker of its own attempt. The CI restore race step drops from about 260 s to 37 s. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 22:07:59 +08:00
Feng Ruohang	0a238589a4	ci: gate releases on the new Object Lock delete tests The release workflow keeps its Object Lock race gate aligned with the matching step in go.yml. Add the replicated-delete and directory bulk delete tests there too. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 21:26:02 +08:00
Feng Ruohang	1f731549e4	docs: prepare SILO 20261109 Record RELEASE.2026-11-09T00-00-00Z (package 20261109000000.0.0) in the changelog: the fixes since 20261026, Console v2.7.0, mcli 20261109 and silo-pkg v3.16.0, upgrade notes and known issues. Point the README quick starts at the new image, replace the 20261026 release section with the 20261109 upgrade notice in both languages, refresh the contributor roll to 2026-11-09, and add a second security contact. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 21:05:22 +08:00
Feng Ruohang	368c52f4dc	build(deps): embed the published Console v2.7.0 Console v2.7.0 was cut again at 4e91ae506e70 with the review fixes that followed 26ce922424f1. Select that commit; no other module changes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 21:02:16 +08:00
Feng Ruohang	f0a4d76ef8	Merge #328 : Console 2.7, mcli 20261109 and silo-pkg 3.16 Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 21:00:20 +08:00
Feng Ruohang	aa2d79daf2	fix(object-lock): check a directory object's lock before a bulk delete A directory object always has the null version. DeleteObjects without a version ID chose whether to read the object before it turned the version into null, so with Object Lock and no replication or tiering it checked retention against empty metadata and removed a directory object under legal hold or retention. Choose the null version before the lookup. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 19:25:33 +08:00
Feng Ruohang	7e9ee6befc	fix(replication): keep Object Lock on a replicated delete A trusted replica DELETE skipped the destination's retention check for every request, not only for the delete marker it is meant to create. A caller allowed s3:ReplicateDelete (root, or the replication account) could purge a version under legal hold or unexpired retention, or name a stored version in a delete marker, which replaces that version. Skip the check only for a delete-marker creation, whose version is not yet stored here, and make that request refuse to replace a protected version with RefuseLockedOverwrite, as batch replication pulls do; the pools layer applies it to the version in any pool. A replicated purge goes through the normal retention check. A source purge that bypassed GOVERNANCE retention is now refused by a target whose copy still carries it; the source records the purge as failed and retries. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 19:13:21 +08:00
Feng Ruohang	47275042f6	chore(helm): deploy SILO and mcli 20261109 by default The 20261026 release moved mcImage.tag but left image.tag, which the templates deploy, at 20260916 (#296). Point the server and mcli images and appVersion at RELEASE.2026-11-09T00-00-00Z and bump the chart to 7.0.4. The Helm check now reads the chart version instead of a fixed file name, and fails when image.tag differs from appVersion, so a release cannot move one without the other again. Upgrades that pass --reuse-values keep the old image.tag; set image.tag explicitly when upgrading such a release. Fixes #296 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 19:03:32 +08:00
Feng Ruohang	d66b3889cd	ci: give the native restore race tests room to finish The lifecycle expiry and native restore race step takes about 260 s on an M-series Mac against its 5 minute limit, and the job holding it was already near 9 of its 15 minutes before that step was added. GitHub's hosted runners are slower, so allow the step 10 minutes and the job 25, as the other jobs have. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 19:03:32 +08:00
Feng Ruohang	099b7ddf37	fix(replication): handle absent pool in worker metrics Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 18:10:19 +08:00
Feng Ruohang	c054569b9f	Update SILO to Console 2.7 and maintained dependency fixes Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 17:24:59 +08:00
Feng Ruohang	1f579afeed	Merge origin/main: conditional multipart initiation (#292 ) Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 13:39:10 +08:00
Feng Ruohang	b14dc454fb	Merge origin/main: one trust store for both server images, mcli notices (#295 ) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-09 09:47:04 +08:00
Feng Ruohang	bb31c26183	fix: close IAM loading and batch recovery review gaps Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-08 18:52:50 +08:00
Feng Ruohang	07ce95dfda	fix(iam): protect stale creates and cache loads Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-08 14:19:49 +08:00
Feng Ruohang	7eed95bc8e	fix(iam): drop a stale cached identity when its edit is refused `70914759e` refused an edit whose stored record no longer matched the caller's view, but left the stale copy in the serving server's cache in two cases: a service-account edit bound to a parent whose account was deleted and created again for another parent, and a replicated edit older than what is stored, which site replication acknowledges. That server kept authenticating the old credential until a notification or the periodic load. Both refusals now drop the cache entry of the addressed kind, as a refusal for an absent or deleted identity does: the service-account closure reports such a record as gone (callers still get errNoSuchServiceAccount), and editStoredIdentity also drops the entry on errIAMStaleUpdate. The LDAP group refresh again treats a stale update as a refusal that writes nothing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-08 14:18:41 +08:00
Feng Ruohang	70914759e5	fix(iam): edit users and service accounts on the stored record, not a cached copy UpdateServiceAccount, SetUserStatus and UpdateUserSecretKey read the identity from the serving server's cache, changed it and wrote it back whole. On a server that missed a notification, an ordinary edit (a description, a status, a secret) undid a rotation made through another server and brought back a user or service account deleted meanwhile, with its old secret. Edits now go through iamIdentityEdit, inside the identity's cross-node revision lock: the stored record is loaded and normalized as the loaders do (access key from the path, claims from the token, a deprecated comment for an empty description), the edit applies to it, and an absent or deleted identity is refused and dropped from the cache of the server that tried. The LDAP group refresh of `a0de92fcb` uses the same mechanism. - The admin handler binds a service-account edit to the parent it authorized: an account deleted and created again for another parent meanwhile is not edited (parentUser in updateServiceAccountOpts; replication does not set it). - A replicated edit older than a deletion this site applied is answered as stale, so site replication acknowledges it instead of retrying. - SetUserStatus refuses a service or temporary credential as before, and a refused edit drops only a cache entry of the addressed kind. - A session-policy edit stores the claims of the token it signs; the token is signed once. - UpdateUserSecretKey addressed to a service-account key finds no user instead of writing the account to the regular-user path. Tests use a second IAM store over the same storage whose cache missed the change: rotation, deletion, status and secret edits of users, parent binding, legacy records, replicated edits around a deletion, and status calls on a service account. Against the cache-based writers every case fails. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-08 13:56:04 +08:00
Feng Ruohang	a0de92fcbe	fix(iam): refresh LDAP groups on the stored credential, not a stale snapshot The hourly LDAP group refresh, which every server runs, reads the STS and service-account credentials from its own cache, queries the directory, and wrote each credential whose groups changed back whole through UpdateUserIdentity. A secret rotation or an edit made meanwhile, on this server or one whose notification it missed, was undone, and a credential deleted meanwhile came back enabled with its old secret. RefreshUserGroups replaces that writer, whose only caller was the refresh. Under the identity's cross-node revision lock it loads the stored record, refuses one that is absent, deleted, expired or now has another parent (whose groups the lookup did not compute), and otherwise changes only the groups of the stored record. A record without its access key gets it from the path, as the loaders do. A refused refresh drops the stale entry from the cache, which loads what is stored on next use; a written or unchanged one caches the stored record. An STS record keeps the lease of its token's remaining lifetime. Tests cover rotation and edit, deletion, a sibling server whose cache did not hear of a rotation or a deletion, recreation under the same and under another parent, unchanged groups, and a stored record without its access key. Written back as the old snapshot write, every case fails. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-08 12:00:20 +08:00
Feng Ruohang	af0afd2321	test(batch): pin the parity of the tiered purge transition helper TestBatchExpirePurgeRefusesTieredKey marks a version transitioned through DecomTieredObject, which takes its parity from the global storage class. Run after tests that leave a parity the one-drive fixture cannot have, it wrote on no drive and still returned nil, and the purge then removed the key. The helper now sets the set's own parity for the test and reads the transition back from every drive. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-08 11:39:03 +08:00
Feng Ruohang	633c9bfe09	fix(batch): keep a job across edits that keep its credential The submitter binding compared the UpdatedAt of the identity record, so any rewrite of the record ended every running job of the credential with "changed since submission": a service account's description, name or later expiry, and the hourly LDAP group refresh of every LDAP-backed service account and STS credential whose groups changed. Compare only what identifies the credential: its type, access key and parent, the revocation boundary of the record, and the keyed binding of its secret and session token. Status, expiry, groups and policies are read again on every start and action. A deletion advances the revocation boundary, so a user created again under the access key with the same secret still fails the binding, and adding an expiry to a service account turns it into a temporary credential, which still ends the job. UpdatedAt stays in the binding for diagnosis. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-08 11:39:03 +08:00
Feng Ruohang	18e09ca219	fix(batch): refuse a job only for a Deny that needs request context A batch job is authorized without a request, so a policy statement that depends on request context the job does not have (source address, transport, headers) was taken to refuse the action. Two parts of that rule refused jobs that no request would refuse: - The scan for unsupplied policy variables ran over the whole statement before its resources were matched. A Deny on an unrelated bucket whose condition value held a variable such as ${aws:UserAgent} refused every job of the user. Resources are now matched first; a variable in a resource still means the statement may apply. - An Allow that needs absent context, such as an IP-restricted read, refused the job even when another Allow granted the action. Such an Allow now grants nothing to the job: it is dropped from the policy that is evaluated, not merely exempted, since IfExists and Null conditions hold for an absent key and would grant. Dropping an Allow can only narrow the decision. A Deny that needs the context still refuses. The narrowed policy is a new policy.Policy, because the statement index of the original points into its own statements and the original may be a cached document. For a session policy, whether one applies and its first decision still come from the claims as before; only an allow is re-evaluated on the narrowed policy. Tests cover an unrelated Deny with a variable, a Deny with an unsupplied variable in its resource, an IP-restricted Allow next to a plain one, a Null Allow that plain evaluation without the address would grant, and the same for session policies. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-08 11:39:01 +08:00
Feng Ruohang	a6571a53e9	chore(batch): keep the policy snapshot out of msgp generation batchPolicySnapshot holds the IAM store, a context and a mutex. Without an ignore directive, go generate emitted msgp encoders for it that do not compile, so make check-gen (and the check-gen step of CI) failed on the generated batch-authz files. The committed generated files are unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Feng Ruohang <rh@vonng.com>	2026-11-08 11:38:59 +08:00