From 80a5669fc74cb1a252c9cef3d56bfd62f7e74ff7 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:23:04 -0600 Subject: [PATCH 1/2] fix: bound maintenance replay state by materialization lifecycle --- Cargo.lock | 1 + data_plane/Cargo.toml | 1 + .../precompute_engine/maintenance_runtime.rs | 247 +++++++++++++++--- .../src/precompute_engine/subdag_scheduler.rs | 16 +- docs/developer_docs/maintenance-replay.md | 28 ++ 5 files changed, 252 insertions(+), 41 deletions(-) create mode 100644 docs/developer_docs/maintenance-replay.md diff --git a/Cargo.lock b/Cargo.lock index 545351fc..f1cb4720 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1205,6 +1205,7 @@ dependencies = [ "serde", "serde_json", "serde_yaml", + "sha2", "snap", "structopt", "tempfile", diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index e517326c..972f993c 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -12,6 +12,7 @@ asap_types.workspace = true # resolver). For now this dep just makes the control_plane crate # compile in the workspace and importable from main.rs. control_plane = { path = "../control_plane" } +sha2 = "0.10" # Step C (data_plane/docs/l4node-plan-executor-design.md): data_plane # vendors its own `SummaryExecutor`/`execute` (see diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index 5fc18cbe..a74340b9 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -8,6 +8,7 @@ use super::subdag_scheduler::{ use crate::storage_engines::types::{AggregateCore, HotReloadStreamingConfig, PrecomputedOutput}; use control_plane::physical::executable_binding::{BackendExecutableBinding, BackendNodeBinding}; use planner_types::post_asap::{ExecutableDagNode, ExecutableOperatorPayload, PostAsapNodeId}; +use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, BTreeSet}; use std::sync::{Arc, Mutex}; @@ -65,14 +66,92 @@ fn merge_inputs(inputs: &[Arc]) -> Result { } struct CommittedState { - value: Arc, + value: Option>, published: bool, } #[derive(Default)] -struct CommitRegistry(Mutex>); +struct CommitRegistryState { + generation: Option<(u64, u64)>, + entries: BTreeMap, + frontiers: BTreeMap, +} + +impl CommitRegistryState { + fn validate_key(&self, key: &MaterializationCommitKey) -> Result<(), String> { + if self + .generation + .is_some_and(|generation| generation != (key.plan_id, key.plan_version)) + { + return Err("maintenance retry belongs to an obsolete plan generation".into()); + } + if self + .frontiers + .get(&key.summary_definition) + .is_some_and(|(latest, horizon)| { + key.window_end_ms + <= latest.saturating_sub(i64::try_from(*horizon).unwrap_or(i64::MAX)) + }) + { + return Err( + "maintenance retry is outside the materialization retention horizon".into(), + ); + } + Ok(()) + } +} + +#[derive(Default)] +struct CommitRegistry(Mutex); impl CommitRegistry { + fn plan_snapshot( + &self, + plans: &HotReloadStreamingConfig, + ) -> Result>, String> { + let mut state = self.0.lock().map_err(|_| "commit registry poisoned")?; + // Read the authoritative generation while holding the registry lock, + // so an old in-flight batch cannot restore an obsolete generation. + let plan = plans.physical_plan_snapshot(); + let generation = plan + .as_ref() + .map(|plan| (plan.plan_id(), plan.plan_version())); + if state.generation != generation { + state.entries.clear(); + state.frontiers.clear(); + state.generation = generation; + } + Ok(plan) + } + + fn advance_retention( + &self, + key: &MaterializationCommitKey, + horizon_ms: u64, + ) -> Result<(), String> { + if horizon_ms == 0 { + return Err("maintenance retry horizon must be positive".into()); + } + let mut state = self.0.lock().map_err(|_| "commit registry poisoned")?; + state.validate_key(key)?; + let (latest, horizon) = state + .frontiers + .entry(key.summary_definition) + .or_insert((key.window_end_ms, horizon_ms)); + *latest = (*latest).max(key.window_end_ms); + *horizon = (*horizon).max(horizon_ms); + let cutoff = latest.saturating_sub(i64::try_from(*horizon).unwrap_or(i64::MAX)); + state.entries.retain(|entry, _| { + entry.summary_definition != key.summary_definition || entry.window_end_ms > cutoff + }); + Ok(()) + } + + fn is_published(&self, key: &MaterializationCommitKey) -> Result { + let state = self.0.lock().map_err(|_| "commit registry poisoned")?; + state.validate_key(key)?; + Ok(state.entries.get(key).is_some_and(|entry| entry.published)) + } fn publish( &self, key: &MaterializationCommitKey, @@ -81,7 +160,9 @@ impl CommitRegistry { // Serialize acknowledgement with publication so a concurrent replay // cannot skip an in-flight write that later fails. let mut commits = self.0.lock().map_err(|_| "commit registry poisoned")?; + commits.validate_key(key)?; let committed = commits + .entries .get_mut(key) .ok_or_else(|| "maintenance result was not committed".to_string())?; if committed.published { @@ -89,6 +170,7 @@ impl CommitRegistry { } else { emit()?; committed.published = true; + committed.value = None; Ok(()) } } @@ -101,12 +183,12 @@ impl IdempotentCommitSink for CommitRegistry { &self, key: &MaterializationCommitKey, ) -> Result>, Self::Error> { - Ok(self - .0 - .lock() - .map_err(|_| "commit registry poisoned")? + let state = self.0.lock().map_err(|_| "commit registry poisoned")?; + state.validate_key(key)?; + Ok(state + .entries .get(key) - .map(|committed| Arc::clone(&committed.value))) + .and_then(|committed| committed.value.as_ref().map(Arc::clone))) } fn commit_if_absent( @@ -115,11 +197,15 @@ impl IdempotentCommitSink for CommitRegistry { value: Arc, ) -> Result, Self::Error> { let mut commits = self.0.lock().map_err(|_| "commit registry poisoned")?; - let committed = commits.entry(key).or_insert_with(|| CommittedState { - value, - published: false, - }); - Ok(Arc::clone(&committed.value)) + commits.validate_key(&key)?; + let committed = commits + .entries + .entry(key) + .or_insert_with(|| CommittedState { + value: Some(Arc::clone(&value)), + published: false, + }); + Ok(committed.value.as_ref().map(Arc::clone).unwrap_or(value)) } } @@ -145,26 +231,29 @@ impl MaintenanceDagSink { output: PrecomputedOutput, state: Box, ) -> Result, String> { - let Some(plan) = self.plans.physical_plan_snapshot() else { + let Some(plan) = self.commits.plan_snapshot(&self.plans)? else { return Ok(vec![(None, output, state)]); }; let source_definition: asap_types::sds::SummaryDefinitionId = output.policy_fp.into(); let source: SummaryState = Arc::from(state); let mut derived = Vec::new(); let mut matched = false; - let mut lineage = Vec::new(); + let mut lineage = Sha256::new(); + lineage.update(b"asap-maintenance-lineage-v1"); let definition_bytes = source_definition.0 .0.to_be_bytes(); - lineage.extend_from_slice(&definition_bytes); + lineage.update(definition_bytes); let group_bytes = output .key .as_ref() .map(|key| key.serialize_to_bytes()) .unwrap_or_default(); - lineage.extend_from_slice(&(group_bytes.len() as u64).to_be_bytes()); - lineage.extend_from_slice(&group_bytes); + lineage.update((group_bytes.len() as u64).to_be_bytes()); + lineage.update(&group_bytes); let state_bytes = source.serialize_to_bytes(); - lineage.extend_from_slice(&(state_bytes.len() as u64).to_be_bytes()); - lineage.extend_from_slice(&state_bytes); + lineage.update((state_bytes.len() as u64).to_be_bytes()); + lineage.update(&state_bytes); + let lineage = lineage.finalize().to_vec(); + drop(state_bytes); for installed in plan.precompute_plan.executable_dags.values() { let dag = installed.document.decode()?; let source_nodes = installed @@ -215,14 +304,38 @@ impl MaintenanceDagSink { ); } matched = true; + let target = match installed.binding.node(*sink_node) { + Some(BackendNodeBinding::Materialization { summary_definition }) => { + *summary_definition + } + _ => return Err("precompute sink lacks materialization binding".into()), + }; let key = MaterializationCommitKey { plan_id: plan.plan_id(), plan_version: plan.plan_version(), - node_id: sink_node.0, + summary_definition: target, window_start_ms: output.start_timestamp as i64, window_end_ms: output.end_timestamp as i64, input_lineage: lineage.clone(), }; + let config = plan + .precompute_plan + .materializations + .iter() + .find(|config| config.policy_fingerprint() == target.fingerprint()) + .ok_or("maintenance sink has no materialization lifecycle")?; + // Keep replay receipts for the installed state-retention span, + // or one complete window when no longer retention is declared. + let horizon_ms = config + .num_aggregates_to_retain + .unwrap_or(1) + .saturating_mul(config.slide_interval) + .max(config.window_size) + .saturating_mul(1_000); + self.commits.advance_retention(&key, horizon_ms)?; + if self.commits.is_published(&key)? { + continue; + } let value = execute_precompute_sink( &dag, &installed.binding, @@ -232,14 +345,8 @@ impl MaintenanceDagSink { &self.commits, ) .map_err(schedule_error)?; - let target = match installed.binding.node(*sink_node) { - Some(BackendNodeBinding::Materialization { summary_definition }) => { - asap_types::PolicyFingerprint::from(*summary_definition) - } - _ => return Err("precompute sink lacks materialization binding".into()), - }; let mut target_output = output.clone(); - target_output.policy_fp = target; + target_output.policy_fp = target.into(); derived.push(( Some(key), target_output, @@ -370,6 +477,67 @@ mod tests { Arc::new(accumulator) } + // Receipts follow the declared event-time horizon and never retain accepted + // summary payloads; expired retries fail instead of becoming duplicate writes. + #[test] + fn maintenance_receipts_are_bounded_and_expired_retries_fail_closed() { + let commits = CommitRegistry::default(); + let key = |end| MaterializationCommitKey { + plan_id: 7, + plan_version: 1, + summary_definition: definition(2), + window_start_ms: end - 10, + window_end_ms: end, + input_lineage: vec![0; 32], + }; + for end in (10..=1_000).step_by(10) { + let key = key(end); + commits.advance_retention(&key, 30).unwrap(); + commits + .commit_if_absent(key.clone(), Arc::new(sum(2.0))) + .unwrap(); + commits.publish(&key, || Ok(())).unwrap(); + let state = commits.0.lock().unwrap(); + assert!(state.entries.len() <= 3); + assert!(state.entries.values().all(|entry| entry.value.is_none())); + } + assert!(commits.advance_retention(&key(970), 30).is_err()); + assert!(commits.get(&key(970)).is_err()); + assert!(commits + .publish(&key(970), || panic!( + "expired output must not reach storage" + )) + .is_err()); + assert!(commits.is_published(&key(980)).unwrap()); + commits.0.lock().unwrap().generation = Some((7, 2)); + assert!(commits + .commit_if_absent(key(1_000), Arc::new(sum(2.0))) + .is_err()); + } + + // Local node IDs are reused in separate query DAGs. Receipts must be scoped + // by materialization identity so neither DAG suppresses the other's output. + #[test] + fn different_materializations_have_independent_publication_receipts() { + let commits = CommitRegistry::default(); + for target in [2, 3] { + let key = MaterializationCommitKey { + plan_id: 7, + plan_version: 1, + summary_definition: definition(target), + window_start_ms: 0, + window_end_ms: 10, + input_lineage: vec![0; 32], + }; + assert!(!commits.is_published(&key).unwrap()); + commits + .commit_if_absent(key.clone(), Arc::new(sum(2.0))) + .unwrap(); + commits.publish(&key, || Ok(())).unwrap(); + } + assert_eq!(commits.0.lock().unwrap().entries.len(), 2); + } + // A failed downstream write must be retried, while accepted outputs remain // idempotent when the same maintenance lineage is replayed. #[test] @@ -404,12 +572,18 @@ mod tests { } } + let mut snapshot: serde_json::Value = serde_json::from_str(include_str!( + "../../../docs/examples/asapquery-planning-snapshot.json" + )) + .unwrap(); + snapshot["query_workload"]["repeating_queries"][0]["query"] = + "sum(sum_over_time(m[1m]))".into(); let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = - serde_json::from_str(include_str!( - "../../../docs/examples/asapquery-planning-snapshot.json" - )) - .unwrap(); + serde_json::from_value(snapshot).unwrap(); let mut bundle = snapshot.compile().unwrap(); + let target_definition = bundle.precompute_plan.materializations[0] + .policy_fingerprint() + .into(); let mut query = node(2); query.output_state = planner_types::post_asap::ExecutionDataState::READ_ROWS; let dag = ExecutableDag { @@ -428,7 +602,7 @@ mod tests { ( PostAsapNodeId(1), BackendNodeBinding::Materialization { - summary_definition: definition(2), + summary_definition: target_definition, }, ), ( @@ -534,7 +708,7 @@ mod tests { let key = MaterializationCommitKey { plan_id: 7, plan_version: 2, - node_id: 3, + summary_definition: definition(4), window_start_ms: 0, window_end_ms: 10, input_lineage: b"batch:1".to_vec(), @@ -550,6 +724,11 @@ mod tests { .unwrap(); assert_eq!(result.as_ref().aux_stats().sum, Some(4.0)); commits.publish(&key, || Ok(())).unwrap(); + assert!( + commits.get(&key).unwrap().is_none(), + "accepted payload must not remain in the retry registry" + ); + assert!(commits.is_published(&key).unwrap()); commits .publish(&key, || panic!("accepted lineage must not publish twice")) .unwrap(); @@ -601,7 +780,7 @@ mod tests { let key = MaterializationCommitKey { plan_id: 7, plan_version: 2, - node_id: 1, + summary_definition: definition(2), window_start_ms: 0, window_end_ms: 10, input_lineage: b"batch:1".to_vec(), diff --git a/data_plane/src/precompute_engine/subdag_scheduler.rs b/data_plane/src/precompute_engine/subdag_scheduler.rs index 2d0036f6..7ac0ee22 100644 --- a/data_plane/src/precompute_engine/subdag_scheduler.rs +++ b/data_plane/src/precompute_engine/subdag_scheduler.rs @@ -10,11 +10,12 @@ use std::{ pub struct MaterializationCommitKey { pub plan_id: u64, pub plan_version: u64, - pub node_id: u32, + pub summary_definition: asap_types::sds::SummaryDefinitionId, pub window_start_ms: i64, pub window_end_ms: i64, - /// Collision-free producer lineage bytes. Callers should include source - /// identity and immutable input payload identity, not a lossy hash. + /// Producer lineage identity, including source and immutable input payload. + /// Production uses a domain-separated SHA-256 digest to avoid retaining + /// another full copy of every source summary. pub input_lineage: Vec, } @@ -59,10 +60,11 @@ where R: PrecomputeOperatorRegistry, S: IdempotentCommitSink, { - if key.node_id != sink_node.0 { + if !matches!(binding.node(sink_node), Some(BackendNodeBinding::Materialization { summary_definition }) if *summary_definition == key.summary_definition) + { return Err(ScheduleError::Invalid(format!( - "commit key node {} does not match sink {}", - key.node_id, sink_node.0 + "commit key materialization {:?} does not match sink {}", + key.summary_definition, sink_node.0 ))); } binding.validate(dag).map_err(ScheduleError::Invalid)?; @@ -250,7 +252,7 @@ mod tests { MaterializationCommitKey { plan_id: 7, plan_version: 1, - node_id, + summary_definition: asap_types::PolicyFingerprint(u64::from(node_id) + 1).into(), window_start_ms: 10, window_end_ms: 20, input_lineage: b"checkpoint:3".to_vec(), diff --git a/docs/developer_docs/maintenance-replay.md b/docs/developer_docs/maintenance-replay.md new file mode 100644 index 00000000..24ed3f5a --- /dev/null +++ b/docs/developer_docs/maintenance-replay.md @@ -0,0 +1,28 @@ +# Maintenance replay and retention + +The maintenance sink retains in-process publication receipts for the active +physical plan. Receipt identity includes the plan generation, target summary +definition, output window, and a SHA-256 digest of the source definition, +group, and serialized input state. Query-local DAG node IDs are not publication +identity: separate query DAGs may use the same node numbers. + +An accepted write releases its cached derived payload. Its compact receipt +remains until the materialization's event-time retry horizon expires. The +horizon is the configured retained-state count times the slide interval, with +a minimum of one complete materialized window. With no retained-state count, +the horizon is one complete window. A later output advances that definition's +event-time frontier. An output ending at or before the frontier minus the +horizon is rejected; it must not be reinserted as a new write after receipt +eviction. Sparse or out-of-order data within the horizon remains eligible. + +When the sink observes a changed active plan generation, it clears old receipts +and cached failures. Previously captured work then fails closed when it attempts +to commit or publish. This uses installed plan identity and event time, not host +wall-clock time, so finite-input replay does not expire state merely because +the dataset is old. + +These receipts do not provide durable exactly-once publication after restart. +Storage must still make uncertain writes idempotent. A failed write retains +its derived state for retry within the same horizon; a successful write is +acknowledged only after the downstream sink accepts it. Retention expiration +may discard failed work once its output is outside the supported horizon. From fb4dc5a5c9fd6261521bac66e786b52b1fee9300 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 17:51:18 -0600 Subject: [PATCH 2/2] fix: retain batch receipts until complete publication --- .../precompute_engine/maintenance_runtime.rs | 250 +++++++++++++++--- docs/developer_docs/maintenance-replay.md | 14 +- 2 files changed, 223 insertions(+), 41 deletions(-) diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index a74340b9..d4f7c1b9 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -14,7 +14,7 @@ use std::sync::{Arc, Mutex}; type SummaryState = Arc; type PendingOutput = ( - Option, + Option<(MaterializationCommitKey, u64)>, PrecomputedOutput, Box, ); @@ -75,6 +75,8 @@ struct CommitRegistryState { generation: Option<(u64, u64)>, entries: BTreeMap, frontiers: BTreeMap, + pending_batch: Option<[u8; 32]>, + batch_has_published: bool, } impl CommitRegistryState { @@ -119,31 +121,78 @@ impl CommitRegistry { if state.generation != generation { state.entries.clear(); state.frontiers.clear(); + state.pending_batch = None; + state.batch_has_published = false; state.generation = generation; } Ok(plan) } - fn advance_retention( + fn begin_batch(&self, digest: [u8; 32]) -> Result<(), String> { + let mut state = self.0.lock().map_err(|_| "commit registry poisoned")?; + match state.pending_batch { + Some(pending) if pending != digest => Err( + "maintenance batch retry is pending; retry that batch before submitting new work" + .into(), + ), + _ => { + if state.pending_batch.is_none() { + state.batch_has_published = false; + } + state.pending_batch = Some(digest); + Ok(()) + } + } + } + + fn finish_batch(&self, digest: [u8; 32]) -> Result<(), String> { + self.complete_batch(digest, &[]) + } + + fn cancel_unpublished_batch(&self) { + if let Ok(mut state) = self.0.lock() { + if !state.batch_has_published { + state.entries.retain(|_, entry| entry.published); + state.pending_batch = None; + } + } + } + + fn complete_batch( &self, - key: &MaterializationCommitKey, - horizon_ms: u64, + digest: [u8; 32], + completed: &[(MaterializationCommitKey, u64)], ) -> Result<(), String> { - if horizon_ms == 0 { - return Err("maintenance retry horizon must be positive".into()); - } let mut state = self.0.lock().map_err(|_| "commit registry poisoned")?; - state.validate_key(key)?; - let (latest, horizon) = state - .frontiers - .entry(key.summary_definition) - .or_insert((key.window_end_ms, horizon_ms)); - *latest = (*latest).max(key.window_end_ms); - *horizon = (*horizon).max(horizon_ms); - let cutoff = latest.saturating_sub(i64::try_from(*horizon).unwrap_or(i64::MAX)); - state.entries.retain(|entry, _| { - entry.summary_definition != key.summary_definition || entry.window_end_ms > cutoff + if state.pending_batch != Some(digest) { + return Err("maintenance batch generation changed before completion".into()); + } + for (key, horizon) in completed { + if *horizon == 0 + || state + .generation + .is_some_and(|generation| generation != (key.plan_id, key.plan_version)) + { + return Err("maintenance batch has invalid completion lifecycle".into()); + } + let frontier = state + .frontiers + .entry(key.summary_definition) + .or_insert((key.window_end_ms, *horizon)); + frontier.0 = frontier.0.max(key.window_end_ms); + frontier.1 = frontier.1.max(*horizon); + } + let frontiers = state.frontiers.clone(); + state.entries.retain(|key, _| { + frontiers + .get(&key.summary_definition) + .is_none_or(|(latest, horizon)| { + key.window_end_ms + > latest.saturating_sub(i64::try_from(*horizon).unwrap_or(i64::MAX)) + }) }); + state.pending_batch = None; + state.batch_has_published = false; Ok(()) } @@ -171,6 +220,7 @@ impl CommitRegistry { emit()?; committed.published = true; committed.value = None; + commits.batch_has_published = true; Ok(()) } } @@ -215,6 +265,7 @@ pub struct MaintenanceDagSink { inner: Arc, plans: HotReloadStreamingConfig, commits: CommitRegistry, + batch_guard: Mutex<()>, } impl MaintenanceDagSink { @@ -223,17 +274,16 @@ impl MaintenanceDagSink { inner, plans, commits: CommitRegistry::default(), + batch_guard: Mutex::new(()), } } fn execute_one( &self, + plan: &crate::storage_engines::types::ActivePhysicalPlan, output: PrecomputedOutput, state: Box, ) -> Result, String> { - let Some(plan) = self.commits.plan_snapshot(&self.plans)? else { - return Ok(vec![(None, output, state)]); - }; let source_definition: asap_types::sds::SummaryDefinitionId = output.policy_fp.into(); let source: SummaryState = Arc::from(state); let mut derived = Vec::new(); @@ -332,7 +382,6 @@ impl MaintenanceDagSink { .saturating_mul(config.slide_interval) .max(config.window_size) .saturating_mul(1_000); - self.commits.advance_retention(&key, horizon_ms)?; if self.commits.is_published(&key)? { continue; } @@ -348,7 +397,7 @@ impl MaintenanceDagSink { let mut target_output = output.clone(); target_output.policy_fp = target.into(); derived.push(( - Some(key), + Some((key, horizon_ms)), target_output, value.as_ref().as_ref().clone_boxed_core(), )); @@ -400,31 +449,101 @@ impl OutputSink for MaintenanceDagSink { &self, outputs: Vec<(PrecomputedOutput, Box)>, ) -> Result<(), Box> { + // One bounded pending batch may be retried. Do not let another worker + // advance its frontier while a partially accepted batch is replayable. + let _guard = self + .batch_guard + .lock() + .map_err(|_| "maintenance batch lock poisoned")?; + let plan = self.commits.plan_snapshot(&self.plans)?; + let sinks = plan.as_ref().map_or(0, |plan| { + plan.precompute_plan + .executable_dags + .values() + .map(|dag| dag.binding.precompute_sinks.len()) + .sum::() + }); + if sinks == 0 { + return self.inner.emit_batch(outputs); + } + const MAX_BATCH_RECEIPTS: usize = 65_536; + const MAX_BATCH_SOURCE_BYTES: usize = 64 * 1024 * 1024; + if outputs.len().saturating_mul(sinks) > MAX_BATCH_RECEIPTS { + return Err( + "maintenance batch exceeds bounded receipt budget; split the input batch".into(), + ); + } + let mut digest = Sha256::new(); + digest.update(b"asap-maintenance-batch-v1"); + let mut source_bytes = 0usize; + for (output, state) in &outputs { + let bytes = state.serialize_to_bytes(); + let group = output + .key + .as_ref() + .map(|key| key.serialize_to_bytes()) + .unwrap_or_default(); + source_bytes = source_bytes + .saturating_add(bytes.len()) + .saturating_add(group.len()); + if source_bytes.saturating_mul(sinks) > MAX_BATCH_SOURCE_BYTES { + return Err( + "maintenance batch exceeds serialized source budget; split the input batch" + .into(), + ); + } + digest.update(output.policy_fp.0.to_be_bytes()); + digest.update(output.start_timestamp.to_be_bytes()); + digest.update(output.end_timestamp.to_be_bytes()); + digest.update((group.len() as u64).to_be_bytes()); + digest.update(group); + digest.update((bytes.len() as u64).to_be_bytes()); + digest.update(bytes); + } + let digest: [u8; 32] = digest.finalize().into(); + self.commits.begin_batch(digest)?; let mut transformed = Vec::new(); for (output, state) in outputs { - transformed.extend( - self.execute_one(output, state) - .map_err(|e| -> Box { e.into() })?, - ); + match self.execute_one( + plan.as_ref() + .expect("maintenance DAG requires an active plan"), + output, + state, + ) { + Ok(outputs) => transformed.extend(outputs), + Err(error) => { + self.commits.cancel_unpublished_batch(); + return Err(error.into()); + } + } } if transformed.iter().all(|(key, _, _)| key.is_none()) { - return self.inner.emit_batch( + self.inner.emit_batch( transformed .into_iter() .map(|(_, output, state)| (output, state)) .collect(), - ); + )?; + self.commits.finish_batch(digest)?; + return Ok(()); } // The generic sink can partially accept a batch. Acknowledge each // maintained output independently so retries skip only accepted writes. + let mut completed = Vec::new(); for (key, output, state) in transformed { match key { - Some(key) => self - .commits - .publish(&key, || self.inner.emit_batch(vec![(output, state)]))?, + Some((key, horizon)) => { + self.commits + .publish(&key, || self.inner.emit_batch(vec![(output, state)]))?; + completed.push((key, horizon)); + } None => self.inner.emit_batch(vec![(output, state)])?, } } + // Publication and partial replay use the previous frontier. Advance + // only after the complete batch was accepted, so an early pane cannot + // lose its receipt merely because a later pane shares its batch. + self.commits.complete_batch(digest, &completed)?; Ok(()) } } @@ -492,16 +611,16 @@ mod tests { }; for end in (10..=1_000).step_by(10) { let key = key(end); - commits.advance_retention(&key, 30).unwrap(); + commits.begin_batch([0; 32]).unwrap(); commits .commit_if_absent(key.clone(), Arc::new(sum(2.0))) .unwrap(); commits.publish(&key, || Ok(())).unwrap(); + commits.complete_batch([0; 32], &[(key, 30)]).unwrap(); let state = commits.0.lock().unwrap(); assert!(state.entries.len() <= 3); assert!(state.entries.values().all(|entry| entry.value.is_none())); } - assert!(commits.advance_retention(&key(970), 30).is_err()); assert!(commits.get(&key(970)).is_err()); assert!(commits .publish(&key(970), || panic!( @@ -581,6 +700,10 @@ mod tests { let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot = serde_json::from_value(snapshot).unwrap(); let mut bundle = snapshot.compile().unwrap(); + let target_config = &bundle.precompute_plan.materializations[0]; + let long_step = target_config.window_size.max( + target_config.slide_interval * target_config.num_aggregates_to_retain.unwrap_or(1), + ) * 1_000; let target_definition = bundle.precompute_plan.materializations[0] .policy_fingerprint() .into(); @@ -632,7 +755,11 @@ mod tests { query_plan: Arc::new(bundle.query_plan), storage_routing: Arc::new(Default::default()), }; - for fail_at in [0, 1] { + // The long batch spans two retention horizons. Its accepted prefix + // must remain replayable until the same whole batch completes. + for (fail_at, count, step, replay_after_success) in + [(0, 2, 10, true), (1, 2, 10, true), (2, 5, long_step, false)] + { let downstream = Arc::new(FailOnceSink { fail_at, ..Default::default() @@ -644,12 +771,12 @@ mod tests { )), ); let batch = || { - (0..2) + (0..count) .map(|i| { ( PrecomputedOutput::new( - i * 10, - (i + 1) * 10, + i * step, + (i + 1) * step, None, asap_types::PolicyFingerprint(1), ), @@ -658,11 +785,54 @@ mod tests { }) .collect() }; + if !replay_after_success { + let oversized = (0..65_537) + .map(|_| { + ( + PrecomputedOutput::new(0, step, None, asap_types::PolicyFingerprint(1)), + sum(2.0).clone_boxed_core(), + ) + }) + .collect(); + assert!(sink + .emit_batch(oversized) + .unwrap_err() + .to_string() + .contains("receipt budget")); + assert_eq!(downstream.attempts.load(Ordering::SeqCst), 0); + } assert!(sink.emit_batch(batch()).is_err()); + if !replay_after_success { + assert!(sink + .emit_batch(vec![( + PrecomputedOutput::new( + 999_000, + 1_000_000, + None, + asap_types::PolicyFingerprint(1), + ), + sum(3.0).clone_boxed_core() + )]) + .unwrap_err() + .to_string() + .contains("retry is pending")); + } sink.emit_batch(batch()).unwrap(); - sink.emit_batch(batch()).unwrap(); - assert_eq!(downstream.accepted.load(Ordering::SeqCst), 2); - assert_eq!(downstream.attempts.load(Ordering::SeqCst), 3); + if replay_after_success { + sink.emit_batch(batch()).unwrap(); + } else { + assert!(sink + .emit_batch(batch()) + .unwrap_err() + .to_string() + .contains("retention horizon")); + assert!(sink.commits.0.lock().unwrap().entries.len() <= 2); + } + assert_eq!(downstream.accepted.load(Ordering::SeqCst), count as usize); + assert_eq!( + downstream.attempts.load(Ordering::SeqCst), + count as usize + 1 + ); } } diff --git a/docs/developer_docs/maintenance-replay.md b/docs/developer_docs/maintenance-replay.md index 24ed3f5a..865341de 100644 --- a/docs/developer_docs/maintenance-replay.md +++ b/docs/developer_docs/maintenance-replay.md @@ -11,10 +11,22 @@ remains until the materialization's event-time retry horizon expires. The horizon is the configured retained-state count times the slide interval, with a minimum of one complete materialized window. With no retained-state count, the horizon is one complete window. A later output advances that definition's -event-time frontier. An output ending at or before the frontier minus the +event-time frontier after its complete output batch is accepted. An output ending at or before the frontier minus the horizon is rejected; it must not be reinserted as a new write after receipt eviction. Sparse or out-of-order data within the horizon remains eligible. +A batch may span more than the retention horizon. Its receipts remain pinned +until every output is accepted, including across a partial downstream failure. +The sink accepts one pending batch at a time; a failure applies backpressure to +different batches until the original batch is retried or the plan generation +changes. Retries use the same ordered outputs and serialized states. Receipt +and source-volume admission budgets are checked before publication: at most +65,536 possible derived outputs and 64 MiB of serialized source and group-key bytes multiplied +by the possible sink count. Oversized batches must be split. These are admission +budgets, not a measured heap-memory limit. Batch completion advances all frontiers +together and evicts expired receipts. Replaying a completed batch after its early +windows expired is rejected, rather than making those writes eligible again. + When the sink observes a changed active plan generation, it clears old receipts and cached failures. Previously captured work then fails closed when it attempts to commit or publish. This uses installed plan identity and event time, not host