fix(scanner): preserve default usage cache wire format (#6477)

crates/scanner/src/data_usage_define.rs

@@ -474,8 +474,14 @@ impl Serialize for DataUsageCacheInfo {
     {
         // Keep this metadata map-encoded so older readers can ignore fields
         // appended by newer scanner versions during rolling upgrades.
-        let field_count =
+        let field_count = 16
-            21 + usize::from(self.tier_registry_generation.is_some()) + usize::from(!self.size_reconciliation.is_empty());
+            + usize::from(self.tier_registry_generation.is_some())
+            + usize::from(!self.size_reconciliation.is_empty())
+            + usize::from(self.lkg_snapshot_complete)
+            + usize::from(self.lkg_next_cycle.is_some())
+            + usize::from(self.lkg_last_update.is_some())
+            + usize::from(self.lkg_leader_epoch.is_some())
+            + usize::from(self.lkg_scan_plan_digest.is_some());
         let mut state = serializer.serialize_map(Some(field_count))?;
         state.serialize_entry("name", &self.name)?;
         state.serialize_entry("next_cycle", &self.next_cycle)?;
@@ -499,11 +505,21 @@ impl Serialize for DataUsageCacheInfo {
         if !self.size_reconciliation.is_empty() {
             state.serialize_entry("size_reconciliation", &self.size_reconciliation)?;
         }
-        state.serialize_entry("lkg_snapshot_complete", &self.lkg_snapshot_complete)?;
+        if self.lkg_snapshot_complete {
-        state.serialize_entry("lkg_next_cycle", &self.lkg_next_cycle)?;
+            state.serialize_entry("lkg_snapshot_complete", &true)?;
-        state.serialize_entry("lkg_last_update", &self.lkg_last_update)?;
+        }
-        state.serialize_entry("lkg_leader_epoch", &self.lkg_leader_epoch)?;
+        if let Some(next_cycle) = self.lkg_next_cycle {
-        state.serialize_entry("lkg_scan_plan_digest", &self.lkg_scan_plan_digest)?;
+            state.serialize_entry("lkg_next_cycle", &next_cycle)?;
+        }
+        if let Some(last_update) = self.lkg_last_update {
+            state.serialize_entry("lkg_last_update", &last_update)?;
+        }
+        if let Some(leader_epoch) = self.lkg_leader_epoch {
+            state.serialize_entry("lkg_leader_epoch", &leader_epoch)?;
+        }
+        if let Some(scan_plan_digest) = self.lkg_scan_plan_digest {
+            state.serialize_entry("lkg_scan_plan_digest", &scan_plan_digest)?;
+        }
         state.end()
     }
 }

crates/scanner/src/data_usage_define/tests.rs

@@ -1326,6 +1326,24 @@ fn usage_cache_wire_format_is_pinned() {
     );
 }
 
+#[test]
+fn usage_cache_lkg_fields_round_trip_when_present() {
+    let mut cache = wire_fixture_cache();
+    cache.info.lkg_snapshot_complete = true;
+    cache.info.lkg_next_cycle = Some(6);
+    cache.info.lkg_last_update = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_699_999_999));
+    cache.info.lkg_leader_epoch = Some(8);
+    cache.info.lkg_scan_plan_digest = Some(DataUsageScanPlanDigest([2; 32]));
+
+    let encoded = cache.marshal_msg().expect("marshal cache with LKG metadata");
+    let decoded = DataUsageCache::unmarshal(&encoded).expect("decode cache with LKG metadata");
+    assert!(decoded.info.lkg_snapshot_complete);
+    assert_eq!(decoded.info.lkg_next_cycle, Some(6));
+    assert_eq!(decoded.info.lkg_last_update, cache.info.lkg_last_update);
+    assert_eq!(decoded.info.lkg_leader_epoch, Some(8));
+    assert_eq!(decoded.info.lkg_scan_plan_digest, Some(DataUsageScanPlanDigest([2; 32])));
+}
+
 #[test]
 fn data_usage_cache_prepare_for_scan_rejects_unscoped_distributed_cache() {
     let mut cache = DataUsageCache {