Merge branch 'main' into overtrue/scanner-peer-bucket-certification

* fix(s3select): match request enum values case-insensitively

ExpressionType, CompressionType, CSV FileHeaderInfo, JSON Type and CSV
QuoteFields were compared exactly, so lowercase values such as
`lines`, `document` or `gzip` were rejected. MinIO accepts any case.

Match them ASCII case-insensitively in request validation and rewrite
them to the canonical uppercase spelling, so every later comparison
keeps working unchanged. Unknown values keep their existing errors.

* fix(s3select): keep source key order for JSON LINES input

Arrow's JSON schema inference walks each record as a serde_json::Map,
which sorts keys unless serde_json's process-wide preserve_order
feature is on. JSON LINES `SELECT *` therefore returned keys sorted,
and CSV output columns were permuted.

Infer the LINES schema from the same object bytes and record limit as
DataFusion's JsonFormat, keep Arrow's types, and reorder fields
(nested structs included) by first-seen key order. Enabling
preserve_order globally was rejected: it changes map ordering for the
whole binary.

Drop the test-only preserve_order dependency so the crate's tests run
with production features; the DOCUMENT inference reference now
compares against Arrow's types in source order.

* fix(s3select): name nested path projections after their last key

`SELECT s.obj0.id` returned the key `s.obj0[id]`, DataFusion's display
name for the field access. AWS names an unaliased path projection after
its last key, e.g. `{"id": ...}`.

Alias such projections to their last key when that name is unique among
the output and ORDER BY column names. Queries with a wildcard keep
today's names, since the expanded columns could collide.

* fix(s3select): preserve names beside computed projections

* fix(s3select): preserve names for ambiguous path projections

* fix(s3select): name path projections from the planned schema

Path projections that the SQL rewrite leaves unaliased are now named
after their last key once DataFusion has planned the query. A path takes
its key only if no other output column keeps or receives that name. A
colliding rename is dropped and the check repeats, so the final names
are unique whenever DataFusion's own names are.

The SQL rewrite is unchanged, so GROUP BY can still refer to the names
it assigns and every query that planned before still plans.

Review cases:
- s.obj0.id, s.obj1.id, s.obj2."s.obj0[id]" keeps the three generated
  names: the third key would duplicate the first retained name.
- s.obj."lower(s.id)", LOWER(s.id), in either order, keeps the generated
  path name: its key is the computed column's name.

Paths beside computed columns, beside duplicated keys, and in
parentheses now take their key, as AWS S3 Select names them. Paths stay
unnamed when an ORDER BY expression is not projected, as before.

---------

Co-authored-by: Hauser <housemecn@gmail.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>
Co-authored-by: GatewayJ <835269233@qq.com>

.config/e2e-distributed-selection.txt

@@ -1,2 +1,2 @@
-sha256-linux=eb33067d90ee66c0a97e8f3658e46c152e47f897d2fbfdc6cb5ad93cfa55414b
-sha256-darwin=eb33067d90ee66c0a97e8f3658e46c152e47f897d2fbfdc6cb5ad93cfa55414b
+sha256-linux=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4
+sha256-darwin=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4

crates/config/src/constants/env.rs

@@ -30,6 +30,16 @@ pub const COMMENT_KEY: &str = "comment";
 pub const ENV_AUDIT_ENABLE: &str = "RUSTFS_AUDIT_ENABLE";
 /// Global switch for enabling the notify module.
 pub const ENV_NOTIFY_ENABLE: &str = "RUSTFS_NOTIFY_ENABLE";
+/// Maximum audit entries per second for authorization denials of authenticated callers.
+/// `0` disables denial auditing for that class of caller.
+pub const ENV_AUDIT_DENIAL_RATE: &str = "RUSTFS_AUDIT_DENIAL_RATE";
+/// Maximum audit entries per second for authorization denials of anonymous callers.
+/// `0` disables denial auditing for that class of caller.
+pub const ENV_AUDIT_DENIAL_ANONYMOUS_RATE: &str = "RUSTFS_AUDIT_DENIAL_ANONYMOUS_RATE";
+/// Default denial audit rate for authenticated callers (entries per second).
+pub const DEFAULT_AUDIT_DENIAL_RATE: u64 = 50;
+/// Default denial audit rate for anonymous callers (entries per second).
+pub const DEFAULT_AUDIT_DENIAL_ANONYMOUS_RATE: u64 = 5;
 /// Default global audit switch (disabled by default).
 pub const DEFAULT_AUDIT_ENABLE: bool = false;
 /// Default global notify switch (disabled by default).
@@ -328,5 +338,7 @@ mod tests {
     fn test_global_audit_notify_switch_constants() {
         assert_eq!(ENV_AUDIT_ENABLE, "RUSTFS_AUDIT_ENABLE");
         assert_eq!(ENV_NOTIFY_ENABLE, "RUSTFS_NOTIFY_ENABLE");
+        assert_eq!(ENV_AUDIT_DENIAL_RATE, "RUSTFS_AUDIT_DENIAL_RATE");
+        assert_eq!(ENV_AUDIT_DENIAL_ANONYMOUS_RATE, "RUSTFS_AUDIT_DENIAL_ANONYMOUS_RATE");
     }
 }

crates/e2e_test/Cargo.toml

@@ -25,6 +25,7 @@ workspace = true
 
 [features]
 default = []
