feat(scanner): certify cold peer buckets before scoped ACK

crates/e2e_test/src/distributed/mod.rs

@@ -36,6 +36,8 @@ mod s3_basic_test;
 mod s3_during_data_movement_test;
 mod scanner_heal_topology_fixture;
 mod scanner_heal_topology_test;
+mod scanner_peer_bucket_fixture;
+mod scanner_peer_bucket_test;
 mod site_replication_test;
 mod upgrade_test;
 mod versioning_test;

crates/e2e_test/src/rpc_fault_proxy.rs

@@ -23,7 +23,7 @@ use hyper_util::{
     server::conn::auto,
 };
 use prost::Message as _;
-use rustfs_protos::proto_gen::node_service::{DeleteRequest, ReadVersionRequest};
+use rustfs_protos::proto_gen::node_service::{DeleteRequest, ReadVersionRequest, ScannerScopedDirtyUsageAckRequest};
 use std::{
     collections::HashSet,
     io,
@@ -43,6 +43,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,
@@ -74,7 +77,8 @@ impl RpcPause {
         self.state.blocked.load(Ordering::SeqCst)
     }
 
-    /// Selected RPCs whose backend returned a response, excluding discarded packets.
+    /// Selected RPCs whose backend returned 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)
     }
@@ -132,6 +136,7 @@ impl Drop for RpcPause {
 pub(crate) struct RpcFaultProxy {
     address: SocketAddr,
     slot: PauseSlot,
+    scoped_ack: scoped_ack::QueueSlot,
     task: JoinHandle<()>,
 }
 
@@ -141,6 +146,8 @@ impl RpcFaultProxy {
         let address = listener.local_addr()?;
         let slot: PauseSlot = Arc::new(Mutex::new(None));
         let active = slot.clone();
+        let scoped_ack = scoped_ack::empty_slot();
+        let active_ack = scoped_ack.clone();
         let task = tokio::spawn(async move {
             let h1: ProxyClient = Client::builder(TokioExecutor::new()).build_http();
             let h2: ProxyClient = Client::builder(TokioExecutor::new()).http2_only(true).build_http();
@@ -152,11 +159,13 @@ impl RpcFaultProxy {
                         let h1 = h1.clone();
                         let h2 = h2.clone();
                         let active = active.clone();
+                        let active_ack = active_ack.clone();
                         connections.spawn(async move {
                             let service = service_fn(move |request: Request<Incoming>| {
                                 let client = if request.version() == http::Version::HTTP_2 { h2.clone() } else { h1.clone() };
                                 let pause = active.lock().expect("RPC pause slot").clone();
-                                forward_request(request, client, target, pause)
+                                let ack = active_ack.lock().expect("scoped ACK queue slot").clone();
+                                forward_request(request, client, target, pause, ack)
 
                             });
                             let _ = auto::Builder::new(TokioExecutor::new())
@@ -167,13 +176,23 @@ impl RpcFaultProxy {
                 }
             }
         });
-        Ok(Self { address, slot, task })
+        Ok(Self {
+            address,
+            slot,
+            scoped_ack,
+            task,
+        })
     }
 
     pub(crate) fn local_addr(&self) -> SocketAddr {
         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> {
@@ -217,6 +236,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 != "Delete" {
         return Ok(body.windows(scope.len()).any(|window| window == scope));
     }
@@ -227,26 +249,50 @@ fn scope_matches(method: &str, body: &[u8], scope: &[u8]) -> io::Result<bool> {
         && options.get("recursive").and_then(serde_json::Value::as_bool) == Some(true))
 }
 
+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>>,
-) -> io::Result<hyper::Response<Incoming>> {
-    let (mut parts, body) = request.into_parts();
-    let mut body = body.map_err(io::Error::other).boxed();
+    ack: Option<Arc<scoped_ack::QueueState>>,
+) -> io::Result<hyper::Response<ProxyBody>> {
+    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;
     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 matches = if let Some(scope) = pause.scope.as_deref() {
+            let matches = if pause.scope.is_some() || method == "ScannerScopedDirtyUsageAck" {
                 // Inspect bounded control routing while preserving the exact
                 // signed bytes and trailers forwarded to the backend.
                 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 = scope_matches(method, &bytes, scope)?;
+                let matches = match pause.scope.as_deref() {
+                    Some(scope) => scope_matches(method, &bytes, scope)?,
+                    None => scoped_ack_matches(&bytes, None)?,
+                };
                 if matches && method == "ReadVersion" {
                     let disk = ReadVersionRequest::decode(unary_payload(&bytes)?)
                         .map_err(io::Error::other)?
@@ -283,6 +329,25 @@ async fn forward_request(
             }
         }
     }
+    let response = forward_to_backend(Request::from_parts(parts, body), client, target).await?;
+    if let Some((pause, route)) = observed
+        && response.status().is_success()
+        && response.headers().get("grpc-status").is_none_or(|status| status == "0")
+    {
+        pause.forwarded.fetch_add(1, Ordering::SeqCst);
+        if let Some(route) = route {
+            pause.forwarded_routes.lock().expect("RPC route observations").insert(route);
+        }
+    }
+    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()
     {
@@ -297,20 +362,10 @@ async fn forward_request(
         .path_and_query(parts.uri.path_and_query().map_or("/", |value| value.as_str()))
         .build()
         .map_err(io::Error::other)?;
-    let response = client
+    client
         .request(Request::from_parts(parts, body))
         .await
-        .map_err(io::Error::other)?;
-    if let Some((pause, route)) = observed
-        && response.status().is_success()
-        && response.headers().get("grpc-status").is_none_or(|status| status == "0")
-    {
-        pause.forwarded.fetch_add(1, Ordering::SeqCst);
-        if let Some(route) = route {
-            pause.forwarded_routes.lock().expect("RPC route observations").insert(route);
-        }
-    }
-    Ok(response)
+        .map_err(io::Error::other)
 }
 
 impl Drop for RpcFaultProxy {
@@ -327,6 +382,7 @@ mod tests {
 
     struct Backend {
         received: Arc<AtomicUsize>,
+        auth_headers: Arc<Mutex<Vec<(String, String)>>>,
         address: SocketAddr,
         task: JoinHandle<()>,
     }
@@ -342,6 +398,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 {
@@ -349,12 +407,18 @@ mod tests {
                     accepted = listener.accept() => {
                         let Ok((socket, _)) = accepted else { break };
                         let received = received_task.clone();
+                        let backend_auth = backend_auth.clone();
                         children.spawn(async move {
                             let service = service_fn(move |request: Request<Incoming>| {
                                 let received = received.clone();
+                                let backend_auth = backend_auth.clone();
                                 async move {
                                 received.fetch_add(1, Ordering::SeqCst);
                                 let authority = request.headers().get(http::header::HOST).cloned();
+                                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 mut trailers = http::HeaderMap::new();
                                 trailers.insert("grpc-status", http::HeaderValue::from_static("0"));
@@ -375,7 +439,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>> {
@@ -389,15 +458,367 @@ 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)?;
         client.request(request).await.map_err(io::Error::other)
     }
 
+    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");
+    }
+
     #[tokio::test]
     async fn preserves_protocol_authority_payload_and_grpc_trailers() {
         tokio::time::timeout(Duration::from_secs(10), async {

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/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,209 @@
+// 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.
+
+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"));
+    }
+    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 {
+        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;
+    }
+    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/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

@@ -792,6 +792,16 @@ async fn scan_and_persist_local_bucket(
     let set_disks = scanner_set_disk_inventory(set.as_ref()).await;
     let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel::<DataUsageCache>(1);
     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 = ScannerIODisk::nsscanner_disk(
         disk.clone(),
         scan_ctx.clone(),

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::{
+        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

@@ -589,6 +589,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

@@ -282,6 +282,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,
@@ -1296,6 +1297,12 @@ impl ScannerIOCache for SetDisks {
                     let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel::<DataUsageCache>(1);
 
                     let scan_ctx = ctx_clone.child_token();
+                    #[cfg(feature = "e2e-test-hooks")]
+                    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(),
+                    })).await;
                     let scan = disk.clone().nsscanner_disk(
                         scan_ctx.clone(),
                         budget_clone.clone(),
@@ -1707,6 +1714,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

@@ -1330,6 +1330,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,
                 },
             ),
@@ -1421,6 +1422,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,
                 },
             ),
@@ -1475,6 +1477,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,
                 },
             ),
@@ -2324,6 +2327,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)),
@@ -3241,6 +3245,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),
@@ -3300,6 +3305,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),
@@ -4868,6 +4874,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
@@ -4975,6 +4982,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 +5130,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

@@ -108,8 +108,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,
 };
@@ -118,8 +120,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,
@@ -127,6 +127,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;

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/storage/rpc/node_service.rs

@@ -1396,8 +1396,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"))??;
@@ -1420,6 +1425,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> {