feat(scanner): observe cache read origins

* fix(ecstore): select the latest version at read quorum like MinIO

GET and HEAD escalated latest-version selection to write quorum when
enough metadata answered and skipped a newer version below write quorum
on degraded reads. LIST and heal already pick the newer side of a split
write at read quorum, so such keys listed but answered 503 until heal.

Select with MinIO's getObjectFileInfo rule: the most common mod time
that reaches the object's data blocks wins, ties go to the newer time,
and one content identity at that time must reach the same quorum. Two
identities tied at the selected time fail instead of resolving by hash
order, and the selected identity's version-stack shape follows disk
order.

* fix(ecstore): resolve selection ties by disk order and guard writes

Review of the read-quorum selection found three gaps:

- Rejecting a same-time half split made explicit-version reads of a
  partially applied tagging update fail and blocked the retry that
  repairs it. Resolve ties to the identity on the lowest-index disk,
  for reads and heal alike, instead of rejecting or using map order.
- Delete markers and empty versions can lower the common read quorum
  below the selected version's data blocks, so HEAD could report a
  version GET cannot decode. Require that many copies, as MinIO's
  getLatestFileInfo does.
- A selection can now return exactly read-quorum disks, and partial
  metadata writes are not rolled back. Refuse an update before writing
  when its target disks cannot reach write quorum.

* fix(ecstore): count identity copies in the data-blocks guard

The first-candidate shortcut counts byte-identical copies, so copies
that differ only in identity-neutral fields such as replication status
were missing from the data-blocks check. When the shortcut's count
falls short, count every copy of the selected content identity before
refusing the version.

* fix(ecstore): keep commit-proof selection for system-bucket reads

Pool metadata's PREPARE/COMMIT recovery and identity epochs treat an
ordinary read of pool.bin as proof that the record committed. Under the
read-quorum rule, a COMMIT that reached only half of a 2+2 set was served
as committed, and the pool_meta_native_physical_* integration tests
failed.

Route reads through select_fileinfo_for_read: S3 buckets keep MinIO's
read-quorum rule, while is_meta_bucketname reads use the previous
algorithm unchanged as select_committed_fileinfo, so a split stays
unreadable until native recovery resolves it.

* fix(ecstore): serve the latest readable version past dangling versions

A failed versioned PUT whose rollback also failed can leave the new
version on top of fewer drives than its data blocks. Reading only each
drive's top version then left the committed version below it short of
read quorum, so GET and HEAD of the latest version answered 503 although
that version is intact in every stack. MinIO resolves the stacks in
pickLatestQuorumFilesInfo.

When every drive answered and an S3 latest read still cannot select a
version, walk the newest version headers of each stack through
exact-version reads: serve the readable version with the newest selected
modification time, skip versions whose exact read proves them dangling
(the NoSuchVersion an explicit GET reports) and queue heal for them, and
keep the read-quorum error on any other outcome. Candidates are ordered
by each version's newest modification time across drives, and at most
eight exact reads are spent.

Test PUTs now wait for every drive so tests that edit one drive's
metadata no longer race the PUT's background tail writes.

* fix(ecstore): resolve dangling latest versions in metadata updates

A failed versioned PUT whose rollback also failed can leave the new
version on fewer drives than its data blocks. GET and HEAD of the
latest version already look past such versions, but read-before-write
callers without a version id still answered 503: object tagging read
each drive's top version with the quorum-error read, and
put_object_metadata and the CopyObject source read selected from
read_all_xl, whose merged latest version is missing on the other
drives, so object_quorum_from_meta failed before selection.

Object tagging now reads through a variant that runs the same walk past
dangling versions as a latest GET while keeping the quorum error for a
dangling key. put_object_metadata and the CopyObject source try the walk
when their quorum or selection step fails with read quorum, before the
existing dangling cleanup could report a key with a committed version
absent. A version-only copy takes the served version's own copies and
write quorum. The walk still applies only to S3 buckets, latest reads,
and sets where every drive answered, and the update still targets only
the served version's disks, so a version below write quorum is refused
before writing.

* fix(ecstore): refuse version-only copies below write quorum up front

Review of the dangling-latest metadata updates found three gaps:

- A version-only CopyObject wrote the new version and reverted on a
  write-quorum miss; refuse it before writing when the served version's
  disks cannot reach its write quorum, like update_object_meta_with_opts.
- Replication status write-back, including the zero-write ValidateOnly
  mode, now keeps its read-quorum error and never walks past dangling
  versions, since the walk queues heal for the versions it skips.
- The erasure-coding doc overstated that these callers never report a
  dangling key as absent; put_object_metadata and CopyObject keep their
  existing dangling cleanup when the walk finds no readable version.

.agents/references/adversarial-validation.md

@@ -27,7 +27,4 @@ Risk and review shape:
   and list verified and unverified target classes in the PR Impact section.
 
 Available domain lenses are security, concurrency/durability, compatibility,
-and performance. Select `.agents/skills/adversarial-validation/SKILL.md` for an
-explicit adversarial request, a high-risk change, or a substantial PR review;
-then read only its matching role references. A routine standard pass does not
-load the playbook unless the reviewer needs a RustFS-specific probe.
+and performance; select only those matched by the change.

.agents/skills/adversarial-validation/SKILL.md

@@ -1,48 +0,0 @@
----
-name: adversarial-validation
-description: Review RustFS diffs or designs for explicit adversarial requests, high-risk changes under the repository review policy, or substantial PR reviews. Skip ordinary questions, diagnosis, planning, status, routine low-risk implementation, and prose with no execution effect.
----
-
-# RustFS Adversarial Validation
-
-Use the [repository risk tiers and review shape](../../references/adversarial-validation.md). This skill
-routes a review to RustFS-specific probes without loading unrelated domains.
-
-## Select Lenses
-
-Read only the references required by the diff:
-
-| Lens | When to read |
-|---|---|
-| [Correctness](references/correctness.md) | Every non-exempt adversarial review |
-| [Simplicity](references/simplicity.md) | Mechanical/standard changes and production growth |
-| [Test coverage](references/test-coverage.md) | Behavior or test changes |
-| [Security](references/security.md) | Authn/authz, IAM, RPC trust, paths, secrets, parsing, browser, encryption |
-| [Concurrency/durability](references/concurrency-durability.md) | Async shared state, locks, storage commit, cancellation, persisted queues |
-| [Compatibility](references/compatibility.md) | S3 surface, MinIO interop, metadata, wire/disk formats, mixed versions |
-| [Performance](references/performance.md) | Request/object hot paths, allocation, blocking work, fsync, fan-out |
-
-Do not read all references as a precaution. A path name alone is insufficient;
-the changed behavior must touch the lens's domain.
-
-For a dedicated security audit or advisory analysis, use
-`security-advisory-lessons` instead of loading it automatically during every
-adversarial review.
-
-## Review Protocol
-
-1. Freeze the exact final diff/head (or the design under review) and list the
-   selected lenses.
-2. Run the review shape required by the repository risk tier.
-3. For each selected lens, either report a concrete finding or a null verdict
-   naming the attacks performed.
-4. Apply root `AGENTS.md`'s finding standard. Test each candidate against callers,
-   existing coverage, and invariants before accepting it; an adversarial role
-   does not have to produce a defect.
-5. Fix or rebut supported findings with code-path, test, or invariant evidence.
-6. After a non-trivial edit, rerun only lenses affected by that edit against the
-   new exact diff.
-
-Do not turn a null verdict into a long checklist. Record concise evidence that
-the relevant failure classes were attacked, then stop under the root completion
-rule. Keep the required per-lens verdicts for high-risk PRs.

.agents/skills/adversarial-validation/references/compatibility.md

@@ -1,38 +0,0 @@
-# Compatibility Lens
-
-- Internal metadata uses `metadata_compat` helpers for dual RustFS/MinIO keys,
-  including mixed casing and removal of both twins.
-- Binary UUID metadata treats absent, empty, and nil as no value. Unversioned
-  remote tiers receive no `versionId`; versioned purge requests retain the real
-  version ID.
-- `xl.meta` changes preserve supported header/meta versions, recompute
-  signatures, decode legacy fixtures, and remain readable by old RustFS/MinIO.
-- Foreign/corrupt metadata validates parallel array lengths and missing fields;
-  it returns a decode error rather than indexing, panicking, or fabricating data.
-- Do not “correct” byte-for-byte MinIO ports without legacy fixture evidence.
-  Bitrot framing, shard math, distribution, and inline prefixes are contracts.
-- Client-visible metadata/events strip both internal prefixes
-  case-insensitively.
-- Proto fields are appended, never reused/renumbered; FlatBuffers tables extend
-  compatibly and absent new fields fail closed where authorization/quorum is
-  involved.
-- Replay real client request shapes and exact pagination boundaries for S3
-  handler changes.
-- Bucket metadata/IAM/config parsing remains compatible with pinned real MinIO
-  fixtures and encrypted migration data.
-- Compatibility shims use `RUSTFS_COMPAT_TODO(<task-id>)`, have a removal
-  condition, and default toward reading old data safely.
-
-## Outbound targets
-
-- A change to what the replication or migration client sends by default
-  (checksum policy, payload framing, headers, version-id addressing) is judged
-  against every target class, not the one it fixes. Name each target-side rule
-  the current default satisfies — checksum required with Object Lock
-  parameters, `aws-chunked` decoding, version-id adoption, ETag equals content
-  MD5 — and show which cell of
-  `crates/e2e_test/src/replication_target_matrix_test.rs` covers each.
-- A test that asserts the fix ("no trailer header") is not evidence; the
-  matrix cell that asserts the target accepted and stored the object is.
-- Every new environment escape hatch appears in
-  `docs/operations/replication-outbound-transport.md` in the same diff.

.agents/skills/adversarial-validation/references/concurrency-durability.md

