From f1a084f09abd19d9d4c9710ed651901527ad25c6 Mon Sep 17 00:00:00 2001
From: Matthew Sanetra <41018997+matthewsanetra@users.noreply.github.com>
Date: Mon, 3 Aug 2026 19:27:23 +0100
Subject: [PATCH] Skip stale compaction progress metrics

---
 slatedb/src/compactor.rs | 51 ++++++++++++++++++++++++++--------------
 1 file changed, 34 insertions(+), 17 deletions(-)

diff --git a/slatedb/src/compactor.rs b/slatedb/src/compactor.rs
index 6e45022a9..37762d644 100644
--- a/slatedb/src/compactor.rs
+++ b/slatedb/src/compactor.rs
@@ -629,8 +629,15 @@ impl CompactorEventHandler {
             .active_compactions()
             .filter(|c| c.status() != CompactionStatus::Compacted)
         {
-            let estimated_source_bytes =
-                Self::calculate_estimated_source_bytes(compaction, db_state);
+            let Some(estimated_source_bytes) =
+                Self::calculate_estimated_source_bytes(compaction, db_state)
+            else {
+                warn!(
+                    "skipping progress metrics for compaction {} because its source state is no longer present in the manifest",
+                    compaction.id()
+                );
+                continue;
+            };
             total_estimated_bytes += estimated_source_bytes;
             total_bytes_processed += compaction.bytes_processed();
 
@@ -684,10 +691,11 @@ impl CompactorEventHandler {
     }
 
     /// Calculates the estimated total source bytes for a compaction.
-    fn calculate_estimated_source_bytes(compaction: &Compaction, db_state: &ManifestCore) -> u64 {
-        let tree = db_state
-            .tree_for_segment(compaction.spec().segment())
-            .expect("compaction target segment missing from manifest");
+    fn calculate_estimated_source_bytes(
+        compaction: &Compaction,
+        db_state: &ManifestCore,
+    ) -> Option<u64> {
+        let tree = db_state.tree_for_segment(compaction.spec().segment())?;
 
         let views_by_id: HashMap<Ulid, &SsTableView> =
             tree.l0.iter().map(|view| (view.id, view)).collect();
@@ -698,17 +706,13 @@ impl CompactorEventHandler {
             .spec()
             .sources()
             .iter()
-            .map(|source| match source {
-                SourceId::SstView(id) => views_by_id
-                    .get(id)
-                    .expect("compaction source view not found in L0")
-                    .estimate_size(),
-                SourceId::SortedRun(id) => srs_by_id
-                    .get(id)
-                    .expect("compaction source sorted run not found")
-                    .estimate_size(),
+            .try_fold(0_u64, |total, source| {
+                let source_bytes = match source {
+                    SourceId::SstView(id) => views_by_id.get(id)?.estimate_size(),
+                    SourceId::SortedRun(id) => srs_by_id.get(id)?.estimate_size(),
+                };
+                Some(total + source_bytes)
             })
-            .sum()
     }
 
     /// Handles a polling tick by refreshing compactions and the manifest, then possibly scheduling compactions.
@@ -3747,7 +3751,20 @@ mod tests {
 
         let expected = segment_l0.estimate_size() + segment_sr.estimate_size();
         let actual = CompactorEventHandler::calculate_estimated_source_bytes(&compaction, &core);
-        assert_eq!(actual, expected);
+        assert_eq!(actual, Some(expected));
+    }
+
+    #[test]
+    fn test_calculate_estimated_source_bytes_tolerates_stale_sources() {
+        let missing_source = Ulid::new();
+        let core = ManifestCore::new();
+        let compaction = Compaction::new(
+            Ulid::new(),
+            CompactionSpec::new(vec![SourceId::SstView(missing_source)], 1),
+        );
+
+        let actual = CompactorEventHandler::calculate_estimated_source_bytes(&compaction, &core);
+        assert_eq!(actual, None);
     }
 
     #[tokio::test]