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..d4f7c1b9 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -8,12 +8,13 @@ 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}; type SummaryState = Arc; type PendingOutput = ( - Option, + Option<(MaterializationCommitKey, u64)>, PrecomputedOutput, Box, ); @@ -65,14 +66,141 @@ 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, + pending_batch: Option<[u8; 32]>, + batch_has_published: bool, +} + +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.pending_batch = None; + state.batch_has_published = false; + state.generation = generation; + } + Ok(plan) + } + + 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, + digest: [u8; 32], + completed: &[(MaterializationCommitKey, u64)], + ) -> Result<(), String> { + let mut state = self.0.lock().map_err(|_| "commit registry poisoned")?; + 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(()) + } + + 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 +209,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 +219,8 @@ impl CommitRegistry { } else { emit()?; committed.published = true; + committed.value = None; + commits.batch_has_published = true; Ok(()) } } @@ -101,12 +233,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 +247,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)) } } @@ -129,6 +265,7 @@ pub struct MaintenanceDagSink { inner: Arc, plans: HotReloadStreamingConfig, commits: CommitRegistry, + batch_guard: Mutex<()>, } impl MaintenanceDagSink { @@ -137,34 +274,36 @@ 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.plans.physical_plan_snapshot() 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 +354,37 @@ 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); + if self.commits.is_published(&key)? { + continue; + } let value = execute_precompute_sink( &dag, &installed.binding, @@ -232,16 +394,10 @@ 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), + Some((key, horizon_ms)), target_output, value.as_ref().as_ref().clone_boxed_core(), )); @@ -293,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(()) } } @@ -370,6 +596,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.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.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 +691,22 @@ 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_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(); let mut query = node(2); query.output_state = planner_types::post_asap::ExecutionDataState::READ_ROWS; let dag = ExecutableDag { @@ -428,7 +725,7 @@ mod tests { ( PostAsapNodeId(1), BackendNodeBinding::Materialization { - summary_definition: definition(2), + summary_definition: target_definition, }, ), ( @@ -458,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() @@ -470,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), ), @@ -484,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 + ); } } @@ -534,7 +878,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 +894,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 +950,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..865341de --- /dev/null +++ b/docs/developer_docs/maintenance-replay.md @@ -0,0 +1,40 @@ +# 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 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 +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.