@@ -1,27 +0,0 @@
-# Concurrency and Durability Lens
-
-- For every changed lock, enumerate overlapping lock sets and construct the
-  ABBA interleaving. Multiple-lock order must be documented and consistent.
-- Mark guard lifetimes and every `.await`, disk,
-  and RPC call; estimate contention/timeouts. Tokio read/write guards across
-  `.await` need bounded hold time; long-lived reads can wedge writers (#4195).
-  Keep `std::sync::Mutex` holds brief and never across `.await`.
-- Use direct `fetch_*` for unconditional atomic read-modify-write and
-  `compare_exchange` for conditional updates.
-- Object commits remain fenced if the distributed lock is lost after shard
-  writes and before metadata rename.
-- For write/rename changes, trace `write tmp -> sync tmp -> rename -> sync parent
-  -> sync required ancestors`; simulate a crash after each step and honor the
-  configured durability gate.
-- Multi-disk fan-out counts every result. Quorum-minus-one cannot become success;
-  heal remains best-effort per target where that is the established contract.
-- At every new cancellable await between mutation and cleanup/commit, drop the
-  future and inspect leftover files, counters, permits, and replay state.
-- Multipart operations on the same upload ID are serialized where required;
-  abort/complete/list races cannot delete parts before durable commit.
-- Post-commit cleanup is best-effort, retry-safe, and cannot fail an already
-  committed write or delete the last surviving copy.
-- Persisted read-modify-write uses serialization/CAS. Queue replay is crash-safe
-  and duplicate delivery has an idempotency contract.
-- Streaming reconstruction failures after partial output surface as errors, not
-  successful EOF.

.agents/skills/adversarial-validation/references/correctness.md

@@ -1,31 +0,0 @@
-# Correctness Lens
-
-Attack the changed behavior, not every subsystem in the repository.
-
-- Trace new error paths to the caller. Inject the ignored/wildcard variants and
-  verify they cannot become success, not-found, or a plausible default.
-- Exercise zero/empty/missing, maximum, and exact-boundary inputs for every
-  changed count, size, index, page limit, or optional value.
-- For aggregation/quorum changes, test exactly quorum and quorum-minus-one with
-  mixed disk errors and nil/placeholder entries.
-- For listing/pagination, test `n == max`, `n == max + 1`, delimiter folding,
-  continuation markers, and object/prefix name collisions.
-- For EC/read/streaming changes, inject failure after partial output and verify
-  the client receives an error rather than a clean truncated body. Assert exact
-  bytes and length.
-- For multipart/object commits, fail before/after rename and cleanup; committed
-  data must remain readable and pre-commit cleanup must not destroy parts.
-- Bound recursion over untrusted or persisted input; handle corrupted/cyclic
-  tree and cache traversals safely.
-- For version/index ordering, test `len - 1`, `len`, equal timestamps, missing
-  versions, and deterministic tie-breaking.
-- For directory-object behavior, trace `__XLDIR__` at the store layer; branches
-  below the layer that sees trailing slashes are dead.
-- For binary UUID metadata, absent, empty, and nil all mean no value. Never send
-  nil/empty `versionId` to an unversioned tier.
-- For agent rules/skill routers, test a trigger matrix covering ordinary
-  inquiry, low-risk implementation, explicit review, high-risk code, PR
-  creation, release, and post-PR monitoring. Each case must select only the
-  intended workflow and retain required safety/authorization boundaries.
-
-Null verdicts name only the probes relevant to the diff.

.agents/skills/adversarial-validation/references/performance.md

@@ -1,34 +0,0 @@
-# Performance Lens
-
-- For added clones/allocations on request/object/block paths, quantify copied
-  data and frequency. Recommend borrowing, move, `Bytes`/`Arc`, `Cow`, or
-  capacity reservation only for a concrete repeated cost.
-- Route every new sync/flush through the durability-mode and bucket override
-  gates; mode `none` must not pay the new fsync.
-- Keep blocking filesystem/CPU work off async runtime threads, but do not split
-  one small operation into many `spawn_blocking` round trips; coalesce `open`,
-  `read`/`write`, `fsync`, and `close` into one blocking closure per disk. When
-  coalescing steps across execution lanes, route the combined closure to the
-  dedicated fsync lane whenever any step performs file or directory sync so
-  stalled syncs cannot starve foreground metadata reads.
-- Check for per-disk syscall amplification (`disks × parts/files`): pin a
-  directory once (`openat2` on Linux) and reuse the handle across sibling reads
-  or same-directory renames instead of re-walking ancestor components per file
-  or re-opening the same parent directory twice.
-- When the file size is already known from `fstat` metadata, pre-size the
-  destination buffer (`len + 1`) before reading to EOF so small metadata reads
-  complete in a single `read()` + EOF check rather than exponential-growth
-  reallocations and syscalls.
-- Avoid redundant disk re-reads or existence checks under the same exclusive
-  lock when the state or raw bytes were just loaded or persisted.
-- Measure lock hold time across I/O and compare acquisition order for ABBA.
-- Keep cleanup, extra stat/rename, and diagnostics out of the PUT commit critical
-  section when they need not be there.
-- Detect per-item serial I/O/RPC in batch APIs and accidental quadratic scans;
-  use a gate or bounded concurrency when the concrete fan-out warrants it.
-- Count buffer growth and byte copies in EC/bitrot paths; preserve pool gauge
-  balance and avoid repeated metadata decode/fetch per object.
-- Repetitive success logs stay at `trace`; metrics/instrumentation on hot paths
-  require an existing gate.
-- Claims of no impact on PUT/GET/commit/erasure paths need relevant benchmark or
-  A/B evidence, especially for 4 KiB objects.

.agents/skills/adversarial-validation/references/security.md

@@ -1,31 +0,0 @@
-# Security Lens
-
-Use `security-advisory-lessons` only for a dedicated advisory/security audit.
-For an ordinary matched diff, attack these boundaries:
-
-- Admin routes: route registration, whitelist, handler authn, and the exact
-  `AdminAction` must agree. Read-only diagnostics still require admin authz.
-- IAM/service accounts: treat parent, claims, keys, groups, status, and policy
-  names as attacker-controlled; prove ownership/root authority before writes.
-- Protocol frontends: every changed/sibling command authorizes the matching S3
-  action before reaching storage.
-- Secrets/signatures: use constant-time comparison, normalize public failures,
-  keep RPC/root/STS keys independent, and fail closed when secrets are absent.
-- RPC: bind signatures to the exact method/path and timestamp; reject replay,
-  stale, malformed, truncated, and invalid-enum payloads without panic.
-- Paths/object/archive entries: reject traversal, absolute/platform escapes,
-  and normalization differences between authz and storage.
-- Copy/multipart/presigned POST: enforce source, destination, version-aware
-  actions, copy-source conditions, and every signed policy condition.
-- Logging/errors: never expose credentials, tokens, expected signatures, raw
-  secret-bearing input, or merged configs—including via `Debug` and parse errors.
-- Untrusted serde: reject unknown fields where compatible and validate
-  security-critical defaults/ranges before numeric conversion.
-- SSE/browser/CORS/trusted proxy: inspect stored ciphertext and wrapper order;
-  isolate user content; never reflect credentialed arbitrary origins or trust
-  forwarded identity from direct clients.
-- Object Lock: unreadable/fabricated/unparsable metadata fails closed across
-  foreground, lifecycle, scanner, and force-delete paths.
-
-Security findings distinguish unauthenticated compromise from a
-low-privileged authenticated bypass.

.agents/skills/adversarial-validation/references/simplicity.md

@@ -1,22 +0,0 @@
-# Simplicity Lens
-
-- Compare the production diff with the smallest equivalent local edit. Fewer
-  lines alone are not evidence; the replacement must preserve correctness,
-  compatibility, readability, and real boundaries.
-- Search the touched crate, domain owner, `crates/utils`, `crates/common`, and
-  relevant dependencies for each new helper, constant, wrapper, or fixture.
-- Reject forced reuse when normalization, error, deadline, or durability
-  semantics differ.
-- Require a concrete trigger for every new defensive branch. Keep boundary
-  checks for disk/RPC/version data and checks immediately before destructive
-  actions.
-- Flag one-caller helpers only when they merely forward or split a short linear
-  flow without adding domain naming, invariant isolation, or useful context.
-- Ensure a replacement removes the superseded in-scope path or keeps one
-  canonical core behind a documented compatibility adapter.
-- Remove narration/change-history comments; preserve concise safety, lock,
-  durability, and compatibility invariants.
-- Treat tests, fixtures, generated code, and documentation separately from
-  production growth. Do not optimize away meaningful regression coverage.
-
-A finding must include a concrete smaller design, not a style preference.

.agents/skills/adversarial-validation/references/test-coverage.md

@@ -1,26 +0,0 @@
-# Test-Coverage Lens
-
-- For every behavior claim, name the focused test/check that fails if the
-  changed hunk is reverted. If none is practical, require the reason and
-  residual risk.
-- Confirm tests exercise the real production path and distinguish the intended
-  behavior from the named regression. A success, `is_err()`, or no-panic check
-  can be sufficient when that is the actual contract; require exact values,
-  bytes, state, or error variants when those distinctions matter to the change.
-- For new flags/modes, verify each branch and ask which test fails if the branch
-  is inverted.
-- For new error propagation, inject the failure and assert the caller observes
-  it; mentally replacing `?`/`return Err` with success must break a test.
-- Streaming GET tests assert the complete body and length under degraded reads.
-- Disk/wire-format tests use pinned foreign/legacy fixtures; same-code
-  round-trips are insufficient for compatibility.
-- Concurrency tests use readiness polling, isolate global state, and avoid fixed
-  sleeps or unrealistically short timeouts. Use nextest groups when process-level
-  serialization is required.
-- Internal metadata tests assert both RustFS and MinIO keys, not only read-back
-  through a helper that prefers one key.
-- Avoid duplicate tests of the same production path and poison-value class.
-  Boundary companions are distinct coverage: `n == max` vs `max + 1`, and
-  absent vs empty vs nil UUID.
-- A focused test proves only the targets/features it builds. Add compilation or
-  Clippy only for uncovered changed targets.

.agents/skills/pr-review/SKILL.md

@@ -13,7 +13,7 @@ description: Review a GitHub PR using its exact base/head and repository risk po
    ```
 
    Record both SHAs; refresh if either moved during fetching. Follow relevant [implementation boundaries](../../references/implementation.md).
-2. Apply the [risk policy](../../references/adversarial-validation.md). Ordinary reviews use the root finding standard directly. For explicit adversarial, substantial, or high-risk reviews, use [the domain probes](../adversarial-validation/SKILL.md). File count does not authorize delegation; use exactly two independent reviewers only when authorized, otherwise two fresh sequential passes for that risk tier. Reviewers do not spawn further agents.
+2. Apply the [risk policy](../../references/adversarial-validation.md). Ordinary reviews use the root finding standard directly. File count does not authorize delegation; use exactly two independent reviewers only when authorized, otherwise two fresh sequential passes for that risk tier. Reviewers do not spawn further agents.
 3. Check `gh pr checks <N> --repo <owner/repo>`. Investigate failures when relevant to a finding or a requested CI/readiness diagnosis. Distinguish pending, unavailable, pre-existing, flaky, and PR-caused states using evidence; a code review does not authorize unrelated repairs.
 4. Report the reviewed SHAs, risk tier, supported findings or `No findings`, material verification limits, check state, and verdict. P0/P1 findings block approval. Severity: P0 = demonstrated data loss, security breach, remote crash, or deadlock; P1 = correctness, compatibility, or material hot-path regression; P2 = a concrete maintainability defect; P3 = optional style, only when requested.
 5. When posting is authorized, follow [Git and PR rules](../../references/pull-requests.md), recheck the remote head, and review any delta before submitting. Use `gh pr review` with `--approve`, `--request-changes`, or `--comment` and `--body-file`; use [the API example](references/posting.md) for authorized inline comments. Reuse existing authorization.

.agents/skills/security-advisory-lessons/SKILL.md

@@ -1,56 +0,0 @@
----
-name: security-advisory-lessons
-description: Perform a dedicated RustFS security/advisory review for authn/authz, IAM, RPC trust, paths, secrets, browser isolation, encryption, Object Lock, or other security boundaries. Use only when the user requests a security/advisory review or an adversarial review explicitly escalates to the full advisory map; do not auto-load solely because code touches a sensitive path.
----
-
-# RustFS Security Advisory Lessons
-
-Use this skill as the deep security lens. For a normal adversarial review with a
-matched security surface, the concise security reference under
-`adversarial-validation` is sufficient.
-
-## Workflow
-
-1. Freeze the exact diff/head and identify the changed trust boundaries.
-2. Inspect the headings in [advisory-patterns.md](references/advisory-patterns.md),
-   then read the matching sections. Read the full map only for a broad security
-   audit. Useful headings are
-   auth/admin, IAM/STS/OIDC, policy/plugins, S3/copy/multipart, protocols, paths,
-   secrets/logging/RPC, browser/CORS/proxy, SSE, Object Lock, and serde.
-3. Trace unauthenticated, low-privilege, wrong-action/owner/bucket, malformed,
-   and default-config cases. Security decisions must fail closed.
-4. Require a focused negative regression test for the bypass/exploit form, not
-   only the intended success path. State residual risk when a test is impractical.
-5. Report proven vulnerabilities separately from defense-in-depth hardening.
-
-When advisory currentness matters, fetch the live inventory instead of treating
-the reference as a status mirror:
-
-```bash
-gh api repos/rustfs/rustfs/security-advisories --paginate \
-  --jq '.[] | {ghsa_id,state,severity,summary,updated_at}'
-```
-
-Fetch an individual advisory only when the live summary indicates a new or
-changed lesson.
-
-## Finding Standard
-
-Every candidate names the lower-trust principal, the boundary it crosses, the
-affected resource, and the observable result. A best-practice gap or an impact
-confined to the caller's own principal is hardening, not a candidate.
-
-Report each candidate under exactly one verdict:
-
-- `confirmed`: a complete `file:line` source trace and a bounded observed
-  result. Only confirmed findings carry severity; each also states attacker
-  prerequisites, concrete input/path, impact, smallest safe fix, and a
-  regression check.
-- `needs_validation`: the trace is blocked on one exact unresolved fact, such
-  as deployment or infrastructure behavior. State that fact; assign no severity
-  and do not substitute an assumption.
-- `rejected`: a disproved candidate, with the caller, invariant, or test that
-  disproves it.
-
-Do not exaggerate unauthenticated impact when the actual issue requires
-authenticated low privilege.

.agents/skills/security-advisory-lessons/references/advisory-patterns.md

@@ -1,158 +0,0 @@
-# RustFS Advisory Pattern Map
-
-This file is a lesson map, not an advisory inventory mirror. It keeps durable security patterns distilled from RustFS GitHub Security Advisories.
-
-When current advisory state, severity, URLs, or full text matters, fetch it live:
-
-```bash
-gh api repos/rustfs/rustfs/security-advisories --paginate \
-  --jq '.[] | {ghsa_id,state,severity,summary,updated_at}'
-gh api repos/rustfs/rustfs/security-advisories/<GHSA_ID>
-```
-
-Update this file only when an advisory adds or changes a reusable lesson, affected surface, validation pattern, or regression-test expectation. Do not update it for state-only, URL-only, count-only, or timestamp-only changes.
-
-## Pattern Index
-
-### Admin authorization and route exposure
-
-- `GHSA-pfcq-4gjr-6gjm`: notification target endpoints accepted authenticated users but skipped admin authorization. Lesson: distinguish authn from authz; admin target CRUD must call the operation-specific admin authorization path.
-- `GHSA-mm2q-qcmx-gw4w`: `ListServiceAccount` used `UpdateServiceAccountAdminAction`, while update lacked target ownership checks. Lesson: exact action constants and ownership checks are both required; information disclosure can chain into secret rotation and takeover.
-- `GHSA-vcwh-pff9-64cc`: `ImportIam` checked `ExportIAMAction` for an import/write operation. Lesson: every admin handler must authorize the action it actually performs.
-- `GHSA-jqmc-mg33-v45g` and `GHSA-8784-9m7f-c6p6`: `/profile/cpu` and `/profile/memory` were whitelisted from auth and allowed expensive diagnostics plus path disclosure. Lesson: profiling/debug endpoints need admin auth, opt-in, rate limits, and non-sensitive responses.
-- `GHSA-f5cv-v44x-2xgf`: `/rustfs/admin/v3/metrics` accepted any authenticated IAM user and skipped admin authorization. Lesson: read-only metrics and diagnostic admin endpoints still require an operation-specific admin action check.
-- `GHSA-796f-j7xp-hwf4`: `/rustfs/admin/v3/list-remote-targets` checked only that credentials existed and leaked replication target credentials. Lesson: replication target reads are privileged admin operations, and stored remote credentials require strict authz plus response redaction review.
-- `GHSA-xp32-gxq2-3v52`: console license metadata endpoint was public and exposed subject and expiration fields. Lesson: management metadata endpoints should require admin auth or return only coarse public status.
-
-### IAM import, service accounts, and privilege boundaries
-
-- `GHSA-566f-q62r-wcr8`: `ImportIam` accepted attacker-controlled service account `parent`, `claims`, `accessKey`, and `secretKey`, enabling persistent backdoor accounts under root. Lesson: imported IAM payloads are untrusted data and must be validated against privilege boundaries.
-- `GHSA-3495-h8r9-gfqg`: `ExportIAM` wrote regular-user and service-account secret keys into exported ZIP data. Lesson: IAM export is a credential-disclosure boundary; redact, seal, or strictly justify every exported secret before treating export permission as safe.
-- `GHSA-5354-r3w2-34m8`: `AddServiceAccount` checked `CreateServiceAccountAdminAction` but trusted caller-supplied `target_user`, allowing service accounts under the root parent. Lesson: service-account create paths must validate parent ownership or root/admin authority, not only the create action.
-- `GHSA-xgr5-qc6w-vcg9`: `deny_only=true` skipped allow checks and let restricted service accounts mint unrestricted children. Lesson: deny-only logic must never become implicit allow for privilege creation.
-- `GHSA-mm2q-qcmx-gw4w`: leaked service account access keys plus update-without-ownership formed an escalation chain. Lesson: service-account identifiers are security-sensitive because update APIs consume them.
-
-### STS, OIDC, and federation flows
-
-- `GHSA-5qfg-mf7r-jp3w` and `GHSA-3473-5353-xhwh`: `AssumeRoleWithWebIdentity` was reachable through unauthenticated `POST /` routing and could issue temporary credentials from crafted web identity input. Lesson: every STS route needs explicit SigV4 or trusted identity-provider validation before role assumption, and unauthenticated exemptions must be narrowed to the exact action with uniform failure responses.
-- `GHSA-jxrr-r6pv-h958`: unsigned JWT issuer data was decoded before verification to select an OIDC provider, and distinguishable failures could expose provider configuration. Lesson: web-identity routing may be unauthenticated, but pre-verification claims are untrusted routing hints; bound and rate-limit the request, normalize public errors, and verify signature, issuer, audience, and expiration before issuing credentials.
-- `GHSA-ccrv-v8v9-ch9q`, `GHSA-48rf-7j3q-3hfv`, and `GHSA-xvfh-7c9g-hpw2`: service-account-controlled material could self-sign JWT session tokens with forged policy claims, and missing `exp` was accepted for service-account tokens. Lesson: session tokens must be signed by a trusted issuer/key path, enforce required claims and expiration, and reject self-signed or principal-controlled tokens.
-- `GHSA-9pjf-w3c2-m32r`, `GHSA-4x2q-cpx9-9h26`, and `GHSA-xvpm-p3f7-34c3`: public OIDC authorize/callback flows trusted request `Host` or forwarded scheme when building credential-bearing redirects. Lesson: OIDC redirects must use configured allowlisted origins and trusted-proxy handling; never derive the post-login credential destination from direct client headers.
-- `GHSA-m479-9x88-94w6`, `GHSA-frwq-mfqx-83p8`, `GHSA-q9q8-rf9r-fg9f`, and `GHSA-j5c2-hhf7-6gf5`: OIDC validation accepted attacker-controlled discovery URLs because hostname checks rejected only literal forbidden IPs, allowing DNS rebinding SSRF. Lesson: outbound federation URL validation must resolve and classify hostnames at the connection boundary and reject loopback, private, link-local, and rebound addresses.
-
-### IAM policy conditions and external policy plugins
-
-- `GHSA-6r96-hmgc-726c`: request headers collided with lowercase server-derived condition keys such as `userid`, `groups`, `versionid`, and JWT/LDAP claims. Lesson: never let caller-controlled headers append to or replace server-derived policy context; reserve trusted condition keys and keep intentional request-header keys separate.
-- `GHSA-v9cp-qfw9-9pfp`: quantified negated string conditions applied negation after aggregation, transposing `ForAllValues` and `ForAnyValue` semantics. Lesson: push negation into the per-value predicate for quantified operators and test partially overlapping multi-value sets.
-- `GHSA-5w8r-p896-6vq2`: OPA policy mode skipped `ExistingObjectTag/*` loading, so tagged objects looked untagged to external policies. Lesson: external authorization plugins need the same object-tag and request context as built-in policy evaluation before they decide.
-- `GHSA-9wgh-5r68-49p7`: substituted policy-variable values could recreate `${...}` and loop inside a single resolution pass, never reaching the outer iteration limit. Lesson: bound substitutions within each pass and test cycles formed across otherwise literal resolver values; confirm whether attacker-controlled claims can reach those values before assigning remote impact.
-
-### S3 object actions, copy, multipart, and upload policy validation
-
-- `GHSA-g8w9-qw9q-fghr` and `GHSA-xm99-m3gq-83g8`: a valid presigned `PutObject` accepted extra `x-amz-*` headers omitted from `SignedHeaders`; `x-amz-copy-source` could turn that upload capability into a cross-bucket read performed as the signer. Lesson: a presigned URL is a bounded capability; inspect all received security-sensitive headers and reject unsigned ones before selecting the S3 operation or reaching storage.
-- `GHSA-3ppv-fx5m-m749`: explicit `versionId` reads and copy sources authorized `s3:GetObject` instead of `s3:GetObjectVersion`. Lesson: version-specific object access must select version-specific actions for direct reads, `CopyObject`, and `UploadPartCopy`, with tests proving the backend is not reached on denial.
-- `GHSA-cg33-pxfp-qcj3` and `GHSA-65wq-gjw8-xr9w`: `HeadObject` and object-tagging paths authorized current-object actions while operating on an explicit historical `versionId`. Lesson: select the corresponding version action for metadata and tag reads and writes too; audit every operation that accepts a version ID.
-- `GHSA-gjph-xcgw-4p7w`: `PutObject` and browser POST accepted a caller-supplied `versionId` and replaced an unlocked historical version with only `s3:PutObject`. Lesson: only replication-authorized writes may choose an existing version ID; ordinary uploads must not rewrite version history.
-- `GHSA-x298-9x87-fvjq`: anonymous `ListObjectVersions` fell back to `ListBucket` and returned before public-access-block gates. Lesson: compatibility fallbacks must converge on the same post-authorization checks as direct grants, especially `RestrictPublicBuckets` and anonymous data-plane denies.
-- `GHSA-mx42-j6wv-px98`: `UploadPartCopy` missed source authorization and allowed cross-bucket object exfiltration. Lesson: multipart copy must enforce the same source and destination contract as `CopyObject`.
-- `GHSA-wfxj-ph3v-7mjf`: `UploadPartCopy` checked source and destination independently but missed destination copy-source policy constraints. Lesson: source read and destination write checks are not sufficient when policy constrains allowed copy sources.
-- `GHSA-w5fh-f8xh-5x3p`: presigned POST accepted uploads without enforcing signed policy conditions. Lesson: parse and enforce all POST policy constraints server-side, including size, key prefix, and content type.
-
-### Protocol frontends and IAM parity
-
-- `GHSA-3g29-xff2-92vp`: FTP `RETR` and `SIZE`/`MDTM` read paths authenticated the user but skipped IAM before calling storage. Lesson: non-HTTP protocol frontends must enforce the same per-operation authorization as the S3 API before backend access.
-- `GHSA-g3vq-vv42-f647`: FTPS `MKD` called `create_bucket` without checking `s3:CreateBucket`. Lesson: protocol command handlers need action-specific checks even when sibling handlers already authorize correctly.
-- `GHSA-3p3x-734c-h5vx`: FTPS and WebDAV compared secret keys with early-return string equality, while FTPS also returned distinguishable invalid-user and invalid-password failures. Lesson: password-style protocol auth needs constant-time secret comparison, indistinguishable failures where practical, and rate limiting.
-- `GHSA-rhg6-jrf2-29c4`: WebDAV, FTP/FTPS, and SFTP used identity-policy checks with empty conditions and bypassed S3 bucket-policy explicit denies and condition-key denies. Lesson: alternate listeners must evaluate the same effective policy and trustworthy request context before storage access.
-
-### Filesystem paths and object key traversal
-
-- `GHSA-pq29-69jg-9mxc`: RPC `read_file_stream` joined untrusted paths under a volume directory without canonical boundary checks. Lesson: `PathBuf::join` plus length checks are not path security.
-- `GHSA-8r6f-hmq2-28rg`: object keys containing traversal sequences bypassed bucket/object authorization when mapped to filesystem paths. Lesson: reject traversal at object-key parsing and verify final storage paths remain under the expected bucket/key root.
-- `GHSA-f4vq-9ffr-m8m3`: Snowball auto-extract accepted archive entries such as `../victim-bucket/object`, authorized the raw attacker-bucket path, then storage path cleaning crossed bucket boundaries. Lesson: archive entries become object keys and need traversal rejection plus consistent authz/storage normalization before writes.
-- `GHSA-g7fq-6xjm-mxf4`: versioned `DeleteObjects` accepted dot-segment keys, authorized the literal prefix, then wrote delete markers under a resolved sibling prefix. Lesson: validate each batch key before authorization and storage normalization, using the same object-name rules as single delete.
-
-### Secrets, defaults, and cryptographic misuse
-
-- `GHSA-j59h-h7q5-q348`, `GHSA-3wm5-wpm5-hmfm`, `GHSA-6wc8-xm48-qhmx`, `GHSA-9gf3-jx4p-4xxf`, `GHSA-63xc-c3w3-m2cf`, and `GHSA-ch63-6q4v-hwp5`: RustFS shipped known default root credentials that could authenticate to S3, admin APIs, IAM, KMS, console, and token-signing surfaces. Lesson: root credentials must be operator-provided or generated per install; known defaults and warnings are not acceptable for network-reachable deployments.
-- `GHSA-h956-rh7x-ppgj`: gRPC used the hard-coded token `rustfs rpc` on both client and server. Lesson: source-visible shared tokens are authentication bypasses.
-- `GHSA-r5qv-rc46-hv8q`: internode RPC HMAC secret fell back to the public default `rustfsadmin`. Lesson: RPC/internode auth must fail closed instead of silently using public defaults.
-- `GHSA-75fx-qg6f-8rm7` and `GHSA-68cw-96m3-h2cf`: internode RPC secrets were derivable from known root credentials, making raw storage RPC signatures forgeable when explicit RPC secrets were unset. Lesson: RPC auth keys must be independent random secrets, never derived from S3 root credentials, and raw storage RPC should not share the public S3 listener without an internode-only boundary.
-- `GHSA-m77q-r63m-pj89`: STS JWTs used the root secret key as the shared token signing key, allowing token forgery when the root secret was known. Lesson: STS signing keys need key separation, rotation, and key IDs; do not reuse root credentials for JWT/HMAC signing.
-- `GHSA-923g-jp7v-f97f`: license verification embedded a production RSA private key and used private-key decryption as authenticity. Lesson: ship verifying/public keys only and use real signature verification.
-
-### Sensitive logging and debug output
-
-- `GHSA-r54g-49rx-98cr`: STS credentials were logged at info level. Lesson: generated credentials must never be logged in plaintext.
-- `GHSA-8cm2-h255-v749`: debug logs leaked session tokens, secret keys, JWT claims, and raw STS response bodies. Lesson: redaction must cover custom `Debug` implementations and dependency response-body logging.
-- `GHSA-333v-68xh-8mmq`: invalid RPC signature logging included the shared HMAC secret and expected signature. Lesson: error paths often leak secrets; never log raw secrets or derived authenticators.
-
-### RPC input validation and panic safety
-
-- `GHSA-gw2x-q739-qhcr`: malformed gRPC `GetMetrics` payloads reached `unwrap()` on deserialization and caused remote DoS. Lesson: every network/RPC deserialization failure returns an error, not a panic.
-- `GHSA-h956-rh7x-ppgj` and `GHSA-r5qv-rc46-hv8q`: weak RPC auth increased reachability of otherwise internal handlers. Lesson: panic bugs become more severe when internode auth is weak or defaulted.
-- `GHSA-c667-rgrv-99vj`: NodeService authentication signed the service prefix instead of the concrete generated method path, so valid metadata for one RPC could be replayed to another method during the timestamp window. Lesson: RPC HMAC payloads must bind exact gRPC method path, HTTP method surrogate, timestamp, and secret.
-
-### Browser, CORS, and console isolation
-
-- `GHSA-v9fg-3cr2-277j`: object preview rendered attacker-controlled HTML in a same-origin iframe, exposing console credentials stored in `localStorage`. Lesson: user content must be origin-isolated from the console and protected with `nosniff`, CSP, and strict content-type handling.
-- `GHSA-7gcx-wg4x-q9x6`: an incomplete preview fix reintroduced extension-based PDF detection and bypassed the sandboxed fallback for attacker-controlled content. Lesson: browser-surface fixes need regression tests for alternate viewers and file-type branches, and preview trust must come from validated content type plus sandboxing rather than object names.
-- `GHSA-x5xv-223c-8vm7`: default CORS reflected arbitrary origins with credentials. Lesson: never combine reflected origins with `Access-Control-Allow-Credentials: true`; default should be fail-closed.
-
-### Trusted proxy and source IP conditions
-
-- `GHSA-fc6g-2gcp-2qrq`: `aws:SourceIp` trusted client-supplied `X-Forwarded-For` or `X-Real-IP`. Lesson: forwarded IP headers are valid only behind configured trusted proxies; direct clients use socket peer IP.
-
-### SSE and on-disk storage invariants
-
-- `GHSA-xrrf-67jm-3c2r`: SSE metadata reported encryption while reader composition bypassed `EncryptReader` and stored plaintext. Lesson: test actual bytes on disk and wrapper order, not only API metadata.
-- `GHSA-wqmc-vjgv-jrpw`: SSE-C persisted and listed the plaintext MD5 as the ETag, allowing list-only users to confirm or brute-force low-entropy content and correlate equal plaintext across keys. Lesson: encrypted-object ETags must not expose deterministic plaintext fingerprints; protect them with per-object key material and keep listing, conditional, copy, and multipart semantics consistent.
-
-### Object Lock and retention invariants
-
-- `GHSA-j548-9grx-fh4f`: Object Lock enforcement treated unreadable, fabricated, or unparsable bucket metadata as absent configuration and allowed retained objects to be deleted or expired. Lesson: retention must fail closed unless Object Lock absence is authoritative, and every delete, lifecycle, scanner, force-delete, and default-retention path needs the same state distinction.
-- `GHSA-28w4-rp2p-636w` and `GHSA-c7hx-vr97-45rg`: an authorized `DeleteBucket` with `x-rustfs-force-delete: true` removed active COMPLIANCE-retained versions even though ordinary version deletion was denied. Lesson: bucket force-delete must enforce retention across every version, including against root/owner callers.
-- `GHSA-q23w-fp7p-472v`: `PutObjectRetention` policy conditions read `x-amz-object-lock-*` headers while the handler applied mode and date from the XML body. Lesson: authorize the parsed retention that will be stored, or reject mismatched header/body values; compare retention dates as instants so offsets cannot change the applied bound.
-
-### Serde deserialization and input validation
-
-- Reject unknown fields in untrusted S3 API XML/JSON, lifecycle, policy, and replication input where compatibility permits. Check the current type and supported payload fixtures; do not infer the repository's current coverage from an older audit. Where extra fields are part of the compatibility contract, validate security-critical values explicitly.
-- `#[serde(default)]` on security-critical fields silently accepts missing values as zero/empty. Lesson: when a field has security implications (retention days, permissions, limits), validate the deserialized value explicitly rather than relying on defaults.
-- Integer fields deserialized from user input and cast with `as` (e.g., `i32 as u32`) can wrap negative values to large positives. Lesson: validate ranges before casting; use `try_into()` with a typed error, or clamp only when the domain explicitly requires saturation.
-- XML config typos (e.g., `"NoncurentDays"` instead of `"NoncurrentDays"`) are silently accepted when `deny_unknown_fields` is absent. Lesson: strict deserialization prevents silent misconfiguration that could cause data loss or unexpected retention behavior.
-
-## Useful Search Seeds
-
-Use these targeted searches when a diff touches security-sensitive code:
-
-```bash
-rg -n "validate_admin_request|check_permissions|AdminAction::|deny_only|is_allowed" rustfs crates
-rg -n "authorize_operation|FtpsDriver|SftpDriver|RETR|MKD|SIZE|MDTM|CreateBucket|GetObject|HeadObject" crates/protocols rustfs
-rg -n "UploadPartCopy|upload_part_copy|CompleteMultipart|PostObject|presign|SignedHeaders|x-amz-copy-source|collect_signed_headers|content-length-range|starts-with" rustfs crates
-rg -n "ListBucketVersions|GetObjectVersion|versionId|VersionId|ExistingObjectTag|ForAllValues|ForAnyValue|POLICY_PLUGIN|opa" rustfs crates
-rg -n "normalize_extract_entry_key|Snowball|auto-extract|PathBuf::join|canonicalize|\\.\\.|x-forwarded-for|x-real-ip|SourceIp" rustfs crates
-rg -n "DEFAULT_SECRET|DEFAULT_ACCESS|TEST_PRIVATE_KEY|rustfs rpc|RUSTFS_RPC_SECRET" rustfs crates
-rg -n "TONIC_RPC_PREFIX|verify_rpc_signature|check_auth|NodeServiceServer|x-rustfs-signature" rustfs crates
-rg -n "debug!|trace!|info!|error!|\\?resp|\\?merged_config|session_token|secret_key" rustfs crates
-rg -n "HashReader|EncryptReader|try_resolve_etag|ETag|SSE|server-side encryption|Access-Control-Allow-Credentials|Origin" rustfs crates
-rg -n "ObjectLock|object_lock|retention|COMPLIANCE|GOVERNANCE|delete_prefix|lifecycle|scanner" rustfs crates
-rg -n "deny_unknown_fields|serde.default|as u32|as usize|as i32" rustfs crates
-```
-
-## Minimum Regression Test Expectations
-
-- Authz fixes: include unauthenticated, valid low-privilege, wrong-action, correct-action, owner, non-owner, and root/admin cases as applicable.
-- Protocol frontend authz fixes: include denied `RETR`, `SIZE`/`MDTM`, `MKD`, bucket probe, and sibling allowed-operation cases; test bucket-policy explicit denies and condition-key denies on each enabled listener against an S3 control, and assert denied paths do not reach storage.
-- IAM fixes: include import/update/list service-account cases with attacker-controlled parent, claims, access key, secret key, and policy.
-- Copy/upload fixes: include cross-bucket, cross-user, source-denied, destination-denied, copy-source-condition, and multipart completion cases.
-- Presigned upload fixes: include a valid presign with extra unsigned tagging, redirect, storage-class, and cross-bucket copy-source headers; require rejection before source or destination storage access, and verify explicitly signed equivalents still work.
-- Version-action fixes: include historical UUID, explicit current version, `null`, range, partNumber, presigned, STS/session, service-account, anonymous bucket-policy, copy source, multipart-copy source, `HeadObject`, and tag read/write cases.
-- Version-write fixes: a `PutObject`-only principal must not replace an existing version through `PutObject` or browser POST `?versionId=`; verify replication-authorized writes and versioned, suspended, and unversioned buckets separately.
-- Policy-condition fixes: include reserved-key header collisions, missing keys, partially overlapping multi-value sets, cyclic variable substitutions, plugin mode, and built-in policy mode.
-- Path fixes: include encoded traversal, absolute path, nested traversal, archive entries with `..`, valid object keys that resemble traversal text but should be rejected, and canonical bucket/prefix boundary checks; test per-key `DeleteObjects` errors and prefix-scoped denials on versioned buckets.
-- Logging fixes: assert redacted output for structs and response bodies that may contain credentials.
-- IAM export fixes: assert exported archives omit plaintext user and service-account secrets unless the format deliberately encrypts or seals them.
-- RPC auth fixes: include captured metadata replay across two concrete methods, stale timestamps, wrong path, wrong method surrogate, wrong secret, and valid same-method calls.
-- Browser/CORS fixes: assert no credentials on reflected/default origins, correct behavior for explicit allowlists, and no same-origin script execution for previewed object content.
-- SSE fixes: inspect stored bytes and verify API metadata, read-back behavior, and on-disk ciphertext together; for ETags, upload equal plaintext under different keys and assert list-only callers cannot derive or correlate the plaintext digest, including multipart objects.
-- Object Lock fixes: include unreadable metadata, fabricated metadata defaults, unparsable config, confirmed absent config, COMPLIANCE/GOVERNANCE retention, lifecycle expiry, scanner sweeps, and force-delete paths; an authorized force-delete must preserve active COMPLIANCE versions. For `PutObjectRetention`, test conflicting header/body values and offset dates against policy bounds and read-back retention.

.github/workflows/windows-filesystem.yml

@@ -1,65 +1,224 @@
-name: Scanner Linux Hosted Preflight
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name: Windows Filesystem Tests
+
 on:
+  push:
+    branches: [ main ]
+    paths:
+      - "crates/ecstore/src/disk/**"
+      - "crates/ecstore/src/layout/**"
+      - "crates/ecstore/src/store/init_format.rs"
+      - "crates/ecstore/src/set_disk/ops/locking.rs"
+      - "crates/ecstore/src/object_api/write_commit_context.rs"
+      - "crates/ecstore/src/bucket/utils.rs"
+      - "crates/ecstore/Cargo.toml"
+      - "crates/utils/src/dunce.rs"
+      - "crates/utils/src/os/**"
+      - "crates/utils/Cargo.toml"
+      - "crates/utils/src/path.rs"
+      - "rustfs/build.rs"
+      - "rustfs/Cargo.toml"
+      - "rustfs/src/main.rs"
+      - "rustfs/src/server/runtime.rs"
+      - "rustfs/src/startup_*.rs"
+      - "Cargo.toml"
+      - "Cargo.lock"
+      - ".github/actions/setup/**"
+      - ".github/workflows/windows-filesystem.yml"
+  pull_request:
+    branches: [ main ]
+    paths:
+      - "crates/ecstore/src/disk/**"
+      - "crates/ecstore/src/layout/**"
+      - "crates/ecstore/src/store/init_format.rs"
+      - "crates/ecstore/src/set_disk/ops/locking.rs"
+      - "crates/ecstore/src/object_api/write_commit_context.rs"
+      - "crates/ecstore/src/bucket/utils.rs"
+      - "crates/ecstore/Cargo.toml"
+      - "crates/utils/src/dunce.rs"
+      - "crates/utils/src/os/**"
+      - "crates/utils/Cargo.toml"
+      - "crates/utils/src/path.rs"
+      - "rustfs/build.rs"
+      - "rustfs/Cargo.toml"
+      - "rustfs/src/main.rs"
+      - "rustfs/src/server/runtime.rs"
+      - "rustfs/src/startup_*.rs"
+      - "Cargo.toml"
+      - "Cargo.lock"
+      - ".github/actions/setup/**"
+      - ".github/workflows/windows-filesystem.yml"
   workflow_dispatch:
+
 permissions:
   contents: read
+
 concurrency:
-  group: scanner-linux-hosted-${{ github.ref }}
-  cancel-in-progress: false
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 env:
-  CARGO_TERM_COLOR: never
+  CARGO_TERM_COLOR: always
+  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
   RUST_BACKTRACE: 1
-  FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: 'true'
-  NO_PROXY: 127.0.0.1,localhost
+
 jobs:
-  scoped-tests:
-    runs-on: ubuntu-latest
-    timeout-minutes: 90
+  rename-safety:
+    name: Rename Safety
+    runs-on: windows-latest
+    timeout-minutes: 60
     steps:
-      - name: Checkout candidate
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
         with:
           persist-credentials: false
-          fetch-depth: 2
-      - name: Setup Rust
+
+      - name: Setup Rust environment
         uses: ./.github/actions/setup
         with:
           rust-version: stable
-          cache-shared-key: ci-e2e-distributed-hosted
+          cache-shared-key: build-x86_64-pc-windows-msvc
           cache-save-if: 'false'
           install-build-packaging-tools: 'false'
           install-test-tools: 'false'
-      - name: Check candidate and resources
-        shell: bash
+
+      - name: Check production Windows dependencies
+        shell: pwsh
+        run: cargo check -p rustfs-ecstore --lib
+
+      - name: Test guarded rename publication
+        shell: pwsh
+        run: cargo test -p rustfs-ecstore --lib rename_all_ -- --nocapture
+
+      - name: Test multipart parent durability barriers
+        shell: pwsh
+        run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture
+
+      - name: Test Windows path conversion
+        shell: pwsh
+        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
+
+      - name: Test Windows volume and disk-space probes
+        shell: pwsh
         run: |
-          set -euo pipefail
-          test "$(git rev-parse HEAD^)" = 03eef1ea79a782e15480a00ba87ed2e256a84f9b
-          test "$(git diff --name-only HEAD^ HEAD)" = .github/workflows/windows-filesystem.yml
-          df -h .
-          free -h
-          du -sh target "$HOME/.cargo" 2>/dev/null || true
-      - name: Run only the named diagnostic unit tests
-        shell: bash
-        env:
-          CARGO_BUILD_JOBS: '1'
-          CARGO_INCREMENTAL: '0'
-        run: |
-          set -euo pipefail
-          run_one() {
-            local package="$1"
-            local test_name="$2"
-            cargo test --locked -p "$package" --lib -- --exact "$test_name" --nocapture 2>&1 | tee scanner-proof-test.log
-            grep -F 'test result: ok. 1 passed; 0 failed;' scanner-proof-test.log
+          # Map a drive letter to the temp drive's administrative share so the
+          # remote-volume probe also covers mapped network drives.
+          $drive = (Split-Path -Qualifier ([System.IO.Path]::GetTempPath())).TrimEnd(':')
+          net use R: "\\localhost\$drive`$" /persistent:no
+          if ($LASTEXITCODE -eq 0) {
+            $env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE = 'R:'
+          } else {
+            Write-Output "::warning::could not map R: to \\localhost\$drive`$; mapped-drive test will skip"
+          }
+          cargo test -p rustfs-utils --features os --lib os:: -- --nocapture
+          $status = $LASTEXITCODE
+          if ($env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE) { net use R: /delete /y | Out-Null }
+          exit $status
+
+      - name: Test Windows handle guards
+        shell: pwsh
+        run: cargo test -p rustfs-ecstore --lib windows_ -- --nocapture
+
+      - name: Test endpoint path parsing
+        shell: pwsh
+        run: "cargo test -p rustfs-ecstore --lib layout::endpoint:: -- --nocapture"
+
+      - name: Test startup temporary-directory cleanup
+        shell: pwsh
+        run: cargo test -p rustfs-ecstore --lib cleanup_tmp_on_startup_ -- --nocapture
+
+      - name: Test object key Windows admission
+        shell: pwsh
+        run: "cargo test -p rustfs-ecstore --lib bucket::utils:: -- --nocapture"
+
+      - name: Test Windows file-name rules
+        shell: pwsh
+        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
+
+      - name: Test fresh format publication
+        shell: pwsh
+        run: cargo test -p rustfs-ecstore --lib fresh_format_load_initializes_all_disks -- --nocapture
+
+  # The native Windows main thread only gets the 1 MiB PE stack reserve, while
+  # Linux and macOS hand out 8 MiB. Startup runs the whole root future on the
+  # thread that calls `block_on`, so a debug server once overflowed before it
+  # listened on Windows only. Start a real server to keep that covered.
+  startup-smoke:
+    name: Startup Smoke
+    runs-on: windows-latest
+    timeout-minutes: 90
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Setup Rust environment
+        uses: ./.github/actions/setup
+        with:
+          rust-version: stable
+          cache-shared-key: build-x86_64-pc-windows-msvc
+          cache-save-if: 'false'
+          install-build-packaging-tools: 'false'
+          install-test-tools: 'false'
+
+      - name: Build debug server binary
+        shell: pwsh
+        run: cargo build --locked -p rustfs --bin rustfs
+
+      - name: Print version
+        shell: pwsh
+        run: |
+          & .\target\debug\rustfs.exe --version
+          if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
+
+      - name: Start server and wait for readiness
+        shell: pwsh
+        env:
+          RUSTFS_ACCESS_KEY: rustfsadmin
+          RUSTFS_SECRET_KEY: rustfsadmin
+        run: |
+          $volume = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke'
+          $stdout = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.out.log'
+          $stderr = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.err.log'
+          New-Item -ItemType Directory -Force -Path $volume | Out-Null
+          $server = Start-Process -FilePath .\target\debug\rustfs.exe `
+            -ArgumentList @('--address', '127.0.0.1:9000', "`"$volume`"") `
+            -RedirectStandardOutput $stdout -RedirectStandardError $stderr `
+            -NoNewWindow -PassThru
+          # Cache the process handle now; otherwise ExitCode reads back empty.
+          $null = $server.Handle
+          $ready = $false
+          try {
+            for ($i = 0; $i -lt 180; $i++) {
+              if ($server.HasExited) { break }
+              try {
+                $response = Invoke-WebRequest -Uri 'http://127.0.0.1:9000/health/ready' -UseBasicParsing -NoProxy -TimeoutSec 2
+                if ($response.StatusCode -eq 200) { $ready = $true; break }
+              } catch {}
+              Start-Sleep -Seconds 1
+            }
+          } finally {
+            if (-not $server.HasExited) { Stop-Process -Id $server.Id -Force }
+            $server.WaitForExit()
+            Write-Host '--- server stdout (tail) ---'
+            Get-Content $stdout -Tail 100 -ErrorAction SilentlyContinue
+            Write-Host '--- server stderr (tail) ---'
+            Get-Content $stderr -Tail 100 -ErrorAction SilentlyContinue
+          }
+          if (-not $ready) {
+            throw ("rustfs did not report ready on /health/ready (exit code: {0})" -f $server.ExitCode)
           }
-          run_one rustfs-scanner scanner::tests::cycle_budget_publication::publication_lease_observations_keep_cycle_epoch_purpose_and_outcome
-          run_one rustfs-scanner scanner::tests::cycle_budget_publication::publication_lease_retry_logs_identify_interleaved_epochs
-          run_one rustfs-scanner scanner::tests::cycle_budget_publication::publication_lease_events_include_context_in_otlp_without_span_export
-          run_one rustfs-scanner scanner::tests::cycle_budget_publication::dropped_lease_wait_does_not_report_a_finished_outcome
-          run_one rustfs-scanner scanner::tests::cycle_budget_publication::publication_lease_retries_local_channel_deadline_and_stops_on_cancel
-          run_one rustfs-ecstore cluster::rpc::peer_rest_client::tests::scanner_publication_lease_validation_errors_are_typed_without_changing_display
-          run_one rustfs-ecstore cluster::rpc::peer_rest_client::tests::scanner_publication_lease_error_class_keeps_remote_reports_separate_from_local_errors
-          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_acquisition_error_identifies_peer_and_preserves_retry_suffix
-          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_failure_info_uses_the_actual_target_and_rejects_unbounded_peers
-          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_failure_info_accepts_real_grid_host_urls
-          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_failure_info_preserves_typed_connection_failures_without_logging_details
-          run_one rustfs-ecstore services::notification_sys::tests::scanner_lease_channel_deadline_preserves_public_error_and_diagnostic

AGENTS.md

@@ -158,9 +158,7 @@ requested adversarial/design reviews, and agent-instruction changes that alter
 execution. Ordinary questions, diagnoses, status reports, non-adversarial code
 reviews, and low-risk planning do not trigger it.
 
-For applicable work and substantial PR reviews, read the [risk tiers and review shape](.agents/references/adversarial-validation.md).
-Load only the matching domain probes; ordinary reviews do not become adversarial
-merely because this reference exists.
+For applicable work and substantial PR reviews, follow the [risk tiers and review shape](.agents/references/adversarial-validation.md) and check changed trust boundaries and relevant correctness, security, concurrency, durability, compatibility, performance, and negative regression coverage.
 
 A review has no finding quota; `No findings` is a complete outcome. A request to
 find problems is not evidence that a defect exists. Before reporting a candidate,

Cargo.lock

@@ -10642,9 +10642,6 @@ dependencies = [
  "hotpath",
  "http 1.5.0",
  "metrics",
- "opentelemetry",
- "opentelemetry-appender-tracing",
- "opentelemetry_sdk",
  "rand 0.10.3",
  "rmp-serde",
  "rustfs-common",

crates/common/src/globals.rs

@@ -261,20 +261,6 @@ pub async fn cached_connection(addr: &str) -> Option<Channel> {
     GLOBAL_CONN_MAP.read().await.get(addr).cloned()
 }
 
-/// Holds the connection cache write lock for an isolated test process.
-#[cfg(any(test, feature = "test-util"))]
-pub struct ConnectionCacheReadBlockerForTest {
-    _guard: tokio::sync::RwLockWriteGuard<'static, HashMap<String, Channel>>,
-}
-
-/// Block cache reads until the returned test guard is dropped.
-#[cfg(any(test, feature = "test-util"))]
-pub async fn block_connection_cache_reads_for_test() -> ConnectionCacheReadBlockerForTest {
-    ConnectionCacheReadBlockerForTest {
-        _guard: GLOBAL_CONN_MAP.write().await,
-    }
-}
-
 /// Cache a gRPC connection for the given address.
 pub async fn cache_connection(addr: String, channel: Channel) {
     GLOBAL_CONN_MAP.write().await.insert(addr, channel);

crates/ecstore/Cargo.toml

@@ -245,7 +245,6 @@ windows-sys = { workspace = true, features = ["Win32_System_Ioctl"] }
 
 [dev-dependencies]
 aws-smithy-async.workspace = true
-rustfs-common = { workspace = true, features = ["test-util"] }
 rustfs-s3-ops.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "test-util", "fs"] }
 criterion = { workspace = true, features = ["html_reports"] }

crates/ecstore/src/api/mod.rs

@@ -527,15 +527,15 @@ pub mod notification {
     pub use crate::services::notification_sys::{
         ClusterTierDailyStats, CrossPoolFenceFleetProofToken, IlmRecoveryExportFleetProofToken,
         LegacyTransitionStateReconcileFleetProofToken, MrfLifecycleWriterFleetProofToken, NotificationPeerErr, NotificationSys,
-        ScannerPublicationLeaseFailureInfo, ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof,
-        acquire_ilm_recovery_export_fleet_proof, acquire_legacy_transition_state_reconcile_fleet_proof,
-        acquire_mrf_lifecycle_writer_fleet_proof, configured_fleet_topology_fingerprint, cross_pool_fence_fleet_proof_matches,
-        get_global_notification_sys, ilm_recovery_export_fleet_proof_matches, ilm_recovery_export_local_process_epoch,
+        ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof, acquire_ilm_recovery_export_fleet_proof,
+        acquire_legacy_transition_state_reconcile_fleet_proof, acquire_mrf_lifecycle_writer_fleet_proof,
+        configured_fleet_topology_fingerprint, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
+        ilm_recovery_export_fleet_proof_matches, ilm_recovery_export_local_process_epoch,
         ilm_recovery_export_member_epochs_sha256, ilm_recovery_export_topology_generation,
         legacy_transition_state_reconcile_fleet_proof_matches, mrf_lifecycle_reader_local_process_epoch,
         mrf_lifecycle_writer_fleet_proof_current, mrf_lifecycle_writer_fleet_proof_matches, new_global_notification_sys,
-        scanner_peer_transport_error_message_is_retryable, scanner_publication_lease_failure_info,
-        start_remote_version_state_fleet_probe, with_mrf_lifecycle_writer_fleet_proof,
+        scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
+        with_mrf_lifecycle_writer_fleet_proof,
     };
     #[cfg(any(test, feature = "test-util"))]
     pub use crate::services::notification_sys::{

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -668,136 +668,6 @@ impl ScannerPublicationLease {
     }
 }
 
-struct ScannerPublicationLeaseLocalError {
-    reason: &'static str,
-    detail: String,
-}
-
-impl std::fmt::Debug for ScannerPublicationLeaseLocalError {
-    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        formatter
-            .debug_struct("ScannerPublicationLeaseLocalError")
-            .field("reason", &self.reason)
-            .finish()
-    }
-}
-
-impl std::fmt::Display for ScannerPublicationLeaseLocalError {
-    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        formatter.write_str(&self.detail)
-    }
-}
-
-impl std::error::Error for ScannerPublicationLeaseLocalError {}
-
-struct ScannerPublicationLeaseRemoteRejection(String);
-
-impl std::fmt::Debug for ScannerPublicationLeaseRemoteRejection {
-    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        formatter.write_str("ScannerPublicationLeaseRemoteRejection")
-    }
-}
-
-impl std::fmt::Display for ScannerPublicationLeaseRemoteRejection {
-    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        formatter.write_str(&self.0)
-    }
-}
-
-impl std::error::Error for ScannerPublicationLeaseRemoteRejection {}
-
-pub(crate) fn scanner_publication_lease_remote_rejection(detail: String) -> Error {
-    Error::other(ScannerPublicationLeaseRemoteRejection(detail))
-}
-
-fn scanner_publication_lease_local_error(reason: &'static str, detail: &'static str) -> Error {
-    Error::other(ScannerPublicationLeaseLocalError {
-        reason,
-        detail: detail.to_owned(),
-    })
-}
-
-fn scanner_publication_lease_local_io_error(reason: &'static str, error: std::io::Error) -> Error {
-    Error::Io(std::io::Error::new(
-        error.kind(),
-        ScannerPublicationLeaseLocalError {
-            reason,
-            detail: error.to_string(),
-        },
-    ))
-}
-
-#[derive(Debug)]
-pub(crate) struct ScannerPublicationLeaseErrorClass {
-    pub reason: Option<&'static str>,
-    pub reported_reason: Option<&'static str>,
-}
-
-pub(crate) fn scanner_publication_lease_error_class(error: &Error) -> ScannerPublicationLeaseErrorClass {
-    let mut result = ScannerPublicationLeaseErrorClass {
-        reason: None,
-        reported_reason: None,
-    };
-    match error {
-        Error::RemoteClientUnavailable(_) => result.reason = Some("client_unavailable"),
-        Error::RemoteNotInitialized => result.reason = Some("peer_not_initialized"),
-        Error::Timeout => result.reason = Some("timeout"),
-        Error::Io(error) => {
-            if let Some(local) = error
-                .get_ref()
-                .and_then(|source| source.downcast_ref::<ScannerPublicationLeaseLocalError>())
-            {
-                result.reason = Some(local.reason);
-            } else if let Some(remote) = error
-                .get_ref()
-                .and_then(|source| source.downcast_ref::<ScannerPublicationLeaseRemoteRejection>())
-            {
-                // These are the exact Display strings currently produced by
-                // the peer's lease acquisition path, not arbitrary suffixes.
-                result.reported_reason = match remote.0.as_str() {
-                    "Io error: scanner publication lease capacity is exhausted" => Some("capacity_exhausted"),
-                    "Io error: scanner publication lease generation is stale" => Some("generation_stale"),
-                    "Io error: scanner publication lease is blocked by data movement" => Some("blocked_by_data_movement"),
-                    _ => None,
-                };
-            } else if let Some(status) = embedded_tonic_status(error) {
-                result.reason = Some(match status.code() {
-                    tonic::Code::Ok => "rpc_ok",
-                    tonic::Code::Cancelled => "rpc_cancelled",
-                    tonic::Code::Unknown => "rpc_unknown",
-                    tonic::Code::InvalidArgument => "rpc_invalid_argument",
-                    tonic::Code::DeadlineExceeded => "rpc_deadline_exceeded",
-                    tonic::Code::NotFound => "rpc_not_found",
-                    tonic::Code::AlreadyExists => "rpc_already_exists",
-                    tonic::Code::PermissionDenied => "rpc_permission_denied",
-                    tonic::Code::ResourceExhausted => "rpc_resource_exhausted",
-                    tonic::Code::FailedPrecondition => "rpc_failed_precondition",
-                    tonic::Code::Aborted => "rpc_aborted",
-                    tonic::Code::OutOfRange => "rpc_out_of_range",
-                    tonic::Code::Unimplemented => "rpc_unimplemented",
-                    tonic::Code::Internal => "rpc_internal",
-                    tonic::Code::Unavailable => "rpc_unavailable",
-                    tonic::Code::DataLoss => "rpc_data_loss",
-                    tonic::Code::Unauthenticated => "rpc_unauthenticated",
-                });
-            } else {
-                result.reason = match error.kind() {
-                    std::io::ErrorKind::TimedOut => Some("io_timeout"),
-                    std::io::ErrorKind::ConnectionRefused => Some("io_connection_refused"),
-                    std::io::ErrorKind::ConnectionReset => Some("io_connection_reset"),
-                    std::io::ErrorKind::ConnectionAborted => Some("io_connection_aborted"),
-                    std::io::ErrorKind::BrokenPipe => Some("io_broken_pipe"),
-                    std::io::ErrorKind::UnexpectedEof => Some("io_unexpected_eof"),
-                    std::io::ErrorKind::NotConnected => Some("io_not_connected"),
-                    _ => None,
-                };
-            }
-        }
-        _ => {}
-    }
-    result
-}
-
 fn validate_scanner_publication_lease_response_fields(
     response: &ScannerPublicationLeaseResponse,
     expected_session_id: &str,
@@ -805,7 +675,7 @@ fn validate_scanner_publication_lease_response_fields(
     expected_observational_only: bool,
 ) -> Result<(Uuid, String)> {
     if !response.success {
-        return Err(scanner_publication_lease_remote_rejection(
+        return Err(Error::other(
             response
                 .error
                 .as_ref()
@@ -814,39 +684,24 @@ fn validate_scanner_publication_lease_response_fields(
         ));
     }
     if response.movement_generation != expected_generation {
-        return Err(scanner_publication_lease_local_error(
-            "generation_mismatch",
-            "peer returned a different scanner publication lease generation",
-        ));
+        return Err(Error::other("peer returned a different scanner publication lease generation"));
     }
     if response.session_id != expected_session_id {
-        return Err(scanner_publication_lease_local_error(
-            "session_mismatch",
-            "peer returned a different scanner publication lease session",
-        ));
+        return Err(Error::other("peer returned a different scanner publication lease session"));
     }
     if response.observational_only != expected_observational_only {
-        return Err(scanner_publication_lease_local_error(
-            "purpose_mismatch",
-            "peer returned a different scanner publication lease purpose",
-        ));
+        return Err(Error::other("peer returned a different scanner publication lease purpose"));
     }
     let owner_id = Uuid::parse_str(&response.owner_id)
         .ok()
         .filter(|owner_id| !owner_id.is_nil())
         .map(|owner_id| owner_id.to_string())
-        .ok_or_else(|| {
-            scanner_publication_lease_local_error("invalid_owner", "peer returned an invalid scanner publication lease owner")
-        })?;
+        .ok_or_else(|| Error::other("peer returned an invalid scanner publication lease owner"))?;
     if response.lease_ttl_ms != crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS {
-        return Err(scanner_publication_lease_local_error(
-            "unsupported_ttl",
-            "peer returned an unsupported scanner publication lease TTL",
-        ));
+        return Err(Error::other("peer returned an unsupported scanner publication lease TTL"));
     }
-    let token = Uuid::from_slice(response.token.as_ref()).map_err(|_| {
-        scanner_publication_lease_local_error("invalid_token", "peer returned an invalid scanner publication lease token")
-    })?;
+    let token = Uuid::from_slice(response.token.as_ref())
+        .map_err(|_| Error::other("peer returned an invalid scanner publication lease token"))?;
     Ok((token, owner_id))
 }
 
@@ -857,27 +712,16 @@ fn scanner_publication_lease_deadline(
 ) -> Result<std::time::Instant> {
     let lease_window = Duration::from_millis(lease_ttl_ms)
         .checked_sub(SCANNER_PUBLICATION_LEASE_SAFETY_MARGIN)
-        .ok_or_else(|| {
-            scanner_publication_lease_local_error(
-                "ttl_below_safety_margin",
-                "scanner publication lease TTL is shorter than its safety margin",
-            )
-        })?;
-    let elapsed = response_received.checked_duration_since(request_started).ok_or_else(|| {
-        scanner_publication_lease_local_error(
-            "response_clock_backwards",
-            "scanner publication lease response clock moved backwards",
-        )
-    })?;
+        .ok_or_else(|| Error::other("scanner publication lease TTL is shorter than its safety margin"))?;
+    let elapsed = response_received
+        .checked_duration_since(request_started)
+        .ok_or_else(|| Error::other("scanner publication lease response clock moved backwards"))?;
     if elapsed >= lease_window {
-        return Err(scanner_publication_lease_local_error(
-            "response_after_safety_window",
-            "scanner publication lease response arrived after its safety window",
-        ));
+        return Err(Error::other("scanner publication lease response arrived after its safety window"));
     }
-    request_started.checked_add(lease_window).ok_or_else(|| {
-        scanner_publication_lease_local_error("deadline_overflow", "scanner publication lease deadline overflowed")
-    })
+    request_started
+        .checked_add(lease_window)
+        .ok_or_else(|| Error::other("scanner publication lease deadline overflowed"))
 }
 
 /// A peer's readiness view of one drive served by another host.
@@ -1032,11 +876,6 @@ fn tier_mutation_failed_response_error(version: u32, failure_class: i32, error_i
     Error::other(message)
 }
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
-pub(crate) enum ScannerLeaseConnectionFailure {
-    ChannelAcquisitionTimedOut,
-}
-
 impl PeerRestClient {
     fn recovery_monitor_span(grid_host: &str) -> tracing::Span {
         tracing::info_span!(
@@ -1164,37 +1003,19 @@ impl PeerRestClient {
     }
 
     pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> {
-        self.get_client_with_connection_failure().await.map_err(|(error, _)| error)
-    }
-
-    async fn get_client_with_connection_failure(
-        &self,
-    ) -> std::result::Result<
-        NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>,
-        (Error, Option<ScannerLeaseConnectionFailure>),
-    > {
         if self.offline.load(Ordering::Acquire) {
             self.mark_offline_and_spawn_recovery();
-            return Err((
-                Error::RemoteClientUnavailable(format!("peer {} is temporarily offline", self.grid_host)),
-                None,
-            ));
+            return Err(Error::RemoteClientUnavailable(format!("peer {} is temporarily offline", self.grid_host)));
         }
 
         node_service_time_out_client(&self.grid_host, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
             .await
             .map_err(|err| {
-                // Channel acquisition reports its deadline as a top-level
-                // typed I/O timeout. Do not infer kinds from dial error text.
-                let failure = err
-                    .downcast_ref::<std::io::Error>()
-                    .filter(|error| error.kind() == std::io::ErrorKind::TimedOut)
-                    .map(|_| ScannerLeaseConnectionFailure::ChannelAcquisitionTimedOut);
                 let storage_err = Error::RemoteClientUnavailable(format!("can not get client, err: {err}"));
                 if Self::is_network_like_error(&storage_err) {
                     self.mark_offline_and_spawn_recovery();
                 }
-                (storage_err, failure)
+                storage_err
             })
     }
 
@@ -2687,91 +2508,54 @@ impl PeerRestClient {
         expected_generation: u64,
         observational_only: bool,
     ) -> Result<ScannerPublicationLease> {
-        self.acquire_scanner_publication_lease_with_connection_failure(
-            expected_session_id,
-            expected_generation,
-            observational_only,
+        let request_started = std::time::Instant::now();
+        self.finalize_result(
+            async {
+                let challenge = Uuid::new_v4();
+                let mut client = self
+                    .get_client()
+                    .await?
+                    .max_decoding_message_size(SCANNER_ACTIVITY_MAX_MESSAGE_SIZE)
+                    .max_encoding_message_size(SCANNER_ACTIVITY_MAX_MESSAGE_SIZE);
+                let mut request = Request::new(ScannerPublicationLeaseRequest {
+                    challenge: challenge.as_bytes().to_vec().into(),
+                    expected_movement_generation: expected_generation,
+                    ttl_ms: crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS,
+                    expected_session_id: expected_session_id.to_string(),
+                    token: Bytes::new(),
+                    observational_only,
+                });
+                let canonical = rustfs_protos::canonical_scanner_publication_lease_request_body(request.get_ref())
+                    .map_err(|_| Error::other("scanner publication lease request is too large to authenticate"))?;
+                set_tonic_canonical_body_digest(&mut request, &canonical)?;
+                let response = client.acquire_scanner_publication_lease(request).await?.into_inner();
+                let response_body =
+                    rustfs_protos::canonical_scanner_publication_lease_response_body(challenge.as_bytes(), &response)
+                        .map_err(|_| Error::other("scanner publication lease response is too large to authenticate"))?;
+                verify_tonic_rpc_response_proof(&response_body, &response.response_proof)
+                    .map_err(|_| Error::other("peer returned an invalid scanner publication lease proof"))?;
+                let (token, owner_id) = validate_scanner_publication_lease_response_fields(
+                    &response,
+                    expected_session_id,
+                    expected_generation,
+                    observational_only,
+                )?;
+                Ok(ScannerPublicationLease {
+                    token,
+                    movement_generation: response.movement_generation,
+                    observational_only,
+                    owner_id,
+                    session_id: response.session_id,
+                    expires_at: scanner_publication_lease_deadline(
+                        request_started,
+                        std::time::Instant::now(),
+                        response.lease_ttl_ms,
+                    )?,
+                })
+            }
+            .await,
         )
         .await
-        .map_err(|(error, _)| error)
-    }
-
-    pub(crate) async fn acquire_scanner_publication_lease_with_connection_failure(
-        &self,
-        expected_session_id: &str,
-        expected_generation: u64,
-        observational_only: bool,
-    ) -> std::result::Result<ScannerPublicationLease, (Error, Option<ScannerLeaseConnectionFailure>)> {
-        let request_started = std::time::Instant::now();
-        let mut connection_failure = None;
-        let result = self
-            .finalize_result(
-                async {
-                    let challenge = Uuid::new_v4();
-                    let mut client = self
-                        .get_client_with_connection_failure()
-                        .await
-                        .map_err(|(error, failure)| {
-                            connection_failure = failure;
-                            error
-                        })?
-                        .max_decoding_message_size(SCANNER_ACTIVITY_MAX_MESSAGE_SIZE)
-                        .max_encoding_message_size(SCANNER_ACTIVITY_MAX_MESSAGE_SIZE);
-                    let mut request = Request::new(ScannerPublicationLeaseRequest {
-                        challenge: challenge.as_bytes().to_vec().into(),
-                        expected_movement_generation: expected_generation,
-                        ttl_ms: crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS,
-                        expected_session_id: expected_session_id.to_string(),
-                        token: Bytes::new(),
-                        observational_only,
-                    });
-                    let canonical =
-                        rustfs_protos::canonical_scanner_publication_lease_request_body(request.get_ref()).map_err(|_| {
-                            scanner_publication_lease_local_error(
-                                "request_too_large",
-                                "scanner publication lease request is too large to authenticate",
-                            )
-                        })?;
-                    set_tonic_canonical_body_digest(&mut request, &canonical)
-                        .map_err(|error| scanner_publication_lease_local_io_error("request_digest_failed", error))?;
-                    let response = client.acquire_scanner_publication_lease(request).await?.into_inner();
-                    let response_body =
-                        rustfs_protos::canonical_scanner_publication_lease_response_body(challenge.as_bytes(), &response)
-                            .map_err(|_| {
-                                scanner_publication_lease_local_error(
-                                    "response_too_large",
-                                    "scanner publication lease response is too large to authenticate",
-                                )
-                            })?;
-                    verify_tonic_rpc_response_proof(&response_body, &response.response_proof).map_err(|_| {
-                        scanner_publication_lease_local_error(
-                            "invalid_proof",
-                            "peer returned an invalid scanner publication lease proof",
-                        )
-                    })?;
-                    let (token, owner_id) = validate_scanner_publication_lease_response_fields(
-                        &response,
-                        expected_session_id,
-                        expected_generation,
-                        observational_only,
-                    )?;
-                    Ok(ScannerPublicationLease {
-                        token,
-                        movement_generation: response.movement_generation,
-                        observational_only,
-                        owner_id,
-                        session_id: response.session_id,
-                        expires_at: scanner_publication_lease_deadline(
-                            request_started,
-                            std::time::Instant::now(),
-                            response.lease_ttl_ms,
-                        )?,
-                    })
-                }
-                .await,
-            )
-            .await;
-        result.map_err(|error| (error, connection_failure))
     }
 
     /// Revalidate the exact token immediately before the coordinator commits
@@ -3362,162 +3146,6 @@ mod tests {
         let error = scanner_publication_lease_deadline(started, started + Duration::from_secs(55), 60_000)
             .expect_err("a response arriving at the safety boundary must fail closed");
         assert!(error.to_string().contains("after its safety window"));
-        assert_eq!(scanner_publication_lease_error_class(&error).reason, Some("response_after_safety_window"));
-        assert!(
-            error
-                .to_string()
-                .ends_with("scanner publication lease response arrived after its safety window")
-        );
-    }
-
-    #[test]
-    fn scanner_publication_lease_validation_errors_are_typed_without_changing_display() {
-        let valid = ScannerPublicationLeaseResponse {
-            success: true,
-            token: Uuid::new_v4().as_bytes().to_vec().into(),
-            movement_generation: 7,
-            lease_ttl_ms: crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS,
-            error: None,
-            response_proof: Bytes::new(),
-            owner_id: Uuid::new_v4().to_string(),
-            session_id: "session-a".to_owned(),
-            observational_only: false,
-        };
-        let cases = [
-            (
-                ScannerPublicationLeaseResponse {
-                    movement_generation: 6,
-                    ..valid.clone()
-                },
-                "generation_mismatch",
-                "peer returned a different scanner publication lease generation",
-            ),
-            (
-                ScannerPublicationLeaseResponse {
-                    session_id: "other".to_owned(),
-                    ..valid.clone()
-                },
-                "session_mismatch",
-                "peer returned a different scanner publication lease session",
-            ),
-            (
-                ScannerPublicationLeaseResponse {
-                    observational_only: true,
-                    ..valid.clone()
-                },
-                "purpose_mismatch",
-                "peer returned a different scanner publication lease purpose",
-            ),
-            (
-                ScannerPublicationLeaseResponse {
-                    owner_id: Uuid::nil().to_string(),
-                    ..valid.clone()
-                },
-                "invalid_owner",
-                "peer returned an invalid scanner publication lease owner",
-            ),
-            (
-                ScannerPublicationLeaseResponse {
-                    lease_ttl_ms: 0,
-                    ..valid.clone()
-                },
-                "unsupported_ttl",
-                "peer returned an unsupported scanner publication lease TTL",
-            ),
-            (
-                ScannerPublicationLeaseResponse {
-                    token: Bytes::from_static(b"short"),
-                    ..valid.clone()
-                },
-                "invalid_token",
-                "peer returned an invalid scanner publication lease token",
-            ),
-        ];
-        for (response, reason, message) in cases {
-            let error = validate_scanner_publication_lease_response_fields(&response, "session-a", 7, false)
-                .expect_err("invalid lease response must be rejected");
-            let class = scanner_publication_lease_error_class(&error);
-            assert_eq!(class.reason, Some(reason));
-            assert_eq!(class.reported_reason, None);
-            assert_eq!(error.to_string(), format!("Io error: {message}"));
-            assert!(matches!(error, Error::Io(ref io) if io.kind() == std::io::ErrorKind::Other));
-        }
-
-        for (reason, message) in [
-            ("request_too_large", "scanner publication lease request is too large to authenticate"),
-            ("response_too_large", "scanner publication lease response is too large to authenticate"),
-            ("invalid_proof", "peer returned an invalid scanner publication lease proof"),
-            ("deadline_overflow", "scanner publication lease deadline overflowed"),
-        ] {
-            let error = scanner_publication_lease_local_error(reason, message);
-            assert_eq!(scanner_publication_lease_error_class(&error).reason, Some(reason));
-            assert_eq!(error.to_string(), format!("Io error: {message}"));
-            assert!(matches!(error, Error::Io(ref io) if io.kind() == std::io::ErrorKind::Other));
-        }
-
-        let original = std::io::Error::new(std::io::ErrorKind::InvalidInput, "digest rejected");
-        let expected = format!("Io error: {original}");
-        let wrapped = scanner_publication_lease_local_io_error("request_digest_failed", original);
-        assert_eq!(scanner_publication_lease_error_class(&wrapped).reason, Some("request_digest_failed"));
-        assert_eq!(wrapped.to_string(), expected);
-        assert!(matches!(wrapped, Error::Io(ref io) if io.kind() == std::io::ErrorKind::InvalidInput));
-    }
-
-    #[test]
-    fn scanner_publication_lease_error_class_keeps_remote_reports_separate_from_local_errors() {
-        for (remote_message, reported) in [
-            ("Io error: scanner publication lease capacity is exhausted", Some("capacity_exhausted")),
-            ("Io error: scanner publication lease generation is stale", Some("generation_stale")),
-            (
-                "Io error: scanner publication lease is blocked by data movement",
-                Some("blocked_by_data_movement"),
-            ),
-            ("remote-secret Io error: scanner publication lease generation is stale", None),
-            ("not Io error: scanner publication lease generation is stale", None),
-            ("scanner publication lease generation is stale", None),
-            ("Io error: scanner publication lease generation is stale ", None),
-            ("Io error: scanner publication lease is blocked by pending namespace commit", None),
-            ("Io error: scanner publication lease namespace changed during acquisition", None),
-        ] {
-            let rejected = ScannerPublicationLeaseResponse {
-                success: false,
-                error: Some(rustfs_protos::proto_gen::node_service::Error {
-                    code: 0,
-                    error_info: remote_message.to_owned(),
-                }),
-                ..Default::default()
-            };
-            let error = validate_scanner_publication_lease_response_fields(&rejected, "session", 0, false)
-                .expect_err("peer rejection must be returned");
-            let class = scanner_publication_lease_error_class(&error);
-            assert_eq!(class.reason, None);
-            assert_eq!(class.reported_reason, reported);
-            assert_eq!(error.to_string(), format!("Io error: {remote_message}"));
-            assert!(!format!("{error:?}").contains("remote-secret"));
-        }
-
-        let forged = Error::other("peer returned a different scanner publication lease generation");
-        let class = scanner_publication_lease_error_class(&forged);
-        assert_eq!(class.reason, None, "untyped error text must not be promoted to a local diagnosis");
-        assert_eq!(class.reported_reason, None, "untyped text must not be attributed to a peer response");
-
-        for (error, expected) in [
-            (Error::RemoteClientUnavailable("remote-secret".to_owned()), "client_unavailable"),
-            (Error::RemoteClientUnavailable("pretend timed out".to_owned()), "client_unavailable"),
-            (Error::Timeout, "timeout"),
-            (tonic::Status::unavailable("remote-secret").into(), "rpc_unavailable"),
-            (tonic::Status::deadline_exceeded("remote-secret").into(), "rpc_deadline_exceeded"),
-            (tonic::Status::permission_denied("remote-secret").into(), "rpc_permission_denied"),
-            (
-                Error::Io(std::io::Error::new(std::io::ErrorKind::TimedOut, "remote-secret")),
-                "io_timeout",
-            ),
-        ] {
-            let class = scanner_publication_lease_error_class(&error);
-            assert_eq!(class.reason, Some(expected));
-            assert_eq!(class.reported_reason, None);
-            assert!(!format!("{class:?}").contains("remote-secret"));
-        }
     }
 
     #[test]

crates/ecstore/src/core/pools.rs

@@ -28300,6 +28300,167 @@ mod tests {
         );
     }
 
+    /// One version commits, then every in-place attempt to copy the next one
+    /// times out on the target namespace lock. The entry is deferred, not
+    /// failed, with both source versions retained; the resumed entry finds the
+    /// committed version and moves both with their exact bytes.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn decommission_defers_target_lookup_lock_timeout_and_retries_without_losing_source() {
+        use tokio::io::AsyncReadExt as _;
+
+        let fixture = NullVersionDecommission::new(NullVersionTop::NewerVersion).await;
+        let object = fixture.object;
+        let injected = Arc::new(AtomicUsize::new(0));
+        let attempted = Arc::new(AtomicUsize::new(0));
+        let fault = DecommissionTestFaultGuard::install_with_error(
+            Arc::new({
+                let injected = Arc::clone(&injected);
+                let attempted = Arc::clone(&attempted);
+                move |stage, _, faulted, _, success| {
+                    let target = stage == "before_target_commit" && faulted == object && success;
+                    let should_inject = target && attempted.fetch_add(1, Ordering::SeqCst) > 0;
+                    if should_inject {
+                        injected.fetch_add(1, Ordering::SeqCst);
+                    }
+                    should_inject
+                }
+            }),
+            Some(Arc::new(|_, bucket, object| {
+                data_movement::data_movement_stage_error_for_test(
+                    "decommission_object",
+                    "prepare_put_object",
+                    bucket,
+                    object,
+                    Error::Lock(rustfs_lock::LockError::timeout(
+                        format!("{bucket}/{object}@latest"),
+                        std::time::Duration::from_secs(5),
+                    )),
+                )
+            })),
+        );
+        let (outcome, _) = fixture.run_entry().await;
+        assert_eq!(injected.load(Ordering::SeqCst), DECOMMISSION_VERSION_COPY_ATTEMPTS);
+        fixture.assert_source_retained().await;
+        assert_eq!(
+            fixture
+                .target_versions()
+                .await
+                .expect("the first version was copied")
+                .versions
+                .len(),
+            1,
+            "one version must commit before the next version times out"
+        );
+        assert!(
+            matches!(&outcome, DecommissionEntryOutcome::Deferred(err) if is_decommission_copy_lock_timeout_deferral(err)),
+            "{outcome:?}"
+        );
+        assert_eq!(
+            fixture.store.pool_meta.read().await.pools[0]
+                .decommission
+                .as_ref()
+                .expect("active decommission")
+                .items_decommission_failed,
+            0,
+            "a deferred entry must not be counted as a permanent failure"
+        );
+        drop(fault);
+        let (outcome, _) = fixture.run_entry().await;
+        assert!(matches!(outcome, DecommissionEntryOutcome::Complete), "{outcome:?}");
+        fixture.assert_moved().await;
+        let mut reader = fixture.store.pools[1]
+            .get_object_reader(
+                fixture.bucket,
+                fixture.object,
+                None,
+                HeaderMap::new(),
+                &ObjectOptions {
+                    version_id: Some(fixture.seeded.versions[0].version_id.expect("newer version").to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("the newer version must remain readable");
+        let mut actual = Vec::new();
+        reader.stream.read_to_end(&mut actual).await.expect("read the newer version");
+        assert_eq!(actual, b"written after versioning was enabled");
+    }
+
+    /// A namespace lock still held on the target key, as by a copy whose node
+    /// was killed mid-write, makes target selection fail with a typed
+    /// object-lock timeout. A real held lock, not an injected error, must defer
+    /// the entry rather than fail it; once the lock is released selection
+    /// recovers and the resumed entry moves both versions.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn decommission_target_selection_returns_timeout_for_a_held_object_lock() {
+        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("1"))], async {
+            let fixture = NullVersionDecommission::new(NullVersionTop::NewerVersion).await;
+            let target_set = fixture.store.pools[1].get_disks_by_key(fixture.object);
+            let lock = target_set
+                .new_ns_lock(fixture.bucket, fixture.object)
+                .await
+                .expect("target lock");
+            let guard = lock
+                .get_write_lock(std::time::Duration::from_secs(30))
+                .await
+                .expect("hold target lock");
+            let opts = ObjectOptions {
+                data_movement: true,
+                src_pool_idx: 0,
+                version_id: Some(uuid::Uuid::nil().to_string()),
+                ..Default::default()
+            };
+            let select = || {
+                fixture
+                    .store
+                    .select_data_movement_pool_idx(fixture.bucket, fixture.object, 1, &opts, false)
+            };
+            let result = tokio::time::timeout(get_lock_acquire_timeout() + std::time::Duration::from_secs(15), select())
+                .await
+                .expect("target selection must terminate");
+            match result {
+                Err(Error::Lock(rustfs_lock::LockError::Timeout { resource, .. })) => {
+                    assert!(resource.contains(fixture.bucket) && resource.contains(fixture.object), "{resource}");
+                }
+                other => panic!("expected an object lock timeout, got {other:?}"),
+            }
+
+            let (outcome, _) = fixture.run_entry().await;
+            assert!(
+                matches!(&outcome, DecommissionEntryOutcome::Deferred(err) if is_decommission_copy_lock_timeout_deferral(err)),
+                "{outcome:?}"
+            );
+            assert_eq!(
+                fixture.store.pool_meta.read().await.pools[0]
+                    .decommission
+                    .as_ref()
+                    .expect("active decommission")
+                    .items_decommission_failed,
+                0,
+                "a held target lock must not be counted as a permanent failure"
+            );
+            fixture.assert_source_retained().await;
+            assert!(
+                fixture.target_versions().await.is_none(),
+                "no version may commit while the target lock is held"
+            );
+
+            drop(guard);
+            assert_eq!(
+                select()
+                    .await
+                    .expect("target selection must recover once the lock is released"),
+                1
+            );
+            let (outcome, _) = fixture.run_entry().await;
+            assert!(matches!(outcome, DecommissionEntryOutcome::Complete), "{outcome:?}");
+            fixture.assert_moved().await;
+        })
+        .await;
+    }
+
     #[test]
     fn decommission_object_attempt_metric_keeps_the_peak() {
         let mut attempts = 0;

crates/ecstore/src/disk/local.rs

@@ -11309,34 +11309,57 @@ impl LocalDisk {
 #[cfg(unix)]
 pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, usize)>) -> Vec<Result<Bytes>> {
     let n = requests.len();
+    let metrics = rustfs_io_metrics::LocalBatchReadMetrics::capture();
+    let submitted = metrics.as_ref().map(|_| std::time::Instant::now());
     tokio::task::spawn_blocking(move || {
+        use rustfs_io_metrics::LocalBatchReadStage;
         use std::os::unix::fs::FileExt;
 
+        let execution_started = metrics.as_ref().map(|_| std::time::Instant::now());
+        if let (Some(metrics), Some(submitted)) = (&metrics, submitted) {
+            metrics.record_stage(LocalBatchReadStage::QueueWait, true, submitted.elapsed());
+            metrics.record_batch(n);
+        }
         let mut results = Vec::with_capacity(n);
         for (file_path, offset, length) in requests {
+            let mut bytes_read = 0;
             let r = (|| -> Result<Bytes> {
-                let meta = std::fs::metadata(&file_path).map_err(DiskError::from)?;
+                let meta = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Metadata, || {
+                    std::fs::metadata(&file_path).map_err(DiskError::from)
+                })?;
                 let end = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?;
                 if meta.len() < u64::try_from(end).unwrap_or(u64::MAX) {
                     return Err(DiskError::FileCorrupt);
                 }
 
-                let file = std::fs::File::open(&file_path).map_err(DiskError::from)?;
+                let file = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Open, || {
+                    std::fs::File::open(&file_path).map_err(DiskError::from)
+                })?;
                 let mut buf = vec![0u8; length];
-                let mut total = 0usize;
-                while total < length {
-                    let nbytes = file
-                        .read_at(&mut buf[total..], u64::try_from(offset + total).unwrap_or(u64::MAX))
-                        .map_err(DiskError::from)?;
-                    if nbytes == 0 {
-                        return Err(DiskError::FileCorrupt);
-                    }
-                    total += nbytes;
+                if length > 0 {
+                    measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Read, || {
+                        while bytes_read < length {
+                            let nbytes = file
+                                .read_at(&mut buf[bytes_read..], u64::try_from(offset + bytes_read).unwrap_or(u64::MAX))
+                                .map_err(DiskError::from)?;
+                            if nbytes == 0 {
+                                return Err(DiskError::FileCorrupt);
+                            }
+                            bytes_read += nbytes;
+                        }
+                        Ok(())
+                    })?;
                 }
                 Ok(Bytes::from(buf))
             })();
+            if let Some(metrics) = &metrics {
+                metrics.record_shard(r.is_ok(), length, bytes_read);
+            }
             results.push(r);
         }
+        if let (Some(metrics), Some(started)) = (&metrics, execution_started) {
+            metrics.record_stage(LocalBatchReadStage::Execution, results.iter().all(Result::is_ok), started.elapsed());
+        }
         results
     })
     .await
@@ -11346,6 +11369,20 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize,
     })
 }
 
+#[cfg(unix)]
+fn measure_local_batch_stage<T>(
+    metrics: Option<&rustfs_io_metrics::LocalBatchReadMetrics>,
+    stage: rustfs_io_metrics::LocalBatchReadStage,
+    operation: impl FnOnce() -> Result<T>,
+) -> Result<T> {
+    let started = metrics.map(|_| std::time::Instant::now());
+    let result = operation();
+    if let (Some(metrics), Some(started)) = (metrics, started) {
+        metrics.record_stage(stage, result.is_ok(), started.elapsed());
+    }
+    result
+}
+
 #[async_trait::async_trait]
 impl DiskAPI for LocalDisk {
     async fn supports_guarded_delete_versions(&self) -> Result<()> {
@@ -30957,6 +30994,109 @@ mod test {
         );
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn local_batch_read_metrics_cover_success_and_unattempted_failure_stages() {
+        use crate::test_metrics::CapturingRecorder;
+
+        let dir = tempfile::tempdir().expect("batch metrics directory");
+        let good = dir.path().join("good.bin");
+        std::fs::write(&good, b"good data").expect("write batch metrics payload");
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("batch metrics runtime");
+        let recorder = CapturingRecorder::default();
+        let previous = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+        let results = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(batch_shard_pread(vec![
+                (good.clone(), 0, 9),
+                (dir.path().join("missing.bin"), 0, 4),
+                (good.clone(), 0, 1024),
+                (good, 0, 0),
+            ]))
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous);
+
+        assert_eq!(results[0].as_ref().expect("valid shard read").as_ref(), b"good data");
+        assert!(results[1].is_err(), "missing shard must remain an error");
+        assert!(matches!(results[2], Err(DiskError::FileCorrupt)), "range failure must preserve its error");
+        assert!(results[3].as_ref().expect("empty range succeeds").is_empty());
+        let counter = "rustfs_io_get_object_local_batch_shards_total";
+        assert_eq!(recorder.counter_value(counter, &[("outcome", "success")]), 2);
+        assert_eq!(recorder.counter_value(counter, &[("outcome", "error")]), 2);
+        assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 9);
+        let stages = "rustfs_io_get_object_local_batch_stage_seconds";
+        for (stage, success, errors) in [("metadata", 3, 1), ("open", 2, 0), ("read", 1, 0), ("execution", 0, 1)] {
+            assert_eq!(
+                recorder
+                    .histogram_values(stages, &[("stage", stage), ("outcome", "success")])
+                    .len(),
+                success
+            );
+            assert_eq!(
+                recorder
+                    .histogram_values(stages, &[("stage", stage), ("outcome", "error")])
+                    .len(),
+                errors
+            );
+        }
+        assert_eq!(recorder.histogram_values(stages, &[("stage", "queue_wait")]).len(), 1);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn local_batch_read_metrics_survive_caller_cancellation() {
+        use crate::test_metrics::CapturingRecorder;
+
+        let dir = tempfile::tempdir().expect("cancelled batch directory");
+        let path = dir.path().join("shard");
+        std::fs::write(&path, b"data").expect("write cancellation fixture");
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .max_blocking_threads(1)
+            .enable_all()
+            .build()
+            .expect("single blocking-worker runtime");
+        let recorder = CapturingRecorder::default();
+        let previous = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+        metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+                let (release_tx, release_rx) = std::sync::mpsc::channel();
+                let blocker = tokio::task::spawn_blocking(move || {
+                    started_tx.send(()).expect("signal occupied blocking worker");
+                    release_rx.recv().expect("release occupied blocking worker");
+                });
+                started_rx.await.expect("blocking worker has started");
+                let mut batch = Box::pin(batch_shard_pread(vec![(path, 0, 4)]));
+                let pending = futures::FutureExt::now_or_never(&mut batch).is_none();
+                drop(batch);
+                rustfs_io_metrics::set_get_stage_metrics_enabled(false);
+                release_tx.send(()).expect("release worker after dropping caller");
+                blocker.await.expect("blocking worker finishes");
+                assert!(pending, "batch must have been submitted while the worker was occupied");
+                tokio::time::timeout(Duration::from_secs(5), async {
+                    while recorder
+                        .histogram_values("rustfs_io_get_object_local_batch_stage_seconds", &[("stage", "execution")])
+                        .is_empty()
+                    {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect("detached batch must still record completion");
+            });
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous);
+        assert_eq!(
+            recorder.counter_value("rustfs_io_get_object_local_batch_shards_total", &[("outcome", "success")]),
+            1
+        );
+        assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 4);
+    }
+
     #[cfg(unix)]
     #[tokio::test]
     async fn test_batch_shard_pread_basic() {

crates/ecstore/src/services/notification_sys.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use crate::bucket::lifecycle::tier_last_day_stats::DailyAllTierStats;
-use crate::cluster::rpc::peer_rest_client::{ScannerLeaseConnectionFailure, scanner_publication_lease_error_class};
 use crate::cluster::rpc::{
     PeerRestClient, ScannerDirtyUsageAcknowledgement, ScannerPeerActivity, ScannerPeerDirtyUsageSnapshot,
     ScannerPublicationLease, TierConfigReloadOutcome,
@@ -159,132 +158,16 @@ struct PeerAdminCache {
     last_server_success: Option<SystemTime>,
 }
 
-enum ScannerPublicationLeaseFailure {
-    DuplicateTarget {
-        peer: String,
-    },
-    PeerUnavailable {
-        peer: String,
-    },
-    RequestFailed {
-        peer: String,
-        purpose: &'static str,
-        detail: String,
-        reason: Option<&'static str>,
-        reported_reason: Option<&'static str>,
-    },
-}
-
-impl std::fmt::Debug for ScannerPublicationLeaseFailure {
-    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        formatter.write_str("ScannerPublicationLeaseFailure")
-    }
-}
-
-impl std::fmt::Display for ScannerPublicationLeaseFailure {
-    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Self::DuplicateTarget { peer } => write!(formatter, "duplicate scanner publication lease target: {peer}"),
-            Self::PeerUnavailable { peer } => write!(formatter, "scanner publication lease peer {peer} is unavailable"),
-            Self::RequestFailed {
-                peer, purpose, detail, ..
-            } => write!(
-                formatter,
-                "scanner publication lease acquisition failed: peer={peer}, purpose={purpose}: {detail}"
-            ),
-        }
-    }
-}
-
-impl std::error::Error for ScannerPublicationLeaseFailure {}
-
-/// Bounded details from the target that actually failed lease acquisition.
-#[derive(Clone, Copy, Debug)]
-pub struct ScannerPublicationLeaseFailureInfo<'a> {
-    pub peer: Option<&'a str>,
-    pub kind: &'static str,
-    pub reason: Option<&'static str>,
-    pub reported_reason: Option<&'static str>,
-}
-
-fn scanner_publication_lease_target_peer(target: &str) -> Option<&str> {
-    if target.len() > 267 {
-        return None;
-    }
-    let (peer, parse_target) = if let Some(peer) = target.strip_prefix("http://").or_else(|| target.strip_prefix("https://")) {
-        (peer, std::borrow::Cow::Borrowed(target))
-    } else {
-        (target, std::borrow::Cow::Owned(format!("http://{target}")))
-    };
-    if peer.is_empty()
-        || peer.len() > 259
-        || peer.ends_with(':')
-        || !peer
-            .bytes()
-            .all(|byte| byte.is_ascii_alphanumeric() || b".-_:[]".contains(&byte))
-    {
-        return None;
-    }
-    let parsed = url::Url::parse(&parse_target).ok()?;
-    if !matches!(parsed.scheme(), "http" | "https")
-        || parsed.host().is_none()
-        || !parsed.username().is_empty()
-        || parsed.password().is_some()
-        || parsed.path() != "/"
-        || parsed.query().is_some()
-        || parsed.fragment().is_some()
-    {
-        return None;
-    }
-    Some(peer)
-}
-
-/// Return structured lease failure context without parsing remote error text.
-pub fn scanner_publication_lease_failure_info(error: &Error) -> Option<ScannerPublicationLeaseFailureInfo<'_>> {
-    let Error::Io(error) = error else { return None };
-    let failure = error.get_ref()?.downcast_ref::<ScannerPublicationLeaseFailure>()?;
-    let (peer, kind, reason, reported_reason) = match failure {
-        ScannerPublicationLeaseFailure::DuplicateTarget { peer } => (peer, "duplicate_target", None, None),
-        ScannerPublicationLeaseFailure::PeerUnavailable { peer } => (peer, "peer_unavailable", None, None),
-        ScannerPublicationLeaseFailure::RequestFailed {
-            peer,
-            reason,
-            reported_reason,
-            ..
-        } => (peer, "peer_request_failed", *reason, *reported_reason),
-    };
-    let safe_peer = scanner_publication_lease_target_peer(peer);
-    Some(ScannerPublicationLeaseFailureInfo {
-        peer: safe_peer,
-        kind,
-        reason,
-        reported_reason,
-    })
-}
-
-fn scanner_publication_lease_acquisition_error(
-    host: &str,
-    observational_only: bool,
-    error: Error,
-    connection_failure: Option<ScannerLeaseConnectionFailure>,
-) -> Error {
+fn scanner_publication_lease_acquisition_error(host: &str, observational_only: bool, error: Error) -> Error {
     let purpose = if observational_only {
         "observational"
     } else {
         "authoritative"
     };
     // Retry classification depends on the original error suffix.
-    let class = scanner_publication_lease_error_class(&error);
-    Error::other(ScannerPublicationLeaseFailure::RequestFailed {
-        peer: host.to_owned(),
-        purpose,
-        detail: error.to_string(),
-        reason: match connection_failure {
-            Some(ScannerLeaseConnectionFailure::ChannelAcquisitionTimedOut) => Some("channel_acquisition_timed_out"),
-            None => class.reason,
-        },
-        reported_reason: class.reported_reason,
-    })
+    Error::other(format!(
+        "scanner publication lease acquisition failed: peer={host}, purpose={purpose}: {error}"
+    ))
 }
 
 #[derive(Default)]
@@ -3278,7 +3161,7 @@ impl NotificationSys {
         targets.sort_by(|left, right| left.0.cmp(&right.0));
         for pair in targets.windows(2) {
             if pair[0].0 == pair[1].0 {
-                return Err(Error::other(ScannerPublicationLeaseFailure::DuplicateTarget { peer: pair[0].0.clone() }));
+                return Err(Error::other(format!("duplicate scanner publication lease target: {}", pair[0].0)));
             }
         }
 
@@ -3292,21 +3175,16 @@ impl NotificationSys {
                 .cloned()
             else {
                 let _ = self.release_scanner_publication_leases(grants).await;
-                return Err(Error::other(ScannerPublicationLeaseFailure::PeerUnavailable { peer: host }));
+                return Err(Error::other(format!("scanner publication lease peer {host} is unavailable")));
             };
             match client
-                .acquire_scanner_publication_lease_with_connection_failure(&session_id, generation, observational_only)
+                .acquire_scanner_publication_lease(&session_id, generation, observational_only)
                 .await
             {
                 Ok(lease) => grants.push(ScannerPublicationLeaseGrant { host, lease }),
-                Err((err, connection_failure)) => {
+                Err(err) => {
                     let _ = self.release_scanner_publication_leases(grants).await;
-                    return Err(scanner_publication_lease_acquisition_error(
-                        &host,
-                        observational_only,
-                        err,
-                        connection_failure,
-                    ));
+                    return Err(scanner_publication_lease_acquisition_error(&host, observational_only, err));
                 }
             }
         }
@@ -4307,243 +4185,20 @@ mod tests {
 
     #[test]
     fn scanner_lease_acquisition_error_identifies_peer_and_preserves_retry_suffix() {
-        use crate::cluster::rpc::peer_rest_client::scanner_publication_lease_remote_rejection;
-
         for (observational_only, purpose) in [(false, "authoritative"), (true, "observational")] {
-            let reason = "scanner publication lease capacity is exhausted";
-            let source = scanner_publication_lease_remote_rejection(format!("Io error: {reason}"));
-            let expected = format!(
-                "Io error: scanner publication lease acquisition failed: peer=peer-a.example.invalid:9000, purpose={purpose}: {source}"
-            );
-            let error =
-                scanner_publication_lease_acquisition_error("peer-a.example.invalid:9000", observational_only, source, None);
-            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
-            assert_eq!(info.peer, Some("peer-a.example.invalid:9000"));
-            assert_eq!(info.kind, "peer_request_failed");
-            assert_eq!(info.reason, None);
-            assert_eq!(info.reported_reason, Some("capacity_exhausted"));
-            assert_eq!(error.to_string(), expected);
-            assert!(
-                error.to_string().ends_with(reason),
-                "retry classification must retain the original suffix"
-            );
-            let cloned = error.clone();
-            assert_eq!(cloned.to_string(), expected);
-            assert!(
-                scanner_publication_lease_failure_info(&cloned).is_none(),
-                "a cloned generic I/O error must not invent lost structured context"
-            );
+            let reason = "scanner publication lease is blocked by pending namespace commit";
+            let error = scanner_publication_lease_acquisition_error(
+                "peer-a.example.invalid:9000",
+                observational_only,
+                Error::other(reason),
+            )
+            .to_string();
+            assert!(error.contains("peer-a.example.invalid:9000"), "missing peer: {error}");
+            assert!(error.contains(purpose), "missing purpose: {error}");
+            assert!(error.ends_with(reason), "retry classification must retain the original suffix");
         }
     }
 
-    #[test]
-    fn scanner_lease_failure_info_uses_the_actual_target_and_rejects_unbounded_peers() {
-        for (failure, expected_display, expected_kind) in [
-            (
-                ScannerPublicationLeaseFailure::DuplicateTarget {
-                    peer: "peer-a:9000".to_owned(),
-                },
-                "Io error: duplicate scanner publication lease target: peer-a:9000",
-                "duplicate_target",
-            ),
-            (
-                ScannerPublicationLeaseFailure::PeerUnavailable {
-                    peer: "peer-a:9000".to_owned(),
-                },
-                "Io error: scanner publication lease peer peer-a:9000 is unavailable",
-                "peer_unavailable",
-            ),
-        ] {
-            let error = Error::other(failure);
-            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
-            assert_eq!(info.peer, Some("peer-a:9000"));
-            assert_eq!(info.kind, expected_kind);
-            assert_eq!(info.reason, None);
-            assert_eq!(info.reported_reason, None);
-            assert_eq!(error.to_string(), expected_display);
-        }
-
-        let forged =
-            Error::other("scanner publication lease acquisition failed: peer=forged:9000, purpose=authoritative: rejected");
-        assert!(
-            scanner_publication_lease_failure_info(&forged).is_none(),
-            "raw error text must not establish peer identity"
-        );
-        for invalid in [
-            "peer-a:9000\nforged",
-            "user:secret@peer-a:9000",
-            "http://user:secret@peer-a:9000",
-            "http://@peer-a:9000",
-            "http://peer-a:65536",
-            "https://peer-a:bad",
-            "https://peer-a:",
-            "http://[2001:db8::1",
-            "http://[2001:db8::1]:bad",
-            "https://peer-a:9000/path",
-            "https://peer-a:9000?token=secret",
-            "https://peer-a:9000#secret",
-            "https://peer-a%2eexample:9000",
-            "ftp://peer-a:9000",
-            "",
-        ] {
-            let error = scanner_publication_lease_acquisition_error(invalid, false, Error::other("remote-secret"), None);
-            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
-            assert_eq!(info.peer, None, "invalid target must not reach logs");
-            assert_eq!(info.reported_reason, None, "unrecognized remote text must not become a reason");
-            assert!(!format!("{error:?}").contains("remote-secret"), "Debug must not expose remote error text");
-        }
-        let oversized = format!("https://{}:9000", "a".repeat(260));
-        let error = scanner_publication_lease_acquisition_error(&oversized, false, Error::other("remote-secret"), None);
-        assert_eq!(scanner_publication_lease_failure_info(&error).and_then(|info| info.peer), None);
-    }
-
-    #[test]
-    fn scanner_lease_failure_info_accepts_real_grid_host_urls() {
-        use crate::cluster::rpc::peer_rest_client::scanner_publication_lease_remote_rejection;
-        use crate::layout::endpoint::Endpoint;
-
-        for (endpoint, expected) in [
-            ("http://peer-a.example:9000/export", "peer-a.example:9000"),
-            ("https://peer-b.example:9443/export", "peer-b.example:9443"),
-            ("http://[2001:db8::1]:9000/export", "[2001:db8::1]:9000"),
-            ("https://peer-c.example/export", "peer-c.example"),
-        ] {
-            let target = Endpoint::try_from(endpoint).expect("valid endpoint").grid_host();
-            let source = scanner_publication_lease_remote_rejection(
-                "Io error: scanner publication lease capacity is exhausted".to_owned(),
-            );
-            let expected_display =
-                format!("Io error: scanner publication lease acquisition failed: peer={target}, purpose=observational: {source}");
-            let error = scanner_publication_lease_acquisition_error(&target, true, source, None);
-            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
-            assert_eq!(info.peer, Some(expected));
-            assert_eq!(info.reason, None);
-            assert_eq!(info.reported_reason, Some("capacity_exhausted"));
-            assert_eq!(error.to_string(), expected_display);
-            assert!(error.to_string().ends_with("scanner publication lease capacity is exhausted"));
-        }
-    }
-
-    #[test]
-    fn scanner_lease_failure_info_preserves_typed_connection_failures_without_logging_details() {
-        for (source, reason) in [
-            (Error::RemoteClientUnavailable("remote-secret".to_owned()), "client_unavailable"),
-            (Error::Timeout, "timeout"),
-            (tonic::Status::unavailable("remote-secret").into(), "rpc_unavailable"),
-            (tonic::Status::deadline_exceeded("remote-secret").into(), "rpc_deadline_exceeded"),
-        ] {
-            let expected = format!(
-                "Io error: scanner publication lease acquisition failed: peer=https://peer-a:9000, purpose=authoritative: {source}"
-            );
-            let error = scanner_publication_lease_acquisition_error("https://peer-a:9000", false, source, None);
-            let info = scanner_publication_lease_failure_info(&error).expect("structured failure source");
-            assert_eq!(info.peer, Some("peer-a:9000"));
-            assert_eq!(info.kind, "peer_request_failed");
-            assert_eq!(info.reason, Some(reason));
-            assert_eq!(info.reported_reason, None);
-            assert_eq!(error.to_string(), expected);
-            assert!(!format!("{info:?}").contains("remote-secret"));
-        }
-
-        let spoofed = scanner_publication_lease_acquisition_error(
-            "https://peer-a:9000",
-            false,
-            Error::other("peer returned a different scanner publication lease generation"),
-            None,
-        );
-        let info = scanner_publication_lease_failure_info(&spoofed).expect("structured failure source");
-        assert_eq!(info.reason, None);
-        assert_eq!(info.reported_reason, None);
-    }
-
-    #[test]
-    fn scanner_lease_channel_deadline_preserves_public_error_and_diagnostic() {
-        const CHILD_MARKER: &str = "RUSTFS_TEST_SCANNER_LEASE_CHANNEL_DEADLINE_CHILD";
-        if std::env::var_os(CHILD_MARKER).is_none() {
-            // Isolate the zero deadline from concurrently running tests.
-            let module = module_path!()
-                .strip_prefix(concat!(env!("CARGO_CRATE_NAME"), "::"))
-                .unwrap_or(module_path!());
-            let name = format!("{module}::scanner_lease_channel_deadline_preserves_public_error_and_diagnostic");
-            let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
-                .args(["--exact", &name, "--nocapture"])
-                .env(CHILD_MARKER, "1")
-                .env(rustfs_config::ENV_INTERNODE_CONNECT_TIMEOUT_SECS, "0")
-                .env("NO_PROXY", "127.0.0.1,localhost")
-                .env("no_proxy", "127.0.0.1,localhost")
-                .output()
-                .expect("deadline test subprocess");
-            assert!(
-                output.status.success() && String::from_utf8_lossy(&output.stdout).contains("1 passed"),
-                "deadline subprocess failed: {} {}",
-                String::from_utf8_lossy(&output.stdout),
-                String::from_utf8_lossy(&output.stderr)
-            );
-            return;
-        }
-
-        let runtime = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("test runtime");
-        runtime.block_on(async {
-            // Keep the address unique to this isolated subprocess.
-            let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("local listener");
-            let address = listener.local_addr().expect("listener address");
-            let target = format!("https://{address}");
-            let make_client = || {
-                PeerRestClient::new(
-                    XHost {
-                        name: "127.0.0.1".to_owned(),
-                        port: address.port(),
-                        is_port_set: true,
-                    },
-                    target.clone(),
-                )
-            };
-            // Block the real channel cache read so the acquisition deadline
-            // wins independently of OS-specific connection or TLS errors.
-            let cache_read_blocker = rustfs_common::block_connection_cache_reads_for_test().await;
-            let public_error = match make_client().get_client().await {
-                Ok(_) => panic!("a stalled connection must exceed the channel deadline"),
-                Err(error) => error,
-            };
-            assert!(matches!(&public_error, Error::RemoteClientUnavailable(_)));
-            assert!(
-                public_error
-                    .to_string()
-                    .contains("internode channel acquisition deadline elapsed")
-            );
-
-            let mut notification = NotificationSys::new(EndpointServerPools::default()).await;
-            notification.peer_clients.push(Some(make_client()));
-            let lease_error = match notification
-                .acquire_scanner_publication_leases(vec![(target.clone(), "session".into(), 3)])
-                .await
-            {
-                Ok(_) => panic!("a stalled connection cannot grant a lease"),
-                Err(error) => error,
-            };
-            assert_eq!(
-                lease_error.to_string(),
-                format!(
-                    "Io error: scanner publication lease acquisition failed: peer={target}, purpose=authoritative: {public_error}"
-                )
-            );
-            let info = scanner_publication_lease_failure_info(&lease_error).expect("structured lease failure");
-            let expected_peer = address.to_string();
-            assert_eq!(info.peer, Some(expected_peer.as_str()));
-            assert_eq!(info.reason, Some("channel_acquisition_timed_out"));
-            assert_eq!(info.reported_reason, None);
-            assert_eq!(
-                scanner_peer_transport_error_message_is_retryable(&lease_error.to_string()),
-                scanner_peer_transport_error_message_is_retryable(&public_error.to_string())
-            );
-            drop(cache_read_blocker);
-            drop(listener);
-        });
-    }
-
     fn ring(total_size: u64) -> LastDayTierStats {
         let mut stats = LastDayTierStats::default();
         stats.add_stats(TierStats {

crates/ecstore/src/services/rebalance/entry.rs

@@ -1782,6 +1782,285 @@ mod tests {
         fixture.assert_moved().await;
     }
 
+    /// Each drive's copy of `version_id` of `object` on `pool`.
+    #[cfg(feature = "test-util")]
+    async fn drive_copies(store: &ECStore, pool: usize, bucket: &str, object: &str, version_id: uuid::Uuid) -> Vec<FileInfo> {
+        use crate::disk::DiskAPI as _;
+        let set = store.pools[pool].get_disks_by_key(object);
+        let mut copies = Vec::new();
+        for disk in set.disks.read().await.iter().flatten() {
+            let path = disk.path().join(bucket).join(object).join(crate::disk::STORAGE_FORMAT_FILE);
+            let Ok(bytes) = tokio::fs::read(&path).await else {
+                continue;
+            };
+            let meta = FileMeta::load(&bytes).expect("drive metadata should decode");
+            if meta.find_version(Some(version_id)).is_ok() {
+                copies.push(
+                    meta.into_fileinfo(bucket, object, &version_id.to_string(), false, false, false)
+                        .expect("drive copy should decode"),
+                );
+            }
+        }
+        copies
+    }
+
+    #[cfg(feature = "test-util")]
+    fn drive_space(total: u64, used: u64) -> crate::disk::DiskInfo {
+        crate::disk::DiskInfo {
+            total,
+            used,
+            free: total - used,
+            free_inodes: 1 << 20,
+            ..Default::default()
+        }
+    }
+
+    #[cfg(feature = "test-util")]
+    const DANGLING_COPY_REBALANCE_ID: &str = "rebalance-dangling-target-copy";
+
+    /// A delete marker over one object version on pool 0 of three four-drive
+    /// pools, plus what an earlier attempt that lost write quorum left on one
+    /// drive of pool 1: both version ids, with older modification times and
+    /// different bytes. Object reads report such copies as absent. Pool 2
+    /// reports far more free space than pool 1, so a free-space choice picks
+    /// pool 2 (the kill-resume e2e case ended with the marker on two pools).
+    #[cfg(feature = "test-util")]
+    struct DanglingTargetCopy {
+        _temp_dirs: Vec<tempfile::TempDir>,
+        store: Arc<ECStore>,
+        _space: crate::store::PoolSpaceForTest,
+        bucket: &'static str,
+        object: &'static str,
+        incarnation: uuid::Uuid,
+        source_set: Arc<SetDisks>,
+        marker: FileInfo,
+        version: FileInfo,
+        payload: Vec<u8>,
+    }
+
+    #[cfg(feature = "test-util")]
+    impl DanglingTargetCopy {
+        async fn new(holder_space: crate::disk::DiskInfo) -> Self {
+            let (temp_dirs, store, _peer) =
+                crate::services::rebalance::test_three_pool_stores_with_four_disk_sets_with_isolated_node_contexts(None).await;
+            let space = crate::store::PoolSpaceForTest::install(store.id, &[(1, holder_space), (2, drive_space(1 << 50, 0))]);
+            let bucket = "rebalance-dangling-target";
+            let object = "versioned/06.bin";
+            Box::pin(store.make_bucket(bucket, &MakeBucketOptions::default()))
+                .await
+                .expect("create bucket");
+            let incarnation = store.bucket_incarnation_id(bucket).await.expect("bucket incarnation");
+            let mut rebalance_meta = active_rebalance_meta(DANGLING_COPY_REBALANCE_ID);
+            rebalance_meta.pool_stats.push(RebalanceStats::default());
+            rebalance_meta
+                .save(Arc::clone(&store.pools[0]))
+                .await
+                .expect("persist the active rebalance run");
+            *store.rebalance_meta.write().await = Some(rebalance_meta);
+            let versioned = ObjectOptions {
+                versioned: true,
+                expected_bucket_incarnation_id: Some(incarnation),
+                ..Default::default()
+            };
+
+            let payload = vec![0x5a; 2048];
+            let source_set = store.pools[0].get_disks_by_key(object);
+            Box::pin(source_set.put_object(bucket, object, &mut PutObjReader::from_vec(payload.clone()), &versioned))
+                .await
+                .expect("seed the object version");
+            let marker = Box::pin(source_set.delete_object(bucket, object, versioned.clone()))
+                .await
+                .expect("seed the delete marker");
+            assert!(marker.delete_marker);
+            let seeded = source_set
+                .load_file_info_versions_exact(bucket, object)
+                .await
+                .expect("read seeded versions")
+                .expect("seeded versions must exist");
+            let [marker, version] = seeded.versions.as_slice() else {
+                panic!("expected a delete marker over one version: {seeded:?}");
+            };
+            let (marker, version) = (marker.clone(), version.clone());
+            let marker_id = marker.version_id.expect("the marker has a version id");
+            let version_id = version.version_id.expect("the object version has a version id");
+            let earlier = |time: Option<OffsetDateTime>| time.map(|time| time - time::Duration::hours(1));
+
+            let holder = store.pools[1].get_disks_by_key(object);
+            let mut marker_opts =
+                super::super::migration::rebalance_delete_marker_opts(&marker, Some(marker_id.to_string()), 0, Some(incarnation));
+            marker_opts.mod_time = earlier(marker.mod_time);
+            Box::pin(holder.delete_object(bucket, object, marker_opts))
+                .await
+                .expect("publish the earlier marker copy");
+            Box::pin(holder.put_object(
+                bucket,
+                object,
+                &mut PutObjReader::from_vec(vec![0x11; 4096]),
+                &ObjectOptions {
+                    version_id: Some(version_id.to_string()),
+                    mod_time: earlier(version.mod_time),
+                    ..versioned
+                },
+            ))
+            .await
+            .expect("publish the earlier object version copy");
+            {
+                use crate::disk::DiskAPI as _;
+                for disk in holder.disks.read().await.iter().flatten().skip(1) {
+                    tokio::fs::remove_dir_all(disk.path().join(bucket).join(object))
+                        .await
+                        .expect("drop the copies from the drives the attempt missed");
+                }
+            }
+            for version_id in [marker_id, version_id] {
+                assert_eq!(drive_copies(&store, 1, bucket, object, version_id).await.len(), 1);
+            }
+
+            Self {
+                _temp_dirs: temp_dirs,
+                store,
+                _space: space,
+                bucket,
+                object,
+                incarnation,
+                source_set,
+                marker,
+                version,
+                payload,
+            }
+        }
+
+        async fn run_entry(&self) -> RebalanceEntryOutcome {
+            let entry = metacache_entry_from_source(&self.source_set, self.bucket, self.object).await;
+            Box::pin(Arc::clone(&self.store).rebalance_entry(
+                RebalanceEntryTarget {
+                    bucket: self.bucket.to_string(),
+                    pool_index: 0,
+                },
+                entry,
+                Arc::clone(&self.source_set),
+                Arc::new(RebalanceBucketConfigs {
+                    bucket_incarnation_id: Some(self.incarnation),
+                    ..Default::default()
+                }),
+                Arc::from(DANGLING_COPY_REBALANCE_ID),
+                CancellationToken::new(),
+            ))
+            .await
+            .expect("rebalance entry")
+        }
+
+        /// Pools whose drives hold `version_id`, with the number of drives.
+        async fn holders(&self, version_id: uuid::Uuid) -> Vec<(usize, usize)> {
+            let mut holders = Vec::new();
+            for pool in 0..self.store.pools.len() {
+                let drives = drive_copies(&self.store, pool, self.bucket, self.object, version_id)
+                    .await
+                    .len();
+                if drives > 0 {
+                    holders.push((pool, drives));
+                }
+            }
+            holders
+        }
+
+        /// The source entry is gone and `pool` serves both versions as the
+        /// source wrote them, on every drive.
+        async fn assert_moved_to(&self, pool: usize) {
+            assert!(
+                self.source_set
+                    .load_file_info_versions_exact(self.bucket, self.object)
+                    .await
+                    .expect("read source versions")
+                    .is_none(),
+                "the source entry must be cleaned up"
+            );
+            let target = self.store.pools[pool]
+                .get_disks_by_key(self.object)
+                .load_file_info_versions_exact(self.bucket, self.object)
+                .await
+                .expect("read target versions")
+                .expect("both versions must reach the target pool");
+            let moved = |id| {
+                target
+                    .versions
+                    .iter()
+                    .find(|version| version.version_id == id)
+                    .unwrap_or_else(|| panic!("version {id:?} must reach pool {pool}: {target:?}"))
+            };
+            let marker = moved(self.marker.version_id);
+            assert!(marker.deleted, "the marker must stay a delete marker");
+            assert_eq!(marker.mod_time, self.marker.mod_time);
+            let version = moved(self.version.version_id);
+            assert!(!version.deleted);
+            assert_eq!(version.mod_time, self.version.mod_time);
+
+            let version_id = self.version.version_id.expect("object version id");
+            let copies = drive_copies(&self.store, pool, self.bucket, self.object, version_id).await;
+            assert_eq!(copies.len(), 4, "the object version must be on every drive of pool {pool}");
+            for copy in &copies {
+                assert_eq!(copy.mod_time, self.version.mod_time, "an earlier copy must be replaced, not kept");
+                assert_eq!(copy.data_dir, copies[0].data_dir, "every drive must hold the same data");
+            }
+            let marker_id = self.marker.version_id.expect("marker version id");
+            let markers = drive_copies(&self.store, pool, self.bucket, self.object, marker_id).await;
+            assert_eq!(markers.len(), 4, "the marker must be on every drive of pool {pool}");
+            assert!(markers.iter().all(|copy| copy.deleted));
+
+            let mut reader = self.store.pools[pool]
+                .get_object_reader(
+                    self.bucket,
+                    self.object,
+                    None,
+                    HeaderMap::new(),
+                    &ObjectOptions {
+                        version_id: Some(version_id.to_string()),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("the moved version must be readable");
+            let mut actual = Vec::new();
+            reader.stream.read_to_end(&mut actual).await.expect("drain the moved version");
+            assert_eq!(actual, self.payload, "the moved version must carry the source bytes");
+        }
+    }
+
+    /// A retry reuses the pool that holds an earlier attempt's copy, even
+    /// though free space favours another pool, and rewrites that copy from
+    /// the source instead of leaving the version in two pools.
+    #[cfg(feature = "test-util")]
+    #[test]
+    #[serial_test::serial]
+    fn real_rebalance_retry_reuses_the_pool_holding_a_dangling_copy() {
+        run_large_stack_async_test("rebalance-dangling-target-copy", || async {
+            let fixture = DanglingTargetCopy::new(drive_space(64 << 20, 0)).await;
+            assert_eq!(fixture.run_entry().await, RebalanceEntryOutcome::Completed);
+            for version in [&fixture.marker, &fixture.version] {
+                let id = version.version_id.expect("version id");
+                assert_eq!(fixture.holders(id).await, vec![(1, 4)], "version {id} must be in one pool");
+            }
+            fixture.assert_moved_to(1).await;
+        });
+    }
+
+    /// A full pool holding an earlier copy must not pin the retry: placement
+    /// falls back to the free-space choice, and the stale copy stays for heal.
+    #[cfg(feature = "test-util")]
+    #[test]
+    #[serial_test::serial]
+    fn real_rebalance_retry_skips_a_full_pool_holding_a_dangling_copy() {
+        run_large_stack_async_test("rebalance-dangling-full-holder", || async {
+            let fixture = DanglingTargetCopy::new(drive_space(1 << 30, 1 << 30)).await;
+            assert_eq!(fixture.run_entry().await, RebalanceEntryOutcome::Completed);
+            for version in [&fixture.marker, &fixture.version] {
+                let id = version.version_id.expect("version id");
+                assert_eq!(fixture.holders(id).await, vec![(1, 1), (2, 4)], "version {id}");
+            }
+            fixture.assert_moved_to(2).await;
+        });
+    }
+
     /// An `ExpiredObjectAllVersions` expiry is a prefix delete that takes the
     /// bucket lifecycle write lock, while the entry holds that lock for read
     /// through its bucket incarnation fence. Applied inline, it waited out the

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -1234,16 +1234,23 @@ async fn try_create_bitrot_readers_via_batch_pread(
     skip_verify_bitrot: bool,
 ) -> Option<BitrotReaderSetup> {
     use crate::disk::local::batch_shard_pread;
+    use rustfs_io_metrics::{LocalBatchReadDecision, record_local_batch_read_decision};
     use std::io::Cursor;
 
+    let prepare_started = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
+    let record_decision = |decision| {
+        record_local_batch_read_decision(decision, prepare_started.map(|started| started.elapsed()));
+    };
     let (adj_off, adj_len) = adjust_shard_read_params(read_offset, read_length, shard_size, &checksum_algo);
     if adj_len > object_mmap_read_max_length() {
+        record_decision(LocalBatchReadDecision::ReadCapExceeded);
         return None;
     }
 
     let mut batch_items: Vec<(usize, std::path::PathBuf, usize, usize)> = Vec::new();
     for (idx, disk_op) in disks.iter().enumerate() {
         if files.get(idx).is_some_and(|fi| fi.data.is_some()) {
+            record_decision(LocalBatchReadDecision::InlineData);
             return None;
         }
         if let Some(disk) = disk_op.as_ref() {
@@ -1251,12 +1258,20 @@ async fn try_create_bitrot_readers_via_batch_pread(
             let path_str = format!("{object}/{data_dir}/part.{part_number}");
             match disk.get_object_path_for_io_if_local(bucket, &path_str) {
                 Some(Ok(p)) => batch_items.push((idx, p, adj_off, adj_len)),
-                _ => return None,
+                Some(Err(_)) => {
+                    record_decision(LocalBatchReadDecision::InvalidPath);
+                    return None;
+                }
+                None => {
+                    record_decision(LocalBatchReadDecision::NonLocalShard);
+                    return None;
+                }
             }
         }
     }
 
     if batch_items.is_empty() {
+        record_decision(LocalBatchReadDecision::NoLocalShards);
         return None;
     }
 
@@ -1265,6 +1280,7 @@ async fn try_create_bitrot_readers_via_batch_pread(
     for (disk_index, _, _, _) in &batch_items {
         disk_call_counters::record(object, disk_call_counters::KIND_DISK_SHARD_OPEN, *disk_index);
     }
+    record_decision(LocalBatchReadDecision::Selected);
     let batch_results = batch_shard_pread(requests).await;
 
     let mut setup = BitrotReaderSetup::new(disks.len());
@@ -1313,6 +1329,7 @@ async fn try_create_bitrot_readers_via_batch_pread(
     _checksum_algo: HashAlgorithm,
     _skip_verify_bitrot: bool,
 ) -> Option<BitrotReaderSetup> {
+    rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::UnsupportedPlatform, None);
     None
 }
 
@@ -1476,11 +1493,19 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_with_prefere
 ) -> BitrotReaderSetup {
     let strategy = get_bitrot_reader_setup_strategy(mode, prefer_data_blocks_first);
 
-    if use_mmap_read
-        && !matches!(
-            crate::set_disk::get_object_read_policy(),
-            crate::set_disk::GetObjectReadPolicy::CopySource
-        )
+    let batch_read_enabled = if !use_mmap_read {
+        rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::MmapDisabled, None);
+        false
+    } else if matches!(
+        crate::set_disk::get_object_read_policy(),
+        crate::set_disk::GetObjectReadPolicy::CopySource
+    ) {
+        rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::CopySource, None);
+        false
+    } else {
+        true
+    };
+    if batch_read_enabled
         && let Some(mut setup) = try_create_bitrot_readers_via_batch_pread(
             files,
             disks,
@@ -7864,6 +7889,228 @@ mod tests {
         (dirs, disks)
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn local_batch_read_decisions_follow_the_actual_reader_setup_entry() {
+        use crate::test_metrics::CapturingRecorder;
+
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("reader setup validation runtime");
+        let recorder = CapturingRecorder::default();
+        let previous = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+        metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(temp_env::async_with_vars(
+                [
+                    (rustfs_config::ENV_OBJECT_MMAP_READ_MAX_LENGTH, Some("4")),
+                    (rustfs_config::ENV_INTERNODE_PREWARM, Some("false")),
+                ],
+                async {
+                    let bucket = "batch-path-attribution";
+                    let object = "object";
+                    let (_dirs, disks) = call_counter_local_disks(bucket, 2).await;
+                    let data_dir = Uuid::new_v4();
+                    let mut files = vec![FileInfo::new(object, 1, 1); 2];
+                    for (file, disk) in files.iter_mut().zip(&disks) {
+                        file.data_dir = Some(data_dir);
+                        disk.as_ref()
+                            .expect("local disk")
+                            .write_all(bucket, &format!("{object}/{data_dir}/part.1"), Bytes::from_static(b"data"))
+                            .await
+                            .expect("write batch fixture");
+                    }
+                    assert_eq!(object_mmap_read_max_length(), 4, "cap must be controlled by this test");
+                    for length in [3, 4] {
+                        let setup = create_bitrot_readers_until_quorum_with_preference(
+                            &files,
+                            &disks,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            length,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                            true,
+                            1,
+                            1,
+                            BitrotReaderSetupMode::ReadQuorum,
+                            false,
+                            None,
+                            None,
+                        )
+                        .await;
+                        assert_eq!(setup.readers.iter().flatten().count(), 2);
+                    }
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &disks,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            5,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none(),
+                        "a read above the adjusted-length cap must fall back"
+                    );
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &disks,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::HighwayHash256S,
+                            false,
+                        )
+                        .await
+                        .is_none(),
+                        "checksum overhead must be included in the read cap"
+                    );
+                    let mut inline_files = files.clone();
+                    inline_files[0].data = Some(Bytes::from_static(b"data"));
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &inline_files,
+                            &disks,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none()
+                    );
+                    let mut endpoint = Endpoint::try_from("http://127.0.0.1:1/disk").expect("remote placeholder endpoint");
+                    endpoint.is_local = false;
+                    let remote = new_disk(
+                        &endpoint,
+                        &DiskOption {
+                            cleanup: false,
+                            health_check: false,
+                        },
+                    )
+                    .await
+                    .expect("remote placeholder must initialize without a connection");
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &[Some(remote), None],
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none(),
+                        "any remote shard must keep the setup out of the local batch path"
+                    );
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &disks,
+                            bucket,
+                            "../outside",
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none()
+                    );
+                    let missing = vec![None, None];
+                    assert!(
+                        try_create_bitrot_readers_via_batch_pread(
+                            &files,
+                            &missing,
+                            bucket,
+                            object,
+                            1,
+                            0,
+                            4,
+                            4,
+                            HashAlgorithm::None,
+                            false,
+                        )
+                        .await
+                        .is_none()
+                    );
+                    for (use_mmap, policy) in [
+                        (false, crate::set_disk::GetObjectReadPolicy::Default),
+                        (true, crate::set_disk::GetObjectReadPolicy::CopySource),
+                    ] {
+                        crate::set_disk::with_get_object_read_policy(policy, async {
+                            create_bitrot_readers_until_quorum_with_preference(
+                                &files,
+                                &missing,
+                                bucket,
+                                object,
+                                1,
+                                0,
+                                4,
+                                4,
+                                HashAlgorithm::None,
+                                false,
+                                use_mmap,
+                                1,
+                                1,
+                                BitrotReaderSetupMode::ReadQuorum,
+                                false,
+                                None,
+                                None,
+                            )
+                            .await;
+                        })
+                        .await;
+                    }
+                },
+            ));
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous);
+        let metric = "rustfs_io_get_object_local_batch_decisions_total";
+        for (decision, expected) in [
+            ("selected", 2),
+            ("read_cap_exceeded", 2),
+            ("inline_data", 1),
+            ("invalid_path", 1),
+            ("non_local_shard", 1),
+            ("no_local_shards", 1),
+            ("mmap_disabled", 1),
+            ("copy_source", 1),
+        ] {
+            assert_eq!(recorder.counter_value(metric, &[("decision", decision)]), expected, "{decision}");
+        }
+        assert_eq!(
+            recorder.counter_value("rustfs_io_get_object_local_batch_shards_total", &[("outcome", "success")]),
+            4
+        );
+        assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 14);
+        assert_eq!(recorder.histogram_sample_count("rustfs_io_get_object_local_batch_prepare_seconds"), 8);
+    }
+
     async fn reopen_local_disk(dir: &TempDir) -> DiskStore {
         let endpoint =
             Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");

crates/ecstore/src/set_disk/mod.rs

@@ -1024,6 +1024,19 @@ impl GetObjectFileInfo {
         self.as_parts().0
     }
 
+    /// Marks an exact-version selection as the object's latest version after
+    /// the versions above it were skipped as dangling.
+    fn mark_selected_as_latest(&mut self) {
+        if let Some(OwnedGetObjectFileInfo {
+            payload: OwnedGetObjectFileInfoPayload::Direct { fi, .. },
+            ..
+        }) = self.owned.as_mut()
+        {
+            fi.is_latest = true;
+            fi.successor_mod_time = None;
+        }
+    }
+
     fn parts_metadata(&self) -> &[FileInfo] {
         self.as_parts().1
     }
@@ -6841,6 +6854,17 @@ impl SetDisks {
         Ok(())
     }
 
+    /// Whether any drive holds `version_id` of the key, including a copy on
+    /// fewer drives than read quorum that object reads report as absent.
+    pub(crate) async fn holds_version_on_any_drive(&self, bucket: &str, object: &str, version_id: Uuid) -> bool {
+        let disks = self.disk_inventory().await;
+        let (copies, _) = Self::read_all_raw_file_info(&disks, bucket, object, false).await;
+        copies
+            .iter()
+            .flatten()
+            .any(|copy| FileMeta::load(&copy.buf).is_ok_and(|meta| meta.find_version(Some(version_id)).is_ok()))
+    }
+
     /// Publish an internal tier free-version record without changing its
     /// delete-marker shape or remote-tier identity. The caller holds the
     /// source and target object locks; a write quorum is required before the

crates/ecstore/src/set_disk/ops/object.rs

@@ -8839,7 +8839,7 @@ impl SetDisks {
         // quorum, failing write quorum on update_object_meta (backlog#872).
         let read_opts = opts.as_read_opts();
         let (mut fi, _, disks) = self
-            .get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
+            .get_object_fileinfo_for_metadata_update(bucket, object, &read_opts)
             .await?
             .into_owned();
 
@@ -9294,36 +9294,56 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             }
         };
 
-        let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) {
-            Ok((r, w)) => (
-                usize::try_from(r)
-                    .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![src_bucket, src_object]))?,
-                usize::try_from(w)
-                    .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![src_bucket, src_object]))?,
-            ),
+        let src_version_id = src_opts.version_id.as_deref().unwrap_or_default();
+        let quorum = Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count);
+        let selection = match &quorum {
+            Ok((read_quorum, write_quorum)) => {
+                let read_quorum = usize::try_from(*read_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![src_bucket, src_object]))?;
+                let write_quorum = usize::try_from(*write_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![src_bucket, src_object]))?;
+                Self::select_fileinfo_for_read(src_bucket, &disks, &metas, &errs, src_version_id, read_quorum, write_quorum)
+                    .map(|(online_disks, fi)| (online_disks, fi, write_quorum))
+            }
+            Err(err) => Err(err.clone()),
+        };
+        let (online_disks, mut fi, write_quorum) = match selection {
+            Ok(selection) => selection,
             Err(mut err) => {
-                if err == DiskError::ErasureReadQuorum
-                    && !src_bucket.starts_with(RUSTFS_META_BUCKET)
-                    && self
-                        .delete_if_dangling(src_bucket, src_object, &metas, &errs, &HashMap::new(), src_opts.clone())
-                        .await
-                        .is_ok()
+                // The merged latest version can be one a failed PUT left below
+                // quorum; resolve it like a latest GET before the dangling
+                // cleanup below can report a key with a committed version absent.
+                if let Some(snapshot) = self
+                    .select_latest_past_dangling_versions(src_bucket, src_object, src_opts, true, &disks, &err)
+                    .await
                 {
-                    if src_opts.version_id.is_some() {
-                        err = DiskError::FileVersionNotFound
-                    } else {
-                        err = DiskError::FileNotFound
+                    // A version-only copy writes the served version's own
+                    // copies, so they and its write quorum replace the latest
+                    // read's.
+                    let (fi, parts_metadata, online_disks) = snapshot.into_owned();
+                    metas = parts_metadata;
+                    let write_quorum = fi.write_quorum(self.default_write_quorum());
+                    (online_disks, fi, write_quorum)
+                } else {
+                    if quorum.is_err()
+                        && err == DiskError::ErasureReadQuorum
+                        && !src_bucket.starts_with(RUSTFS_META_BUCKET)
+                        && self
+                            .delete_if_dangling(src_bucket, src_object, &metas, &errs, &HashMap::new(), src_opts.clone())
+                            .await
+                            .is_ok()
+                    {
+                        if src_opts.version_id.is_some() {
+                            err = DiskError::FileVersionNotFound
+                        } else {
+                            err = DiskError::FileNotFound
+                        }
                     }
+                    return Err(to_object_err(err.into(), vec![src_bucket, src_object]));
                 }
-                return Err(to_object_err(err.into(), vec![src_bucket, src_object]));
             }
         };
 
-        let src_version_id = src_opts.version_id.as_deref().unwrap_or_default();
-        let (online_disks, mut fi) =
-            Self::select_fileinfo_for_read(src_bucket, &disks, &metas, &errs, src_version_id, read_quorum, write_quorum)
-                .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
-
         if fi.deleted {
             if src_opts.version_id.is_none() {
                 return Err(to_object_err(Error::FileNotFound, vec![src_bucket, src_object]));
@@ -9461,6 +9481,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 }
             }
 
+            // A version-only copy writes and then reverts on a quorum miss;
+            // refuse it before writing when the served version's disks cannot
+            // reach its write quorum even if every write succeeds.
+            let reachable = online_disks
+                .iter()
+                .map(|disk| disk.is_none().then_some(DiskError::DiskNotFound))
+                .collect::<Vec<_>>();
+            if let Some(err) = reduce_write_quorum_errs(&reachable, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+                return Err(to_object_err(err.into(), vec![src_bucket, src_object]));
+            }
             Self::write_unique_file_info(&online_disks, "", src_bucket, src_object, &metas, write_quorum)
                 .await
                 .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
@@ -11147,45 +11177,63 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             }
         };
 
-        let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) {
-            Ok((read_quorum, write_quorum)) => (read_quorum, write_quorum),
+        let version_id = opts.version_id.as_deref().unwrap_or_default();
+        let quorum = Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count);
+        let selection = match &quorum {
+            Ok((read_quorum, write_quorum)) => {
+                let read_quorum = usize::try_from(*read_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
+                let write_quorum = usize::try_from(*write_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
+                Self::select_fileinfo_for_read(bucket, &disks, &metas, &errs, version_id, read_quorum, write_quorum)
+            }
+            Err(err) => Err(err.clone()),
+        };
+        let (online_disks, mut fi) = match selection {
+            Ok(selection) => selection,
             Err(mut err) => {
-                // A replication terminal CAS is a validation/write-back
-                // attempt, never a heal operation. In particular,
-                // ValidateOnly must remain zero-write, and a worker whose
-                // outer replication lease is already lost must not delete the
-                // surviving xl.meta copies before the namespace fence below
-                // can reject it. Leave dangling cleanup to heal/scanner.
-                if opts.replication_status_writeback.is_some() {
+                // The merged latest version can be one a failed PUT left below
+                // quorum; resolve it like a latest GET before the dangling
+                // cleanup below can report a key with a committed version absent.
+                // A replication terminal CAS stays side-effect free: the walk
+                // queues heal for the versions it skips, so it does not run.
+                let resolved = if opts.replication_status_writeback.is_none() {
+                    self.select_latest_past_dangling_versions(bucket, object, opts, false, &disks, &err)
+                        .await
+                } else {
+                    None
+                };
+                if let Some(snapshot) = resolved {
+                    let (fi, _, online_disks) = snapshot.into_owned();
+                    (online_disks, fi)
+                } else {
+                    // A replication terminal CAS is a validation/write-back
+                    // attempt, never a heal operation. In particular,
+                    // ValidateOnly must remain zero-write, and a worker whose
+                    // outer replication lease is already lost must not delete
+                    // the surviving xl.meta copies before the namespace fence
+                    // below can reject it. Leave dangling cleanup to
+                    // heal/scanner.
+                    if quorum.is_err()
+                        && opts.replication_status_writeback.is_none()
+                        && err == DiskError::ErasureReadQuorum
+                        && !bucket.starts_with(RUSTFS_META_BUCKET)
+                        && self
+                            .delete_if_dangling(bucket, object, &metas, &errs, &HashMap::new(), opts.clone())
+                            .await
+                            .is_ok()
+                    {
+                        if opts.version_id.is_some() {
+                            err = DiskError::FileVersionNotFound
+                        } else {
+                            err = DiskError::FileNotFound
+                        }
+                    }
                     return Err(to_object_err(err.into(), vec![bucket, object]));
                 }
-                if err == DiskError::ErasureReadQuorum
-                    && !bucket.starts_with(RUSTFS_META_BUCKET)
-                    && self
-                        .delete_if_dangling(bucket, object, &metas, &errs, &HashMap::new(), opts.clone())
-                        .await
-                        .is_ok()
-                {
-                    if opts.version_id.is_some() {
-                        err = DiskError::FileVersionNotFound
-                    } else {
-                        err = DiskError::FileNotFound
-                    }
-                }
-                return Err(to_object_err(err.into(), vec![bucket, object]));
             }
         };
 
-        let read_quorum =
-            usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
-        let write_quorum = usize::try_from(write_quorum)
-            .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
-
-        let version_id = opts.version_id.as_deref().unwrap_or_default();
-        let (online_disks, mut fi) =
-            Self::select_fileinfo_for_read(bucket, &disks, &metas, &errs, version_id, read_quorum, write_quorum)
-                .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
-
         if fi.deleted {
             return Err(to_object_err(Error::MethodNotAllowed, vec![bucket, object]));
         }

crates/ecstore/src/set_disk/read.rs

@@ -264,6 +264,11 @@ enum MetadataReadPurpose {
     GetObjectReader,
 }
 
+/// Upper bound on exact-version reads a latest read spends looking past
+/// dangling versions before it keeps the read-quorum error; each read is one
+/// metadata fan-out.
+const MAX_DANGLING_LATEST_VERSIONS: usize = 8;
+
 /// How a metadata read-quorum failure is reported when the replies prove the
 /// object or version dangling. Only the object read entry points
 /// (`get_object_info`, `get_object_reader` and prepared GET metadata) report
@@ -273,6 +278,10 @@ enum MetadataReadPurpose {
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum DanglingRead {
     KeepQuorumError,
+    /// Metadata updates of the latest version keep the quorum error but, like
+    /// object reads, look past newer versions proven dangling
+    /// (`read_latest_past_dangling_versions`).
+    SkipDanglingVersions,
     ReportNotFound,
 }
 
@@ -724,9 +733,10 @@ impl SetDisks {
     }
 
     /// Like `get_object_fileinfo`, but `allow_early_stop=false` forces the full
-    /// quorum fanout. Read-before-write callers (object tagging) must use this:
-    /// the returned online-disk set is the write target, and the early-stop
-    /// subset would fail write quorum (backlog#872 regression fix).
+    /// quorum fanout. Read-before-write callers must use this or
+    /// `get_object_fileinfo_for_metadata_update`: the returned online-disk set
+    /// is the write target, and the early-stop subset would fail write quorum
+    /// (backlog#872 regression fix).
     #[tracing::instrument(level = "debug", skip(self))]
     pub(in crate::set_disk) async fn get_object_fileinfo_gated(
         &self,
@@ -748,6 +758,64 @@ impl SetDisks {
         .await
     }
 
+    /// The full-fanout read behind a metadata update such as object tagging.
+    /// A latest read whose newest versions never committed resolves to the
+    /// version a latest GET serves, while a dangling key keeps the quorum
+    /// error. The returned online disks are the write target.
+    #[tracing::instrument(level = "debug", skip(self))]
+    pub(in crate::set_disk) async fn get_object_fileinfo_for_metadata_update(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+    ) -> Result<GetObjectFileInfo> {
+        self.get_object_fileinfo_gated_inner(
+            bucket,
+            object,
+            opts,
+            false,
+            false,
+            MetadataReadPurpose::FullLayout,
+            DanglingRead::SkipDanglingVersions,
+        )
+        .await
+    }
+
+    /// Recovers a failed latest selection in read-before-write callers that
+    /// read and select metadata themselves (`put_object_metadata` and the
+    /// CopyObject source) the way a latest GET does. `None` keeps the caller's
+    /// error: a failure other than read quorum, an explicit version, the
+    /// system bucket's commit-proof rule, or a walk that finds no readable
+    /// version or cannot read every drive. The selected metadata keeps its
+    /// part checksums because a version-only copy carries them into the new
+    /// version.
+    pub(in crate::set_disk) async fn select_latest_past_dangling_versions(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+        read_data: bool,
+        disks: &[Option<DiskStore>],
+        err: &DiskError,
+    ) -> Option<GetObjectFileInfo> {
+        if *err != DiskError::ErasureReadQuorum || opts.version_id.is_some() || crate::bucket::utils::is_meta_bucketname(bucket) {
+            return None;
+        }
+        let opts = ObjectOptions {
+            include_part_checksums: true,
+            ..opts.clone()
+        };
+        Box::pin(self.read_latest_past_dangling_versions(
+            bucket,
+            object,
+            &opts,
+            read_data,
+            MetadataReadPurpose::FullLayout,
+            disks,
+        ))
+        .await
+    }
+
     #[allow(clippy::too_many_arguments)]
     async fn get_object_fileinfo_gated_inner(
         &self,
@@ -971,6 +1039,18 @@ impl SetDisks {
             {
                 Ok(selection) => selection,
                 Err(err) => {
+                    if err == DiskError::ErasureReadQuorum
+                        && vid.is_empty()
+                        && dangling_read != DanglingRead::KeepQuorumError
+                        && metadata_fanout_complete
+                        && !crate::bucket::utils::is_meta_bucketname(bucket)
+                        && let Some(snapshot) =
+                            Box::pin(self.read_latest_past_dangling_versions(bucket, object, opts, read_data, purpose, &disks))
+                                .await
+                    {
+                        rustfs_io_metrics::record_get_object_metadata_quorum_result(metadata_metrics_path, "reached");
+                        return Ok(snapshot);
+                    }
                     let err = StorageError::from(err);
                     rustfs_io_metrics::record_get_object_metadata_quorum_result(
                         metadata_metrics_path,
@@ -1051,6 +1131,168 @@ impl SetDisks {
         Ok(snapshot)
     }
 
+    /// A latest read whose split cannot reach read quorum may still have a
+    /// readable older version under newer versions that never committed;
+    /// MinIO resolves this in `pickLatestQuorumFilesInfo`. With every drive
+    /// answering, try versions newest first through exact-version reads and
+    /// serve the readable one with the newest selected modification time. A
+    /// version whose exact read proves it dangling (the NoSuchVersion an
+    /// explicit GET reports) is skipped and queued for heal, which owns its
+    /// cleanup; any other outcome keeps the read-quorum error.
+    async fn read_latest_past_dangling_versions(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+        read_data: bool,
+        purpose: MetadataReadPurpose,
+        disks: &[Option<DiskStore>],
+    ) -> Option<GetObjectFileInfo> {
+        let disk_object = rustfs_utils::path::encode_dir_object(object);
+        let (raw, errs) = Self::read_all_raw_file_info(disks, bucket, &disk_object, false).await;
+        if errs
+            .iter()
+            .any(|err| err.as_ref().is_some_and(|err| !matches!(err, DiskError::FileNotFound)))
+        {
+            return None;
+        }
+        // Stacks are newest first, so only each drive's first few headers can
+        // matter. Keep a version's newest modification time across drives: a
+        // rewritten null version can carry different times on different drives.
+        let mut newest = std::collections::HashMap::<Option<uuid::Uuid>, Option<time::OffsetDateTime>>::new();
+        let mut truncated = false;
+        let mut metadata = Vec::with_capacity(raw.len());
+        for raw in raw.into_iter().flatten() {
+            let mut meta = rustfs_filemeta::FileMeta::load(&raw.buf).ok()?;
+            let mut versions = meta
+                .versions
+                .iter()
+                .enumerate()
+                .filter(|(_, version)| !version.header.free_version());
+            let mut retained_end = 0;
+            for (index, version) in versions.by_ref().take(MAX_DANGLING_LATEST_VERSIONS) {
+                let mod_time = newest.entry(version.header.version_id).or_insert(version.header.mod_time);
+                *mod_time = (*mod_time).max(version.header.mod_time);
+                retained_end = index + 1;
+            }
+            truncated |= versions.next().is_some();
+            meta.versions.truncate(retained_end);
+            metadata.push(meta);
+        }
+        let mut candidates = newest.into_iter().collect::<Vec<_>>();
+        candidates.sort_by(|left, right| right.1.cmp(&left.1).then_with(|| left.0.cmp(&right.0)));
+
+        // Use the stored header for equal-time ordering: re-encoding a selected
+        // FileInfo can change legacy signatures or omit part-checksum metadata.
+        let selected_header = |snapshot: &GetObjectFileInfo| {
+            let mut selected = snapshot.fi().clone();
+            selected.data = None;
+            let version_id = selected.version_id.unwrap_or_else(uuid::Uuid::nil).to_string();
+            let include_part_checksums =
+                opts.include_part_checksums || opts.part_number.is_some() || opts.data_movement || opts.raw_data_movement_read;
+            metadata.iter().find_map(|meta| {
+                let version = meta
+                    .versions
+                    .iter()
+                    .filter(|version| !version.header.free_version())
+                    .take(MAX_DANGLING_LATEST_VERSIONS)
+                    .find(|version| {
+                        version.header.version_id.unwrap_or_else(uuid::Uuid::nil)
+                            == selected.version_id.unwrap_or_else(uuid::Uuid::nil)
+                            && version.header.mod_time == selected.mod_time
+                    })?;
+                let mut original = if include_part_checksums {
+                    meta.into_fileinfo(&selected.volume, &selected.name, &version_id, false, opts.incl_free_versions, true)
+                } else {
+                    meta.into_fileinfo_without_part_checksums(
+                        &selected.volume,
+                        &selected.name,
+                        &version_id,
+                        false,
+                        opts.incl_free_versions,
+                    )
+                }
+                .ok()?;
+                if !include_part_checksums {
+                    rustfs_utils::http::remove_str(&mut original.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
+                }
+                original.data = None;
+                original.is_latest = selected.is_latest;
+                original.successor_mod_time = selected.successor_mod_time;
+                original.num_versions = selected.num_versions;
+                (original == selected).then_some(version.header.clone())
+            })
+        };
+
+        let mut best: Option<(Option<time::OffsetDateTime>, GetObjectFileInfo)> = None;
+        let mut skipped = Vec::new();
+        let mut settled = false;
+        for (reads, (version_id, newest_mod_time)) in candidates.into_iter().enumerate() {
+            if best.as_ref().is_some_and(|(selected, _)| newest_mod_time < *selected) {
+                settled = true;
+                break;
+            }
+            if reads == MAX_DANGLING_LATEST_VERSIONS {
+                return None;
+            }
+            // The nil UUID addresses the null version slot at this layer.
+            let version_id = version_id.unwrap_or_else(uuid::Uuid::nil).to_string();
+            let exact = ObjectOptions {
+                version_id: Some(version_id.clone()),
+                ..opts.clone()
+            };
+            match Box::pin(self.get_object_fileinfo_gated_inner(
+                bucket,
+                object,
+                &exact,
+                read_data,
+                false,
+                purpose,
+                DanglingRead::ReportNotFound,
+            ))
+            .await
+            {
+                Ok(snapshot) => {
+                    let selected = snapshot.fi().mod_time;
+                    let replace = match best.as_ref() {
+                        Some((current, previous)) if selected == *current => {
+                            selected_header(&snapshot)?.sorts_before(&selected_header(previous)?)
+                        }
+                        Some((current, _)) => selected > *current,
+                        None => true,
+                    };
+                    if replace {
+                        best = Some((selected, snapshot));
+                    }
+                }
+                Err(StorageError::VersionNotFound(..)) => skipped.push(version_id),
+                Err(_) => return None,
+            }
+        }
+        // Headers past each drive's first few were not compared; only a walk
+        // that proved no remaining candidate can be newer may stop early.
+        if !settled && truncated {
+            return None;
+        }
+        let (_, mut snapshot) = best?;
+        snapshot.mark_selected_as_latest();
+        if !opts.suppress_read_repair {
+            for version_id in &skipped {
+                submit_read_repair_heal(
+                    bucket,
+                    object,
+                    Some(version_id),
+                    self.pool_index,
+                    self.set_index,
+                    None,
+                    "dangling_latest_version",
+                )
+                .await;
+            }
+        }
+        Some(snapshot)
+    }
+
     #[hotpath::measure(impl_type = "SetDisks")]
     pub(super) async fn get_object_info_fileinfo_and_quorum(
         &self,
@@ -9751,6 +9993,7 @@ mod dangling_read_tests {
     use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
     use crate::storage_api_contracts::object::ObjectOperations;
     use http::HeaderMap;
+    use std::collections::HashMap;
     use std::path::PathBuf;
     use tempfile::TempDir;
     use tokio::io::AsyncReadExt;
@@ -9761,9 +10004,13 @@ mod dangling_read_tests {
     const PARITY: usize = 2;
 
     async fn dangling_test_set(bucket: &str) -> (Vec<TempDir>, Arc<SetDisks>) {
+        dangling_test_set_with(bucket, DRIVES, PARITY).await
+    }
+
+    async fn dangling_test_set_with(bucket: &str, drives: usize, parity: usize) -> (Vec<TempDir>, Arc<SetDisks>) {
         let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
         ctx.update_erasure_type(crate::layout::endpoints::SetupType::Erasure).await;
-        let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(DRIVES, PARITY, ctx).await;
+        let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(drives, parity, ctx).await;
         set.make_bucket(bucket, &MakeBucketOptions::default())
             .await
             .expect("test bucket should be created on every drive");
@@ -9777,6 +10024,9 @@ mod dangling_read_tests {
             &mut PutObjReader::from_vec(body.to_vec()),
             &ObjectOptions {
                 versioned,
+                // Wait for every drive so tests that edit one drive's metadata
+                // do not race the PUT's background tail writes.
+                write_completion: crate::object_api::WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )
@@ -9796,6 +10046,25 @@ mod dangling_read_tests {
         }
     }
 
+    /// Drops `version_id` from `disks`, modelling drives a failed PUT never
+    /// reached while the older versions below it stay intact.
+    async fn drop_version_from(set: &SetDisks, bucket: &str, object: &str, version_id: Uuid, disks: std::ops::Range<usize>) {
+        let all = set.disks.read().await.clone();
+        for disk in &all[disks] {
+            let disk = disk.as_ref().expect("every drive should be online");
+            let mut remove = FileInfo {
+                volume: bucket.to_string(),
+                name: object.to_string(),
+                version_id: Some(version_id),
+                ..Default::default()
+            };
+            remove.set_skip_tier_free_version();
+            disk.delete_version(bucket, object, remove, false, DeleteOptions::default())
+                .await
+                .expect("dropping the version should model a drive that never received it");
+        }
+    }
+
     fn read_opts(version_id: Option<&str>) -> ObjectOptions {
         ObjectOptions {
             version_id: version_id.map(str::to_owned),
@@ -9955,12 +10224,642 @@ mod dangling_read_tests {
         head(&set, bucket, object, Some(&v1))
             .await
             .expect("HEAD of the committed version must succeed");
-        // The latest read sees two generations (version two on three drives,
-        // version one on the rest), which is not proof of absence.
-        let Err(err) = head(&set, bucket, object, None).await else {
-            panic!("a split latest version must not resolve without quorum");
+        // The latest read sees version two on three drives and version one on
+        // the rest. Version two's exact read proves it dangling, so the latest
+        // read serves version one, the newest version an exact read can serve.
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("the latest read must serve the newest readable version");
+        assert_eq!(latest.version_id.map(|id| id.to_string()).as_deref(), Some(v1.as_str()));
+        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"version one");
+    }
+
+    #[tokio::test]
+    async fn latest_read_skips_dangling_versions_above_a_committed_version() {
+        // EC 12+4: a failed PUT that left version two on 5..=11 drives keeps
+        // version one, present in every stack, below its read quorum of twelve
+        // when only the top of each stack is read. Up to seven copies the
+        // listing quorum already shows version one; from eight it lists the
+        // dangling version two.
+        let bucket = "dangling-latest-12-4";
+        let (_dirs, set) = dangling_test_set_with(bucket, 16, 4).await;
+        for copies in 5..=11 {
+            let object = format!("torn/latest-{copies:02}");
+            let v1 = put(&set, bucket, &object, b"committed version", true)
+                .await
+                .version_id
+                .expect("versioned PUT should assign a version id");
+            let v2 = put(&set, bucket, &object, b"failed overwrite", true)
+                .await
+                .version_id
+                .expect("versioned PUT should assign a version id");
+            drop_version_from(&set, bucket, &object, v2, copies..16).await;
+
+            let context = format!("version two on {copies} of 16 drives");
+            let latest = head(&set, bucket, &object, None)
+                .await
+                .unwrap_or_else(|err| panic!("{context}: HEAD must serve version one: {err:?}"));
+            assert_eq!(latest.version_id, Some(v1), "{context}");
+            assert!(latest.is_latest, "{context}: the served version is the latest one");
+            assert_eq!(
+                get(&set, bucket, &object, None)
+                    .await
+                    .unwrap_or_else(|err| panic!("{context}: GET must serve version one: {err:?}")),
+                b"committed version",
+                "{context}"
+            );
+            let Err(err) = head(&set, bucket, &object, Some(&v2.to_string())).await else {
+                panic!("{context}: the dangling version must not be served");
+            };
+            assert!(matches!(err, StorageError::VersionNotFound(..)), "{context}: {err:?}");
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_read_keeps_quorum_error_when_the_newest_version_is_uncertain() {
+        // Version two is on every drive, but a torn metadata update left two
+        // identities at one modification time, three copies each. Its exact
+        // read fails with a quorum error rather than NoSuchVersion, so the
+        // walk must not fall through to version one.
+        let bucket = "dangling-latest-uncertain";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        put(&set, bucket, object, b"version one", true).await;
+        let v2 = put(&set, bucket, object, b"version two", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id")
+            .to_string();
+        let disks = set.disks.read().await.clone();
+        for disk in disks.iter().take(3) {
+            let disk = disk.as_ref().expect("every drive should be online");
+            let mut fi = disk
+                .read_version("", bucket, object, &v2, &ReadOptions::default())
+                .await
+                .expect("version two should be on every drive");
+            fi.metadata.insert("x-amz-meta-torn".to_string(), "half-applied".to_string());
+            disk.update_metadata(bucket, object, fi, &crate::disk::UpdateMetadataOpts::default())
+                .await
+                .expect("a half-applied metadata update should be stored");
+        }
+
+        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "torn metadata update on version two").await;
+    }
+
+    #[tokio::test]
+    async fn latest_read_serves_a_committed_null_version_below_a_dangling_version() {
+        // A null version written while versioning was suspended sits between
+        // version one and a dangling version three. The walk reaches the null
+        // slot through the nil UUID and serves it, not version one.
+        let bucket = "dangling-latest-null";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let v1 = put(&set, bucket, object, b"version one", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        set.put_object(
+            bucket,
+            object,
+            &mut PutObjReader::from_vec(b"null version".to_vec()),
+            &ObjectOptions {
+                version_suspended: true,
+                write_completion: crate::object_api::WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("a suspended-versioning PUT should commit the null version");
+        let v3 = put(&set, bucket, object, b"version three", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        drop_version_from(&set, bucket, object, v3, 3..DRIVES).await;
+
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("the committed null version must be served");
+        assert_ne!(latest.version_id, Some(v1));
+        assert_ne!(latest.version_id, Some(v3));
+        assert!(latest.is_latest);
+        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"null version");
+    }
+
+    #[tokio::test]
+    async fn latest_read_orders_versions_by_their_newest_modification_time() {
+        // Drive 0 missed the suspended-versioning rewrite of the null version,
+        // so its null copy predates version one while every other drive holds
+        // the committed rewrite. Ordering by the first drive's time would try
+        // version one before the null version and serve it.
+        let bucket = "dangling-latest-order";
+        let object = "torn/latest";
+        let (dirs, set) = dangling_test_set(bucket).await;
+        let suspended = ObjectOptions {
+            version_suspended: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
+            ..Default::default()
         };
-        assert!(is_read_quorum_error(&err), "a split latest version keeps the quorum error: {err:?}");
+        set.put_object(bucket, object, &mut PutObjReader::from_vec(b"stale null".to_vec()), &suspended)
+            .await
+            .expect("the first null version should commit");
+        let v1 = put(&set, bucket, object, b"version one", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        let drive_zero = object_dir(&dirs, 0, bucket, object).join(STORAGE_FORMAT_FILE);
+        let stale = tokio::fs::read(&drive_zero)
+            .await
+            .expect("drive zero metadata should be readable");
+        set.put_object(bucket, object, &mut PutObjReader::from_vec(b"committed null".to_vec()), &suspended)
+            .await
+            .expect("the null rewrite should commit");
+        tokio::fs::write(&drive_zero, stale)
+            .await
+            .expect("drive zero should keep the pre-rewrite stack");
+        let v5 = put(&set, bucket, object, b"version five", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        drop_version_from(&set, bucket, object, v5, 3..DRIVES).await;
+
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("the committed null rewrite must be served");
+        assert_ne!(latest.version_id, Some(v1));
+        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"committed null");
+    }
+
+    #[tokio::test]
+    async fn latest_read_preserves_object_before_equal_time_marker() {
+        let bucket = "dangling-latest-equal-time-marker";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let same_time = time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixed source mtime");
+        for suspended in [false, true] {
+            let object = if suspended {
+                "torn/equal-time-null"
+            } else {
+                "torn/equal-time-versioned"
+            };
+            let object_id = if suspended { Uuid::nil() } else { Uuid::from_u128(2) };
+            set.put_object(
+                bucket,
+                object,
+                &mut PutObjReader::from_vec(b"committed object".to_vec()),
+                &ObjectOptions {
+                    versioned: !suspended,
+                    version_suspended: suspended,
+                    replication_request: true,
+                    version_id: Some(object_id.to_string()),
+                    mod_time: Some(same_time),
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("commit object version");
+            set.delete_object(
+                bucket,
+                object,
+                ObjectOptions {
+                    versioned: true,
+                    version_id: Some(Uuid::from_u128(1).to_string()),
+                    mod_time: Some(same_time),
+                    data_movement: true,
+                    delete_marker: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("recreate source delete marker during data movement");
+            let expected = head(&set, bucket, object, None)
+                .await
+                .expect("canonical latest before dangling write");
+            assert_eq!(expected.version_id, (!suspended).then_some(object_id));
+            assert!(!expected.delete_marker, "canonical header order prefers object to equal-time marker");
+            assert_eq!(get(&set, bucket, object, None).await.expect("canonical GET"), b"committed object");
+            let dangling_id = Uuid::from_u128(3);
+            set.put_object(
+                bucket,
+                object,
+                &mut PutObjReader::from_vec(b"failed newer version".to_vec()),
+                &ObjectOptions {
+                    versioned: true,
+                    replication_request: true,
+                    version_id: Some(dangling_id.to_string()),
+                    mod_time: Some(same_time + time::Duration::seconds(1)),
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("commit newer version before modeling failed rollback");
+            drop_version_from(&set, bucket, object, dangling_id, 3..DRIVES).await;
+            let actual = head(&set, bucket, object, None)
+                .await
+                .expect("latest HEAD past dangling version");
+            let actual_get = get(&set, bucket, object, None).await;
+            assert_eq!(
+                actual.version_id, expected.version_id,
+                "{object}: unreadable newer version must preserve canonical equal-time latest; GET result: {actual_get:?}"
+            );
+            assert!(
+                !actual.delete_marker,
+                "{object}: fallback must preserve the object before its equal-time marker"
+            );
+            assert_eq!(actual_get.expect("latest GET"), b"committed object");
+
+            let checksum_opts = ObjectOptions {
+                include_part_checksums: true,
+                ..Default::default()
+            };
+            let checksum_head = set
+                .get_object_info(bucket, object, &checksum_opts)
+                .await
+                .expect("latest HEAD with part checksums");
+            assert_eq!(checksum_head.version_id, expected.version_id, "{object}: checksum HEAD version");
+            assert!(!checksum_head.delete_marker, "{object}: checksum HEAD must return the object");
+            let mut checksum_reader = set
+                .get_object_reader(bucket, object, None, HeaderMap::new(), &checksum_opts)
+                .await
+                .expect("latest GET with part checksums");
+            let mut checksum_body = Vec::new();
+            checksum_reader
+                .stream
+                .read_to_end(&mut checksum_body)
+                .await
+                .expect("checksum GET should stream the committed bytes");
+            assert_eq!(checksum_body, b"committed object", "{object}: checksum GET bytes");
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_read_past_dangling_versions_requires_every_drive_to_answer() {
+        let bucket = "dangling-latest-offline";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        put(&set, bucket, object, b"version one", true).await;
+        let v2 = put(&set, bucket, object, b"version two", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        drop_version_from(&set, bucket, object, v2, 3..DRIVES).await;
+        // The offline drive may hold the copy that makes version two readable.
+        let offline = set.disks.write().await[5].take().expect("drive five should start online");
+
+        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "one drive offline").await;
+
+        set.disks.write().await[5] = Some(offline);
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("with every drive answering the dangling version is skipped");
+        assert_ne!(latest.version_id, Some(v2));
+    }
+
+    /// Version one on every drive under a version two that a failed PUT left
+    /// on drives `0..3`, below the read quorum of four.
+    async fn put_committed_below_dangling(set: &SetDisks, bucket: &str, object: &str) -> (Uuid, Uuid) {
+        let v1 = put(set, bucket, object, b"version one", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        let v2 = put(set, bucket, object, b"version two", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        drop_version_from(set, bucket, object, v2, 3..DRIVES).await;
+        (v1, v2)
+    }
+
+    /// Each drive's copy of `version_id`, or `None` where the drive lacks it.
+    async fn version_metadata(
+        set: &SetDisks,
+        bucket: &str,
+        object: &str,
+        version_id: Uuid,
+    ) -> Vec<Option<HashMap<String, String>>> {
+        let disks = set.disks.read().await.clone();
+        let mut copies = Vec::with_capacity(disks.len());
+        for disk in &disks {
+            let disk = disk.as_ref().expect("every drive should be online");
+            copies.push(
+                disk.read_version("", bucket, object, &version_id.to_string(), &ReadOptions::default())
+                    .await
+                    .ok()
+                    .map(|fi| fi.metadata),
+            );
+        }
+        copies
+    }
+
+    #[tokio::test]
+    async fn latest_tagging_skips_dangling_versions_above_a_committed_version() {
+        let bucket = "dangling-latest-tagging";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let (v1, v2) = put_committed_below_dangling(&set, bucket, object).await;
+        let tag_of = |metadata: &Option<HashMap<String, String>>| {
+            metadata
+                .as_ref()
+                .and_then(|metadata| metadata.get(rustfs_utils::http::headers::AMZ_OBJECT_TAGGING).cloned())
+        };
+
+        let tagged = set
+            .put_object_tags(bucket, object, "stage=committed", &ObjectOptions::default())
+            .await
+            .expect("PutObjectTagging must update the version a latest GET serves");
+        assert_eq!(tagged.version_id, Some(v1));
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            assert_eq!(
+                tag_of(metadata).as_deref(),
+                Some("stage=committed"),
+                "drive {disk} must carry the tag on version one"
+            );
+        }
+        for (disk, metadata) in version_metadata(&set, bucket, object, v2).await.iter().enumerate() {
+            assert_eq!(tag_of(metadata), None, "drive {disk} must leave the dangling version untouched");
+        }
+        assert_eq!(*head(&set, bucket, object, None).await.expect("latest HEAD").user_tags, "stage=committed");
+
+        let untagged = set
+            .delete_object_tags(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("DeleteObjectTagging must update the version a latest GET serves");
+        assert_eq!(untagged.version_id, Some(v1));
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            assert_eq!(tag_of(metadata).as_deref(), Some(""), "drive {disk} must clear the tag on version one");
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_metadata_update_skips_dangling_versions_above_a_committed_version() {
+        let bucket = "dangling-latest-metadata";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let (v1, v2) = put_committed_below_dangling(&set, bucket, object).await;
+
+        let updated = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "updated".to_string())])),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a latest metadata update must target the version a latest GET serves");
+        assert_eq!(updated.version_id, Some(v1));
+        assert!(updated.is_latest, "the updated version is the latest one");
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            let metadata = metadata.as_ref().expect("version one is on every drive");
+            assert_eq!(metadata.get("x-amz-meta-probe").map(String::as_str), Some("updated"), "drive {disk}");
+        }
+        for (disk, metadata) in version_metadata(&set, bucket, object, v2).await.iter().enumerate() {
+            assert!(
+                metadata
+                    .as_ref()
+                    .is_none_or(|metadata| !metadata.contains_key("x-amz-meta-probe")),
+                "drive {disk} must leave the dangling version untouched"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_copy_source_skips_dangling_versions_above_a_committed_version() {
+        let bucket = "dangling-latest-copy";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+
+        // A metadata-only self-copy rewrites the latest version in place.
+        let object = "torn/in-place";
+        let (v1, _) = put_committed_below_dangling(&set, bucket, object).await;
+        let mut source = head(&set, bucket, object, None).await.expect("latest HEAD");
+        assert_eq!(source.version_id, Some(v1));
+        source.metadata_only = true;
+        source.user_defined = Arc::new(HashMap::from([("x-amz-meta-copied".to_string(), "in-place".to_string())]));
+        let copied = set
+            .copy_object(
+                bucket,
+                object,
+                bucket,
+                object,
+                &mut source,
+                &ObjectOptions::default(),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("a CopyObject source without a version id must resolve like a latest GET");
+        assert_eq!(copied.version_id, Some(v1));
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            let metadata = metadata.as_ref().expect("version one is on every drive");
+            assert_eq!(metadata.get("x-amz-meta-copied").map(String::as_str), Some("in-place"), "drive {disk}");
+        }
+
+        // A version-only self-copy commits a new version that shares the
+        // served version's data.
+        let object = "torn/new-version";
+        let (_, v2) = put_committed_below_dangling(&set, bucket, object).await;
+        let v3 = Uuid::new_v4();
+        let mut source = head(&set, bucket, object, None).await.expect("latest HEAD");
+        source.version_only = true;
+        let copied = set
+            .copy_object(
+                bucket,
+                object,
+                bucket,
+                object,
+                &mut source,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+                &ObjectOptions {
+                    version_id: Some(v3.to_string()),
+                    versioned: true,
+                    object_lock_config_snapshot: Some(Arc::new(crate::object_api::ObjectLockConfigSnapshot::new(
+                        crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+                    ))),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a version-only CopyObject source without a version id must resolve like a latest GET");
+        assert_eq!(copied.version_id, Some(v3));
+        let latest = head(&set, bucket, object, None).await.expect("latest HEAD");
+        assert_eq!(latest.version_id, Some(v3));
+        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"version one");
+        assert!(
+            version_metadata(&set, bucket, object, v2).await[..3]
+                .iter()
+                .all(Option::is_some),
+            "the copy must not remove the dangling version; heal owns its cleanup"
+        );
+    }
+
+    #[tokio::test]
+    async fn latest_metadata_update_past_dangling_versions_keeps_write_quorum() {
+        // EC 3+3: read quorum three, write quorum four. Version one reached
+        // three drives, so it is readable but below write quorum; versions two
+        // and three are dangling on two drives each. A latest read serves
+        // version one, and a metadata update of it is refused before writing.
+        let bucket = "dangling-latest-write-quorum";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set_with(bucket, 6, 3).await;
+        let mut versions = Vec::new();
+        for body in [b"version zero".as_slice(), b"version one", b"version two", b"version three"] {
+            versions.push(
+                put(&set, bucket, object, body, true)
+                    .await
+                    .version_id
+                    .expect("versioned PUT should assign a version id"),
+            );
+        }
+        let [_, v1, v2, v3] = versions[..] else {
+            unreachable!("four versions were written");
+        };
+        drop_version_from(&set, bucket, object, v1, 3..6).await;
+        drop_version_from(&set, bucket, object, v2, 2..6).await;
+        drop_version_from(&set, bucket, object, v3, 0..3).await;
+        drop_version_from(&set, bucket, object, v3, 5..6).await;
+
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("version one is readable at read quorum");
+        assert_eq!(latest.version_id, Some(v1));
+
+        let is_write_quorum_error =
+            |err: &StorageError| matches!(err, StorageError::ErasureWriteQuorum | StorageError::InsufficientWriteQuorum(_, _));
+        let Err(err) = set
+            .put_object_tags(bucket, object, "stage=refused", &ObjectOptions::default())
+            .await
+        else {
+            panic!("tagging a version below write quorum must be refused");
+        };
+        assert!(is_write_quorum_error(&err), "tagging keeps the write-quorum error: {err:?}");
+        let Err(err) = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "refused".to_string())])),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a metadata update of a version below write quorum must be refused");
+        };
+        assert!(is_write_quorum_error(&err), "the metadata update keeps the write-quorum error: {err:?}");
+        let v4 = Uuid::new_v4();
+        let mut source = head(&set, bucket, object, None).await.expect("latest HEAD");
+        source.version_only = true;
+        let Err(err) = set
+            .copy_object(
+                bucket,
+                object,
+                bucket,
+                object,
+                &mut source,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+                &ObjectOptions {
+                    version_id: Some(v4.to_string()),
+                    versioned: true,
+                    object_lock_config_snapshot: Some(Arc::new(crate::object_api::ObjectLockConfigSnapshot::new(
+                        crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+                    ))),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a version-only copy of a version below write quorum must be refused");
+        };
+        assert!(is_write_quorum_error(&err), "the version-only copy keeps the write-quorum error: {err:?}");
+        assert!(
+            version_metadata(&set, bucket, object, v4).await.iter().all(Option::is_none),
+            "a refused version-only copy must not write the new version to any drive"
+        );
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            assert!(
+                metadata.as_ref().is_none_or(|metadata| {
+                    !metadata.contains_key(rustfs_utils::http::headers::AMZ_OBJECT_TAGGING)
+                        && !metadata.contains_key("x-amz-meta-probe")
+                }),
+                "drive {disk} must not be written by a refused update"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn replication_status_writeback_never_walks_past_dangling_versions() {
+        // A replication terminal CAS stays side-effect free: it keeps the
+        // read-quorum error instead of resolving the latest version, so it
+        // neither queues heal for the dangling version nor writes version one.
+        let bucket = "dangling-latest-replication-cas";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let (v1, _) = put_committed_below_dangling(&set, bucket, object).await;
+
+        let Err(err) = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "written".to_string())])),
+                    replication_status_writeback: Some(Box::new(crate::object_api::ReplicationStatusWritebackCondition {
+                        mode: crate::object_api::ReplicationStatusWritebackMode::ValidateOnly,
+                        ..Default::default()
+                    })),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a replication terminal CAS must not resolve past a dangling version");
+        };
+        assert!(is_read_quorum_error(&err), "the CAS keeps the read-quorum error: {err:?}");
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            let metadata = metadata.as_ref().expect("version one is on every drive");
+            assert!(!metadata.contains_key("x-amz-meta-probe"), "drive {disk} must not be written");
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_metadata_updates_past_dangling_versions_require_every_drive_to_answer() {
+        let bucket = "dangling-latest-update-offline";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        put_committed_below_dangling(&set, bucket, object).await;
+        let offline = set.disks.write().await[5].take().expect("drive five should start online");
+
+        let Err(err) = set
+            .put_object_tags(bucket, object, "stage=uncertain", &ObjectOptions::default())
+            .await
+        else {
+            panic!("tagging must not resolve past a version an offline drive may commit");
+        };
+        assert!(is_read_quorum_error(&err), "tagging keeps the read-quorum error: {err:?}");
+        let Err(err) = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "uncertain".to_string())])),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a metadata update must not resolve past a version an offline drive may commit");
+        };
+        assert!(is_read_quorum_error(&err), "the metadata update keeps the read-quorum error: {err:?}");
+        set.disks.write().await[5] = Some(offline);
     }
 
     #[tokio::test]

crates/ecstore/src/store/mod.rs

@@ -452,6 +452,8 @@ pub use object::{
 };
 mod peer;
 mod rebalance;
+#[cfg(test)]
+pub(crate) use rebalance::PoolSpaceForTest;
 pub(crate) mod utils;
 
 use peer::init_local_peer;

crates/ecstore/src/store/object.rs

@@ -4291,8 +4291,40 @@ impl ECStore {
                 .select_decommission_capacity_target_pool(owner, expected_data_bytes)
                 .await;
         }
+        let holder = self
+            .data_movement_pool_holding_version(bucket, object, opts.version_id.as_deref(), opts.src_pool_idx)
+            .await;
+        self.get_available_pool_idx_preferring(bucket, object, size, None, holder)
+            .await
+            .ok_or(Error::DiskFull)
+    }
 
-        self.get_available_pool_idx(bucket, object, size).await.ok_or(Error::DiskFull)
+    /// The target pool that already holds a copy of the exact version being
+    /// moved. An attempt that lost write quorum can leave that copy on fewer
+    /// drives than read quorum, which the pool lookup reports as absent; a
+    /// retry that placed the version on another pool would leave it in two.
+    /// Callers still subject the holder to the free-space checks, so a full
+    /// holder cannot pin every retry of a deferred entry.
+    async fn data_movement_pool_holding_version(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: Option<&str>,
+        src_pool_idx: usize,
+    ) -> Option<usize> {
+        let version_id = Uuid::parse_str(version_id?).ok()?;
+        let holders = join_all(self.pools.iter().enumerate().map(|(pool_idx, pool)| async move {
+            // Match the pools the free-space choice can return.
+            if pool_idx == src_pool_idx || self.is_suspended(pool_idx).await || self.is_pool_rebalancing(pool_idx).await {
+                return None;
+            }
+            pool.get_disks_by_key(object)
+                .holds_version_on_any_drive(bucket, object, version_id)
+                .await
+                .then_some(pool_idx)
+        }))
+        .await;
+        holders.into_iter().flatten().next()
     }
 
     async fn find_data_movement_target_info(
@@ -5420,7 +5452,10 @@ impl ECStore {
                         if let Some(owner) = DecommissionCapacityOwner::from_options(&opts) {
                             self.select_decommission_capacity_target_pool(owner, 0).await?
                         } else {
-                            self.get_available_pool_idx_excluding(bucket, object, 0, opts.src_pool_idx)
+                            let holder = self
+                                .data_movement_pool_holding_version(bucket, object, opts.version_id.as_deref(), opts.src_pool_idx)
+                                .await;
+                            self.get_available_pool_idx_preferring(bucket, object, 0, Some(opts.src_pool_idx), holder)
                                 .await
                                 .ok_or(Error::DiskFull)?
                         }

crates/ecstore/src/store/rebalance.rs

@@ -122,6 +122,51 @@ async fn pause_prepared_pool_read_before_refetch(object: &str) {
         state.refetch_release.notified().await;
     }
 }
+/// Capacity that every online drive of one pool reports to free-space target
+/// selection, per store, until the guard drops.
+#[cfg(test)]
+static POOL_SPACE_FOR_TEST: std::sync::Mutex<Vec<(Uuid, usize, crate::disk::DiskInfo)>> = std::sync::Mutex::new(Vec::new());
+
+#[cfg(test)]
+pub(crate) struct PoolSpaceForTest(Uuid);
+
+#[cfg(test)]
+impl PoolSpaceForTest {
+    pub(crate) fn install(store: Uuid, pools: &[(usize, crate::disk::DiskInfo)]) -> Self {
+        POOL_SPACE_FOR_TEST
+            .lock()
+            .expect("pool space override must not be poisoned")
+            .extend(pools.iter().map(|(pool, info)| (store, *pool, info.clone())));
+        Self(store)
+    }
+}
+
+#[cfg(test)]
+impl Drop for PoolSpaceForTest {
+    fn drop(&mut self) {
+        POOL_SPACE_FOR_TEST
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .retain(|(store, _, _)| *store != self.0);
+    }
+}
+
+#[cfg(test)]
+fn pool_space_for_test(
+    store: Uuid,
+    pool: usize,
+    infos: Vec<Option<crate::disk::DiskInfo>>,
+) -> Vec<Option<crate::disk::DiskInfo>> {
+    let overrides = POOL_SPACE_FOR_TEST.lock().expect("pool space override must not be poisoned");
+    let Some((_, _, info)) = overrides
+        .iter()
+        .find(|(override_store, override_pool, _)| *override_store == store && *override_pool == pool)
+    else {
+        return infos;
+    };
+    infos.into_iter().map(|disk| disk.map(|_| info.clone())).collect()
+}
+
 #[cfg(test)]
 use support::resolve_latest_object_info_candidates;
 use support::{
@@ -438,30 +483,7 @@ impl ECStore {
     }
 
     pub(super) async fn get_available_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Option<usize> {
-        // // Return a random one first
-
-        let mut server_pools = self.get_server_pools_available_space(bucket, object, size).await;
-        server_pools.filter_max_used(100 - (100_f64 * DISK_RESERVE_FRACTION) as u64);
-        let total = server_pools.total_available();
-
-        if total == 0 {
-            return None;
-        }
-
-        let mut rng = rand::rng();
-        let random_u64: u64 = rng.random_range(0..total);
-
-        let choose = random_u64 % total;
-        let mut at_total = 0;
-
-        for pool in server_pools.iter() {
-            at_total += pool.available;
-            if at_total > choose && pool.available > 0 {
-                return Some(pool.index);
-            }
-        }
-
-        None
+        self.get_available_pool_idx_preferring(bucket, object, size, None, None).await
     }
 
     pub(super) async fn get_available_pool_idx_excluding(
@@ -470,13 +492,33 @@ impl ECStore {
         object: &str,
         size: i64,
         excluded_pool_idx: usize,
+    ) -> Option<usize> {
+        self.get_available_pool_idx_preferring(bucket, object, size, Some(excluded_pool_idx), None)
+            .await
+    }
+
+    /// Free-space choice among the pools that can take `size` more bytes,
+    /// weighted by available space. `preferred_pool_idx` is returned whenever
+    /// it passes the same checks, so a full or excluded pool is never forced.
+    pub(super) async fn get_available_pool_idx_preferring(
+        &self,
+        bucket: &str,
+        object: &str,
+        size: i64,
+        excluded_pool_idx: Option<usize>,
+        preferred_pool_idx: Option<usize>,
     ) -> Option<usize> {
         let mut server_pools = self.get_server_pools_available_space(bucket, object, size).await;
         server_pools.filter_max_used(100 - (100_f64 * DISK_RESERVE_FRACTION) as u64);
 
-        if let Some(pool) = server_pools.0.get_mut(excluded_pool_idx) {
+        if let Some(pool) = excluded_pool_idx.and_then(|idx| server_pools.0.get_mut(idx)) {
             pool.available = 0;
         }
+        if let Some(pool_idx) = preferred_pool_idx
+            && server_pools.0.get(pool_idx).is_some_and(|pool| pool.available > 0)
+        {
+            return Some(pool_idx);
+        }
 
         let total = server_pools.total_available();
         if total == 0 {
@@ -509,6 +551,8 @@ impl ECStore {
 
             let disks = pool.get_disks_by_key(object).disk_inventory().await;
             let disk_infos = get_disk_infos(&disks).await;
+            #[cfg(test)]
+            let disk_infos = pool_space_for_test(self.id, idx, disk_infos);
 
             (idx, pool.set_count, disk_infos)
         }))

crates/io-metrics/benches/metrics_pipeline.rs

@@ -1,10 +1,67 @@
 use criterion::{Criterion, criterion_group, criterion_main};
+use metrics_util::debugging::DebuggingRecorder;
+use rustfs_io_metrics::{
+    LocalBatchReadDecision, LocalBatchReadMetrics, LocalBatchReadStage, record_local_batch_read_decision,
+    set_get_stage_metrics_enabled,
+};
 use rustfs_io_metrics::{MetricsCollector, PerformanceMetrics, record_get_object_timeout_operation_start};
 use rustfs_io_metrics::{record_s3_op, s3_http_metrics::S3HttpRequestGuard};
 use rustfs_s3_ops::S3Operation;
 use std::hint::black_box;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
+
+fn local_batch_attribution_sample() {
+    let prepare = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now);
+    record_local_batch_read_decision(LocalBatchReadDecision::Selected, prepare.map(|started| started.elapsed()));
+    if let Some(metrics) = LocalBatchReadMetrics::capture() {
+        metrics.record_batch(4);
+        let execution = Instant::now();
+        metrics.record_stage(LocalBatchReadStage::QueueWait, true, execution.elapsed());
+        for _ in 0..4 {
+            for stage in [
+                LocalBatchReadStage::Metadata,
+                LocalBatchReadStage::Open,
+                LocalBatchReadStage::Read,
+            ] {
+                let started = Instant::now();
+                metrics.record_stage(stage, true, started.elapsed());
+            }
+            metrics.record_shard(true, 32 * 1024, 32 * 1024);
+        }
+        metrics.record_stage(LocalBatchReadStage::Execution, true, execution.elapsed());
+    }
+}
+
+fn bench_local_batch_attribution(c: &mut Criterion) {
+    let mut group = c.benchmark_group("local_batch_attribution_4_shards_debug_recorder");
+    for (name, enabled) in [("disabled", false), ("enabled", true)] {
+        set_get_stage_metrics_enabled(enabled);
+        group.bench_function(name, |b| {
+            b.iter_custom(|iterations| {
+                let mut remaining = iterations;
+                let mut elapsed = Duration::ZERO;
+                while remaining > 0 {
+                    // Bound captured histogram storage independently of Criterion's
+                    // iteration count. Recorder setup/teardown is not measured.
+                    let chunk = remaining.min(128);
+                    let recorder = DebuggingRecorder::new();
+                    metrics::with_local_recorder(&recorder, || {
+                        let started = Instant::now();
+                        for _ in 0..chunk {
+                            local_batch_attribution_sample();
+                        }
+                        elapsed += started.elapsed();
+                    });
+                    remaining -= chunk;
+                }
+                elapsed
+            });
+        });
+    }
+    group.finish();
+    set_get_stage_metrics_enabled(false);
+}
 
 fn bench_record_get_object_timeout_operation_start(c: &mut Criterion) {
     c.bench_function("record_get_object_timeout_operation_start", |b| {
@@ -54,6 +111,7 @@ criterion_group!(
     bench_record_get_object_timeout_operation_start,
     bench_s3_http_outcomes,
     bench_update_concurrent_requests,
-    bench_metrics_collector_record_io_operation
+    bench_metrics_collector_record_io_operation,
+    bench_local_batch_attribution
 );
 criterion_main!(benches);

crates/io-metrics/src/lib.rs

@@ -49,6 +49,11 @@
 #[macro_use]
 extern crate metrics;
 
+mod local_batch_read;
+pub use local_batch_read::{
+    LocalBatchReadDecision, LocalBatchReadMetrics, LocalBatchReadStage, record_local_batch_read_decision,
+};
+
 use std::collections::HashMap;
 use std::sync::{
     LazyLock, Mutex,

crates/io-metrics/src/local_batch_read.rs

@@ -0,0 +1,210 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Opt-in attribution for the buffered local shard batch path.
+
+use crate::get_stage_metrics_enabled;
+use metrics::{Counter, Histogram};
+use std::time::Duration;
+
+/// One decision per reader-setup attempt, rather than per S3 request.
+#[derive(Clone, Copy, Debug)]
+pub enum LocalBatchReadDecision {
+    Selected,
+    MmapDisabled,
+    CopySource,
+    ReadCapExceeded,
+    InlineData,
+    NonLocalShard,
+    InvalidPath,
+    NoLocalShards,
+    UnsupportedPlatform,
+}
+
+impl LocalBatchReadDecision {
+    const fn as_str(self) -> &'static str {
+        match self {
+            Self::Selected => "selected",
+            Self::MmapDisabled => "mmap_disabled",
+            Self::CopySource => "copy_source",
+            Self::ReadCapExceeded => "read_cap_exceeded",
+            Self::InlineData => "inline_data",
+            Self::NonLocalShard => "non_local_shard",
+            Self::InvalidPath => "invalid_path",
+            Self::NoLocalShards => "no_local_shards",
+            Self::UnsupportedPlatform => "unsupported_platform",
+        }
+    }
+}
+
+/// Record eligibility, including path preparation only when it actually ran.
+pub fn record_local_batch_read_decision(decision: LocalBatchReadDecision, preparation: Option<Duration>) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
+    let decision = decision.as_str();
+    counter!("rustfs_io_get_object_local_batch_decisions_total", "decision" => decision).increment(1);
+    if let Some(duration) = preparation {
+        histogram!("rustfs_io_get_object_local_batch_prepare_seconds", "decision" => decision).record(duration.as_secs_f64());
+    }
+}
+
+/// Stages of the existing buffered positioned-read implementation.
+#[derive(Clone, Copy)]
+pub enum LocalBatchReadStage {
+    QueueWait,
+    Execution,
+    Metadata,
+    Open,
+    Read,
+}
+
+impl LocalBatchReadStage {
+    const fn index(self) -> usize {
+        match self {
+            Self::QueueWait => 0,
+            Self::Execution => 1,
+            Self::Metadata => 2,
+            Self::Open => 3,
+            Self::Read => 4,
+        }
+    }
+}
+
+/// Handles captured before dispatch, kept alive by the actual blocking worker.
+///
+/// Capturing the recorder once avoids per-I/O registration and also preserves
+/// completion samples when the caller stops waiting. No timers or handles are
+/// created with GET-stage metrics disabled. The snapshot applies to the whole
+/// submitted batch, including work that finishes after the switch is disabled.
+pub struct LocalBatchReadMetrics {
+    stages: [[Histogram; 2]; 5],
+    shards: [Counter; 2],
+    requested_bytes: [Counter; 2],
+    read_bytes: [Counter; 2],
+    batch_size: Histogram,
+}
+
+fn outcome(success: bool) -> &'static str {
+    if success { "success" } else { "error" }
+}
+
+impl LocalBatchReadMetrics {
+    /// Resolve bounded-label handles only for an explicitly instrumented batch.
+    /// Boxing the enabled state keeps the blocking task small when disabled.
+    pub fn capture() -> Option<Box<Self>> {
+        if !get_stage_metrics_enabled() {
+            return None;
+        }
+        let stages = ["queue_wait", "execution", "metadata", "open", "read"].map(|stage| {
+            [false, true].map(|success| {
+                histogram!("rustfs_io_get_object_local_batch_stage_seconds", "stage" => stage, "outcome" => outcome(success))
+            })
+        });
+        Some(Box::new(Self {
+            stages,
+            shards: [false, true]
+                .map(|success| counter!("rustfs_io_get_object_local_batch_shards_total", "outcome" => outcome(success))),
+            requested_bytes: [false, true]
+                .map(|success| counter!("rustfs_io_get_object_local_batch_requested_bytes_total", "outcome" => outcome(success))),
+            read_bytes: [false, true]
+                .map(|success| counter!("rustfs_io_get_object_local_batch_read_bytes_total", "outcome" => outcome(success))),
+            batch_size: histogram!("rustfs_io_get_object_local_batch_shards"),
+        }))
+    }
+
+    /// Count shards actually submitted to this blocking batch.
+    pub fn record_batch(&self, shards: usize) {
+        self.batch_size.record(crate::shard_read_fanout_to_f64(shards));
+    }
+
+    /// Record an executed stage; unattempted stages receive no sample.
+    ///
+    /// Queue wait succeeds when the worker starts. Execution covers the whole
+    /// batch, while metadata/open/read describe individual I/O operations.
+    /// Read timing excludes buffer allocation; execution timing includes it.
+    pub fn record_stage(&self, stage: LocalBatchReadStage, success: bool, duration: Duration) {
+        self.stages[stage.index()][usize::from(success)].record(duration.as_secs_f64());
+    }
+
+    /// Separate completed shard outcomes and bytes actually returned by read_at.
+    pub fn record_shard(&self, success: bool, requested: usize, read: usize) {
+        let index = usize::from(success);
+        self.shards[index].increment(1);
+        self.requested_bytes[index].increment(u64::try_from(requested).unwrap_or(u64::MAX));
+        self.read_bytes[index].increment(u64::try_from(read).unwrap_or(u64::MAX));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::set_get_stage_metrics_enabled;
+    use crate::tests::{METRICS_FLAG_LOCK, counter_total, histogram_samples};
+    use metrics_util::debugging::DebuggingRecorder;
+    use std::time::Duration;
+
+    #[test]
+    fn local_batch_read_metrics_are_disabled_without_stage_opt_in() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|error| error.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        metrics::with_local_recorder(&recorder, || {
+            set_get_stage_metrics_enabled(false);
+            assert!(LocalBatchReadMetrics::capture().is_none());
+            record_local_batch_read_decision(LocalBatchReadDecision::Selected, Some(Duration::from_millis(1)));
+        });
+        assert!(snapshotter.snapshot().into_vec().is_empty());
+    }
+
+    #[test]
+    fn local_batch_read_handles_record_worker_costs_and_actual_bytes() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|error| error.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let observation = metrics::with_local_recorder(&recorder, || {
+            set_get_stage_metrics_enabled(true);
+            record_local_batch_read_decision(LocalBatchReadDecision::Selected, Some(Duration::from_millis(2)));
+            LocalBatchReadMetrics::capture().expect("enabled batch metrics must capture recorder handles")
+        });
+        set_get_stage_metrics_enabled(false);
+        std::thread::spawn(move || {
+            observation.record_stage(LocalBatchReadStage::QueueWait, true, Duration::from_millis(3));
+            observation.record_stage(LocalBatchReadStage::Metadata, false, Duration::from_millis(4));
+            observation.record_shard(false, 1024, 0);
+            observation.record_shard(true, 16, 16);
+            observation.record_batch(2);
+        })
+        .join()
+        .expect("worker should record through the captured handles");
+        set_get_stage_metrics_enabled(false);
+
+        let rows = snapshotter.snapshot().into_vec();
+        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_decisions_total"), Some(1));
+        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_shards_total"), Some(2));
+        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_requested_bytes_total"), Some(1040));
+        assert_eq!(counter_total(&rows, "rustfs_io_get_object_local_batch_read_bytes_total"), Some(16));
+        assert_eq!(histogram_samples(&rows, "rustfs_io_get_object_local_batch_prepare_seconds"), vec![0.002]);
+        assert_eq!(
+            histogram_samples(&rows, "rustfs_io_get_object_local_batch_stage_seconds"),
+            vec![0.003, 0.004]
+        );
+        assert_eq!(histogram_samples(&rows, "rustfs_io_get_object_local_batch_shards"), vec![2.0]);
+        for (key, _, _, _) in rows {
+            for label in key.key().labels() {
+                assert!(matches!(label.key(), "decision" | "stage" | "outcome"));
+            }
+        }
+    }
+}

crates/obs/src/telemetry/otel.rs

@@ -83,6 +83,16 @@ const GET_OBJECT_DURATION_HISTOGRAM_BUCKETS: &[f64] = &[
     0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
 ];
 
+const LOCAL_BATCH_DURATION_HISTOGRAM_METRICS: &[&str] = &[
+    "rustfs_io_get_object_local_batch_prepare_seconds",
+    "rustfs_io_get_object_local_batch_stage_seconds",
+];
+
+const LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS: &[f64] = &[
+    0.0000001, 0.00000025, 0.0000005, 0.000001, 0.0000025, 0.000005, 0.00001, 0.000025, 0.00005, 0.0001, 0.00025, 0.0005, 0.001,
+    0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
+];
+
 const DEFAULT_OTLP_HTTP_TIMEOUT: Duration = Duration::from_secs(10);
 const OTEL_EXPORTER_OTLP_TIMEOUT: &str = "OTEL_EXPORTER_OTLP_TIMEOUT";
 const OTEL_EXPORTER_OTLP_TRACES_TIMEOUT: &str = "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT";
@@ -481,21 +491,25 @@ fn build_meter_provider(
 }
 
 fn get_object_duration_histogram_view(instrument: &Instrument) -> Option<Stream> {
-    if !is_get_object_duration_histogram_metric(instrument.name()) {
-        return None;
-    }
+    let boundaries = get_object_duration_histogram_buckets(instrument.name())?;
 
     Stream::builder()
         .with_aggregation(Aggregation::ExplicitBucketHistogram {
-            boundaries: GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.to_vec(),
+            boundaries: boundaries.to_vec(),
             record_min_max: true,
         })
         .build()
         .ok()
 }
 
-fn is_get_object_duration_histogram_metric(name: &str) -> bool {
-    GET_OBJECT_DURATION_HISTOGRAM_METRICS.contains(&name)
+fn get_object_duration_histogram_buckets(name: &str) -> Option<&'static [f64]> {
+    if LOCAL_BATCH_DURATION_HISTOGRAM_METRICS.contains(&name) {
+        Some(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS)
+    } else if GET_OBJECT_DURATION_HISTOGRAM_METRICS.contains(&name) {
+        Some(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS)
+    } else {
+        None
+    }
 }
 
 /// Build an optional [`SdkLoggerProvider`] for the given log endpoint.
@@ -1098,10 +1112,22 @@ mod tests {
 
     #[test]
     fn test_get_object_duration_histogram_metric_match_is_scoped() {
-        assert!(is_get_object_duration_histogram_metric("rustfs_io_get_object_stage_duration_seconds"));
-        assert!(is_get_object_duration_histogram_metric("rustfs_io_get_object_request_duration_seconds"));
-        assert!(!is_get_object_duration_histogram_metric("rustfs_io_put_object_request_duration_seconds"));
-        assert!(!is_get_object_duration_histogram_metric("rustfs_io_get_object_response_size_bytes"));
+        for name in [
+            "rustfs_io_get_object_stage_duration_seconds",
+            "rustfs_io_get_object_request_duration_seconds",
+        ] {
+            assert_eq!(get_object_duration_histogram_buckets(name), Some(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS));
+        }
+        for name in LOCAL_BATCH_DURATION_HISTOGRAM_METRICS {
+            assert_eq!(get_object_duration_histogram_buckets(name), Some(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS));
+        }
+        for name in [
+            "rustfs_io_put_object_request_duration_seconds",
+            "rustfs_io_get_object_response_size_bytes",
+            "rustfs_io_get_object_local_batch_shards",
+        ] {
+            assert!(get_object_duration_histogram_buckets(name).is_none());
+        }
     }
 
     #[test]
@@ -1111,6 +1137,17 @@ mod tests {
         assert_eq!(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.last(), Some(&10.0));
     }
 
+    #[test]
+    fn local_batch_duration_buckets_resolve_sub_microsecond_operations() {
+        assert!(
+            LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS
+                .windows(2)
+                .all(|pair| pair[0] < pair[1])
+        );
+        assert_eq!(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS.first(), Some(&0.0000001));
+        assert_eq!(LOCAL_BATCH_DURATION_HISTOGRAM_BUCKETS.last(), Some(&10.0));
+    }
+
     #[cfg(all(
         feature = "pyroscope",
         any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))

crates/scanner/Cargo.toml

@@ -107,10 +107,6 @@ bytes.workspace = true
 hex-simd.workspace = true
 
 [dev-dependencies]
-opentelemetry.workspace = true
-opentelemetry-appender-tracing = { workspace = true, features = ["experimental_span_attributes", "experimental_metadata_attributes"] }
-opentelemetry_sdk.workspace = true
-rustfs-common = { workspace = true, features = ["test-util"] }
 rustfs-heal.workspace = true
 rustfs-test-utils.workspace = true
 tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }

crates/scanner/src/data_usage_define.rs

@@ -67,11 +67,12 @@ const METRIC_CACHE_SAVE_RETRY_TOTAL: &str = "rustfs_scanner_cache_save_retry_tot
 const METRIC_CACHE_SAVE_PAYLOAD_BYTES_TOTAL: &str = "rustfs_scanner_cache_save_payload_bytes_total";
 const METRIC_CACHE_SAVE_DURATION_SECONDS: &str = "rustfs_scanner_cache_save_duration_seconds";
 const METRIC_CACHE_BACKUP_REVISION_FAILURE_TOTAL: &str = "rustfs_scanner_cache_backup_revision_failure_total";
+const METRIC_CACHE_READ_ORIGIN_TOTAL: &str = "rustfs_scanner_cache_read_origin_total";
 const LOG_COMPONENT_SCANNER: &str = "scanner";
 const LOG_SUBSYSTEM_CACHE: &str = "cache";
 const EVENT_SCANNER_CACHE_LOAD_STATE: &str = "scanner_cache_load_state";
 const EVENT_SCANNER_CACHE_SAVE_STATE: &str = "scanner_cache_save_state";
-static CACHE_SAVE_METRICS_ONCE: Once = Once::new();
+static CACHE_METRICS_ONCE: Once = Once::new();
 
 pub const DATA_USAGE_SCAN_CHECKPOINT_VERSION: u16 = 1;
 pub const DATA_USAGE_RAW_ENUMERATION_CURSOR_VERSION: u16 = 1;
@@ -204,6 +205,54 @@ pub(crate) struct DataUsageCacheRevisions {
     backup: Option<DataUsageCacheRevision>,
 }
 
+/// The observed result of trying the logical primary cache slot.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum DataUsageCachePrimaryRead {
+    Missing,
+    Corrupt,
+}
+
+/// Process-local read provenance, not evidence of freshness or authority.
+/// Legacy objects may come from the old data-usage bucket. Empty may mean
+/// missing or corrupt objects; neither state certifies a complete snapshot.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum DataUsageCacheReadOrigin {
+    Primary {
+        legacy: bool,
+    },
+    Backup {
+        primary: DataUsageCachePrimaryRead,
+        legacy: bool,
+    },
+    Empty,
+}
+
+impl DataUsageCacheReadOrigin {
+    fn metric_label(self) -> &'static str {
+        match self {
+            Self::Primary { legacy: false } => "primary",
+            Self::Primary { legacy: true } => "legacy_primary",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: false,
+            } => "backup_primary_missing",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: true,
+            } => "legacy_backup_primary_missing",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: false,
+            } => "backup_primary_corrupt",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: true,
+            } => "legacy_backup_primary_corrupt",
+            Self::Empty => "empty",
+        }
+    }
+}
+
 pub static DATA_USAGE_BUCKET: LazyLock<String> =
     LazyLock::new(|| format!("{RUSTFS_META_BUCKET}{SLASH_SEPARATOR}{BUCKET_META_PREFIX}"));
 
@@ -1271,8 +1320,12 @@ impl DataUsageCache {
         Ok(())
     }
 
-    fn ensure_cache_save_metrics_registered() {
-        CACHE_SAVE_METRICS_ONCE.call_once(|| {
+    fn ensure_cache_metrics_registered() {
+        CACHE_METRICS_ONCE.call_once(|| {
+            describe_counter!(
+                METRIC_CACHE_READ_ORIGIN_TOTAL,
+                "Successful scanner cache loads by observed logical origin; does not prove cache freshness."
+            );
             describe_counter!(
                 METRIC_CACHE_SAVE_ATTEMPT_TOTAL,
                 "Total scanner data usage cache save attempts by result and cache type."

crates/scanner/src/data_usage_define/persistence.rs

@@ -70,6 +70,7 @@ struct DataUsageCacheLoadResult {
     cache: DataUsageCache,
     main_revision: DataUsageCacheRevision,
     backup_revision: Option<DataUsageCacheRevision>,
+    origin: DataUsageCacheReadOrigin,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -97,6 +98,7 @@ impl DataUsageCache {
     /// If the object is not found, a nil error with empty data usage cache is returned.
     pub async fn load<S: ScannerObjectIO>(&mut self, store: Arc<S>, name: &str) -> StorageResult<()> {
         let loaded = Self::load_cache(store, name).await?;
+        Self::record_cache_read_origin(loaded.origin);
         *self = loaded.cache;
         Ok(())
     }
@@ -106,6 +108,17 @@ impl DataUsageCache {
         store: Arc<S>,
         name: &str,
     ) -> StorageResult<DataUsageCacheRevisions> {
+        self.load_with_revisions_and_origin(store, name)
+            .await
+            .map(|(revisions, _)| revisions)
+    }
+
+    /// Origin is an observation of this load, not proof of current coverage.
+    pub(crate) async fn load_with_revisions_and_origin<S: ScannerObjectIO>(
+        &mut self,
+        store: Arc<S>,
+        name: &str,
+    ) -> StorageResult<(DataUsageCacheRevisions, DataUsageCacheReadOrigin)> {
         let backup_name = format!("{name}.bkp");
         let backup_path = path_join_buf(&[BUCKET_META_PREFIX, &backup_name]);
         let loaded = Self::load_cache(store.clone(), name).await?;
@@ -131,9 +144,16 @@ impl DataUsageCache {
             },
         };
         let main = loaded.main_revision;
+        let origin = loaded.origin;
+        Self::record_cache_read_origin(origin);
         *self = loaded.cache;
 
-        Ok(DataUsageCacheRevisions { main, backup })
+        Ok((DataUsageCacheRevisions { main, backup }, origin))
+    }
+
+    fn record_cache_read_origin(origin: DataUsageCacheReadOrigin) {
+        Self::ensure_cache_metrics_registered();
+        counter!(METRIC_CACHE_READ_ORIGIN_TOTAL, "origin" => origin.metric_label()).increment(1);
     }
 
     pub(crate) async fn read_revisions<S: ScannerObjectIO>(store: Arc<S>, name: &str) -> StorageResult<DataUsageCacheRevisions> {
@@ -156,10 +176,12 @@ impl DataUsageCache {
                     cache,
                     revision: Some(main_revision),
                 } => {
+                    let legacy = matches!(&main_revision, DataUsageCacheRevision::Missing);
                     return Ok(DataUsageCacheLoadResult {
                         cache: *cache,
                         main_revision,
                         backup_revision: None,
+                        origin: DataUsageCacheReadOrigin::Primary { legacy },
                     });
                 }
                 DataUsageCacheLoadAttempt::Loaded { revision: None, .. } => {
@@ -193,10 +215,17 @@ impl DataUsageCache {
                         let main_revision = main_revision.ok_or_else(|| {
                             Error::other(format!("scanner cache main revision is unavailable while loading backup: {name}"))
                         })?;
+                        let primary = if main_state == DataUsageCacheLoadState::Corrupt {
+                            DataUsageCachePrimaryRead::Corrupt
+                        } else {
+                            DataUsageCachePrimaryRead::Missing
+                        };
+                        let legacy = matches!(&backup_revision, DataUsageCacheRevision::Missing);
                         return Ok(DataUsageCacheLoadResult {
                             cache: *cache,
                             main_revision,
                             backup_revision: Some(backup_revision),
+                            origin: DataUsageCacheReadOrigin::Backup { primary, legacy },
                         });
                     }
                     DataUsageCacheLoadState::Loaded
@@ -220,6 +249,7 @@ impl DataUsageCache {
                         main_revision: main_revision
                             .ok_or_else(|| Error::other(format!("scanner cache missing state has no revision: {name}")))?,
                         backup_revision,
+                        origin: DataUsageCacheReadOrigin::Empty,
                     });
                 }
                 (DataUsageCacheLoadState::Corrupt, DataUsageCacheLoadState::Missing)
@@ -239,6 +269,7 @@ impl DataUsageCache {
                         main_revision: main_revision
                             .ok_or_else(|| Error::other(format!("scanner cache corrupt state has no revision: {name}")))?,
                         backup_revision,
+                        origin: DataUsageCacheReadOrigin::Empty,
                     });
                 }
                 _ => {}
@@ -518,7 +549,7 @@ impl DataUsageCache {
         revision: Option<DataUsageCacheRevision>,
         expected_epoch: Option<u64>,
     ) -> StorageResult<()> {
-        Self::ensure_cache_save_metrics_registered();
+        Self::ensure_cache_metrics_registered();
         let path_type = Self::cache_path_type(path);
         let path = path.to_string();
         let sha256hex = (revision.is_some() && !buf.is_empty())

crates/scanner/src/data_usage_define/tests.rs

@@ -168,8 +168,10 @@ impl AsyncRead for PrefixThenErrorReader {
 #[derive(Debug)]
 struct CacheReadStore {
     main: CacheReadBody,
+    main_missing: bool,
     backup: Option<Vec<u8>>,
     legacy: Option<Vec<u8>>,
+    legacy_backup: Option<Vec<u8>>,
     puts: AtomicUsize,
     #[cfg(feature = "diagnostic-read-version-caller")]
     entry_sources: Mutex<Vec<Option<crate::storage_api::ecstore_diagnostic::ReadVersionEntrySource>>>,
@@ -185,8 +187,10 @@ impl CacheReadStore {
     fn new(main: CacheReadBody, backup: Option<Vec<u8>>) -> Self {
         Self {
             main,
+            main_missing: false,
             backup,
             legacy: None,
+            legacy_backup: None,
             puts: AtomicUsize::new(0),
             #[cfg(feature = "diagnostic-read-version-caller")]
             entry_sources: Mutex::new(Vec::new()),
@@ -239,8 +243,12 @@ impl ObjectIO for CacheReadStore {
             .await
             .push(crate::storage_api::ecstore_diagnostic::current_read_version_entry_source());
         if bucket == DATA_USAGE_BUCKET.as_str() {
-            return self
-                .legacy
+            let legacy = if object.ends_with(".bkp") {
+                &self.legacy_backup
+            } else {
+                &self.legacy
+            };
+            return legacy
                 .clone()
                 .map(|data| Self::reader(CacheReadBody::Bytes(data), "legacy-etag"))
                 .ok_or(Error::FileNotFound);
@@ -252,6 +260,9 @@ impl ObjectIO for CacheReadStore {
         let main_path = path_join_buf(&[BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME]);
         let backup_path = format!("{main_path}.bkp");
         if object == main_path {
+            if self.main_missing {
+                return Err(Error::FileNotFound);
+            }
             return Ok(Self::reader(self.main.clone(), "main-etag"));
         }
         if object == backup_path {
@@ -712,6 +723,143 @@ async fn corrupt_primary_cache_loads_valid_backup() {
     assert_eq!(store.puts.load(Ordering::SeqCst), 0);
 }
 
+#[tokio::test]
+async fn cache_read_origin_reports_logical_slot_and_legacy_location() {
+    let mut primary = DataUsageCache::default();
+    primary.info.name = "primary".to_owned();
+    let mut backup = DataUsageCache::default();
+    backup.info.name = "backup".to_owned();
+    backup.info.snapshot_complete = true;
+    let primary_bytes = primary.marshal_msg().expect("serialize primary");
+    let backup_bytes = backup.marshal_msg().expect("serialize backup");
+    let empty_bytes = DataUsageCache::default().marshal_msg().expect("serialize empty cache");
+    let corrupt = b"not-msgpack".to_vec();
+
+    let cases = [
+        (
+            "primary",
+            Some(primary_bytes.clone()),
+            Some(backup_bytes.clone()),
+            None,
+            None,
+            DataUsageCacheReadOrigin::Primary { legacy: false },
+            primary_bytes.clone(),
+        ),
+        (
+            "legacy primary",
+            None,
+            None,
+            Some(primary_bytes.clone()),
+            None,
+            DataUsageCacheReadOrigin::Primary { legacy: true },
+            primary_bytes,
+        ),
+        (
+            "backup after missing",
+            None,
+            Some(backup_bytes.clone()),
+            None,
+            None,
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: false,
+            },
+            backup_bytes.clone(),
+        ),
+        (
+            "legacy backup after missing",
+            None,
+            None,
+            None,
+            Some(backup_bytes.clone()),
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: true,
+            },
+            backup_bytes.clone(),
+        ),
+        (
+            "backup after corrupt",
+            Some(corrupt.clone()),
+            Some(backup_bytes.clone()),
+            None,
+            None,
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: false,
+            },
+            backup_bytes.clone(),
+        ),
+        (
+            "legacy backup after corrupt",
+            Some(corrupt.clone()),
+            None,
+            None,
+            Some(backup_bytes.clone()),
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: true,
+            },
+            backup_bytes,
+        ),
+        (
+            "missing objects",
+            None,
+            None,
+            None,
+            None,
+            DataUsageCacheReadOrigin::Empty,
+            empty_bytes.clone(),
+        ),
+        (
+            "corrupt primary without backup",
+            Some(corrupt),
+            None,
+            None,
+            None,
+            DataUsageCacheReadOrigin::Empty,
+            empty_bytes,
+        ),
+    ];
+
+    for (name, main, current_backup, legacy_primary, legacy_backup, expected_origin, expected_bytes) in cases {
+        let has_main = main.is_some();
+        let has_backup = current_backup.is_some();
+        let mut fixture = CacheReadStore::new(CacheReadBody::Bytes(main.unwrap_or_default()), current_backup);
+        fixture.main_missing = !has_main;
+        fixture.legacy = legacy_primary;
+        fixture.legacy_backup = legacy_backup;
+        let store = Arc::new(fixture);
+        let mut loaded = DataUsageCache::default();
+        let (revisions, origin) = loaded
+            .load_with_revisions_and_origin(store.clone(), DATA_USAGE_CACHE_NAME)
+            .await
+            .expect("fixture cache load");
+
+        assert_eq!(origin, expected_origin, "origin for {name}");
+        assert_eq!(loaded.marshal_msg().expect("re-encode cache"), expected_bytes, "cache body for {name}");
+        assert_eq!(
+            revisions.main,
+            if has_main {
+                DataUsageCacheRevision::Etag("main-etag".to_owned())
+            } else {
+                DataUsageCacheRevision::Missing
+            },
+            "primary CAS revision for {name}"
+        );
+        assert_eq!(
+            revisions.backup,
+            Some(if has_backup {
+                DataUsageCacheRevision::Etag("backup-etag".to_owned())
+            } else {
+                DataUsageCacheRevision::Missing
+            }),
+            "backup CAS revision for {name}"
+        );
+        assert_eq!(store.puts.load(Ordering::SeqCst), 0, "reading must not write for {name}");
+    }
+}
+
 #[tokio::test]
 async fn corrupt_cache_without_valid_backup_rebuilds_with_cas_revisions() {
     for backup in [None, Some(b"also-not-msgpack".to_vec())] {

crates/scanner/src/data_usage_define/tests/save_metrics.rs

@@ -7,17 +7,18 @@ use std::sync::atomic::AtomicU64;
 
 const PAYLOAD_BYTES: &str = "rustfs_scanner_cache_save_payload_bytes_total";
 const ENCODED_BYTES: &str = "rustfs_scanner_cache_serialized_bytes_total";
+const READ_ORIGIN: &str = "rustfs_scanner_cache_read_origin_total";
 
 #[derive(Default)]
-struct SaveRecorder(StdMutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
+struct CacheRecorder(StdMutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
 
-impl metrics::Recorder for SaveRecorder {
+impl metrics::Recorder for CacheRecorder {
     fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
     fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
     fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
 
     fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
-        if ![PAYLOAD_BYTES, ENCODED_BYTES].contains(&key.name()) {
+        if ![PAYLOAD_BYTES, ENCODED_BYTES, READ_ORIGIN].contains(&key.name()) {
             return metrics::Counter::noop();
         }
         metrics::Counter::from_arc(self.0.lock().expect("save recorder").entry(key.clone()).or_default().clone())
@@ -36,7 +37,7 @@ async fn check_save(store: BackupFallbackStore, revisions: Option<DataUsageCache
     let store = Arc::new(store);
     let cache = DataUsageCache::default();
     let encoded = rmp_serde::to_vec(&cache).expect("independent encoded body");
-    let recorder = SaveRecorder::default();
+    let recorder = CacheRecorder::default();
     let save = async {
         match revisions {
             Some(revisions) => {
@@ -85,6 +86,49 @@ async fn check_save(store: BackupFallbackStore, revisions: Option<DataUsageCache
     );
 }
 
+#[tokio::test(start_paused = true)]
+async fn read_origin_counts_successful_loads_with_fixed_labels() {
+    let bytes = DataUsageCache::default().marshal_msg().expect("serialize backup");
+    let corrupt_primary = CacheReadStore::new(CacheReadBody::Bytes(b"not-msgpack".to_vec()), Some(bytes.clone()));
+    let mut legacy_backup = CacheReadStore::new(CacheReadBody::Bytes(Vec::new()), None);
+    legacy_backup.main_missing = true;
+    legacy_backup.legacy_backup = Some(bytes);
+    let mut empty = CacheReadStore::new(CacheReadBody::Bytes(Vec::new()), None);
+    empty.main_missing = true;
+    let recorder = CacheRecorder::default();
+
+    for (store, with_revisions) in [(corrupt_primary, false), (legacy_backup, true), (empty, true)] {
+        let store = Arc::new(store);
+        let mut loaded = DataUsageCache::default();
+        let read = async {
+            if with_revisions {
+                loaded.load_with_revisions(store, DATA_USAGE_CACHE_NAME).await.map(|_| ())
+            } else {
+                loaded.load(store, DATA_USAGE_CACHE_NAME).await
+            }
+        };
+        let mut read = std::pin::pin!(read);
+        std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || read.as_mut().poll(cx)))
+            .await
+            .expect("fixture load");
+    }
+
+    let unavailable = Arc::new(BackupFallbackStore::new(Vec::new(), false));
+    let mut loaded = DataUsageCache::default();
+    let mut failed_read = std::pin::pin!(loaded.load_with_revisions(unavailable, DATA_USAGE_CACHE_NAME));
+    let error = std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || failed_read.as_mut().poll(cx)))
+        .await
+        .expect_err("an unreadable primary must not count as a fallback");
+    assert!(matches!(error, Error::ErasureReadQuorum));
+
+    let rows = recorder.0.lock().expect("cache recorder");
+    assert_eq!(rows.len(), 3, "one fixed origin label per successful load");
+    for origin in ["backup_primary_corrupt", "legacy_backup_primary_missing", "empty"] {
+        let key = metrics::Key::from_parts(READ_ORIGIN, vec![metrics::Label::new("origin", origin)]);
+        assert_eq!(rows[&key].load(Ordering::Relaxed), 1, "one observation for {origin}");
+    }
+}
+
 fn revisions(backup: bool) -> DataUsageCacheRevisions {
     DataUsageCacheRevisions {
         main: DataUsageCacheRevision::Missing,

crates/scanner/src/scanner.rs

@@ -2082,7 +2082,7 @@ where
         None
     } else if let Some(notification_system) = storeapi.scanner_notification_system() {
         let lease_result = publication_plan
-            .await_leases(ctx, &cycle_budget, cycle_info.current, leader_epoch, |observational_lease| {
+            .await_leases(ctx, &cycle_budget, cycle_info.current, |observational_lease| {
                 let targets = remote_publication_lease_targets.clone();
                 let notification_system = Arc::clone(&notification_system);
                 async move {
@@ -3892,17 +3892,17 @@ impl ScannerCyclePublicationPlan {
         self != Self::Authoritative
     }
 
-    async fn await_leases<T, F, Fut>(
+    async fn await_leases<T, E, F, Fut>(
         self,
         parent: &CancellationToken,
         budget: &ScannerCycleBudget,
         cycle: u64,
-        leader_epoch: u64,
         mut acquire: F,
-    ) -> ScannerPublicationProofWait<T, EcstoreError>
+    ) -> ScannerPublicationProofWait<T, E>
     where
+        E: std::fmt::Display,
         F: FnMut(bool) -> Fut,
-        Fut: Future<Output = Result<T, EcstoreError>>,
+        Fut: Future<Output = Result<T, E>>,
     {
         // Exhausting the walk budget must not cancel the movement proof needed
         // to save partial progress. No usage update means namespace stability
@@ -3912,17 +3912,6 @@ impl ScannerCyclePublicationPlan {
         } else {
             budget.token()
         };
-        let started_at = Instant::now();
-        let publication_kind = match self {
-            Self::ProgressOnly => "progress_only",
-            Self::Observational => "observational",
-            Self::Authoritative => "authoritative",
-        };
-        let lease_purpose = if self.uses_observational_lease() {
-            "observational"
-        } else {
-            "authoritative"
-        };
         debug!(
             target: "rustfs::scanner",
             event = EVENT_SCANNER_CYCLE_STATE,
@@ -3930,66 +3919,30 @@ impl ScannerCyclePublicationPlan {
             subsystem = LOG_SUBSYSTEM_RUNTIME,
             state = "publication_proof_started",
             cycle,
-            leader_epoch,
             stage = "lease_acquire",
-            publication_kind,
-            lease_purpose,
+            publication_kind = match self {
+                Self::ProgressOnly => "progress_only",
+                Self::Observational => "observational",
+                Self::Authoritative => "authoritative",
+            },
             "Scanner is acquiring peer publication proof"
         );
         let proof = await_scanner_publication_proof(
             &proof_ctx,
-            ScannerPublicationProofContext {
-                cycle,
-                stage: "lease_acquire",
-                leader_epoch: Some(leader_epoch),
-                lease_purpose: Some(lease_purpose),
-            },
+            cycle,
+            "lease_acquire",
             || acquire(self.uses_observational_lease()),
-            |err| {
-                // The structured reason comes from local acquisition, not a peer report.
-                crate::storage_api::scan::scanner_publication_lease_failure_info(err)
-                    .is_some_and(|info| info.reason == Some("channel_acquisition_timed_out"))
-                    || scanner_publication_lease_error_is_retryable(&err.to_string())
-            },
-            crate::storage_api::scan::scanner_publication_lease_failure_info,
+            |err| scanner_publication_lease_error_is_retryable(&err.to_string()),
         );
-        let (proof, timed_out) = if self == Self::ProgressOnly {
+        if self == Self::ProgressOnly {
             // Object/directory budgets can expire with no runtime deadline.
             // Bound the control-plane drain independently of the walk budget.
-            match tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, proof).await {
-                Ok(proof) => (proof, false),
-                Err(_) => (ScannerPublicationProofWait::Cancelled, true),
-            }
+            tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, proof)
+                .await
+                .unwrap_or(ScannerPublicationProofWait::Cancelled)
         } else {
-            (proof.await, false)
-        };
-        let outcome = if timed_out {
-            "timed_out"
-        } else {
-            match &proof {
-                ScannerPublicationProofWait::Ready(_) => "acquired",
-                ScannerPublicationProofWait::Rejected(_) => "rejected",
-                ScannerPublicationProofWait::Cancelled => "cancelled",
-            }
-        };
-        debug!(
-            target: "rustfs::scanner",
-            event = EVENT_SCANNER_CYCLE_STATE,
-            component = LOG_COMPONENT_SCANNER,
-            subsystem = LOG_SUBSYSTEM_RUNTIME,
-            state = "publication_lease_wait_finished",
-            cycle,
-            leader_epoch,
-            stage = "lease_acquire",
-            publication_kind,
-            lease_purpose,
-            outcome,
-            parent_cancelled_at_return = parent.is_cancelled(),
-            proof_token_cancelled_at_return = proof_ctx.is_cancelled(),
-            elapsed = ?started_at.elapsed(),
-            "Scanner peer publication lease wait finished"
-        );
-        proof
+            proof.await
+        }
     }
 }
 

crates/scanner/src/scanner/activity.rs

@@ -14,7 +14,7 @@
 /// Cycle wake/backoff policy and scanner activity observation (probing, generations, topology digest).
 use super::*;
 use crate::storage_api::ScannerStorage;
-use crate::storage_api::scan::{SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, ScannerPublicationLeaseFailureInfo};
+use crate::storage_api::scan::SCANNER_ACTIVITY_V6_PROTOCOL_VERSION;
 use std::collections::HashSet;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -239,7 +239,7 @@ pub(crate) fn scanner_publication_proof_retry_delay(consecutive_failures: u32) -
 }
 
 pub(crate) fn scanner_publication_activity_error_is_retryable(error: &str) -> bool {
-    crate::storage_api::scan::scanner_peer_transport_error_message_is_retryable(error)
+    crate::storage_api::scanner_peer_transport_error_message_is_retryable(error)
 }
 
 pub(crate) fn scanner_publication_lease_error_is_retryable(error: &str) -> bool {
@@ -256,19 +256,12 @@ pub(crate) enum ScannerPublicationProofWait<T, E> {
     Cancelled,
 }
 
-pub(crate) struct ScannerPublicationProofContext {
-    pub cycle: u64,
-    pub stage: &'static str,
-    pub leader_epoch: Option<u64>,
-    pub lease_purpose: Option<&'static str>,
-}
-
 pub(crate) async fn await_scanner_publication_proof<T, E, F, Fut, Retryable>(
     ctx: &CancellationToken,
-    context: ScannerPublicationProofContext,
+    cycle: u64,
+    stage: &'static str,
     mut proof: F,
     retryable: Retryable,
-    failure_info: for<'a> fn(&'a E) -> Option<ScannerPublicationLeaseFailureInfo<'a>>,
 ) -> ScannerPublicationProofWait<T, E>
 where
     F: FnMut() -> Fut,
@@ -276,12 +269,6 @@ where
     E: std::fmt::Display,
     Retryable: Fn(&E) -> bool,
 {
-    let ScannerPublicationProofContext {
-        cycle,
-        stage,
-        leader_epoch,
-        lease_purpose,
-    } = context;
     let started_at = Instant::now();
     let mut consecutive_failures = 0u32;
 
@@ -301,8 +288,6 @@ where
                         state = "publication_proof_recovered",
                         cycle,
                         stage,
-                        leader_epoch,
-                        lease_purpose,
                         attempts = consecutive_failures.saturating_add(1),
                         pending_duration = ?started_at.elapsed(),
                         "Scanner publication proof recovered"
@@ -311,11 +296,6 @@ where
                 return ScannerPublicationProofWait::Ready(value);
             }
             Err(err) if !retryable(&err) => {
-                let failure = failure_info(&err);
-                let peer = failure.and_then(|info| info.peer);
-                let failure_kind = failure.map(|info| info.kind);
-                let reason = failure.and_then(|info| info.reason);
-                let reported_reason = failure.and_then(|info| info.reported_reason);
                 warn!(
                     target: "rustfs::scanner",
                     event = EVENT_SCANNER_CYCLE_STATE,
@@ -324,12 +304,7 @@ where
                     state = "publication_proof_rejected",
                     cycle,
                     stage,
-                    leader_epoch,
-                    lease_purpose,
-                    peer,
-                    failure_kind,
-                    reason,
-                    reported_reason,
+                    error = %err,
                     "Scanner publication proof failed with a non-retryable cluster state"
                 );
                 return ScannerPublicationProofWait::Rejected(err);
@@ -337,11 +312,6 @@ where
             Err(err) => {
                 consecutive_failures = consecutive_failures.saturating_add(1);
                 let retry_delay = scanner_publication_proof_retry_delay(consecutive_failures);
-                let failure = failure_info(&err);
-                let peer = failure.and_then(|info| info.peer);
-                let failure_kind = failure.map(|info| info.kind);
-                let reason = failure.and_then(|info| info.reason);
-                let reported_reason = failure.and_then(|info| info.reported_reason);
                 if consecutive_failures == 1 || consecutive_failures.is_multiple_of(20) {
                     warn!(
                         target: "rustfs::scanner",
@@ -351,15 +321,10 @@ where
                         state = "publication_proof_pending",
                         cycle,
                         stage,
-                        leader_epoch,
-                        lease_purpose,
-                        peer,
-                        failure_kind,
-                        reason,
-                        reported_reason,
                         attempt = consecutive_failures,
                         retry_delay = ?retry_delay,
                         pending_duration = ?started_at.elapsed(),
+                        error = %err,
                         "Scanner is waiting for publication proof"
                     );
                 } else {
@@ -371,14 +336,9 @@ where
                         state = "publication_proof_retry",
                         cycle,
                         stage,
-                        leader_epoch,
-                        lease_purpose,
-                        peer,
-                        failure_kind,
-                        reason,
-                        reported_reason,
                         attempt = consecutive_failures,
                         retry_delay = ?retry_delay,
+                        error = %err,
                         "Scanner publication activity proof retry scheduled"
                     );
                 }
@@ -403,18 +363,9 @@ where
     F: FnMut() -> Fut,
     Fut: Future<Output = Result<ScannerActivitySnapshot, String>>,
 {
-    match await_scanner_publication_proof(
-        ctx,
-        ScannerPublicationProofContext {
-            cycle,
-            stage,
-            leader_epoch: None,
-            lease_purpose: None,
-        },
-        probe,
-        |err: &String| scanner_publication_activity_error_is_retryable(err),
-        |_| None,
-    )
+    match await_scanner_publication_proof(ctx, cycle, stage, probe, |err: &String| {
+        scanner_publication_activity_error_is_retryable(err)
+    })
     .await
     {
         ScannerPublicationProofWait::Ready(snapshot) => Ok(snapshot),

crates/scanner/src/scanner/tests.rs

@@ -9204,12 +9204,8 @@ async fn publication_lease_retries_pending_namespace_commit_until_it_clears() {
 
     let result = await_scanner_publication_proof(
         &ctx,
-        ScannerPublicationProofContext {
-            cycle: 17,
-            stage: "lease_acquire",
-            leader_epoch: None,
-            lease_purpose: None,
-        },
+        17,
+        "lease_acquire",
         move || {
             let attempt = proof_attempts.fetch_add(1, Ordering::SeqCst);
             async move {
@@ -9221,7 +9217,6 @@ async fn publication_lease_retries_pending_namespace_commit_until_it_clears() {
             }
         },
         |error| scanner_publication_lease_error_is_retryable(error),
-        |_| None,
     )
     .await;
 

crates/scanner/src/scanner/tests/cycle_budget_publication.rs

@@ -13,629 +13,8 @@
 // limitations under the License.
 
 use super::*;
-use crate::storage_api::scan::ScannerPublicationLeaseFailureInfo;
 use rustfs_utils::path::path_join_buf;
 
-#[derive(Clone)]
-struct LeaseLogWriter(Arc<std::sync::Mutex<Vec<u8>>>);
-
-impl std::io::Write for LeaseLogWriter {
-    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
-        self.0.lock().expect("lease log mutex").extend_from_slice(bytes);
-        Ok(bytes.len())
-    }
-
-    fn flush(&mut self) -> std::io::Result<()> {
-        Ok(())
-    }
-}
-
-#[tokio::test(start_paused = true)]
-async fn publication_lease_observations_keep_cycle_epoch_purpose_and_outcome() {
-    use tracing::instrument::WithSubscriber as _;
-
-    let output = Arc::new(std::sync::Mutex::new(Vec::new()));
-    let writer = LeaseLogWriter(Arc::clone(&output));
-    let subscriber = tracing_subscriber::fmt()
-        .json()
-        .with_max_level(tracing::Level::DEBUG)
-        .with_writer(move || writer.clone())
-        .with_ansi(false)
-        .without_time()
-        .finish();
-    let _interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
-
-    async {
-        let ctx = CancellationToken::new();
-        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
-        let ready = ScannerCyclePublicationPlan::ProgressOnly
-            .await_leases(&ctx, &budget, 501, 17, |_| async { Ok::<_, EcstoreError>(()) })
-            .await;
-        assert!(matches!(ready, ScannerPublicationProofWait::Ready(())));
-        let next_leader = ScannerCyclePublicationPlan::Observational
-            .await_leases(&ctx, &budget, 501, 22, |_| async { Ok::<_, EcstoreError>(()) })
-            .await;
-        assert!(matches!(next_leader, ScannerPublicationProofWait::Ready(())));
-        let rejected = ScannerCyclePublicationPlan::Observational
-            .await_leases(&ctx, &budget, 502, 18, |_| async {
-                Err::<(), _>(EcstoreError::other("protocol mismatch"))
-            })
-            .await;
-        assert!(matches!(rejected, ScannerPublicationProofWait::Rejected(_)));
-        ctx.cancel();
-        let cancelled = ScannerCyclePublicationPlan::Authoritative
-            .await_leases(&ctx, &budget, 503, 19, |_| async { Ok::<_, EcstoreError>(()) })
-            .await;
-        assert!(matches!(cancelled, ScannerPublicationProofWait::Cancelled));
-        let timeout_ctx = CancellationToken::new();
-        let timeout_budget = ScannerCycleBudget::new(&timeout_ctx, ScannerCycleBudgetConfig::default());
-        let timed_out = ScannerCyclePublicationPlan::ProgressOnly
-            .await_leases(&timeout_ctx, &timeout_budget, 504, 20, |_| async {
-                std::future::pending::<Result<(), EcstoreError>>().await
-            })
-            .await;
-        assert!(matches!(timed_out, ScannerPublicationProofWait::Cancelled));
-        let budget_ctx = CancellationToken::new();
-        let stopped_budget = ScannerCycleBudget::new(&budget_ctx, ScannerCycleBudgetConfig::default());
-        stopped_budget.cancel_for_runtime();
-        let budget_cancelled = ScannerCyclePublicationPlan::Observational
-            .await_leases(&budget_ctx, &stopped_budget, 505, 21, |_| async { Ok::<_, EcstoreError>(()) })
-            .await;
-        assert!(matches!(budget_cancelled, ScannerPublicationProofWait::Cancelled));
-    }
-    .with_subscriber(subscriber)
-    .await;
-
-    let bytes = output.lock().expect("lease log mutex").clone();
-    let logs = String::from_utf8(bytes).expect("lease logs are UTF-8");
-    let entries: Vec<serde_json::Value> = logs
-        .lines()
-        .map(|line| serde_json::from_str(line).expect("lease logs are JSON"))
-        .collect();
-    for (cycle, epoch, kind, purpose, outcome, parent_cancelled, proof_cancelled) in [
-        (501, 17, "progress_only", "observational", "acquired", false, false),
-        (502, 18, "observational", "observational", "rejected", false, false),
-        (503, 19, "authoritative", "authoritative", "cancelled", true, true),
-        (504, 20, "progress_only", "observational", "timed_out", false, false),
-        (505, 21, "observational", "observational", "cancelled", false, true),
-        (501, 22, "observational", "observational", "acquired", false, false),
-    ] {
-        let started: Vec<_> = entries
-            .iter()
-            .filter_map(|entry| entry.get("fields"))
-            .filter(|fields| {
-                fields["state"] == "publication_proof_started" && fields["cycle"] == cycle && fields["leader_epoch"] == epoch
-            })
-            .collect();
-        assert_eq!(started.len(), 1, "missing or duplicated started event for cycle {cycle}: {logs}");
-        assert_eq!(started[0]["leader_epoch"], epoch);
-        assert_eq!(started[0]["publication_kind"], kind);
-        assert_eq!(started[0]["lease_purpose"], purpose);
-        let finished: Vec<_> = entries
-            .iter()
-            .filter_map(|entry| entry.get("fields"))
-            .filter(|fields| {
-                fields["state"] == "publication_lease_wait_finished"
-                    && fields["cycle"] == cycle
-                    && fields["leader_epoch"] == epoch
-            })
-            .collect();
-        assert_eq!(finished.len(), 1, "missing or duplicated finished event for cycle {cycle}: {logs}");
-        let fields = finished[0];
-        assert_eq!(fields["leader_epoch"], epoch);
-        assert_eq!(fields["publication_kind"], kind);
-        assert_eq!(fields["lease_purpose"], purpose);
-        assert_eq!(fields["outcome"], outcome);
-        assert_eq!(fields["parent_cancelled_at_return"], parent_cancelled);
-        assert_eq!(fields["proof_token_cancelled_at_return"], proof_cancelled);
-    }
-}
-
-#[tokio::test(start_paused = true)]
-async fn publication_lease_retry_logs_identify_interleaved_epochs() {
-    use tracing::instrument::WithSubscriber as _;
-
-    for level in [tracing::Level::WARN, tracing::Level::INFO, tracing::Level::DEBUG] {
-        let output = Arc::new(std::sync::Mutex::new(Vec::new()));
-        let writer = LeaseLogWriter(Arc::clone(&output));
-        let subscriber = tracing_subscriber::fmt()
-            .json()
-            .flatten_event(true)
-            .with_env_filter(format!("off,rustfs::scanner={level}"))
-            .with_current_span(true)
-            .with_writer(move || writer.clone())
-            .with_ansi(false)
-            .without_time()
-            .finish();
-        let _interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
-
-        exercise_interleaved_lease_waits().with_subscriber(subscriber).await;
-
-        let bytes = output.lock().expect("lease log mutex").clone();
-        let logs = String::from_utf8(bytes).expect("lease logs are UTF-8");
-        assert!(!logs.contains("remote-secret"), "untrusted error text must not enter logs");
-        assert!(!logs.contains("forged:9000"), "unstructured error text must not establish a peer");
-        let entries: Vec<serde_json::Value> = logs
-            .lines()
-            .map(|line| serde_json::from_str(line).expect("lease logs are JSON"))
-            .collect();
-        let find_lease_event = |state: &str| {
-            entries
-                .iter()
-                .position(|entry| entry["state"] == state && entry["stage"] == "lease_acquire")
-                .unwrap_or_else(|| panic!("missing {state} at {level}: {logs}"))
-        };
-        let pending = find_lease_event("publication_proof_pending");
-        let rejected = find_lease_event("publication_proof_rejected");
-        assert!(pending < rejected, "waits must interleave: {logs}");
-        for (index, epoch, purpose) in [(pending, 41, "observational"), (rejected, 42, "authoritative")] {
-            assert_eq!(entries[index]["cycle"], 601);
-            assert_eq!(entries[index]["leader_epoch"], epoch, "{logs}");
-            assert_eq!(entries[index]["lease_purpose"], purpose, "{logs}");
-        }
-        assert_eq!(entries[rejected]["peer"], "peer-a:9000", "{logs}");
-        assert_eq!(entries[rejected]["failure_kind"], "peer_unavailable", "{logs}");
-        let ipv6 = entries
-            .iter()
-            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 604)
-            .expect("IPv6 target failure should be logged");
-        assert_eq!(ipv6["peer"], "[2001:db8::1]:9443", "{logs}");
-        let secret_target = entries
-            .iter()
-            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 605)
-            .expect("credential-bearing target failure should be logged");
-        assert!(secret_target["peer"].is_null(), "credential-bearing target must not be logged");
-        assert!(!logs.contains("target-secret"), "target credentials must not enter logs");
-        let local = entries
-            .iter()
-            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 606)
-            .expect("local diagnostic should be logged");
-        assert_eq!(local["reason"], "invalid_proof");
-        assert!(local["reported_reason"].is_null());
-        let reported = entries
-            .iter()
-            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 607)
-            .expect("remote report should be logged");
-        assert!(reported["reason"].is_null());
-        assert_eq!(reported["reported_reason"], "generation_stale");
-        let timeout = entries
-            .iter()
-            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["cycle"] == 608)
-            .expect("channel deadline should be logged");
-        assert_eq!(timeout["reason"], "channel_acquisition_timed_out");
-        assert!(timeout["reported_reason"].is_null());
-        assert!(!logs.contains("diagnostic-secret"), "error text must not enter logs");
-        if level != tracing::Level::WARN {
-            let recovered = find_lease_event("publication_proof_recovered");
-            assert!(rejected < recovered, "waits must interleave: {logs}");
-            assert_eq!(entries[recovered]["leader_epoch"], 41, "{logs}");
-            assert_eq!(entries[recovered]["lease_purpose"], "observational", "{logs}");
-        }
-        if level == tracing::Level::DEBUG {
-            let retry = find_lease_event("publication_proof_retry");
-            let recovered = find_lease_event("publication_proof_recovered");
-            assert!(rejected < retry && retry < recovered, "retry must remain interleaved: {logs}");
-            assert_eq!(entries[retry]["leader_epoch"], 41, "{logs}");
-            assert_eq!(entries[retry]["lease_purpose"], "observational", "{logs}");
-        }
-        let unknown = entries
-            .iter()
-            .find(|entry| entry["state"] == "publication_proof_rejected" && entry["stage"] == "postscan")
-            .expect("activity rejection should be logged");
-        assert!(unknown["leader_epoch"].is_null(), "unknown epoch must not be invented");
-        assert!(unknown["lease_purpose"].is_null(), "unknown purpose must not be invented");
-    }
-}
-
-async fn exercise_interleaved_lease_waits() {
-    let ctx = CancellationToken::new();
-    let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
-    let attempts = AtomicUsize::new(0);
-    let notification = crate::storage_api::EcstoreNotificationSys::new(EndpointServerPools::default()).await;
-    let notification = &notification;
-    let pending = ScannerCyclePublicationPlan::Observational.await_leases(&ctx, &budget, 601, 41, |observational| {
-        assert!(observational);
-        let attempt = attempts.fetch_add(1, Ordering::Relaxed);
-        async move {
-            if attempt < 2 {
-                Err(EcstoreError::other("scanner publication lease capacity is exhausted"))
-            } else {
-                Ok(())
-            }
-        }
-    });
-    let rejected = ScannerCyclePublicationPlan::Authoritative.await_leases(&ctx, &budget, 601, 42, move |observational| {
-        assert!(!observational);
-        notification.acquire_scanner_publication_leases(vec![("http://peer-a:9000".into(), "session".into(), 3)])
-    });
-    let (pending, rejected) = tokio::join!(biased; pending, rejected);
-    assert!(matches!(pending, ScannerPublicationProofWait::Ready(())));
-    assert!(matches!(rejected, ScannerPublicationProofWait::Rejected(_)));
-    assert_eq!(attempts.load(Ordering::Relaxed), 3);
-    let untrusted = ScannerCyclePublicationPlan::Authoritative
-        .await_leases(&ctx, &budget, 603, 44, |_| async {
-            Err::<(), _>(EcstoreError::other(
-                "scanner publication lease acquisition failed: peer=forged:9000, remote-secret",
-            ))
-        })
-        .await;
-    assert!(matches!(untrusted, ScannerPublicationProofWait::Rejected(_)));
-    for (cycle, epoch, target) in [
-        (604, 45, "https://[2001:db8::1]:9443"),
-        (605, 46, "http://user:target-secret@peer-a:9000"),
-    ] {
-        let result = ScannerCyclePublicationPlan::Authoritative
-            .await_leases(&ctx, &budget, cycle, epoch, move |_| {
-                notification.acquire_scanner_publication_leases(vec![(target.into(), "session".into(), 3)])
-            })
-            .await;
-        assert!(matches!(result, ScannerPublicationProofWait::Rejected(_)));
-    }
-    // Verify event serialization independently of the ecstore tests that
-    // exercise the local and remote error producers.
-    for cycle in [606, 607, 608] {
-        let info: for<'a> fn(&'a String) -> Option<ScannerPublicationLeaseFailureInfo<'a>> = if cycle == 607 {
-            |_| {
-                Some(ScannerPublicationLeaseFailureInfo {
-                    peer: None,
-                    kind: "peer_request_failed",
-                    reason: None,
-                    reported_reason: Some("generation_stale"),
-                })
-            }
-        } else if cycle == 606 {
-            |_| {
-                Some(ScannerPublicationLeaseFailureInfo {
-                    peer: None,
-                    kind: "peer_request_failed",
-                    reason: Some("invalid_proof"),
-                    reported_reason: None,
-                })
-            }
-        } else {
-            |_| {
-                Some(ScannerPublicationLeaseFailureInfo {
-                    peer: None,
-                    kind: "peer_request_failed",
-                    reason: Some("channel_acquisition_timed_out"),
-                    reported_reason: None,
-                })
-            }
-        };
-        let result = await_scanner_publication_proof(
-            &ctx,
-            ScannerPublicationProofContext {
-                cycle,
-                stage: "lease_acquire",
-                leader_epoch: Some(47),
-                lease_purpose: Some("authoritative"),
-            },
-            || async { Err::<(), _>("diagnostic-secret".to_owned()) },
-            |_| false,
-            info,
-        )
-        .await;
-        assert!(matches!(result, ScannerPublicationProofWait::Rejected(_)));
-    }
-    let unknown =
-        await_scanner_publication_activity(&ctx, 601, "postscan", || async { Err("protocol mismatch".to_string()) }).await;
-    assert!(unknown.is_err());
-}
-
-#[derive(Clone, Debug, Default)]
-struct CapturedOtelLogs(Arc<std::sync::Mutex<Vec<opentelemetry_sdk::logs::SdkLogRecord>>>);
-
-impl opentelemetry_sdk::logs::LogProcessor for CapturedOtelLogs {
-    fn emit(&self, data: &mut opentelemetry_sdk::logs::SdkLogRecord, _: &opentelemetry::InstrumentationScope) {
-        self.0.lock().expect("OTLP log mutex").push(data.clone());
-    }
-
-    fn force_flush(&self) -> opentelemetry_sdk::error::OTelSdkResult {
-        Ok(())
-    }
-
-    fn shutdown_with_timeout(&self, _: std::time::Duration) -> opentelemetry_sdk::error::OTelSdkResult {
-        Ok(())
-    }
-}
-
-#[tokio::test(start_paused = true)]
-async fn publication_lease_events_include_context_in_otlp_without_span_export() {
-    use opentelemetry::logs::AnyValue;
-    use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
-    use tracing::{Instrument as _, instrument::WithSubscriber as _};
-    use tracing_subscriber::prelude::*;
-
-    for level in [tracing::Level::WARN, tracing::Level::INFO, tracing::Level::DEBUG] {
-        let captured = CapturedOtelLogs::default();
-        let provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder()
-            .with_log_processor(captured.clone())
-            .build();
-        let subscriber = tracing_subscriber::registry()
-            .with(tracing_subscriber::EnvFilter::new(format!("off,rustfs::scanner={level}")))
-            .with(OpenTelemetryTracingBridge::new(&provider));
-        let _interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
-        async {
-            // The bridge is configured like production: no tracing exporter or
-            // span attribute copying. A parent span must not supply these fields.
-            let unrelated = tracing::warn_span!(
-                target: "rustfs::scanner",
-                "unrelated_parent",
-                leader_epoch = 999,
-                lease_purpose = "unrelated"
-            );
-            exercise_interleaved_lease_waits().instrument(unrelated).await;
-        }
-        .with_subscriber(subscriber)
-        .await;
-
-        let records = captured.0.lock().expect("OTLP log mutex").clone();
-        let attribute = |record: &opentelemetry_sdk::logs::SdkLogRecord, name: &str| {
-            record
-                .attributes_iter()
-                .find(|(key, _)| key.as_str() == name)
-                .map(|(_, value)| value.clone())
-        };
-        let find = |state: &'static str, cycle: i64, stage: &'static str| {
-            records
-                .iter()
-                .find(|record| {
-                    attribute(record, "state") == Some(AnyValue::from(state))
-                        && attribute(record, "cycle") == Some(AnyValue::Int(cycle))
-                        && attribute(record, "stage") == Some(AnyValue::from(stage))
-                })
-                .unwrap_or_else(|| panic!("missing {state} at {level}"))
-        };
-        let pending = find("publication_proof_pending", 601, "lease_acquire");
-        let rejected = find("publication_proof_rejected", 601, "lease_acquire");
-        for (record, epoch, purpose) in [(pending, 41, "observational"), (rejected, 42, "authoritative")] {
-            assert_eq!(attribute(record, "leader_epoch"), Some(AnyValue::Int(epoch)));
-            assert_eq!(attribute(record, "lease_purpose"), Some(AnyValue::from(purpose)));
-        }
-        assert_eq!(attribute(rejected, "peer"), Some(AnyValue::from("peer-a:9000")));
-        assert_eq!(attribute(rejected, "failure_kind"), Some(AnyValue::from("peer_unavailable")));
-        let ipv6 = find("publication_proof_rejected", 604, "lease_acquire");
-        assert_eq!(attribute(ipv6, "peer"), Some(AnyValue::from("[2001:db8::1]:9443")));
-        let secret_target = find("publication_proof_rejected", 605, "lease_acquire");
-        assert!(attribute(secret_target, "peer").is_none());
-        let local = find("publication_proof_rejected", 606, "lease_acquire");
-        assert_eq!(attribute(local, "reason"), Some(AnyValue::from("invalid_proof")));
-        assert!(attribute(local, "reported_reason").is_none());
-        let reported = find("publication_proof_rejected", 607, "lease_acquire");
-        assert!(attribute(reported, "reason").is_none());
-        assert_eq!(attribute(reported, "reported_reason"), Some(AnyValue::from("generation_stale")));
-        let timeout = find("publication_proof_rejected", 608, "lease_acquire");
-        assert_eq!(attribute(timeout, "reason"), Some(AnyValue::from("channel_acquisition_timed_out")));
-        assert!(attribute(timeout, "reported_reason").is_none());
-        if level != tracing::Level::WARN {
-            let recovered = find("publication_proof_recovered", 601, "lease_acquire");
-            assert_eq!(attribute(recovered, "leader_epoch"), Some(AnyValue::Int(41)));
-            assert_eq!(attribute(recovered, "lease_purpose"), Some(AnyValue::from("observational")));
-        }
-        if level == tracing::Level::DEBUG {
-            let retry = find("publication_proof_retry", 601, "lease_acquire");
-            assert_eq!(attribute(retry, "leader_epoch"), Some(AnyValue::Int(41)));
-            assert_eq!(attribute(retry, "lease_purpose"), Some(AnyValue::from("observational")));
-        }
-        let unknown = find("publication_proof_rejected", 601, "postscan");
-        assert!(attribute(unknown, "leader_epoch").is_none());
-        assert!(attribute(unknown, "lease_purpose").is_none());
-        let untrusted = find("publication_proof_rejected", 603, "lease_acquire");
-        assert!(attribute(untrusted, "peer").is_none());
-        assert!(!format!("{records:?}").contains("remote-secret"));
-        assert!(!format!("{records:?}").contains("forged:9000"));
-        assert!(!format!("{records:?}").contains("target-secret"));
-        assert!(!format!("{records:?}").contains("diagnostic-secret"));
-    }
-}
-
-#[tokio::test]
-async fn dropped_lease_wait_does_not_report_a_finished_outcome() {
-    use tracing::instrument::WithSubscriber as _;
-
-    let output = Arc::new(std::sync::Mutex::new(Vec::new()));
-    let writer = LeaseLogWriter(Arc::clone(&output));
-    let subscriber = tracing_subscriber::fmt()
-        .json()
-        .with_max_level(tracing::Level::DEBUG)
-        .with_writer(move || writer.clone())
-        .with_ansi(false)
-        .without_time()
-        .finish();
-    let _interest_guard = tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default());
-
-    async {
-        let ctx = CancellationToken::new();
-        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
-        let entered = AtomicBool::new(false);
-        {
-            let proof = ScannerCyclePublicationPlan::Observational.await_leases(&ctx, &budget, 602, 43, |_| {
-                entered.store(true, Ordering::Relaxed);
-                std::future::pending::<Result<(), EcstoreError>>()
-            });
-            tokio::pin!(proof);
-            assert!(futures::poll!(&mut proof).is_pending());
-            assert!(entered.load(Ordering::Relaxed));
-        }
-        assert!(!ctx.is_cancelled());
-        assert!(!budget.token().is_cancelled());
-    }
-    .with_subscriber(subscriber)
-    .await;
-
-    let bytes = output.lock().expect("lease log mutex").clone();
-    let logs = String::from_utf8(bytes).expect("lease logs are UTF-8");
-    let entries: Vec<serde_json::Value> = logs
-        .lines()
-        .map(|line| serde_json::from_str(line).expect("lease logs are JSON"))
-        .collect();
-    let started = entries
-        .iter()
-        .filter(|entry| entry["fields"]["state"] == "publication_proof_started" && entry["fields"]["cycle"] == 602)
-        .count();
-    let finished = entries
-        .iter()
-        .filter(|entry| entry["fields"]["state"] == "publication_lease_wait_finished" && entry["fields"]["cycle"] == 602)
-        .count();
-    assert_eq!(started, 1, "the wait was polled: {logs}");
-    assert_eq!(finished, 0, "a dropped future has no observed outcome: {logs}");
-}
-
-#[test]
-fn publication_lease_retries_local_channel_deadline_and_stops_on_cancel() {
-    const CHILD_MARKER: &str = "RUSTFS_TEST_SCANNER_LEASE_DIAL_RETRY_CHILD";
-    if std::env::var_os(CHILD_MARKER).is_none() {
-        let module = module_path!()
-            .strip_prefix(concat!(env!("CARGO_CRATE_NAME"), "::"))
-            .unwrap_or(module_path!());
-        let name = format!("{module}::publication_lease_retries_local_channel_deadline_and_stops_on_cancel");
-        let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
-            .args(["--exact", &name, "--nocapture"])
-            .env(CHILD_MARKER, "1")
-            .env(rustfs_config::ENV_INTERNODE_CONNECT_TIMEOUT_SECS, "0")
-            .env("NO_PROXY", "127.0.0.1,localhost")
-            .env("no_proxy", "127.0.0.1,localhost")
-            .output()
-            .expect("channel deadline test subprocess");
-        assert!(
-            output.status.success() && String::from_utf8_lossy(&output.stdout).contains("1 passed"),
-            "channel deadline subprocess failed: {} {}",
-            String::from_utf8_lossy(&output.stdout),
-            String::from_utf8_lossy(&output.stderr)
-        );
-        return;
-    }
-
-    let runtime = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .start_paused(true)
-        .build()
-        .expect("test runtime");
-    runtime.block_on(async {
-        use crate::storage_api::scan::EcstorePeerRestClient;
-        use rustfs_utils::net::XHost;
-
-        async fn channel_deadline_error(
-            notification: &crate::storage_api::EcstoreNotificationSys,
-            target: &str,
-        ) -> EcstoreError {
-            let error = match notification
-                .acquire_scanner_publication_leases(vec![(target.to_owned(), "session".into(), 3)])
-                .await
-            {
-                Ok(_) => panic!("the stalled channel cannot grant a lease"),
-                Err(error) => error,
-            };
-            let info = crate::storage_api::scan::scanner_publication_lease_failure_info(&error)
-                .expect("local channel deadline must carry structured diagnostics");
-            assert_eq!(info.kind, "peer_request_failed");
-            assert_eq!(info.reason, Some("channel_acquisition_timed_out"));
-            assert_eq!(info.reported_reason, None);
-            assert!(!scanner_publication_lease_error_is_retryable(&error.to_string()));
-            error
-        }
-
-        // Keep the address unique to this isolated subprocess.
-        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("local listener");
-        let address = listener.local_addr().expect("listener address");
-        let target = format!("https://{address}");
-        let client = EcstorePeerRestClient::new(
-            XHost {
-                name: "127.0.0.1".to_owned(),
-                port: address.port(),
-                is_port_set: true,
-            },
-            target.clone(),
-        );
-        let mut notification = crate::storage_api::EcstoreNotificationSys::new(EndpointServerPools::default()).await;
-        notification.peer_clients.push(Some(client));
-
-        // Block the real channel cache read so the acquisition deadline wins
-        // independently of OS-specific connect and TLS error ordering.
-        let cache_read_blocker = rustfs_common::block_connection_cache_reads_for_test().await;
-
-        let ctx = CancellationToken::new();
-        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
-        let attempts = AtomicUsize::new(0);
-        let started_at = Instant::now();
-        let result = ScannerCyclePublicationPlan::Authoritative
-            .await_leases(&ctx, &budget, 701, 51, |_| {
-                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
-                let target = target.clone();
-                let notification = &notification;
-                async move {
-                    if attempt > 0 {
-                        return Ok(());
-                    }
-                    Err(channel_deadline_error(notification, &target).await)
-                }
-            })
-            .await;
-        assert!(matches!(result, ScannerPublicationProofWait::Ready(())));
-        assert_eq!(attempts.load(Ordering::SeqCst), 2);
-        assert!(started_at.elapsed() >= scanner_publication_proof_retry_delay(1));
-
-        for detail in [
-            "channel_acquisition_timed_out",
-            "scanner publication lease acquisition failed: peer=peer-a:9000, purpose=authoritative: Io error: internode channel acquisition deadline elapsed",
-        ] {
-            let attempts = AtomicUsize::new(0);
-            let result = ScannerCyclePublicationPlan::Authoritative
-                .await_leases(&ctx, &budget, 702, 52, |_| {
-                    attempts.fetch_add(1, Ordering::SeqCst);
-                    async move { Err::<(), _>(EcstoreError::other(detail)) }
-                })
-                .await;
-            assert!(matches!(result, ScannerPublicationProofWait::Rejected(_)));
-            assert_eq!(attempts.load(Ordering::SeqCst), 1, "remote text must not create a local diagnosis");
-        }
-
-        let cancel_ctx = CancellationToken::new();
-        let cancel_budget = ScannerCycleBudget::new(&cancel_ctx, ScannerCycleBudgetConfig::default());
-        let attempts = AtomicUsize::new(0);
-        let started_at = Instant::now();
-        let result = ScannerCyclePublicationPlan::Authoritative
-            .await_leases(&cancel_ctx, &cancel_budget, 703, 53, |_| {
-                let attempt = attempts.fetch_add(1, Ordering::SeqCst);
-                let target = target.clone();
-                let notification = &notification;
-                let cancel_ctx = cancel_ctx.clone();
-                async move {
-                    assert_eq!(attempt, 0, "cancellation must stop the scheduled retry");
-                    let error = channel_deadline_error(notification, &target).await;
-                    tokio::spawn(async move {
-                        tokio::time::sleep(Duration::from_secs(1)).await;
-                        cancel_ctx.cancel();
-                    });
-                    Err::<(), _>(error)
-                }
-            })
-            .await;
-        assert!(matches!(result, ScannerPublicationProofWait::Cancelled));
-        assert_eq!(attempts.load(Ordering::SeqCst), 1);
-        assert!(started_at.elapsed() < scanner_publication_proof_retry_delay(1));
-
-        let progress_ctx = CancellationToken::new();
-        let progress_budget = ScannerCycleBudget::new(&progress_ctx, ScannerCycleBudgetConfig::default());
-        let attempts = AtomicUsize::new(0);
-        let started_at = Instant::now();
-        let result = ScannerCyclePublicationPlan::ProgressOnly
-            .await_leases(&progress_ctx, &progress_budget, 704, 54, |observational| {
-                assert!(observational);
-                attempts.fetch_add(1, Ordering::SeqCst);
-                let target = target.clone();
-                let notification = &notification;
-                async move { Err::<(), _>(channel_deadline_error(notification, &target).await) }
-            })
-            .await;
-        assert!(matches!(result, ScannerPublicationProofWait::Cancelled));
-        assert_eq!(attempts.load(Ordering::SeqCst), 1);
-        assert!(started_at.elapsed() < scanner_publication_proof_retry_delay(1));
-        drop(cache_read_blocker);
-        drop(listener);
-    });
-}
-
 #[tokio::test]
 #[serial]
 async fn exhausted_budget_still_proves_peer_movement_without_namespace_authority() {
@@ -654,7 +33,7 @@ async fn exhausted_budget_still_proves_peer_movement_without_namespace_authority
     let calls = AtomicUsize::new(0);
 
     let proof = plan
-        .await_leases(&ctx, &budget, 352, 19, |observational| {
+        .await_leases(&ctx, &budget, 352, |observational| {
             calls.fetch_add(1, Ordering::Relaxed);
             store.acquire_scanner_publication_lease_with_purpose(generation, Duration::from_secs(60), observational)
         })
@@ -678,7 +57,7 @@ async fn exhausted_budget_still_proves_peer_movement_without_namespace_authority
     assert!(store.release_scanner_publication_lease(token).await);
 
     let stale = plan
-        .await_leases(&ctx, &budget, 352, 19, |observational| {
+        .await_leases(&ctx, &budget, 352, |observational| {
             store.acquire_scanner_publication_lease_with_purpose(generation + 1, Duration::from_secs(60), observational)
         })
         .await;
@@ -692,7 +71,7 @@ async fn exhausted_budget_still_proves_peer_movement_without_namespace_authority
     assert!(store.scanner_data_movement_pause_status().await.paused);
     let current_generation = store.scanner_data_movement_generation();
     let moving = plan
-        .await_leases(&ctx, &budget, 352, 19, |observational| {
+        .await_leases(&ctx, &budget, 352, |observational| {
             store.acquire_scanner_publication_lease_with_purpose(current_generation, Duration::from_secs(60), observational)
         })
         .await;
@@ -722,21 +101,15 @@ async fn emitted_updates_keep_budget_cancellation_and_publication_purpose() {
         let ctx = CancellationToken::new();
         let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
         let ready = plan
-            .await_leases(
-                &ctx,
-                &budget,
-                352,
-                19,
-                |observational| async move { Ok::<_, EcstoreError>(observational) },
-            )
+            .await_leases(&ctx, &budget, 352, |observational| async move { Ok::<_, String>(observational) })
             .await;
         assert!(matches!(ready, ScannerPublicationProofWait::Ready(observational) if observational == plan.is_observational()));
         budget.cancel_for_runtime();
         let calls = AtomicUsize::new(0);
         let cancelled = plan
-            .await_leases(&ctx, &budget, 352, 19, |_| {
+            .await_leases(&ctx, &budget, 352, |_| {
                 calls.fetch_add(1, Ordering::Relaxed);
-                async { Ok::<(), EcstoreError>(()) }
+                async { Ok::<(), String>(()) }
             })
             .await;
         assert!(matches!(cancelled, ScannerPublicationProofWait::Cancelled));
@@ -751,9 +124,9 @@ async fn progress_proof_honors_parent_cancellation() {
     ctx.cancel();
     let calls = AtomicUsize::new(0);
     let proof = ScannerCyclePublicationPlan::ProgressOnly
-        .await_leases(&ctx, &budget, 352, 19, |_| {
+        .await_leases(&ctx, &budget, 352, |_| {
             calls.fetch_add(1, Ordering::Relaxed);
-            async { Ok::<_, EcstoreError>(()) }
+            async { Ok::<_, String>(()) }
         })
         .await;
     assert!(matches!(proof, ScannerPublicationProofWait::Cancelled));
@@ -776,13 +149,13 @@ async fn progress_proof_is_bounded_without_a_runtime_budget() {
         let calls = AtomicUsize::new(0);
         let started = Instant::now();
         let proof = ScannerCyclePublicationPlan::ProgressOnly
-            .await_leases(&ctx, &budget, 352, 19, |_| {
+            .await_leases(&ctx, &budget, 352, |_| {
                 calls.fetch_add(1, Ordering::Relaxed);
                 async move {
                     if stalled {
                         std::future::pending::<()>().await;
                     }
-                    Err::<(), _>(EcstoreError::other("scanner publication lease capacity is exhausted"))
+                    Err::<(), _>("scanner publication lease capacity is exhausted")
                 }
             })
             .await;

crates/scanner/src/storage_api.rs

@@ -109,6 +109,7 @@ pub(crate) use rustfs_ecstore::api::layout::{
     SetupType as EcstoreSetupType,
 };
 pub(crate) use rustfs_ecstore::api::notification::NotificationSys as EcstoreNotificationSys;
+pub(crate) use rustfs_ecstore::api::notification::scanner_peer_transport_error_message_is_retryable;
 pub(crate) use rustfs_ecstore::api::object::{
     SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitState,
 };
@@ -364,12 +365,6 @@ pub(crate) mod scan {
         SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE,
         SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES,
     };
-    pub(crate) use rustfs_ecstore::api::notification::{
-        ScannerPublicationLeaseFailureInfo, scanner_peer_transport_error_message_is_retryable,
-        scanner_publication_lease_failure_info,
-    };
-    #[cfg(test)]
-    pub(crate) use rustfs_ecstore::api::rpc::PeerRestClient as EcstorePeerRestClient;
 }
 
 pub(crate) mod scanner_io {

docs/architecture/erasure-coding.md

@@ -5,7 +5,7 @@
 
 Status: normative. This document is the source of truth for how RustFS erasure-codes, stores, reads, reconstructs, and heals user data, and for the on-disk / on-wire compatibility contract that every future change must preserve. It governs the highest-risk code in the system: a regression here can silently corrupt or lose all user data, or make existing (and MinIO-migrated) objects permanently unreadable.
 
-Erasure coding, quorum/heal, and metadata/on-disk formats are **High-risk** per [AGENTS.md](../../AGENTS.md) ("Broad or High-Risk Changes"). Any behavior-affecting change to code this document governs requires adversarial review with the `adversarial-validation` skill and, for anything touching decode or the on-disk format, a regression test against real on-disk and MinIO-migrated samples before merge.
+Erasure coding, quorum/heal, and metadata/on-disk formats are **High-risk** per [AGENTS.md](../../AGENTS.md). Any behavior-affecting change to code this document governs requires adversarial review and, for anything touching decode or the on-disk format, a regression test against real on-disk and MinIO-migrated samples before merge.
 
 This document describes the algorithm as implemented on `main`. The *invariant* stated is always the rule the code must satisfy; where the code enforces it, the enforcing symbol is cited.
 
@@ -313,7 +313,7 @@ Small objects store their payload inline after the container CRC ([filemeta_inli
 ## 8. Read path and read quorum
 
 - **INVARIANT — read quorum = `data_blocks`.** `object_quorum_from_meta` returns `(read_quorum = data_blocks, write_quorum)` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)); `parity_blocks = common_parity(...)` is the parity value held by the most disks that still reaches its own read quorum. When `default_parity_count == 0`, read = write = all shards.
-- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. Reads select at read quorum through `select_valid_fileinfo`, the same rule heal uses and MinIO's `getObjectFileInfo`: the most common mod_time wins and a tie goes to the newer one, so a write that failed after reaching exactly `data_blocks` copies is served as listing shows it, instead of `ErasureReadQuorum` until heal runs. Two identities tied at the selected mod_time (a data = parity set split in half, such as a metadata update that failed after reaching half the drives) resolve to the one on the lowest-index disk, in reads and heal alike, so a retried metadata write can repair the split. A selected version with local shards also needs as many copies as its own `data_blocks`, since delete markers and empty versions can lower the common parity's read quorum. A metadata update whose target disks cannot reach write quorum is refused before any write. A version replaced by a committed write keeps fewer than `data_blocks` copies and can never be selected again; the newer side of a failed write can be served and then lose to the older side if one of its disks fails before heal converges the set.
+- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. S3 reads select at read quorum through `select_valid_fileinfo`, the same rule heal uses and MinIO's `getObjectFileInfo`: the most common mod_time wins and a tie goes to the newer one, so a write that failed after reaching exactly `data_blocks` copies is served as listing shows it, instead of `ErasureReadQuorum` until heal runs. Two identities tied at the selected mod_time (a data = parity set split in half, such as a metadata update that failed after reaching half the drives) resolve to the one on the lowest-index disk, in reads and heal alike, so a retried metadata write can repair the split. A selected version with local shards also needs as many copies as its own `data_blocks`, since delete markers and empty versions can lower the common parity's read quorum. A metadata update whose target disks cannot reach write quorum is refused before any write. A version replaced by a committed write keeps fewer than `data_blocks` copies and can never be selected again; the newer side of a failed write can be served and then lose to the older side if one of its disks fails before heal converges the set. Reads in the system bucket (`is_meta_bucketname`) keep the commit-proof rule in `select_committed_fileinfo`: when enough copies answer to prove write quorum, the latest version must hold it, and a degraded read never serves a newer version below write quorum. Pool metadata's PREPARE/COMMIT recovery and identity epochs rely on an ordinary read implying a committed record, so a split stays unreadable until native recovery resolves it. When every drive answered and an S3 latest read still cannot select a version, `read_latest_past_dangling_versions` walks the version stack newest first through exact-version reads, as MinIO's `pickLatestQuorumFilesInfo` resolves stacks: it serves the readable version with the newest selected mod_time, skips versions whose exact read proves them dangling (the NoSuchVersion an explicit GET reports) and queues heal for them, and keeps the read-quorum error on any other outcome. A failed versioned PUT that left the new version on fewer drives than its data blocks therefore no longer hides the committed version below it; ListObjectVersions can still show the dangling version as latest until heal removes it. Metadata updates of the latest version (object tagging, `put_object_metadata`) and the CopyObject source read resolve a failed selection through the same walk; when the walk finds no readable version, tagging keeps the read-quorum error while `put_object_metadata` and CopyObject keep their existing `delete_if_dangling` cleanup. Replication status write-back never walks. The update targets only the served version's disks and is refused before writing when they cannot reach its write quorum, including a version-only copy.
 - **INVARIANT — decode needs ≥ `data_blocks` shards.** The stripe reader requires `available_shards ≥ data_shards`; below that the read fails closed with a read-quorum error (never silent truncation) ([set_disk/read.rs](../../crates/ecstore/src/set_disk/read.rs), [set_disk/shard_source.rs](../../crates/ecstore/src/set_disk/shard_source.rs)). Codec geometry is validated at construction: `Erasure::try_new` / `try_new_with_options` ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)) return `ErasureConstructionError` for `data_shards == 0`, `block_size == 0`, shard-count overflow, or an unsupported shard configuration, so a read never reaches a `block_size` / `data_shards` division with invalid geometry (§13).
 - If `available ≥ data_blocks` but some shards are missing, the read is served **and** a background read-repair heal is enqueued.
 - **INVARIANT — cross-stripe read verification.** When a data shard is missing and `available > data_blocks`, reconstruction regenerates parity and compares it to the surviving parity; a mismatch is `InvalidData "inconsistent read source shards"` (backlog#832), catching corruption that passed per-shard bitrot but disagrees across the stripe ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)).
@@ -400,7 +400,7 @@ Decode tolerance
 
 ## 13. Change procedure and guardrails
 
-- **Risk tier.** All of the above is High-risk ([AGENTS.md](../../AGENTS.md) "Broad or High-Risk Changes"). Any behavior-affecting change requires adversarial review with the `adversarial-validation` skill.
+- **Risk tier.** All of the above is High-risk ([AGENTS.md](../../AGENTS.md)). Any behavior-affecting change requires adversarial review.
 - **Keep this document in sync.** A change to any governed behavior, formula, format field, or invariant must update this spec in the same PR; renaming a cited symbol must update its reference here. The spec is normative and is the checklist the next change is reviewed against, so drift is a correctness defect. References are symbol-based (not line numbers) specifically so ordinary refactors do not invalidate them — but semantic changes still must.
 - **Adding an on-disk field** must be additive: new msgpack key or a `minor`/`meta_ver` bump with a read path for the old value; keep decoders skipping unknown keys; write both internal-key prefixes; never repurpose or reorder existing keys or header array positions.
 - **Never make a decode boundary stricter** than what §11 allows without (a) proving no legitimate older-RustFS or MinIO-migrated shape is rejected, and (b) a regression test against real on-disk and MinIO fixtures. New validation belongs at the trust boundary and must fail *open to a tolerant default*, not closed to `FileCorrupt`, for anything recoverable. (Concretely: rejecting a negative `actual_size`, or hard-failing a non-16-byte `transitioned-versionID`, breaks existing data — see §11.)

docs/operations/hotpath-warp-ab-runbook.md

@@ -143,6 +143,39 @@ Rules that override the table: a candidate result is actionable only when the `A
 
 ### CPU and memory evidence
 
+#### Local shard batch attribution
+
+The existing GET-stage metric gate also controls local batch attribution. This
+path performs buffered positioned reads, even if O_DIRECT or io_uring is enabled
+for another backend. Keep the gate, exporter, and build features identical
+between compared binaries; calibrate instrumentation overhead separately.
+
+| Metric suffix after `rustfs_io_get_object_local_batch_` | Meaning |
+| --- | --- |
+| `decisions_total` | Reader-setup decisions, labelled by a fixed decision enum; this is not an S3 request counter. |
+| `prepare_seconds` | Eligibility and path preparation, including pre-worker path checks; absent when preparation was skipped. |
+| `stage_seconds` | Queue wait, whole-worker execution, and per-shard metadata/open/read durations, labelled by stage and operation outcome. |
+| `shards` / `shards_total` | Batch size and completed shard outcomes. |
+| `requested_bytes_total` / `read_bytes_total` | Requested bytes and bytes actually returned by positioned reads, split by shard outcome. |
+
+Worker handles are captured before dispatch and remain owned by the worker, so
+completion samples can arrive after the caller stops waiting. A stage that did
+not execute contributes no sample. Queue wait ends when the worker starts;
+execution includes allocation and instrumentation, while read timing excludes
+buffer allocation. These nested and per-shard durations cannot be added to
+request latency, and stage p99 values cannot be summed. Match request-boundary
+samples and successful-request counters to the same workload window when
+normalizing costs. Keep failed requests and partial-shard failures visible.
+Inline GET and HEAD can bypass this reader-setup entry entirely; absence of a
+decision sample does not imply a failed request. The two local duration metrics
+use dedicated 0.1-microsecond-to-10-second histogram boundaries in the OTLP view;
+request-duration histogram boundaries are separate.
+
+For a bounded, four-shard instrumentation microbenchmark, run
+`cargo bench -p rustfs-io-metrics --bench metrics_pipeline -- local_batch_attribution_4_shards_debug_recorder`.
+It performs no disk I/O and uses a debugging recorder, so its result calibrates
+the sampling code rather than production exporter overhead or S3 latency.
+
 Warp output says whether throughput or latency changed; host telemetry says why. Collect it for the whole run and stop the collectors after the script exits.
 
 | Tool | Command | Answers |

docs/postmortems/2026-09-03-replication-checksum-default-regression.md

@@ -30,7 +30,7 @@ The same thing was missing at every layer: an inventory of what remote targets m
 
 - **Change design.** A default that applies to every target was changed to satisfy one target class. The PR text considered checksums only for objects that already carried one ("object-level `x-amz-checksum-*` forwarding is unaffected") and never asked the inverse question: which target-side rules depend on a checksum being present.
 - **Tests.** The new unit test asserted the fix ("no `x-amz-trailer`"), not the contract ("the target accepted the PUT"). The fake target modeled neither failure mode: it did not store `aws-chunked` verbatim and it did not demand a checksum on locked PUTs. `replication-check` has an ObjectLock phase, but its probe PUT carries no retention headers, so it would have reported OK against the customer's target.
-- **Review.** The compatibility lens (`.agents/skills/adversarial-validation/references/compatibility.md`) listed only inbound and on-disk concerns: `xl.meta`, proto, MinIO fixtures. It contained no line about outbound target behavior, so the lens passed while blind to the class.
+- **Review.** The compatibility review guidance listed only inbound and on-disk concerns: `xl.meta`, proto, MinIO fixtures. It contained no line about outbound target behavior, so the review passed while blind to the class.
 - **Release.** Two new environment escape hatches shipped undocumented. The customer's second question in rustfs#7082 was whether a configuration option exists; it did, and it was unfindable.
 
 ## Corrective actions
@@ -39,7 +39,7 @@ The same thing was missing at every layer: an inventory of what remote targets m
 | --- | --- |
 | Outbound target matrix e2e: every target failure mode times every object shape, with an explicit expectation table. Known-red cells are pinned to an open issue and fail loudly when they start passing. | `crates/e2e_test/src/replication_target_matrix_test.rs` |
 | Fake target gains the three failure modes a real fleet has shown: rejects `aws-chunked` uploads, requires a checksum on Object Lock PUTs (and verifies `Content-MD5`), mints its own version ids. | `crates/e2e_test/src/fake_s3_target/mod.rs` |
-| Compatibility lens gains an outbound target section. | `.agents/skills/adversarial-validation/references/compatibility.md` |
+| Review policy covers outbound target behavior. | `.agents/references/adversarial-validation.md` |
 | `AGENTS.md` names outbound client defaults as high risk and requires the matrix plus documented escape hatches. | `AGENTS.md`, Adversarial Validation |
 | The two escape hatches from rustfs#6895 are documented. | `docs/operations/replication-outbound-transport.md` |
 | Product fix for rustfs#7082: a locked PUT carries `Content-MD5` derived from the source ETag when that ETag is the MD5 of the wire bytes, and an SDK CRC32 checksum otherwise; the two matrix cells flipped to `Completed` in the same change. | `crates/ecstore/src/bucket/bucket_target_sys.rs` (`object_lock_put_integrity_for`), `crates/replication/src/object.rs` (`object_lock_put_integrity`) |

fuzz/Cargo.toml

@@ -19,6 +19,18 @@ rustfs-security-governance = { path = "../crates/security-governance" }
 rustfs-utils = { path = "../crates/utils", features = ["compress", "path"] }
 serde_json = "1"
 
+[patch.crates-io]
+# Standalone workspaces must repeat the root workspace's fixed SigV4a backend.
+aws-sigv4 = { git = "https://github.com/smithy-lang/smithy-rs", rev = "5c1c4bca2ea7af97496e077c9f1a718eaac64bf8" }
+
+[patch."https://github.com/smithy-lang/smithy-rs"]
+# Keep the signer's path dependencies on the SDK's existing registry types.
+aws-credential-types = { version = "=1.3.1" }
+aws-smithy-eventstream = { version = "=0.61.6" }
+aws-smithy-http = { version = "=0.64.3" }
+aws-smithy-runtime-api = { version = "=1.19.2" }
+aws-smithy-types = { version = "=1.9.1" }
+
 [[bin]]
 name = "bucket_validation"
 path = "fuzz_targets/bucket_validation.rs"

scripts/check_embedded_secrets.sh

@@ -3,9 +3,8 @@ set -euo pipefail
 
 # Guard: no private key material and no long-lived provider credential may
 # exist as a literal anywhere in the repository — see AGENTS.md "Security
-# Baseline" ("Never commit secrets, credentials, or key material") and
-# .agents/skills/security-advisory-lessons ("Do not ship hard-coded shared
-# tokens, HMAC secrets, private keys, or production test keys").
+# Baseline" ("Never commit secrets, credentials, or key material"). This includes
+# shared tokens, HMAC secrets, private keys, and production test keys.
 #
 # This replaces the unit test `test_source_does_not_embed_private_key` that
 # used to live in crates/crypto/src/license_token.rs (rustfs/backlog#1884).

scripts/error-other-format-baseline.txt

@@ -45,9 +45,9 @@
 3|crates/ecstore/src/runtime/sources.rs
 4|crates/ecstore/src/services/batch_processor.rs
 12|crates/ecstore/src/services/notification_sys.rs
-16|crates/ecstore/src/services/rebalance/control.rs
+15|crates/ecstore/src/services/rebalance/control.rs
 1|crates/ecstore/src/services/rebalance/entry.rs
-8|crates/ecstore/src/services/rebalance/meta.rs
+6|crates/ecstore/src/services/rebalance/meta.rs
 8|crates/ecstore/src/services/rebalance/runtime.rs
 17|crates/ecstore/src/services/rebalance/worker.rs
 33|crates/ecstore/src/services/tier/tier.rs