fix(replication): forward single-part object checksums as headers

crates/e2e_test/src/replication_extension_test.rs

@@ -3837,6 +3837,164 @@ async fn test_bucket_replication_converges_delete_marker_and_version_purge() ->
     Ok(())
 }
 
+/// Regression for rustfs/backlog#2340 (not Wasabi specific): a directory
+/// marker (`prefix/` with a body) in a versioned bucket is stored as the null
+/// version, like MinIO (`putOpts`: "for directory objects skip creating new
+/// versions"), and must still replicate to completion instead of staying
+/// `PENDING`.
+#[tokio::test]
+async fn test_bucket_replication_replicates_directory_marker_in_versioned_bucket() -> TestResult {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    let mut source_env_vars = replication_fast_env();
+    source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = "replication-dir-marker-src";
+    let target_bucket = "replication-dir-marker-dst";
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    source_client.create_bucket().bucket(source_bucket).send().await?;
+    target_client.create_bucket().bucket(target_bucket).send().await?;
+    enable_bucket_versioning(&source_env, source_bucket).await?;
+    enable_bucket_versioning(&target_env, target_bucket).await?;
+    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
+    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
+
+    let marker_key = "dir/trailing/";
+    let body = b"directory marker body";
+    let put = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(marker_key)
+        .body(ByteStream::from_static(body))
+        .send()
+        .await?;
+    assert!(
+        put.version_id()
+            .is_none_or(|id| id == "null" || id == uuid::Uuid::nil().to_string()),
+        "a directory marker is the null version even in a versioned bucket: {:?}",
+        put.version_id()
+    );
+
+    wait_for_source_replication_status(&source_client, source_bucket, marker_key, "COMPLETED", false).await?;
+
+    let replica = target_client
+        .get_object()
+        .bucket(target_bucket)
+        .key(marker_key)
+        .send()
+        .await?;
+    assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body);
+    let listed = target_client
+        .list_object_versions()
+        .bucket(target_bucket)
+        .prefix(marker_key)
+        .send()
+        .await?;
+    let marker_versions: Vec<_> = listed.versions().iter().filter(|v| v.key() == Some(marker_key)).collect();
+    assert_eq!(marker_versions.len(), 1, "the marker must land exactly once: {marker_versions:?}");
+    assert_eq!(
+        marker_versions[0].version_id(),
+        Some("null"),
+        "the replica keeps the null version identity"
+    );
+
+    Ok(())
+}
+
+/// Regression for rustfs/backlog#2340 (not Wasabi specific): a single-part
+/// object uploaded with `x-amz-checksum-*` must reach the target with the same
+/// checksum. The outbound options keyed the stored record by algorithm name,
+/// which the target client sent as `x-amz-meta-*` user metadata, so a replica
+/// never carried a checksum although the source HEAD returned one.
+#[tokio::test]
+async fn test_bucket_replication_forwards_single_part_object_checksums() -> TestResult {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    let mut source_env_vars = replication_fast_env();
+    source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = "replication-checksum-src";
+    let target_bucket = "replication-checksum-dst";
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    source_client.create_bucket().bucket(source_bucket).send().await?;
+    target_client.create_bucket().bucket(target_bucket).send().await?;
+    enable_bucket_versioning(&source_env, source_bucket).await?;
+    enable_bucket_versioning(&target_env, target_bucket).await?;
+    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
+    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
+
+    let body = b"123456789";
+    let crc32_key = "checksum-crc32.txt";
+    let sha256_key = "checksum-sha256.txt";
+    let crc32_put = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(crc32_key)
+        .body(ByteStream::from_static(body))
+        .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Crc32)
+        .send()
+        .await?;
+    let expected_crc32 = crc32_put.checksum_crc32().ok_or("source PUT omitted CRC32")?.to_string();
+    let sha256_put = source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(sha256_key)
+        .body(ByteStream::from_static(body))
+        .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Sha256)
+        .send()
+        .await?;
+    let expected_sha256 = sha256_put.checksum_sha256().ok_or("source PUT omitted SHA256")?.to_string();
+
+    for key in [crc32_key, sha256_key] {
+        wait_for_source_replication_status(&source_client, source_bucket, key, "COMPLETED", false).await?;
+    }
+
+    let replica = target_client
+        .head_object()
+        .bucket(target_bucket)
+        .key(crc32_key)
+        .checksum_mode(aws_sdk_s3::types::ChecksumMode::Enabled)
+        .send()
+        .await?;
+    assert_eq!(replica.checksum_crc32(), Some(expected_crc32.as_str()), "replica lost the CRC32 checksum");
+    let replica = target_client
+        .head_object()
+        .bucket(target_bucket)
+        .key(sha256_key)
+        .checksum_mode(aws_sdk_s3::types::ChecksumMode::Enabled)
+        .send()
+        .await?;
+    assert_eq!(
+        replica.checksum_sha256(),
+        Some(expected_sha256.as_str()),
+        "replica lost the SHA256 checksum"
+    );
+    // The bare algorithm name must not leak as user metadata either.
+    assert!(
+        replica
+            .metadata()
+            .is_none_or(|meta| !meta.keys().any(|k| k.eq_ignore_ascii_case("sha256"))),
+        "replica carries the checksum as user metadata: {:?}",
+        replica.metadata()
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_bucket_replication_disabled_delete_marker_does_not_propagate() -> TestResult {
     init_logging();

crates/e2e_test/src/replication_target_matrix_test.rs

@@ -42,7 +42,8 @@ use crate::replication_extension_test::{
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::{ByteStream, DateTime};
 use aws_sdk_s3::types::{
-    Checksum, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus, ObjectLockMode,
+    Checksum, ChecksumAlgorithm, CompletedMultipartUpload, CompletedPart, ObjectAttributes, ObjectLockLegalHoldStatus,
+    ObjectLockMode,
 };
 use bytes::Bytes;
 use std::error::Error;
@@ -114,10 +115,13 @@ enum ObjectShape {
     LockedMultipart,
     /// ODM stores two local parts while preserving a single-PUT source's MD5 ETag.
     OdmPreservedMd5Multipart,
+    /// Single-part object uploaded with `x-amz-checksum-sha256`; the replica
+    /// must carry the same header (rustfs/backlog#2340).
+    Checksummed,
 }
 
 impl ObjectShape {
-    const ALL: [ObjectShape; 7] = [
+    const ALL: [ObjectShape; 8] = [
         ObjectShape::Empty,
         ObjectShape::Plain,
         ObjectShape::Retention,
@@ -125,6 +129,7 @@ impl ObjectShape {
         ObjectShape::Multipart,
         ObjectShape::LockedMultipart,
         ObjectShape::OdmPreservedMd5Multipart,
+        ObjectShape::Checksummed,
     ];
 
     fn key(self) -> &'static str {
@@ -136,6 +141,16 @@ impl ObjectShape {
             ObjectShape::Multipart => "matrix/multipart.bin",
             ObjectShape::LockedMultipart => "matrix/locked-multipart.bin",
             ObjectShape::OdmPreservedMd5Multipart => "matrix/odm-preserved-md5.bin",
+            ObjectShape::Checksummed => "matrix/checksummed.bin",
+        }
+    }
+
+    /// The `x-amz-checksum-*` header the source stored and every upload of
+    /// the replica must repeat.
+    fn forwarded_checksum_header(self) -> Option<&'static str> {
+        match self {
+            ObjectShape::Checksummed => Some("x-amz-checksum-sha256"),
+            _ => None,
         }
     }
 
@@ -198,6 +213,18 @@ impl ObjectShape {
             ObjectShape::Multipart => multipart_put(client, bucket, key, 0x44, false).await,
             ObjectShape::LockedMultipart => multipart_put(client, bucket, key, 0x55, true).await,
             ObjectShape::OdmPreservedMd5Multipart => odm_preserved_md5_multipart(env, bucket, key).await,
+            ObjectShape::Checksummed => {
+                let body = payload(40 * 1024, 0x66);
+                client
+                    .put_object()
+                    .bucket(bucket)
+                    .key(key)
+                    .body(ByteStream::from(body.clone()))
+                    .checksum_algorithm(ChecksumAlgorithm::Sha256)
+                    .send()
+                    .await?;
+                Ok(body)
+            }
         }
     }
 }
@@ -450,6 +477,19 @@ async fn check_completed_cell(
     }) {
         return Err(format!("a locked PutObject went out without any integrity header (rustfs#7082): {bare:?}").into());
     }
+    // rustfs/backlog#2340 contract: a source checksum reaches the target as
+    // the `x-amz-checksum-*` header, not as user metadata; every PutObject of
+    // the shape carries it.
+    if let Some(header) = shape.forwarded_checksum_header()
+        && let Some(missing) = uploads.iter().find(|record| {
+            record.operation == FakeTargetOperation::PutObject
+                && !record.transport.checksum_headers.iter().any(|name| name == header)
+        })
+    {
+        return Err(
+            format!("a PutObject went out without the source's {header} header (rustfs/backlog#2340): {missing:?}").into(),
+        );
+    }
     Ok(())
 }
 

crates/ecstore/src/api/mod.rs

