feat(scanner): send scoped dirty usage acknowledgements (#7322)

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

@@ -1257,6 +1257,8 @@ 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 {
@@ -1282,6 +1284,8 @@ 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,11 +575,13 @@ 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())
 }
@@ -1842,13 +1844,16 @@ 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");
@@ -1865,6 +1870,9 @@ 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);
@@ -1874,6 +1882,9 @@ 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,6 +885,7 @@ message ScannerActivityResponse {
 message ScannerDirtyUsageBucket {
   string bucket = 1;
   uint64 generation = 2;
+  bytes bucket_incarnation = 3;
 }
 
 message ScannerDirtyUsageSnapshotRequest {
@@ -901,6 +902,7 @@ 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