diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index a8d6e415..7e65edad 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/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,7 +1769,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, @@ -1786,7 +1777,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 c1ec4991..ef9f115f 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/admission.rs b/data_plane/src/storage_engines/sketch_db/index/admission.rs index f0d6ecee..19c32510 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/epoch_columnar.rs b/data_plane/src/storage_engines/sketch_db/index/epoch_columnar.rs index 4442dfcb..695c7505 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 498fd540..e6f355a4 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,9 @@ impl Drop for StateMutation<'_> { #[derive(Default)] 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, @@ -879,17 +882,65 @@ 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 + // 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 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) + .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(()) + Ok(true) } pub(crate) fn summary_window_known_empty( @@ -1058,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, @@ -1104,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, }; @@ -1206,7 +1264,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 +1278,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 +1312,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 +1347,7 @@ impl SketchStore { retention_horizon_ms, ); } + true } /// Read a category through the derived in-memory rollup. Returns `None` @@ -2416,6 +2484,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 +2862,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 +2878,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 @@ -3005,6 +3074,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; } @@ -3102,6 +3179,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 { @@ -4745,6 +4829,168 @@ 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!(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(); + 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, + 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 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 c2aad589..e774b39e 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/metadata.rs b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs index 65c0e144..c432e07f 100644 --- a/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs +++ b/data_plane/src/storage_engines/sketch_db/persistence/metadata.rs @@ -319,6 +319,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 { @@ -343,6 +346,7 @@ impl SidMetaRecord { retired_at_ms: None, expires_at_ms: None, removed: false, + completed_through_ms: None, } } @@ -410,6 +414,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 @@ -480,6 +486,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, }, ); } @@ -533,6 +540,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() @@ -650,6 +658,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/data_plane/src/storage_engines/sketch_db/persistence/source.rs b/data_plane/src/storage_engines/sketch_db/persistence/source.rs index 117c6854..734cfaf6 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 048dcbac..59e1fb72 100644 --- a/docs/design_docs/summary-catalog-sds-architecture.md +++ b/docs/design_docs/summary-catalog-sds-architecture.md @@ -429,3 +429,27 @@ 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. 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 +cannot reopen a completed window. The guard belongs to the physical lifetime; +a catalog-authorized replacement SeriesId has its own boundary. + +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.