Make completed summary windows immutable after durable flush

data_plane/src/drivers/ingest/otel.rs

@@ -1754,15 +1754,6 @@ async fn route_modified_otlp_sketches_to_precompute(
                         }
                     }
 
-                    ingest_state.sketch_snapshots.insert(
-                        series_key.clone(),
-                        crate::precompute_engine::ingest_handler::SnapshotCacheEntry {
-                            core: accumulator.clone_boxed_core(),
-                            window_start: dp.start_time_unix_nano,
-                        },
-                    );
-                    ingest_state.note_window_and_sweep(dp.start_time_unix_nano);
-
                     use crate::storage_engines::sketch_db::index::{
                         SketchEncoding, SketchSampleState,
                     };
@@ -1778,15 +1769,26 @@ async fn route_modified_otlp_sketches_to_precompute(
                     );
                     let encoding =
                         encoding_to_handle(dp.encoding).unwrap_or(SketchEncoding::ProtoFull);
-                    ingest_state.sketch_index.append_sample(
+                    if !ingest_state.sketch_index.append_sample(
                         sid,
                         label_values,
                         window,
                         SketchSampleState {
                             bytes: dp.sketch.clone(),
                             encoding,
                         },
+                    ) {
+                        return Err("summary window is immutable after completion".into());
+                    }
+
+                    ingest_state.sketch_snapshots.insert(
+                        series_key.clone(),
+                        crate::precompute_engine::ingest_handler::SnapshotCacheEntry {
+                            core: accumulator.clone_boxed_core(),
+                            window_start: dp.start_time_unix_nano,
+                        },
                     );
+                    ingest_state.note_window_and_sweep(dp.start_time_unix_nano);
 
                     // Collect the configs whose metric matches this DP.
                     // Detection is independent of the legacy dual-write

data_plane/src/drivers/ingest/prometheus_remote_write.rs

@@ -199,10 +199,23 @@ impl PrometheusRemoteWriteReceiver {
             .and_then(|plan| plan.precompute_plan.summary_catalog.clone())
             .ok_or("finite completion requires a catalog generation")?;
         self.inner.ingest.router.drain().await?;
-        self.inner
-            .ingest
-            .sketch_index
-            .seal_finite_summary_input(&generation)?;
+        let flush_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30);
+        loop {
+            if self
+                .inner
+                .ingest
+                .sketch_index
+                .seal_finite_summary_input(&generation)?
+            {
+                break;
+            }
+            if tokio::time::Instant::now() >= flush_deadline {
+                return Err(
+                    "finite completion is waiting for durable summary payloads; retry drain".into(),
+                );
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+        }
         trim_process_allocator();
         Ok(())
     }

data_plane/src/storage_engines/sketch_db/index/admission.rs

@@ -22,7 +22,7 @@ pub(super) struct AdmissionInventory {
     replay_floors: BTreeMap<SummaryDefinitionId, i64>,
     observed_extent: Option<HalfOpenTimeRange>,
     finite_complete: bool,
-    published_series: BTreeSet<u64>,
+    published_series: BTreeMap<u64, u64>,
     pending_revisions: usize,
 }
 
@@ -182,7 +182,7 @@ impl AdmissionInventory {
         if self.generation.as_ref() != Some(generation) {
             return Err("summary series publication catalog generation differs".into());
         }
-        if !self.published_series.contains(&series_id)
+        if !self.published_series.contains_key(&series_id)
             && self.published_series.len() >= Self::MAX_WINDOWS
         {
             return Err("summary admission series capacity exceeded".into());
@@ -198,11 +198,16 @@ impl AdmissionInventory {
             return Err("summary coordinate changed series identity".into());
         }
         window.series_id = Some(series_id);
-        self.published_series.insert(series_id);
+        let end = u64::try_from(coordinate.time_range.end_ms)
+            .map_err(|_| "published window end is outside storage timestamp range")?;
+        self.published_series
+            .entry(series_id)
+            .and_modify(|current| *current = (*current).max(end))
+            .or_insert(end);
         Ok(())
     }
 
-    pub(super) fn seal_finite(&mut self, generation: &CatalogGeneration) -> Result<(), String> {
+    pub(super) fn validate_finite(&self, generation: &CatalogGeneration) -> Result<u64, String> {
         if self.generation.as_ref() != Some(generation) {
             return Err("finite completion catalog generation differs".into());
         }
@@ -213,11 +218,15 @@ impl AdmissionInventory {
         {
             return Err("finite source has unpublished summary windows".into());
         }
-        self.finite_complete = true;
-        self.revision = self
-            .revision
+        self.revision
             .checked_add(1)
-            .ok_or("summary admission revision exhausted")?;
+            .ok_or_else(|| "summary admission revision exhausted".into())
+    }
+
+    pub(super) fn seal_finite(&mut self, generation: &CatalogGeneration) -> Result<(), String> {
+        let revision = self.validate_finite(generation)?;
+        self.finite_complete = true;
+        self.revision = revision;
         Ok(())
     }
 
@@ -228,7 +237,7 @@ impl AdmissionInventory {
         range: HalfOpenTimeRange,
     ) -> bool {
         self.finite_complete
-            && self.published_series.contains(&series_id)
+            && self.published_series.contains_key(&series_id)
             && self.observed_extent.is_some_and(|extent| {
                 range.start_ms >= extent.start_ms && range.end_ms <= extent.end_ms
             })
@@ -244,6 +253,10 @@ impl AdmissionInventory {
             })
     }
 
+    pub(super) fn published_frontiers(&self) -> &BTreeMap<u64, u64> {
+        &self.published_series
+    }
+
     pub(super) fn revision(&self) -> u64 {
         self.revision
     }
@@ -358,11 +371,18 @@ mod tests {
         assert!(inventory.has_pending(coordinate.summary_definition_id, coordinate.time_range));
         inventory.retire_completed_before(coordinate.summary_definition_id, 1000);
         assert_eq!(inventory.windows.len(), 1);
+        inventory
+            .record_series(&generation, &coordinate, 42)
+            .unwrap();
         inventory
             .acknowledge(&generation, &coordinate, second)
             .unwrap();
         inventory.retire_completed_before(coordinate.summary_definition_id, 1000);
         assert!(inventory.windows.is_empty());
+        assert_eq!(
+            inventory.published_frontiers().get(&42),
+            Some(&(coordinate.time_range.end_ms as u64))
+        );
     }
 
     #[test]

data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs

@@ -956,6 +956,19 @@ impl<K: Eq + std::hash::Hash + Clone, P> SidStoreData<K, P> {
         }
     }
 
+    /// Whether any hot or queued-for-flush state belongs to this completion prefix.
+    pub(crate) fn contains_window_ending_at_or_before(&self, end_ms: u64) -> bool {
+        self.current_epoch
+            .iter_entries()
+            .any(|(window, _, _)| window.1 <= end_ms)
+            || self.sealed_epochs.values().any(|epoch| {
+                epoch
+                    .entries
+                    .iter()
+                    .any(|(window, _, _)| window.1 <= end_ms)
+            })
+    }
+
     /// Time-driven seal for the persistence tier: roll every window in
     /// `current_epoch` whose END is at or before `cutoff_end` into a
     /// freshly-sealed epoch, leaving the more-recent windows in
@@ -1372,6 +1385,21 @@ mod tests {
         assert_eq!(s.current_epoch.distinct_windows(), 0);
     }
 
+    #[test]
+    fn completion_prefix_does_not_wait_for_future_windows() {
+        let mut store = SidStoreData::<Vec<String>, i32>::new();
+        store.insert((30, 60), vec![], 1);
+        assert!(!store.contains_window_ending_at_or_before(30));
+        store.insert((0, 30), vec![], 2);
+        assert!(store.contains_window_ending_at_or_before(30));
+        store.persistence_enabled = true;
+        store.seal_aged_windows(31);
+        assert!(store.contains_window_ending_at_or_before(30));
+        store.sealed_epochs.clear();
+        assert!(!store.contains_window_ending_at_or_before(30));
+        assert!(store.contains_window_ending_at_or_before(60));
+    }
+
     #[test]
     fn seal_aged_windows_noop_when_persistence_disabled() {
         let mut s = SidStoreData::<String, u32>::new();