+scanner-peer-mixed-version = []
 hotpath = [
     "hotpath/hotpath",
     "hotpath/tokio",

crates/e2e_test/README.md

@@ -64,6 +64,25 @@ Each build/run holds an exclusive `rustfs.e2e.lock` marker beside the binary; co
 
 The protocols suite has its own fixed-port and single-worker contract in [`src/protocols/README.md`](src/protocols/README.md). Use its command under [Troubleshooting](#troubleshooting).
 
+### Scanner mixed-version gate
+
+The Linux N/O scanner gate needs a separately qualified `33482bb` server, which supports scoped ACKs. Set `RUSTFS_SCANNER_COMPAT_OLD_BINARY` to that executable and `RUSTFS_SCANNER_COMPAT_OLD_SHA256` to its independently verified digest. The release binary used by ordinary upgrade CI is a different baseline. Missing or mismatched inputs fail the test.
+
+Enable the harness-only `scanner-peer-mixed-version` feature explicitly; keep the server wrapper features identical between build and run:
+
+```bash
+python3 scripts/e2e_binary.py build --features e2e-test-hooks
+python3 scripts/e2e_binary.py run --features e2e-test-hooks -- \
+  cargo nextest run --profile e2e-distributed -p e2e_test \
+  --features scanner-peer-mixed-version --ignore-default-filter \
+  -E 'test(=distributed::scanner_peer_bucket_test::scanner_peer_bucket_new_coordinator_certifies_old_peers_and_scopes_next_cycle)' \
+  --test-threads 1 --retries 0 --no-tests=fail
+```
+
+The separate `scanner_peer_bucket_new_coordinator_certifies_old_worker_peers_and_scopes_next_cycle` case uses the same feature and qualified old binary inputs. Select its exact name in the command above. It keeps the original N process enabled and starts the three official O replacements with `RUSTFS_SCANNER_ENABLED=false`: their autonomous scanner loops are disabled, while real S3 dirty tracking, authenticated snapshots, publication leases, scoped ACKs, and N-dispatched RemoteScanner workers remain active. It records each node's actual scanner status and retains the same identity, durability, root, ACK, data-conservation, and physical-walk requirements. This fixed-role case does not validate autonomous O scanning or all-enabled election behavior. The original all-enabled N/O test remains separate and has not passed acceptance; a fixed-role result cannot satisfy that gate.
+
+For this fixed-role case, the `e2e-test-hooks` server also records the N client's complete bounded worker inventory. Requests are observed before opening the normal stream and paired only after authenticated source/plan validation and terminal EOF. Closed manifests and final health receipts must bind both cycles to their actual candidate/root; the next cycle needs exactly four sets of hot-only execution and at least one positive-progress worker mapped to a measured official O process. A zero-progress Current response, missing/extra request, unhealthy recorder, or missing physical coverage fails the gate. The recorder does not add O hooks or change any protocol or timeout. Final health reads check immutable closed-cycle evidence, not current N liveness.
+
 ### `#[ignore]` semantics
 
 Ignored tests are excluded from the default `cargo nextest run` pass because

crates/e2e_test/src/distributed/mod.rs

@@ -39,6 +39,8 @@ mod s3_during_data_movement_test;
 mod scanner_heal_topology_fixture;
 mod scanner_heal_topology_test;
 mod scanner_heal_topology_tier_test;
+mod scanner_peer_bucket_fixture;
+mod scanner_peer_bucket_test;
 mod site_replication_test;
 mod table_catalog_test;
 mod upgrade_test;

crates/e2e_test/src/rpc_fault_proxy.rs

@@ -28,7 +28,8 @@ use hyper_util::{
 use prost::Message as _;
 use rustfs_protos::proto_gen::node_service::{
     BatchGenerallyLockRequest, DeleteRequest, GenerallyLockRequest, PreparePartTransactionResponse, ReadVersionRequest,
-    RecoverMultipartSourceRequest, RenameDataResponse, SettlePartTransactionRequest, SettlePartTransactionResponse,
+    RecoverMultipartSourceRequest, RenameDataResponse, ScannerScopedDirtyUsageAckRequest, SettlePartTransactionRequest,
+    SettlePartTransactionResponse,
 };
 use std::{
     collections::HashSet,
@@ -49,6 +50,9 @@ use tokio::{
 type ProxyBody = BoxBody<Bytes, io::Error>;
 type ProxyClient = Client<HttpConnector, ProxyBody>;
 
+mod scoped_ack;
+pub(crate) use scoped_ack::{ScopedAckFault, ScopedAckQueue, ScopedAckScope};
+
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum Disposition {
     Hold,
@@ -121,7 +125,8 @@ impl RpcPause {
         self.state.blocked.load(Ordering::SeqCst)
     }
 
-    /// Selected RPCs with successful HTTP/gRPC headers, excluding discarded packets.
+    /// Selected RPCs with successful HTTP/gRPC response headers, excluding
+    /// discarded packets. This does not prove a successful gRPC trailer or ACK apply.
     pub(crate) fn forwarded(&self) -> usize {
         self.state.forwarded.load(Ordering::SeqCst)
     }
@@ -233,6 +238,7 @@ pub(crate) struct RpcFaultProxy {
     address: SocketAddr,
     slot: PauseSlot,
     lock_slot: PauseSlot,
+    scoped_ack: scoped_ack::QueueSlot,
     task: JoinHandle<()>,
     late_requests: LateRequests,
 }
@@ -245,6 +251,8 @@ impl RpcFaultProxy {
         let active = slot.clone();
         let lock_slot: PauseSlot = Arc::new(Mutex::new(None));
         let active_lock = lock_slot.clone();
+        let scoped_ack = scoped_ack::empty_slot();
+        let active_ack = scoped_ack.clone();
         let late_requests: LateRequests = Arc::new(Mutex::new(Some(Vec::new())));
         let late = late_requests.clone();
         let task = tokio::spawn(async move {
@@ -259,6 +267,7 @@ impl RpcFaultProxy {
                         let h2 = h2.clone();
                         let active = active.clone();
                         let active_lock = active_lock.clone();
+                        let active_ack = active_ack.clone();
                         let late = late.clone();
                         connections.spawn(async move {
                             let service = service_fn(move |request: Request<Incoming>| {
@@ -272,6 +281,7 @@ impl RpcFaultProxy {
                                     None
                                 };
                                 let pause = owner_pause.or_else(|| active.lock().expect("RPC pause slot").clone());
+                                let ack = active_ack.lock().expect("scoped ACK queue slot").clone();
                                 let retain = pause.as_ref().is_some_and(|pause| {
                                     pause.retain_after_disconnect
                                         && request.uri().path().strip_prefix("/node_service.NodeService/")
@@ -280,11 +290,11 @@ impl RpcFaultProxy {
                                 let late = late.clone();
                                 async move {
                                     if !retain {
-                                        return forward_request(request, client, target, pause).await;
+                                        return forward_request(request, client, target, pause, ack).await;
                                     }
                                     // This opt-in fault owns the packet after client cancellation.
                                     // Proxy teardown still aborts every retained request.
-                                    let task = tokio::spawn(forward_request(request, client, target, pause));
+                                    let task = tokio::spawn(forward_request(request, client, target, pause, ack));
                                     {
                                         let mut requests = late.lock().expect("late RPC ownership");
                                         if let Some(requests) = requests.as_mut() {
@@ -309,6 +319,7 @@ impl RpcFaultProxy {
             address,
             slot,
             lock_slot,
+            scoped_ack,
             task,
             late_requests,
         })
@@ -318,6 +329,11 @@ impl RpcFaultProxy {
         self.address
     }
 
+    /// Retain only genuine scoped ACKs for one receiver/session and bucket set.
+    pub(crate) fn scoped_ack_queue(&self, scope: ScopedAckScope, capacity: usize) -> io::Result<ScopedAckQueue> {
+        ScopedAckQueue::arm(self.scoped_ack.clone(), scope, capacity)
+    }
+
     /// Pause at most limit requests to the named unary methods; other streams
     /// on the same HTTP/2 connection continue forwarding.
     pub(crate) fn pause(&self, methods: &[&'static str], limit: usize, scope: Option<&str>) -> io::Result<RpcPause> {
@@ -469,6 +485,9 @@ fn unary_payload(body: &[u8]) -> io::Result<&[u8]> {
 }
 
 fn scope_matches(method: &str, body: &[u8], scope: &[u8]) -> io::Result<bool> {
+    if method == "ScannerScopedDirtyUsageAck" {
+        return scoped_ack_matches(body, Some(scope));
+    }
     if method == "RecoverMultipartSource" {
         let request = RecoverMultipartSourceRequest::decode(unary_payload(body)?).map_err(io::Error::other)?;
         if request.protocol_version != u32::from(MULTIPART_SOURCE_PROTOCOL_VERSION) || request.path.as_bytes() != scope {
@@ -525,7 +544,9 @@ fn scoped_lock_owners(method: &str, body: &[u8], scope: &[u8]) -> io::Result<Vec
     let args = match method {
         "Lock" => vec![GenerallyLockRequest::decode(payload).map_err(io::Error::other)?.args],
         "LockBatch" => BatchGenerallyLockRequest::decode(payload).map_err(io::Error::other)?.args,
-        _ => return Err(io::Error::new(io::ErrorKind::InvalidInput, "lock observer requires a lock method")),
+        _ => {
+            return Err(io::Error::new(io::ErrorKind::InvalidInput, "lock observer requires a lock method"));
+        }
     };
     args.into_iter()
         .filter(|args| args.as_bytes().windows(scope.len()).any(|window| window == scope))
@@ -540,18 +561,39 @@ fn scoped_lock_owners(method: &str, body: &[u8], scope: &[u8]) -> io::Result<Vec
         .collect()
 }
 
+fn scoped_ack_matches(body: &[u8], scope: Option<&[u8]>) -> io::Result<bool> {
+    let request = ScannerScopedDirtyUsageAckRequest::decode(unary_payload(body)?).map_err(io::Error::other)?;
+    Ok(!request.probe_only && scope.is_none_or(|scope| request.entries.iter().any(|entry| entry.bucket.as_bytes() == scope)))
+}
+
 async fn forward_request(
     request: Request<Incoming>,
     client: ProxyClient,
     target: SocketAddr,
     pause: Option<Arc<PauseState>>,
+    ack: Option<Arc<scoped_ack::QueueState>>,
 ) -> io::Result<hyper::Response<ProxyBody>> {
-    let (mut parts, body) = request.into_parts();
-    let mut body = body.map_err(io::Error::other).boxed();
+    let request = request.map(|body| body.map_err(io::Error::other).boxed());
+    let request = if let Some(ack) = ack
+        && request.uri().path() == "/node_service.ScannerControlService/ScannerScopedDirtyUsageAck"
+    {
+        match scoped_ack::intercept(request, client.clone(), target, ack).await? {
+            scoped_ack::Interception::Pass(request) => request,
+            scoped_ack::Interception::Handled(response) => return Ok(response),
+        }
+    } else {
+        request
+    };
+    let (parts, body) = request.into_parts();
+    let mut body = body;
     let mut observed = None;
     let mut inspection = None;
     if let Some(pause) = pause {
-        let method = parts.uri.path().strip_prefix("/node_service.NodeService/");
+        let method = parts
+            .uri
+            .path()
+            .strip_prefix("/node_service.NodeService/")
+            .or_else(|| parts.uri.path().strip_prefix("/node_service.ScannerControlService/"));
         if let Some(method) = method.filter(|name| pause.methods.contains(name)) {
             let mut route = None;
             let mut lock_owners = Vec::new();
@@ -599,6 +641,16 @@ async fn forward_request(
                     .with_trailers(async move { trailers.map(Ok::<_, io::Error>) })
                     .boxed();
                 matches
+            } else if method == "ScannerScopedDirtyUsageAck" {
+                let collected = Limited::new(body, 1024 * 1024).collect().await.map_err(io::Error::other)?;
+                let trailers = collected.trailers().cloned();
+                let bytes = collected.to_bytes();
+                let matches = scoped_ack_matches(&bytes, None)?;
+                body = Full::new(bytes)
+                    .map_err(|never| match never {})
+                    .with_trailers(async move { trailers.map(Ok::<_, io::Error>) })
+                    .boxed();
+                matches
             } else {
                 true
             };
@@ -642,24 +694,7 @@ async fn forward_request(
             }
         }
     }
-    if !parts.headers.contains_key(http::header::HOST)
-        && let Some(authority) = parts.uri.authority()
-    {
-        parts.headers.insert(
-            http::header::HOST,
-            http::HeaderValue::from_str(authority.as_str()).map_err(io::Error::other)?,
-        );
-    }
-    parts.uri = http::Uri::builder()
-        .scheme("http")
-        .authority(target.to_string())
-        .path_and_query(parts.uri.path_and_query().map_or("/", |value| value.as_str()))
-        .build()
-        .map_err(io::Error::other)?;
-    let response = client
-        .request(Request::from_parts(parts, body))
-        .await
-        .map_err(io::Error::other)?;
+    let response = forward_to_backend(Request::from_parts(parts, body), client, target).await?;
     if let Some((pause, disk)) = inspection
         && response.status().is_success()
         && response.headers().get("grpc-status").is_some_and(|status| status == "12")
@@ -712,6 +747,32 @@ async fn forward_request(
     Ok(response.map(|body| body.map_err(io::Error::other).boxed()))
 }
 
+async fn forward_to_backend(
+    request: Request<ProxyBody>,
+    client: ProxyClient,
+    target: SocketAddr,
+) -> io::Result<hyper::Response<Incoming>> {
+    let (mut parts, body) = request.into_parts();
+    if !parts.headers.contains_key(http::header::HOST)
+        && let Some(authority) = parts.uri.authority()
+    {
+        parts.headers.insert(
+            http::header::HOST,
+            http::HeaderValue::from_str(authority.as_str()).map_err(io::Error::other)?,
+        );
+    }
+    parts.uri = http::Uri::builder()
+        .scheme("http")
+        .authority(target.to_string())
+        .path_and_query(parts.uri.path_and_query().map_or("/", |value| value.as_str()))
+        .build()
+        .map_err(io::Error::other)?;
+    client
+        .request(Request::from_parts(parts, body))
+        .await
+        .map_err(io::Error::other)
+}
+
 impl Drop for RpcFaultProxy {
     fn drop(&mut self) {
         if let Some(requests) = self.late_requests.lock().expect("late RPC ownership").take() {
@@ -731,6 +792,7 @@ mod tests {
 
     struct Backend {
         received: Arc<AtomicUsize>,
+        auth_headers: Arc<Mutex<Vec<(String, String)>>>,
         address: SocketAddr,
         task: JoinHandle<()>,
     }
@@ -754,6 +816,8 @@ mod tests {
         let address = listener.local_addr()?;
         let received = Arc::new(AtomicUsize::new(0));
         let received_task = received.clone();
+        let auth_headers = Arc::new(Mutex::new(Vec::new()));
+        let backend_auth = auth_headers.clone();
         let task = tokio::spawn(async move {
             let mut children = JoinSet::new();
             loop {
@@ -761,10 +825,12 @@ mod tests {
                     accepted = listener.accept() => {
                         let Ok((socket, _)) = accepted else { break };
                         let received = received_task.clone();
+                        let backend_auth = backend_auth.clone();
                         let reply = reply.clone();
                         children.spawn(async move {
                             let service = service_fn(move |request: Request<Incoming>| {
                                 let received = received.clone();
+                                let backend_auth = backend_auth.clone();
                                 let reply = reply.clone();
                                 async move {
                                 received.fetch_add(1, Ordering::SeqCst);
@@ -773,6 +839,10 @@ mod tests {
                                         http::HeaderValue::from_str(authority.as_str()).expect("valid HTTP authority")
                                     })
                                 });
+                                if let (Some(nonce), Some(signature)) = (request.headers().get("x-rustfs-rpc-nonce"), request.headers().get("x-rustfs-rpc-signature-v2")) {
+                                    backend_auth.lock().expect("backend auth header observations").push((
+                                        nonce.to_str().expect("test nonce").to_string(), signature.to_str().expect("test signature").to_string()));
+                                }
                                 let body = request.into_body().collect().await?.to_bytes();
                                 let body = reply.unwrap_or(body);
                                 let mut trailers = http::HeaderMap::new();
@@ -795,7 +865,12 @@ mod tests {
                 }
             }
         });
-        Ok(Backend { received, address, task })
+        Ok(Backend {
+            received,
+            auth_headers,
+            address,
+            task,
+        })
     }
 
     async fn call(client: &ProxyClient, address: SocketAddr, method: &str) -> io::Result<hyper::Response<Incoming>> {
@@ -809,9 +884,14 @@ mod tests {
         bytes: Bytes,
     ) -> io::Result<hyper::Response<Incoming>> {
         let body: ProxyBody = Full::new(bytes).map_err(|never| match never {}).boxed();
+        let service = if method == "ScannerScopedDirtyUsageAck" {
+            "ScannerControlService"
+        } else {
+            "NodeService"
+        };
         let request = Request::builder()
             .method("POST")
-            .uri(format!("http://{address}/node_service.NodeService/{method}"))
+            .uri(format!("http://{address}/node_service.{service}/{method}"))
             .header(http::header::TE, "trailers")
             .body(body)
             .map_err(io::Error::other)?;
@@ -826,6 +906,353 @@ mod tests {
         frame.into()
     }
 
+    const ACK_OWNER: &str = "11111111-1111-1111-1111-111111111111";
+    const ACK_INSTANCE: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+
+    fn ack_scope() -> ScopedAckScope {
+        ScopedAckScope {
+            nonce: uuid::Uuid::new_v4().to_string(),
+            owner_id: ACK_OWNER.to_string(),
+            instance_id: ACK_INSTANCE.to_string(),
+            buckets: std::collections::BTreeSet::from(["cold-bucket".to_string()]),
+        }
+    }
+
+    fn ack_request(bucket: &str, generation: u64) -> ScannerScopedDirtyUsageAckRequest {
+        ScannerScopedDirtyUsageAckRequest {
+            challenge: vec![7; 16].into(),
+            protocol_version: 1,
+            owner_id: ACK_OWNER.to_string(),
+            instance_id: ACK_INSTANCE.to_string(),
+            scope: 1,
+            probe_only: false,
+            entries: vec![rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageEntry {
+                bucket: bucket.to_string(),
+                bucket_incarnation: vec![4; 16].into(),
+                generation,
+            }],
+        }
+    }
+
+    fn ack_frame(request: &ScannerScopedDirtyUsageAckRequest) -> Bytes {
+        let payload = request.encode_to_vec();
+        let mut frame = vec![0];
+        frame.extend_from_slice(&u32::try_from(payload.len()).expect("bounded test frame").to_be_bytes());
+        frame.extend_from_slice(&payload);
+        Bytes::from(frame)
+    }
+
+    fn ack_hash(bytes: &[u8]) -> String {
+        use sha2::Digest as _;
+        hex_simd::encode_to_string(&sha2::Sha256::digest(bytes)[..], hex_simd::AsciiCase::Lower)
+    }
+
+    // Transport-only fixtures: the distributed tests supply production signatures.
+    async fn call_ack_frame(
+        client: &ProxyClient,
+        address: SocketAddr,
+        frame: Bytes,
+        nonce: &str,
+    ) -> io::Result<hyper::Response<Incoming>> {
+        let request = Request::builder()
+            .method("POST")
+            .uri(format!("http://{address}/node_service.ScannerControlService/ScannerScopedDirtyUsageAck"))
+            .header(http::header::TE, "trailers")
+            .header("x-rustfs-rpc-nonce", nonce)
+            .header("x-rustfs-rpc-signature-v2", format!("opaque-signature-{nonce}"))
+            .body(Full::new(frame).map_err(|never| match never {}).boxed())
+            .map_err(io::Error::other)?;
+        client.request(request).await.map_err(io::Error::other)
+    }
+
+    #[tokio::test]
+    async fn scoped_ack_queue_loss_preserves_backend_boundary() {
+        tokio::time::timeout(Duration::from_secs(10), async {
+            for fault in [ScopedAckFault::RequestLoss, ScopedAckFault::ResponseLoss] {
+                let backend = backend().await.expect("backend");
+                let proxy = RpcFaultProxy::start_on("127.0.0.1:0".parse().expect("loopback"), backend.address)
+                    .await
+                    .expect("proxy");
+                let queue = proxy.scoped_ack_queue(ack_scope(), 1).expect("ACK queue");
+                let client: ProxyClient = Client::builder(TokioExecutor::new()).http2_only(true).build_http();
+                let frame = ack_frame(&ack_request("cold-bucket", 7));
+                let address = proxy.local_addr();
+                let body = frame.clone();
+                let caller = tokio::spawn(async move { call_ack_frame(&client, address, body, "loss").await });
+                queue.wait_for_packets(1).await.expect("captured ACK");
+                assert_eq!(backend.received.load(Ordering::SeqCst), 0, "retention must precede backend dispatch");
+                queue.settle(0, fault).expect("settle actual retained packet");
+                assert!(
+                    caller.await.expect("caller task").is_err(),
+                    "the caller must observe the injected transport failure"
+                );
+                let receipt = queue.wait_for_completion(0).await.expect("complete fault receipt");
+                let response_loss = fault == ScopedAckFault::ResponseLoss;
+                assert_eq!(receipt["forwarded"], response_loss);
+                assert_eq!(receipt["backend_complete"], response_loss);
+                assert_eq!(backend.received.load(Ordering::SeqCst), usize::from(response_loss));
+                if response_loss {
+                    assert_eq!(receipt["outcome"], "response_lost");
+                    assert_eq!(receipt["response"]["grpc_status"], "0", "status exists only in real backend trailers");
+                    assert_eq!(receipt["response"]["bytes"], frame.len());
+                    assert_eq!(receipt["response"]["sha256"], ack_hash(&frame));
+                    assert!(
+                        receipt["response"]["trailers_sha256"]
+                            .as_str()
+                            .is_some_and(|value| value.len() == 64)
+                    );
+                } else {
+                    assert_eq!(receipt["outcome"], "request_lost");
+                }
+            }
+        })
+        .await
+        .expect("bounded ACK loss checks");
+    }
+
+    #[tokio::test]
+    async fn scoped_ack_queue_keeps_cancelled_packets_and_releases_out_of_order() {
+        tokio::time::timeout(Duration::from_secs(10), async {
+            let backend = backend().await.expect("backend");
+            let proxy = RpcFaultProxy::start_on("127.0.0.1:0".parse().expect("loopback"), backend.address)
+                .await
+                .expect("proxy");
+            let queue = proxy.scoped_ack_queue(ack_scope(), 2).expect("two-packet ACK queue");
+            let client: ProxyClient = Client::builder(TokioExecutor::new()).http2_only(true).build_http();
+            let first = ack_frame(&ack_request("cold-bucket", 7));
+            let second = ack_frame(&ack_request("cold-bucket", 9));
+            let address = proxy.local_addr();
+            let old_client = client.clone();
+            let old_body = first.clone();
+            let old_caller = tokio::spawn(async move { call_ack_frame(&old_client, address, old_body, "first").await });
+            queue.wait_for_packets(1).await.expect("first retained packet");
+            old_caller.abort();
+            assert!(old_caller.await.expect_err("cancel original caller").is_cancelled());
+            let new_body = second.clone();
+            let new_caller = tokio::spawn(async move { call_ack_frame(&client, address, new_body, "second").await });
+            queue.wait_for_packets(2).await.expect("second retained packet");
+            assert_eq!(backend.received.load(Ordering::SeqCst), 0);
+            queue.settle(1, ScopedAckFault::Forward).expect("release newer packet first");
+            let response = new_caller
+                .await
+                .expect("new caller task")
+                .expect("new response")
+                .into_body()
+                .collect()
+                .await
+                .expect("complete body");
+            assert_eq!(response.trailers().expect("response trailers")["x-proof"], "unchanged-proof");
+            assert_eq!(response.to_bytes(), second);
+            queue.wait_for_completion(1).await.expect("new packet complete");
+            queue
+                .settle(0, ScopedAckFault::Forward)
+                .expect("release old packet after its caller disappeared");
+            let old = queue.wait_for_completion(0).await.expect("detached old packet complete");
+            assert_eq!(old["backend_complete"], true);
+            assert_eq!(old["request_sha256"], ack_hash(&first));
+            assert_eq!(old["response"]["sha256"], ack_hash(&first));
+            assert_eq!(backend.received.load(Ordering::SeqCst), 2);
+            assert_eq!(
+                *backend.auth_headers.lock().expect("backend header observations"),
+                vec![
+                    ("second".to_string(), "opaque-signature-second".to_string()),
+                    ("first".to_string(), "opaque-signature-first".to_string())
+                ]
+            );
+            assert!(
+                queue.settle(0, ScopedAckFault::Forward).is_err(),
+                "one original packet may only be forwarded once"
+            );
+        })
+        .await
+        .expect("bounded detached retention check");
+    }
+
+    #[tokio::test]
+    async fn scoped_ack_queue_bounds_capacity_and_excludes_other_scopes() {
+        tokio::time::timeout(Duration::from_secs(10), async {
+            let backend = backend().await.expect("backend");
+            let proxy = RpcFaultProxy::start_on("127.0.0.1:0".parse().expect("loopback"), backend.address)
+                .await
+                .expect("proxy");
+            assert!(proxy.scoped_ack_queue(ack_scope(), 0).is_err());
+            assert!(proxy.scoped_ack_queue(ack_scope(), 5).is_err());
+            let mut invalid = ack_scope();
+            invalid.nonce.clear();
+            assert!(proxy.scoped_ack_queue(invalid, 1).is_err());
+            let mut scope = ack_scope();
+            scope.buckets.insert("cold-bucket-z".to_string());
+            let queue = proxy.scoped_ack_queue(scope, 1).expect("bounded ACK queue");
+            let client: ProxyClient = Client::builder(TokioExecutor::new()).http2_only(true).build_http();
+            let mut probe = ack_request("cold-bucket", 7);
+            probe.probe_only = true;
+            let mut other_peer = ack_request("cold-bucket", 7);
+            other_peer.owner_id = uuid::Uuid::new_v4().to_string();
+            let mut other_instance = ack_request("cold-bucket", 7);
+            other_instance.instance_id = "b".repeat(32);
+            let mut invalid_challenge = ack_request("cold-bucket", 7);
+            invalid_challenge.challenge = vec![7; 15].into();
+            let mut noncanonical_entries = ack_request("cold-bucket-z", 7);
+            noncanonical_entries.entries.extend(ack_request("cold-bucket", 7).entries);
+            let unselected = [
+                probe,
+                other_peer,
+                other_instance,
+                ack_request("cold-bucket-other", 7),
+                ack_request("cold-bucket", u64::MAX),
+                invalid_challenge,
+                noncanonical_entries,
+            ];
+            let unselected_count = unselected.len();
+            for (index, request) in unselected.into_iter().enumerate() {
+                let body = ack_frame(&request);
+                let response = call_ack_frame(&client, proxy.local_addr(), body.clone(), "unselected")
+                    .await
+                    .expect("unselected ACK progresses");
+                assert_eq!(response.into_body().collect().await.expect("unselected body").to_bytes(), body);
+                assert!(queue.receipts().is_empty(), "unselected or protocol-invalid ACK must not consume a slot");
+                assert_eq!(backend.received.load(Ordering::SeqCst), index + 1);
+            }
+            assert!(queue.receipts().is_empty());
+            let address = proxy.local_addr();
+            let held_client = client.clone();
+            let valid_frame = ack_frame(&ack_request("cold-bucket", 7));
+            let held_body = valid_frame.clone();
+            let held = tokio::spawn(async move { call_ack_frame(&held_client, address, held_body, "held").await });
+            queue.wait_for_packets(1).await.expect("one matching packet");
+            assert!(!held.is_finished(), "the subsequent valid ACK must be retained");
+            assert_eq!(queue.receipts()[0]["request_sha256"], ack_hash(&valid_frame));
+            assert_eq!(backend.received.load(Ordering::SeqCst), unselected_count);
+            let extra = call_ack_frame(&client, address, ack_frame(&ack_request("cold-bucket", 8)), "beyond-capacity")
+                .await
+                .expect("capacity overflow passes without retention");
+            extra.into_body().collect().await.expect("overflow response");
+            assert_eq!(queue.receipts().len(), 1);
+            assert_eq!(backend.received.load(Ordering::SeqCst), unselected_count + 1);
+            drop(queue);
+            assert!(
+                held.await.expect("held caller").is_err(),
+                "dropping the session must cancel its retained packet"
+            );
+            assert_eq!(
+                backend.received.load(Ordering::SeqCst),
+                unselected_count + 1,
+                "session drop must never forward the held packet"
+            );
+        })
+        .await
+        .expect("bounded scope and lifecycle checks");
+    }
+
+    #[tokio::test]
+    async fn scoped_ack_queue_expiry_never_forwards_retained_packet() {
+        let (backend, _proxy, queue, caller) = tokio::time::timeout(Duration::from_secs(10), async {
+            let backend = backend().await.expect("backend");
+            let proxy = RpcFaultProxy::start_on("127.0.0.1:0".parse().expect("loopback"), backend.address)
+                .await
+                .expect("proxy");
+            let queue = proxy.scoped_ack_queue(ack_scope(), 1).expect("ACK queue");
+            let client: ProxyClient = Client::builder(TokioExecutor::new()).http2_only(true).build_http();
+            let address = proxy.local_addr();
+            let caller = tokio::spawn(async move {
+                call_ack_frame(&client, address, ack_frame(&ack_request("cold-bucket", 7)), "expires").await
+            });
+            queue.wait_for_packets(1).await.expect("retained packet");
+            (backend, proxy, queue, caller)
+        })
+        .await
+        .expect("bounded queue setup");
+        tokio::time::pause();
+        tokio::time::advance(Duration::from_secs(120)).await;
+        tokio::time::resume();
+        assert!(
+            tokio::time::timeout(Duration::from_secs(10), caller)
+                .await
+                .expect("expiry completes caller")
+                .expect("caller task")
+                .is_err()
+        );
+        let receipt = queue.receipts().into_iter().next().expect("expired receipt");
+        assert_eq!(receipt["completed"], true);
+        assert_eq!(receipt["forwarded"], false);
+        assert_eq!(receipt["backend_complete"], false);
+        assert_eq!(backend.received.load(Ordering::SeqCst), 0);
+        assert_eq!(
+            queue
+                .settle(0, ScopedAckFault::Forward)
+                .expect_err("expired packet cannot be released")
+                .kind(),
+            io::ErrorKind::TimedOut
+        );
+    }
+
+    #[tokio::test]
+    async fn scanner_ack_fault_matches_exact_bucket_and_excludes_probes() {
+        use rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageEntry;
+
+        tokio::time::timeout(Duration::from_secs(10), async {
+            let backend = backend().await.expect("backend listener");
+            let proxy = RpcFaultProxy::start_on("127.0.0.1:0".parse().expect("loopback"), backend.address)
+                .await
+                .expect("proxy listener");
+            let client: ProxyClient = Client::builder(TokioExecutor::new()).http2_only(true).build_http();
+            let pause = proxy
+                .pause(&["ScannerScopedDirtyUsageAck"], 1, Some("cold-bucket"))
+                .expect("scoped ACK fault");
+            let frame = |bucket: &str, probe_only: bool| {
+                let payload = ScannerScopedDirtyUsageAckRequest {
+                    challenge: vec![3; 16].into(),
+                    protocol_version: 1,
+                    owner_id: "11111111-1111-1111-1111-111111111111".to_string(),
+                    instance_id: "a".repeat(32),
+                    scope: 1,
+                    probe_only,
+                    entries: vec![ScannerScopedDirtyUsageEntry {
+                        bucket: bucket.to_string(),
+                        bucket_incarnation: vec![4; 16].into(),
+                        generation: 17,
+                    }],
+                }
+                .encode_to_vec();
+                let mut frame = vec![0];
+                frame.extend_from_slice(&u32::try_from(payload.len()).expect("small ACK frame").to_be_bytes());
+                frame.extend_from_slice(&payload);
+                Bytes::from(frame)
+            };
+            for bytes in [frame("cold-bucket", true), frame("cold-bucket-other", false)] {
+                let response = call_body(&client, proxy.local_addr(), "ScannerScopedDirtyUsageAck", bytes.clone())
+                    .await
+                    .expect("probe and other bucket must progress");
+                assert_eq!(response.into_body().collect().await.expect("unselected response").to_bytes(), bytes);
+            }
+            assert_eq!(pause.hits(), 0);
+            let signed_bytes = frame("cold-bucket", false);
+            let task_bytes = signed_bytes.clone();
+            let address = proxy.local_addr();
+            let blocked =
+                tokio::spawn(async move { call_body(&client, address, "ScannerScopedDirtyUsageAck", task_bytes).await });
+            pause
+                .wait_for_hits(1)
+                .await
+                .expect("real ScannerControlService method must reach the fault");
+            assert!(!blocked.is_finished());
+            assert_eq!(backend.received.load(Ordering::SeqCst), 2, "held ACK must not apply at the backend");
+            pause.release();
+            let collected = blocked
+                .await
+                .expect("ACK task")
+                .expect("released ACK")
+                .into_body()
+                .collect()
+                .await
+                .expect("ACK body and trailers");
+            assert_eq!(collected.trailers().expect("ACK trailers")["x-proof"], "unchanged-proof");
+            assert_eq!(collected.to_bytes(), signed_bytes, "proxy must preserve the complete signed ACK frame");
+        })
+        .await
+        .expect("scanner ACK fault deadline");
+    }
+
     fn native_abort_request(path: &str) -> RecoverMultipartSourceRequest {
         use crate::storage_api::multipart_source_rpc::{
             MultipartAbortCertificate, MultipartAbortProof, MultipartAbortProofKind, MultipartAbortRound, MultipartSourcePhase,

crates/e2e_test/src/rpc_fault_proxy/scoped_ack.rs

@@ -0,0 +1,404 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+
+//! Bounded retention of original scoped ACK frames, independent of caller cancellation.
+
+use super::{ProxyBody, ProxyClient, forward_to_backend, unary_payload};
+use bytes::Bytes;
+use http_body_util::{BodyExt as _, Full, Limited};
+use hyper::{Request, Response};
+use prost::Message as _;
+use rustfs_protos::proto_gen::node_service::{ScannerScopedDirtyUsageAckRequest, ScannerScopedDirtyUsageAckResponse};
+use rustfs_protos::scoped_dirty_usage::validate_scoped_dirty_usage_request;
+use serde_json::{Value, json};
+use sha2::{Digest as _, Sha256};
+use std::{
+    collections::BTreeSet,
+    io,
+    net::SocketAddr,
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::Duration,
+};
+use tokio::{
+    sync::{Notify, oneshot, watch},
+    task::AbortHandle,
+    time::Instant,
+};
+
+const MAX_RETAINED: usize = 4;
+const MAX_BYTES: usize = 1024 * 1024;
+const RETENTION: Duration = Duration::from_secs(120);
+
+pub(crate) struct ScopedAckScope {
+    pub nonce: String,
+    pub owner_id: String,
+    pub instance_id: String,
+    pub buckets: BTreeSet<String>,
+}
+
+impl ScopedAckScope {
+    fn validate(&self) -> io::Result<()> {
+        let valid_uuid = |value: &str| uuid::Uuid::parse_str(value).is_ok_and(|value| !value.is_nil());
+        if !valid_uuid(&self.nonce)
+            || !valid_uuid(&self.owner_id)
+            || self.instance_id.is_empty()
+            || self.instance_id.len() > 128
+            || self.buckets.is_empty()
+            || self.buckets.len() > 16
+            || self.buckets.iter().any(|bucket| bucket.is_empty() || bucket.len() > 63)
+        {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "scoped ACK queue needs a bounded session, receiver and bucket scope",
+            ));
+        }
+        Ok(())
+    }
+
+    fn matches(&self, request: &ScannerScopedDirtyUsageAckRequest) -> bool {
+        !request.probe_only
+            && validate_scoped_dirty_usage_request(request).is_ok()
+            && request.owner_id == self.owner_id
+            && request.instance_id == self.instance_id
+            && request.entries.iter().all(|entry| self.buckets.contains(&entry.bucket))
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum ScopedAckFault {
+    Forward,
+    RequestLoss,
+    ResponseLoss,
+}
+
+struct Packet {
+    action: watch::Sender<Option<ScopedAckFault>>,
+    receipt: Mutex<Value>,
+    task: Mutex<Option<AbortHandle>>,
+}
+
+pub(super) struct QueueState {
+    scope: ScopedAckScope,
+    capacity: usize,
+    started: Instant,
+    deadline: Instant,
+    closed: AtomicBool,
+    // The packet inventory lock is released before any packet lock or await.
+    packets: Mutex<Vec<Arc<Packet>>>,
+    changed: Notify,
+}
+
+pub(super) type QueueSlot = Arc<Mutex<Option<Arc<QueueState>>>>;
+
+pub(super) fn empty_slot() -> QueueSlot {
+    Arc::new(Mutex::new(None))
+}
+
+pub(crate) struct ScopedAckQueue {
+    state: Arc<QueueState>,
+    slot: QueueSlot,
+}
+
+impl ScopedAckQueue {
+    pub(super) fn arm(slot: QueueSlot, scope: ScopedAckScope, capacity: usize) -> io::Result<Self> {
+        scope.validate()?;
+        if !(1..=MAX_RETAINED).contains(&capacity) {
+            return Err(io::Error::new(io::ErrorKind::InvalidInput, "scoped ACK capacity must be 1..=4"));
+        }
+        let mut active = slot.lock().expect("scoped ACK queue slot");
+        if active.is_some() {
+            return Err(io::Error::new(io::ErrorKind::AlreadyExists, "scoped ACK queue is already armed"));
+        }
+        let started = Instant::now();
+        let state = Arc::new(QueueState {
+            scope,
+            capacity,
+            started,
+            deadline: started + RETENTION,
+            closed: AtomicBool::new(false),
+            packets: Mutex::new(Vec::new()),
+            changed: Notify::new(),
+        });
+        *active = Some(state.clone());
+        drop(active);
+        Ok(Self { state, slot })
+    }
+
+    pub(crate) fn receipts(&self) -> Vec<Value> {
+        let packets = self.state.packets.lock().expect("scoped ACK packets").clone();
+        packets
+            .iter()
+            .map(|packet| packet.receipt.lock().expect("scoped ACK receipt").clone())
+            .collect()
+    }
+
+    pub(crate) async fn wait_for_packets(&self, count: usize) -> io::Result<()> {
+        self.wait_until(|| self.receipts().len() >= count).await
+    }
+
+    pub(crate) async fn wait_for_completion(&self, id: usize) -> io::Result<Value> {
+        self.wait_until(|| {
+            self.receipts()
+                .get(id)
+                .is_some_and(|receipt| receipt["completed"].as_bool() == Some(true))
+        })
+        .await?;
+        self.receipts()
+            .get(id)
+            .cloned()
+            .ok_or_else(|| io::Error::other("scoped ACK receipt disappeared"))
+    }
+
+    async fn wait_until(&self, ready: impl Fn() -> bool) -> io::Result<()> {
+        tokio::time::timeout_at(self.state.deadline, async {
+            loop {
+                let changed = self.state.changed.notified();
+                if ready() {
+                    return;
+                }
+                changed.await;
+            }
+        })
+        .await
+        .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "scoped ACK queue lifetime elapsed"))
+    }
+
+    pub(crate) fn settle(&self, id: usize, fault: ScopedAckFault) -> io::Result<()> {
+        if Instant::now() >= self.state.deadline {
+            return Err(io::Error::new(io::ErrorKind::TimedOut, "scoped ACK retention expired"));
+        }
+        let packet = self
+            .state
+            .packets
+            .lock()
+            .expect("scoped ACK packets")
+            .get(id)
+            .cloned()
+            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "scoped ACK packet is not retained"))?;
+        if !packet.action.send_if_modified(|current| {
+            if current.is_none() {
+                *current = Some(fault);
+                true
+            } else {
+                false
+            }
+        }) {
+            return Err(io::Error::new(io::ErrorKind::AlreadyExists, "scoped ACK packet was already settled"));
+        }
+        Ok(())
+    }
+}
+
+impl Drop for ScopedAckQueue {
+    fn drop(&mut self) {
+        self.state.closed.store(true, Ordering::SeqCst);
+        let mut slot = self.slot.lock().expect("scoped ACK queue slot");
+        if slot.as_ref().is_some_and(|current| Arc::ptr_eq(current, &self.state)) {
+            *slot = None;
+        }
+        drop(slot);
+        let packets = self.state.packets.lock().expect("scoped ACK packets").clone();
+        for packet in packets {
+            if let Some(task) = packet.task.lock().expect("scoped ACK task").take() {
+                task.abort();
+            }
+        }
+    }
+}
+
+pub(super) enum Interception {
+    Pass(Request<ProxyBody>),
+    Handled(Response<ProxyBody>),
+}
+
+fn restored_body(bytes: Bytes, trailers: Option<http::HeaderMap>) -> ProxyBody {
+    Full::new(bytes)
+        .map_err(|never| match never {})
+        .with_trailers(async move { trailers.map(Ok::<_, io::Error>) })
+        .boxed()
+}
+
+fn hash(bytes: &[u8]) -> String {
+    hex(Sha256::digest(bytes).as_ref())
+}
+
+fn hex(bytes: &[u8]) -> String {
+    hex_simd::encode_to_string(bytes, hex_simd::AsciiCase::Lower)
+}
+
+fn trailer_receipt(trailers: Option<&http::HeaderMap>) -> Value {
+    trailers.map_or(Value::Null, |headers| {
+        json!(
+            headers
+                .iter()
+                .map(|(name, value)| { json!({"name": name.as_str(), "value_hex": hex(value.as_bytes())}) })
+                .collect::<Vec<_>>()
+        )
+    })
+}
+
+fn header_hash(headers: &http::HeaderMap) -> String {
+    let mut hasher = Sha256::new();
+    for (name, value) in headers {
+        hasher.update(name.as_str().as_bytes());
+        hasher.update([0]);
+        hasher.update(value.as_bytes());
+        hasher.update([0]);
+    }
+    hex(hasher.finalize().as_ref())
+}
+
+fn millis(state: &QueueState) -> u64 {
+    u64::try_from(state.started.elapsed().as_millis()).unwrap_or(u64::MAX)
+}
+
+pub(super) async fn intercept(
+    request: Request<ProxyBody>,
+    client: ProxyClient,
+    target: SocketAddr,
+    state: Arc<QueueState>,
+) -> io::Result<Interception> {
+    if Instant::now() >= state.deadline || state.closed.load(Ordering::SeqCst) {
+        return Ok(Interception::Pass(request));
+    }
+    let (parts, body) = request.into_parts();
+    let collected = tokio::time::timeout_at(state.deadline, Limited::new(body, MAX_BYTES).collect())
+        .await
+        .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "scoped ACK frame collection expired"))?
+        .map_err(io::Error::other)?;
+    let trailers = collected.trailers().cloned();
+    let bytes = collected.to_bytes();
+    let decoded = ScannerScopedDirtyUsageAckRequest::decode(unary_payload(&bytes)?).map_err(io::Error::other)?;
+    let request = Request::from_parts(parts, restored_body(bytes.clone(), trailers.clone()));
+    if Instant::now() >= state.deadline || !state.scope.matches(&decoded) {
+        return Ok(Interception::Pass(request));
+    }
+    let header = |name| request.headers().get(name).and_then(|value| value.to_str().ok());
+    let (action, mut selected) = watch::channel(None);
+    let packet = {
+        let mut packets = state.packets.lock().expect("scoped ACK packets");
+        if packets.len() >= state.capacity || state.closed.load(Ordering::SeqCst) {
+            return Ok(Interception::Pass(request));
+        }
+        let id = packets.len();
+        let packet = Arc::new(Packet {
+            action,
+            task: Mutex::new(None),
+            receipt: Mutex::new(json!({
+                "id": id, "session_nonce": state.scope.nonce, "receiver": target.to_string(),
+                "authority": request.uri().authority().map(|value| value.as_str()),
+                "captured_at_ms": millis(&state), "completed": false, "forwarded": false,
+                "backend_complete": false, "outcome": "held", "request_bytes": bytes.len(),
+            "request_sha256": hash(&bytes), "request_frame_hex": hex(&bytes), "request_headers_sha256": header_hash(request.headers()),
+                "request_trailers_sha256": trailers.as_ref().map(header_hash),
+            "request_trailers": trailer_receipt(trailers.as_ref()),
+                "auth_nonce": header("x-rustfs-rpc-nonce"), "replay_nonce": header("x-rustfs-rpc-replay-nonce"),
+                "signature_sha256": header("x-rustfs-rpc-signature-v2").map(|value| hash(value.as_bytes())),
+                "replay_signature_sha256": header("x-rustfs-rpc-signature-v3").map(|value| hash(value.as_bytes())),
+                "request": {"owner_id": decoded.owner_id, "instance_id": decoded.instance_id,
+                    "probe_only": decoded.probe_only, "entries": decoded.entries.iter().map(|entry| json!({
+                        "bucket": entry.bucket, "bucket_incarnation": uuid::Uuid::from_slice(&entry.bucket_incarnation).ok().map(|value| value.to_string()),
+                        "generation": entry.generation })).collect::<Vec<_>>() }
+            })),
+        });
+        packets.push(packet.clone());
+        packet
+    };
+    let (reply, response) = oneshot::channel();
+    let worker_packet = packet.clone();
+    let worker_state = state.clone();
+    // This owner survives an HTTP/2 stream reset from the original 5 s caller.
+    // Every request is settled once and is discarded at the session deadline.
+    let task = tokio::spawn(async move {
+        let result = tokio::time::timeout_at(worker_state.deadline, async {
+            let fault = loop {
+                if let Some(fault) = *selected.borrow_and_update() {
+                    break fault;
+                }
+                selected
+                    .changed()
+                    .await
+                    .map_err(|_| io::Error::other("scoped ACK queue closed"))?;
+            };
+            if fault == ScopedAckFault::RequestLoss {
+                worker_packet.receipt.lock().expect("scoped ACK receipt")["outcome"] = json!("request_lost");
+                return Err(io::Error::new(io::ErrorKind::ConnectionAborted, "injected scoped ACK request loss"));
+            }
+            if worker_state.closed.load(Ordering::SeqCst) || Instant::now() >= worker_state.deadline {
+                return Err(io::Error::new(
+                    io::ErrorKind::ConnectionAborted,
+                    "scoped ACK session ended or expired before forwarding",
+                ));
+            }
+            {
+                let mut receipt = worker_packet.receipt.lock().expect("scoped ACK receipt");
+                receipt["forwarded"] = json!(true);
+                receipt["forwarded_at_ms"] = json!(millis(&worker_state));
+            }
+            let backend = forward_to_backend(request, client, target).await?;
+            let (parts, body) = backend.into_parts();
+            let collected = Limited::new(body, MAX_BYTES).collect().await.map_err(io::Error::other)?;
+            let trailers = collected.trailers().cloned();
+            let bytes = collected.to_bytes();
+            let grpc_status = trailers
+                .as_ref()
+                .and_then(|values| values.get("grpc-status"))
+                .or_else(|| parts.headers.get("grpc-status"))
+                .and_then(|value| value.to_str().ok());
+            let response = unary_payload(&bytes)
+                .ok()
+                .and_then(|payload| ScannerScopedDirtyUsageAckResponse::decode(payload).ok())
+                .map(|response| {
+                    json!({"owner_id": response.owner_id, "instance_id": response.instance_id,
+                    "supported": response.supported, "cleared": response.cleared,
+                    "response_proof_bytes": response.response_proof.len()})
+                });
+            {
+                let mut receipt = worker_packet.receipt.lock().expect("scoped ACK receipt");
+                receipt["backend_complete"] = json!(true);
+                receipt["backend_completed_at_ms"] = json!(millis(&worker_state));
+                receipt["response"] = json!({"status": parts.status.as_u16(), "grpc_status": grpc_status,
+                    "bytes": bytes.len(), "sha256": hash(&bytes), "frame_hex": hex(&bytes), "headers_sha256": header_hash(&parts.headers),
+                    "trailers": trailer_receipt(trailers.as_ref()),
+                    "trailers_sha256": trailers.as_ref().map(header_hash), "ack": response});
+                receipt["outcome"] = json!(if fault == ScopedAckFault::ResponseLoss {
+                    "response_lost"
+                } else {
+                    "forwarded"
+                });
+            }
+            if fault == ScopedAckFault::ResponseLoss {
+                return Err(io::Error::new(
+                    io::ErrorKind::ConnectionAborted,
+                    "injected scoped ACK response loss after backend trailers",
+                ));
+            }
+            Ok(Response::from_parts(parts, restored_body(bytes, trailers)))
+        })
+        .await
+        .unwrap_or_else(|_| Err(io::Error::new(io::ErrorKind::TimedOut, "scoped ACK retention expired")));
+        {
+            let mut receipt = worker_packet.receipt.lock().expect("scoped ACK receipt");
+            if result.is_err() && receipt["outcome"] == "held" {
+                receipt["outcome"] = json!("transport_failed_or_expired");
+            }
+            receipt["completed"] = json!(true);
+            receipt["completed_at_ms"] = json!(millis(&worker_state));
+        }
+        let _ = reply.send(result);
+        worker_state.changed.notify_one();
+    });
+    *packet.task.lock().expect("scoped ACK task") = Some(task.abort_handle());
+    if state.closed.load(Ordering::SeqCst) {
+        task.abort();
+    }
+    drop(task);
+    state.changed.notify_one();
+    response
+        .await
+        .map_err(|_| io::Error::new(io::ErrorKind::ConnectionAborted, "scoped ACK queue ended"))?
+        .map(Interception::Handled)
+}

crates/e2e_test/src/storage_api.rs

@@ -80,3 +80,14 @@ pub(crate) mod multipart_source_rpc {
         PublicationIdentity, SourceGeneration, SourceGeometry, SourceLineage,
     };
 }
+
+#[cfg(test)]
+pub(crate) mod scanner_peer_bucket {
+    use rustfs_ecstore::api::disk as ecstore_disk;
+    use rustfs_ecstore::api::erasure as ecstore_erasure;
+    use rustfs_ecstore::api::rpc as ecstore_rpc;
+
+    pub(crate) use ecstore_disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
+    pub(crate) use ecstore_erasure::{ReedSolomonEncoder, calc_shard_size};
+    pub(crate) use ecstore_rpc::PeerRestClient;
+}

crates/s3select-query/Cargo.toml

@@ -57,8 +57,6 @@ tracing = { workspace = true }
 
 [dev-dependencies]
 rustfs-test-utils = { workspace = true, features = ["put-object-commit-barrier"] }
-# Match the streaming inference's field order in Arrow's JSON reference.
-serde_json = { workspace = true, features = ["preserve_order"] }
 
 [lib]
 doctest = false

crates/s3select-query/src/dispatcher/json_document_schema.rs

@@ -426,14 +426,18 @@ impl InferredObject {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::dispatcher::json_lines_schema;
     use datafusion::arrow::json::reader::infer_json_schema as arrow_infer_json_schema;
 
     const TEST_SCHEMA_LIMIT: usize = 4 * 1024 * 1024;
 
     fn assert_matches_arrow(input: &[u8]) -> Schema {
         let records = input.iter().filter(|byte| **byte == b'\n').count();
-        let (expected, expected_records) = arrow_infer_json_schema(io::BufReader::new(input), Some(records))
+        let (_, expected_records) = arrow_infer_json_schema(io::BufReader::new(input), Some(records))
             .expect("Arrow should infer the compatibility fixture");
+        // Arrow's types in source key order, which is what JSON LINES uses.
+        let expected = json_lines_schema::infer_schema(input, records)
+            .expect("JSON LINES inference should accept the compatibility fixture");
         let (actual, actual_records, _) = infer_schema(input, records, &AtomicBool::new(false), TEST_SCHEMA_LIMIT)
             .expect("streaming inference should accept the compatibility fixture");
         assert_eq!(actual_records, expected_records);

crates/s3select-query/src/dispatcher/json_lines_schema.rs

@@ -0,0 +1,185 @@
+// 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.
+
+use datafusion::arrow::{
+    datatypes::{DataType, FieldRef, Fields, Schema},
+    error::ArrowError,
+    json::reader::{ValueIter, infer_json_schema_from_iterator},
+};
+use serde::de::{self, DeserializeSeed, MapAccess, SeqAccess, Visitor};
+use std::{collections::HashMap, fmt, sync::Arc};
+
+/// Infers a JSON LINES schema with Arrow's types and the source's first-seen
+/// key order.
+///
+/// Arrow walks each record as a `serde_json::Map`, which sorts keys unless the
+/// process-wide `serde_json/preserve_order` feature is enabled, so `SELECT *`
+/// would otherwise reorder fields and CSV output columns.
+pub(super) fn infer_schema(bytes: &[u8], max_records: usize) -> Result<Schema, ArrowError> {
+    let schema = infer_json_schema_from_iterator(ValueIter::new(bytes, Some(max_records)))?;
+    let mut order = KeyOrder::default();
+    // Same records as `ValueIter`: one value per line, blank lines skipped.
+    // Arrow has already rejected invalid UTF-8 and JSON within this prefix.
+    let records = bytes
+        .split(|byte| *byte == b'\n')
+        .filter_map(|line| std::str::from_utf8(line).ok())
+        .map(str::trim)
+        .filter(|line| !line.is_empty())
+        .take(max_records);
+    for record in records {
+        let mut deserializer = serde_json::Deserializer::from_str(record);
+        (&mut order)
+            .deserialize(&mut deserializer)
+            .map_err(|error| ArrowError::JsonError(format!("Not valid JSON: {error}")))?;
+    }
+    Ok(Schema::new_with_metadata(order.reorder(schema.fields()), schema.metadata().clone()))
+}
+
+/// First-seen key order of a JSON object, merged across records. Array
+/// elements share their parent's node, matching Arrow's `List<Struct>` shape.
+#[derive(Default)]
+struct KeyOrder {
+    positions: HashMap<String, usize>,
+    children: Vec<KeyOrder>,
+}
+
+impl KeyOrder {
+    fn child(&mut self, key: String) -> &mut KeyOrder {
+        let next = self.children.len();
+        let position = *self.positions.entry(key).or_insert(next);
+        if position == next {
+            self.children.push(KeyOrder::default());
+        }
+        &mut self.children[position]
+    }
+
+    fn reorder(&self, fields: &Fields) -> Fields {
+        let mut fields: Vec<FieldRef> = fields
+            .iter()
+            .map(|field| match self.positions.get(field.name()) {
+                Some(&position) => {
+                    let data_type = self.children[position].reorder_type(field.data_type());
+                    Arc::new(field.as_ref().clone().with_data_type(data_type))
+                }
+                None => Arc::clone(field),
+            })
+            .collect();
+        fields.sort_by_key(|field| self.positions.get(field.name()).copied().unwrap_or(usize::MAX));
+        fields.into()
+    }
+
+    fn reorder_type(&self, data_type: &DataType) -> DataType {
+        match data_type {
+            DataType::Struct(fields) => DataType::Struct(self.reorder(fields)),
+            DataType::List(item) => {
+                DataType::List(Arc::new(item.as_ref().clone().with_data_type(self.reorder_type(item.data_type()))))
+            }
+            other => other.clone(),
+        }
+    }
+}
+
+impl<'de> DeserializeSeed<'de> for &mut KeyOrder {
+    type Value = ();
+
+    fn deserialize<D: de::Deserializer<'de>>(self, deserializer: D) -> Result<(), D::Error> {
+        deserializer.deserialize_any(self)
+    }
+}
+
+impl<'de> Visitor<'de> for &mut KeyOrder {
+    type Value = ();
+
+    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+        formatter.write_str("a JSON value")
+    }
+
+    fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<(), A::Error> {
+        while let Some(key) = map.next_key::<String>()? {
+            map.next_value_seed(self.child(key))?;
+        }
+        Ok(())
+    }
+
+    fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<(), A::Error> {
+        while seq.next_element_seed(&mut *self)?.is_some() {}
+        Ok(())
+    }
+
+    fn visit_bool<E: de::Error>(self, _: bool) -> Result<(), E> {
+        Ok(())
+    }
+
+    fn visit_i64<E: de::Error>(self, _: i64) -> Result<(), E> {
+        Ok(())
+    }
+
+    fn visit_u64<E: de::Error>(self, _: u64) -> Result<(), E> {
+        Ok(())
+    }
+
+    fn visit_f64<E: de::Error>(self, _: f64) -> Result<(), E> {
+        Ok(())
+    }
+
+    fn visit_str<E: de::Error>(self, _: &str) -> Result<(), E> {
+        Ok(())
+    }
+
+    fn visit_unit<E: de::Error>(self) -> Result<(), E> {
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::infer_schema;
+    use datafusion::arrow::datatypes::{DataType, Fields};
+
+    fn names(fields: &Fields) -> Vec<&str> {
+        fields.iter().map(|field| field.name().as_str()).collect()
+    }
+
+    #[test]
+    fn keeps_first_seen_key_order_at_every_level() {
+        let input = b"{\"z\":1,\"a\":{\"y\":[{\"k2\":1,\"k1\":2}],\"b\":2}}\n\n{\"m\":null,\"a\":{\"c\":3}}\n";
+        let schema = infer_schema(input, 1000).expect("schema");
+
+        assert_eq!(names(schema.fields()), ["z", "a", "m"]);
+        let DataType::Struct(a) = schema.field(1).data_type() else {
+            panic!("a should be a struct");
+        };
+        assert_eq!(names(a), ["y", "b", "c"]);
+        let DataType::List(item) = a[0].data_type() else {
+            panic!("y should be a list");
+        };
+        let DataType::Struct(y) = item.data_type() else {
+            panic!("y items should be structs");
+        };
+        assert_eq!(names(y), ["k2", "k1"]);
+    }
+
+    #[test]
+    fn ignores_records_after_the_inference_prefix_like_arrow() {
+        let schema = infer_schema(b"{\"b\":1,\"a\":2}\nnot json\n", 1).expect("schema");
+
+        assert_eq!(names(schema.fields()), ["b", "a"]);
+    }
+
+    #[test]
+    fn keeps_arrow_errors_for_invalid_records() {
+        assert!(infer_schema(b"{\"a\":1}\n[1]\n", 1000).is_err());
+        assert!(infer_schema(b"{\"a\":\n", 1000).is_err());
+    }
+}

crates/s3select-query/src/dispatcher/manager.rs

@@ -28,11 +28,12 @@ use datafusion::{
         record_batch::RecordBatch,
     },
     datasource::{
-        file_format::{csv::CsvFormat, json::JsonFormat},
+        file_format::{DEFAULT_SCHEMA_INFER_MAX_RECORD, csv::CsvFormat, json::JsonFormat},
         listing::{ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl},
     },
     error::Result as DFResult,
     execution::{RecordBatchStream, SendableRecordBatchStream},
+    object_store::ObjectStoreExt as _,
     sql::sqlparser::parser::ParserError,
 };
 use futures::Stream;
@@ -61,7 +62,7 @@ use tokio::{
 };
 
 use crate::{
-    dispatcher::{json_document_table::JsonDocumentTable, parquet_table::ParquetSelectTable},
+    dispatcher::{json_document_table::JsonDocumentTable, json_lines_schema, parquet_table::ParquetSelectTable},
     execution::factory::QueryExecutionFactoryRef,
     instance::{DEFAULT_MAX_CONCURRENT_QUERIES, DEFAULT_QUERY_TIMEOUT_SECS},
     metadata::{ContextProviderExtension, MetadataProvider, TableHandleProviderRef, base_table::BaseTableProvider},
@@ -508,7 +509,11 @@ impl SimpleQueryDispatcher {
                 return Err(SelectError::InvalidDataSource.into());
             };
 
-        let resolve_schema = listing_options.infer_schema(session.inner(), &table_path).await?;
+        let resolve_schema = if self.input.request.input_serialization.json.is_some() {
+            infer_json_lines_schema(session, &table_path).await?
+        } else {
+            listing_options.infer_schema(session.inner(), &table_path).await?
+        };
         let config = if need_rename_volume_name {
             let mut new_fields = Vec::new();
             for (i, field) in resolve_schema.fields().iter().enumerate() {
@@ -549,6 +554,14 @@ impl SimpleQueryDispatcher {
     }
 }
 
+/// Reads the object like DataFusion's `JsonFormat::infer_schema` does, but
+/// keeps the source key order (see `json_lines_schema::infer_schema`).
+async fn infer_json_lines_schema(session: &SessionCtx, table_path: &ListingTableUrl) -> DFResult<SchemaRef> {
+    let store = session.inner().runtime_env().object_store(table_path.object_store())?;
+    let bytes = store.get(table_path.prefix()).await?.bytes().await?;
+    Ok(Arc::new(json_lines_schema::infer_schema(&bytes, DEFAULT_SCHEMA_INFER_MAX_RECORD)?))
+}
+
 fn is_json_document_input(input: &SelectObjectContentInput) -> bool {
     input
         .request
@@ -856,6 +869,7 @@ mod tests {
             array::{Int32Array, StringArray},
             datatypes::{DataType, Field, Schema, SchemaRef},
             record_batch::RecordBatch,
+            util::display::array_value_to_string,
         },
         common::DataFusionError,
         execution::object_store::ObjectStoreUrl,
@@ -1727,6 +1741,334 @@ mod tests {
         }
     }
 
+    async fn json_lines_plan_schema(data: &'static [u8], sql: &str) -> SchemaRef {
+        let mut input = test_input();
+        input.key = "records.json".to_owned();
+        input.request.input_serialization.csv = None;
+        input.request.input_serialization.json = Some(JSONInput {
+            type_: Some(JSONType::from_static(JSONType::LINES)),
+        });
+        let input = Arc::new(input);
+        let optimizer = Arc::new(CascadeOptimizerBuilder::default().build());
+        let dispatcher = test_dispatcher_for_input(
+            Arc::clone(&input),
+            Arc::new(Semaphore::new(1)),
+            Duration::from_secs(30),
+            Arc::new(SqlQueryExecutionFactory::new(optimizer, Arc::new(LocalScheduler {}))),
+        );
+        let query = Query::new(QueryContext { input }, sql.to_owned());
+        let machine = dispatcher.build_query_state_machine(query).await.expect("build query state");
+        let store_url = ObjectStoreUrl::parse("s3://test-bucket").expect("test store URL");
+        let store = machine
+            .session
+            .inner()
+            .runtime_env()
+            .object_store(&store_url)
+            .expect("test store");
+        store
+            .put(&Path::from("records.json"), data.into())
+            .await
+            .expect("write JSON LINES object");
+        let Plan::Query(plan) = dispatcher
+            .build_logical_plan(machine)
+            .await
+            .expect("plan JSON LINES query")
+            .expect("select plan");
+        Arc::new(plan.df_plan.schema().as_arrow().clone())
+    }
+
+    fn field_names(fields: &datafusion::arrow::datatypes::Fields) -> Vec<&str> {
+        fields.iter().map(|field| field.name().as_str()).collect()
+    }
+
+    fn struct_fields(data_type: &DataType) -> &datafusion::arrow::datatypes::Fields {
+        match data_type {
+            DataType::Struct(fields) => fields,
+            DataType::List(item) => struct_fields(item.data_type()),
+            other => panic!("expected a struct or list of structs, got {other}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn json_lines_select_star_keeps_source_field_order() {
+        let schema = json_lines_plan_schema(
+            b"{\"z\":1,\"a b\":{\"y\":true,\"B\":[{\"k2\":1,\"k1\":2}]},\"A\":\"x\"}\n\n  \r\n{\"A\":\"y\",\"$c\":2,\"a b\":{\"x\":1}}\r\n",
+            "SELECT * FROM S3Object",
+        )
+        .await;
+
+        assert_eq!(field_names(schema.fields()), ["z", "a b", "A", "$c"]);
+        let nested = struct_fields(schema.field_with_name("a b").expect("nested field").data_type());
+        assert_eq!(field_names(nested), ["y", "B", "x"]);
+        let listed = struct_fields(nested.find("B").expect("list field").1.data_type());
+        assert_eq!(field_names(listed), ["k2", "k1"]);
+    }
+
+    #[tokio::test]
+    async fn path_projections_are_named_after_their_last_key() {
+        const DATA: &[u8] = b"{\"obj0\":{\"id\":\"id0\",\"inner\":{\"Key\":1}}}\n";
+        for (sql, expected) in [
+            ("SELECT s.obj0.id, s.obj0.inner.\"Key\" FROM S3Object s", ["id", "Key"].as_slice()),
+            ("SELECT obj0.id FROM S3Object", &["id"]),
+            ("SELECT s.obj0.id FROM S3Object s ORDER BY s.obj0.id", &["id"]),
+            ("SELECT s.obj0.id, COUNT(*) FROM S3Object s GROUP BY s.obj0.id", &["id", "count(*)"]),
+        ] {
+            let schema = json_lines_plan_schema(DATA, sql).await;
+            assert_eq!(field_names(schema.fields()), expected, "{sql}");
+        }
+    }
+
+    #[tokio::test]
+    async fn path_projection_names_do_not_clash_with_other_columns() {
+        const DATA: &[u8] = b"{\"id\":\"top\",\"obj0\":{\"id\":\"id0\"}}\n";
+        for sql in ["SELECT *, s.obj0.id FROM S3Object s", "SELECT s.*, s.obj0.id FROM S3Object s"] {
+            let schema = json_lines_plan_schema(DATA, sql).await;
+            assert_eq!(field_names(schema.fields()), ["id", "obj0", "s.obj0[id]"], "{sql}");
+        }
+        for sql in [
+            "SELECT s.obj0.id FROM S3Object s ORDER BY s.id",
+            "SELECT s.obj0.id FROM S3Object s ORDER BY id",
+            "SELECT s.obj0.id FROM S3Object s ORDER BY upper(s.id)",
+        ] {
+            let schema = json_lines_plan_schema(DATA, sql).await;
+            assert_eq!(field_names(schema.fields()), ["s.obj0[id]"], "{sql}");
+        }
+        let schema = json_lines_plan_schema(DATA, "SELECT s.obj0.id, (s.id) FROM S3Object s").await;
+        assert_eq!(field_names(schema.fields()), ["s.obj0[id]", "id"]);
+    }
+
+    #[tokio::test]
+    async fn path_aliases_do_not_collide_with_retained_path_names() {
+        const DATA: &[u8] = br#"{"id":"top","obj0":{"id":"a"},"obj1":{"id":"b"},"obj2":{"s.obj0[id]":"c","obj0":"d"}}"#;
+        for (sql, expected) in [
+            (
+                "SELECT s.obj0.id, s.obj1.id, s.obj2.\"s.obj0[id]\" FROM S3Object s",
+                vec!["s.obj0[id]", "s.obj1[id]", "s.obj2[s.obj0[id]]"],
+            ),
+            (
+                "SELECT s.obj2.\"s.obj0[id]\", s.obj1.id, s.obj0.id FROM S3Object s",
+                vec!["s.obj2[s.obj0[id]]", "s.obj1[id]", "s.obj0[id]"],
+            ),
+            ("SELECT (s.obj0.id), s.obj2.\"s.obj0[id]\" FROM S3Object s", vec!["id", "s.obj0[id]"]),
+            (
+                "SELECT s.obj0.id, s.obj2.\"s.obj0[id]\" FROM S3Object s ORDER BY s.id",
+                vec!["s.obj0[id]", "s.obj2[s.obj0[id]]"],
+            ),
+            ("SELECT s.obj2.obj0 FROM S3Object s ORDER BY s.obj0.id", vec!["s.obj2[obj0]"]),
+            ("SELECT s.obj2.obj0 FROM S3Object s ORDER BY (s.obj0.id)", vec!["s.obj2[obj0]"]),
+        ] {
+            let schema = json_lines_plan_schema(DATA, sql).await;
+            assert_eq!(field_names(schema.fields()), expected, "{sql}");
+        }
+    }
+
+    const PATH_NAMES_DATA: &[u8] = br#"{"id":"A","x":"B","n":1,"obj":{"lower(s.id)":7,"count(*)":3,"id":"x","ID":"X","s.n":2,"s.n + Int64(1)":3},"obj0":{"id":"a"},"obj1":{"id":"b"},"obj2":{"s.obj0[id]":"c","obj0":"d"},"a":{"id":1},"b":{"id":2}}
+"#;
+
+    /// Plans and runs `sql` over a stored JSON LINES object through the
+    /// production dispatcher. Returns the planned output names and the rows.
+    async fn production_json_lines_query(bucket: &str, key: &str, sql: &str) -> QueryResult<(Vec<String>, Vec<Vec<String>>)> {
+        let mut input = (*json_snapshot_input()).clone();
+        input.bucket = bucket.to_owned();
+        input.key = key.to_owned();
+        input.request.expression = sql.to_owned();
+        let input = Arc::new(input);
+        let snapshot = snapshot_test_env().await.prepare_select_object_snapshot(bucket, key).await;
+        let dispatcher = production_dispatcher(Arc::clone(&input));
+        let query = Query::new_with_snapshot(QueryContext { input }, sql.to_owned(), snapshot);
+        let machine = dispatcher.build_query_state_machine(query).await?;
+        let plan = dispatcher
+            .build_logical_plan(Arc::clone(&machine))
+            .await?
+            .expect("SELECT should produce a logical plan");
+        let Plan::Query(query_plan) = &plan;
+        let names = query_plan
+            .df_plan
+            .schema()
+            .fields()
+            .iter()
+            .map(|field| field.name().clone())
+            .collect();
+        let Output::StreamData(stream) = dispatcher.execute_logical_plan(plan, machine).await? else {
+            panic!("SELECT should return rows");
+        };
+        let batches: Vec<_> = stream.try_collect().await?;
+        let rows = batches
+            .iter()
+            .flat_map(|batch| {
+                (0..batch.num_rows()).map(|row| {
+                    batch
+                        .columns()
+                        .iter()
+                        .map(|column| array_value_to_string(column, row).expect("display a result value"))
+                        .collect()
+                })
+            })
+            .collect();
+        Ok((names, rows))
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn production_path_alias_does_not_duplicate_a_retained_path_name() {
+        const BUCKET: &str = "s3select-retained-path-name";
+        const KEY: &str = "retained.jsonl";
+        let env = snapshot_test_env().await;
+        env.make_bucket(BUCKET, false).await;
+        env.put_object_bytes(
+            BUCKET,
+            KEY,
+            br#"{"obj0":{"id":"a"},"obj1":{"id":"b"},"obj2":{"s.obj0[id]":"c"}}
+"#
+            .to_vec(),
+        )
+        .await;
+        let (names, rows) =
+            production_json_lines_query(BUCKET, KEY, "SELECT s.obj0.id, s.obj1.id, s.obj2.\"s.obj0[id]\" FROM S3Object s")
+                .await
+                .expect("the query plans and runs");
+        assert_eq!(names, ["s.obj0[id]", "s.obj1[id]", "s.obj2[s.obj0[id]]"]);
+        assert_eq!(rows, [["a", "b", "c"]]);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn production_path_projections_take_their_key_only_when_it_is_a_unique_output_name() {
+        const BUCKET: &str = "s3select-path-projection-names";
+        const KEY: &str = "names.jsonl";
+        let env = snapshot_test_env().await;
+        env.make_bucket(BUCKET, false).await;
+        env.put_object_bytes(BUCKET, KEY, PATH_NAMES_DATA.to_vec()).await;
+        let cases: &[(&str, &[&str])] = &[
+            (
+                "SELECT s.obj.\"lower(s.id)\", LOWER(s.id) FROM S3Object s",
+                &["s.obj[lower(s.id)]", "lower(s.id)"],
+            ),
+            (
+                "SELECT LOWER(s.id), s.obj.\"lower(s.id)\" FROM S3Object s",
+                &["lower(s.id)", "s.obj[lower(s.id)]"],
+            ),
+            (
+                "SELECT s.obj.\"count(*)\", COUNT(*) FROM S3Object s GROUP BY s.obj.\"count(*)\"",
+                &["s.obj[count(*)]", "count(*)"],
+            ),
+            ("SELECT s.obj0.id, COUNT(*) FROM S3Object s GROUP BY s.obj0.id", &["id", "count(*)"]),
+            ("SELECT s.obj0.id, LOWER(s.id) FROM S3Object s", &["id", "lower(s.id)"]),
+            ("SELECT s.obj0.id, s.x AS id FROM S3Object s", &["s.obj0[id]", "id"]),
+            ("SELECT s.x AS id, s.obj0.id FROM S3Object s", &["id", "s.obj0[id]"]),
+            (
+                "SELECT *, s.obj0.id FROM S3Object s",
+                &["id", "x", "n", "obj", "obj0", "obj1", "obj2", "a", "b", "s.obj0[id]"],
+            ),
+            ("SELECT s.obj.ID, s.id FROM S3Object s", &["s.obj[id]", "id"]),
+            ("SELECT s.obj.\"ID\", s.id FROM S3Object s", &["ID", "id"]),
+            ("SELECT s.obj.\"s.n\", CAST(s.n AS INT) FROM S3Object s", &["s.obj[s.n]", "s.n"]),
+            (
+                "SELECT s.n + 1, s.obj.\"s.n + Int64(1)\" FROM S3Object s",
+                &["s.n + Int64(1)", "s.obj[s.n + Int64(1)]"],
+            ),
+            ("SELECT s.obj0.id FROM S3Object s ORDER BY s.id", &["s.obj0[id]"]),
+            ("SELECT s.obj0.id FROM S3Object s ORDER BY s.obj0.id", &["id"]),
+            ("SELECT (s.obj0.id) FROM S3Object s", &["id"]),
+            ("SELECT s.obj0.id, (s.id) FROM S3Object s", &["s.obj0[id]", "id"]),
+            ("SELECT s.a.id, s.b.id FROM S3Object s", &["s.a[id]", "s.b[id]"]),
+            ("SELECT s.a.id, s.b.id, s.obj2.obj0 FROM S3Object s", &["s.a[id]", "s.b[id]", "obj0"]),
+            ("SELECT s.obj2.\"s.obj0[id]\", s.obj0.id FROM S3Object s", &["s.obj0[id]", "id"]),
+            (
+                "SELECT s.obj2.\"s.obj0[id]\", s.obj1.id, s.obj0.id FROM S3Object s",
+                &["s.obj2[s.obj0[id]]", "s.obj1[id]", "s.obj0[id]"],
+            ),
+            (
+                "SELECT s.obj2.\"s.obj0[id]\", s.obj0.id, LOWER(s.id) FROM S3Object s",
+                &["s.obj0[id]", "id", "lower(s.id)"],
+            ),
+        ];
+        for (sql, expected) in cases {
+            let (names, _) = production_json_lines_query(BUCKET, KEY, sql)
+                .await
+                .unwrap_or_else(|error| panic!("{sql}: {error}"));
+            assert_eq!(names, *expected, "{sql}");
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn production_path_projection_naming_keeps_queries_planning() {
+        const BUCKET: &str = "s3select-path-projection-planning";
+        const KEY: &str = "planning.jsonl";
+        let env = snapshot_test_env().await;
+        env.make_bucket(BUCKET, false).await;
+        env.put_object_bytes(BUCKET, KEY, PATH_NAMES_DATA.to_vec()).await;
+        env.put_object_bytes(BUCKET, "no-top-level-id.jsonl", br#"{"obj0":{"id":"a"},"obj1":{"id":"b"}}"#.to_vec())
+            .await;
+        let mut failures = Vec::new();
+        // GROUP BY resolves a SELECT alias when no input column has its name.
+        for sql in [
+            "SELECT s.obj0.id FROM S3Object s GROUP BY id",
+            "SELECT s.obj0.id, s.obj1.id AS k FROM S3Object s GROUP BY id, k",
+        ] {
+            if let Err(error) = production_json_lines_query(BUCKET, "no-top-level-id.jsonl", sql).await {
+                failures.push(format!("{sql}: {error}"));
+            }
+        }
+        // Each query also plans when every path projection keeps DataFusion's name.
+        for sql in [
+            "SELECT s.obj0.id FROM S3Object s",
+            "SELECT obj0.id FROM S3Object",
+            "SELECT s.obj0.id FROM S3Object s WHERE s.obj0.id = 'a'",
+            "SELECT s.obj0.id FROM S3Object s WHERE s.obj0.id IN ('a', 'b') LIMIT 1",
+            "SELECT s.obj0.id FROM S3Object s LIMIT 0",
+            "SELECT s.obj0.id FROM S3Object s ORDER BY s.obj0.id DESC LIMIT 1",
+            "SELECT s.obj0.id FROM S3Object s ORDER BY id",
+            "SELECT s.obj0.id FROM S3Object s ORDER BY s.id",
+            "SELECT s.obj0.id FROM S3Object s ORDER BY upper(s.id)",
+            "SELECT s.obj0.id AS k FROM S3Object s ORDER BY k",
+            "SELECT s.obj0.id, s.obj2.obj0 FROM S3Object s ORDER BY s.obj0.id",
+            "SELECT s.obj2.obj0 FROM S3Object s ORDER BY (s.obj0.id)",
+            "SELECT s.obj0.id, s.id FROM S3Object s ORDER BY 2",
+            "SELECT s.obj0.id FROM S3Object s GROUP BY s.obj0.id",
+            "SELECT s.obj0.id, COUNT(*) FROM S3Object s GROUP BY s.obj0.id",
+            "SELECT s.obj0.id, COUNT(*) AS id FROM S3Object s GROUP BY s.obj0.id",
+            "SELECT COUNT(*), s.obj0.id FROM S3Object s GROUP BY s.obj0.id ORDER BY COUNT(*)",
+            "SELECT COUNT(s.obj0.id) FROM S3Object s",
+            "SELECT MAX(s.obj0.id) FROM S3Object s",
+            "SELECT UPPER(s.obj0.id), s.obj0.id FROM S3Object s",
+            "SELECT s.obj0.id, s.obj0.id = 'a' FROM S3Object s",
+            "SELECT s.obj0.id, s.obj0.id AS x FROM S3Object s",
+            "SELECT s.obj0.id, s.id FROM S3Object s",
+            "SELECT s.obj0.id, (s.id) FROM S3Object s",
+            "SELECT (s.obj0.id), s.obj2.\"s.obj0[id]\" FROM S3Object s",
+            "SELECT s.obj0.id AS id, s.a.id FROM S3Object s",
+            "SELECT s.obj0.id, s.x AS id FROM S3Object s",
+            "SELECT s.obj0.id AS \"s.a[id]\", s.b.id FROM S3Object s",
+            "SELECT s.obj.ID, s.id FROM S3Object s",
+            "SELECT s.obj.\"ID\", LOWER(s.id) FROM S3Object s",
+            "SELECT s.obj.id, s.obj0.id FROM S3Object s",
+            "SELECT s.a.id, s.b.id FROM S3Object s",
+            "SELECT s.a.id, s.b.id, s.obj2.obj0 FROM S3Object s",
+            "SELECT *, s.obj0.id FROM S3Object s",
+            "SELECT s.*, s.obj0.id FROM S3Object s",
+            "SELECT s.obj.\"lower(s.id)\", LOWER(s.id) FROM S3Object s",
+            "SELECT LOWER(s.id), s.obj.\"lower(s.id)\" FROM S3Object s",
+            "SELECT s.obj.\"count(*)\", COUNT(*) FROM S3Object s GROUP BY s.obj.\"count(*)\"",
+            "SELECT s.obj.\"s.n\", CAST(s.n AS INT) FROM S3Object s",
+            "SELECT s.obj.\"s.n + Int64(1)\", s.n + 1 FROM S3Object s",
+            "SELECT s.obj0.id, CAST(s.n AS INT) FROM S3Object s",
+            "SELECT s.obj0.id, s.obj1.id, s.obj2.\"s.obj0[id]\" FROM S3Object s",
+            "SELECT s.obj2.\"s.obj0[id]\", s.obj1.id, s.obj0.id FROM S3Object s",
+            "SELECT s.obj2.\"s.obj0[id]\", s.obj0.id FROM S3Object s",
+            "SELECT s.obj2.\"s.obj0[id]\", s.obj0.id, LOWER(s.id) FROM S3Object s",
+            "SELECT (s.obj0.id), (s.obj1.id) FROM S3Object s",
+            "SELECT s.obj.\"ID\", s.obj0.id FROM S3Object s",
+        ] {
+            match production_json_lines_query(BUCKET, KEY, sql).await {
+                Ok((names, _)) if names.iter().collect::<std::collections::HashSet<_>>().len() != names.len() => {
+                    failures.push(format!("{sql}: duplicate names {names:?}"));
+                }
+                Ok(_) => {}
+                Err(error) => failures.push(format!("{sql}: {error}")),
+            }
+        }
+        assert!(failures.is_empty(), "{failures:#?}");
+    }
+
     #[tokio::test]
     async fn csv_query_uses_custom_record_delimiter_across_file_partitions() {
         const ROW_COUNT: usize = 200_000;
@@ -2841,4 +3183,110 @@ mod tests {
         assert_eq!(admission.available_permits(), 1);
         assert!(output.next().await.is_none());
     }
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn path_alias_does_not_collide_with_computed_output_name() {
+        let mut input = (*json_snapshot_input()).clone();
+        input.bucket = "s3select-review-name-collision".to_string();
+        input.key = "collision.jsonl".to_string();
+        input.request.expression = "SELECT s.obj.\"lower(s.id)\", LOWER(s.id) FROM S3Object s".to_string();
+        let input = Arc::new(input);
+        let env = snapshot_test_env().await;
+        env.make_bucket(&input.bucket, false).await;
+        env.put_object_bytes(&input.bucket, &input.key, b"{\"id\":\"A\",\"obj\":{\"lower(s.id)\":7}}\n".to_vec())
+            .await;
+        let snapshot = env.prepare_select_object_snapshot(&input.bucket, &input.key).await;
+        let dispatcher = production_dispatcher(Arc::clone(&input));
+        let query = Query::new_with_snapshot(
+            QueryContext {
+                input: Arc::clone(&input),
+            },
+            input.request.expression.clone(),
+            snapshot,
+        );
+        let machine = dispatcher.build_query_state_machine(query).await.expect("production session");
+        let result = dispatcher.build_logical_plan(Arc::clone(&machine)).await;
+        if let Err(error) = &result {
+            panic!(
+                "real production planner rejected a previously valid query: {error}; S3 classification={:?}",
+                error.s3_select_policy_error()
+            );
+        }
+        let plan = result.expect("valid production query").expect("query plan");
+        let Plan::Query(query_plan) = &plan;
+        let schema = query_plan.df_plan.schema();
+        assert_eq!(schema.fields().len(), 2);
+        assert_ne!(schema.field(0).name(), schema.field(1).name());
+        let output = dispatcher
+            .execute_logical_plan(plan, machine)
+            .await
+            .expect("execute production query");
+        let Output::StreamData(stream) = output else {
+            panic!("query must return rows");
+        };
+        let batches: Vec<_> = stream.try_collect().await.expect("read production result");
+        assert_eq!(batches.iter().map(|batch| batch.num_rows()).sum::<usize>(), 1);
+        let batch = batches.iter().find(|batch| batch.num_rows() == 1).expect("one result row");
+        let key = batch
+            .column(0)
+            .as_any()
+            .downcast_ref::<datafusion::arrow::array::Int64Array>()
+            .expect("integer key");
+        let lower = batch
+            .column(1)
+            .as_any()
+            .downcast_ref::<datafusion::arrow::array::StringArray>()
+            .expect("lowercase string");
+        assert_eq!(key.value(0), 7);
+        assert_eq!(lower.value(0), "a");
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn path_alias_does_not_collide_with_computed_sort_name() {
+        let mut input = (*json_snapshot_input()).clone();
+        input.bucket = "s3select-sort-name-collision".to_string();
+        input.key = "collision.jsonl".to_string();
+        input.request.expression = "SELECT s.obj.\"lower(s.id)\" FROM S3Object s ORDER BY LOWER(s.id)".to_string();
+        let input = Arc::new(input);
+        let env = snapshot_test_env().await;
+        env.make_bucket(&input.bucket, false).await;
+        env.put_object_bytes(&input.bucket, &input.key, b"{\"id\":\"A\",\"obj\":{\"lower(s.id)\":7}}\n".to_vec())
+            .await;
+        let snapshot = env.prepare_select_object_snapshot(&input.bucket, &input.key).await;
+        let dispatcher = production_dispatcher(Arc::clone(&input));
+        let query = Query::new_with_snapshot(
+            QueryContext {
+                input: Arc::clone(&input),
+            },
+            input.request.expression.clone(),
+            snapshot,
+        );
+        let machine = dispatcher.build_query_state_machine(query).await.expect("production session");
+        let result = dispatcher.build_logical_plan(Arc::clone(&machine)).await;
+        if let Err(error) = &result {
+            panic!(
+                "real production planner rejected a previously valid query: {error}; S3 classification={:?}",
+                error.s3_select_policy_error()
+            );
+        }
+        let plan = result.expect("valid production query").expect("query plan");
+        let Plan::Query(query_plan) = &plan;
+        let schema = query_plan.df_plan.schema();
+        assert_eq!(schema.fields().len(), 1);
+        let output = dispatcher
+            .execute_logical_plan(plan, machine)
+            .await
+            .expect("execute production query");
+        let Output::StreamData(stream) = output else {
+            panic!("query must return rows");
+        };
+        let batches: Vec<_> = stream.try_collect().await.expect("read production result");
+        assert_eq!(batches.iter().map(|batch| batch.num_rows()).sum::<usize>(), 1);
+        let batch = batches.iter().find(|batch| batch.num_rows() == 1).expect("one result row");
+        let key = batch
+            .column(0)
+            .as_any()
+            .downcast_ref::<datafusion::arrow::array::Int64Array>()
+            .expect("integer key");
+        assert_eq!(key.value(0), 7);
+    }
 }

crates/s3select-query/src/dispatcher/mod.rs

@@ -14,5 +14,6 @@
 
 mod json_document_schema;
 mod json_document_table;
+mod json_lines_schema;
 pub mod manager;
 mod parquet_table;

crates/s3select-query/src/sql/planner.rs

@@ -12,14 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{convert::Infallible, ops::ControlFlow};
+use std::{collections::HashMap, convert::Infallible, ops::ControlFlow};
 
 use async_trait::async_trait;
-use datafusion::sql::{
-    planner::{IdentNormalizer, SqlToRel},
-    sqlparser::ast::{
-        AccessExpr, Expr, GroupByExpr, Ident, JsonPath, JsonPathElem, ObjectNamePart, OrderByKind, Query, Select, SelectFlavor,
-        SetExpr, Statement, Subscript, TableAlias, TableFactor, Value, Visit, VisitMut, Visitor, VisitorMut,
+use datafusion::{
+    common::{Column, Result as DFResult},
+    logical_expr::{Expr as LogicalExpr, LogicalPlan, LogicalPlanBuilder},
+    sql::{
+        planner::{IdentNormalizer, SqlToRel},
+        sqlparser::ast::{
+            AccessExpr, Expr, GroupByExpr, Ident, JsonPath, JsonPathElem, ObjectNamePart, OrderBy, OrderByKind, Query, Select,
+            SelectFlavor, SelectItem, SetExpr, Statement, Subscript, TableAlias, TableFactor, Value, Visit, VisitMut, Visitor,
+            VisitorMut,
+        },
     },
 };
 use rustfs_s3select_api::{
@@ -75,7 +80,9 @@ impl<'a, S: ContextProviderExtension + Send + Sync + 'a> SqlPlanner<'a, S> {
     }
 
     async fn df_prepared_sql_to_plan(&self, stmt: Statement, _session: &SessionCtx) -> QueryResult<Plan> {
+        let keys = path_projection_keys(&stmt);
         let df_plan = self.df_planner.sql_statement_to_plan(stmt).map_err(classify_planner_error)?;
+        let df_plan = name_path_projections(df_plan, &keys)?;
         Ok(Plan::Query(QueryPlan {
             df_plan,
             is_tag_scan: false,
@@ -150,9 +157,203 @@ pub(crate) fn prepare_s3_select_statement(statement: &mut Statement) -> QueryRes
     };
     let mut normalizer = PartiQlSubscriptNormalizer;
     let _ = VisitMut::visit(query, &mut normalizer);
+    if let SetExpr::Select(select) = query.body.as_mut() {
+        alias_path_projections(select, query.order_by.as_ref());
+    }
     Ok(source)
 }
 
+/// Names an unaliased path projection after its last key, as AWS does
+/// (`s.obj0.id` -> `id`), instead of DataFusion's `s.obj0[id]`. Names that
+/// could collide with another output name keep DataFusion's name. Wildcard
+/// columns, computed expressions, and retained paths have unknown output names
+/// here, so they disable aliasing. Unprojected ORDER BY expressions also disable
+/// aliasing because DataFusion adds their input columns to the projection.
+/// GROUP BY can refer to these aliases; [`name_path_projections`] names the
+/// paths left here once the planned output names are known.
+fn alias_path_projections(select: &mut Select, order_by: Option<&OrderBy>) {
+    if has_wildcard(select) || sorts_by_unprojected_expr(select, order_by) {
+        return;
+    }
+    let Some(qualifier) = source_qualifier(select) else {
+        return;
+    };
+    let normalizer = IdentNormalizer::default();
+    let names: Vec<Option<String>> = select
+        .projection
+        .iter()
+        .map(|item| {
+            let ident = match item {
+                SelectItem::UnnamedExpr(Expr::Nested(_)) => None,
+                SelectItem::UnnamedExpr(expr) => column_name(expr),
+                SelectItem::ExprWithAlias { alias, .. } => Some(alias),
+                _ => None,
+            };
+            ident.map(|ident| normalizer.normalize(ident.clone()))
+        })
+        .collect();
+    if names.iter().any(Option::is_none) {
+        return;
+    }
+    let mut counts = HashMap::<String, usize>::new();
+    for name in names.iter().flatten() {
+        *counts.entry(name.clone()).or_default() += 1;
+    }
+    // Retained paths have generated names that cannot be inferred from their last key.
+    // Keep the whole projection unchanged so a sibling alias cannot collide with them.
+    if names.iter().flatten().any(|name| counts.get(name) != Some(&1)) {
+        return;
+    }
+    for item in &mut select.projection {
+        let SelectItem::UnnamedExpr(expr) = item else {
+            continue;
+        };
+        let Some(alias) = path_key(expr, &qualifier).cloned() else {
+            continue;
+        };
+        let expr = std::mem::replace(expr, Expr::Identifier(Ident::new("")));
+        *item = SelectItem::ExprWithAlias { expr, alias };
+    }
+}
+
+/// The output name DataFusion gives a column reference: its last part.
+fn column_name(expr: &Expr) -> Option<&Ident> {
+    match expr {
+        Expr::Identifier(ident) => Some(ident),
+        Expr::CompoundIdentifier(parts) => parts.last(),
+        Expr::Nested(inner) => column_name(inner),
+        _ => None,
+    }
+}
+
+fn has_wildcard(select: &Select) -> bool {
+    select
+        .projection
+        .iter()
+        .any(|item| matches!(item, SelectItem::Wildcard(_) | SelectItem::QualifiedWildcard(..)))
+}
+
+fn sorts_by_unprojected_expr(select: &Select, order_by: Option<&OrderBy>) -> bool {
+    let Some(OrderBy {
+        kind: OrderByKind::Expressions(sort_exprs),
+        ..
+    }) = order_by
+    else {
+        return false;
+    };
+    sort_exprs.iter().any(|sort_expr| {
+        !select
+            .projection
+            .iter()
+            .any(|item| matches!(item, SelectItem::UnnamedExpr(projected) if *projected == sort_expr.expr))
+    })
+}
+
+/// The normalized table alias, or table name, that may prefix a path.
+fn source_qualifier(select: &Select) -> Option<String> {
+    let normalizer = IdentNormalizer::default();
+    match &select.from.first()?.relation {
+        TableFactor::Table { alias: Some(alias), .. } => Some(normalizer.normalize(alias.name.clone())),
+        TableFactor::Table { name, .. } => match name.0.first() {
+            Some(ObjectNamePart::Identifier(table_name)) => Some(normalizer.normalize(table_name.clone())),
+            _ => None,
+        },
+        _ => None,
+    }
+}
+
+/// The last key of a path of at least two keys after the source qualifier
+/// (`s.obj0.id` -> `id`), the name AWS gives the projection.
+fn path_key<'a>(expr: &'a Expr, qualifier: &str) -> Option<&'a Ident> {
+    let parts = match expr {
+        Expr::CompoundIdentifier(parts) => parts,
+        Expr::Nested(inner) => return path_key(inner, qualifier),
+        _ => return None,
+    };
+    let path = match parts.split_first() {
+        Some((first, rest)) if IdentNormalizer::default().normalize(first.clone()) == qualifier => rest,
+        _ => parts.as_slice(),
+    };
+    match path {
+        [_, _, ..] => path.last(),
+        _ => None,
+    }
+}
+
+/// The normalized key of each path projection left unaliased by
+/// [`alias_path_projections`]. Empty when a wildcard hides the output
+/// positions, or when an ORDER BY expression is not projected.
+fn path_projection_keys(statement: &Statement) -> Vec<Option<String>> {
+    let Statement::Query(query) = statement else {
+        return Vec::new();
+    };
+    let SetExpr::Select(select) = query.body.as_ref() else {
+        return Vec::new();
+    };
+    if has_wildcard(select) || sorts_by_unprojected_expr(select, query.order_by.as_ref()) {
+        return Vec::new();
+    }
+    let Some(qualifier) = source_qualifier(select) else {
+        return Vec::new();
+    };
+    let normalizer = IdentNormalizer::default();
+    select
+        .projection
+        .iter()
+        .map(|item| match item {
+            SelectItem::UnnamedExpr(expr) => path_key(expr, &qualifier).map(|key| normalizer.normalize(key.clone())),
+            _ => None,
+        })
+        .collect()
+}
+
+/// Renames each path projection to its key from [`path_projection_keys`]
+/// unless another output column keeps or receives that name. Names come from
+/// the planned schema, so computed, aliased and retained columns all count.
+fn name_path_projections(plan: LogicalPlan, keys: &[Option<String>]) -> DFResult<LogicalPlan> {
+    let schema = plan.schema();
+    if keys.len() != schema.fields().len() {
+        return Ok(plan);
+    }
+    let mut renames: Vec<Option<&str>> = keys
+        .iter()
+        .zip(schema.fields())
+        .map(|(key, field)| key.as_deref().filter(|key| key != field.name()))
+        .collect();
+    // Dropping a rename brings back the generated name, which can collide with another rename.
+    loop {
+        let mut counts = HashMap::<&str, usize>::new();
+        for (rename, field) in renames.iter().zip(schema.fields()) {
+            *counts.entry(rename.unwrap_or(field.name())).or_default() += 1;
+        }
+        let mut dropped = false;
+        for rename in &mut renames {
+            if rename.is_some_and(|name| counts[name] > 1) {
+                *rename = None;
+                dropped = true;
+            }
+        }
+        if !dropped {
+            break;
+        }
+    }
+    if renames.iter().all(Option::is_none) {
+        return Ok(plan);
+    }
+    let columns: Vec<_> = schema
+        .iter()
+        .zip(renames)
+        .map(|(column, rename)| {
+            let column = LogicalExpr::Column(Column::from(column));
+            match rename {
+                Some(name) => column.alias(name),
+                None => column,
+            }
+        })
+        .collect();
+    LogicalPlanBuilder::from(plan).project(columns)?.build()
+}
+
 struct PartiQlSubscriptNormalizer;
 
 impl VisitorMut for PartiQlSubscriptNormalizer {
@@ -374,7 +575,7 @@ impl Visitor for SubqueryDetector {
 
 #[cfg(test)]
 mod tests {
-    use super::prepare_s3_select_statement;
+    use super::{path_projection_keys, prepare_s3_select_statement};
     use crate::sql::parser::ExtParser;
     use datafusion::sql::sqlparser::ast::{AccessExpr, Expr, Statement, Visit, Visitor};
     use rustfs_s3select_api::{
@@ -596,6 +797,86 @@ mod tests {
         }
     }
 
+    fn prepared(sql: &str) -> String {
+        let mut statement = parse_statement(sql);
+        prepare_s3_select_statement(&mut statement).expect("statement should be supported");
+        statement.to_string()
+    }
+
+    #[test]
+    fn unaliased_path_projections_are_named_after_their_last_key() {
+        for (sql, expected) in [
+            ("SELECT s.obj0.id FROM S3Object AS s", "SELECT s.obj0.id AS id FROM S3Object AS s"),
+            (
+                "SELECT s.a.b.c, s.x.\"Y\" FROM S3Object AS s",
+                "SELECT s.a.b.c AS c, s.x.\"Y\" AS \"Y\" FROM S3Object AS s",
+            ),
+            ("SELECT obj0.id FROM S3Object", "SELECT obj0.id AS id FROM S3Object"),
+            (
+                "SELECT s.obj0.id FROM S3Object AS s WHERE s.obj0.id = 'a' ORDER BY s.obj0.id",
+                "SELECT s.obj0.id AS id FROM S3Object AS s WHERE s.obj0.id = 'a' ORDER BY s.obj0.id",
+            ),
+        ] {
+            assert_eq!(prepared(sql), expected, "{sql}");
+        }
+    }
+
+    #[test]
+    fn projections_keep_their_names_when_a_key_alias_is_absent_or_ambiguous() {
+        for sql in [
+            "SELECT s.id FROM S3Object AS s",
+            "SELECT S3Object.id FROM S3Object",
+            "SELECT s.tags[1] FROM S3Object AS s",
+            "SELECT s.obj0.id AS x FROM S3Object AS s",
+            "SELECT s.obj0.id, s.obj1.id FROM S3Object AS s",
+            "SELECT s.obj0.ID, s.obj1.id FROM S3Object AS s",
+            "SELECT s.obj0.id, s.id FROM S3Object AS s",
+            "SELECT s.obj0.id, s.name AS id FROM S3Object AS s",
+            "SELECT s.obj0.\"Id\", s.obj1.\"Id\" FROM S3Object AS s",
+            "SELECT *, s.obj0.id FROM S3Object AS s",
+            "SELECT s.*, s.obj0.id FROM S3Object AS s",
+            "SELECT s.obj0.id FROM S3Object AS s ORDER BY s.id",
+            "SELECT s.obj0.id FROM S3Object AS s ORDER BY upper(id)",
+            "SELECT s.obj0.id, (s.id) FROM S3Object AS s",
+            "SELECT s.obj.\"lower(s.id)\", LOWER(s.id) FROM S3Object AS s",
+            "SELECT s.obj.\"count(*)\", COUNT(*) FROM S3Object AS s GROUP BY s.obj.\"count(*)\"",
+            "SELECT s.obj0.id, COUNT(*) FROM S3Object AS s GROUP BY s.obj0.id",
+        ] {
+            assert_eq!(prepared(sql), sql);
+        }
+    }
+
+    #[test]
+    fn path_projection_keys_cover_paths_left_unaliased() {
+        let keys = |sql: &str| {
+            let mut statement = parse_statement(sql);
+            prepare_s3_select_statement(&mut statement).expect("statement should be supported");
+            path_projection_keys(&statement)
+        };
+        let key = |key: &str| Some(key.to_owned());
+        for (sql, expected) in [
+            ("SELECT s.obj0.id FROM S3Object AS s", vec![None]),
+            ("SELECT s.obj0.id, s.obj1.id FROM S3Object AS s", vec![key("id"), key("id")]),
+            (
+                "SELECT s.a.b.c, s.x.\"Y\", s.obj0.ID, lower(s.id) FROM S3Object AS s",
+                vec![key("c"), key("Y"), key("id"), None],
+            ),
+            ("SELECT obj0.id, COUNT(*) FROM S3Object GROUP BY obj0.id", vec![key("id"), None]),
+            ("SELECT (s.obj0.id), ((s.obj1.id)) FROM S3Object AS s", vec![key("id"), key("id")]),
+            (
+                "SELECT s.id, s.tags[1], s.obj0.id AS x, (s.id) FROM S3Object AS s",
+                vec![None, None, None, None],
+            ),
+            ("SELECT S3Object.id, lower(id) FROM S3Object", vec![None, None]),
+            ("SELECT *, s.obj0.id FROM S3Object AS s", vec![]),
+            ("SELECT s.*, s.obj0.id FROM S3Object AS s", vec![]),
+            ("SELECT s.obj0.id FROM S3Object AS s ORDER BY s.id", vec![]),
+            ("SELECT s.obj0.id FROM S3Object AS s ORDER BY upper(id)", vec![]),
+        ] {
+            assert_eq!(keys(sql), expected, "{sql}");
+        }
+    }
+
     #[test]
     fn accepts_group_by_and_order_by() {
         let statement = parse_statement("SELECT department, COUNT(*) FROM S3Object GROUP BY department ORDER BY department");

crates/scanner/Cargo.toml

@@ -31,6 +31,7 @@ workspace = true
 
 [features]
 default = []
+e2e-test-hooks = []
 diagnostic-read-version-caller = ["rustfs-ecstore/diagnostic-read-version-caller"]
 hotpath = [
     "hotpath/hotpath",

crates/scanner/src/e2e_barrier.rs

@@ -0,0 +1,222 @@
+// Copyright 2026 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.
+
+//! Test-cluster observations and bounded pauses at real scanner boundaries.
+//! This module is absent unless `e2e-test-hooks` is explicitly enabled.
+
+pub(crate) mod worker_observation;
+
+use serde::Deserialize;
+use serde_json::{Value, json};
+use std::io::{self, Write};
+use std::path::{Component, Path};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+
+const MAX_BYTES: u64 = 1024 * 1024;
+const MAX_EVENTS: u64 = 512;
+const MAX_WAIT: Duration = Duration::from_secs(120);
+static SEQUENCE: AtomicU64 = AtomicU64::new(0);
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Session {
+    nonce: String,
+    buckets: Vec<String>,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Arm {
+    nonce: String,
+    stage: String,
+    pid: u32,
+    owner_id: Option<String>,
+    instance_id: Option<String>,
+    epoch: Option<u64>,
+    cycle: Option<u64>,
+}
+
+fn valid_component(value: &str) -> bool {
+    !value.is_empty()
+        && value.len() <= 64
+        && value
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_')
+}
+
+fn read_control<T: serde::de::DeserializeOwned>(path: &Path) -> io::Result<Option<T>> {
+    let metadata = match std::fs::symlink_metadata(path) {
+        Ok(metadata) => metadata,
+        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => return Err(error),
+    };
+    if !metadata.is_file() || metadata.len() > MAX_BYTES {
+        return Err(io::Error::new(io::ErrorKind::InvalidData, "invalid scanner test control file"));
+    }
+    let data = std::fs::read(path)?;
+    if data.len() > MAX_BYTES as usize {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidData,
+            "scanner test control file grew past its bound",
+        ));
+    }
+    serde_json::from_slice(&data).map(Some).map_err(io::Error::other)
+}
+
+fn write_observation(directory: &Path, name: &str, bytes: &[u8]) -> io::Result<()> {
+    let temporary = directory.join(format!(".{name}.tmp"));
+    let mut output = std::fs::OpenOptions::new().write(true).create_new(true).open(&temporary)?;
+    output.write_all(bytes)?;
+    output.sync_all()?;
+    std::fs::rename(temporary, directory.join(name))
+}
+
+/// Observe a real event and, only for an explicitly armed matching session,
+/// wait for its release file. No test control can supply scanner state.
+pub async fn checkpoint(stage: &str, detail: Value) -> io::Result<()> {
+    let Some(directory) = std::env::var_os("RUSTFS_E2E_SCANNER_BARRIER_DIR") else {
+        return Ok(());
+    };
+    let directory = Path::new(&directory);
+    if !directory.is_absolute()
+        || directory.components().any(|part| matches!(part, Component::ParentDir))
+        || !valid_component(stage)
+    {
+        return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid scanner test barrier path or stage"));
+    }
+    let Some(session) = read_control::<Session>(&directory.join("session.json"))? else {
+        return Ok(());
+    };
+    if !valid_component(&session.nonce)
+        || session.buckets.is_empty()
+        || session.buckets.len() > 32
+        || session.buckets.iter().any(|bucket| bucket.is_empty() || bucket.len() > 255)
+    {
+        return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid scanner test session"));
+    }
+    // Admission may enable the recorder while this actual start gate is held.
+    // Begin on release, before any snapshot or worker; other boundaries stay synchronous here.
+    if stage != "before_start_snapshot" {
+        worker_observation::boundary(stage, &detail);
+    }
+    if detail
+        .get("bucket")
+        .and_then(Value::as_str)
+        .is_some_and(|bucket| !session.buckets.iter().any(|allowed| allowed == bucket))
+    {
+        return Ok(());
+    }
+    let arm = read_control::<Arm>(&directory.join("arm.json"))?;
+    let armed = arm.is_some_and(|arm| {
+        arm.nonce == session.nonce
+            && arm.stage == stage
+            && arm.pid == std::process::id()
+            && arm
+                .owner_id
+                .as_ref()
+                .is_none_or(|owner| detail.get("owner_id").and_then(Value::as_str) == Some(owner.as_str()))
+            && arm
+                .instance_id
+                .as_ref()
+                .is_none_or(|instance| detail.get("instance_id").and_then(Value::as_str) == Some(instance.as_str()))
+            && arm
+                .epoch
+                .is_none_or(|epoch| detail.get("epoch").and_then(Value::as_u64) == Some(epoch))
+            && arm
+                .cycle
+                .is_none_or(|cycle| detail.get("cycle").and_then(Value::as_u64) == Some(cycle))
+    });
+    let sequence = SEQUENCE.fetch_add(1, Ordering::Relaxed);
+    if sequence >= MAX_EVENTS {
+        return Err(io::Error::other("scanner test observation limit reached"));
+    }
+    let pid = std::process::id();
+    let event = json!({
+        "nonce": session.nonce,
+        "pid": pid,
+        "sequence": sequence,
+        "stage": stage,
+        "armed": armed,
+        "detail": detail,
+    });
+    let encoded = serde_json::to_vec(&event).map_err(io::Error::other)?;
+    if encoded.len() > MAX_BYTES as usize {
+        return Err(io::Error::other("scanner test observation exceeds its byte bound"));
+    }
+    let suffix = format!("{}.{}.{}.json", session.nonce, pid, sequence);
+    write_observation(directory, &format!("observed.{suffix}"), &encoded)?;
+    if !armed {
+        if stage == "before_start_snapshot" {
+            worker_observation::boundary(stage, &event["detail"]);
+        }
+        return Ok(());
+    }
+    write_observation(directory, &format!("reached.{suffix}"), &encoded)?;
+    let release = directory.join(format!("release.{}.{}.{}", session.nonce, stage, pid));
+    let started = Instant::now();
+    while !release.try_exists()? {
+        if started.elapsed() >= MAX_WAIT {
+            return Err(io::Error::new(io::ErrorKind::TimedOut, "scanner test barrier release timed out"));
+        }
+        tokio::time::sleep(Duration::from_millis(25)).await;
+    }
+    if stage == "before_start_snapshot" {
+        worker_observation::boundary(stage, &event["detail"]);
+    }
+    Ok(())
+}
+
+pub(crate) fn hex(bytes: &[u8]) -> String {
+    hex_simd::encode_to_string(bytes, hex_simd::AsciiCase::Lower)
+}
+
+pub(crate) fn acknowledgement_observations(acknowledgements: &[crate::scanner::ScannerDirtyUsageAcknowledgement]) -> Value {
+    Value::Array(
+        acknowledgements
+            .iter()
+            .map(|acknowledgement| {
+                use crate::scanner::ScannerDirtyUsageAcknowledgementKind;
+                match &acknowledgement.kind {
+                    ScannerDirtyUsageAcknowledgementKind::Generation(generation) => json!({
+                        "host": acknowledgement.host,
+                        "instance_id": acknowledgement.instance_id,
+                        "kind": "generation",
+                        "generation": generation,
+                    }),
+                    ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries } => json!({
+                        "host": acknowledgement.host,
+                        "owner_id": owner_id,
+                        "instance_id": acknowledgement.instance_id,
+                        "kind": "scoped",
+                        "entries": entries.iter().map(|entry| json!({
+                            "bucket": entry.bucket,
+                            "bucket_incarnation": entry.bucket_incarnation,
+                            "generation": entry.generation,
+                        })).collect::<Vec<_>>(),
+                    }),
+                }
+            })
+            .collect(),
+    )
+}
+
+pub(crate) fn activity_observations(activity: &crate::scanner::ScannerActivitySnapshot) -> Value {
+    Value::Object(
+        activity
+            .iter()
+            .map(|(host, activity)| (host.clone(), activity.e2e_observation()))
+            .collect(),
+    )
+}

crates/scanner/src/e2e_barrier/worker_observation.rs

@@ -0,0 +1,558 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+
+//! N-client evidence only. This recorder never waits for a release or filters buckets.
+
+use super::{read_control, valid_component, write_observation};
+use serde::{Deserialize, Serialize};
+use serde_json::{Value, json};
+use sha2::{Digest, Sha256};
+use std::collections::{BTreeMap, BTreeSet};
+use std::future::Future;
+use std::io;
+use std::path::{Component, Path, PathBuf};
+use std::sync::{Mutex, OnceLock};
+use uuid::Uuid;
+
+const MAX_EVENTS: usize = 512;
+const MAX_CYCLES: usize = 8;
+const MAX_BYTES: usize = 1024 * 1024;
+static OBSERVER: OnceLock<Mutex<Option<Observer>>> = OnceLock::new();
+tokio::task_local! { static INCARNATION: Option<Uuid>; }
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct Control {
+    nonce: String,
+    pid: u32,
+}
+
+#[derive(Serialize)]
+struct Request {
+    context: Value,
+    opening_sequence: usize,
+    complete_sequence: Option<usize>,
+    progress: Option<[u64; 3]>,
+}
+
+#[derive(Default)]
+struct Cycle {
+    start_sequence: usize,
+    events: Vec<Value>,
+    requests: BTreeMap<String, Request>,
+    local_walks: Vec<Value>,
+    candidate: Option<Value>,
+    closed: bool,
+}
+
+struct Observer {
+    directory: PathBuf,
+    nonce: String,
+    pid: u32,
+    sequence: usize,
+    errors: BTreeSet<&'static str>,
+    cycles: BTreeMap<(u64, u64), Cycle>,
+    manifests: Vec<Value>,
+}
+
+fn invalid(code: &'static str) -> io::Error {
+    io::Error::other(code)
+}
+
+fn cycle_key(detail: &Value) -> io::Result<(u64, u64)> {
+    Ok((
+        detail["epoch"].as_u64().ok_or_else(|| invalid("malformed"))?,
+        detail["cycle"].as_u64().ok_or_else(|| invalid("malformed"))?,
+    ))
+}
+
+fn uuid_field(detail: &Value, name: &str) -> io::Result<()> {
+    let value = detail[name].as_str().ok_or_else(|| invalid("malformed"))?;
+    if Uuid::parse_str(value).is_ok_and(|id| !id.is_nil()) {
+        Ok(())
+    } else {
+        Err(invalid("malformed"))
+    }
+}
+
+impl Observer {
+    fn new(directory: PathBuf, control: Control) -> Self {
+        Self {
+            directory,
+            nonce: control.nonce,
+            pid: control.pid,
+            sequence: 0,
+            errors: BTreeSet::new(),
+            cycles: BTreeMap::new(),
+            manifests: Vec::new(),
+        }
+    }
+
+    fn health(&self) -> Value {
+        json!({"schema": 1, "nonce": self.nonce, "pid": self.pid, "event_count": self.sequence,
+            "max_events": MAX_EVENTS, "max_cycles": MAX_CYCLES, "max_bytes": MAX_BYTES,
+            "errors": self.errors, "manifests": self.manifests})
+    }
+
+    fn persist(&self, name: &str, value: &Value) -> io::Result<String> {
+        let bytes = serde_json::to_vec(value)?;
+        if bytes.len() > MAX_BYTES {
+            return Err(invalid("overflow"));
+        }
+        write_observation(&self.directory, name, &bytes)?;
+        Ok(super::hex(&Sha256::digest(&bytes)))
+    }
+
+    fn event(&mut self, key: (u64, u64), kind: &str, detail: Value) -> io::Result<usize> {
+        if self.sequence >= MAX_EVENTS {
+            return Err(invalid("overflow"));
+        }
+        let sequence = self.sequence;
+        self.sequence += 1;
+        let event = json!({"schema": 1, "nonce": self.nonce, "pid": self.pid,
+            "sequence": sequence, "kind": kind, "epoch": key.0, "cycle": key.1, "detail": detail});
+        let name = format!("worker-observed.{}.{}.{sequence}.json", self.nonce, self.pid);
+        let sha256 = self.persist(&name, &event)?;
+        self.cycles
+            .get_mut(&key)
+            .ok_or_else(|| invalid("outside_cycle"))?
+            .events
+            .push(json!({"file": name, "sha256": sha256, "sequence": sequence, "kind": kind}));
+        Ok(sequence)
+    }
+
+    fn active(&self, key: (u64, u64)) -> io::Result<()> {
+        match self.cycles.get(&key) {
+            Some(cycle) if !cycle.closed => Ok(()),
+            Some(_) => Err(invalid("late")),
+            None => Err(invalid("outside_cycle")),
+        }
+    }
+
+    fn boundary(&mut self, stage: &str, detail: Value) -> io::Result<()> {
+        let key = cycle_key(&detail)?;
+        if stage == "before_start_snapshot" {
+            if self.cycles.len() >= MAX_CYCLES {
+                return Err(invalid("overflow"));
+            }
+            if self.cycles.contains_key(&key) {
+                return Err(invalid("duplicate"));
+            }
+            if self.cycles.values().any(|cycle| !cycle.closed) {
+                return Err(invalid("cross_cycle"));
+            }
+            self.cycles.insert(
+                key,
+                Cycle {
+                    start_sequence: self.sequence,
+                    ..Default::default()
+                },
+            );
+        }
+        self.active(key)?;
+        match stage {
+            "before_start_snapshot" => {
+                self.event(key, "cycle_begin", detail)?;
+            }
+            "bucket_walk" => {
+                if self.cycles[&key].candidate.is_some() {
+                    return Err(invalid("late"));
+                }
+                uuid_field(&detail, "bucket_incarnation")?;
+                uuid_field(&detail, "disk_uuid")?;
+                let sequence = self.event(key, "local_walk", detail.clone())?;
+                self.cycles
+                    .get_mut(&key)
+                    .ok_or_else(|| invalid("outside_cycle"))?
+                    .local_walks
+                    .push(json!({"sequence": sequence, "context": detail}));
+            }
+            "cycle_candidate" => {
+                if self.cycles[&key].candidate.is_some() {
+                    return Err(invalid("duplicate"));
+                }
+                self.event(key, "candidate", detail.clone())?;
+                self.cycles.get_mut(&key).ok_or_else(|| invalid("outside_cycle"))?.candidate = Some(detail);
+            }
+            "root_durable" => self.close(key, detail)?,
+            _ => return Err(invalid("malformed")),
+        }
+        Ok(())
+    }
+
+    fn opening(&mut self, mut context: Value, incarnation: Option<Uuid>) -> io::Result<()> {
+        let key = cycle_key(&context)?;
+        // A removed session ends the measurement window, but never hides a late event for a known cycle.
+        if !self.cycles.contains_key(&key) && !self.directory.join("session.json").try_exists()? {
+            return Ok(());
+        }
+        self.active(key)?;
+        if self.cycles[&key].candidate.is_some() {
+            return Err(invalid("late"));
+        }
+        context["bucket_incarnation"] = json!(incarnation);
+        for field in [
+            "request_id",
+            "wire_session_id",
+            "server_epoch",
+            "bucket_incarnation",
+            "disk_uuid",
+        ] {
+            uuid_field(&context, field)?;
+        }
+        for field in ["pool", "set", "disk_index", "wire_session_sequence"] {
+            context[field].as_u64().ok_or_else(|| invalid("malformed"))?;
+        }
+        for field in [
+            "bucket",
+            "worker_endpoint",
+            "worker_path",
+            "reuse_plan_sha256",
+            "coverage_plan_sha256",
+        ] {
+            if context[field].as_str().is_none_or(str::is_empty) {
+                return Err(invalid("malformed"));
+            }
+        }
+        for field in ["reuse_plan_sha256", "coverage_plan_sha256"] {
+            if context[field]
+                .as_str()
+                .is_none_or(|value| value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()))
+            {
+                return Err(invalid("malformed"));
+            }
+        }
+        let id = context["request_id"]
+            .as_str()
+            .ok_or_else(|| invalid("malformed"))?
+            .to_string();
+        if self.cycles.values().any(|cycle| cycle.requests.contains_key(&id)) {
+            return Err(invalid("duplicate"));
+        }
+        let sequence = self.event(key, "remote_scan_opening", context.clone())?;
+        self.cycles
+            .get_mut(&key)
+            .ok_or_else(|| invalid("outside_cycle"))?
+            .requests
+            .insert(
+                id,
+                Request {
+                    context,
+                    opening_sequence: sequence,
+                    complete_sequence: None,
+                    progress: None,
+                },
+            );
+        Ok(())
+    }
+
+    fn complete(&mut self, id: Uuid, progress: [u64; 3]) -> io::Result<()> {
+        let id = id.to_string();
+        let key = self
+            .cycles
+            .iter()
+            .find_map(|(key, cycle)| cycle.requests.contains_key(&id).then_some(*key));
+        let Some(key) = key else {
+            if !self.directory.join("session.json").try_exists()? {
+                return Ok(());
+            }
+            return Err(invalid("unpaired"));
+        };
+        self.active(key)?;
+        if self.cycles[&key].candidate.is_some() {
+            return Err(invalid("late"));
+        }
+        let request = &self.cycles[&key].requests[&id];
+        if request.complete_sequence.is_some() {
+            return Err(invalid("duplicate"));
+        }
+        let detail = json!({"request_id": id, "context": request.context, "progress": progress,
+            "eof_verified": true, "mac_source_plan_verified": true});
+        let sequence = self.event(key, "remote_scan_complete", detail)?;
+        let request = self
+            .cycles
+            .get_mut(&key)
+            .and_then(|cycle| cycle.requests.get_mut(&id))
+            .ok_or_else(|| invalid("unpaired"))?;
+        request.complete_sequence = Some(sequence);
+        request.progress = Some(progress);
+        Ok(())
+    }
+
+    // Hooks are synchronous; all worker/set tasks join before candidate/root.
+    // A closed manifest is per-cycle evidence, not a continuous live-health attestation.
+    fn close(&mut self, key: (u64, u64), root: Value) -> io::Result<()> {
+        if !self.errors.is_empty() {
+            return Err(invalid("unhealthy"));
+        }
+        let cycle = &self.cycles[&key];
+        let candidate = cycle.candidate.as_ref().ok_or_else(|| invalid("unpaired"))?;
+        for field in ["candidate_sha256", "coverage_sha256"] {
+            if root[field].as_str().is_none() || root[field] != candidate[field] {
+                return Err(invalid("coverage"));
+            }
+        }
+        let mut observed = BTreeSet::new();
+        for context in cycle
+            .requests
+            .values()
+            .map(|request| &request.context)
+            .chain(cycle.local_walks.iter().map(|walk| &walk["context"]))
+        {
+            let pool = context["pool"].as_u64().ok_or_else(|| invalid("malformed"))?;
+            let set = context["set"].as_u64().ok_or_else(|| invalid("malformed"))?;
+            let bucket = context["bucket"].as_str().ok_or_else(|| invalid("malformed"))?;
+            if !observed.insert((pool, set, bucket)) {
+                return Err(invalid("duplicate"));
+            }
+            let coverage = candidate["set_coverage"].as_array().ok_or_else(|| invalid("coverage"))?;
+            if !coverage.iter().any(|row| {
+                row["pool"].as_u64() == Some(pool)
+                    && row["set"].as_u64() == Some(set)
+                    && row["bucket_incarnations"][bucket] == context["bucket_incarnation"]
+                    && !context["bucket_incarnation"].is_null()
+            }) {
+                return Err(invalid("coverage"));
+            }
+        }
+        if cycle.requests.values().any(|request| request.complete_sequence.is_none()) {
+            return Err(invalid("unpaired"));
+        }
+        let sequence = self.event(key, "cycle_close", root.clone())?;
+        self.cycles.get_mut(&key).ok_or_else(|| invalid("outside_cycle"))?.closed = true;
+        let cycle = &self.cycles[&key];
+        let manifest = json!({"schema": 1, "nonce": self.nonce, "pid": self.pid, "epoch": key.0, "cycle": key.1,
+            "start_sequence": cycle.start_sequence, "close_sequence": sequence, "events": cycle.events,
+            "requests": cycle.requests, "local_walks": cycle.local_walks, "candidate": cycle.candidate,
+            "root": root, "errors": self.errors, "outstanding": 0, "closed": true});
+        let name = format!("worker-manifest.{}.{}.{}.{}.json", self.nonce, self.pid, key.0, key.1);
+        let sha256 = self.persist(&name, &manifest)?;
+        self.manifests
+            .push(json!({"file": name, "sha256": sha256, "epoch": key.0, "cycle": key.1,
+            "close_sequence": sequence}));
+        Ok(())
+    }
+    fn apply(&mut self, action: impl FnOnce(&mut Self) -> io::Result<()>) {
+        if let Err(error) = action(self) {
+            let code = match error.to_string().as_str() {
+                "overflow" => "overflow",
+                "late" => "late",
+                "duplicate" => "duplicate",
+                "coverage" => "coverage",
+                "unpaired" => "unpaired",
+                "outside_cycle" => "outside_cycle",
+                "cross_cycle" => "cross_cycle",
+                _ => "io_or_malformed",
+            };
+            self.errors.insert(code);
+        }
+        let name = format!("worker-health.{}.{}.json", self.nonce, self.pid);
+        if self.persist(&name, &self.health()).is_err() {
+            self.errors.insert("health_io");
+        }
+    }
+}
+
+fn control(directory: &Path) -> io::Result<Option<Control>> {
+    if !directory.is_absolute() || directory.components().any(|part| matches!(part, Component::ParentDir)) {
+        return Err(invalid("malformed"));
+    }
+    let control = read_control::<Control>(&directory.join("worker-observer.json"))?;
+    if control
+        .as_ref()
+        .is_some_and(|control| !valid_component(&control.nonce) || control.pid == 0)
+    {
+        return Err(invalid("malformed"));
+    }
+    Ok(control)
+}
+
+/// The scope covers the normal open and consume future; it never consults O again.
+pub(crate) async fn with_incarnation<F: Future>(incarnation: Option<Uuid>, future: F) -> F::Output {
+    INCARNATION.scope(incarnation, future).await
+}
+
+pub(crate) fn enabled() -> bool {
+    // Once begun, a transient control read failure must never hide a real opening or completion.
+    if OBSERVER.get().is_some_and(|state| {
+        state
+            .lock()
+            .unwrap_or_else(|poison| poison.into_inner())
+            .as_ref()
+            .is_some_and(|observer| observer.pid == std::process::id())
+    }) {
+        return true;
+    }
+    let Some(directory) = std::env::var_os("RUSTFS_E2E_SCANNER_BARRIER_DIR") else { return false };
+    control(Path::new(&directory))
+        .ok()
+        .flatten()
+        .is_some_and(|control| control.pid == std::process::id())
+}
+
+fn update(action: impl FnOnce(&mut Observer) -> io::Result<()>) {
+    let Some(directory) = std::env::var_os("RUSTFS_E2E_SCANNER_BARRIER_DIR") else { return };
+    let directory = PathBuf::from(directory);
+    let (mut guard, poisoned) = match OBSERVER.get_or_init(|| Mutex::new(None)).lock() {
+        Ok(guard) => (guard, false),
+        Err(poison) => (poison.into_inner(), true),
+    };
+    let configured = control(&directory);
+    match configured {
+        Ok(Some(control)) if control.pid == std::process::id() => {
+            if let Some(observer) = guard.as_mut() {
+                if observer.nonce != control.nonce {
+                    observer.errors.insert("session_changed");
+                }
+            } else {
+                *guard = Some(Observer::new(directory, control));
+            }
+        }
+        Ok(_) => {
+            let Some(observer) = guard.as_mut() else { return };
+            observer.errors.insert("control_changed");
+        }
+        Err(_) => {
+            if let Some(observer) = guard.as_mut() {
+                observer.errors.insert("control_io");
+            }
+        }
+    }
+    let Some(observer) = guard.as_mut() else { return };
+    if poisoned {
+        observer.errors.insert("poisoned");
+    }
+    observer.apply(action);
+}
+
+pub(crate) fn boundary(stage: &str, detail: &Value) {
+    if !matches!(stage, "before_start_snapshot" | "cycle_candidate" | "root_durable" | "bucket_walk") {
+        return;
+    }
+    if stage == "bucket_walk" && detail["worker_kind"] != "local" {
+        return;
+    }
+    update(|observer| observer.boundary(stage, detail.clone()));
+}
+
+pub(crate) fn opening(context: Value) {
+    let incarnation = INCARNATION.try_with(|incarnation| *incarnation).ok().flatten();
+    update(|observer| observer.opening(context, incarnation));
+}
+
+/// Called only after the ordinary client's authenticated terminal EOF validation.
+pub(crate) fn complete(id: Uuid, progress: [u64; 3]) {
+    update(|observer| observer.complete(id, progress));
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn observer(path: &Path) -> Observer {
+        std::fs::write(path.join("session.json"), b"{}").expect("active observation window");
+        Observer::new(
+            path.to_path_buf(),
+            Control {
+                nonce: "unit".into(),
+                pid: 1,
+            },
+        )
+    }
+    fn context(id: Uuid, cycle: u64) -> Value {
+        json!({"epoch": 1, "cycle": cycle, "request_id": id, "wire_session_id": Uuid::new_v4(),
+            "server_epoch": Uuid::new_v4(), "wire_session_sequence": 0, "pool": 0, "set": 0,
+            "disk_index": 0, "disk_uuid": Uuid::new_v4(), "bucket": "hot", "worker_endpoint": "http://worker:9000",
+            "worker_path": "/disk", "reuse_plan_sha256": "a".repeat(64), "coverage_plan_sha256": "b".repeat(64)})
+    }
+    fn candidate(incarnation: Uuid) -> Value {
+        json!({"epoch":1,"cycle":2,"candidate_sha256":"c".repeat(64),"coverage_sha256":"d".repeat(64),
+            "set_coverage":[{"pool":0,"set":0,"bucket_incarnations":{"hot":incarnation}}]})
+    }
+    fn root() -> Value {
+        json!({"epoch":1,"cycle":2,"candidate_sha256":"c".repeat(64),"coverage_sha256":"d".repeat(64),
+            "root_sha256":"e".repeat(64),"root_version":"root"})
+    }
+    #[test]
+    fn worker_inventory_rejects_unpaired_partial_and_late_completions() {
+        let dir = tempfile::tempdir().expect("temporary evidence directory");
+        let mut observer = observer(dir.path());
+        observer
+            .boundary("before_start_snapshot", json!({"epoch":1,"cycle":2}))
+            .expect("begin");
+        let id = Uuid::new_v4();
+        let incarnation = Uuid::new_v4();
+        observer.opening(context(id, 2), Some(incarnation)).expect("opening");
+        observer
+            .boundary("cycle_candidate", candidate(incarnation))
+            .expect("candidate");
+        assert_eq!(
+            observer
+                .close((1, 2), root())
+                .expect_err("Partial/MAC/EOF failure cannot pair opening")
+                .to_string(),
+            "unpaired"
+        );
+        observer.apply(|state| state.complete(id, [1, 0, 0]));
+        assert!(observer.errors.contains("late"));
+        assert!(observer.close((1, 2), root()).is_err());
+        assert!(observer.complete(Uuid::new_v4(), [1, 0, 0]).is_err());
+    }
+    #[test]
+    fn worker_inventory_preserves_zero_progress_and_rejects_cross_cycle_and_overflow() {
+        let dir = tempfile::tempdir().expect("temporary evidence directory");
+        let mut observer = observer(dir.path());
+        observer
+            .boundary("before_start_snapshot", json!({"epoch":1,"cycle":2}))
+            .expect("begin");
+        let id = Uuid::new_v4();
+        assert!(observer.opening(context(id, 3), Some(Uuid::new_v4())).is_err());
+        observer.opening(context(id, 2), Some(Uuid::new_v4())).expect("opening");
+        observer
+            .complete(id, [0, 0, 0])
+            .expect("verified Current may make no progress");
+        assert_eq!(observer.cycles[&(1, 2)].requests[&id.to_string()].progress, Some([0, 0, 0]));
+        let incarnation = observer.cycles[&(1, 2)].requests[&id.to_string()].context["bucket_incarnation"]
+            .as_str()
+            .expect("captured incarnation")
+            .parse()
+            .expect("UUID");
+        observer
+            .boundary("cycle_candidate", candidate(incarnation))
+            .expect("candidate");
+        observer
+            .close((1, 2), root())
+            .expect("zero-progress Current is recorded, never silently treated as a walk");
+        observer.apply(|state| state.complete(id, [1, 0, 0]));
+        assert!(observer.errors.contains("late"));
+        // Reserve exhaustion on a new active cycle; closed-cycle rejection must not mask overflow.
+        observer
+            .boundary("before_start_snapshot", json!({"epoch":1,"cycle":3}))
+            .expect("second begin");
+        observer.sequence = MAX_EVENTS;
+        observer.apply(|state| state.opening(context(Uuid::new_v4(), 3), Some(Uuid::new_v4())));
+        assert!(observer.errors.contains("overflow"));
+    }
+    #[test]
+    fn worker_inventory_write_failure_and_missing_incarnation_cannot_close() {
+        let dir = tempfile::tempdir().expect("temporary evidence directory");
+        let mut observer = observer(dir.path());
+        observer
+            .boundary("before_start_snapshot", json!({"epoch":1,"cycle":2}))
+            .expect("begin");
+        observer.apply(|state| state.opening(context(Uuid::new_v4(), 2), None));
+        assert!(observer.errors.contains("io_or_malformed"));
+        std::fs::remove_dir_all(dir.path()).expect("remove evidence destination");
+        observer.apply(|state| state.opening(context(Uuid::new_v4(), 2), Some(Uuid::new_v4())));
+        assert!(observer.errors.contains("health_io"));
+        assert_eq!(
+            observer
+                .close((1, 2), root())
+                .expect_err("sticky recorder failure prevents closure")
+                .to_string(),
+            "unhealthy"
+        );
+        assert!(observer.manifests.is_empty());
+    }
+}

crates/scanner/src/lib.rs

@@ -58,6 +58,9 @@ use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
 pub mod data_usage_define;
+#[cfg(feature = "e2e-test-hooks")]
+#[doc(hidden)]
+pub mod e2e_barrier;
 pub mod error;
 mod heal_completion;
 pub mod prefix_usage;

crates/scanner/src/remote_scanner/stream.rs

@@ -793,6 +793,16 @@ async fn scan_and_persist_local_bucket(
     let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel::<DataUsageCache>(1);
     let (node_tx, mut node_rx) = crate::workload_admission::scanner_node_request_channel();
     let scan_ctx = ctx.child_token();
+    #[cfg(feature = "e2e-test-hooks")]
+    let _ = crate::e2e_barrier::checkpoint(
+        "bucket_walk",
+        serde_json::json!({
+            "epoch": leader_epoch, "cycle": next_cycle,
+            "pool": source.pool_index, "set": source.set_index,
+            "bucket": bucket, "drive": disk.path().display().to_string(),
+        }),
+    )
+    .await;
     let scan = crate::workload_admission::with_scanner_node_requests(
         node_tx,
         ScannerIODisk::nsscanner_disk(
@@ -1077,6 +1087,25 @@ pub(crate) async fn scan_remote_bucket(
     let result = async {
         let rpc_deadline = Instant::now() + NS_SCANNER_MAX_RPC_LIFETIME;
         let open_deadline = (Instant::now() + NS_SCANNER_STALL_TIMEOUT).min(rpc_deadline);
+        #[cfg(feature = "e2e-test-hooks")]
+        if crate::e2e_barrier::worker_observation::enabled() {
+            let location = disk.get_disk_location();
+            // RemoteDisk::get_disk_id reads the same cached ID used by disk_ref; no RPC.
+            let disk_uuid = crate::storage_api::owner::EcstoreDiskAPI::get_disk_id(disk)
+                .await
+                .ok()
+                .flatten();
+            crate::e2e_barrier::worker_observation::opening(serde_json::json!({
+                "epoch": leader_epoch, "cycle": next_cycle, "request_id": request_id,
+                "wire_session_id": session_id, "wire_session_sequence": session_sequence,
+                "server_epoch": server_epoch, "bucket": bucket, "pool": expected_source.pool_index,
+                "set": expected_source.set_index, "disk_index": location.disk_idx, "disk_uuid": disk_uuid,
+                "worker_endpoint": crate::storage_api::owner::EcstoreDiskAPI::endpoint(disk).grid_host(),
+                "worker_path": disk.path().display().to_string(),
+                "reuse_plan_sha256": crate::e2e_barrier::hex(&scan_plan_digest.0),
+                "coverage_plan_sha256": crate::e2e_barrier::hex(&checkpoint_plan_digest.0),
+            }));
+        }
         let open_stream = disk.open_ns_scanner_stream(crate::NsScannerOpenRequest {
             request_id,
             server_epoch,
@@ -1428,6 +1457,17 @@ where
                     last_phase,
                 ));
             }
+            #[cfg(feature = "e2e-test-hooks")]
+            if matches!(&result, Ok(RemoteScannerOutcome::Complete { .. })) && crate::e2e_barrier::worker_observation::enabled() {
+                crate::e2e_barrier::worker_observation::complete(
+                    authenticator.request_id,
+                    [
+                        last_progress.objects_scanned,
+                        last_progress.directories_started,
+                        last_progress.entries_visited,
+                    ],
+                );
+            }
             return result;
         }
     }

crates/scanner/src/scanner.rs

@@ -2024,6 +2024,8 @@ where
             resolved_scope_observer: None,
             #[cfg(test)]
             post_walk_gate: None,
+            #[cfg(test)]
+            peer_observations: None,
         },
     )
     .await;
@@ -2164,6 +2166,9 @@ where
         scanner_post_lease_activity_defer_reason(
             result.activity_baseline(),
             probe_scanner_activity(storeapi.as_ref(), true).await,
+            result
+                .remote_dirty_usage_certification()
+                .zip(remote_publication_leases.as_ref().map(|(_, grants)| grants.as_slice())),
         )
     } else {
         None
@@ -2515,9 +2520,24 @@ where
     } else {
         "incomplete_coverage"
     };
+    #[cfg(feature = "e2e-test-hooks")]
+    let durable_observation = scan_cycle_result
+        .publication_expectation()
+        .as_ref()
+        .and_then(|expected| usage_publication_result.e2e_durable_observation(leader_epoch, cycle_info.current, expected));
+    #[cfg(feature = "e2e-test-hooks")]
+    if let Some(observation) = &durable_observation {
+        let _ = crate::e2e_barrier::checkpoint("root_durable", observation.clone()).await;
+    }
     let pending_dirty_usage_buckets = scan_cycle_result.pending_dirty_usage_buckets();
     let (completion_outcome, scanner_pending_maintenance_work, remote_dirty_usage_acknowledgements) =
         finalize_scanner_cycle_result(scan_cycle_result, usage_publication_result);
+    #[cfg(feature = "e2e-test-hooks")]
+    if let Some(mut observation) = durable_observation {
+        observation["acks"] = crate::e2e_barrier::acknowledgement_observations(&remote_dirty_usage_acknowledgements);
+        observation["pending_bucket_count"] = serde_json::json!(pending_dirty_usage_buckets);
+        let _ = crate::e2e_barrier::checkpoint("ack_send", observation).await;
+    }
     let remote_dirty_usage_pending = if remote_dirty_usage_acknowledgements.is_empty() {
         false
     } else if let Some(notification_system) = storeapi.scanner_notification_system() {
@@ -3949,18 +3969,26 @@ impl ScannerCyclePublicationPlan {
 /// Re-prove the guardrail after every peer granted a publication lease. The
 /// comparison uses the guardrail fields (`scanner_activity_guardrail_veto`),
 /// not the full digest: local generation movement in this window is the next
-/// cycle's work and its buckets stay in the dirty map. A peer whose dirty
-/// usage moved keeps deferring in this phase, as it does during the walk.
+/// cycle's work and its buckets stay in the dirty map. A sealed producer
+/// proof may cover peer dirty movement too, but must bind every lease owner
+/// and session without changing its candidate or acknowledged tuples.
 fn scanner_post_lease_activity_defer_reason(
     baseline: Option<&ScannerActivitySnapshot>,
     activity: Result<ScannerActivitySnapshot, String>,
+    certification: Option<(
+        &crate::scanner_io::ScannerRemoteDirtyUsageCertification,
+        &[crate::storage_api::EcstoreScannerPublicationLeaseGrant],
+    )>,
 ) -> Option<ScannerCycleDeferReason> {
     match activity {
         Ok(snapshot)
             if scanner_activity_allows_usage_publication(&snapshot)
                 && baseline.is_some_and(|baseline| {
                     scanner_activity_guardrail_veto(baseline, &snapshot).is_none()
-                        && scanner_peer_dirty_usage_moved(baseline, &snapshot).is_none()
+                        && certification.map_or_else(
+                            || scanner_peer_dirty_usage_moved(baseline, &snapshot).is_none(),
+                            |(proof, grants)| proof.matches_publication_leases(grants) && proof.matches_activity(&snapshot),
+                        )
                 }) =>
         {
             None
@@ -3970,6 +3998,13 @@ fn scanner_post_lease_activity_defer_reason(
                 Err(_) => "probe_failed",
                 Ok(snapshot) if !scanner_activity_allows_usage_publication(snapshot) => "publication_blocked",
                 Ok(_) if baseline.is_none() => "baseline_missing",
+                Ok(snapshot)
+                    if certification.is_some_and(|(proof, grants)| {
+                        !proof.matches_publication_leases(grants) || !proof.matches_activity(snapshot)
+                    }) =>
+                {
+                    "peer_certification"
+                }
                 Ok(snapshot) => baseline
                     .and_then(|baseline| {
                         scanner_activity_guardrail_veto(baseline, snapshot)
@@ -4133,6 +4168,7 @@ fn finalize_scanner_cycle_result(
                 .is_some_and(|proof| proof.verified_version_for(expected).is_some())
         });
     let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work()
+        || scan_cycle_result.has_remote_dirty_usage_pending()
         || (scan_cycle_result.has_dirty_usage_to_acknowledge() && !durable_complete_snapshot);
     let remote_dirty_usage_acknowledgements = if durable_complete_snapshot {
         match proof {

crates/scanner/src/scanner/activity.rs

@@ -546,6 +546,24 @@ pub(crate) struct ScannerNodeActivity {
     pub(super) publication_blocked: bool,
 }
 
+#[cfg(feature = "e2e-test-hooks")]
+impl ScannerNodeActivity {
+    pub(crate) fn e2e_observation(&self) -> serde_json::Value {
+        serde_json::json!({
+            "instance_id": self.instance_id,
+            "namespace_generation": self.namespace_generation,
+            "maintenance_generation": self.maintenance_generation,
+            "protocol_version": self.protocol_version,
+            "topology_digest": crate::e2e_barrier::hex(&self.topology_digest),
+            "data_movement_active": self.data_movement_active,
+            "dirty_usage_generation": self.dirty_usage_generation,
+            "dirty_usage_pending": self.dirty_usage_pending,
+            "movement_generation": self.movement_generation,
+            "publication_blocked": self.publication_blocked,
+        })
+    }
+}
+
 pub(crate) type ScannerActivitySnapshot = BTreeMap<String, ScannerNodeActivity>;
 
 #[cfg(test)]

crates/scanner/src/scanner/tests.rs

@@ -10362,7 +10362,7 @@ fn post_lease_activity_proof_rejects_a_peer_whose_dirty_usage_moved_before_lease
         (LOCAL_SCANNER_ACTIVITY_NODE.to_string(), scanner_node_activity("local", 5, 2)),
         ("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3)),
     ]);
-    assert_eq!(scanner_post_lease_activity_defer_reason(Some(&before), Ok(before.clone())), None);
+    assert_eq!(scanner_post_lease_activity_defer_reason(Some(&before), Ok(before.clone()), None), None);
 
     // Local generation movement in the lease window is the next cycle's
     // work: its buckets stay in the dirty map and the walk remains certified.
@@ -10373,7 +10373,7 @@ fn post_lease_activity_proof_rejects_a_peer_whose_dirty_usage_moved_before_lease
     local.namespace_generation += 1;
     local.dirty_usage_generation += 1;
     local.dirty_usage_pending = true;
-    assert_eq!(scanner_post_lease_activity_defer_reason(Some(&before), Ok(local_shaded)), None);
+    assert_eq!(scanner_post_lease_activity_defer_reason(Some(&before), Ok(local_shaded), None), None);
 
     // A peer PUT tail that drained before the lease leaves the movement
     // generation accepted by the lease unchanged, but the peer's dirty state
@@ -10387,14 +10387,14 @@ fn post_lease_activity_proof_rejects_a_peer_whose_dirty_usage_moved_before_lease
         "the existing movement-only lease remains valid after a PUT tail drains"
     );
     assert!(scanner_activity_allows_usage_publication(&after));
-    let reason = scanner_post_lease_activity_defer_reason(Some(&before), Ok(after));
+    let reason = scanner_post_lease_activity_defer_reason(Some(&before), Ok(after), None);
     assert_eq!(reason, Some(ScannerCycleDeferReason::ActivityBaselineUnavailable));
 
     // A guardrail field moving in the window still defers.
     let mut moved = before.clone();
     moved.get_mut("node-2").expect("peer should be present").movement_generation += 1;
     assert_eq!(
-        scanner_post_lease_activity_defer_reason(Some(&before), Ok(moved)),
+        scanner_post_lease_activity_defer_reason(Some(&before), Ok(moved), None),
         Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
     );
 
@@ -10489,7 +10489,7 @@ fn post_lease_activity_proof_requires_a_complete_matching_baseline() {
         (Some(&blocked), Ok(blocked.clone())),
     ] {
         assert_eq!(
-            scanner_post_lease_activity_defer_reason(expected, observed),
+            scanner_post_lease_activity_defer_reason(expected, observed, None),
             Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
         );
     }

crates/scanner/src/scanner/tests/scoped_ack_publication.rs

@@ -49,6 +49,7 @@ async fn complete_candidate(store: &Arc<ECStore>, cycle: u64) -> (crate::scanner
             service_cohort: None,
             resolved_scope_observer: None,
             post_walk_gate: None,
+            peer_observations: None,
         },
     )
     .await
@@ -285,6 +286,7 @@ async fn scoped_ack_publication_observed_candidate_reuse_requires_a_new_root_pro
             service_cohort: None,
             resolved_scope_observer: Some(observer),
             post_walk_gate: None,
+            peer_observations: None,
         },
     )
     .await
@@ -456,6 +458,7 @@ async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish
         "remote_scoped_ack_target",
         "publication_epoch",
         "remote_lease_targets",
+        "remote_pending",
     ] {
         let (_directory, store) = candidate_store().await;
         let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await;
@@ -502,6 +505,7 @@ async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish
                 },
             }]),
             "publication_epoch" => scan.with_publication_epoch(Some(changed_epoch)),
+            "remote_pending" => scan.with_remote_dirty_usage_pending(true),
             "remote_lease_targets" => scan.with_remote_publication_lease_targets(vec![(
                 "proof-peer:9000".to_string(),
                 crate::scanner_activity_epoch().to_string(),
@@ -524,3 +528,142 @@ async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish
     }
     crate::scanner_io::clear_dirty_usage_buckets_for_tests();
 }
+
+#[test]
+#[serial]
+fn certified_empty_remote_ack_keeps_known_remote_work_pending() {
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+    let scan = ScannerCycleResult::new(ScannerCycleStatus::Complete, None).with_remote_dirty_usage_pending(true);
+    let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(scan, DataUsagePersistOutcome::Current.into());
+    assert_eq!(outcome, ScannerCycleOutcome::Completed);
+    assert!(acknowledgements.is_empty());
+    assert!(pending, "an empty ACK cannot hide the remote buckets excluded from certification");
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+}
+
+#[test]
+fn sealed_peer_proof_binds_every_lease_and_keeps_all_post_lease_guardrails() {
+    use crate::storage_api::scan::{
+        EcstoreScannerPeerDirtyUsageBucket, EcstoreScannerPeerDirtyUsageSnapshot, EcstoreScannerPublicationLease,
+        EcstoreScannerPublicationLeaseGrant,
+    };
+    let peer_rows = [("peer-a", 101_u128, 201_u128, true), ("peer-b", 102, 202, false)];
+    let mut before = BTreeMap::new();
+    let mut peers = Vec::new();
+    let mut grants = Vec::new();
+    for (host, owner, instance, pending) in peer_rows {
+        let owner_id = uuid::Uuid::from_u128(owner).simple().to_string();
+        let instance_id = uuid::Uuid::from_u128(instance).simple().to_string();
+        before.insert(host.to_string(), scanner_node_activity_for_tests(&instance_id, 10, 10, pending));
+        let buckets = if pending {
+            BTreeMap::from([(
+                "cold-bucket".to_string(),
+                EcstoreScannerPeerDirtyUsageBucket {
+                    bucket_incarnation: uuid::Uuid::from_u128(301),
+                    generation: 8,
+                },
+            )])
+        } else {
+            BTreeMap::new()
+        };
+        peers.push((
+            host.to_string(),
+            EcstoreScannerPeerDirtyUsageSnapshot {
+                owner_id: owner_id.clone(),
+                instance_id: instance_id.clone(),
+                generation: 10,
+                pending_bucket_count: buckets.len() as u64,
+                protocol_version: crate::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION,
+                complete: true,
+                buckets,
+            },
+        ));
+        grants.push(EcstoreScannerPublicationLeaseGrant {
+            host: host.to_string(),
+            lease: EcstoreScannerPublicationLease {
+                token: uuid::Uuid::new_v4(),
+                movement_generation: 0,
+                observational_only: false,
+                owner_id,
+                session_id: instance_id,
+                expires_at: std::time::Instant::now() + Duration::from_secs(60),
+            },
+        });
+    }
+    let proof = crate::scanner_io::ScannerRemoteDirtyUsageCertification::for_tests(before.clone(), peers)
+        .expect("complete producer evidence");
+    let mut late_write = before.clone();
+    let writer = late_write.get_mut("peer-a").expect("writer");
+    writer.namespace_generation += 1;
+    writer.dirty_usage_generation += 1;
+    assert_eq!(
+        scanner_post_lease_activity_defer_reason(Some(&before), Ok(late_write.clone()), Some((&proof, &grants))),
+        None
+    );
+
+    for case in [
+        "missing",
+        "duplicate",
+        "owner",
+        "instance",
+        "movement",
+        "expired",
+        "observational",
+        "token",
+    ] {
+        let mut invalid = grants.clone();
+        match case {
+            "missing" => {
+                invalid.pop();
+            }
+            "duplicate" => invalid[1] = invalid[0].clone(),
+            "owner" => invalid[1].lease.owner_id = uuid::Uuid::from_u128(999).to_string(),
+            "instance" => invalid[1].lease.session_id = uuid::Uuid::from_u128(999).to_string(),
+            "movement" => invalid[1].lease.movement_generation += 1,
+            "expired" => invalid[1].lease.expires_at = std::time::Instant::now() - Duration::from_secs(1),
+            "observational" => invalid[1].lease.observational_only = true,
+            "token" => invalid[1].lease.token = uuid::Uuid::nil(),
+            _ => unreachable!(),
+        }
+        assert_eq!(
+            scanner_post_lease_activity_defer_reason(Some(&before), Ok(before.clone()), Some((&proof, &invalid))),
+            Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
+            "{case}"
+        );
+    }
+    for field in [
+        "instance",
+        "protocol",
+        "topology",
+        "maintenance",
+        "data_movement",
+        "movement",
+        "blocked",
+        "missing_peer",
+        "overflow",
+        "rollback",
+    ] {
+        let mut changed = late_write.clone();
+        let peer = changed.get_mut("peer-a").expect("peer");
+        match field {
+            "instance" => peer.instance_id = uuid::Uuid::from_u128(999).to_string(),
+            "protocol" => peer.protocol_version += 1,
+            "topology" => peer.topology_digest[0] ^= 1,
+            "maintenance" => peer.maintenance_generation += 1,
+            "data_movement" => peer.data_movement_active = true,
+            "movement" => peer.movement_generation += 1,
+            "blocked" => peer.publication_blocked = true,
+            "missing_peer" => {
+                changed.remove("peer-b");
+            }
+            "overflow" => peer.dirty_usage_generation = u64::MAX,
+            "rollback" => peer.dirty_usage_generation = 9,
+            _ => unreachable!(),
+        }
+        assert_eq!(
+            scanner_post_lease_activity_defer_reason(Some(&before), Ok(changed), Some((&proof, &grants))),
+            Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
+            "{field}"
+        );
+    }
+}

crates/scanner/src/scanner/usage_store.rs

@@ -84,6 +84,27 @@ impl DataUsagePublicationResult {
         (self.outcome, self.proof)
     }
 
+    #[cfg(feature = "e2e-test-hooks")]
+    pub(super) fn e2e_durable_observation(
+        &self,
+        epoch: u64,
+        cycle: u64,
+        expected: &crate::scanner_io::ScannerPublicationExpectation,
+    ) -> Option<serde_json::Value> {
+        let outcome = match self.outcome {
+            DataUsagePersistOutcome::Saved => "saved",
+            DataUsagePersistOutcome::AlreadyDurable => "already_durable",
+            _ => return None,
+        };
+        let (root_version, root_digest) = self.proof.as_ref()?.verified_version_for(expected)?;
+        let (candidate_digest, coverage_digest) = expected.e2e_digests();
+        Some(serde_json::json!({
+            "epoch": epoch, "cycle": cycle, "outcome": outcome,
+            "candidate_sha256": candidate_digest, "coverage_sha256": coverage_digest,
+            "root_version": root_version, "root_sha256": crate::e2e_barrier::hex(root_digest),
+        }))
+    }
+
     pub(super) fn observation_for_attempt(
         &self,
         scanner_epoch: u64,
@@ -1067,6 +1088,23 @@ where
                 break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded);
             }
 
+            #[cfg(feature = "e2e-test-hooks")]
+            if !observational
+                && target_path == DATA_USAGE_OBJ_NAME_PATH.as_str()
+                && let (Some(expected), Some(epoch), Some(cycle)) =
+                    (ack_expectation.as_ref(), data_usage_info.scanner_epoch, data_usage_info.scanner_cycle)
+                && expected.matches_encoded_candidate(&data_digest)
+            {
+                let (candidate_digest, coverage_digest) = expected.e2e_digests();
+                let _ = crate::e2e_barrier::checkpoint(
+                    "root_before_publish",
+                    serde_json::json!({
+                        "epoch": epoch, "cycle": cycle,
+                        "candidate_sha256": candidate_digest, "coverage_sha256": coverage_digest,
+                    }),
+                )
+                .await;
+            }
             let done_save = Metrics::time(Metric::SaveUsage);
             let (save_result, commit_state) = {
                 let publication_scope = storeapi

crates/scanner/src/scanner_io.rs

@@ -938,6 +938,7 @@ pub struct ScannerBucketScanPlan {
     publication_epoch: Option<u64>,
     dirty_usage_buckets: Arc<DirtyUsageBuckets>,
     bucket_failures: ScannerBucketFailureState,
+    bucket_completions: Option<Arc<Mutex<ScannerBucketCompletions>>>,
     pending_maintenance_work: Arc<AtomicBool>,
     cache_cycle_floor: Arc<AtomicU64>,
     cold_zero_walk_reuse_observed: Arc<AtomicBool>,
@@ -1272,20 +1273,14 @@ enum ScannerCycleActivityStatus {
     Unverified,
 }
 
-async fn scanner_cycle_activity_status<S>(
-    store: &S,
-    distributed: bool,
+fn scanner_cycle_activity_status(
     before: &crate::scanner::ScannerActivitySnapshot,
-) -> (ScannerCycleActivityStatus, Vec<(String, String, u64)>)
-where
-    S: ScannerStorage,
-{
-    // Read the storage movement barrier before sampling the activity
-    // generations. A movement that starts during this await must invalidate
-    // the earlier baseline. Ordinary PUT tails are not part of the barrier:
-    // they shade their bucket and are certified per bucket below.
-    let publication_blocked = store.scanner_data_usage_publication_blocked().await;
-    match crate::scanner::probe_scanner_activity(store, distributed).await {
+    observed: std::result::Result<crate::scanner::ScannerActivitySnapshot, String>,
+    publication_blocked: bool,
+    certification: Option<&ScannerRemoteDirtyUsageCertification>,
+    certification_required: bool,
+) -> (ScannerCycleActivityStatus, Vec<(String, String, u64)>) {
+    match observed {
         Ok(after) => {
             let veto = if publication_blocked {
                 Some(crate::scanner::ScannerActivityGuardrailVeto::new(
@@ -1309,10 +1304,15 @@ where
                 );
                 global_metrics().record_scanner_guardrail_veto(veto.reason());
                 ScannerCycleActivityStatus::Changed
-            } else if let Some(moved) = crate::scanner::scanner_peer_dirty_usage_moved(before, &after) {
-                // Per-bucket certification covers the local dirty map only;
-                // a peer whose dirty state moved keeps today's whole-cycle
-                // outcome until scoped acknowledgements can name its buckets.
+            } else if certification_required && certification.is_none()
+                || certification.is_some_and(|proof| !proof.matches_activity(&after))
+            {
+                ScannerCycleActivityStatus::Unverified
+            } else if let Some(moved) =
+                crate::scanner::scanner_peer_dirty_usage_moved(before, &after).filter(|_| certification.is_none())
+            {
+                // Without a complete producer proof, retain the original
+                // whole-cycle fallback for a moving peer.
                 tracing::info!(
                     target: "rustfs::scanner::io",
                     event = EVENT_SCANNER_SET_STATE,
@@ -1743,6 +1743,8 @@ pub(crate) struct ScannerCycleResult {
     partial_bucket_progress: bool,
     dirty_usage_clear: Option<DirtyUsageBuckets>,
     remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
+    remote_dirty_usage_certification: Option<ScannerRemoteDirtyUsageCertification>,
+    remote_dirty_usage_pending: bool,
     distributed_segment_invalidation_evidence: Option<DistributedSegmentInvalidationEvidence>,
     segment_reuse_activation_preflight: ScannerSegmentReuseActivationPreflight,
     remote_publication_lease_targets: Vec<(String, String, u64)>,
@@ -1765,6 +1767,8 @@ impl ScannerCycleResult {
             partial_bucket_progress: false,
             dirty_usage_clear,
             remote_dirty_usage_acknowledgements: Vec::new(),
+            remote_dirty_usage_certification: None,
+            remote_dirty_usage_pending: false,
             distributed_segment_invalidation_evidence: None,
             segment_reuse_activation_preflight: scanner_segment_reuse_activation_preflight(),
             remote_publication_lease_targets: Vec::new(),
@@ -1888,6 +1892,26 @@ impl ScannerCycleResult {
         self
     }
 
+    fn with_remote_dirty_usage_certification(mut self, certification: Option<ScannerRemoteDirtyUsageCertification>) -> Self {
+        self.publication_expectation = None;
+        self.remote_dirty_usage_certification = certification;
+        self
+    }
+
+    pub(crate) fn remote_dirty_usage_certification(&self) -> Option<&ScannerRemoteDirtyUsageCertification> {
+        self.remote_dirty_usage_certification.as_ref()
+    }
+
+    pub(crate) fn with_remote_dirty_usage_pending(mut self, pending: bool) -> Self {
+        self.publication_expectation = None;
+        self.remote_dirty_usage_pending = pending;
+        self
+    }
+
+    pub(crate) fn has_remote_dirty_usage_pending(&self) -> bool {
+        self.remote_dirty_usage_pending
+    }
+
     fn with_distributed_segment_invalidation_evidence(
         mut self,
         evidence: Option<DistributedSegmentInvalidationEvidence>,
@@ -1969,9 +1993,12 @@ mod guards;
 pub(crate) use guards::ScannerServiceCohort;
 mod io_cache;
 mod io_cycle;
+mod peer_certification;
 #[cfg(test)]
 use io_cache::{ScannerSetCacheGeneration, prepare_scoped_set_scan};
 pub(crate) use io_cycle::{ScannerCycleRequest, nsscanner_with_storage_status_scoped};
+pub(crate) use peer_certification::ScannerRemoteDirtyUsageCertification;
+use peer_certification::{ScannerBucketCompletions, ScannerRemoteDirtyUsageWindow};
 mod io_disk;
 #[cfg(test)]
 mod publish_gate_tests;

crates/scanner/src/scanner_io/cache.rs

@@ -641,6 +641,11 @@ impl ScannerPublicationExpectation {
         Arc::ptr_eq(&self.candidate, &other.candidate) && self.candidate.1 == other.candidate.1
     }
 
+    #[cfg(feature = "e2e-test-hooks")]
+    pub(crate) fn e2e_digests(&self) -> (String, String) {
+        (crate::e2e_barrier::hex(&self.candidate.0), crate::e2e_barrier::hex(&self.candidate.1.0))
+    }
+
     #[cfg(test)]
     pub(crate) fn for_tests(candidate_digest: [u8; 32], coverage_digest: DataUsageScanPlanDigest) -> Self {
         Self {

crates/scanner/src/scanner_io/io_cache.rs

@@ -308,6 +308,7 @@ impl ScannerIOCache for SetDisks {
             publication_epoch,
             dirty_usage_buckets,
             bucket_failures,
+            bucket_completions,
             pending_maintenance_work,
             cache_cycle_floor,
             cold_zero_walk_reuse_observed,
@@ -809,6 +810,8 @@ impl ScannerIOCache for SetDisks {
             let scope_clone = scope.clone();
             let cache_cycle_floor_clone = cache_cycle_floor.clone();
             let expected_publication_epoch_clone = expected_publication_epoch;
+            #[cfg(feature = "e2e-test-hooks")]
+            let worker_incarnations = current_bucket_incarnations.clone();
             let remote_server_epoch = match worker_mode {
                 NamespaceScannerWorkerMode::RemoteV4(server_epoch) => Some(server_epoch),
                 NamespaceScannerWorkerMode::Coordinator => None,
@@ -909,7 +912,7 @@ impl ScannerIOCache for SetDisks {
                             break;
                         };
                         remote_session_sequence = next_sequence;
-                        let remote_outcome = crate::remote_scanner::scan_remote_bucket(
+                        let remote_scan = crate::remote_scanner::scan_remote_bucket(
                             &disk,
                             ctx_clone.clone(),
                             budget_clone.clone(),
@@ -925,8 +928,13 @@ impl ScannerIOCache for SetDisks {
                                 skip_healing: healing,
                                 scan_mode,
                             },
-                        )
-                        .await;
+                        );
+                        #[cfg(feature = "e2e-test-hooks")]
+                        let remote_scan = crate::e2e_barrier::worker_observation::with_incarnation(
+                            worker_incarnations.as_ref().and_then(|values| values.get(&bucket.name)).copied(),
+                            remote_scan,
+                        );
+                        let remote_outcome = remote_scan.await;
                         match remote_outcome {
                             Ok(crate::remote_scanner::RemoteScannerOutcome::Complete {
                                 usage,
@@ -1323,6 +1331,23 @@ impl ScannerIOCache for SetDisks {
                     let (node_tx, mut node_rx) = crate::workload_admission::scanner_node_request_channel();
 
                     let scan_ctx = ctx_clone.child_token();
+                    #[cfg(feature = "e2e-test-hooks")]
+                    {
+                        let disk_uuid = if crate::e2e_barrier::worker_observation::enabled() {
+                            crate::storage_api::owner::EcstoreDiskAPI::get_disk_id(disk.as_ref()).await.ok().flatten()
+                        } else { None };
+                        let location = disk.get_disk_location();
+                        let _ = crate::e2e_barrier::checkpoint("bucket_walk", serde_json::json!({
+                            "epoch": leader_epoch, "cycle": want_cycle,
+                            "pool": source.pool_index, "set": source.set_index,
+                            "bucket": bucket.name, "drive": disk.path().display().to_string(),
+                            "worker_kind": "local", "worker_path": disk.path().display().to_string(),
+                            "worker_endpoint": crate::storage_api::owner::EcstoreDiskAPI::endpoint(disk.as_ref()).grid_host(),
+                            "disk_uuid": disk_uuid,
+                            "disk_index": location.disk_idx,
+                            "bucket_incarnation": worker_incarnations.as_ref().and_then(|values| values.get(&bucket.name)),
+                        })).await;
+                    }
                     let scan = crate::workload_admission::with_scanner_node_requests(node_tx, disk.clone().nsscanner_disk(
                         scan_ctx.clone(),
                         budget_clone.clone(),
@@ -1752,6 +1777,22 @@ impl ScannerIOCache for SetDisks {
             return Err(err);
         }
 
+        // Record only actual worker completions, before adding carried values.
+        // The root's cold entries cannot certify a newly observed peer tuple.
+        if let Some(completions) = bucket_completions {
+            let completed_incarnations = completed_buckets
+                .iter()
+                .filter_map(|bucket| {
+                    current_bucket_incarnations
+                        .as_ref()?
+                        .get(bucket)
+                        .filter(|incarnation| !incarnation.is_nil())
+                        .map(|incarnation| (bucket.clone(), *incarnation))
+                })
+                .collect();
+            completions.lock().await.insert(source, completed_incarnations);
+        }
+
         let carried_buckets = carry_partial_bucket_values(
             &mut cache,
             &old_cache,

crates/scanner/src/scanner_io/io_cycle.rs

@@ -79,6 +79,8 @@ where
         resolved_scope_observer: None,
         #[cfg(test)]
         post_walk_gate: None,
+        #[cfg(test)]
+        peer_observations: None,
     };
     nsscanner_with_storage_status_scoped(store, request).await
 }
@@ -103,10 +105,133 @@ pub(crate) struct ScannerCycleRequest {
     /// certification regardless of how fast the walk finishes.
     #[cfg(test)]
     pub(crate) post_walk_gate: Option<tokio::sync::oneshot::Receiver<()>>,
+    #[cfg(test)]
+    pub(crate) peer_observations: Option<ScannerCyclePeerObservations>,
+}
+
+#[cfg(test)]
+#[derive(Clone)]
+pub(crate) struct ScannerCyclePeerObservations {
+    pub(crate) activity: Arc<dyn Fn() -> crate::scanner::ScannerActivitySnapshot + Send + Sync>,
+    pub(crate) dirty_usage: Arc<dyn Fn() -> Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)> + Send + Sync>,
+    pub(crate) scoped_ack_capable: bool,
+}
+
+pub(super) async fn probe_scanner_cycle_activity<S>(
+    store: &S,
+    distributed: bool,
+    #[cfg(test)] peer_observations: Option<&ScannerCyclePeerObservations>,
+) -> std::result::Result<crate::scanner::ScannerActivitySnapshot, String>
+where
+    S: ScannerStorage,
+{
+    #[cfg(test)]
+    if let Some(peers) = peer_observations {
+        let mut snapshot = crate::scanner::probe_scanner_activity(store, false).await?;
+        for (host, activity) in (peers.activity)() {
+            if snapshot.insert(host, activity).is_some() {
+                return Err("duplicate test peer activity".to_string());
+            }
+        }
+        return Ok(snapshot);
+    }
+    crate::scanner::probe_scanner_activity(store, distributed).await
+}
+
+async fn observe_remote_dirty_usage_window<S>(
+    store: &S,
+    before: &crate::scanner::ScannerActivitySnapshot,
+    _stage: &str,
+    _epoch: u64,
+    _cycle: u64,
+    #[cfg(test)] peer_observations: Option<&ScannerCyclePeerObservations>,
+) -> (
+    Option<ScannerRemoteDirtyUsageWindow>,
+    std::result::Result<crate::scanner::ScannerActivitySnapshot, String>,
+)
+where
+    S: ScannerStorage,
+{
+    #[cfg(feature = "e2e-test-hooks")]
+    let _ =
+        crate::e2e_barrier::checkpoint(&format!("before_{_stage}"), serde_json::json!({"epoch": _epoch, "cycle": _cycle})).await;
+    #[cfg(test)]
+    let snapshots = if let Some(peers) = peer_observations {
+        Ok((peers.dirty_usage)())
+    } else if let Some(notification) = store.scanner_notification_system() {
+        notification
+            .scanner_dirty_usage_snapshots()
+            .await
+            .map_err(|error| error.to_string())
+    } else {
+        Err("scanner peer snapshot service is unavailable".to_string())
+    };
+    #[cfg(not(test))]
+    let snapshots = if let Some(notification) = store.scanner_notification_system() {
+        notification
+            .scanner_dirty_usage_snapshots()
+            .await
+            .map_err(|error| error.to_string())
+    } else {
+        Err("scanner peer snapshot service is unavailable".to_string())
+    };
+    let after = probe_scanner_cycle_activity(
+        store,
+        true,
+        #[cfg(test)]
+        peer_observations,
+    )
+    .await;
+    let window = match (snapshots, &after) {
+        (Ok(snapshots), Ok(after)) => ScannerRemoteDirtyUsageWindow::verify(before.clone(), snapshots, after.clone()),
+        _ => None,
+    };
+    #[cfg(feature = "e2e-test-hooks")]
+    let _ = crate::e2e_barrier::checkpoint(
+        _stage,
+        window.as_ref().map_or_else(
+            || serde_json::json!({"epoch": _epoch, "cycle": _cycle, "verified": false}),
+            |window| window.observation(_epoch, _cycle),
+        ),
+    )
+    .await;
+    (window, after)
+}
+
+async fn remote_dirty_usage_window_supports_scoped_ack<S>(
+    store: &S,
+    window: &ScannerRemoteDirtyUsageWindow,
+    #[cfg(test)] peer_observations: Option<&ScannerCyclePeerObservations>,
+) -> bool
+where
+    S: ScannerStorage,
+{
+    let acknowledgements = window.acknowledgements();
+    if scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&acknowledgements) {
+        return false;
+    }
+    // Empty peers remain in the proof; they need no empty capability request.
+    if acknowledgements.is_empty() {
+        return true;
+    }
+    #[cfg(test)]
+    if let Some(peers) = peer_observations {
+        return peers.scoped_ack_capable;
+    }
+    let Some(notification) = store.scanner_notification_system() else {
+        return false;
+    };
+    matches!(
+        notification
+            .scanner_scoped_dirty_usage_capabilities(acknowledgements.into_iter().map(Into::into).collect(),)
+            .await,
+        Ok(true)
+    )
 }
 
 pub(super) struct ScannerBucketScopeResolution<'a> {
     pub(super) requested_scope: ScannerBucketScanScope,
+    pub(super) remote_window: Option<&'a ScannerRemoteDirtyUsageWindow>,
     pub(super) baseline_proof: ScannerCacheBaselineProof<'a>,
     pub(super) activity_before: &'a crate::scanner::ScannerActivitySnapshot,
     pub(super) dirty_usage_snapshot: &'a DirtyUsageSnapshot,
@@ -149,6 +274,29 @@ where
         .cloned()
         .collect::<HashSet<_>>();
     if distributed {
+        if let Some(window) = resolution.remote_window {
+            if let Some(strict_remote_usage) = window.strict_remote_usage() {
+                // Only the original quiet window can supply segment reuse evidence.
+                return resolve_remote_dirty_usage_scope(
+                    resolution.requested_scope,
+                    resolution.dirty_usage_snapshot,
+                    strict_remote_usage,
+                    resolution.all_buckets,
+                    resolution.baseline_proof,
+                );
+            }
+            let mut dirty_buckets = dirty_buckets;
+            dirty_buckets.extend(window.dirty_buckets());
+            return default_result(scoped_scan_scope_from_dirty_buckets(
+                resolution.requested_scope,
+                dirty_buckets,
+                None,
+                true,
+                false,
+                resolution.all_buckets,
+                resolution.baseline_proof,
+            ));
+        }
         let notification_system = store.scanner_notification_system();
         #[cfg(test)]
         let peer_snapshots = if let Some(peer_snapshots) = resolution.test_peer_snapshots.clone() {
@@ -322,7 +470,11 @@ where
         resolved_scope_observer,
         #[cfg(test)]
         post_walk_gate,
+        #[cfg(test)]
+        peer_observations,
     } = request;
+    #[cfg(feature = "e2e-test-hooks")]
+    let requested_scope = e2e_scope_observation(&scan_scope);
     let child_token = ctx.child_token();
     let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);
 
@@ -359,7 +511,17 @@ where
     };
 
     let distributed = store.setup_is_dist_erasure().await;
-    let activity_before = match scanner_activity_preflight(crate::scanner::probe_scanner_activity(store, distributed).await) {
+    #[cfg(test)]
+    let distributed = distributed || peer_observations.is_some();
+    let activity_before = match scanner_activity_preflight(
+        probe_scanner_cycle_activity(
+            store,
+            distributed,
+            #[cfg(test)]
+            peer_observations.as_ref(),
+        )
+        .await,
+    ) {
         ScannerActivityPreflight::Ready(snapshot) => Arc::new(snapshot),
         ScannerActivityPreflight::ActivityBaselineUnavailable(err) => {
             warn!(
@@ -420,6 +582,39 @@ where
             .unwrap_or_else(|poisoned| poisoned.into_inner())
             .refresh(&buckets_by_source);
     }
+    let full_scan_required = requires_full_scan || scan_mode == HealScanMode::Deep;
+    let scope_eligible = !full_scan_required && scan_scope.is_default();
+    let mut remote_start = if distributed && scope_eligible && !all_buckets.is_empty() {
+        observe_remote_dirty_usage_window(
+            store,
+            &activity_before,
+            "start_snapshot",
+            leader_epoch,
+            want_cycle,
+            #[cfg(test)]
+            peer_observations.as_ref(),
+        )
+        .await
+        .0
+    } else {
+        None
+    };
+    if let Some(window) = remote_start.as_ref()
+        && !remote_dirty_usage_window_supports_scoped_ack(
+            store,
+            window,
+            #[cfg(test)]
+            peer_observations.as_ref(),
+        )
+        .await
+    {
+        remote_start = None;
+    }
+    // Work and Current-cache identities include writes admitted by the start
+    // window. Its earlier edge remains in the producer proof, not the cache key.
+    let activity_before = remote_start
+        .as_ref()
+        .map_or(activity_before, ScannerRemoteDirtyUsageWindow::activity_for_walk);
     let structural_scan_plan_digest =
         scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before));
     let scan_plan_digest = scanner_bucket_work_digest(structural_scan_plan_digest, scan_mode, requires_full_scan);
@@ -437,22 +632,21 @@ where
         want_cycle,
         scan_plan_digest: structural_scan_plan_digest,
     };
-    let full_scan_required = requires_full_scan || scan_mode == HealScanMode::Deep;
-    let scope_eligible = !full_scan_required && scan_scope.is_default();
     let scope_resolution = resolve_scanner_bucket_scan_scope(
         store,
         distributed,
         ScannerBucketScopeResolution {
             requested_scope: scan_scope,
+            remote_window: remote_start.as_ref(),
             baseline_proof,
             activity_before: &activity_before,
             dirty_usage_snapshot: &dirty_usage_snapshot,
             all_buckets: &all_buckets,
             requires_full_scan: full_scan_required,
             #[cfg(test)]
-            test_peer_snapshots: None,
+            test_peer_snapshots: peer_observations.as_ref().map(|peers| (peers.dirty_usage)()),
             #[cfg(test)]
-            test_scoped_dirty_usage_capability: None,
+            test_scoped_dirty_usage_capability: peer_observations.as_ref().map(|peers| peers.scoped_ack_capable),
         },
     )
     .await;
@@ -496,8 +690,16 @@ where
         if !bucket_plan_complete {
             return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch));
         }
+        let publication_blocked = store.scanner_data_usage_publication_blocked().await;
+        let activity_after = probe_scanner_cycle_activity(
+            store,
+            distributed,
+            #[cfg(test)]
+            peer_observations.as_ref(),
+        )
+        .await;
         let (activity_status, remote_publication_lease_targets) =
-            scanner_cycle_activity_status(store, distributed, &activity_before).await;
+            scanner_cycle_activity_status(&activity_before, activity_after, publication_blocked, None, false);
         let dirty_usage_status = dirty_usage_snapshot_status(&dirty_usage_snapshot);
         let status = classify_nsscanner_cycle(
             true,
@@ -583,6 +785,15 @@ where
     let serial_progress_accounting = budget.requires_serial_progress_accounting();
     let set_scan_limit = scanner_budgeted_concurrency_limit(dispatch_limit, serial_progress_accounting);
     let bucket_failures = ScannerBucketFailureState::default();
+    let bucket_completions = remote_start
+        .as_ref()
+        .map(|_| Arc::new(Mutex::new(ScannerBucketCompletions::new())));
+    // Recovery sweeps have no remote certification window, but test sessions
+    // still need the actual worker completions before any values are carried.
+    #[cfg(feature = "e2e-test-hooks")]
+    let bucket_completions = bucket_completions.or_else(|| {
+        std::env::var_os("RUSTFS_E2E_SCANNER_BARRIER_DIR").map(|_| Arc::new(Mutex::new(ScannerBucketCompletions::new())))
+    });
     let pending_maintenance_work = Arc::new(AtomicBool::new(false));
     let cold_zero_walk_reuse_observed = Arc::new(AtomicBool::new(false));
     record_set_scan_concurrency_limit(set_scan_limit);
@@ -679,6 +890,7 @@ where
             publication_epoch,
             dirty_usage_buckets: dirty_usage_snapshot.buckets.clone(),
             bucket_failures: bucket_failures.clone(),
+            bucket_completions: bucket_completions.clone(),
             pending_maintenance_work: pending_maintenance_work.clone(),
             cache_cycle_floor: cache_cycle_floor.clone(),
             cold_zero_walk_reuse_observed: cold_zero_walk_reuse_observed.clone(),
@@ -827,8 +1039,43 @@ where
             !namespace_not_found_buckets.is_empty(),
         );
     let dirty_usage_status = dirty_usage_snapshot_status(&dirty_usage_snapshot);
-    let (activity_status, remote_publication_lease_targets) =
-        scanner_cycle_activity_status(store, distributed, &activity_before).await;
+    // Check the movement barrier before sampling either edge. The two RPC
+    // windows are single observations; publication never waits for silence.
+    let publication_blocked = store.scanner_data_usage_publication_blocked().await;
+    let post_activity_before = probe_scanner_cycle_activity(
+        store,
+        distributed,
+        #[cfg(test)]
+        peer_observations.as_ref(),
+    )
+    .await;
+    let remote_certification_attempted = remote_start.is_some();
+    let (remote_certification, post_activity) = match (remote_start, post_activity_before) {
+        (Some(start), Ok(before)) => {
+            let (post_walk, after) = observe_remote_dirty_usage_window(
+                store,
+                &before,
+                "post_walk_snapshot",
+                leader_epoch,
+                want_cycle,
+                #[cfg(test)]
+                peer_observations.as_ref(),
+            )
+            .await;
+            (
+                post_walk.and_then(|post_walk| ScannerRemoteDirtyUsageCertification::new(start, post_walk)),
+                after,
+            )
+        }
+        (_, after) => (None, after),
+    };
+    let (activity_status, remote_publication_lease_targets) = scanner_cycle_activity_status(
+        &activity_before,
+        post_activity,
+        publication_blocked,
+        remote_certification.as_ref(),
+        remote_certification_attempted,
+    );
     let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::<Vec<_>>();
     let completed_usage = completed_usage_candidate(
         &results,
@@ -877,17 +1124,93 @@ where
     // stopped early is pending for the same reason: the publication carries
     // its last certified value, not this walk's (a `Complete` cycle has only
     // carried partial buckets, see `classify_nsscanner_cycle`).
+    let mut uncertified_buckets = failed_buckets.union(&partial_buckets).cloned().collect::<HashSet<_>>();
+    let local_pending_buckets = pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets);
+    uncertified_buckets.extend(local_pending_buckets.iter().cloned());
+    let completions = match bucket_completions {
+        Some(completions) => completions.lock().await.clone(),
+        None => ScannerBucketCompletions::new(),
+    };
+    let (certified_acknowledgements, remote_pending_buckets) = remote_certification.as_ref().map_or_else(
+        || (Vec::new(), ScannerCyclePendingBuckets::new()),
+        |proof| proof.classify_buckets(&expected_sources, &completions, &uncertified_buckets),
+    );
+    let remote_dirty_usage_pending = !remote_pending_buckets.is_empty();
+    uncertified_buckets.extend(remote_pending_buckets.iter().cloned());
     let pending_buckets = if cycle_status == ScannerCycleStatus::Complete {
-        let mut pending = pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets);
+        let mut pending = local_pending_buckets;
         pending.extend(partial_buckets.iter().cloned());
+        pending.extend(remote_pending_buckets);
         pending
     } else {
         ScannerCyclePendingBuckets::new()
     };
+    // A verified but empty scoped ACK is deliberate. It must never become a
+    // whole-generation ACK that could clear a hot or carried partial bucket.
+    let remote_dirty_usage_acknowledgements = if cycle_status != ScannerCycleStatus::Complete {
+        Vec::new()
+    } else if remote_certification.is_some() {
+        certified_acknowledgements
+    } else if remote_certification_attempted {
+        Vec::new()
+    } else if !remote_dirty_usage_acknowledgements.is_empty() {
+        remote_dirty_usage_acknowledgements
+    } else if scan_scope.is_default() {
+        crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before)
+    } else {
+        Vec::new()
+    };
     let mut publication_expectation = None;
     let usage_update_emitted = if let Some(candidate) = completed_usage {
         let (data_usage_info, expectation) = candidate.prepare(cycle_status, &pending_buckets);
         publication_expectation = expectation;
+        #[cfg(feature = "e2e-test-hooks")]
+        {
+            let digests = publication_expectation
+                .as_ref()
+                .map(ScannerPublicationExpectation::e2e_digests);
+            let scanned_buckets = completions
+                .values()
+                .flat_map(|buckets| buckets.keys().cloned())
+                .collect::<BTreeSet<_>>();
+            let mut set_coverage = completions.iter().collect::<Vec<_>>();
+            set_coverage.sort_by_key(|(source, _)| (source.pool_index, source.set_index));
+            let _ = crate::e2e_barrier::checkpoint(
+                "cycle_candidate",
+                serde_json::json!({
+                    "epoch": leader_epoch,
+                    "cycle": want_cycle,
+                    "scan_mode": if scan_mode == HealScanMode::Deep { "deep" } else { "normal" },
+                    "requested_scope": requested_scope,
+                    "selected_scope": e2e_scope_observation(&scan_scope),
+                    "scanned_buckets": scanned_buckets,
+                    "pending_buckets": data_usage_info.usage_snapshot_pending_buckets,
+                    "converged": data_usage_info.usage_snapshot_converged,
+                    "status": match cycle_status {
+                        ScannerCycleStatus::Complete => "complete",
+                        ScannerCycleStatus::Incomplete => "incomplete",
+                        ScannerCycleStatus::Superseded => "superseded",
+                        ScannerCycleStatus::Deferred(_) => "deferred",
+                    },
+                    "candidate_sha256": digests.as_ref().map(|digests| &digests.0),
+                    "coverage_sha256": digests.as_ref().map(|digests| &digests.1),
+                    "peer_certification": remote_certification.is_some(),
+                    "certified_peers": remote_certification.as_ref().map(|proof| {
+                        proof.post_walk.peers.iter().map(|(host, peer)| serde_json::json!({
+                            "host": host, "owner_id": peer.owner_id, "instance_id": peer.instance_id,
+                        })).collect::<Vec<_>>()
+                    }).unwrap_or_default(),
+                    "set_coverage": set_coverage.into_iter().map(|(source, buckets)| serde_json::json!({
+                        "pool": source.pool_index,
+                        "set": source.set_index,
+                        "buckets": buckets.keys().cloned().collect::<BTreeSet<_>>(),
+                        "bucket_incarnations": buckets,
+                    })).collect::<Vec<_>>(),
+                    "acks": crate::e2e_barrier::acknowledgement_observations(&remote_dirty_usage_acknowledgements),
+                }),
+            )
+            .await;
+        }
         if should_publish_observational_snapshot(cycle_status) {
             publish_observational_snapshot(&updates, data_usage_info).await?
         } else {
@@ -902,7 +1225,6 @@ where
     };
     // A walk that stopped early certifies nothing: the bucket keeps its
     // dirty generation like a failed one and is walked again next cycle.
-    let uncertified_buckets = failed_buckets.union(&partial_buckets).cloned().collect::<HashSet<_>>();
     let dirty_usage_clear = should_clear_dirty_usage_snapshot(
         result.is_ok(),
         structurally_complete_snapshot,
@@ -915,14 +1237,6 @@ where
     if cycle_status == ScannerCycleStatus::Complete {
         complete_tier_registry_cycle(want_cycle, leader_epoch);
     }
-    let remote_dirty_usage_acknowledgements =
-        if cycle_status == ScannerCycleStatus::Complete && !remote_dirty_usage_acknowledgements.is_empty() {
-            remote_dirty_usage_acknowledgements
-        } else if cycle_status == ScannerCycleStatus::Complete && scan_scope.is_default() {
-            crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before)
-        } else {
-            Vec::new()
-        };
     Ok(ScannerCycleResult::new(cycle_status, dirty_usage_clear)
         .with_publication_epoch(publication_epoch)
         .with_activity_digest(activity_digest)
@@ -932,6 +1246,8 @@ where
         .with_partial_bucket_progress(partial_bucket_progress)
         .with_remote_publication_lease_targets(remote_publication_lease_targets)
         .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
+        .with_remote_dirty_usage_certification(remote_certification)
+        .with_remote_dirty_usage_pending(remote_dirty_usage_pending)
         .with_distributed_segment_invalidation_evidence(distributed_segment_invalidation_evidence)
         .with_segment_reuse_activation_preflight(segment_reuse_activation_preflight)
         .with_failed_dirty_usage(!failed_buckets.is_empty())
@@ -940,3 +1256,11 @@ where
         .with_scope_fallback(scope_fallback)
         .with_publication_expectation(publication_expectation))
 }
+
+#[cfg(feature = "e2e-test-hooks")]
+fn e2e_scope_observation(scope: &ScannerBucketScanScope) -> serde_json::Value {
+    match &scope.selected_buckets {
+        Some(buckets) => serde_json::json!({"kind": "buckets", "buckets": buckets.iter().collect::<BTreeSet<_>>()}),
+        None => serde_json::json!({"kind": "all"}),
+    }
+}

crates/scanner/src/scanner_io/peer_certification.rs

@@ -0,0 +1,582 @@
+// Copyright 2026 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.
+
+use super::*;
+use crate::scanner::{
+    ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, ScannerDirtyUsageAcknowledgementKind,
+    scanner_activity_allows_usage_publication, scanner_activity_dirty_usage_state_for_host, scanner_activity_guardrail_veto,
+    scanner_activity_publication_lease_targets,
+};
+use crate::storage_api::EcstoreScannerPublicationLeaseGrant;
+
+/// Only worker results received in this cycle enter this map. A carried value
+/// or a cold bucket copied from the old set root is not a completion.
+pub(super) type ScannerBucketCompletions = HashMap<DataUsageCacheSource, HashMap<String, uuid::Uuid>>;
+
+#[derive(Debug)]
+pub(super) struct ScannerRemoteDirtyUsageWindow {
+    pub(super) before: ScannerActivitySnapshot,
+    pub(super) after: ScannerActivitySnapshot,
+    pub(super) peers: BTreeMap<String, EcstoreScannerPeerDirtyUsageSnapshot>,
+}
+
+fn remote_generations_advance(before: &ScannerActivitySnapshot, after: &ScannerActivitySnapshot) -> bool {
+    scanner_activity_allows_usage_publication(before)
+        && scanner_activity_allows_usage_publication(after)
+        && scanner_activity_guardrail_veto(before, after).is_none()
+        && scanner_activity_publication_lease_targets(before).iter().all(|(host, _, _)| {
+            let Some((_, before_generation, _)) = scanner_activity_dirty_usage_state_for_host(before, host) else {
+                return false;
+            };
+            scanner_activity_dirty_usage_state_for_host(after, host)
+                .is_some_and(|(_, after_generation, _)| before_generation <= after_generation && after_generation < u64::MAX)
+        })
+}
+
+impl ScannerRemoteDirtyUsageWindow {
+    /// The caller obtains these in order: before activity, dirty snapshot,
+    /// after activity. The generation can advance without waiting for silence.
+    pub(super) fn verify(
+        before: ScannerActivitySnapshot,
+        snapshots: Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)>,
+        after: ScannerActivitySnapshot,
+    ) -> Option<Self> {
+        let expected = scanner_activity_publication_lease_targets(&before);
+        if expected.is_empty() || expected.len() != snapshots.len() || !remote_generations_advance(&before, &after) {
+            return None;
+        }
+        let mut peers = BTreeMap::new();
+        for (host, snapshot) in snapshots {
+            let (instance_id, lower, lower_pending) = scanner_activity_dirty_usage_state_for_host(&before, &host)?;
+            let (_, upper, upper_pending) = scanner_activity_dirty_usage_state_for_host(&after, &host)?;
+            let pending = !snapshot.buckets.is_empty();
+            if host.is_empty()
+                || snapshot.instance_id != instance_id
+                || uuid::Uuid::parse_str(&snapshot.owner_id)
+                    .ok()
+                    .is_none_or(|owner| owner.is_nil())
+                || uuid::Uuid::parse_str(&snapshot.instance_id)
+                    .ok()
+                    .is_none_or(|instance| instance.is_nil())
+                || snapshot.protocol_version != crate::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION
+                || !snapshot.complete
+                || snapshot.buckets.len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES
+                || snapshot.pending_bucket_count != u64::try_from(snapshot.buckets.len()).ok()?
+                || !(lower <= snapshot.generation && snapshot.generation <= upper && upper < u64::MAX)
+                || (lower == snapshot.generation && lower_pending != pending)
+                || (upper == snapshot.generation && upper_pending != pending)
+                || snapshot.buckets.iter().any(|(bucket, entry)| {
+                    bucket.is_empty()
+                        || bucket == DATA_USAGE_ROOT
+                        || entry.bucket_incarnation.is_nil()
+                        || entry.generation == 0
+                        || entry.generation > snapshot.generation
+                })
+                || peers.insert(host, snapshot).is_some()
+            {
+                return None;
+            }
+        }
+        (peers.len() == expected.len() && expected.iter().all(|(host, _, _)| peers.contains_key(host))).then_some(Self {
+            before,
+            after,
+            peers,
+        })
+    }
+
+    pub(super) fn activity_for_walk(&self) -> Arc<ScannerActivitySnapshot> {
+        Arc::new(self.after.clone())
+    }
+
+    pub(super) fn strict_remote_usage(&self) -> Option<VerifiedRemoteDirtyUsage> {
+        if crate::scanner::scanner_peer_dirty_usage_moved(&self.before, &self.after).is_some() {
+            return None;
+        }
+        let expected = self
+            .peers
+            .keys()
+            .map(|host| {
+                let (instance_id, generation, pending) = scanner_activity_dirty_usage_state_for_host(&self.before, host)?;
+                Some((
+                    host.clone(),
+                    ScannerPeerDirtyUsageExpectation {
+                        instance_id: instance_id.to_string(),
+                        generation,
+                        pending,
+                    },
+                ))
+            })
+            .collect::<Option<HashMap<_, _>>>()?;
+        verified_remote_dirty_usage(&expected, self.peers.iter().map(|(host, peer)| (host.clone(), peer.clone())).collect())
+    }
+
+    #[cfg(feature = "e2e-test-hooks")]
+    pub(super) fn observation(&self, epoch: u64, cycle: u64) -> serde_json::Value {
+        serde_json::json!({
+            "epoch": epoch,
+            "cycle": cycle,
+            "verified": true,
+            "activity_before": crate::e2e_barrier::activity_observations(&self.before),
+            "activity_after": crate::e2e_barrier::activity_observations(&self.after),
+            "peers": self.peers.iter().map(|(host, peer)| serde_json::json!({
+                "host": host,
+                "owner_id": peer.owner_id,
+                "instance_id": peer.instance_id,
+                "generation": peer.generation,
+                "protocol_version": peer.protocol_version,
+                "complete": peer.complete,
+                "pending_bucket_count": peer.pending_bucket_count,
+                "buckets": peer.buckets.iter().map(|(bucket, entry)| serde_json::json!({
+                    "bucket": bucket,
+                    "bucket_incarnation": entry.bucket_incarnation,
+                    "generation": entry.generation,
+                })).collect::<Vec<_>>(),
+            })).collect::<Vec<_>>(),
+        })
+    }
+
+    pub(super) fn dirty_buckets(&self) -> HashSet<String> {
+        self.peers.values().flat_map(|peer| peer.buckets.keys().cloned()).collect()
+    }
+
+    pub(super) fn acknowledgements(&self) -> Vec<ScannerDirtyUsageAcknowledgement> {
+        self.peers
+            .iter()
+            .filter(|(_, peer)| !peer.buckets.is_empty())
+            .map(|(host, peer)| ScannerDirtyUsageAcknowledgement {
+                host: host.clone(),
+                instance_id: peer.instance_id.clone(),
+                kind: ScannerDirtyUsageAcknowledgementKind::Scoped {
+                    owner_id: peer.owner_id.clone(),
+                    entries: peer
+                        .buckets
+                        .iter()
+                        .map(|(bucket, entry)| crate::storage_api::EcstoreScannerScopedDirtyUsageAckEntry {
+                            bucket: bucket.clone(),
+                            bucket_incarnation: entry.bucket_incarnation,
+                            generation: entry.generation,
+                        })
+                        .collect(),
+                },
+            })
+            .collect()
+    }
+}
+
+/// A complete producer proof includes clean peers even when it yields no ACK.
+/// The start and post-walk maps remain sealed until publication and lease
+/// validation; post-lease writes are retained by exact receiver comparison.
+#[derive(Debug)]
+pub(crate) struct ScannerRemoteDirtyUsageCertification {
+    pub(super) start: ScannerRemoteDirtyUsageWindow,
+    pub(super) post_walk: ScannerRemoteDirtyUsageWindow,
+}
+
+impl ScannerRemoteDirtyUsageCertification {
+    pub(super) fn new(start: ScannerRemoteDirtyUsageWindow, post_walk: ScannerRemoteDirtyUsageWindow) -> Option<Self> {
+        if !remote_generations_advance(&start.after, &post_walk.before)
+            || start.peers.len() != post_walk.peers.len()
+            || start.peers.iter().any(|(host, before)| {
+                post_walk.peers.get(host).is_none_or(|after| {
+                    before.owner_id != after.owner_id
+                        || before.instance_id != after.instance_id
+                        || before.protocol_version != after.protocol_version
+                        || before.generation > after.generation
+                })
+            })
+        {
+            return None;
+        }
+        Some(Self { start, post_walk })
+    }
+
+    #[cfg(test)]
+    pub(crate) fn for_tests(
+        activity: ScannerActivitySnapshot,
+        peers: Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)>,
+    ) -> Option<Self> {
+        let start = ScannerRemoteDirtyUsageWindow::verify(activity.clone(), peers.clone(), activity.clone())?;
+        let post = ScannerRemoteDirtyUsageWindow::verify(activity.clone(), peers, activity)?;
+        Self::new(start, post)
+    }
+
+    pub(super) fn classify_buckets(
+        &self,
+        expected_sources: &HashSet<DataUsageCacheSource>,
+        completions: &ScannerBucketCompletions,
+        uncertified: &HashSet<String>,
+    ) -> (Vec<ScannerDirtyUsageAcknowledgement>, ScannerCyclePendingBuckets) {
+        let mut buckets = self.start.dirty_buckets();
+        buckets.extend(self.post_walk.dirty_buckets());
+        let sources_complete = !expected_sources.is_empty()
+            && completions.len() == expected_sources.len()
+            && expected_sources.iter().all(|source| completions.contains_key(source));
+        let pending = buckets
+            .iter()
+            .filter(|bucket| {
+                !sources_complete
+                    || uncertified.contains(*bucket)
+                    || self.start.peers.iter().any(|(host, before)| {
+                        let start = before.buckets.get(*bucket);
+                        let post_walk = self.post_walk.peers[host].buckets.get(*bucket);
+                        start != post_walk
+                            || start.is_some_and(|entry| {
+                                expected_sources.iter().any(|source| {
+                                    completions.get(source).and_then(|set| set.get(*bucket)) != Some(&entry.bucket_incarnation)
+                                })
+                            })
+                    })
+            })
+            .cloned()
+            .collect::<ScannerCyclePendingBuckets>();
+        let acknowledgements = self
+            .start
+            .acknowledgements()
+            .into_iter()
+            .filter_map(|mut acknowledgement| {
+                let ScannerDirtyUsageAcknowledgementKind::Scoped { entries, .. } = &mut acknowledgement.kind else {
+                    return None;
+                };
+                entries.retain(|entry| !pending.contains(&entry.bucket));
+                (!entries.is_empty()).then_some(acknowledgement)
+            })
+            .collect();
+        (acknowledgements, pending)
+    }
+
+    pub(crate) fn matches_activity(&self, activity: &ScannerActivitySnapshot) -> bool {
+        remote_generations_advance(&self.post_walk.after, activity)
+    }
+
+    pub(crate) fn matches_publication_leases(&self, grants: &[EcstoreScannerPublicationLeaseGrant]) -> bool {
+        if grants.len() != self.post_walk.peers.len() {
+            return false;
+        }
+        let targets = scanner_activity_publication_lease_targets(&self.post_walk.after)
+            .into_iter()
+            .map(|(host, instance, movement)| (host, (instance, movement)))
+            .collect::<BTreeMap<_, _>>();
+        let mut hosts = HashSet::new();
+        grants.iter().all(|grant| {
+            let Some(peer) = self.post_walk.peers.get(&grant.host) else {
+                return false;
+            };
+            hosts.insert(grant.host.as_str())
+                && grant.lease.is_valid()
+                && !grant.lease.observational_only
+                && !grant.lease.token.is_nil()
+                && grant.lease.owner_id == peer.owner_id
+                && grant.lease.session_id == peer.instance_id
+                && targets.get(&grant.host).is_some_and(|(instance, movement)| {
+                    instance == &grant.lease.session_id && *movement == grant.lease.movement_generation
+                })
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const HOT: uuid::Uuid = uuid::Uuid::from_u128(71);
+    const COLD: uuid::Uuid = uuid::Uuid::from_u128(72);
+
+    fn peers(generation: u64, hot_generation: u64) -> Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)> {
+        ["peer-a", "peer-b"]
+            .into_iter()
+            .enumerate()
+            .map(|(index, host)| {
+                let buckets = if index == 0 {
+                    BTreeMap::from([
+                        (
+                            "cold-bucket".to_string(),
+                            crate::storage_api::EcstoreScannerPeerDirtyUsageBucket {
+                                bucket_incarnation: COLD,
+                                generation: 8,
+                            },
+                        ),
+                        (
+                            "hot-bucket".to_string(),
+                            crate::storage_api::EcstoreScannerPeerDirtyUsageBucket {
+                                bucket_incarnation: HOT,
+                                generation: hot_generation,
+                            },
+                        ),
+                    ])
+                } else {
+                    BTreeMap::new()
+                };
+                (
+                    host.to_string(),
+                    EcstoreScannerPeerDirtyUsageSnapshot {
+                        owner_id: uuid::Uuid::from_u128(101 + index as u128).simple().to_string(),
+                        instance_id: uuid::Uuid::from_u128(201 + index as u128).simple().to_string(),
+                        generation: if index == 0 { generation } else { 7 },
+                        pending_bucket_count: buckets.len() as u64,
+                        protocol_version: crate::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION,
+                        complete: true,
+                        buckets,
+                    },
+                )
+            })
+            .collect()
+    }
+
+    fn activity(generation: u64) -> ScannerActivitySnapshot {
+        peers(generation, generation)
+            .into_iter()
+            .map(|(host, peer)| {
+                (
+                    host,
+                    crate::scanner::scanner_node_activity_for_tests(
+                        &peer.instance_id,
+                        peer.generation,
+                        peer.generation,
+                        !peer.buckets.is_empty(),
+                    ),
+                )
+            })
+            .collect()
+    }
+
+    fn window(lower: u64, generation: u64, upper: u64) -> ScannerRemoteDirtyUsageWindow {
+        ScannerRemoteDirtyUsageWindow::verify(activity(lower), peers(generation, generation), activity(upper))
+            .expect("ordered complete producer window")
+    }
+
+    fn completions() -> (HashSet<DataUsageCacheSource>, ScannerBucketCompletions) {
+        let sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(0, 1)]);
+        let completed = sources
+            .iter()
+            .map(|source| {
+                (
+                    *source,
+                    HashMap::from([("cold-bucket".to_string(), COLD), ("hot-bucket".to_string(), HOT)]),
+                )
+            })
+            .collect();
+        (sources, completed)
+    }
+
+    #[test]
+    fn hot_writes_cross_both_bounded_windows_while_the_clean_peer_stays_in_the_proof() {
+        let start = window(9, 10, 11);
+        assert!(start.strict_remote_usage().is_none(), "the relaxed window grants no segment reuse proof");
+        let proof = ScannerRemoteDirtyUsageCertification::new(start, window(12, 13, 14)).expect("same producers");
+        let (sources, completed) = completions();
+        let (acks, pending) = proof.classify_buckets(&sources, &completed, &HashSet::new());
+        assert_eq!(pending, BTreeSet::from(["hot-bucket".to_string()]));
+        assert_eq!(acks.len(), 1);
+        assert_eq!(proof.post_walk.peers.len(), 2);
+        assert!(proof.post_walk.peers["peer-b"].buckets.is_empty());
+        let ScannerDirtyUsageAcknowledgementKind::Scoped { entries, .. } = &acks[0].kind else {
+            panic!("scoped ACK")
+        };
+        assert_eq!(entries.len(), 1);
+        assert_eq!(
+            (entries[0].bucket.as_str(), entries[0].bucket_incarnation, entries[0].generation),
+            ("cold-bucket", COLD, 8)
+        );
+    }
+
+    #[test]
+    fn producer_windows_reject_missing_duplicate_stale_future_overflow_and_incomplete_evidence() {
+        for case in [
+            "missing_peer",
+            "duplicate_peer",
+            "old_snapshot",
+            "future_snapshot",
+            "overflow",
+            "owner",
+            "instance",
+            "protocol",
+            "incomplete",
+            "count",
+            "future_entry",
+            "nil_incarnation",
+            "empty_bucket",
+            "entry_limit",
+            "pending",
+        ] {
+            let mut snapshots = peers(10, 10);
+            let before = activity(9);
+            let mut after = activity(11);
+            match case {
+                "missing_peer" => {
+                    snapshots.pop();
+                }
+                "duplicate_peer" => {
+                    snapshots[1] = snapshots[0].clone();
+                }
+                "old_snapshot" => snapshots[0].1.generation = 8,
+                "future_snapshot" => snapshots[0].1.generation = 12,
+                "overflow" => after = activity(u64::MAX),
+                "owner" => snapshots[0].1.owner_id = uuid::Uuid::nil().to_string(),
+                "instance" => snapshots[0].1.instance_id = uuid::Uuid::from_u128(999).to_string(),
+                "protocol" => snapshots[0].1.protocol_version += 1,
+                "incomplete" => snapshots[0].1.complete = false,
+                "count" => snapshots[0].1.pending_bucket_count += 1,
+                "future_entry" => snapshots[0].1.buckets.get_mut("cold-bucket").expect("cold").generation = 11,
+                "nil_incarnation" => {
+                    snapshots[0]
+                        .1
+                        .buckets
+                        .get_mut("cold-bucket")
+                        .expect("cold")
+                        .bucket_incarnation = uuid::Uuid::nil()
+                }
+                "empty_bucket" => {
+                    let entry = snapshots[0].1.buckets.remove("cold-bucket").expect("cold");
+                    snapshots[0].1.buckets.insert(String::new(), entry);
+                }
+                "entry_limit" => {
+                    let peer = &mut snapshots[0].1;
+                    peer.buckets = (0..=crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES)
+                        .map(|index| {
+                            (
+                                format!("bucket-{index}"),
+                                crate::storage_api::EcstoreScannerPeerDirtyUsageBucket {
+                                    bucket_incarnation: COLD,
+                                    generation: 8,
+                                },
+                            )
+                        })
+                        .collect();
+                    peer.pending_bucket_count = peer.buckets.len() as u64;
+                }
+                "pending" => {
+                    snapshots[0].1.generation = 11;
+                    snapshots[0].1.buckets.clear();
+                    snapshots[0].1.pending_bucket_count = 0;
+                }
+                _ => unreachable!(),
+            }
+            assert!(ScannerRemoteDirtyUsageWindow::verify(before, snapshots, after).is_none(), "{case}");
+        }
+    }
+
+    #[test]
+    fn certification_rejects_owner_replacement_and_a_generation_rollback_between_windows() {
+        let mut changed = window(12, 13, 14);
+        changed.peers.get_mut("peer-b").expect("clean peer").owner_id = uuid::Uuid::from_u128(999).to_string();
+        assert!(ScannerRemoteDirtyUsageCertification::new(window(9, 10, 11), changed).is_none());
+        assert!(ScannerRemoteDirtyUsageCertification::new(window(9, 10, 11), window(8, 9, 10)).is_none());
+    }
+
+    #[test]
+    fn certification_requires_actual_coverage_and_the_same_incarnation_on_every_set() {
+        let proof = ScannerRemoteDirtyUsageCertification::new(window(10, 10, 10), window(10, 10, 10)).expect("stable peers");
+        for case in ["missing_set", "missing_bucket", "wrong_incarnation", "partial"] {
+            let (sources, mut completed) = completions();
+            let mut uncertified = HashSet::new();
+            match case {
+                "missing_set" => {
+                    completed.remove(&DataUsageCacheSource::new(0, 1));
+                }
+                "missing_bucket" => {
+                    completed
+                        .get_mut(&DataUsageCacheSource::new(0, 1))
+                        .expect("set")
+                        .remove("cold-bucket");
+                }
+                "wrong_incarnation" => {
+                    completed
+                        .get_mut(&DataUsageCacheSource::new(0, 1))
+                        .expect("set")
+                        .insert("cold-bucket".to_string(), uuid::Uuid::from_u128(999));
+                }
+                "partial" => {
+                    uncertified.insert("cold-bucket".to_string());
+                }
+                _ => unreachable!(),
+            }
+            let (acks, pending) = proof.classify_buckets(&sources, &completed, &uncertified);
+            assert!(pending.contains("cold-bucket"), "{case}");
+            assert!(acks.iter().all(|ack| matches!(&ack.kind, ScannerDirtyUsageAcknowledgementKind::Scoped { entries, .. } if entries.iter().all(|entry| entry.bucket != "cold-bucket"))), "{case}");
+        }
+    }
+
+    #[test]
+    fn a_bucket_changed_on_one_peer_cannot_be_acknowledged_on_another() {
+        let start = window(10, 10, 10);
+        let mut post_peers = peers(10, 10);
+        post_peers[1].1.generation = 8;
+        post_peers[1].1.pending_bucket_count = 1;
+        post_peers[1].1.buckets.insert(
+            "cold-bucket".to_string(),
+            crate::storage_api::EcstoreScannerPeerDirtyUsageBucket {
+                bucket_incarnation: COLD,
+                generation: 8,
+            },
+        );
+        let mut after = activity(10);
+        after.insert(
+            "peer-b".to_string(),
+            crate::scanner::scanner_node_activity_for_tests(&post_peers[1].1.instance_id, 8, 8, true),
+        );
+        let post = ScannerRemoteDirtyUsageWindow::verify(activity(10), post_peers, after).expect("clean peer became dirty");
+        let proof = ScannerRemoteDirtyUsageCertification::new(start, post).expect("same producers");
+        let (sources, completed) = completions();
+        let (acks, pending) = proof.classify_buckets(&sources, &completed, &HashSet::new());
+        assert!(pending.contains("cold-bucket"));
+        assert!(acks.iter().all(|ack| matches!(&ack.kind, ScannerDirtyUsageAcknowledgementKind::Scoped { entries, .. } if entries.iter().all(|entry| entry.bucket != "cold-bucket"))));
+    }
+
+    #[test]
+    fn start_window_trailing_activity_invalidates_same_cycle_current_cache_without_segment_reuse() {
+        let window = window(9, 10, 11);
+        let old_plan = scanner_bucket_cache_digest(
+            DataUsageScanPlanDigest(crate::scanner::scanner_activity_snapshot_digest(&window.before)),
+            None,
+        );
+        let new_plan = scanner_bucket_cache_digest(
+            DataUsageScanPlanDigest(crate::scanner::scanner_activity_snapshot_digest(&window.activity_for_walk())),
+            None,
+        );
+        let source = DataUsageCacheSource::new(0, 0);
+        let mut cache = DataUsageCache {
+            info: DataUsageCacheInfo {
+                name: "cold-bucket".to_string(),
+                next_cycle: 7,
+                leader_epoch: 11,
+                source: Some(source),
+                last_update: Some(SystemTime::UNIX_EPOCH),
+                snapshot_complete: true,
+                scan_plan_digest: Some(old_plan),
+                cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        cache.replace("cold-bucket", "", DataUsageEntry::default());
+        let options = || DataUsageCacheReuseOptions {
+            require_source: true,
+            tier_registry_generation: None,
+            checkpoint_identity: None,
+            checkpoint_plan_digest: None,
+        };
+        assert!(matches!(
+            current_cache_root_or_prepare_with_generation(&mut cache.clone(), "cold-bucket", source, 7, 11, old_plan, options()),
+            DataUsageCacheScanState::Current(_)
+        ));
+        assert!(matches!(
+            current_cache_root_or_prepare_with_generation(&mut cache, "cold-bucket", source, 7, 11, new_plan, options()),
+            DataUsageCacheScanState::Prepared {
+                outcome: DataUsageCachePrepareOutcome::Reset,
+                ..
+            }
+        ));
+        assert!(window.strict_remote_usage().is_none());
+    }
+}

crates/scanner/src/scanner_io/tests.rs

@@ -1332,6 +1332,7 @@ async fn scoped_scan_production_entry_preserves_deep_and_full_maintenance_work()
                     requires_full_scan,
                     resolved_scope_observer: Some(observer),
                     post_walk_gate: None,
+                    peer_observations: None,
                     service_cohort: None,
                 },
             ),
@@ -1423,6 +1424,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
                     requires_full_scan: false,
                     resolved_scope_observer: None,
                     post_walk_gate: None,
+                    peer_observations: None,
                     service_cohort: None,
                 },
             ),
@@ -1477,6 +1479,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure(
                     requires_full_scan,
                     resolved_scope_observer: None,
                     post_walk_gate: None,
+                    peer_observations: None,
                     service_cohort: None,
                 },
             ),
@@ -2468,6 +2471,7 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers
             publication_epoch: Some(epoch),
             dirty_usage_buckets: Arc::new(HashMap::new()),
             bucket_failures: ScannerBucketFailureState::default(),
+            bucket_completions: None,
             pending_maintenance_work: Arc::new(AtomicBool::new(false)),
             cache_cycle_floor: Arc::new(AtomicU64::new(8)),
             cold_zero_walk_reuse_observed: Arc::new(AtomicBool::new(false)),
@@ -3385,6 +3389,7 @@ async fn distributed_scoped_scan_falls_back_when_remote_ack_exceeds_protocol_bat
         store.as_ref(),
         true,
         super::io_cycle::ScannerBucketScopeResolution {
+            remote_window: None,
             requested_scope: ScannerBucketScanScope::default(),
             baseline_proof: ScannerCacheBaselineProof {
                 authoritative_data: Some(&baseline),
@@ -3444,6 +3449,7 @@ async fn distributed_scoped_scan_falls_back_when_remote_scoped_ack_capability_is
             store.as_ref(),
             true,
             super::io_cycle::ScannerBucketScopeResolution {
+                remote_window: None,
                 requested_scope: ScannerBucketScanScope::default(),
                 baseline_proof: ScannerCacheBaselineProof {
                     authoritative_data: Some(&baseline),
@@ -5015,6 +5021,7 @@ async fn replication_maintenance_refreshes_compacted_stats_across_cycles_and_ret
                 service_cohort: None,
                 resolved_scope_observer: None,
                 post_walk_gate: None,
+                peer_observations: None,
             },
         )
         .await
@@ -5122,6 +5129,7 @@ async fn replication_maintenance_refreshes_compacted_stats_across_cycles_and_ret
                 service_cohort: None,
                 resolved_scope_observer: None,
                 post_walk_gate: None,
+                peer_observations: None,
             },
         )
         .await
@@ -5269,6 +5277,7 @@ async fn run_full_cycle(store: &Arc<ECStore>, cycle: u64) -> ScannerCycleStatus
                 resolved_scope_observer: None,
                 service_cohort: None,
                 post_walk_gate: None,
+                peer_observations: None,
             },
         ),
     )

crates/scanner/src/scanner_io/tests/checkpoint_coverage_identity.rs

@@ -54,6 +54,7 @@ async fn run_full_cycle(
                 service_cohort: None,
                 resolved_scope_observer: None,
                 post_walk_gate: None,
+                peer_observations: None,
             },
         ),
     )

crates/scanner/src/scanner_io/tests/collector_lifecycle.rs

@@ -140,6 +140,7 @@ async fn scan_plan(set: &Arc<SetDisks>, names: &[&str], cycle: u64) -> ScannerBu
         ),
         dirty_usage_buckets: Arc::new(dirty_usage_buckets_for_tests()),
         bucket_failures: ScannerBucketFailureState::default(),
+        bucket_completions: None,
         pending_maintenance_work: Arc::new(AtomicBool::new(false)),
         cache_cycle_floor: Arc::new(AtomicU64::new(cycle)),
         cold_zero_walk_reuse_observed: Arc::new(AtomicBool::new(false)),

crates/scanner/src/scanner_io/tests/listing_window_scope.rs

@@ -47,6 +47,7 @@ async fn run_cycle(
                 service_cohort: None,
                 resolved_scope_observer: Some(observer),
                 post_walk_gate: None,
+                peer_observations: None,
             },
         ),
     )

crates/scanner/src/scanner_io/tests/partial_bucket_carry.rs

@@ -57,6 +57,7 @@ async fn run_cycle(store: &Arc<ECStore>, cycle: u64, options: CycleOptions) -> (
                 service_cohort: None,
                 resolved_scope_observer: options.resolved_scope_observer,
                 post_walk_gate: None,
+                peer_observations: None,
             },
         ),
     )

crates/scanner/src/scanner_io/tests/per_bucket_certification.rs

@@ -22,6 +22,7 @@ async fn put_fixture_object(store: &Arc<ECStore>, bucket: &str, object: &str) {
 struct CycleHooks {
     resolved_scope_observer: Option<tokio::sync::oneshot::Sender<ScannerBucketScanScope>>,
     post_walk_gate: Option<tokio::sync::oneshot::Receiver<()>>,
+    peer_observations: Option<super::super::io_cycle::ScannerCyclePeerObservations>,
 }
 
 async fn run_cycle(
@@ -31,6 +32,7 @@ async fn run_cycle(
     requires_full_scan: bool,
     hooks: CycleHooks,
 ) -> (ScannerCycleResult, DataUsageInfo) {
+    let peer_cycle = hooks.peer_observations.is_some();
     let ctx = CancellationToken::new();
     let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
     let (updates, mut receiver) = mpsc::channel(1);
@@ -52,12 +54,20 @@ async fn run_cycle(
                 service_cohort: None,
                 resolved_scope_observer: hooks.resolved_scope_observer,
                 post_walk_gate: hooks.post_walk_gate,
+                peer_observations: hooks.peer_observations,
             },
         ),
     )
     .await
     .expect("cycle should finish within the test deadline")
     .expect("cycle should succeed");
+    if peer_cycle {
+        assert_eq!(
+            result.status,
+            ScannerCycleStatus::Complete,
+            "one remote hot bucket cannot supersede certified cold buckets"
+        );
+    }
     let snapshot = receiver.recv().await.expect("cycle should publish a snapshot");
     assert!(receiver.recv().await.is_none(), "a cycle publishes one terminal snapshot");
     (result, snapshot)
@@ -91,6 +101,7 @@ async fn write_during_walk_leaves_only_that_bucket_pending_and_scopes_the_next_c
         CycleHooks {
             resolved_scope_observer: None,
             post_walk_gate: None,
+            peer_observations: None,
         },
     )
     .await;
@@ -128,6 +139,7 @@ async fn write_during_walk_leaves_only_that_bucket_pending_and_scopes_the_next_c
         CycleHooks {
             resolved_scope_observer: Some(observer),
             post_walk_gate: Some(gate),
+            peer_observations: None,
         },
     )
     .await;
@@ -175,6 +187,7 @@ async fn write_during_walk_leaves_only_that_bucket_pending_and_scopes_the_next_c
         CycleHooks {
             resolved_scope_observer: Some(observer),
             post_walk_gate: None,
+            peer_observations: None,
         },
     )
     .await;
@@ -226,6 +239,7 @@ async fn pending_put_tail_across_the_walk_no_longer_vetoes_the_cycle() {
         CycleHooks {
             resolved_scope_observer: None,
             post_walk_gate: None,
+            peer_observations: None,
         },
     )
     .await;
@@ -285,6 +299,7 @@ async fn guardrail_veto_still_supersedes_the_cycle_onto_the_observed_route() {
         CycleHooks {
             resolved_scope_observer: Some(observer),
             post_walk_gate: Some(gate),
+            peer_observations: None,
         },
     )
     .await;
@@ -306,3 +321,230 @@ async fn guardrail_veto_still_supersedes_the_cycle_onto_the_observed_route() {
     );
     clear_dirty_usage_buckets_for_tests();
 }
+
+#[tokio::test]
+#[serial]
+async fn remote_hot_bucket_keeps_cold_buckets_certified_and_scopes_the_next_cycle() {
+    run_remote_certification_cycle(false).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn a_verified_empty_ack_on_a_full_scope_keeps_the_remote_hot_bucket_pending() {
+    run_remote_certification_cycle(true).await;
+}
+
+async fn run_remote_certification_cycle(empty_ack: bool) {
+    let (_temp_dir, store) = setup_local_scanner_store(1, 1).await;
+    clear_dirty_usage_buckets_for_tests();
+    let mut incarnations = BTreeMap::new();
+    for bucket in BUCKETS {
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("fixture bucket");
+        put_fixture_object(&store, bucket, "initial").await;
+        record_dirty_usage_bucket(bucket);
+        incarnations.insert(
+            bucket.to_string(),
+            store.all_set_disks()[0]
+                .find_bucket_incarnation_id_from_disk(bucket)
+                .await
+                .expect("read actual incarnation")
+                .expect("fixture bucket has an incarnation"),
+        );
+    }
+    wait_for_namespace_commit_tails(store.as_ref()).await;
+    let peers = Arc::new(StdMutex::new(BTreeMap::from([
+        (
+            "peer-a:9000".to_string(),
+            peer_dirty_usage_snapshot(&Uuid::from_u128(101).to_string(), 7, true, &[]),
+        ),
+        (
+            "peer-b:9000".to_string(),
+            peer_dirty_usage_snapshot(&Uuid::from_u128(102).to_string(), 7, true, &[]),
+        ),
+    ])));
+    // These are the RPC observations only. Bucket listing, traversal, cache
+    // coverage, classification and candidate production use the real cycle.
+    let activity_peers = Arc::clone(&peers);
+    let dirty_peers = Arc::clone(&peers);
+    let observations = super::super::io_cycle::ScannerCyclePeerObservations {
+        activity: Arc::new(move || {
+            activity_peers
+                .lock()
+                .expect("peer fixture lock")
+                .iter()
+                .map(|(host, peer)| {
+                    (
+                        host.clone(),
+                        crate::scanner::scanner_node_activity_for_tests(
+                            &peer.instance_id,
+                            peer.generation,
+                            peer.generation,
+                            !peer.buckets.is_empty(),
+                        ),
+                    )
+                })
+                .collect()
+        }),
+        dirty_usage: Arc::new(move || {
+            dirty_peers
+                .lock()
+                .expect("peer fixture lock")
+                .iter()
+                .map(|(host, snapshot)| (host.clone(), snapshot.clone()))
+                .collect()
+        }),
+        scoped_ack_capable: true,
+    };
+    let (result, baseline) = run_cycle(
+        &store,
+        1,
+        None,
+        true,
+        CycleHooks {
+            resolved_scope_observer: None,
+            post_walk_gate: None,
+            peer_observations: Some(observations.clone()),
+        },
+    )
+    .await;
+    assert_eq!(result.status, ScannerCycleStatus::Complete);
+    assert!(result.clear_verified_usage().is_empty());
+    assert!(baseline.is_complete_bucket_usage_snapshot());
+    let initial_dirty = if empty_ack {
+        vec!["hot-bucket"]
+    } else {
+        vec!["cold-bucket", "idle-bucket"]
+    };
+    for bucket in initial_dirty {
+        put_fixture_object(&store, bucket, "second").await;
+        let mut locked = peers.lock().expect("peer fixture lock");
+        let peer = locked.get_mut("peer-a:9000").expect("remote producer exists");
+        peer.generation += 1;
+        peer.buckets.insert(
+            bucket.to_string(),
+            crate::storage_api::EcstoreScannerPeerDirtyUsageBucket {
+                bucket_incarnation: incarnations[bucket],
+                generation: peer.generation,
+            },
+        );
+        peer.pending_bucket_count = u64::try_from(peer.buckets.len()).expect("bounded fixture");
+    }
+    wait_for_namespace_commit_tails(store.as_ref()).await;
+    let (observer, resolved_scope) = tokio::sync::oneshot::channel();
+    let (release, gate) = tokio::sync::oneshot::channel();
+    let writer_store = Arc::clone(&store);
+    let writer_peers = Arc::clone(&peers);
+    let hot_incarnation = incarnations["hot-bucket"];
+    let writer = tokio::spawn(async move {
+        let scope = resolved_scope.await.expect("real cycle resolved its scope");
+        put_fixture_object(&writer_store, "hot-bucket", "during-walk").await;
+        {
+            let mut locked = writer_peers.lock().expect("peer fixture lock");
+            let peer = locked.get_mut("peer-a:9000").expect("remote producer exists");
+            peer.generation += 1;
+            peer.buckets.insert(
+                "hot-bucket".to_string(),
+                crate::storage_api::EcstoreScannerPeerDirtyUsageBucket {
+                    bucket_incarnation: hot_incarnation,
+                    generation: peer.generation,
+                },
+            );
+            peer.pending_bucket_count = u64::try_from(peer.buckets.len()).expect("bounded fixture");
+        }
+        wait_for_namespace_commit_tails(writer_store.as_ref()).await;
+        release.send(()).expect("cycle awaits its actual post-walk proof");
+        scope
+    });
+    let (result, published) = run_cycle(
+        &store,
+        2,
+        (!empty_ack).then(|| baseline_bytes(&baseline)),
+        false,
+        CycleHooks {
+            resolved_scope_observer: Some(observer),
+            post_walk_gate: Some(gate),
+            peer_observations: Some(observations.clone()),
+        },
+    )
+    .await;
+    let scope = writer.await.expect("writer completed");
+    if empty_ack {
+        assert!(scope.is_default(), "this case exercises the full-scope generation fallback");
+    } else {
+        assert_eq!(
+            scope.selected_buckets_for_tests(),
+            Some(&HashSet::from(["cold-bucket".to_string(), "idle-bucket".to_string()]))
+        );
+    }
+    assert_eq!(
+        result.status,
+        ScannerCycleStatus::Complete,
+        "one remote hot bucket cannot supersede certified cold buckets"
+    );
+    assert_eq!(published.usage_snapshot_pending_buckets, vec!["hot-bucket".to_string()]);
+    assert_eq!(published.usage_snapshot_converged, Some(false));
+    assert!(!published.usage_snapshot_observational);
+    assert_eq!(published.buckets_usage["cold-bucket"].objects_count, if empty_ack { 1 } else { 2 });
+    assert_eq!(published.buckets_usage["idle-bucket"].objects_count, if empty_ack { 1 } else { 2 });
+    if !empty_ack {
+        assert_eq!(published.buckets_usage["hot-bucket"].objects_count, 1);
+    }
+    assert!(result.has_remote_dirty_usage_pending());
+    let acknowledgements = &result.remote_dirty_usage_acknowledgements;
+    if empty_ack {
+        assert!(acknowledgements.is_empty(), "a verified empty scoped ACK cannot become a generation ACK");
+    } else {
+        assert_eq!(acknowledgements.len(), 1, "the clean peer is proved without a fabricated ACK");
+        let crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries } = &acknowledgements[0].kind else {
+            panic!("certified remote buckets require exact scoped ACKs");
+        };
+        assert_eq!(
+            entries.iter().map(|entry| entry.bucket.as_str()).collect::<BTreeSet<_>>(),
+            BTreeSet::from(["cold-bucket", "idle-bucket"])
+        );
+        {
+            let mut locked = peers.lock().expect("peer fixture lock");
+            let peer = locked
+                .get_mut(&acknowledgements[0].host)
+                .expect("ACK targets the original peer");
+            assert_eq!(&peer.owner_id, owner_id);
+            assert_eq!(peer.instance_id, acknowledgements[0].instance_id);
+            // Model the RPC receiver only after asserting the production sender's
+            // exact scope; this does not claim runtime publication/receiver proof.
+            for entry in entries {
+                let current = peer.buckets.get(&entry.bucket).expect("ACK names an observed tuple");
+                assert_eq!(
+                    (current.bucket_incarnation, current.generation),
+                    (entry.bucket_incarnation, entry.generation)
+                );
+                peer.buckets.remove(&entry.bucket);
+            }
+            peer.pending_bucket_count = u64::try_from(peer.buckets.len()).expect("bounded fixture");
+        }
+    }
+    let _ = result.clear_verified_usage();
+    let (observer, resolved_scope) = tokio::sync::oneshot::channel();
+    let (result, next) = run_cycle(
+        &store,
+        3,
+        Some(baseline_bytes(&published)),
+        false,
+        CycleHooks {
+            resolved_scope_observer: Some(observer),
+            post_walk_gate: None,
+            peer_observations: Some(observations),
+        },
+    )
+    .await;
+    assert_eq!(
+        resolved_scope.await.expect("next cycle scope").selected_buckets_for_tests(),
+        Some(&HashSet::from(["hot-bucket".to_string()]))
+    );
+    assert_eq!(result.status, ScannerCycleStatus::Complete);
+    assert_eq!(next.buckets_usage["hot-bucket"].objects_count, if empty_ack { 3 } else { 2 });
+    assert_eq!(next.usage_snapshot_converged, Some(true));
+    clear_dirty_usage_buckets_for_tests();
+}

crates/scanner/src/scanner_io/tests/scoped_entry_fallback.rs

@@ -139,6 +139,7 @@ async fn run_entry(
                 service_cohort: None,
                 resolved_scope_observer: Some(observer),
                 post_walk_gate: None,
+                peer_observations: None,
             },
         ),
     )

crates/scanner/src/scanner_io/tests/service_cohort.rs

@@ -74,6 +74,7 @@ async fn run_cohort_cycle(
                 service_cohort: Some(cohort),
                 resolved_scope_observer: None,
                 post_walk_gate: None,
+                peer_observations: None,
             },
         ),
     )

crates/scanner/src/storage_api.rs

@@ -116,8 +116,10 @@ pub(crate) use rustfs_ecstore::api::layout::{
     EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints,
     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::notification::{
+    NotificationSys as EcstoreNotificationSys, ScannerPublicationLeaseGrant as EcstoreScannerPublicationLeaseGrant,
+};
 pub(crate) use rustfs_ecstore::api::object::{
     SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitState,
 };
@@ -126,8 +128,6 @@ pub(crate) use rustfs_ecstore::api::rebalance::{
     RebalStatus as EcstoreRebalStatus, RebalanceInfo as EcstoreRebalanceInfo, RebalanceMeta as EcstoreRebalanceMeta,
     RebalanceStats as EcstoreRebalanceStats,
 };
-#[cfg(test)]
-pub(crate) use rustfs_ecstore::api::rpc::ScannerPeerDirtyUsageBucket as EcstoreScannerPeerDirtyUsageBucket;
 pub(crate) use rustfs_ecstore::api::rpc::{
     ScannerBucketListing as EcstoreScannerBucketListing,
     ScannerDirtyUsageAcknowledgement as EcstoreScannerDirtyUsageAcknowledgement,
@@ -135,6 +135,10 @@ pub(crate) use rustfs_ecstore::api::rpc::{
     ScannerScopedDirtyUsageAckEntry as EcstoreScannerScopedDirtyUsageAckEntry,
 };
 #[cfg(test)]
+pub(crate) use rustfs_ecstore::api::rpc::{
+    ScannerPeerDirtyUsageBucket as EcstoreScannerPeerDirtyUsageBucket, ScannerPublicationLease as EcstoreScannerPublicationLease,
+};
+#[cfg(test)]
 pub(crate) use rustfs_ecstore::api::runtime::InstanceContext as EcstoreInstanceContext;
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::runtime::set_object_store_resolver as ecstore_set_object_store_resolver;
@@ -380,6 +384,11 @@ 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,
     };
+    #[cfg(test)]
+    pub(crate) use super::{
+        EcstoreScannerPeerDirtyUsageBucket, EcstoreScannerPeerDirtyUsageSnapshot, EcstoreScannerPublicationLease,
+        EcstoreScannerPublicationLeaseGrant,
+    };
 }
 
 pub(crate) mod scanner_io {

docs/operations/audit-denial-entries.md

@@ -0,0 +1,39 @@
+# Audit Entries for Authorization Denials
+
+**Use this when:** you consume the audit stream and need to know which authorization denials appear in it, how many, and how to tune the limit.
+**Source of truth:** `rustfs/src/storage/denial_audit.rs` (`DenialAuditor`, `DenialRateLimiter`), `rustfs/src/storage/access.rs` (`DenialContext`), `crates/config/src/constants/env.rs` (`ENV_AUDIT_DENIAL_RATE`, `ENV_AUDIT_DENIAL_ANONYMOUS_RATE`).
+
+With audit enabled (`RUSTFS_AUDIT_ENABLE=true`), a request refused by the authorization layer produces an audit entry in addition to the `s3_authorization_denied` log event. The S3 response is unchanged: a bare `Access Denied`.
+
+## Entry shape
+
+| Field | Value |
+| --- | --- |
+| `event` | `audit:S3:Operation` |
+| `api.name` | the refused action, for example `s3:PutObject` |
+| `api.bucket`, `api.object` | the request's bucket and object (the object is omitted for a bulk entry) |
+| `api.status`, `api.statusCode` | `failure`, `403` |
+| `accessKey` | the caller's access key, absent for an anonymous caller |
+| `error` | `Access Denied` |
+| `tags.denialReason` | the internal reason, for example `iam_explicit_deny`, `bucket_policy_explicit_deny`, `no_policy_allows_action` |
+| `tags.deniedKeys` | bulk entries only: how many keys of a DeleteObjects request were refused for the action in `api.name` |
+| `tags.suppressedDenials` | present when denials were dropped by the rate limit since the previous entry |
+
+## What is not audited
+
+- Probe denials that callers treat as an expected filter outcome (the per-bucket ListBuckets fallback and per-object metadata permission collection).
+- A DeleteObjects request audits its refused keys as one entry per denied action (for example `s3:DeleteObject`, `s3:GetObject`, `s3:BypassGovernanceRetention`) with `tags.deniedKeys`, not one entry per key.
+- Denials that are later overridden, such as an implicit IAM deny followed by a bucket policy allow.
+
+## Rate limit
+
+A caller can cause denials at will, so the number of entries is bounded by a process-wide token bucket per caller class. The burst equals the per-second rate.
+
+| Variable | Default | Meaning |
+| --- | --- | --- |
+| `RUSTFS_AUDIT_DENIAL_RATE` | `50` | Entries per second for authenticated callers. `0` turns denial entries off for them. |
+| `RUSTFS_AUDIT_DENIAL_ANONYMOUS_RATE` | `5` | Entries per second for anonymous callers. `0` turns denial entries off for them. |
+
+Denials dropped by the limit are counted per caller class, and the count is reported in `tags.suppressedDenials` on the next entry of the same class that is let through. The limit is per process, not per node group.
+
+Trade-off: all authenticated callers share one bucket. A valid, low-privilege account that causes more denials per second than `RUSTFS_AUDIT_DENIAL_RATE` can use up the budget and crowd out other principals' denial entries during that time. Its own drops are still counted in `tags.suppressedDenials`, so the loss is visible, but the other principals' denials in that window are not recorded. A per-principal sub-limit would remove this and is a follow-up, not part of this change.

rustfs/Cargo.toml

@@ -68,7 +68,7 @@ license = []
 io-scheduler-debug = []  # Enable debug information in I/O scheduler
 tracing-chunk-debug = []  # Enable per-chunk tracing in data plane (high noise, for debugging only)
 full = ["metrics-gpu", "ftps", "swift", "webdav", "sftp", "pyroscope", "gcs"]
-e2e-test-hooks = ["rustfs-ecstore/e2e-test-hooks"]
+e2e-test-hooks = ["rustfs-ecstore/e2e-test-hooks", "rustfs-scanner/e2e-test-hooks"]
 # Shortens Connect credentials only in debug E2E builds.
 connect-e2e-short-credentials = []
 # Builds the dedicated rustfs-cli-e2e target with a build-time public enrollment root.

rustfs/src/admin/handlers/object_zip_download.rs

@@ -210,6 +210,7 @@ impl From<ObjectZipDownloadReqInfoSnapshot> for ReqInfo {
             region: current_region(),
             request_context: None,
             suppress_denial_log: false,
+            suppress_denial_audit: false,
         }
     }
 }
@@ -1430,6 +1431,7 @@ mod tests {
             region: current_region(),
             request_context: None,
             suppress_denial_log: false,
+            suppress_denial_audit: false,
         }
     }
 

rustfs/src/app/object/delete.rs

@@ -568,6 +568,8 @@ impl DefaultObjectUsecase {
         // warn and demote the rest to debug, so a denied 1000-key DeleteObjects
         // cannot flood the log.
         let mut bulk_denial_logged = false;
+        // Per-key denials are audited once per denied action, in aggregate, after the loop.
+        let mut bulk_denials = crate::storage::access::BulkDenialAudit::start(&mut req)?;
         for (idx, obj_id) in delete.objects.iter().enumerate() {
             if let Err(err) = validate_object_key(&obj_id.key, "DeleteObjects") {
                 delete_results[idx].error = Some(s3s::dto::Error {
@@ -606,15 +608,18 @@ impl DefaultObjectUsecase {
                 req_info.version_id = version_id.clone();
             }
 
-            let mut auth_res = authorize_request(&mut req, delete_object_authorize_action(version_id.as_deref())).await;
+            let mut denied_action = delete_object_authorize_action(version_id.as_deref());
+            let mut auth_res = authorize_request(&mut req, denied_action).await;
             if auth_res.is_ok()
                 && delete_condition
                     .as_ref()
                     .is_some_and(DeleteObjectCondition::requires_read_permission)
             {
-                auth_res = authorize_request(&mut req, Action::S3Action(S3Action::GetObjectAction)).await;
+                denied_action = Action::S3Action(S3Action::GetObjectAction);
+                auth_res = authorize_request(&mut req, denied_action).await;
             }
             if auth_res.is_err() {
+                bulk_denials.denied(denied_action);
                 if !bulk_denial_logged {
                     bulk_denial_logged = true;
                     req_info_mut(&mut req)?.suppress_denial_log = true;
@@ -631,6 +636,7 @@ impl DefaultObjectUsecase {
             if bypass_governance {
                 let auth_res = authorize_request(&mut req, Action::S3Action(S3Action::BypassGovernanceRetentionAction)).await;
                 if auth_res.is_err() {
+                    bulk_denials.denied(Action::S3Action(S3Action::BypassGovernanceRetentionAction));
                     if !bulk_denial_logged {
                         bulk_denial_logged = true;
                         req_info_mut(&mut req)?.suppress_denial_log = true;
@@ -669,6 +675,8 @@ impl DefaultObjectUsecase {
             authorized_deletes.push(AuthorizedDelete { idx, object });
         }
 
+        bulk_denials.finish(&mut req)?;
+
         if authorized_deletes.is_empty() {
             let output = DeleteObjectsOutput {
                 deleted: (!delete.quiet.unwrap_or_default()).then(Vec::new),
@@ -3686,3 +3694,29 @@ mod tests {
         ));
     }
 }
+
+/// DeleteObjects refusals are audited once per denied action by `BulkDenialAudit`. The handler needs a
+/// live store to run, so this source-level guard fails if the wiring is removed; it does not execute it.
+#[cfg(test)]
+mod bulk_denial_audit_wiring_tests {
+    #[test]
+    fn delete_objects_audits_refused_keys_in_aggregate() {
+        // Only the code above this module counts; the guard's own text must not satisfy it.
+        let source = include_str!("delete.rs")
+            .split("\n#[cfg(test)]\nmod bulk_denial_audit_wiring_tests")
+            .next()
+            .unwrap();
+        let started = source
+            .find("BulkDenialAudit::start(&mut req)")
+            .expect("per-key denial audit is not suppressed");
+        let finished = source
+            .find("bulk_denials.finish(&mut req)")
+            .expect("the aggregate entry is not emitted");
+        assert!(started < finished, "the aggregate must be emitted after the per-key checks");
+        assert_eq!(
+            source.matches("bulk_denials.denied(").count(),
+            2,
+            "both authorization checks must count their denials"
+        );
+    }
+}

rustfs/src/app/select_object.rs

@@ -600,9 +600,10 @@ fn validate_select_request(headers: &http::HeaderMap, input: &mut SelectObjectCo
     if input.request.expression.len() > MAX_SELECT_EXPRESSION_BYTES {
         return Err(S3Error::new(S3ErrorCode::ExpressionTooLong));
     }
-    if input.request.expression_type.as_str() != ExpressionType::SQL {
+    let Some(expression_type) = canonical_enum(input.request.expression_type.as_str(), &[ExpressionType::SQL]) else {
         return Err(S3Error::new(S3ErrorCode::InvalidExpressionType));
-    }
+    };
+    input.request.expression_type = ExpressionType::from_static(expression_type);
 
     normalize_input_serialization(&mut input.request.input_serialization)?;
     let compressed_input = is_compressed_input(&input.request.input_serialization);
@@ -649,6 +650,15 @@ fn is_compressed_input(input: &InputSerialization) -> bool {
         .is_some_and(|compression| compression.as_str() != CompressionType::NONE)
 }
 
+/// Matches an S3 enum value in any ASCII case, as MinIO does, and returns the
+/// canonical spelling that downstream code compares against.
+fn canonical_enum(value: &str, allowed: &[&'static str]) -> Option<&'static str> {
+    allowed
+        .iter()
+        .copied()
+        .find(|candidate| candidate.eq_ignore_ascii_case(value))
+}
+
 fn normalize_input_serialization(input: &mut InputSerialization) -> S3Result<()> {
     let format_count =
         usize::from(input.csv.is_some()) + usize::from(input.json.is_some()) + usize::from(input.parquet.is_some());
@@ -659,19 +669,20 @@ fn normalize_input_serialization(input: &mut InputSerialization) -> S3Result<()>
         return Err(S3Error::new(S3ErrorCode::ObjectSerializationConflict));
     }
 
-    if let Some(compression) = input.compression_type.as_ref() {
-        match compression.as_str() {
-            CompressionType::NONE => {}
-            CompressionType::GZIP | CompressionType::BZIP2 => {
-                if input.parquet.is_some() {
-                    return Err(S3Error::with_message(
-                        S3ErrorCode::InvalidRequestParameter,
-                        INVALID_REQUEST_PARAMETER_MESSAGE,
-                    ));
-                }
-            }
-            _ => return Err(map_select_error_to_s3(&SelectError::InvalidCompressionFormat)),
+    if let Some(compression) = input.compression_type.as_mut() {
+        let Some(canonical) = canonical_enum(
+            compression.as_str(),
+            &[CompressionType::NONE, CompressionType::GZIP, CompressionType::BZIP2],
+        ) else {
+            return Err(map_select_error_to_s3(&SelectError::InvalidCompressionFormat));
+        };
+        if canonical != CompressionType::NONE && input.parquet.is_some() {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequestParameter,
+                INVALID_REQUEST_PARAMETER_MESSAGE,
+            ));
         }
+        *compression = CompressionType::from_static(canonical);
     }
     input
         .compression_type
@@ -681,15 +692,16 @@ fn normalize_input_serialization(input: &mut InputSerialization) -> S3Result<()>
         let file_header_info = csv
             .file_header_info
             .get_or_insert_with(|| FileHeaderInfo::from_static(FileHeaderInfo::NONE));
-        if !matches!(
+        let Some(canonical) = canonical_enum(
             file_header_info.as_str(),
-            FileHeaderInfo::NONE | FileHeaderInfo::USE | FileHeaderInfo::IGNORE
-        ) {
+            &[FileHeaderInfo::NONE, FileHeaderInfo::USE, FileHeaderInfo::IGNORE],
+        ) else {
             return Err(S3Error::with_message(
                 S3ErrorCode::InvalidFileHeaderInfo,
                 "The FileHeaderInfo value is not valid. Only NONE, USE, and IGNORE are supported.",
             ));
-        }
+        };
+        *file_header_info = FileHeaderInfo::from_static(canonical);
         validate_single_byte(csv.comments.as_deref(), S3ErrorCode::InvalidRequestParameter)?;
         validate_single_character(csv.quote_character.as_deref())?;
         validate_single_character(csv.quote_escape_character.as_deref())?;
@@ -699,9 +711,10 @@ fn normalize_input_serialization(input: &mut InputSerialization) -> S3Result<()>
 
     if let Some(json) = input.json.as_mut() {
         let json_type = json.type_.get_or_insert_with(|| JSONType::from_static(JSONType::LINES));
-        if !matches!(json_type.as_str(), JSONType::DOCUMENT | JSONType::LINES) {
+        let Some(canonical) = canonical_enum(json_type.as_str(), &[JSONType::DOCUMENT, JSONType::LINES]) else {
             return Err(S3Error::new(S3ErrorCode::InvalidJsonType));
-        }
+        };
+        *json_type = JSONType::from_static(canonical);
     }
 
     Ok(())
@@ -716,15 +729,16 @@ fn normalize_output_serialization(output: &mut OutputSerialization) -> S3Result<
         return Err(S3Error::new(S3ErrorCode::ObjectSerializationConflict));
     }
 
-    if let Some(csv) = output.csv.as_ref() {
+    if let Some(csv) = output.csv.as_mut() {
         validate_single_byte(csv.field_delimiter.as_deref(), S3ErrorCode::InvalidRequestParameter)?;
         validate_single_byte(csv.quote_character.as_deref(), S3ErrorCode::InvalidRequestParameter)?;
         validate_single_byte(csv.quote_escape_character.as_deref(), S3ErrorCode::InvalidRequestParameter)?;
         validate_output_record_delimiter(csv.record_delimiter.as_deref())?;
-        if let Some(quote_fields) = csv.quote_fields.as_ref()
-            && !matches!(quote_fields.as_str(), QuoteFields::ALWAYS | QuoteFields::ASNEEDED)
-        {
-            return Err(S3Error::new(S3ErrorCode::InvalidQuoteFields));
+        if let Some(quote_fields) = csv.quote_fields.as_mut() {
+            let Some(canonical) = canonical_enum(quote_fields.as_str(), &[QuoteFields::ALWAYS, QuoteFields::ASNEEDED]) else {
+                return Err(S3Error::new(S3ErrorCode::InvalidQuoteFields));
+            };
+            *quote_fields = QuoteFields::from_static(canonical);
         }
         return Ok(SelectOutputFormat::Csv(csv.clone()));
     }
@@ -3639,6 +3653,132 @@ mod tests {
         );
     }
 
+    fn json_input(json_type: &str) -> SelectObjectContentInput {
+        let mut input = base_input();
+        input.request.input_serialization.csv = None;
+        input.request.input_serialization.json = Some(JSONInput {
+            type_: Some(JSONType::from(json_type.to_string())),
+        });
+        input
+    }
+
+    #[test]
+    fn validate_matches_enum_values_case_insensitively_and_canonicalizes_them() {
+        for (value, expected) in [
+            ("gzip", CompressionType::GZIP),
+            ("bZip2", CompressionType::BZIP2),
+            ("none", CompressionType::NONE),
+        ] {
+            let mut input = base_input();
+            input.request.input_serialization.compression_type = Some(CompressionType::from(value.to_string()));
+            validate_select_request(&HeaderMap::new(), &mut input).expect("CompressionType should match any case");
+            assert_eq!(
+                input
+                    .request
+                    .input_serialization
+                    .compression_type
+                    .as_ref()
+                    .map(|v| v.as_str()),
+                Some(expected)
+            );
+        }
+
+        for (value, expected) in [("lines", JSONType::LINES), ("Document", JSONType::DOCUMENT)] {
+            let mut input = json_input(value);
+            validate_select_request(&HeaderMap::new(), &mut input).expect("JSON Type should match any case");
+            let json_type = input
+                .request
+                .input_serialization
+                .json
+                .as_ref()
+                .and_then(|json| json.type_.as_ref());
+            assert_eq!(json_type.map(|v| v.as_str()), Some(expected));
+        }
+        let mut input = json_input("document");
+        validate_select_request(&HeaderMap::new(), &mut input).expect("lowercase DOCUMENT should be accepted");
+        assert!(is_json_document(input.request.input_serialization.json.as_ref().expect("JSON input")));
+
+        for (value, expected) in [
+            ("use", FileHeaderInfo::USE),
+            ("Ignore", FileHeaderInfo::IGNORE),
+            ("none", FileHeaderInfo::NONE),
+        ] {
+            let mut input = base_input();
+            input
+                .request
+                .input_serialization
+                .csv
+                .as_mut()
+                .expect("base input should use CSV")
+                .file_header_info = Some(FileHeaderInfo::from(value.to_string()));
+            validate_select_request(&HeaderMap::new(), &mut input).expect("FileHeaderInfo should match any case");
+            let csv = input.request.input_serialization.csv.as_ref().expect("CSV input");
+            assert_eq!(csv.file_header_info.as_ref().map(|v| v.as_str()), Some(expected));
+        }
+
+        for (value, expected) in [("always", QuoteFields::ALWAYS), ("AsNeeded", QuoteFields::ASNEEDED)] {
+            let mut input = base_input();
+            input
+                .request
+                .output_serialization
+                .csv
+                .as_mut()
+                .expect("base output should use CSV")
+                .quote_fields = Some(QuoteFields::from(value.to_string()));
+            let validation = validate_select_request(&HeaderMap::new(), &mut input).expect("QuoteFields should match any case");
+            let SelectOutputFormat::Csv(csv) = validation.output_format else {
+                panic!("CSV output expected");
+            };
+            assert_eq!(csv.quote_fields.as_ref().map(|v| v.as_str()), Some(expected));
+        }
+
+        let mut input = base_input();
+        input.request.expression_type = ExpressionType::from("sql".to_string());
+        validate_select_request(&HeaderMap::new(), &mut input).expect("ExpressionType should match any case");
+        assert_eq!(input.request.expression_type.as_str(), ExpressionType::SQL);
+    }
+
+    #[test]
+    fn validate_still_rejects_unknown_enum_values_in_any_case() {
+        for value in ["gzipx", "snappy", " GZIP"] {
+            let mut input = base_input();
+            input.request.input_serialization.compression_type = Some(CompressionType::from(value.to_string()));
+            let error = validate_select_request(&HeaderMap::new(), &mut input).expect_err("unknown compression must fail");
+            assert_eq!(error.code(), &S3ErrorCode::InvalidCompressionFormat, "{value}");
+        }
+        for value in ["ndjson", "lines "] {
+            let error =
+                validate_select_request(&HeaderMap::new(), &mut json_input(value)).expect_err("unknown JSON type must fail");
+            assert_eq!(error.code(), &S3ErrorCode::InvalidJsonType, "{value}");
+        }
+        let mut input = base_input();
+        input
+            .request
+            .input_serialization
+            .csv
+            .as_mut()
+            .expect("base input should use CSV")
+            .file_header_info = Some(FileHeaderInfo::from("used".to_string()));
+        let error = validate_select_request(&HeaderMap::new(), &mut input).expect_err("unknown header mode must fail");
+        assert_eq!(error.code(), &S3ErrorCode::InvalidFileHeaderInfo);
+
+        let mut input = base_input();
+        input
+            .request
+            .output_serialization
+            .csv
+            .as_mut()
+            .expect("base output should use CSV")
+            .quote_fields = Some(QuoteFields::from("never".to_string()));
+        let error = validate_select_request(&HeaderMap::new(), &mut input).expect_err("unknown QuoteFields must fail");
+        assert_eq!(error.code(), &S3ErrorCode::InvalidQuoteFields);
+
+        let mut input = base_input();
+        input.request.expression_type = ExpressionType::from("sqlite".to_string());
+        let error = validate_select_request(&HeaderMap::new(), &mut input).expect_err("unknown ExpressionType must fail");
+        assert_eq!(error.code(), &S3ErrorCode::InvalidExpressionType);
+    }
+
     #[test]
     fn validate_accepts_two_byte_csv_input_record_delimiter() {
         let mut input = base_input();

rustfs/src/protocols/client.rs

@@ -130,6 +130,7 @@ fn session_list_buckets_request(
         region: None,
         request_context: Some(RequestContext::fallback()),
         suppress_denial_log: false,
+        suppress_denial_audit: false,
     });
 
     S3Request {
@@ -324,6 +325,7 @@ impl ProtocolStorageClient {
             region: None,
             request_context: Some(RequestContext::fallback()),
             suppress_denial_log: false,
+            suppress_denial_audit: false,
         });
 
         let req = S3Request {

rustfs/src/storage/access.rs

@@ -25,6 +25,8 @@ use crate::auth::{
 use crate::error::{ApiError, normalize_read_quorum_error};
 use crate::license::license_check;
 use crate::shared_types::RemoteAddr;
+use crate::storage::denial_audit::DenialAuditor;
+use crate::storage::helper::DenialScope;
 use crate::storage::request_context::RequestContext;
 use crate::storage::storage_api::contract::bucket::BUCKET_LIFECYCLE_LOCK_OBJECT;
 use crate::storage::storage_api::contract::namespace::NamespaceLocking as _;
@@ -91,6 +93,9 @@ pub(crate) struct ReqInfo {
     /// collection) so `authorize_request` logs those routine denials at `debug`
     /// instead of `warn` (issue #5740).
     pub suppress_denial_log: bool,
+    /// Set by callers that audit their denials in aggregate (DeleteObjects), so each per-key
+    /// denial is not audited on its own. The denial is still logged.
+    pub suppress_denial_audit: bool,
 }
 
 pub(crate) fn replication_request_authorized<T>(req: &S3Request<T>) -> bool {
@@ -1191,6 +1196,8 @@ pub(crate) fn log_list_buckets_iam_implicit_deny<T>(req: &S3Request<T>) -> S3Res
     let req_info = req_info_ref(req)?;
     let denial = DenialContext {
         quiet: true,
+        suppress_audit: true,
+        auditor: DenialAuditor::global(),
         bucket: req_info.bucket.as_deref().unwrap_or_default(),
         object: req_info.object.as_deref().unwrap_or_default(),
         version_id: req_info.version_id.as_deref(),
@@ -1306,6 +1313,8 @@ pub(crate) fn owner_can_bypass_policy_deny(is_owner: bool, action: &Action) -> b
 /// instead of `warn`.
 struct DenialContext<'a> {
     quiet: bool,
+    suppress_audit: bool,
+    auditor: &'a DenialAuditor,
     bucket: &'a str,
     object: &'a str,
     version_id: Option<&'a str>,
@@ -1342,14 +1351,68 @@ impl DenialContext<'_> {
         );
     }
 
-    /// Build the generic AccessDenied response after recording the structured
-    /// denial event.
-    fn deny(&self, reason: &'static str, action: Action) -> S3Error {
+    /// Log the denial and, unless it is a quiet probe denial or audited in aggregate by the
+    /// caller, audit it.
+    fn record<T>(&self, req: &S3Request<T>, reason: &'static str, action: Action) {
         self.log(reason, action);
+        if !self.quiet && !self.suppress_audit {
+            self.auditor.audit(req, &action, reason, DenialScope::Request);
+        }
+    }
+
+    /// Build the generic AccessDenied response after recording the structured
+    /// denial event and, unless this is a quiet probe denial, an audit entry.
+    fn deny<T>(&self, req: &S3Request<T>, reason: &'static str, action: Action) -> S3Error {
+        self.record(req, reason, action);
         s3_error!(AccessDenied, "Access Denied")
     }
 }
 
+/// Audits a denial at an exit that has no [`DenialContext`] of its own. It only audits: those exits
+/// keep their existing logging. The request's own probe flags are honored.
+fn audit_request_denial<T>(req: &S3Request<T>, auditor: &DenialAuditor, reason: &'static str, action: Action) {
+    let info = req.extensions.get::<ReqInfo>();
+    if info.is_some_and(|info| info.suppress_denial_log || info.suppress_denial_audit) {
+        return;
+    }
+    auditor.audit(req, &action, reason, DenialScope::Request);
+}
+
+/// Audits the refused keys of a bulk request (DeleteObjects) in aggregate instead of once per key.
+///
+/// `start` marks the request so per-key denials are logged but not audited; `finish` clears the
+/// mark and emits one entry per denied action with the number of keys refused for it.
+pub(crate) struct BulkDenialAudit {
+    denied: Vec<(Action, u64)>,
+}
+
+impl BulkDenialAudit {
+    pub(crate) fn start<T>(req: &mut S3Request<T>) -> S3Result<Self> {
+        ext_req_info_mut(&mut req.extensions)?.suppress_denial_audit = true;
+        Ok(Self { denied: Vec::new() })
+    }
+
+    /// Counts one refused key for `action`.
+    pub(crate) fn denied(&mut self, action: Action) {
+        match self.denied.iter_mut().find(|(denied_action, _)| *denied_action == action) {
+            Some((_, count)) => *count += 1,
+            None => self.denied.push((action, 1)),
+        }
+    }
+
+    pub(crate) fn finish<T>(self, req: &mut S3Request<T>) -> S3Result<()> {
+        self.finish_with(req, DenialAuditor::global())
+    }
+
+    fn finish_with<T>(self, req: &mut S3Request<T>, auditor: &DenialAuditor) -> S3Result<()> {
+        ext_req_info_mut(&mut req.extensions)?.suppress_denial_audit = false;
+        for (action, keys) in self.denied {
+            auditor.audit(req, &action, "bulk_request_keys_denied", DenialScope::Keys(keys));
+        }
+        Ok(())
+    }
+}
+
 /// Authorizes the request based on the action and credentials.
 pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3Result<()> {
     authorize_request_with_location_constraint(req, action, None).await
@@ -1369,9 +1432,12 @@ async fn authorize_request_with_location_constraint<T>(
     let object = req_info.object.clone().unwrap_or_default();
     let version_id = req_info.version_id.clone();
     let quiet_denial = req_info.suppress_denial_log;
+    let suppress_denial_audit = req_info.suppress_denial_audit;
     let store = request_object_store(req)?;
     let denial = DenialContext {
         quiet: quiet_denial,
+        suppress_audit: suppress_denial_audit,
+        auditor: DenialAuditor::global(),
         bucket: bucket.as_str(),
         object: object.as_str(),
         version_id: version_id.as_deref(),
@@ -1472,7 +1538,7 @@ async fn authorize_request_with_location_constraint<T>(
             .await
             .map_err(ApiError::from)?
         {
-            return Err(denial.deny("bucket_policy_explicit_deny", action));
+            return Err(denial.deny(req, "bucket_policy_explicit_deny", action));
         }
 
         let iam_allowed = {
@@ -1514,7 +1580,7 @@ async fn authorize_request_with_location_constraint<T>(
                     .await
                     .map_err(iam_authorization_error)?
                 {
-                    return Err(denial.deny("iam_explicit_deny", action));
+                    return Err(denial.deny(req, "iam_explicit_deny", action));
                 }
             }
             allowed
@@ -1628,7 +1694,7 @@ async fn authorize_request_with_location_constraint<T>(
             .await
             .map_err(ApiError::from)?
         {
-            return Err(denial.deny("bucket_policy_explicit_deny", action));
+            return Err(denial.deny(req, "bucket_policy_explicit_deny", action));
         }
 
         if action != Action::S3Action(S3Action::ListAllMyBucketsAction) {
@@ -1670,11 +1736,11 @@ async fn authorize_request_with_location_constraint<T>(
                 deny_anonymous_table_data_plane_if_needed(req, action, bucket.as_str(), object.as_str()).await?;
                 // RestrictPublicBuckets: when true, deny public access even if bucket policy allows it.
                 match store.restricts_public_bucket_access(bucket_name).await {
-                    Ok(true) => return Err(denial.deny("restrict_public_buckets", action)),
+                    Ok(true) => return Err(denial.deny(req, "restrict_public_buckets", action)),
                     Ok(false) => {}
                     Err(StorageError::ConfigNotFound) => {}
                     Err(_) => {
-                        return Err(denial.deny("public_access_block_unavailable", action));
+                        return Err(denial.deny(req, "public_access_block_unavailable", action));
                     }
                 }
                 return Ok(());
@@ -1682,7 +1748,7 @@ async fn authorize_request_with_location_constraint<T>(
         }
     }
 
-    Err(denial.deny("no_policy_allows_action", action))
+    Err(denial.deny(req, "no_policy_allows_action", action))
 }
 
 // Multipart parts are staged and become visible only after CompleteMultipartUpload,
@@ -2178,6 +2244,7 @@ async fn authorize_table_data_plane_if_needed<T>(
         ?admin_action,
         "table data-plane access denied by table resource policy"
     );
+    audit_request_denial(req, DenialAuditor::global(), "table_resource_policy_denied", action);
     Err(s3_error!(AccessDenied, "Access Denied"))
 }
 
@@ -2206,6 +2273,7 @@ async fn deny_anonymous_table_data_plane_if_needed<T>(
         .await?
         .is_some()
     {
+        audit_request_denial(req, DenialAuditor::global(), "anonymous_table_data_plane_access", action);
         return Err(s3_error!(AccessDenied, "Access Denied"));
     }
     Ok(())
@@ -2642,6 +2710,12 @@ impl S3Access for FS {
             authorize_request(req, Action::S3Action(S3Action::ForceDeleteObjectAction)).await?;
         }
         if !recursive_force_delete_has_authenticated_caller(&req.headers, authenticated, replica_request) {
+            audit_request_denial(
+                req,
+                DenialAuditor::global(),
+                "recursive_force_delete_requires_authenticated_caller",
+                Action::S3Action(S3Action::ForceDeleteObjectAction),
+            );
             return Err(s3_error!(AccessDenied, "Recursive force-delete requires an authenticated caller"));
         }
 
@@ -3605,12 +3679,14 @@ mod tests {
         table_data_plane_content_mutation, table_data_plane_resource_for_request, table_publication_guard_error,
         validate_post_object_success_controls, versioned_read_action, versioned_tagging_action,
     };
+    use super::{BulkDenialAudit, audit_request_denial};
     use super::{
         remaining_retention_days, requested_bucket_location_constraint, set_requested_object_lock_retain_until,
         set_requested_object_lock_retention,
     };
     use crate::auth::OBJECT_LOCK_REMAINING_RETENTION_DAYS_CONDITION;
     use crate::error::ApiError;
+    use crate::storage::denial_audit::{DenialAuditor, DenialRateLimiter};
     use crate::storage::storage_api::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
     use crate::storage::storage_api::contract::multipart::MultipartOperations as _;
     use crate::storage::storage_api::contract::object::{ObjectIO as _, ObjectOperations as _};
@@ -5107,21 +5183,268 @@ mod tests {
         ));
     }
 
+    type CapturedAudit = Arc<std::sync::Mutex<Vec<rustfs_audit::AuditEntry>>>;
+
+    /// An auditor wired like the global one except that entries land in the returned vector.
+    fn test_auditor(enabled: bool, authenticated_rate: u64, anonymous_rate: u64) -> (DenialAuditor, CapturedAudit) {
+        let captured: CapturedAudit = Arc::default();
+        let sink = Arc::clone(&captured);
+        // A frozen clock: no wall-clock refill can change what the limiter admits.
+        let t0 = std::time::Instant::now();
+        let limiter = DenialRateLimiter::new(authenticated_rate, anonymous_rate, t0);
+        let auditor =
+            DenialAuditor::new(limiter, move || enabled, move |_, entry| sink.lock().unwrap().push(entry)).with_clock(move || t0);
+        (auditor, captured)
+    }
+
+    fn denial_request(request_id: &str, authenticated: bool) -> S3Request<()> {
+        let mut req = build_request((), Method::GET);
+        req.uri = Uri::from_static("/audit-bucket/audit-object");
+        req.extensions.insert(crate::storage::request_context::RequestContext {
+            request_id: request_id.to_string(),
+            x_amz_request_id: request_id.to_string(),
+            trace_id: None,
+            span_id: None,
+            start_time: std::time::Instant::now(),
+        });
+        req.extensions.insert(ReqInfo {
+            cred: authenticated.then(|| rustfs_credentials::Credentials {
+                access_key: "denied-user".to_string(),
+                ..Default::default()
+            }),
+            bucket: Some("audit-bucket".to_string()),
+            object: Some("audit-object".to_string()),
+            ..Default::default()
+        });
+        req
+    }
+
+    fn denial_context(auditor: &DenialAuditor, quiet: bool) -> DenialContext<'_> {
+        DenialContext {
+            quiet,
+            suppress_audit: false,
+            auditor,
+            bucket: "audit-bucket",
+            object: "audit-object",
+            version_id: Some("version"),
+            account: Some("denied-user"),
+            is_owner: false,
+        }
+    }
+
+    const GET_OBJECT: Action = Action::S3Action(S3Action::GetObjectAction);
+
     /// Issue #5740: the denial helper must keep the client-facing error identical
     /// to the historical bare response — the added diagnostics are log-only.
     #[test]
     fn test_access_denied_helper_keeps_generic_wire_response() {
-        let denial = DenialContext {
-            quiet: false,
-            bucket: "bucket",
-            object: "object",
-            version_id: Some("version"),
-            account: Some("account"),
-            is_owner: false,
-        };
-        let err = denial.deny("bucket_policy_explicit_deny", Action::S3Action(S3Action::GetObjectAction));
+        let (auditor, captured) = test_auditor(true, 10, 10);
+        let req = denial_request("denial-wire-response", true);
+        let err = denial_context(&auditor, false).deny(&req, "bucket_policy_explicit_deny", GET_OBJECT);
         assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
         assert_eq!(err.message(), Some("Access Denied"));
+        assert_eq!(captured.lock().unwrap().len(), 1);
+    }
+
+    /// A real denial yields exactly one audit entry naming the account, bucket, object,
+    /// denied action and internal reason.
+    #[test]
+    fn denial_is_audited_once_with_account_resource_action_and_reason() {
+        let (auditor, captured) = test_auditor(true, 10, 10);
+        let req = denial_request("denial-audited-once", true);
+        let _ = denial_context(&auditor, false).deny(&req, "iam_explicit_deny", Action::S3Action(S3Action::PutObjectAction));
+
+        let entries = captured.lock().unwrap();
+        assert_eq!(entries.len(), 1);
+        let entry = &entries[0];
+        assert_eq!(entry.request_id.as_deref(), Some("denial-audited-once"));
+        assert_eq!(entry.access_key.as_deref(), Some("denied-user"));
+        assert_eq!(entry.api.name.as_deref(), Some("s3:PutObject"));
+        assert_eq!(entry.api.bucket.as_deref(), Some("audit-bucket"));
+        assert_eq!(entry.api.object.as_deref(), Some("audit-object"));
+        assert_eq!(entry.api.status.as_deref(), Some("failure"));
+        assert_eq!(entry.api.status_code, Some(403));
+        assert_eq!(entry.error.as_deref(), Some("Access Denied"));
+        let tags = entry.tags.as_ref().expect("denial entry carries its reason");
+        assert_eq!(tags.get("denialReason").and_then(|value| value.as_str()), Some("iam_explicit_deny"));
+        assert!(!tags.contains_key("suppressedDenials"));
+    }
+
+    /// Probe denials (per-bucket ListBuckets fallback, per-object metadata collection) are
+    /// expected filter outcomes and must not flood the audit stream.
+    #[test]
+    fn quiet_probe_denial_is_not_audited() {
+        let (auditor, captured) = test_auditor(true, 10, 10);
+        let req = denial_request("denial-quiet-probe", true);
+        let err =
+            denial_context(&auditor, true).deny(&req, "no_policy_allows_action", Action::S3Action(S3Action::ListBucketAction));
+        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
+        assert!(captured.lock().unwrap().is_empty());
+    }
+
+    #[test]
+    fn denial_is_not_audited_when_the_audit_module_is_off() {
+        let (auditor, captured) = test_auditor(false, 10, 10);
+        let req = denial_request("denial-audit-off", true);
+        let _ = denial_context(&auditor, false).deny(&req, "no_policy_allows_action", GET_OBJECT);
+        assert!(captured.lock().unwrap().is_empty());
+    }
+
+    #[test]
+    fn denial_audited_in_aggregate_by_the_caller_is_not_audited_per_key() {
+        let (auditor, captured) = test_auditor(true, 10, 10);
+        let req = denial_request("denial-per-key", true);
+        let mut denial = denial_context(&auditor, false);
+        denial.suppress_audit = true;
+        let _ = denial.deny(&req, "iam_explicit_deny", GET_OBJECT);
+        assert!(captured.lock().unwrap().is_empty());
+    }
+
+    /// A flood of denials emits a bounded number of entries; the rest are only counted.
+    #[test]
+    fn denial_flood_emits_bounded_entries() {
+        let (auditor, captured) = test_auditor(true, 3, 3);
+        let req = denial_request("denial-flood", true);
+        for _ in 0..20 {
+            let _ = denial_context(&auditor, false).deny(&req, "no_policy_allows_action", GET_OBJECT);
+        }
+        assert_eq!(captured.lock().unwrap().len(), 3);
+    }
+
+    /// The entry that follows a suppressed run reports how many denials were dropped.
+    #[test]
+    fn entry_after_a_flood_reports_the_suppressed_count() {
+        use crate::storage::helper::DenialScope;
+        let (auditor, captured) = test_auditor(true, 1, 1);
+        let req = denial_request("denial-suppressed-count", true);
+        let t0 = std::time::Instant::now();
+        for _ in 0..5 {
+            auditor.audit_at(&req, &GET_OBJECT, "no_policy_allows_action", DenialScope::Request, t0);
+        }
+        auditor.audit_at(
+            &req,
+            &GET_OBJECT,
+            "no_policy_allows_action",
+            DenialScope::Request,
+            t0 + std::time::Duration::from_secs(2),
+        );
+        let entries = captured.lock().unwrap();
+        assert_eq!(entries.len(), 2);
+        let tag = |index: usize| {
+            entries[index]
+                .tags
+                .as_ref()
+                .unwrap()
+                .get("suppressedDenials")
+                .and_then(|v| v.as_u64())
+        };
+        assert_eq!(tag(0), None);
+        assert_eq!(tag(1), Some(4));
+    }
+
+    /// Anonymous callers have their own, lower limit, so an unauthenticated scanner cannot
+    /// use up the budget for authenticated callers.
+    #[test]
+    fn anonymous_denials_are_limited_separately_from_authenticated_ones() {
+        let (auditor, captured) = test_auditor(true, 10, 1);
+        let anonymous = denial_request("denial-anonymous", false);
+        let authenticated = denial_request("denial-authenticated", true);
+        for _ in 0..5 {
+            let _ = denial_context(&auditor, false).deny(&anonymous, "no_policy_allows_action", GET_OBJECT);
+        }
+        for _ in 0..5 {
+            let _ = denial_context(&auditor, false).deny(&authenticated, "no_policy_allows_action", GET_OBJECT);
+        }
+        let entries = captured.lock().unwrap();
+        let count = |id: &str| entries.iter().filter(|entry| entry.request_id.as_deref() == Some(id)).count();
+        assert_eq!(count("denial-anonymous"), 1);
+        assert_eq!(count("denial-authenticated"), 5);
+    }
+
+    /// Exits that have no `DenialContext` of their own audit through the request's own state
+    /// and honor the same probe flags.
+    #[test]
+    fn request_denial_exit_is_audited_and_honors_the_probe_flags() {
+        let (auditor, captured) = test_auditor(true, 10, 10);
+        let mut req = denial_request("denial-exit", true);
+        audit_request_denial(&req, &auditor, "table_resource_policy_denied", GET_OBJECT);
+        assert_eq!(captured.lock().unwrap().len(), 1);
+
+        req.extensions.get_mut::<ReqInfo>().unwrap().suppress_denial_log = true;
+        audit_request_denial(&req, &auditor, "table_resource_policy_denied", GET_OBJECT);
+        assert_eq!(captured.lock().unwrap().len(), 1);
+
+        req.extensions.get_mut::<ReqInfo>().unwrap().suppress_denial_log = false;
+        req.extensions.get_mut::<ReqInfo>().unwrap().suppress_denial_audit = true;
+        audit_request_denial(&req, &auditor, "table_resource_policy_denied", GET_OBJECT);
+        assert_eq!(captured.lock().unwrap().len(), 1);
+    }
+
+    /// The table-policy, anonymous table and force-delete exits live deep in the authorization
+    /// flow and need a table catalog to reach. This source-level guard fails if one stops
+    /// auditing; it does not execute them.
+    #[test]
+    fn denial_exits_without_a_context_call_the_audit_function() {
+        // Only the production code counts; the tests' own calls must not satisfy the guard.
+        let source = include_str!("access.rs").split("\nmod tests {").next().unwrap();
+        for reason in [
+            "\"table_resource_policy_denied\"",
+            "\"anonymous_table_data_plane_access\"",
+            "\"recursive_force_delete_requires_authenticated_caller\"",
+        ] {
+            let call = source
+                .match_indices("audit_request_denial(")
+                .any(|(at, _)| source[at..].chars().take(260).collect::<String>().contains(reason));
+            assert!(call, "{reason} exit does not call audit_request_denial");
+        }
+    }
+
+    /// DeleteObjects audits its refused keys as one entry per denied action, carrying the count and
+    /// no single object, and the per-key mark is cleared afterwards.
+    #[test]
+    fn bulk_denial_is_one_entry_per_denied_action_with_the_key_count() {
+        let (auditor, captured) = test_auditor(true, 10, 10);
+        let mut req = denial_request("denial-bulk", true);
+
+        let nothing_denied = BulkDenialAudit::start(&mut req).unwrap();
+        assert!(req.extensions.get::<ReqInfo>().unwrap().suppress_denial_audit);
+        nothing_denied.finish_with(&mut req, &auditor).unwrap();
+        assert!(!req.extensions.get::<ReqInfo>().unwrap().suppress_denial_audit);
+        assert!(captured.lock().unwrap().is_empty());
+
+        let mut bulk = BulkDenialAudit::start(&mut req).unwrap();
+        let delete = Action::S3Action(S3Action::DeleteObjectAction);
+        let bypass = Action::S3Action(S3Action::BypassGovernanceRetentionAction);
+        for action in [delete, delete, bypass, delete] {
+            bulk.denied(action);
+        }
+        bulk.finish_with(&mut req, &auditor).unwrap();
+
+        let entries = captured.lock().unwrap();
+        assert_eq!(entries.len(), 2);
+        let summary: Vec<_> = entries
+            .iter()
+            .map(|entry| {
+                assert_eq!(entry.api.object, None);
+                (
+                    entry.api.name.clone().unwrap(),
+                    entry
+                        .tags
+                        .as_ref()
+                        .unwrap()
+                        .get("deniedKeys")
+                        .and_then(|value| value.as_u64())
+                        .unwrap(),
+                )
+            })
+            .collect();
+        assert_eq!(
+            summary,
+            vec![
+                ("s3:DeleteObject".to_string(), 3),
+                ("s3:BypassGovernanceRetention".to_string(), 1)
+            ]
+        );
     }
 
     #[test]

rustfs/src/storage/denial_audit.rs

@@ -0,0 +1,254 @@
+// 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.
+
+//! Audit entries for authorization denials.
+//!
+//! A denial is caused by the caller, not by an operation the server chose to run, so the
+//! rate of entries is bounded: a process-wide token bucket per caller class (authenticated
+//! and anonymous), with a lower default for anonymous callers. Suppressed denials are
+//! counted and the count is carried on the next entry that is let through.
+
+use crate::module_switches::is_audit_module_enabled;
+use crate::storage::access::{ReqInfo, request_context_from_req};
+use crate::storage::helper::{DenialRequest, DenialScope, authorization_denial_entry, spawn_background_with_context};
+use crate::storage::request_context::RequestContext;
+use rustfs_audit::{AuditEntry, global::AuditLogger};
+use rustfs_policy::policy::action::Action;
+use serde_json::Value;
+use std::sync::{Mutex, OnceLock};
+use std::time::Instant;
+
+/// Tag on an entry that reports how many denials were suppressed since the previous entry.
+pub(crate) const SUPPRESSED_DENIALS_TAG: &str = "suppressedDenials";
+
+struct Bucket {
+    rate: f64,
+    burst: f64,
+    tokens: f64,
+    /// Denials dropped since this class last emitted an entry.
+    suppressed: u64,
+}
+
+impl Bucket {
+    fn new(rate_per_sec: u64) -> Self {
+        let rate = rate_per_sec as f64;
+        let burst = rate.max(1.0);
+        Self {
+            rate,
+            burst,
+            tokens: burst,
+            suppressed: 0,
+        }
+    }
+
+    fn refill(&mut self, elapsed_secs: f64) {
+        self.tokens = (self.tokens + elapsed_secs * self.rate).min(self.burst);
+    }
+
+    /// Takes one token. `Suppress` drops the denial; a bucket with rate 0 is switched
+    /// off and does not count what it drops.
+    fn admit(&mut self) -> Admission {
+        if self.rate > 0.0 && self.tokens >= 1.0 {
+            self.tokens -= 1.0;
+            Admission::Emit {
+                suppressed: std::mem::take(&mut self.suppressed),
+            }
+        } else {
+            if self.rate > 0.0 {
+                self.suppressed = self.suppressed.saturating_add(1);
+            }
+            Admission::Suppress
+        }
+    }
+}
+
+struct LimiterState {
+    authenticated: Bucket,
+    anonymous: Bucket,
+    last: Instant,
+}
+
+/// Outcome of asking the limiter for one denial entry.
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum Admission {
+    /// Emit the entry; `suppressed` denials were dropped since the previous one.
+    Emit {
+        suppressed: u64,
+    },
+    Suppress,
+}
+
+pub(crate) struct DenialRateLimiter {
+    state: Mutex<LimiterState>,
+}
+
+impl DenialRateLimiter {
+    pub(crate) fn new(authenticated_rate: u64, anonymous_rate: u64, now: Instant) -> Self {
+        Self {
+            state: Mutex::new(LimiterState {
+                authenticated: Bucket::new(authenticated_rate),
+                anonymous: Bucket::new(anonymous_rate),
+                last: now,
+            }),
+        }
+    }
+
+    pub(crate) fn admit(&self, anonymous: bool, now: Instant) -> Admission {
+        let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        let elapsed = now.saturating_duration_since(state.last).as_secs_f64();
+        state.last = now.max(state.last);
+        state.authenticated.refill(elapsed);
+        state.anonymous.refill(elapsed);
+
+        if anonymous {
+            state.anonymous.admit()
+        } else {
+            state.authenticated.admit()
+        }
+    }
+}
+
+type Emit = Box<dyn Fn(Option<RequestContext>, AuditEntry) + Send + Sync>;
+
+/// Builds, bounds and dispatches denial audit entries.
+pub(crate) struct DenialAuditor {
+    limiter: DenialRateLimiter,
+    enabled: Box<dyn Fn() -> bool + Send + Sync>,
+    emit: Emit,
+    clock: Box<dyn Fn() -> Instant + Send + Sync>,
+}
+
+impl DenialAuditor {
+    pub(crate) fn new(
+        limiter: DenialRateLimiter,
+        enabled: impl Fn() -> bool + Send + Sync + 'static,
+        emit: impl Fn(Option<RequestContext>, AuditEntry) + Send + Sync + 'static,
+    ) -> Self {
+        Self {
+            limiter,
+            enabled: Box::new(enabled),
+            emit: Box::new(emit),
+            clock: Box::new(Instant::now),
+        }
+    }
+
+    /// Replaces the clock the limiter is driven by; tests use a frozen one so no wall-clock
+    /// refill can change an outcome.
+    #[cfg(test)]
+    pub(crate) fn with_clock(mut self, clock: impl Fn() -> Instant + Send + Sync + 'static) -> Self {
+        self.clock = Box::new(clock);
+        self
+    }
+
+    /// The process-wide auditor: audit switch from the module state, rates from the environment,
+    /// entries dispatched to the audit system in the background.
+    pub(crate) fn global() -> &'static Self {
+        static AUDITOR: OnceLock<DenialAuditor> = OnceLock::new();
+        AUDITOR.get_or_init(|| {
+            let limiter = DenialRateLimiter::new(
+                rustfs_utils::get_env_u64(rustfs_config::ENV_AUDIT_DENIAL_RATE, rustfs_config::DEFAULT_AUDIT_DENIAL_RATE),
+                rustfs_utils::get_env_u64(
+                    rustfs_config::ENV_AUDIT_DENIAL_ANONYMOUS_RATE,
+                    rustfs_config::DEFAULT_AUDIT_DENIAL_ANONYMOUS_RATE,
+                ),
+                Instant::now(),
+            );
+            DenialAuditor::new(limiter, is_audit_module_enabled, |context, entry| {
+                spawn_background_with_context(context, async move {
+                    AuditLogger::log(entry).await;
+                });
+            })
+        })
+    }
+
+    pub(crate) fn audit<T>(&self, req: &DenialRequest<T>, action: &Action, reason: &'static str, scope: DenialScope) {
+        self.audit_at(req, action, reason, scope, (self.clock)());
+    }
+
+    pub(crate) fn audit_at<T>(
+        &self,
+        req: &DenialRequest<T>,
+        action: &Action,
+        reason: &'static str,
+        scope: DenialScope,
+        now: Instant,
+    ) {
+        if !(self.enabled)() {
+            return;
+        }
+        let anonymous = req.extensions.get::<ReqInfo>().is_none_or(|info| info.cred.is_none());
+        let Admission::Emit { suppressed } = self.limiter.admit(anonymous, now) else {
+            return;
+        };
+        let mut entry = authorization_denial_entry(req, action, reason, scope);
+        if suppressed > 0 {
+            entry
+                .tags
+                .get_or_insert_with(Default::default)
+                .insert(SUPPRESSED_DENIALS_TAG.to_string(), Value::from(suppressed));
+        }
+        (self.emit)(request_context_from_req(req), entry);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{Admission, DenialRateLimiter};
+    use std::time::{Duration, Instant};
+
+    #[test]
+    fn burst_is_bounded_and_suppressed_denials_are_counted_on_the_next_entry() {
+        let t0 = Instant::now();
+        let limiter = DenialRateLimiter::new(2, 1, t0);
+        assert_eq!(limiter.admit(false, t0), Admission::Emit { suppressed: 0 });
+        assert_eq!(limiter.admit(false, t0), Admission::Emit { suppressed: 0 });
+        for _ in 0..5 {
+            assert_eq!(limiter.admit(false, t0), Admission::Suppress);
+        }
+        // One second later the bucket has refilled: the next entry reports what was dropped.
+        assert_eq!(limiter.admit(false, t0 + Duration::from_secs(1)), Admission::Emit { suppressed: 5 });
+        assert_eq!(limiter.admit(false, t0 + Duration::from_secs(1)), Admission::Emit { suppressed: 0 });
+    }
+
+    #[test]
+    fn suppressed_counts_are_kept_per_caller_class() {
+        let t0 = Instant::now();
+        let limiter = DenialRateLimiter::new(1, 1, t0);
+        assert_eq!(limiter.admit(true, t0), Admission::Emit { suppressed: 0 });
+        for _ in 0..4 {
+            assert_eq!(limiter.admit(true, t0), Admission::Suppress);
+        }
+        // The authenticated class dropped nothing, so it must not inherit the anonymous drops.
+        assert_eq!(limiter.admit(false, t0), Admission::Emit { suppressed: 0 });
+        assert_eq!(limiter.admit(true, t0 + Duration::from_secs(1)), Admission::Emit { suppressed: 4 });
+    }
+
+    #[test]
+    fn a_zero_rate_disables_that_class_without_counting() {
+        let t0 = Instant::now();
+        let limiter = DenialRateLimiter::new(2, 0, t0);
+        assert_eq!(limiter.admit(true, t0), Admission::Suppress);
+        assert_eq!(limiter.admit(false, t0), Admission::Emit { suppressed: 0 });
+    }
+
+    #[test]
+    fn tokens_never_exceed_the_burst() {
+        let t0 = Instant::now();
+        let limiter = DenialRateLimiter::new(2, 2, t0);
+        let later = t0 + Duration::from_secs(3600);
+        assert_eq!(limiter.admit(false, later), Admission::Emit { suppressed: 0 });
+        assert_eq!(limiter.admit(false, later), Admission::Emit { suppressed: 0 });
+        assert_eq!(limiter.admit(false, later), Admission::Suppress);
+    }
+}

rustfs/src/storage/helper.rs

@@ -22,12 +22,13 @@ use hashbrown::HashMap;
 use http::StatusCode;
 use metrics::counter;
 use rustfs_audit::{
-    ObjectVersion,
+    ApiDetails, AuditEntry, ObjectVersion,
     entity::{ApiDetailsBuilder, AuditEntryBuilder},
     global::AuditLogger,
 };
 use rustfs_io_metrics::record_s3_op;
 use rustfs_notify::EventArgsBuilder;
+use rustfs_policy::policy::action::Action;
 use rustfs_s3_ops::{S3Operation, operation_matches_event_name};
 use rustfs_s3_types::EventName;
 use rustfs_targets::{
@@ -39,6 +40,9 @@ use s3s::{S3Request, S3Response, S3Result};
 use serde_json::Value;
 use std::future::Future;
 use tokio::runtime::{Builder, Handle};
+
+/// Request type seen by the denial auditor, so `denial_audit` stays off the direct s3s surface.
+pub(crate) type DenialRequest<T> = S3Request<T>;
 use tracing::{Instrument, info_span, warn};
 
 #[cfg(feature = "diagnostic-read-version-caller")]
@@ -97,6 +101,42 @@ pub(crate) fn build_event_resp_elements<T>(response: &S3Response<T>, request_id:
     resp_elements
 }
 
+/// Builds the audit entry shared by every request-scoped audit source: caller identity,
+/// client address and the request line, without any outcome.
+fn request_audit_builder<T>(
+    req: &S3Request<T>,
+    req_info: Option<&ReqInfo>,
+    event: EventName,
+    trigger: &str,
+    api: ApiDetails,
+    request_id: &str,
+) -> AuditEntryBuilder {
+    let remote_host = req
+        .extensions
+        .get::<ClientInfo>()
+        .map(|info| info.real_ip.to_string())
+        .or_else(|| {
+            req.extensions
+                .get::<Option<RemoteAddr>>()
+                .and_then(|addr| addr.map(|addr| addr.0.ip().to_string()))
+        })
+        .unwrap_or_default();
+    let mut builder = AuditEntryBuilder::new("1.0", event, trigger, api)
+        .remote_host(remote_host)
+        .user_agent(get_request_user_agent(&req.headers))
+        .req_host(get_request_host(&req.headers))
+        .req_path(req.uri.path().to_string())
+        .req_query(extract_req_params(req))
+        .request_id(request_id);
+    if let Some(cred) = req_info
+        .and_then(|info| info.cred.as_ref())
+        .filter(|cred| !cred.access_key.is_empty())
+    {
+        builder = builder.access_key(&cred.access_key);
+    }
+    builder
+}
+
 /// A unified helper structure for building and distributing audit logs and event notifications via RAII mode at the end of an S3 operation scope.
 pub enum OperationHelper {
     Disabled,
@@ -205,34 +245,8 @@ impl OperationHelper {
         let request_context = request_context.unwrap_or_else(|| RequestContext::from_external_headers(&req.headers));
         let request_id = request_context.request_id.clone();
 
-        let audit_builder = if audit_enabled {
-            let remote_host = req
-                .extensions
-                .get::<ClientInfo>()
-                .map(|info| info.real_ip.to_string())
-                .or_else(|| {
-                    req.extensions
-                        .get::<Option<RemoteAddr>>()
-                        .and_then(|addr| addr.map(|addr| addr.0.ip().to_string()))
-                })
-                .unwrap_or_default();
-            let mut builder = AuditEntryBuilder::new("1.0", event, trigger, api_builder.clone().build())
-                .remote_host(remote_host)
-                .user_agent(get_request_user_agent(&req.headers))
-                .req_host(get_request_host(&req.headers))
-                .req_path(req.uri.path().to_string())
-                .req_query(extract_req_params(req))
-                .request_id(&request_id);
-            if let Some(cred) = req_info
-                .and_then(|info| info.cred.as_ref())
-                .filter(|cred| !cred.access_key.is_empty())
-            {
-                builder = builder.access_key(&cred.access_key);
-            }
-            Some(builder)
-        } else {
-            None
-        };
+        let audit_builder =
+            audit_enabled.then(|| request_audit_builder(req, req_info, event, trigger, api_builder.clone().build(), &request_id));
 
         let event_object = ObjectInfo {
             bucket: bucket.clone(),
@@ -440,6 +454,73 @@ impl OperationHelper {
     }
 }
 
+/// What an authorization denial entry describes.
+#[derive(Clone, Copy, Debug)]
+pub(crate) enum DenialScope {
+    /// One refused request for the request's own bucket and object.
+    Request,
+    /// A bulk request (DeleteObjects) in which this many keys were refused; no single object applies.
+    Keys(u64),
+}
+
+/// Builds the audit entry for a request the authorization layer refused.
+///
+/// The entry carries the action that was denied and the internal denial reason, so an
+/// audit consumer can tell a refused request from a handler that never ran.
+pub(crate) fn authorization_denial_entry<T>(
+    req: &S3Request<T>,
+    action: &Action,
+    reason: &'static str,
+    scope: DenialScope,
+) -> AuditEntry {
+    let req_info = req.extensions.get::<ReqInfo>();
+    let path = req.uri.path().trim_start_matches('/');
+    let mut segs = path.splitn(2, '/');
+    let path_bucket = segs.next().unwrap_or("");
+    let path_object = segs.next().unwrap_or("");
+    let bucket = req_info
+        .and_then(|info| info.bucket.as_deref())
+        .filter(|value| !value.is_empty())
+        .unwrap_or(path_bucket);
+    let object = match scope {
+        DenialScope::Request => req_info
+            .and_then(|info| info.object.as_deref())
+            .filter(|value| !value.is_empty())
+            .unwrap_or(path_object),
+        DenialScope::Keys(_) => "",
+    };
+
+    let action_name: &str = action.into();
+    let mut api = ApiDetailsBuilder::new()
+        .name(action_name)
+        .status("failure")
+        .status_code(StatusCode::FORBIDDEN.as_u16() as i32);
+    if !bucket.is_empty() {
+        api = api.bucket(bucket);
+    }
+    if !object.is_empty() {
+        api = api.object(object);
+    }
+
+    let request_context = request_context_from_req(req).unwrap_or_else(|| RequestContext::from_external_headers(&req.headers));
+    let mut tags = HashMap::new();
+    tags.insert("denialReason".to_string(), Value::String(reason.to_string()));
+    if let DenialScope::Keys(denied) = scope {
+        tags.insert("deniedKeys".to_string(), Value::from(denied));
+    }
+    request_audit_builder(
+        req,
+        req_info,
+        EventName::AuditS3Operation,
+        action_name,
+        api.build(),
+        &request_context.request_id,
+    )
+    .error("Access Denied")
+    .tags(tags)
+    .build()
+}
+
 /// Runs `handler` on `req` and records the outcome as an audit-only entry for `op`.
 ///
 /// The helper is built before the request is handed over so it sees the original headers

rustfs/src/storage/mod.rs

@@ -15,6 +15,7 @@
 pub mod access;
 pub mod concurrency;
 pub mod deadlock_detector;
+pub(crate) mod denial_audit;
 pub mod ecfs;
 pub(crate) mod helper;
 pub mod options;

rustfs/src/storage/rpc/node_service.rs

@@ -1398,8 +1398,13 @@ impl scanner_control_service_server::ScannerControlService for NodeService {
                 .zip(&request.entries)
                 .map(|(guard, entry)| (guard, entry.generation))
                 .collect::<Vec<_>>();
-            rustfs_scanner::acknowledge_scoped_dirty_usage(&request.instance_id, &entries, request.probe_only)
-                .map_err(|err| Status::failed_precondition(err.to_string()))
+            #[cfg(feature = "e2e-test-hooks")]
+            observe_scoped_dirty_usage_ack("ack_before_apply", &request, None).await?;
+            let cleared = rustfs_scanner::acknowledge_scoped_dirty_usage(&request.instance_id, &entries, request.probe_only)
+                .map_err(|err| Status::failed_precondition(err.to_string()))?;
+            #[cfg(feature = "e2e-test-hooks")]
+            observe_scoped_dirty_usage_ack("ack_after_apply", &request, Some(cleared)).await?;
+            Ok::<_, Status>(cleared)
         })
         .await
         .map_err(|_| Status::deadline_exceeded("scoped dirty usage incarnation validation timed out"))??;
@@ -1422,6 +1427,44 @@ impl scanner_control_service_server::ScannerControlService for NodeService {
     }
 }
 
+/// Capture the real compare-and-remove boundary while the production
+/// incarnation fences are still held. Capability probes never pause here.
+#[cfg(feature = "e2e-test-hooks")]
+async fn observe_scoped_dirty_usage_ack(
+    stage: &str,
+    request: &ScannerScopedDirtyUsageAckRequest,
+    cleared: Option<u64>,
+) -> Result<(), Status> {
+    if request.probe_only {
+        return Ok(());
+    }
+    let dirty =
+        rustfs_scanner::scanner_dirty_usage_snapshot(rustfs_protos::scoped_dirty_usage::SCOPED_DIRTY_USAGE_MAX_ENTRIES as usize);
+    let detail = serde_json::json!({
+        "owner_id": request.owner_id,
+        "instance_id": request.instance_id,
+        "probe_only": request.probe_only,
+        "entries": request.entries.iter().map(|entry| serde_json::json!({
+            "bucket": entry.bucket,
+            "bucket_incarnation": Uuid::from_slice(entry.bucket_incarnation.as_ref()).ok().map(|value| value.to_string()),
+            "generation": entry.generation,
+        })).collect::<Vec<_>>(),
+        "cleared": cleared,
+        "dirty_snapshot": {
+            "generation": dirty.generation,
+            "pending_bucket_count": dirty.pending_bucket_count,
+            "complete": dirty.complete,
+            "buckets": dirty.buckets.iter().map(|bucket| serde_json::json!({
+                "bucket": bucket.bucket,
+                "generation": bucket.generation,
+            })).collect::<Vec<_>>(),
+        },
+    });
+    rustfs_scanner::e2e_barrier::checkpoint(stage, detail)
+        .await
+        .map_err(|error| Status::internal(format!("scoped dirty usage test barrier failed: {error}")))
+}
+
 #[tonic::async_trait]
 impl Node for NodeService {
     async fn ping(&self, request: Request<PingRequest>) -> Result<Response<PingResponse>, Status> {