From 911a472c1e8e0d7941b023fb55145ba7408c9c0a Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 06:56:53 -0600 Subject: [PATCH 1/4] fix(storage): make completed summary windows immutable --- .../sketch_db/index/admission.rs | 38 +++-- .../storage_engines/sketch_db/index/mod.rs | 140 +++++++++++++++++- .../sketch_db/persistence/metadata.rs | 10 ++ .../summary-catalog-sds-architecture.md | 20 +++ 4 files changed, 193 insertions(+), 15 deletions(-) diff --git a/data_plane/src/storage_engines/sketch_db/index/admission.rs b/data_plane/src/storage_engines/sketch_db/index/admission.rs index f0d6eceed..19c325108 100644 --- a/data_plane/src/storage_engines/sketch_db/index/admission.rs +++ b/data_plane/src/storage_engines/sketch_db/index/admission.rs @@ -22,7 +22,7 @@ pub(super) struct AdmissionInventory { replay_floors: BTreeMap, observed_extent: Option, finite_complete: bool, - published_series: BTreeSet, + published_series: BTreeMap, 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 { 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 { + &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] diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 72d172468..a7175b2e1 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -605,6 +605,8 @@ impl Drop for StateMutation<'_> { #[derive(Default)] pub struct SketchStore { + /// Held through each state append; completion takes the exclusive guard. + completed_windows: RwLock>, admission: RwLock, mutation_revision: std::sync::atomic::AtomicU64, active_mutations: std::sync::atomic::AtomicUsize, @@ -881,13 +883,43 @@ impl SketchStore { generation: &CatalogGeneration, ) -> Result<(), String> { use std::sync::atomic::Ordering::SeqCst; + // Same order as admitted publication: admission -> metadata -> append fence. + // Closing a receiver alone is insufficient: the fence also rejects writes + // from every other producer once these physical windows are complete. + let mut inventory = self.admission.write().unwrap(); + let frontiers = inventory.published_frontiers().clone(); + let instances = self.instances.read().unwrap(); + let mut records = Vec::new(); + for (sid, end) in &frontiers { + let instance = instances + .get(sid) + .ok_or("completed series has no identity")?; + let mut record = self + .metadata_record(instance) + .ok_or("completed series has no catalog provenance")?; + record.completed_through_ms = record.completed_through_ms.max(Some(*end)); + records.push(record); + } + let mut completed = self.completed_windows.write().unwrap(); let mutation = self.mutation_revision.load(SeqCst); if self.active_mutations.load(SeqCst) != 0 || mutation != self.admitted_mutations.load(SeqCst) { return Err("finite summary completion cannot certify untracked state writes".into()); } - self.admission.write().unwrap().seal_finite(generation)?; + inventory.validate_finite(generation)?; + if let Some(writer) = self.persistence_metadata.read().unwrap().as_ref() { + writer + .upsert_all(&records) + .map_err(|error| error.to_string())?; + } + inventory.seal_finite(generation)?; + for (sid, end) in frontiers { + completed + .entry(sid) + .and_modify(|value| *value = (*value).max(end)) + .or_insert(end); + } self.finite_mutation_revision.store(mutation, SeqCst); Ok(()) } @@ -1206,7 +1238,11 @@ impl SketchStore { series_label_values: BTreeMap, window: TimestampRange, sample: SketchSampleState, - ) { + ) -> bool { + let completed = self.completed_windows.read().unwrap(); + if completed.get(&sid).is_some_and(|end| window.1 <= *end) { + return false; + } let _mutation = self.begin_state_mutation(); let store = self .series @@ -1216,6 +1252,7 @@ impl SketchStore { let mut guard = store.write().unwrap(); guard.insert(window, series_label_values, AggPayload::Sketch(sample)); guard.last_write_unix_ms = now_ms(); + true } /// Build a `SidStoreData` pre-configured for the store's current @@ -1249,7 +1286,11 @@ impl SketchStore { series_label_values: BTreeMap, window: TimestampRange, payload: Box, - ) { + ) -> bool { + let completed = self.completed_windows.read().unwrap(); + if completed.get(&sid).is_some_and(|end| window.1 <= *end) { + return false; + } let _mutation = self.begin_state_mutation(); let max_value = payload .as_any() @@ -1280,6 +1321,7 @@ impl SketchStore { retention_horizon_ms, ); } + true } /// Read a category through the derived in-memory rollup. Returns `None` @@ -2416,6 +2458,7 @@ impl SketchStore { record.summary_definition_id = Some(SummaryDefinitionId::from(m.policy_fp)); record.catalog_generation = Some(Arc::clone(m.catalog_generation.as_ref()?)); } + record.completed_through_ms = self.completed_windows.read().unwrap().get(&m.sid).copied(); record.retired_at_ms = m.retired_at_ms; record.expires_at_ms = m.expires_at_ms; Some(record) @@ -2793,7 +2836,7 @@ impl SketchStore { } let window = (output.start_timestamp, output.end_timestamp); - match crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg) { + let accepted = match crate::storage_engines::sketch_db::data::agg_kind_for_config(agg_cfg) { AggKind::Sketch { .. } => self.append_sample( sid, label_values_map, @@ -2809,8 +2852,8 @@ impl SketchStore { window, accumulator.clone_boxed_core(), ), - } - Some(sid) + }; + accepted.then_some(sid) } /// Phase 5 M2.3.6d — eviction-side helper. Removes every sid in the @@ -3006,6 +3049,14 @@ impl SketchStore { let mut registered = 0usize; for rec in records { + if let Some(end) = rec.completed_through_ms { + self.completed_windows + .write() + .unwrap() + .entry(rec.sid) + .and_modify(|current| *current = (*current).max(end)) + .or_insert(end); + } if rec.removed || rec.expires_at_ms.is_some_and(|expiry| expiry <= now_ms()) { continue; } @@ -4746,6 +4797,83 @@ mod tests { assert_eq!(expired.expires_at_ms, Some(2)); } + #[test] + fn completed_windows_reject_late_updates_after_restart() { + // Completion is a storage admission rule, including legacy producers, + // and survives restart without allowing a correction into consumed state. + let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + serde_json::from_str(include_str!( + "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let plan = snapshot.compile().unwrap(); + let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); + let directory = tempfile::tempdir().unwrap(); + let store = SketchStore::new(); + store + .install_summary_catalog(Arc::new(plan.summary_catalog.clone())) + .unwrap(); + store.register(meta_with_policy(850, fingerprint)); + let generation = store.active_catalog_generation().unwrap(); + let writer = Arc::new(persistence::metadata::SidMetadataStore::new( + directory.path(), + )); + *store.persistence_metadata.write().unwrap() = Some(writer.clone()); + let coordinate = asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id: fingerprint.into(), + time_range: HalfOpenTimeRange { + start_ms: 0, + end_ms: 30_000, + }, + group_values: BTreeMap::new(), + }; + let revision = store + .admit_summary_updates(&generation, [coordinate.clone()].into()) + .unwrap(); + assert!(store.seal_finite_summary_input(&generation).is_err()); + store + .publish_admitted_summary_update( + &generation, + &coordinate, + revision, + revision, + 120_000, + || { + store + .append_sample(850, BTreeMap::new(), (0, 30_000), sample(1)) + .then_some(850) + }, + ) + .unwrap(); + let stale_record = store + .metadata_record(&store.instances.read().unwrap()[&850]) + .unwrap(); + let before_failed_seal = store.summary_update_revision(); + std::fs::create_dir(writer.path()).unwrap(); + assert!(store.seal_finite_summary_input(&generation).is_err()); + assert_eq!(store.summary_update_revision(), before_failed_seal); + assert!(!store.completed_windows.read().unwrap().contains_key(&850)); + std::fs::remove_dir(writer.path()).unwrap(); + store.seal_finite_summary_input(&generation).unwrap(); + assert!(!store.append_sample(850, BTreeMap::new(), (0, 30_000), sample(2))); + assert!(!store.append_precompute( + 850, + BTreeMap::new(), + (0, 30_000), + Box::new(crate::precompute_engine::operators::SumAccumulator::new()) + )); + // A flusher that captured metadata before completion cannot reopen it. + writer.upsert_all(&[stale_record]).unwrap(); + assert_eq!(writer.load().unwrap()[0].completed_through_ms, Some(30_000)); + let restored = SketchStore::new(); + restored + .install_summary_catalog(Arc::new(plan.summary_catalog)) + .unwrap(); + restored.register_recovered_disk_series(directory.path()); + assert!(!restored.append_sample(850, BTreeMap::new(), (0, 30_000), sample(3))); + assert!(restored.append_sample(850, BTreeMap::new(), (30_000, 60_000), sample(4))); + } + #[test] fn failed_durable_lifecycle_write_preserves_live_instance() { let store = SketchStore::new(); diff --git a/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs index 51e6e683d..93caef534 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs @@ -277,6 +277,9 @@ pub struct SidMetaRecord { pub expires_at_ms: Option, #[serde(default)] pub removed: bool, + /// No further publication may change a window ending at or before this bound. + #[serde(default)] + pub completed_through_ms: Option, } impl SidMetaRecord { @@ -301,6 +304,7 @@ impl SidMetaRecord { retired_at_ms: None, expires_at_ms: None, removed: false, + completed_through_ms: None, } } @@ -368,6 +372,8 @@ struct SidBindingRec { expires_at_ms: Option, #[serde(default)] removed: bool, + #[serde(default)] + completed_through_ms: Option, } /// Version-3 normalized sidecar with authoritative catalog provenance. Descriptors appear once and SeriesId bindings hold @@ -438,6 +444,7 @@ impl SdsSidecar { retired_at_ms: record.retired_at_ms, expires_at_ms: record.expires_at_ms, removed: record.removed, + completed_through_ms: record.completed_through_ms, }, ); } @@ -491,6 +498,7 @@ impl SdsSidecar { retired_at_ms: binding.retired_at_ms, expires_at_ms: binding.expires_at_ms, removed: binding.removed, + completed_through_ms: binding.completed_through_ms, }) }) .collect() @@ -608,6 +616,8 @@ impl SidMetadataStore { // Lifecycle is monotone for a SeriesId. An older flush snapshot // must not resurrect a retired or removed persisted instance. next.removed |= existing.removed; + next.completed_through_ms = + existing.completed_through_ms.max(next.completed_through_ms); next.retired_at_ms = existing.retired_at_ms.or(next.retired_at_ms); next.expires_at_ms = match (existing.expires_at_ms, next.expires_at_ms) { (Some(a), Some(b)) => Some(a.min(b)), diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index abc38b97b..0ef4724c6 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -403,3 +403,23 @@ This is an explicit lifetime transition, not cross-generation recovery of arbitr summary state. Legacy records without trustworthy catalog provenance remain unbound. Tombstone reclamation still requires coordinated removal of old physical parts and is not implemented by this transition. + +### Immutable completed windows + +Finite Remote Write completion now fences the SummaryStore append boundary, +not just the receiver queue. After all admitted outputs are published, the store +records the greatest published window end for each physical SeriesId. Sketch and +exact-state writes ending at or before that boundary are rejected, including +writes arriving through other producers. A later window remains writable. + +The boundary is monotone in the existing SeriesId metadata sidecar and is restored +before recovered identities become writable. A stale background metadata flush +cannot reopen a completed window. The guard belongs to the physical lifetime; +a catalog-authorized replacement SeriesId has its own boundary. + +This is an immutability guarantee, not a promise that every payload has reached +disk. Maintenance consumers must separately verify durable state availability and +atomically publish their output identity before claiming replay-safe consumption. +The existing finite-source completeness proof still rejects untracked writes or +pending admitted work. Continuous producer watermarks and derived-state commit +transactions are separate from this finite-input boundary. From 7312e244d56ea39ddaadba3632bfa8cd42ced383 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 06:57:47 -0600 Subject: [PATCH 2/4] test(storage): check unchanged completion revision with snapshot fence --- data_plane/src/storage_engines/sketch_db/index/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index a7175b2e1..014ea8e75 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -4851,7 +4851,7 @@ mod tests { let before_failed_seal = store.summary_update_revision(); std::fs::create_dir(writer.path()).unwrap(); assert!(store.seal_finite_summary_input(&generation).is_err()); - assert_eq!(store.summary_update_revision(), before_failed_seal); + assert!(store.summary_update_revision().matches(before_failed_seal)); assert!(!store.completed_windows.read().unwrap().contains_key(&850)); std::fs::remove_dir(writer.path()).unwrap(); store.seal_finite_summary_input(&generation).unwrap(); From 875486e3c68e47e1f62853f91a522fa35356c8d8 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 07:05:05 -0600 Subject: [PATCH 3/4] fix(storage): flush completed payloads before sealing their windows --- data_plane/src/drivers/ingest/otel.rs | 22 ++-- .../drivers/ingest/prometheus_remote_write.rs | 21 +++- .../sketch_db/index/epoch_columnar.rs | 28 +++++ .../storage_engines/sketch_db/index/mod.rs | 106 +++++++++++++++++- .../sketch_db/persistence/flusher.rs | 10 +- .../sketch_db/persistence/source.rs | 5 + .../summary-catalog-sds-architecture.md | 9 +- 7 files changed, 178 insertions(+), 23 deletions(-) diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 1419df0b6..cc38990ad 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -1753,15 +1753,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, }; @@ -1777,7 +1768,7 @@ 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, @@ -1785,7 +1776,18 @@ async fn route_modified_otlp_sketches_to_precompute( 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 diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index 367ac15f1..3808c9468 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/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(()) } diff --git a/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs b/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs index 4442dfcbc..695c75052 100644 --- a/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs +++ b/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs @@ -956,6 +956,19 @@ impl SidStoreData { } } + /// 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::, 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::::new(); diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 014ea8e75..556c9749c 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -607,6 +607,7 @@ impl Drop for StateMutation<'_> { pub struct SketchStore { /// Held through each state append; completion takes the exclusive guard. completed_windows: RwLock>, + completion_flush_before: std::sync::atomic::AtomicU64, admission: RwLock, mutation_revision: std::sync::atomic::AtomicU64, active_mutations: std::sync::atomic::AtomicUsize, @@ -881,7 +882,7 @@ impl SketchStore { pub(crate) fn seal_finite_summary_input( &self, generation: &CatalogGeneration, - ) -> Result<(), String> { + ) -> Result { use std::sync::atomic::Ordering::SeqCst; // Same order as admitted publication: admission -> metadata -> append fence. // Closing a receiver alone is insufficient: the fence also rejects writes @@ -908,6 +909,24 @@ impl SketchStore { return Err("finite summary completion cannot certify untracked state writes".into()); } inventory.validate_finite(generation)?; + if self.persistence_read.read().unwrap().is_some() { + if let Some(end) = frontiers.values().max() { + self.completion_flush_before + .fetch_max(end.saturating_add(1), SeqCst); + } + // The flusher evicts an epoch only after its payload and manifest + // are durable. Until then a restart must remain able to replay it. + let pending = frontiers.iter().any(|(sid, end)| { + self.series.get(sid).is_some_and(|data| { + data.read() + .unwrap() + .contains_window_ending_at_or_before(*end) + }) + }); + if pending { + return Ok(false); + } + } if let Some(writer) = self.persistence_metadata.read().unwrap().as_ref() { writer .upsert_all(&records) @@ -921,7 +940,7 @@ impl SketchStore { .or_insert(end); } self.finite_mutation_revision.store(mutation, SeqCst); - Ok(()) + Ok(true) } pub(crate) fn summary_window_known_empty( @@ -3154,6 +3173,13 @@ impl SketchStore { // the trait keeps the historical name so the flusher / manifest / // part-writer stay untouched. impl crate::storage_engines::sketch_db::index::persistence::EpochSource for SketchStore { + fn flush_before_ms(&self) -> Option { + let cutoff = self + .completion_flush_before + .load(std::sync::atomic::Ordering::SeqCst); + (cutoff != 0).then_some(cutoff) + } + fn list_sealed_epochs( &self, ) -> Vec { @@ -4874,6 +4900,82 @@ mod tests { assert!(restored.append_sample(850, BTreeMap::new(), (30_000, 60_000), sample(4))); } + #[test] + fn finite_completion_flushes_payload_before_persisting_immutability() { + // With neither memory pressure nor a hot-tier deadline, completion must + // explicitly flush its payload before persisting a non-replayable window. + let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + serde_json::from_str(include_str!( + "../../../../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let plan = snapshot.compile().unwrap(); + let fingerprint = plan.precompute_plan.materializations[0].policy_fingerprint(); + let directory = tempfile::tempdir().unwrap(); + { + let store = Arc::new(SketchStore::new()); + store + .install_summary_catalog(Arc::new(plan.summary_catalog.clone())) + .unwrap(); + store.register(meta_with_policy(851, fingerprint)); + let mut config = durable_cfg(directory.path().to_path_buf()); + config.hot_window_ms = None; + config.seal_window_count = 100; + let mut persistence = store.start_persistence(config).unwrap(); + let generation = store.active_catalog_generation().unwrap(); + let coordinate = asap_types::sds::SummaryInstanceCoordinates { + summary_definition_id: fingerprint.into(), + time_range: HalfOpenTimeRange { + start_ms: 0, + end_ms: 30_000, + }, + group_values: BTreeMap::new(), + }; + let revision = store + .admit_summary_updates(&generation, [coordinate.clone()].into()) + .unwrap(); + store + .publish_admitted_summary_update( + &generation, + &coordinate, + revision, + revision, + 120_000, + || { + store + .append_sample(851, BTreeMap::new(), (0, 30_000), sample(1)) + .then_some(851) + }, + ) + .unwrap(); + assert!(!store.seal_finite_summary_input(&generation).unwrap()); + assert!(!store.completed_windows.read().unwrap().contains_key(&851)); + assert!(wait_until( + || store.seal_finite_summary_input(&generation).unwrap(), + Duration::from_secs(5) + )); + assert!(!persistence.manifest.live_parts().is_empty()); + persistence.shutdown(); + } + let restored = Arc::new(SketchStore::new()); + restored + .install_summary_catalog(Arc::new(plan.summary_catalog)) + .unwrap(); + let _persistence = restored + .start_persistence(durable_cfg(directory.path().to_path_buf())) + .unwrap(); + assert!(!restored.append_sample(851, BTreeMap::new(), (0, 30_000), sample(2))); + let rows = restored.query_range(851, 0, 30_000); + assert_eq!(rows.len(), 1); + let payloads: Vec<_> = rows + .iter() + .flat_map(|row| row.samples.values()) + .flatten() + .collect(); + assert_eq!(payloads.len(), 1); + assert_eq!(payloads[0].bytes, vec![1]); + } + #[test] fn failed_durable_lifecycle_write_preserves_live_instance() { let store = SketchStore::new(); diff --git a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs index c2aad589b..e774b39e6 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/flusher.rs @@ -296,8 +296,11 @@ fn run_tick(shared: &Arc, source: &S) -> PersistR // it becomes flushable this same tick. (Without `hot_window_ms` the // durable tier is purely memory-pressure driven and phase 1 below // handles eviction.) - if let Some(hot) = cfg.hot_window_ms { - let cutoff = now.saturating_sub(hot); + let flush_cutoff = cfg + .hot_window_ms + .map(|hot| now.saturating_sub(hot)) + .max(source.flush_before_ms()); + if let Some(cutoff) = flush_cutoff { source.seal_aged_epochs(cutoff); } @@ -325,8 +328,7 @@ fn run_tick(shared: &Arc, source: &S) -> PersistR } // Phase 2: time watermark (any epoch older than now - hot_window). - if let Some(hot) = cfg.hot_window_ms { - let cutoff = now.saturating_sub(hot); + if let Some(cutoff) = flush_cutoff { for r in &all { if r.end_ts < cutoff && !selected diff --git a/data_plane/src/storage_engines/sketch_db/persistence/source.rs b/data_plane/src/storage_engines/sketch_db/persistence/source.rs index 117c68544..734cfaf6a 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/source.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/source.rs @@ -97,6 +97,11 @@ pub struct EpochSnapshotEntry { /// * [`approx_memory_bytes`] is cheap (atomic load) and is kept in sync /// with what the flusher has evicted. pub trait EpochSource: Send + Sync { + /// Explicit durability demand from a completed source, independent of the hot tier. + fn flush_before_ms(&self) -> Option { + None + } + fn list_sealed_epochs(&self) -> Vec; /// Time-driven seal: roll every un-sealed `current_epoch` window diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 0ef4724c6..4b25499a3 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -417,9 +417,12 @@ before recovered identities become writable. A stale background metadata flush cannot reopen a completed window. The guard belongs to the physical lifetime; a catalog-authorized replacement SeriesId has its own boundary. -This is an immutability guarantee, not a promise that every payload has reached -disk. Maintenance consumers must separately verify durable state availability and -atomically publish their output identity before claiming replay-safe consumption. +With persistence enabled, completion explicitly requests the existing flusher to +make the completed prefix durable, even if it is still inside the hot tier. +Completion waits until the corresponding epochs have been evicted after part and +manifest publication; only then does it persist the immutable boundary. An +in-memory deployment provides no restart guarantee. Maintenance consumers still +must atomically publish their output identity before claiming replay-safe consumption. The existing finite-source completeness proof still rejects untracked writes or pending admitted work. Continuous producer watermarks and derived-state commit transactions are separate from this finite-input boundary. From c5ae4a606d04296b6d0e3f49f23e016d8932c29d Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 11 Sep 2026 07:16:12 -0600 Subject: [PATCH 4/4] fix(storage): report proven immutable instances as complete --- .../storage_engines/sketch_db/index/mod.rs | 24 +++++++++++++++---- .../summary-catalog-sds-architecture.md | 3 ++- 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index f2fbe573d..e6f355a4b 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -1109,6 +1109,12 @@ impl SketchStore { .series .get(series_id) .map(|entry| Arc::clone(entry.value())); + let completed_through = self + .completed_windows + .read() + .unwrap() + .get(series_id) + .copied(); let status = match binding.metadata.status() { AggStatus::Active => SummaryInstanceStatus::Ready, AggStatus::Retired | AggStatus::Expired => SummaryInstanceStatus::Retiring, @@ -1155,10 +1161,11 @@ impl SketchStore { checksum: None, }, status: status.clone(), - // The current payload row does not distinguish a normal - // pane close from a late standalone correction. Report the - // concrete instance without inventing a completeness proof. - completeness: InstanceCompleteness::Unknown, + completeness: if completed_through.is_some_and(|end| window.1 <= end) { + InstanceCompleteness::Complete + } else { + InstanceCompleteness::Unknown + }, lifecycle: InstanceLifecycle::Persistent, observed_at_ms, }; @@ -4880,6 +4887,15 @@ mod tests { assert!(!store.completed_windows.read().unwrap().contains_key(&850)); std::fs::remove_dir(writer.path()).unwrap(); store.seal_finite_summary_input(&generation).unwrap(); + let producers = BTreeMap::from([(fingerprint.into(), "producer".to_string())]); + let inventory = store + .observed_summary_inventory("backend", "store", &producers, 1, 30_000) + .unwrap(); + assert_eq!(inventory.instances.len(), 1); + assert_eq!( + inventory.instances.values().next().unwrap().completeness, + InstanceCompleteness::Complete + ); assert!(!store.append_sample(850, BTreeMap::new(), (0, 30_000), sample(2))); assert!(!store.append_precompute( 850, diff --git a/docs/design_docs/summary-catalog-sds-architecture.md b/docs/design_docs/summary-catalog-sds-architecture.md index 9534cb31b..59e1fb72f 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -436,7 +436,8 @@ Finite Remote Write completion now fences the SummaryStore append boundary, not just the receiver queue. After all admitted outputs are published, the store records the greatest published window end for each physical SeriesId. Sketch and exact-state writes ending at or before that boundary are rejected, including -writes arriving through other producers. A later window remains writable. +writes arriving through other producers. A later window remains writable. Observed SDS inventory reports only these frozen +instances as `Complete`; ordinary emitted panes remain `Unknown`. The boundary is monotone in the existing SeriesId metadata sidecar and is restored before recovered identities become writable. A stale background metadata flush