@@ -520,8 +520,7 @@ pub mod rpc {
     pub use crate::cluster::rpc::{
         AuthenticatedChannel, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS,
         PeerRestClient, PeerS3Client, S3PeerSys, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC,
-        ScannerBucketListing, ScannerDirtyUsageAcknowledgement, ScannerPeerActivity, ScannerPeerDirtyUsageBucket,
-        ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease, ScannerScopedDirtyUsageAckEntry, TONIC_RPC_PREFIX,
+        ScannerBucketListing, ScannerPeerActivity, ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease, TONIC_RPC_PREFIX,
         TonicInterceptor, build_put_file_auth_trailer, check_and_record_signed_rpc_nonce, decode_heal_bucket_rpc_options,
         encode_heal_bucket_rpc_options, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers,
         gen_tonic_signature_interceptor, node_service_time_out_client, node_service_time_out_client_no_auth,

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -2064,7 +2064,15 @@ impl TargetClient {
             }
         }
 
-        match builder
+        // A forwarded source checksum is this PUT's integrity header. In
+        // streaming-checksum mode (`RUSTFS_REPLICATION_STREAMING_CHECKSUMS`)
+        // the SDK would still add its default CRC32 trailer, and a target that
+        // receives both keeps the trailer's algorithm: a forwarded SHA256
+        // vanished from the replica while the source reported COMPLETED. Pin
+        // this request to WhenRequired so nothing is sent beside the source's
+        // own checksum.
+        let forwards_source_checksum = headers.keys().any(|name| name.as_str().starts_with("x-amz-checksum-"));
+        let mut operation = builder
             .bucket(bucket)
             .key(object)
             .content_length(size)
@@ -2084,10 +2092,14 @@ impl TargetClient {
                 }
 
                 Result::<_, aws_smithy_types::error::operation::BuildError>::Ok(req)
-            })
-            .send()
-            .await
-        {
+            });
+        if forwards_source_checksum {
+            operation = operation.config_override(
+                aws_sdk_s3::config::Builder::new()
+                    .request_checksum_calculation(aws_sdk_s3::config::RequestChecksumCalculation::WhenRequired),
+            );
+        }
+        match operation.send().await {
             Ok(output) => {
                 // Under SSE-KMS/DSSE or SSE-C the target's ETag is not the MD5
                 // of the stored plaintext, so it cannot be compared against the
@@ -2507,13 +2519,21 @@ mod tests {
     }
 
     fn header_recording_target_client(response_headers: Vec<(String, String)>) -> (TargetClient, RecordedHeaders) {
+        header_recording_target_client_with_checksums(response_headers, replication_request_checksum_calculation())
+    }
+
+    fn header_recording_target_client_with_checksums(
+        response_headers: Vec<(String, String)>,
+        checksums: RequestChecksumCalculation,
+    ) -> (TargetClient, RecordedHeaders) {
         let request_headers: RecordedHeaders = Arc::new(std::sync::Mutex::new(Vec::new()));
         let connector = SharedHttpConnector::new(RecordingHeaderConnector {
             request_headers: Arc::clone(&request_headers),
             response_headers,
         });
         let http_client = http_client_fn(move |_settings, _components| connector.clone());
-        let client = s3_client_for_test(443, Some(http_client));
+        let client =
+            s3_client_for_endpoint_test_with_checksums("https://localhost:443".to_string(), Some(http_client), checksums);
         (
             TargetClient {
                 endpoint: "https://localhost:443".to_string(),
@@ -2680,6 +2700,47 @@ mod tests {
         }
     }
 
+    /// With streaming checksums enabled the SDK adds a CRC32 trailer to every
+    /// upload. A PUT that forwards the source's checksum must not get that
+    /// second algorithm: a target that receives both keeps the trailer's and
+    /// the forwarded SHA256 never reaches the replica (rustfs/backlog#2340).
+    #[tokio::test]
+    async fn streaming_put_object_with_forwarded_checksum_sends_no_sdk_checksum() {
+        let (client, recorded) =
+            header_recording_target_client_with_checksums(Vec::new(), RequestChecksumCalculation::WhenSupported);
+        let mut forwarded = PutObjectOptions::default();
+        forwarded.user_metadata.insert(
+            "x-amz-checksum-sha256".to_string(),
+            "OoJ3yNhRwv3wwtZoGqEIPrPX9xwTnfLl+ka0wStN1g0=".to_string(),
+        );
+        client
+            .put_object("target-bucket", "object", 4, streaming_test_body(b"data"), &forwarded)
+            .await
+            .expect("recorded put_object should succeed");
+        client
+            .put_object("target-bucket", "object", 4, streaming_test_body(b"data"), &PutObjectOptions::default())
+            .await
+            .expect("recorded put_object should succeed");
+        let recorded = recorded.lock().expect("recorded header lock should not be poisoned");
+        let with_forwarded = &recorded[0];
+        assert_eq!(
+            recorded_header(with_forwarded, "x-amz-checksum-sha256"),
+            Some("OoJ3yNhRwv3wwtZoGqEIPrPX9xwTnfLl+ka0wStN1g0=")
+        );
+        assert_eq!(
+            recorded_header(with_forwarded, "x-amz-trailer"),
+            None,
+            "the SDK must not add a trailer checksum"
+        );
+        assert_eq!(recorded_header(with_forwarded, "x-amz-sdk-checksum-algorithm"), None);
+        // Control: the same client still streams a trailer when nothing is forwarded.
+        let without_forwarded = &recorded[1];
+        assert!(
+            recorded_header(without_forwarded, "x-amz-trailer").is_some(),
+            "streaming mode must still apply to uploads without a forwarded checksum: {without_forwarded:?}"
+        );
+    }
+
     /// A forwarded source checksum already satisfies the rule; nothing is added.
     #[tokio::test]
     async fn locked_put_object_keeps_a_forwarded_source_checksum() {
@@ -3045,6 +3106,14 @@ mod tests {
     }
 
     fn s3_client_for_endpoint_test(endpoint: String, http_client: Option<SharedHttpClient>) -> S3Client {
+        s3_client_for_endpoint_test_with_checksums(endpoint, http_client, replication_request_checksum_calculation())
+    }
+
+    fn s3_client_for_endpoint_test_with_checksums(
+        endpoint: String,
+        http_client: Option<SharedHttpClient>,
+        checksums: RequestChecksumCalculation,
+    ) -> S3Client {
         let credentials = SdkCredentials::builder()
             .access_key_id("test-access")
             .secret_access_key("test-secret")
@@ -3058,7 +3127,7 @@ mod tests {
             .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
             // Mirror the production remote-target builder so recorded requests
             // exercise the same checksum/framing behavior (#6853).
-            .request_checksum_calculation(replication_request_checksum_calculation());
+            .request_checksum_calculation(checksums);
         if let Some(http_client) = http_client {
             config = config.http_client(http_client);
         }

crates/ecstore/src/bucket/replication/replication_target_boundary.rs

@@ -281,12 +281,6 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
             // record may only add multipart-ness, never take it away.
             is_multipart = base_is_multipart || checksum_record_is_multipart;
 
-            for (key, value) in checksum_meta.iter() {
-                if key != AMZ_CHECKSUM_TYPE {
-                    meta.insert(key.clone(), value.clone());
-                }
-            }
-
             if !base_is_multipart
                 && checksum_meta
                     .get(AMZ_CHECKSUM_TYPE)
@@ -294,6 +288,26 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo)
             {
                 is_multipart = false;
             }
+
+            // The record keys each checksum by algorithm name ("CRC32"); the
+            // target only reads `x-amz-checksum-<algorithm>`. Inserting the bare
+            // name here made `PutObjectOptions::header()` send it as user
+            // metadata (`x-amz-meta-crc32`), so no replica ever carried the
+            // source checksum (rustfs/backlog#2340). The object-level record
+            // describes one PUT body: a multipart replica is rebuilt part by
+            // part, and its CreateMultipartUpload must not announce a checksum
+            // the parts do not carry, so the record is forwarded on the
+            // single-PUT route only (MinIO `getCRCMeta` parity).
+            if !is_multipart {
+                for (key, value) in checksum_meta.iter() {
+                    if key == AMZ_CHECKSUM_TYPE {
+                        continue;
+                    }
+                    if let Some(header) = rustfs_rio::ChecksumType::from_string(key).key() {
+                        meta.insert(header.to_string(), value.clone());
+                    }
+                }
+            }
         }
     }
 
@@ -1468,12 +1482,63 @@ mod tests {
                 ..Default::default()
             };
 
-            let (opts, _is_multipart) = replication_put_object_options("", &object_info).expect("build replication put options");
+            let (opts, is_multipart) = replication_put_object_options("", &object_info).expect("build replication put options");
 
+            assert!(!is_multipart, "{name}: a single-part checksum record must keep the single-PUT route");
+            let header = ty.key().expect("every forwarded algorithm has an x-amz-checksum header");
             assert_eq!(
-                opts.user_metadata.get(name),
+                opts.user_metadata.get(header),
                 Some(&checksum.encoded),
-                "replication must forward the {name} checksum into user_metadata identically to the classic algorithms"
+                "replication must forward the {name} checksum as the {header} header"
+            );
+            assert!(
+                !opts.user_metadata.contains_key(name),
+                "{name}: the bare algorithm name would leave as x-amz-meta user metadata"
+            );
+        }
+    }
+
+    /// The object-level record of a multipart upload (composite or full-object)
+    /// must not become a PutObject checksum header: the replica is rebuilt
+    /// through CreateMultipartUpload/UploadPart, and a checksum announced there
+    /// that the parts do not carry would be rejected by the target.
+    #[test]
+    fn replication_put_object_options_keeps_multipart_checksum_records_off_the_wire() {
+        let mut composite_type = rustfs_rio::ChecksumType::from_string("crc32");
+        composite_type
+            .merge(rustfs_rio::ChecksumType::MULTIPART)
+            .merge(rustfs_rio::ChecksumType::INCLUDES_MULTIPART);
+        let mut combined = Vec::new();
+        for part in [b"part-one".as_slice(), b"part-two".as_slice()] {
+            let part_checksum =
+                rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::from_string("crc32"), part).expect("part checksum");
+            combined.extend_from_slice(part_checksum.raw.as_slice());
+        }
+        let composite = rustfs_rio::Checksum::new_from_data(composite_type, &combined)
+            .expect("composite checksum")
+            .to_bytes(&combined);
+
+        for (label, checksum, etag) in [
+            ("composite", composite, "0123456789abcdef0123456789abcdef-2"),
+            (
+                "full-object",
+                full_object_multipart_checksum_record(),
+                "0123456789abcdef0123456789abcdef-3",
+            ),
+        ] {
+            let object_info = ObjectInfo {
+                etag: Some(etag.to_string()),
+                checksum: Some(checksum),
+                ..Default::default()
+            };
+            let (opts, is_multipart) = replication_put_object_options("", &object_info).expect("build replication put options");
+            assert!(is_multipart, "{label}: a multipart object must keep the multipart route");
+            assert!(
+                opts.user_metadata
+                    .keys()
+                    .all(|key| !key.starts_with("x-amz-checksum-") && key != "CRC32"),
+                "{label}: no object-level checksum may reach the target's CreateMultipartUpload: {:?}",
+                opts.user_metadata
             );
         }
     }

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

@@ -48,8 +48,7 @@ pub use internode_data_transport::build_internode_data_transport_from_env;
 pub(crate) use peer_rest_client::TierConfigReloadOutcome;
 pub use peer_rest_client::{
     KMS_SIGNAL_SUBSYSTEM, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, SERVICE_SIGNAL_REFRESH_CONFIG,
-    SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerDirtyUsageAcknowledgement, ScannerPeerActivity, ScannerPeerDirtyUsageBucket,
-    ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease, ScannerScopedDirtyUsageAckEntry,
+    SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerPeerActivity, ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease,
 };
 pub(crate) use peer_s3_client::heal_bucket_local_on_disks;
 pub use peer_s3_client::{

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

@@ -49,11 +49,10 @@ use rustfs_protos::proto_gen::node_service::{
     LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest,
     LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ReplacementRecoveryStatusRequest,
     ScannerActivityRequest, ScannerActivityResponse, ScannerDirtyUsageSnapshotRequest, ScannerDirtyUsageSnapshotResponse,
-    ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest, ScannerPublicationLeaseResponse,
-    ScannerScopedDirtyUsageAckRequest, ScannerScopedDirtyUsageEntry, ServerInfoRequest, SignalServiceRequest,
-    SignalServiceResponse, StartDecommissionRequest, StartProfilingRequest, StopRebalanceRequest, TierDailyStatsRequest,
-    TierMutationAbortRequest, TierMutationCommitRequest, TierMutationControlResponse, TierMutationFailureClass,
-    TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient,
+    ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest, ScannerPublicationLeaseResponse, ServerInfoRequest,
+    SignalServiceRequest, SignalServiceResponse, StartDecommissionRequest, StartProfilingRequest, StopRebalanceRequest,
+    TierDailyStatsRequest, TierMutationAbortRequest, TierMutationCommitRequest, TierMutationControlResponse,
+    TierMutationFailureClass, TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient,
     tier_mutation_control_service_client::TierMutationControlServiceClient,
 };
 pub use rustfs_protos::{PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS};
@@ -93,7 +92,6 @@ const HEAL_CONTROL_PAYLOAD_MAX_SIZE: usize = 64 * 1024;
 const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60;
 const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30);
 const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024;
-const SCANNER_SCOPED_DIRTY_USAGE_STAGE_TIMEOUT: Duration = Duration::from_secs(5);
 /// Reserve time for the acquire response's network/clock uncertainty.  The
 /// server owns the real expiry; this local deadline is intentionally earlier
 /// so a coordinator never starts a bounded persistence operation at the edge
@@ -194,102 +192,12 @@ pub struct ScannerPeerActivity {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ScannerPeerDirtyUsageSnapshot {
-    pub owner_id: String,
     pub instance_id: String,
     pub generation: u64,
     pub pending_bucket_count: u64,
     pub protocol_version: u32,
     pub complete: bool,
-    pub buckets: BTreeMap<String, ScannerPeerDirtyUsageBucket>,
-}
-
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub struct ScannerPeerDirtyUsageBucket {
-    pub bucket_incarnation: Uuid,
-    pub generation: u64,
-}
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub struct ScannerScopedDirtyUsageAckEntry {
-    pub bucket: String,
-    pub bucket_incarnation: Uuid,
-    pub generation: u64,
-}
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub enum ScannerDirtyUsageAcknowledgement {
-    Generation {
-        host: String,
-        instance_id: String,
-        generation: u64,
-    },
-    Scoped {
-        host: String,
-        owner_id: String,
-        instance_id: String,
-        entries: Vec<ScannerScopedDirtyUsageAckEntry>,
-    },
-}
-
-fn scanner_scoped_dirty_usage_ack_payloads(
-    owner_id: String,
-    instance_id: String,
-    probe_only: bool,
-    entries: Vec<ScannerScopedDirtyUsageAckEntry>,
-) -> Result<Vec<ScannerScopedDirtyUsageAckRequest>> {
-    use rustfs_protos::scoped_dirty_usage::*;
-
-    if entries.is_empty() {
-        return Err(Error::other("scoped dirty usage acknowledgement entries must be nonempty"));
-    }
-
-    let mut payloads = Vec::with_capacity(entries.len().div_ceil(SCOPED_DIRTY_USAGE_MAX_ENTRIES as usize));
-    let mut batch = Vec::with_capacity(SCOPED_DIRTY_USAGE_MAX_ENTRIES as usize);
-    for entry in entries {
-        batch.push(ScannerScopedDirtyUsageEntry {
-            bucket: entry.bucket,
-            bucket_incarnation: entry.bucket_incarnation.as_bytes().to_vec().into(),
-            generation: entry.generation,
-        });
-        if batch.len() == SCOPED_DIRTY_USAGE_MAX_ENTRIES as usize {
-            payloads.push(scanner_scoped_dirty_usage_ack_payload(
-                &owner_id,
-                &instance_id,
-                probe_only,
-                std::mem::take(&mut batch),
-            )?);
-        }
-    }
-    if !batch.is_empty() {
-        payloads.push(scanner_scoped_dirty_usage_ack_payload(&owner_id, &instance_id, probe_only, batch)?);
-    }
-
-    Ok(payloads)
-}
-
-fn scanner_scoped_dirty_usage_ack_payload(
-    owner_id: &str,
-    instance_id: &str,
-    probe_only: bool,
-    entries: Vec<ScannerScopedDirtyUsageEntry>,
-) -> Result<ScannerScopedDirtyUsageAckRequest> {
-    use rustfs_protos::scoped_dirty_usage::*;
-
-    let payload = ScannerScopedDirtyUsageAckRequest {
-        challenge: Uuid::new_v4().as_bytes().to_vec().into(),
-        protocol_version: SCOPED_DIRTY_USAGE_PROTOCOL_VERSION,
-        owner_id: owner_id.to_string(),
-        instance_id: instance_id.to_string(),
-        scope: SCOPED_DIRTY_USAGE_BUCKET_SCOPE,
-        probe_only,
-        entries,
-    };
-    canonical_scoped_dirty_usage_request(&payload).map_err(|err| Error::other(err.to_string()))?;
-    Ok(payload)
-}
-
-fn scanner_scoped_dirty_usage_ack_reconciled(activity: &ScannerPeerActivity, expected_instance_id: &str) -> bool {
-    activity.instance_id == expected_instance_id && activity.dirty_usage_pending == Some(false)
+    pub buckets: BTreeMap<String, u64>,
 }
 
 fn scanner_instance_id_is_valid(instance_id: &str) -> bool {
@@ -443,11 +351,6 @@ fn decode_scanner_dirty_usage_snapshot_with_verifier(
     if !scanner_instance_id_is_valid(&response.instance_id) {
         return Err(Error::other("peer returned an invalid scanner dirty usage snapshot instance ID"));
     }
-    let owner_id = Uuid::parse_str(&response.owner_id)
-        .ok()
-        .filter(|owner_id| !owner_id.is_nil())
-        .map(|owner_id| owner_id.to_string())
-        .ok_or_else(|| Error::other("peer returned an invalid scanner dirty usage snapshot owner"))?;
     if response.generation == u64::MAX {
         return Err(Error::other("peer scanner dirty usage snapshot exhausted its generation"));
     }
@@ -483,14 +386,9 @@ fn decode_scanner_dirty_usage_snapshot_with_verifier(
         if bucket.generation == 0 || bucket.generation > response.generation {
             return Err(Error::other("peer scanner dirty usage snapshot contains an invalid bucket generation"));
         }
-        Uuid::from_slice(bucket.bucket_incarnation.as_ref())
-            .ok()
-            .filter(|bucket_incarnation| !bucket_incarnation.is_nil())
-            .ok_or_else(|| Error::other("peer scanner dirty usage snapshot contains an invalid bucket incarnation"))?;
     }
 
     Ok(ScannerPeerDirtyUsageSnapshot {
-        owner_id,
         instance_id: response.instance_id,
         generation: response.generation,
         pending_bucket_count: response.pending_bucket_count,
@@ -499,16 +397,7 @@ fn decode_scanner_dirty_usage_snapshot_with_verifier(
         buckets: response
             .buckets
             .into_iter()
-            .map(|bucket| {
-                (
-                    bucket.bucket,
-                    ScannerPeerDirtyUsageBucket {
-                        bucket_incarnation: Uuid::from_slice(bucket.bucket_incarnation.as_ref())
-                            .expect("bucket incarnation was validated"),
-                        generation: bucket.generation,
-                    },
-                )
-            })
+            .map(|bucket| (bucket.bucket, bucket.generation))
             .collect(),
     })
 }
@@ -2188,10 +2077,19 @@ impl PeerRestClient {
         &self,
         owner_id: String,
         instance_id: String,
-        entries: Vec<ScannerScopedDirtyUsageAckEntry>,
+        entries: Vec<rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageEntry>,
     ) -> Result<bool> {
         use rustfs_protos::scoped_dirty_usage::*;
-        let payloads = scanner_scoped_dirty_usage_ack_payloads(owner_id, instance_id, true, entries)?;
+        let payload = rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageAckRequest {
+            challenge: Uuid::new_v4().as_bytes().to_vec().into(),
+            protocol_version: SCOPED_DIRTY_USAGE_PROTOCOL_VERSION,
+            owner_id,
+            instance_id,
+            scope: SCOPED_DIRTY_USAGE_BUCKET_SCOPE,
+            probe_only: true,
+            entries,
+        };
+        let canonical = canonical_scoped_dirty_usage_request(&payload).map_err(|err| Error::other(err.to_string()))?;
         self.finalize_result(
             async {
                 let mut client = super::client::scanner_control_time_out_client(
@@ -2199,106 +2097,26 @@ impl PeerRestClient {
                     TonicInterceptor::Signature(gen_tonic_signature_interceptor()),
                 )
                 .await?;
-                for payload in payloads {
-                    let canonical =
-                        canonical_scoped_dirty_usage_request(&payload).map_err(|err| Error::other(err.to_string()))?;
-                    let mut request = Request::new(payload.clone());
-                    set_tonic_canonical_body_digest(&mut request, &canonical)?;
-                    let response = client.scanner_scoped_dirty_usage_ack(request).await?.into_inner();
-                    let body = canonical_scoped_dirty_usage_response(&canonical, &response)
-                        .map_err(|_| Error::other("scoped dirty usage capability response is too large"))?;
-                    verify_tonic_rpc_response_proof(&body, response.response_proof.as_ref())?;
-                    if response.protocol_version != SCOPED_DIRTY_USAGE_PROTOCOL_VERSION
-                        || response.owner_id != payload.owner_id
-                        || response.instance_id != payload.instance_id
-                        || response.max_entries != SCOPED_DIRTY_USAGE_MAX_ENTRIES
-                        || response.max_request_bytes != SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES
-                        || response.cleared != 0
-                    {
-                        return Err(Error::other("scoped dirty usage capability response does not match request"));
-                    }
-                    if !response.supported {
-                        return Ok(false);
-                    }
-                }
-                Ok(true)
-            }
-            .await,
-        )
-        .await
-    }
-
-    pub async fn acknowledge_scanner_scoped_dirty_usage(
-        &self,
-        owner_id: String,
-        instance_id: String,
-        entries: Vec<ScannerScopedDirtyUsageAckEntry>,
-    ) -> Result<ScannerPeerActivity> {
-        use rustfs_protos::scoped_dirty_usage::*;
-        let payloads = scanner_scoped_dirty_usage_ack_payloads(owner_id, instance_id.clone(), false, entries)?;
-        let ack_attempt = async {
-            let mut client = super::client::scanner_control_time_out_client(
-                &self.grid_host,
-                TonicInterceptor::Signature(gen_tonic_signature_interceptor()),
-            )
-            .await?;
-            for payload in payloads {
-                let canonical = canonical_scoped_dirty_usage_request(&payload).map_err(|err| Error::other(err.to_string()))?;
                 let mut request = Request::new(payload.clone());
                 set_tonic_canonical_body_digest(&mut request, &canonical)?;
                 let response = client.scanner_scoped_dirty_usage_ack(request).await?.into_inner();
                 let body = canonical_scoped_dirty_usage_response(&canonical, &response)
-                    .map_err(|_| Error::other("scoped dirty usage acknowledgement response is too large"))?;
+                    .map_err(|_| Error::other("scoped dirty usage capability response is too large"))?;
                 verify_tonic_rpc_response_proof(&body, response.response_proof.as_ref())?;
                 if response.protocol_version != SCOPED_DIRTY_USAGE_PROTOCOL_VERSION
                     || response.owner_id != payload.owner_id
                     || response.instance_id != payload.instance_id
                     || response.max_entries != SCOPED_DIRTY_USAGE_MAX_ENTRIES
                     || response.max_request_bytes != SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES
-                    || !response.supported
+                    || response.cleared != 0
                 {
-                    return Err(Error::other("scoped dirty usage acknowledgement response does not match request"));
+                    return Err(Error::other("scoped dirty usage capability response does not match request"));
                 }
+                Ok(response.supported)
             }
-            Ok(())
-        };
-        let result = match timeout(SCANNER_SCOPED_DIRTY_USAGE_STAGE_TIMEOUT, ack_attempt).await {
-            Ok(result) => self.finalize_result(result).await,
-            Err(_) => {
-                self.prepare_retry_with_timeout(SCANNER_SCOPED_DIRTY_USAGE_STAGE_TIMEOUT)
-                    .await;
-                Err(Error::other("scoped dirty usage acknowledgement deadline elapsed"))
-            }
-        };
-
-        match result {
-            Ok(()) => {
-                let activity = self.scanner_scoped_dirty_usage_activity_confirmation().await?;
-                if activity.instance_id == instance_id {
-                    Ok(activity)
-                } else {
-                    Err(Error::other(
-                        "scoped dirty usage acknowledgement peer restarted before activity confirmation",
-                    ))
-                }
-            }
-            Err(err) => {
-                if Self::is_network_like_error(&err) {
-                    self.prepare_retry_with_timeout(SCANNER_SCOPED_DIRTY_USAGE_STAGE_TIMEOUT)
-                        .await;
-                }
-                match self.scanner_scoped_dirty_usage_activity_confirmation().await {
-                    Ok(activity) if scanner_scoped_dirty_usage_ack_reconciled(&activity, &instance_id) => Ok(activity),
-                    _ => Err(err),
-                }
-            }
-        }
-    }
-
-    async fn scanner_scoped_dirty_usage_activity_confirmation(&self) -> Result<ScannerPeerActivity> {
-        timeout(SCANNER_SCOPED_DIRTY_USAGE_STAGE_TIMEOUT, self.scanner_activity())
-            .await
-            .map_err(|_| Error::other("scoped dirty usage activity confirmation timed out"))?
+            .await,
+        )
+        .await
     }
 
     pub async fn acknowledge_scanner_dirty_usage(&self, instance_id: String, generation: u64) -> Result<ScannerPeerActivity> {
@@ -3027,79 +2845,16 @@ mod tests {
                 rustfs_protos::proto_gen::node_service::ScannerDirtyUsageBucket {
                     bucket: "archive".to_string(),
                     generation: 3,
-                    bucket_incarnation: Uuid::from_u128(0x11111111111111111111111111111111).as_bytes().to_vec().into(),
                 },
                 rustfs_protos::proto_gen::node_service::ScannerDirtyUsageBucket {
                     bucket: "photos".to_string(),
                     generation: 7,
-                    bucket_incarnation: Uuid::from_u128(0x22222222222222222222222222222222).as_bytes().to_vec().into(),
                 },
             ],
             response_proof: b"proof".to_vec().into(),
-            owner_id: "33333333-3333-3333-3333-333333333333".to_string(),
         }
     }
 
-    #[test]
-    fn scanner_scoped_dirty_usage_ack_payloads_split_at_protocol_limit() {
-        use rustfs_protos::scoped_dirty_usage::{SCOPED_DIRTY_USAGE_MAX_ENTRIES, canonical_scoped_dirty_usage_request};
-
-        let entries = (0..=SCOPED_DIRTY_USAGE_MAX_ENTRIES)
-            .map(|index| ScannerScopedDirtyUsageAckEntry {
-                bucket: format!("bucket-{index:02}"),
-                bucket_incarnation: Uuid::from_u128(0x11111111111111111111111111111111),
-                generation: 9,
-            })
-            .collect::<Vec<_>>();
-
-        let payloads = scanner_scoped_dirty_usage_ack_payloads(
-            "33333333-3333-3333-3333-333333333333".to_string(),
-            "0123456789abcdef0123456789abcdef".to_string(),
-            false,
-            entries,
-        )
-        .expect("33 entries should split into valid scoped dirty usage requests");
-
-        assert_eq!(payloads.len(), 2);
-        assert_eq!(payloads[0].entries.len(), SCOPED_DIRTY_USAGE_MAX_ENTRIES as usize);
-        assert_eq!(payloads[1].entries.len(), 1);
-        assert_eq!(payloads[0].entries.first().map(|entry| entry.bucket.as_str()), Some("bucket-00"));
-        assert_eq!(payloads[0].entries.last().map(|entry| entry.bucket.as_str()), Some("bucket-31"));
-        assert_eq!(payloads[1].entries.first().map(|entry| entry.bucket.as_str()), Some("bucket-32"));
-        for payload in payloads {
-            canonical_scoped_dirty_usage_request(&payload).expect("each split scoped ACK payload should be canonical");
-        }
-    }
-
-    #[test]
-    fn scanner_scoped_dirty_usage_ack_reconciliation_requires_same_clean_instance() {
-        let activity = |instance_id: &str, pending| ScannerPeerActivity {
-            instance_id: instance_id.to_string(),
-            namespace_generation: 1,
-            maintenance_generation: 1,
-            protocol_version: SCANNER_ACTIVITY_PROTOCOL_VERSION,
-            topology_digest: Some([1; 32]),
-            data_movement_active: Some(false),
-            dirty_usage_generation: Some(9),
-            dirty_usage_pending: pending,
-            movement_generation: Some(1),
-            publication_blocked: Some(false),
-        };
-
-        assert!(scanner_scoped_dirty_usage_ack_reconciled(
-            &activity("0123456789abcdef0123456789abcdef", Some(false)),
-            "0123456789abcdef0123456789abcdef"
-        ));
-        assert!(!scanner_scoped_dirty_usage_ack_reconciled(
-            &activity("0123456789abcdef0123456789abcdef", Some(true)),
-            "0123456789abcdef0123456789abcdef"
-        ));
-        assert!(!scanner_scoped_dirty_usage_ack_reconciled(
-            &activity("fedcba9876543210fedcba9876543210", Some(false)),
-            "0123456789abcdef0123456789abcdef"
-        ));
-    }
-
     #[test]
     fn scanner_dirty_usage_snapshot_requires_a_complete_authenticated_ordered_view() {
         let decoded = decode_test_scanner_dirty_usage_snapshot(test_scanner_dirty_usage_snapshot_response())
@@ -3108,18 +2863,9 @@ mod tests {
         assert_eq!(decoded.generation, 7);
         assert_eq!(decoded.pending_bucket_count, 2);
         assert_eq!(decoded.protocol_version, SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION);
-        assert_eq!(decoded.owner_id, "33333333-3333-3333-3333-333333333333");
         assert!(decoded.complete);
-        assert_eq!(
-            decoded.buckets.get("archive").map(|bucket| bucket.bucket_incarnation),
-            Some(Uuid::from_u128(0x11111111111111111111111111111111))
-        );
-        assert_eq!(decoded.buckets.get("archive").map(|bucket| bucket.generation), Some(3));
-        assert_eq!(
-            decoded.buckets.get("photos").map(|bucket| bucket.bucket_incarnation),
-            Some(Uuid::from_u128(0x22222222222222222222222222222222))
-        );
-        assert_eq!(decoded.buckets.get("photos").map(|bucket| bucket.generation), Some(7));
+        assert_eq!(decoded.buckets.get("archive"), Some(&3));
+        assert_eq!(decoded.buckets.get("photos"), Some(&7));
 
         let overflow_count =
             u64::try_from(SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES + 1).expect("the test snapshot entry limit should fit in u64");
@@ -3170,14 +2916,6 @@ mod tests {
         empty_bucket.buckets[0].bucket.clear();
         cases.push((empty_bucket, "empty bucket name"));
 
-        let mut invalid_owner = test_scanner_dirty_usage_snapshot_response();
-        invalid_owner.owner_id.clear();
-        cases.push((invalid_owner, "snapshot owner"));
-
-        let mut invalid_incarnation = test_scanner_dirty_usage_snapshot_response();
-        invalid_incarnation.buckets[0].bucket_incarnation = Uuid::nil().as_bytes().to_vec().into();
-        cases.push((invalid_incarnation, "bucket incarnation"));
-
         let mut partial = test_scanner_dirty_usage_snapshot_response();
         partial.complete = false;
         cases.push((partial, "entry-limit overflow"));
@@ -3190,7 +2928,6 @@ mod tests {
                 .map(|index| rustfs_protos::proto_gen::node_service::ScannerDirtyUsageBucket {
                     bucket: format!("bucket-{index:04}"),
                     generation: 1,
-                    bucket_incarnation: Uuid::from_u128(0x11111111111111111111111111111111).as_bytes().to_vec().into(),
                 })
                 .collect(),
             ..test_scanner_dirty_usage_snapshot_response()

crates/ecstore/src/services/notification_sys.rs

@@ -14,8 +14,7 @@
 
 use crate::bucket::lifecycle::tier_last_day_stats::DailyAllTierStats;
 use crate::cluster::rpc::{
-    PeerRestClient, ScannerDirtyUsageAcknowledgement, ScannerPeerActivity, ScannerPeerDirtyUsageSnapshot,
-    ScannerPublicationLease, TierConfigReloadOutcome,
+    PeerRestClient, ScannerPeerActivity, ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease, TierConfigReloadOutcome,
 };
 use crate::diagnostics::admin_server_info::get_commit_id;
 use crate::disk::DiskAPI;
@@ -2553,70 +2552,11 @@ impl NotificationSys {
         Ok(snapshots)
     }
 
-    pub async fn scanner_scoped_dirty_usage_capabilities(
-        &self,
-        acknowledgements: Vec<ScannerDirtyUsageAcknowledgement>,
-    ) -> Result<bool> {
+    pub async fn acknowledge_scanner_dirty_usage(&self, acknowledgements: Vec<(String, String, u64)>) -> Result<bool> {
         let mut by_host = HashMap::with_capacity(acknowledgements.len());
-        for acknowledgement in acknowledgements {
-            let host = match &acknowledgement {
-                ScannerDirtyUsageAcknowledgement::Scoped { host, .. } => host.clone(),
-                ScannerDirtyUsageAcknowledgement::Generation { .. } => {
-                    return Err(Error::other("scanner scoped dirty usage capability requires scoped acknowledgements"));
-                }
-            };
-            if by_host.insert(host.clone(), acknowledgement).is_some() {
-                return Err(Error::other("duplicate scanner dirty usage acknowledgement target"));
-            }
-        }
-
-        let clients = self
-            .peer_clients
-            .iter()
-            .flatten()
-            .map(|client| (client.grid_host.clone(), client.clone()))
-            .collect::<HashMap<_, _>>();
-        let mut futures = Vec::with_capacity(by_host.len());
-        for (host, acknowledgement) in by_host {
-            let Some(client) = clients.get(&host).cloned() else {
-                return Err(Error::other("scanner scoped dirty usage capability failed: peer is not reachable"));
-            };
-            futures.push(async move {
-                let ScannerDirtyUsageAcknowledgement::Scoped {
-                    owner_id,
-                    instance_id,
-                    entries,
-                    ..
-                } = acknowledgement
-                else {
-                    unreachable!("scoped acknowledgement was validated before probing");
-                };
-                timeout(
-                    SCANNER_ACTIVITY_PROBE_TIMEOUT,
-                    client.scanner_scoped_dirty_usage_capability(owner_id, instance_id, entries),
-                )
-                .await
-                .map_err(|_| Error::other("scanner scoped dirty usage capability timed out"))?
-            });
-        }
-
-        for result in join_all(futures).await {
-            if !result? {
-                return Ok(false);
-            }
-        }
-        Ok(true)
-    }
-
-    pub async fn acknowledge_scanner_dirty_usage(&self, acknowledgements: Vec<ScannerDirtyUsageAcknowledgement>) -> Result<bool> {
-        let mut by_host = HashMap::with_capacity(acknowledgements.len());
-        for acknowledgement in acknowledgements {
-            let host = match &acknowledgement {
-                ScannerDirtyUsageAcknowledgement::Generation { host, .. }
-                | ScannerDirtyUsageAcknowledgement::Scoped { host, .. } => host.clone(),
-            };
-            if by_host.insert(host.clone(), acknowledgement).is_some() {
-                return Err(Error::other("duplicate scanner dirty usage acknowledgement target"));
+        for (host, instance_id, generation) in acknowledgements {
+            if by_host.insert(host.clone(), (instance_id, generation)).is_some() {
+                return Err(Error::other(format!("duplicate scanner dirty usage acknowledgement target: {host}")));
             }
         }
 
@@ -2628,34 +2568,18 @@ impl NotificationSys {
             .collect::<HashMap<_, _>>();
         let mut failures = Vec::new();
         let mut futures = Vec::with_capacity(by_host.len());
-        for (host, acknowledgement) in by_host {
+        for (host, (instance_id, generation)) in by_host {
             let Some(client) = clients.get(&host).cloned() else {
                 failures.push(format!("peer {host} scanner dirty usage acknowledgement failed: peer is not reachable"));
                 continue;
             };
             futures.push(async move {
-                let result = match acknowledgement {
-                    ScannerDirtyUsageAcknowledgement::Generation {
-                        instance_id, generation, ..
-                    } => {
-                        scanner_activity_with_timeout(
-                            SCANNER_ACTIVITY_PROBE_TIMEOUT,
-                            &host,
-                            client.acknowledge_scanner_dirty_usage(instance_id, generation),
-                        )
-                        .await
-                    }
-                    ScannerDirtyUsageAcknowledgement::Scoped {
-                        owner_id,
-                        instance_id,
-                        entries,
-                        ..
-                    } => {
-                        client
-                            .acknowledge_scanner_scoped_dirty_usage(owner_id, instance_id, entries)
-                            .await
-                    }
-                };
+                let result = scanner_activity_with_timeout(
+                    SCANNER_ACTIVITY_PROBE_TIMEOUT,
+                    &host,
+                    client.acknowledge_scanner_dirty_usage(instance_id, generation),
+                )
+                .await;
                 (host, result)
             });
         }
@@ -4820,28 +4744,15 @@ mod tests {
             peer_topology_hosts: Vec::new(),
         };
         let missing = sys
-            .acknowledge_scanner_dirty_usage(vec![ScannerDirtyUsageAcknowledgement::Generation {
-                host: "peer-1".to_string(),
-                instance_id: "0123456789abcdef0123456789abcdef".to_string(),
-                generation: 7,
-            }])
+            .acknowledge_scanner_dirty_usage(vec![("peer-1".to_string(), "0123456789abcdef0123456789abcdef".to_string(), 7)])
             .await
             .expect_err("a missing acknowledgement target must remain pending");
         assert!(missing.to_string().contains("peer is not reachable"));
 
         let duplicate = sys
             .acknowledge_scanner_dirty_usage(vec![
-                ScannerDirtyUsageAcknowledgement::Generation {
-                    host: "peer-1".to_string(),
-                    instance_id: "0123456789abcdef0123456789abcdef".to_string(),
-                    generation: 7,
-                },
-                ScannerDirtyUsageAcknowledgement::Scoped {
-                    host: "peer-1".to_string(),
-                    owner_id: "11111111-1111-1111-1111-111111111111".to_string(),
-                    instance_id: "0123456789abcdef0123456789abcdef".to_string(),
-                    entries: Vec::new(),
-                },
+                ("peer-1".to_string(), "0123456789abcdef0123456789abcdef".to_string(), 7),
+                ("peer-1".to_string(), "0123456789abcdef0123456789abcdef".to_string(), 7),
             ])
             .await
             .expect_err("duplicate acknowledgement targets must be rejected");

crates/protos/src/generated/proto_gen/node_service.rs

@@ -1257,8 +1257,6 @@ pub struct ScannerDirtyUsageBucket {
     pub bucket: ::prost::alloc::string::String,
     #[prost(uint64, tag = "2")]
     pub generation: u64,
-    #[prost(bytes = "bytes", tag = "3")]
-    pub bucket_incarnation: ::prost::bytes::Bytes,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ScannerDirtyUsageSnapshotRequest {
@@ -1284,8 +1282,6 @@ pub struct ScannerDirtyUsageSnapshotResponse {
     pub buckets: ::prost::alloc::vec::Vec<ScannerDirtyUsageBucket>,
     #[prost(bytes = "bytes", tag = "7")]
     pub response_proof: ::prost::bytes::Bytes,
-    #[prost(string, tag = "8")]
-    pub owner_id: ::prost::alloc::string::String,
 }
 /// Receiver-only protocol. Producers must retain whole-cycle ACK until they
 /// have a durable per-bucket publication proof.

crates/protos/src/lib.rs

@@ -575,13 +575,11 @@ pub fn canonical_scanner_dirty_usage_snapshot_response_body(
     body.push_u64(response.generation);
     body.push_u64(response.pending_bucket_count);
     body.push_u32(response.protocol_version);
-    body.push_str(&response.owner_id)?;
     body.push_bool(response.complete);
     body.push_count(response.buckets.len())?;
     for bucket in &response.buckets {
         body.push_str(&bucket.bucket)?;
         body.push_u64(bucket.generation);
-        body.push_bytes(bucket.bucket_incarnation.as_ref())?;
     }
     Ok(body.finish())
 }
@@ -1844,16 +1842,13 @@ mod scanner_activity_tests {
                 ScannerDirtyUsageBucket {
                     bucket: "archive".to_string(),
                     generation: 3,
-                    bucket_incarnation: vec![1; 16].into(),
                 },
                 ScannerDirtyUsageBucket {
                     bucket: "photos".to_string(),
                     generation: 7,
-                    bucket_incarnation: vec![2; 16].into(),
                 },
             ],
             response_proof: vec![9; 32].into(),
-            owner_id: "11111111-1111-1111-1111-111111111111".to_string(),
         };
         let baseline = canonical_scanner_dirty_usage_snapshot_response_body(&[1; 16], &response)
             .expect("scanner dirty usage snapshot response should encode");
@@ -1870,9 +1865,6 @@ mod scanner_activity_tests {
         let mut protocol = response.clone();
         protocol.protocol_version = 2;
         variants.push(protocol);
-        let mut owner = response.clone();
-        owner.owner_id = "22222222-2222-2222-2222-222222222222".to_string();
-        variants.push(owner);
         let mut complete = response.clone();
         complete.complete = false;
         variants.push(complete);
@@ -1882,9 +1874,6 @@ mod scanner_activity_tests {
         let mut bucket_generation = response.clone();
         bucket_generation.buckets[0].generation = 4;
         variants.push(bucket_generation);
-        let mut bucket_incarnation = response.clone();
-        bucket_incarnation.buckets[0].bucket_incarnation = vec![3; 16].into();
-        variants.push(bucket_incarnation);
         let mut bucket_order = response.clone();
         bucket_order.buckets.reverse();
         variants.push(bucket_order);

crates/protos/src/node.proto

@@ -885,7 +885,6 @@ message ScannerActivityResponse {
 message ScannerDirtyUsageBucket {
   string bucket = 1;
   uint64 generation = 2;
-  bytes bucket_incarnation = 3;
 }
 
 message ScannerDirtyUsageSnapshotRequest {
@@ -902,7 +901,6 @@ message ScannerDirtyUsageSnapshotResponse {
   bool complete = 5;
   repeated ScannerDirtyUsageBucket buckets = 6;
   bytes response_proof = 7;
-  string owner_id = 8;
 }
 
 // Receiver-only protocol. Producers must retain whole-cycle ACK until they

crates/scanner/src/lib.rs

@@ -100,7 +100,6 @@ pub use storage_api::ScannerReplicationConfig as ReplicationConfig;
 pub use storage_api::scan::{
     SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES,
     SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE,
-    SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES,
 };
 pub use workload_admission::set_scanner_workload_admission_snapshot_provider;
 

crates/scanner/src/scanner.rs

@@ -2216,7 +2216,10 @@ where
         false
     } else if let Some(notification_system) = storeapi.scanner_notification_system() {
         let acknowledgement_count = remote_dirty_usage_acknowledgements.len();
-        let acknowledgements = remote_dirty_usage_acknowledgements.into_iter().map(Into::into).collect();
+        let acknowledgements = remote_dirty_usage_acknowledgements
+            .into_iter()
+            .map(|acknowledgement| (acknowledgement.host, acknowledgement.instance_id, acknowledgement.generation))
+            .collect();
         remote_dirty_usage_acknowledgement_pending(
             cycle_info.current,
             acknowledgement_count,
@@ -3575,10 +3578,9 @@ use usage_store::*;
 
 pub use activity::scanner_topology_digest;
 pub(crate) use activity::{
-    ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, ScannerDirtyUsageAcknowledgementKind, probe_scanner_activity,
-    scanner_activity_allows_usage_publication, scanner_activity_dirty_usage_state_for_host,
-    scanner_activity_publication_lease_targets, scanner_activity_snapshot_digest, scanner_activity_structural_digest,
-    scanner_dirty_usage_acknowledgements,
+    ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, probe_scanner_activity, scanner_activity_allows_usage_publication,
+    scanner_activity_dirty_usage_state_for_host, scanner_activity_publication_lease_targets, scanner_activity_snapshot_digest,
+    scanner_activity_structural_digest, scanner_dirty_usage_acknowledgements,
 };
 pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance};
 pub use backlog::{

crates/scanner/src/scanner/activity.rs

@@ -445,34 +445,7 @@ pub(crate) type ScannerActivitySnapshot = BTreeMap<String, ScannerNodeActivity>;
 pub(crate) struct ScannerDirtyUsageAcknowledgement {
     pub(crate) host: String,
     pub(crate) instance_id: String,
-    pub(crate) kind: ScannerDirtyUsageAcknowledgementKind,
-}
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub(crate) enum ScannerDirtyUsageAcknowledgementKind {
-    Generation(u64),
-    Scoped {
-        owner_id: String,
-        entries: Vec<crate::storage_api::EcstoreScannerScopedDirtyUsageAckEntry>,
-    },
-}
-
-impl From<ScannerDirtyUsageAcknowledgement> for crate::storage_api::EcstoreScannerDirtyUsageAcknowledgement {
-    fn from(acknowledgement: ScannerDirtyUsageAcknowledgement) -> Self {
-        match acknowledgement.kind {
-            ScannerDirtyUsageAcknowledgementKind::Generation(generation) => Self::Generation {
-                host: acknowledgement.host,
-                instance_id: acknowledgement.instance_id,
-                generation,
-            },
-            ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries } => Self::Scoped {
-                host: acknowledgement.host,
-                owner_id,
-                instance_id: acknowledgement.instance_id,
-                entries,
-            },
-        }
-    }
+    pub(crate) generation: u64,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -1001,7 +974,7 @@ pub(crate) fn scanner_dirty_usage_acknowledgements(snapshot: &ScannerActivitySna
         .map(|(host, activity)| ScannerDirtyUsageAcknowledgement {
             host: host.clone(),
             instance_id: activity.instance_id.clone(),
-            kind: ScannerDirtyUsageAcknowledgementKind::Generation(activity.dirty_usage_generation),
+            generation: activity.dirty_usage_generation,
         })
         .collect()
 }

crates/scanner/src/scanner/tests.rs

@@ -7433,7 +7433,7 @@ fn finalizing_a_saved_enum_without_proof_keeps_dirty_pending() {
     let remote_acknowledgement = ScannerDirtyUsageAcknowledgement {
         host: "node-2".to_string(),
         instance_id: "0123456789abcdef0123456789abcdef".to_string(),
-        kind: ScannerDirtyUsageAcknowledgementKind::Generation(11),
+        generation: 11,
     };
     let unsaved = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot.clone()))
         .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement.clone()]);
@@ -8982,7 +8982,7 @@ fn post_lease_activity_proof_rejects_a_put_tail_that_finished_before_lease_acqui
         ScannerDirtyUsageAcknowledgement {
             host: "node-2".to_string(),
             instance_id: "epoch-a".to_string(),
-            kind: ScannerDirtyUsageAcknowledgementKind::Generation(5),
+            generation: 5,
         },
     ]);
     let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(
@@ -9142,7 +9142,7 @@ fn scanner_dirty_usage_acknowledgements_exclude_local_and_clean_nodes() {
         vec![ScannerDirtyUsageAcknowledgement {
             host: "node-3".to_string(),
             instance_id: "epoch-dirty".to_string(),
-            kind: ScannerDirtyUsageAcknowledgementKind::Generation(11),
+            generation: 11,
         }]
     );
 }

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

@@ -463,7 +463,7 @@ async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish
             "remote_ack_target" => scan.with_remote_dirty_usage_acknowledgements(vec![ScannerDirtyUsageAcknowledgement {
                 host: "proof-peer:9000".to_string(),
                 instance_id: crate::scanner_activity_epoch().to_string(),
-                kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Generation(changed_generation),
+                generation: changed_generation,
             }]),
             "publication_epoch" => scan.with_publication_epoch(Some(changed_epoch)),
             "remote_lease_targets" => scan.with_remote_publication_lease_targets(vec![(

crates/scanner/src/scanner_io.rs

@@ -148,26 +148,19 @@ struct ScannerPeerDirtyUsageExpectation {
     pending: bool,
 }
 
-#[derive(Debug, PartialEq, Eq)]
-struct VerifiedRemoteDirtyUsage {
-    dirty_buckets: HashSet<String>,
-    acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
-}
-
-fn verified_remote_dirty_usage(
+fn verified_remote_dirty_usage_buckets(
     expected_peers: &HashMap<String, ScannerPeerDirtyUsageExpectation>,
     peer_snapshots: Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)>,
-) -> Option<VerifiedRemoteDirtyUsage> {
+) -> Option<HashSet<String>> {
     if expected_peers.is_empty() || peer_snapshots.len() != expected_peers.len() {
         return None;
     }
 
     let mut received_peers = HashSet::with_capacity(peer_snapshots.len());
     let mut dirty_buckets = HashSet::new();
-    let mut acknowledgements = Vec::new();
     for (host, snapshot) in peer_snapshots {
         let expected = expected_peers.get(&host)?;
-        if !received_peers.insert(host.clone())
+        if !received_peers.insert(host)
             || snapshot.instance_id != expected.instance_id
             || snapshot.generation != expected.generation
             || snapshot.generation == u64::MAX
@@ -178,44 +171,10 @@ fn verified_remote_dirty_usage(
         {
             return None;
         }
-        let entries = snapshot
-            .buckets
-            .iter()
-            .map(|(bucket, state)| crate::storage_api::EcstoreScannerScopedDirtyUsageAckEntry {
-                bucket: bucket.clone(),
-                bucket_incarnation: state.bucket_incarnation,
-                generation: state.generation,
-            })
-            .collect::<Vec<_>>();
-        dirty_buckets.extend(snapshot.buckets.keys().cloned());
-        if !entries.is_empty() {
-            acknowledgements.push(crate::scanner::ScannerDirtyUsageAcknowledgement {
-                host,
-                instance_id: snapshot.instance_id,
-                kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped {
-                    owner_id: snapshot.owner_id,
-                    entries,
-                },
-            });
-        }
+        dirty_buckets.extend(snapshot.buckets.into_keys());
     }
 
-    (received_peers.len() == expected_peers.len()).then_some(VerifiedRemoteDirtyUsage {
-        dirty_buckets,
-        acknowledgements,
-    })
-}
-
-fn scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(
-    acknowledgements: &[crate::scanner::ScannerDirtyUsageAcknowledgement],
-) -> bool {
-    acknowledgements.iter().any(|acknowledgement| {
-        matches!(
-            &acknowledgement.kind,
-            crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { entries, .. }
-                if entries.len() > crate::SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES
-        )
-    })
+    (received_peers.len() == expected_peers.len()).then_some(dirty_buckets)
 }
 
 fn complete_scanner_cache_snapshot_plan_digest(

crates/scanner/src/scanner_io/io_cycle.rs

@@ -107,32 +107,23 @@ struct ScannerBucketScopeResolution<'a> {
     requires_full_scan: bool,
 }
 
-struct ScannerBucketScopeResolutionResult {
-    scope: ScannerBucketScanScope,
-    remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
-}
-
 async fn resolve_scanner_bucket_scan_scope<S>(
     store: &S,
     distributed: bool,
     resolution: ScannerBucketScopeResolution<'_>,
-) -> ScannerBucketScopeResolutionResult
+) -> ScannerBucketScanScope
 where
     S: ScannerStorage,
 {
-    let default_result = |scope: ScannerBucketScanScope| ScannerBucketScopeResolutionResult {
-        scope,
-        remote_dirty_usage_acknowledgements: Vec::new(),
-    };
     if resolution.requires_full_scan {
-        return default_result(ScannerBucketScanScope::default());
+        return ScannerBucketScanScope::default();
     }
     if !resolution.requested_scope.is_default()
         || !resolution.dirty_usage_snapshot.covers_all_pending
         || resolution.dirty_usage_snapshot.generation == u64::MAX
         || resolution.dirty_usage_snapshot.buckets.len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES
     {
-        return default_result(resolution.requested_scope);
+        return resolution.requested_scope;
     }
 
     let mut dirty_buckets = resolution
@@ -143,10 +134,10 @@ where
         .collect::<HashSet<_>>();
     if distributed {
         let Some(notification_system) = store.scanner_notification_system() else {
-            return default_result(resolution.requested_scope);
+            return resolution.requested_scope;
         };
         let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else {
-            return default_result(resolution.requested_scope);
+            return resolution.requested_scope;
         };
         let mut expected_peers = HashMap::new();
         for (host, lease_instance_id, _) in crate::scanner::scanner_activity_publication_lease_targets(resolution.activity_before)
@@ -154,10 +145,10 @@ where
             let Some((activity_instance_id, generation, pending)) =
                 crate::scanner::scanner_activity_dirty_usage_state_for_host(resolution.activity_before, &host)
             else {
-                return default_result(resolution.requested_scope);
+                return resolution.requested_scope;
             };
             if activity_instance_id != lease_instance_id || expected_peers.contains_key(&host) {
-                return default_result(resolution.requested_scope);
+                return resolution.requested_scope;
             }
             expected_peers.insert(
                 host,
@@ -168,76 +159,19 @@ where
                 },
             );
         }
-        let Some(remote_dirty_usage) = verified_remote_dirty_usage(&expected_peers, peer_snapshots) else {
-            return default_result(resolution.requested_scope);
-        };
-        dirty_buckets.extend(remote_dirty_usage.dirty_buckets);
-        let scope = scoped_scan_scope_from_dirty_buckets(
-            resolution.requested_scope,
-            dirty_buckets,
-            true,
-            resolution.all_buckets,
-            resolution.baseline_proof,
-        );
-        if scope.is_default() {
-            return default_result(scope);
-        }
-        let Some(selected_buckets) = scope.selected_buckets.as_ref() else {
-            return default_result(scope);
-        };
-        let mut scoped_acknowledgements = Vec::with_capacity(remote_dirty_usage.acknowledgements.len());
-        for acknowledgement in remote_dirty_usage.acknowledgements {
-            let crate::scanner::ScannerDirtyUsageAcknowledgement {
-                host,
-                instance_id,
-                kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries },
-            } = acknowledgement
-            else {
-                return default_result(scope);
-            };
-            let entries = entries
-                .into_iter()
-                .filter(|entry| selected_buckets.contains(&entry.bucket))
-                .collect::<Vec<_>>();
-            if !entries.is_empty() {
-                scoped_acknowledgements.push(crate::scanner::ScannerDirtyUsageAcknowledgement {
-                    host,
-                    instance_id,
-                    kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped { owner_id, entries },
-                });
-            }
-        }
-        if super::scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&scoped_acknowledgements) {
-            return default_result(ScannerBucketScanScope::default());
-        }
-        if !scoped_acknowledgements.is_empty() {
-            let capability_acknowledgements = scoped_acknowledgements
-                .clone()
-                .into_iter()
-                .map(Into::into)
-                .collect::<Vec<crate::storage_api::EcstoreScannerDirtyUsageAcknowledgement>>();
-            if !matches!(
-                notification_system
-                    .scanner_scoped_dirty_usage_capabilities(capability_acknowledgements)
-                    .await,
-                Ok(true)
-            ) {
-                return default_result(ScannerBucketScanScope::default());
-            }
-        }
-        return ScannerBucketScopeResolutionResult {
-            scope,
-            remote_dirty_usage_acknowledgements: scoped_acknowledgements,
+        let Some(remote_dirty_buckets) = verified_remote_dirty_usage_buckets(&expected_peers, peer_snapshots) else {
+            return resolution.requested_scope;
         };
+        dirty_buckets.extend(remote_dirty_buckets);
     }
 
-    default_result(scoped_scan_scope_from_dirty_buckets(
+    scoped_scan_scope_from_dirty_buckets(
         resolution.requested_scope,
         dirty_buckets,
         true,
         resolution.all_buckets,
         resolution.baseline_proof,
-    ))
+    )
 }
 
 pub(crate) async fn nsscanner_with_storage_status_scoped<S>(store: &S, request: ScannerCycleRequest) -> Result<ScannerCycleResult>
@@ -360,7 +294,7 @@ where
     let bucket_coverage_digest = scanner_bucket_plan_digest(&all_buckets, activity_digest);
     let execution_digest = scanner_bucket_work_digest(bucket_coverage_digest, scan_mode, requires_full_scan);
     let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list));
-    let scope_resolution = resolve_scanner_bucket_scan_scope(
+    let scan_scope = resolve_scanner_bucket_scan_scope(
         store,
         distributed,
         ScannerBucketScopeResolution {
@@ -380,8 +314,6 @@ where
         },
     )
     .await;
-    let remote_dirty_usage_acknowledgements = scope_resolution.remote_dirty_usage_acknowledgements;
-    let scan_scope = scope_resolution.scope;
     #[cfg(test)]
     if let Some(observer) = resolved_scope_observer {
         let _ = observer.send(scan_scope.clone());
@@ -742,14 +674,11 @@ 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()
-        };
+    let remote_dirty_usage_acknowledgements = if cycle_status == ScannerCycleStatus::Complete {
+        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)

crates/scanner/src/scanner_io/tests.rs

@@ -1561,7 +1561,6 @@ fn peer_dirty_usage_snapshot(
     buckets: &[(&str, u64)],
 ) -> EcstoreScannerPeerDirtyUsageSnapshot {
     EcstoreScannerPeerDirtyUsageSnapshot {
-        owner_id: uuid::Uuid::from_u128(0x11111111111111111111111111111111).to_string(),
         instance_id: instance_id.to_string(),
         generation,
         pending_bucket_count: u64::try_from(buckets.len()).expect("test bucket count should fit"),
@@ -1569,15 +1568,7 @@ fn peer_dirty_usage_snapshot(
         complete,
         buckets: buckets
             .iter()
-            .map(|(bucket, generation)| {
-                (
-                    (*bucket).to_string(),
-                    crate::storage_api::EcstoreScannerPeerDirtyUsageBucket {
-                        bucket_incarnation: uuid::Uuid::from_u128(0x22222222222222222222222222222222),
-                        generation: *generation,
-                    },
-                )
-            })
+            .map(|(bucket, generation)| ((*bucket).to_string(), *generation))
             .collect(),
     }
 }
@@ -1604,7 +1595,7 @@ fn verified_remote_dirty_usage_buckets_merges_only_complete_current_snapshots()
     ]);
 
     assert_eq!(
-        verified_remote_dirty_usage(
+        verified_remote_dirty_usage_buckets(
             &expected_peers,
             vec![
                 (
@@ -1617,63 +1608,10 @@ fn verified_remote_dirty_usage_buckets_merges_only_complete_current_snapshots()
                 ),
             ],
         ),
-        Some(VerifiedRemoteDirtyUsage {
-            dirty_buckets: HashSet::from(["photos".to_string(), "archive".to_string()]),
-            acknowledgements: vec![
-                crate::scanner::ScannerDirtyUsageAcknowledgement {
-                    host: "node-a:9000".to_string(),
-                    instance_id: "instance-a".to_string(),
-                    kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped {
-                        owner_id: uuid::Uuid::from_u128(0x11111111111111111111111111111111).to_string(),
-                        entries: vec![crate::storage_api::EcstoreScannerScopedDirtyUsageAckEntry {
-                            bucket: "photos".to_string(),
-                            bucket_incarnation: uuid::Uuid::from_u128(0x22222222222222222222222222222222),
-                            generation: 7,
-                        }],
-                    },
-                },
-                crate::scanner::ScannerDirtyUsageAcknowledgement {
-                    host: "node-b:9000".to_string(),
-                    instance_id: "instance-b".to_string(),
-                    kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped {
-                        owner_id: uuid::Uuid::from_u128(0x11111111111111111111111111111111).to_string(),
-                        entries: vec![crate::storage_api::EcstoreScannerScopedDirtyUsageAckEntry {
-                            bucket: "archive".to_string(),
-                            bucket_incarnation: uuid::Uuid::from_u128(0x22222222222222222222222222222222),
-                            generation: 3,
-                        }],
-                    },
-                },
-            ],
-        })
+        Some(HashSet::from(["photos".to_string(), "archive".to_string()]))
     );
 }
 
-#[test]
-fn scanner_scoped_dirty_usage_ack_cost_threshold_is_single_protocol_batch() {
-    let acknowledgement = |entry_count: usize| crate::scanner::ScannerDirtyUsageAcknowledgement {
-        host: "node-a:9000".to_string(),
-        instance_id: "instance-a".to_string(),
-        kind: crate::scanner::ScannerDirtyUsageAcknowledgementKind::Scoped {
-            owner_id: uuid::Uuid::from_u128(0x11111111111111111111111111111111).to_string(),
-            entries: (0..entry_count)
-                .map(|index| crate::storage_api::EcstoreScannerScopedDirtyUsageAckEntry {
-                    bucket: format!("bucket-{index:02}"),
-                    bucket_incarnation: uuid::Uuid::from_u128(0x22222222222222222222222222222222),
-                    generation: 7,
-                })
-                .collect(),
-        },
-    };
-
-    assert!(!scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&[acknowledgement(
-        crate::SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES
-    )]));
-    assert!(scanner_scoped_dirty_usage_ack_exceeds_cost_threshold(&[acknowledgement(
-        crate::SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES + 1
-    )]));
-}
-
 #[test]
 fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state() {
     let expected_peers = HashMap::from([(
@@ -1692,7 +1630,7 @@ fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state()
         peer_dirty_usage_snapshot("instance-a", 7, true, &[]),
     ] {
         assert!(
-            verified_remote_dirty_usage(&expected_peers, vec![("node-a:9000".to_string(), snapshot)]).is_none(),
+            verified_remote_dirty_usage_buckets(&expected_peers, vec![("node-a:9000".to_string(), snapshot)]).is_none(),
             "incomplete, stale, mismatched, or empty pending peer state must fall back to a full scan"
         );
     }

crates/scanner/src/storage_api.rs

@@ -103,13 +103,8 @@ 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,
-    ScannerPeerDirtyUsageSnapshot as EcstoreScannerPeerDirtyUsageSnapshot,
-    ScannerScopedDirtyUsageAckEntry as EcstoreScannerScopedDirtyUsageAckEntry,
+    ScannerBucketListing as EcstoreScannerBucketListing, ScannerPeerDirtyUsageSnapshot as EcstoreScannerPeerDirtyUsageSnapshot,
 };
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::runtime::InstanceContext as EcstoreInstanceContext;
@@ -320,7 +315,6 @@ pub(crate) mod scan {
     pub use super::storage_contracts::{
         SCANNER_ACTIVITY_PROTOCOL_VERSION, SCANNER_ACTIVITY_V6_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES,
         SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE,
-        SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES,
     };
 }
 

crates/storage-api/src/lib.rs

@@ -56,9 +56,8 @@ pub const SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION: u32 = 5;
 /// state is not authenticated by that version.
 pub const SCANNER_ACTIVITY_V6_PROTOCOL_VERSION: u32 = 6;
 pub const SCANNER_ACTIVITY_PROTOCOL_VERSION: u32 = 7;
-pub const SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION: u32 = 2;
-pub const SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES: usize = 32;
-pub const SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES: usize = SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES;
+pub const SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION: u32 = 1;
+pub const SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES: usize = 4096;
 pub const SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE: usize = 512 * 1024;
 
 #[derive(Debug, serde::Deserialize, serde::Serialize)]

docs/architecture/s3-compatibility-matrix.md

@@ -62,6 +62,7 @@ Object keys are stored as file-system paths under each drive (`{drive}/{bucket}/
 | Behavior | RustFS | AWS S3 | Why |
 |---|---|---|---|
 | Object key with a `.` or `..` path segment, or an empty segment (`//`), such as `a//b/./c/../d` | `400 InvalidArgument` (`check_object_args` in `crates/ecstore/src/bucket/utils.rs`, mirroring MinIO `IsValidObjectPrefix`) | Accepted as an opaque key | A `..` segment would resolve to a parent directory and `.`/`//` segments would alias other keys on disk; encoding them would change the MinIO-compatible on-disk format. |
+| Directory marker (key ending in `/`, with or without a body) in a versioned bucket | Stored as the null version: `PutObject`/`HeadObject` report version id `00000000-0000-0000-0000-000000000000`, `ListObjectVersions` reports `null`, and a later PUT of the same key overwrites in place (`put_opts` in `rustfs/src/storage/options.rs`, mirroring MinIO `putOpts`: "for directory objects skip creating new versions") | A real version id per PUT, with a version history | The marker only exists to make an empty prefix listable; keeping a history for it would leave hidden versions behind every prefix delete. Replication still copies the marker as its null version (`test_bucket_replication_replicates_directory_marker_in_versioned_bucket` in `crates/e2e_test/src/replication_extension_test.rs`). |
 
 ## Update Rule
 

docs/operations/replication-outbound-transport.md

@@ -6,7 +6,7 @@
 ## What a replication PUT carries by default
 
 - A plain signed body with an exact `Content-Length`. The SDK does not add a streaming trailer checksum, so the body is never wrapped in `aws-chunked` framing (rustfs#6853: a target that does not decode that framing stored the frames verbatim while RustFS recorded COMPLETED).
-- Any object-level checksum the source object was uploaded with, forwarded as its `x-amz-checksum-*` header.
+- For a single-part object, the checksum the source object was uploaded with, forwarded as its `x-amz-checksum-<algorithm>` header (the value the source verified on upload). A multipart replica is rebuilt through CreateMultipartUpload/UploadPart and carries no object-level checksum header. Managed-SSE objects forward none.
 - On a PUT that carries Object Lock parameters and no forwarded checksum: `Content-MD5` derived from the source ETag, or an SDK CRC32 checksum when the ETag is not the MD5 of the wire bytes (rustfs#7082).
 - The source ETag, mtime and version id on `x-rustfs-source-*` headers (with `x-minio-source-*` twins), and the Object Lock mode, retain-until date and legal hold of the source version when present.
 - After the PUT, the target's ETag is compared with the source ETag when both are plain single-part MD5s; a mismatch fails the replication instead of reporting a corrupted replica as COMPLETED.
@@ -17,6 +17,7 @@
 | --- | --- | --- |
 | Rejects or mis-stores `aws-chunked` bodies (SeaweedFS 3.97) | Handled by the plain-payload default above. | Outbound target matrix, `RejectAwsChunked` mode |
 | Requires `Content-MD5` or `x-amz-checksum-*` on a PutObject with Object Lock parameters (AWS S3, MinIO, Impossible Cloud, most compatible stores) | Satisfied: a locked single PUT carries `Content-MD5` derived from the source ETag (plaintext objects whose ETag is the MD5 of the wire bytes) or an SDK CRC32 checksum (multipart-layout ETags, managed SSE, SSE-C passthrough — this one is an `aws-chunked` trailer, so a target that also rejects that framing cannot take such objects). Releases before this fix (`1.0.0-rc.5`) need `RUSTFS_REPLICATION_STREAMING_CHECKSUMS=true` as a workaround. | Outbound target matrix, `RequireChecksumWithObjectLock` mode |
+| Stores `x-amz-checksum-*` from a PutObject and returns it on `HEAD ?ChecksumMode=ENABLED` (AWS S3, Wasabi, RustFS) | Satisfied for single-part objects: the replica answers with the source's checksum. Before this fix (`1.0.0-rc.5`) the checksum left the source as `x-amz-meta-<algorithm>` user metadata and no replica carried it (rustfs/backlog#2340). | Outbound target matrix, `Checksummed` shape |
 | Mints its own version ids (AWS S3, Wasabi, Impossible Cloud) | Data lands; version-addressed convergence does not. See rustfs/backlog#2085 and `docs/operations/replication-check.md` (VersionFidelity). | `replication-check`, outbound target matrix, `MintOwnVersionIds` mode |
 | Returns an ETag that is not the content MD5 without announcing SSE | Every single-part object fails ETag verification. Set `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY=false`. | Replication status FAILED with `replica etag mismatch` |
 
@@ -24,7 +25,7 @@
 
 | Variable | Default | Meaning |
 | --- | --- | --- |
-| `RUSTFS_REPLICATION_STREAMING_CHECKSUMS` | unset (plain payloads) | `true` or `1` restores SDK trailer checksums (`RequestChecksumCalculation::WhenSupported`). Every streaming upload is then `aws-chunked` with an `x-amz-trailer`; use only when every target decodes that framing. |
+| `RUSTFS_REPLICATION_STREAMING_CHECKSUMS` | unset (plain payloads) | `true` or `1` restores SDK trailer checksums (`RequestChecksumCalculation::WhenSupported`). Every streaming upload is then `aws-chunked` with an `x-amz-trailer`, except a single-part PUT that forwards the source's `x-amz-checksum-*` header, which is sent plain so the target does not receive a second algorithm; use only when every target decodes that framing. |
 | `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY` | enabled | `false` or `0` disables the post-PUT ETag comparison for targets whose 32-hex ETags are legitimately not the content MD5. |
 
 Both knobs are read by the RustFS process that owns the replication target, at client build time; restart the server after changing them.

rustfs/src/server/http.rs

@@ -2418,7 +2418,6 @@ fn get_default_tcp_keepalive() -> TcpKeepalive {
 mod tests {
     use super::*;
     use crate::server::compress::RequestPathCategory;
-    use crate::storage_api::server::http::ScannerScopedDirtyUsageAckEntry;
     use bytes::Bytes;
     use http::Request as HttpRequest;
     use http::{HeaderMap, StatusCode};
@@ -3455,9 +3454,9 @@ mod tests {
             .scanner_scoped_dirty_usage_capability(
                 "11111111-1111-1111-1111-111111111111".to_string(),
                 "a".repeat(32),
-                vec![ScannerScopedDirtyUsageAckEntry {
-                    bucket: "photos".to_string(),
-                    bucket_incarnation: uuid::Uuid::from_u128(0x11111111111111111111111111111111),
+                vec![rustfs_protos::proto_gen::node_service::ScannerScopedDirtyUsageEntry {
+                    bucket: "photos".into(),
+                    bucket_incarnation: vec![1; 16].into(),
                     generation: 8,
                 }],
             )

rustfs/src/storage/rpc/node_service.rs

@@ -303,38 +303,25 @@ fn scanner_activity_response(
     }
 }
 
-async fn scanner_dirty_usage_snapshot_response(
-    store: &ECStore,
+fn scanner_dirty_usage_snapshot_response(
     snapshot: rustfs_scanner::ScannerDirtyUsageSnapshot,
-) -> Result<ScannerDirtyUsageSnapshotResponse, Status> {
-    if store.id.is_nil() {
-        return Err(Status::failed_precondition("scanner dirty usage snapshot owner is unavailable"));
-    }
-    let mut buckets = Vec::with_capacity(snapshot.buckets.len());
-    for bucket in snapshot.buckets {
-        let bucket_incarnation = store
-            .bucket_incarnation_id_from_disk(&bucket.bucket)
-            .await
-            .map_err(|_| Status::failed_precondition("scanner dirty usage bucket incarnation is unavailable"))?;
-        if bucket_incarnation.is_nil() {
-            return Err(Status::failed_precondition("scanner dirty usage bucket incarnation is unavailable"));
-        }
-        buckets.push(ScannerDirtyUsageBucket {
-            bucket: bucket.bucket,
-            generation: bucket.generation,
-            bucket_incarnation: bucket_incarnation.as_bytes().to_vec().into(),
-        });
-    }
-    Ok(ScannerDirtyUsageSnapshotResponse {
+) -> ScannerDirtyUsageSnapshotResponse {
+    ScannerDirtyUsageSnapshotResponse {
         instance_id: rustfs_scanner::scanner_activity_epoch().to_string(),
         generation: snapshot.generation,
         pending_bucket_count: snapshot.pending_bucket_count,
         protocol_version: rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION,
         complete: snapshot.complete,
-        buckets,
+        buckets: snapshot
+            .buckets
+            .into_iter()
+            .map(|bucket| ScannerDirtyUsageBucket {
+                bucket: bucket.bucket,
+                generation: bucket.generation,
+            })
+            .collect(),
         response_proof: Bytes::new(),
-        owner_id: store.id.to_string(),
-    })
+    }
 }
 
 fn scanner_activity_response_v7(
@@ -2232,14 +2219,11 @@ impl Node for NodeService {
             .as_ref()
             .try_into()
             .map_err(|_| Status::invalid_argument("scanner dirty usage snapshot challenge must be 16 bytes"))?;
-        let store = self
-            .resolve_object_store()
-            .ok_or_else(|| Status::unavailable("storage layer is not initialized"))?;
-        let snapshot = rustfs_scanner::scanner_dirty_usage_snapshot(rustfs_scanner::SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES);
+        let snapshot = rustfs_scanner::scanner_dirty_usage_snapshot(rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES);
         if snapshot.generation == u64::MAX {
             return Err(Status::resource_exhausted("scanner dirty usage generation is exhausted"));
         }
-        let mut response = scanner_dirty_usage_snapshot_response(&store, snapshot).await?;
+        let mut response = scanner_dirty_usage_snapshot_response(snapshot);
         let canonical = rustfs_protos::canonical_scanner_dirty_usage_snapshot_response_body(&challenge, &response)
             .map_err(|_| Status::internal("scanner dirty usage snapshot response is too large to authenticate"))?;
         response.response_proof = sign_tonic_rpc_response_proof(&canonical)
@@ -6526,28 +6510,7 @@ mod tests {
     #[tokio::test]
     async fn test_scanner_dirty_usage_snapshot_requires_body_bound_auth_and_signs_a_consistent_view() {
         let _ = rustfs_credentials::set_global_rpc_secret("scanner-dirty-usage-snapshot-test-secret".to_string());
-        let _ = rustfs_credentials::init_global_action_credentials(
-            Some("TESTROOTACCESSKEY".to_string()),
-            Some("TESTROOTSECRET123".to_string()),
-        );
-        let temp_dir = tempfile::tempdir().expect("scanner dirty usage snapshot RPC test directory");
-        let env = rustfs_test_utils::TestECStoreEnv::builder()
-            .base_dir(temp_dir.path())
-            .build()
-            .await;
-        ObjectStore::new(Arc::clone(&env.ecstore))
-            .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX))
-            .await
-            .expect("seed IAM format");
-        let iam = rustfs_iam::build_iam_sys(Arc::clone(&env.ecstore))
-            .await
-            .expect("build isolated IAM");
-        let context = Arc::new(crate::runtime_sources::AppContext::with_default_interfaces(
-            Arc::clone(&env.ecstore),
-            iam,
-            Arc::new(KmsServiceManager::new()),
-        ));
-        let service = make_server_for_context(Some(context));
+        let service = create_test_node_service();
         let unsigned = service
             .scanner_dirty_usage_snapshot(Request::new(ScannerDirtyUsageSnapshotRequest {
                 challenge: vec![7; 16].into(),
@@ -6601,7 +6564,6 @@ mod tests {
             .into_inner();
         assert_eq!(response.instance_id, rustfs_scanner::scanner_activity_epoch());
         assert_eq!(response.protocol_version, rustfs_scanner::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION);
-        assert!(Uuid::parse_str(&response.owner_id).is_ok_and(|owner_id| !owner_id.is_nil()));
         let bucket_count = u64::try_from(response.buckets.len()).expect("snapshot bucket count should fit in u64");
         assert_eq!(response.complete, response.pending_bucket_count == bucket_count);
         let canonical = rustfs_protos::canonical_scanner_dirty_usage_snapshot_response_body(&challenge, &response)

rustfs/src/storage/storage_api.rs

@@ -550,8 +550,8 @@ pub(crate) mod ecstore_rpc {
     };
     #[cfg(test)]
     pub(crate) use rustfs_ecstore::api::rpc::{
-        ScannerScopedDirtyUsageAckEntry, build_put_file_auth_trailer, gen_signature_headers, gen_tonic_signature_headers,
-        set_tonic_canonical_body_digest, verify_put_file_capability, verify_tonic_rpc_response_proof,
+        build_put_file_auth_trailer, gen_signature_headers, gen_tonic_signature_headers, set_tonic_canonical_body_digest,
+        verify_put_file_capability, verify_tonic_rpc_response_proof,
     };
 }
 
@@ -694,8 +694,6 @@ pub(crate) type ServerContextSlot = crate::storage::runtime_sources::ServerConte
 pub(crate) type LocalPeerS3Client = ecstore_rpc::LocalPeerS3Client;
 #[cfg(test)]
 pub(crate) type PeerRestClient = ecstore_rpc::PeerRestClient;
-#[cfg(test)]
-pub(crate) type ScannerScopedDirtyUsageAckEntry = ecstore_rpc::ScannerScopedDirtyUsageAckEntry;
 pub(crate) type MetricType = ecstore_metrics::MetricType;
 pub(crate) type ObjectPartInfo = rustfs_filemeta::ObjectPartInfo;
 pub(crate) type ObjectLockBlockReason = ecstore_bucket::object_lock::objectlock_sys::ObjectLockBlockReason;

rustfs/src/storage_api.rs

@@ -147,8 +147,8 @@ pub(crate) mod server {
 
         #[cfg(test)]
         pub(crate) use crate::storage::storage_api::{
-            Endpoint, EndpointServerPools, Endpoints, PeerRestClient, PoolEndpoints, ScannerScopedDirtyUsageAckEntry,
-            gen_signature_headers, gen_tonic_signature_headers,
+            Endpoint, EndpointServerPools, Endpoints, PeerRestClient, PoolEndpoints, gen_signature_headers,
+            gen_tonic_signature_headers,
         };
 
         pub(crate) mod ecfs {

scripts/error-other-format-baseline.txt

@@ -45,7 +45,7 @@
 1|crates/ecstore/src/object_api/types.rs
 3|crates/ecstore/src/runtime/sources.rs
 4|crates/ecstore/src/services/batch_processor.rs
-13|crates/ecstore/src/services/notification_sys.rs
+14|crates/ecstore/src/services/notification_sys.rs
 16|crates/ecstore/src/services/rebalance/control.rs
 1|crates/ecstore/src/services/rebalance/entry.rs
 8|crates/ecstore/src/services/rebalance/meta.rs