From df553f676de1b912d33f3ce652d3169f7d709cb8 Mon Sep 17 00:00:00 2001 From: Beinan Date: Tue, 20 Oct 2026 19:05:17 +0000 Subject: [PATCH] feat(wal): retire closed source payloads while retaining retry receipts --- crates/lance-context-wal/src/replication.rs | 5 +- .../src/replication/checkpoint.rs | 29 +++- .../src/replication/checkpoint/tests.rs | 1 + .../src/replication/source/closure.rs | 22 ++- .../src/replication/source/tests.rs | 13 +- .../src/replication/source/tests/closed.rs | 129 +++++++++++++++- .../src/replication/state.rs | 144 ++++++++++++++++-- .../replication_process/source_closure.rs | 53 ++++++- docs/design/wal-manifest-authority.md | 2 +- docs/design/wal-prepared-descriptors.md | 2 +- docs/design/wal-publication-outcomes.md | 2 +- docs/design/wal-publication-plans.md | 2 +- docs/design/wal-publisher-authority.md | 2 +- docs/design/wal-source-closure.md | 12 +- docs/design/wal-source-pins.md | 2 +- docs/design/wal-source-retirement.md | 38 +++++ 16 files changed, 410 insertions(+), 48 deletions(-) create mode 100644 docs/design/wal-source-retirement.md diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 61eced9..1458216 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -93,7 +93,7 @@ impl Limits { "invalid replication limits" ); Ok(bincode::serialize(&( - "lance-context/replicated-append/v8", + "lance-context/replicated-append/v9", self, ))?) } @@ -459,7 +459,8 @@ impl Node { } /// Persist a catalog-authorized source closure and reject delayed pin - /// requests for the closed plan. Bodies remain retained until retirement. + /// requests for the closed plan. Retire selected payload metadata while + /// preserving latest producer receipts; physical collection is separate. pub fn close_source(&mut self, request: u64, permit: source::ClosurePermit) -> Result<()> { let group = permit.closure().group; let batch = permit.batch(&self.layout, &self.limits)?; diff --git a/crates/lance-context-wal/src/replication/checkpoint.rs b/crates/lance-context-wal/src/replication/checkpoint.rs index cd358df..2a4f555 100644 --- a/crates/lance-context-wal/src/replication/checkpoint.rs +++ b/crates/lance-context-wal/src/replication/checkpoint.rs @@ -2,7 +2,7 @@ //! building and loading an image never decode Arrow or concatenate its bodies. //! File I/O belongs on a bounded worker, outside the Raft owner. Only a later //! durable Raft transition can install a prepared image. Files are retained on -//! errors; physical collection and publication-aware retirement are separate. +//! errors; images omit retired bodies but physical collection is separate. use super::*; use bincode::Options; @@ -14,8 +14,8 @@ use std::{ path::PathBuf, }; -const FORMAT: u32 = 8; -const MAGIC: &[u8; 8] = b"LCWALCP8"; +const FORMAT: u32 = 9; +const MAGIC: &[u8; 8] = b"LCWALCP9"; pub const CHUNK_BYTES: usize = 1024 * 1024; fn codec() -> impl Options { @@ -74,6 +74,7 @@ struct Descriptor { tables: u64, pins: u64, closed: u64, + retired: u64, metadata_bytes: u64, payload_bytes: u64, metadata_digest: [u8; 32], @@ -91,6 +92,7 @@ fn metadata_limit(limits: &Limits) -> u64 { + limits.max_manifest_records * 2048 + limits.max_tables * 256 + limits.max_tables * (source::MAX_CLOSED_BYTES as u64 + 4) + + limits.max_producers * (state::MAX_RETIRED_BYTES as u64 + 4) + 1024 } @@ -102,7 +104,7 @@ impl Descriptor { && !self.id.is_nil() && self.records <= limits.max_records && self.producers <= limits.max_producers - && self.producers <= self.records + && self.retired <= self.producers && self.tables <= limits.max_tables && self.pins <= limits.max_tables.min(limits.max_active_plans) && self.closed <= limits.max_tables @@ -453,6 +455,7 @@ pub(super) fn build( tables: state.catalog.tables.len() as u64, pins: state.pins.len() as u64, closed: state.closed.len() as u64, + retired: state.retired.len() as u64, metadata_bytes: 0, payload_bytes, metadata_digest: [0; 32], @@ -467,6 +470,7 @@ pub(super) fn build( write_u64(&mut metadata, descriptor.tables)?; write_u64(&mut metadata, descriptor.pins)?; write_u64(&mut metadata, descriptor.closed)?; + write_u64(&mut metadata, descriptor.retired)?; for (index, meta) in &state.accepted { write_meta(&mut metadata, *index, meta)?; } @@ -482,6 +486,9 @@ pub(super) fn build( for closed in state.closed.values() { write_frame(&mut metadata, &source::encode_closed(closed)?)?; } + for (key, retired) in &state.retired { + write_frame(&mut metadata, &state::encode_retired(*key, *retired)?)?; + } (descriptor.metadata_bytes, descriptor.metadata_digest) = metadata.finish()?; let mut pack = Writer::new(&paths[1], payload_bytes)?; for (index, meta) in &state.accepted { @@ -530,7 +537,8 @@ pub(super) fn load( && read_u64(&mut metadata)? == descriptor.producers && read_u64(&mut metadata)? == descriptor.tables && read_u64(&mut metadata)? == descriptor.pins - && read_u64(&mut metadata)? == descriptor.closed, + && read_u64(&mut metadata)? == descriptor.closed + && read_u64(&mut metadata)? == descriptor.retired, "image metadata count mismatch" ); let mut state = State { @@ -594,6 +602,17 @@ pub(super) fn load( state.closed.insert(table, Arc::new(closed)); previous = Some(table); } + let mut previous = None; + for _ in 0..descriptor.retired { + let (bytes, n) = read_frame::<{ state::MAX_RETIRED_BYTES }>(&mut metadata)?; + let (key, retired) = state::decode_retired(&bytes[..n])?; + ensure!( + previous.is_none_or(|old| old < key), + "unordered or duplicate retired boundary" + ); + state.retired.insert(key, retired); + previous = Some(key); + } let metadata_file = metadata.finish(descriptor.metadata_bytes, descriptor.metadata_digest)?; state.validate_checkpoint(layout, group, limits, descriptor.binding.index)?; let (locations, payload_bytes) = locations(&state)?; diff --git a/crates/lance-context-wal/src/replication/checkpoint/tests.rs b/crates/lance-context-wal/src/replication/checkpoint/tests.rs index 7acac8e..cf77495 100644 --- a/crates/lance-context-wal/src/replication/checkpoint/tests.rs +++ b/crates/lance-context-wal/src/replication/checkpoint/tests.rs @@ -403,6 +403,7 @@ fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() -> state.catalog.tables.len(), state.pins.len(), state.closed.len(), + state.retired.len(), ] { assert_eq!(read_u64(&mut header)?, count as u64); } diff --git a/crates/lance-context-wal/src/replication/source/closure.rs b/crates/lance-context-wal/src/replication/source/closure.rs index 7891316..8445d73 100644 --- a/crates/lance-context-wal/src/replication/source/closure.rs +++ b/crates/lance-context-wal/src/replication/source/closure.rs @@ -1,5 +1,5 @@ //! Catalog-authorized closure of an exact source pin. Closure persists a -//! monotonic fence before releasing the pin. Payload retirement is separate. +//! monotonic fence before releasing the pin and retiring its exact payload range. use super::*; const DOMAIN: &str = "lance-context/catalog-source-closure/v1"; @@ -217,6 +217,12 @@ pub(super) fn apply( if !state.closed.contains_key(&c.table) && state.closed.len() as u64 >= limits.max_tables { return Ok(Reply::Rejected(Rejection::Capacity)); } + for (i, meta) in &selection.range.records { + ensure!( + state.accepted.get(i) == Some(meta), + "pinned source payload metadata missing" + ); + } let table = c.table; let pin_index = pin.index; state.closed.insert( @@ -227,8 +233,18 @@ pub(super) fn apply( permit, }), ); - state.pins.remove(&table); - // Accepted bodies and producer tails remain intact in this increment. + let pin = state.pins.remove(&table).expect("verified source pin"); + for (i, meta) in &pin.permit.selection().range.records { + state.retired.insert( + (meta.stream, meta.bucket, meta.producer), + state::Retired { + sequence: meta.sequence, + index: *i, + }, + ); + state.accepted.remove(i); + } + // Latest producer receipts survive even if their payload has been retired. Ok(Reply::Closed { index }) } diff --git a/crates/lance-context-wal/src/replication/source/tests.rs b/crates/lance-context-wal/src/replication/source/tests.rs index 2ca42f5..9483c71 100644 --- a/crates/lance-context-wal/src/replication/source/tests.rs +++ b/crates/lance-context-wal/src/replication/source/tests.rs @@ -18,6 +18,9 @@ impl Cluster { Self::with_plan_limit(128) } fn with_plan_limit(max_active_plans: u64) -> Result { + Self::with_policy(max_active_plans, 100_000) + } + fn with_policy(max_active_plans: u64, max_records: u64) -> Result { let root = tempfile::tempdir()?; let layout = Arc::new(Layout { cluster: Uuid::from_u128(91), @@ -33,6 +36,7 @@ impl Cluster { let limits = Limits { catalog_key: Some(signer.public_key()), max_active_plans, + max_records, ..Default::default() }; let mut c = Self { @@ -137,7 +141,14 @@ impl Cluster { .context("missing publisher completion") } fn plan(&mut self, mutate: impl FnOnce(&mut SourceRange)) -> Result { - for seq in 1..=12 { + self.plan_with_background(mutate, 12) + } + fn plan_with_background( + &mut self, + mutate: impl FnOnce(&mut SourceRange), + background: u64, + ) -> Result { + for seq in 1..=background { self.append(Uuid::from_u128(999), seq)?; } self.append(self.table, 1)?; diff --git a/crates/lance-context-wal/src/replication/source/tests/closed.rs b/crates/lance-context-wal/src/replication/source/tests/closed.rs index 9bbee99..ca24bc0 100644 --- a/crates/lance-context-wal/src/replication/source/tests/closed.rs +++ b/crates/lance-context-wal/src/replication/source/tests/closed.rs @@ -16,7 +16,13 @@ fn publisher(events: Events) -> Result { } fn fixture() -> Result<(Cluster, Permit, Uuid, Uuid)> { - let mut c = Cluster::new()?; + fixture_with_background(100_000, 12) +} +fn fixture_with_background( + max_records: u64, + background: u64, +) -> Result<(Cluster, Permit, Uuid, Uuid)> { + let mut c = Cluster::with_policy(128, max_records)?; let authority = c.acquire(0)?; let incarnation = Uuid::from_u128(800); c.nodes[0].bootstrap_manifest( @@ -37,7 +43,7 @@ fn fixture() -> Result<(Cluster, Permit, Uuid, Uuid)> { let publication::Reply::ManifestReserved(base) = publisher(c.pump()?)? else { anyhow::bail!("bootstrap missing") }; - let plan = c.plan(|_| {})?; + let plan = c.plan_with_background(|_| {}, background)?; let permit = c.permit(plan)?; assert!(matches!(c.pin(permit.clone())?, Reply::Pinned { .. })); Ok((c, permit, incarnation, base.identity)) @@ -141,7 +147,7 @@ fn close(c: &mut Cluster, permit: ClosurePermit) -> Result { } #[test] -fn durable_closure_releases_pin_rejects_delayed_permits_and_preserves_data() -> Result<()> { +fn closure_retires_selected_payloads_preserves_suffix_and_rejects_delayed_pins() -> Result<()> { let (mut c, pin, incarnation, predecessor) = fixture()?; assert!(closure_permit(&mut c, pin.selection().plan_index)?.is_none()); let permit = publish(&mut c, incarnation, predecessor)?; @@ -158,7 +164,8 @@ fn durable_closure_releases_pin_rejects_delayed_permits_and_preserves_data() -> 1, "raw streams must not allocate closure history" ); - assert_eq!(state.accepted, before.accepted); + assert_eq!(state.accepted.len() + 2, before.accepted.len()); + assert_eq!(state.retired.len(), 1); assert_eq!(state.producers, before.producers); assert_eq!( state.closed[&c.table].pin_index, @@ -186,6 +193,10 @@ fn durable_closure_releases_pin_rejects_delayed_permits_and_preserves_data() -> assert!(state.pins.is_empty()); state.validate_checkpoint(&c.layout, 20, &c.limits, state.applied)?; for (index, meta) in &before.accepted { + if meta.stream == c.table && *index <= pin.selection().range.through { + assert!(node.read_accepted(20, *index)?.is_none()); + continue; + } let record = node .read_accepted(20, *index)? .context("retained source disappeared")?; @@ -196,6 +207,116 @@ fn durable_closure_releases_pin_rejects_delayed_permits_and_preserves_data() -> Ok(()) } +#[test] +fn empty_payload_images_preserve_latest_retry_and_captured_old_images() -> Result<()> { + use checkpoint::Part; + let (mut c, pin, incarnation, predecessor) = fixture_with_background(2, 0)?; + let original_index = pin.selection().range.records.last().unwrap().0; + let permit = publish(&mut c, incarnation, predecessor)?; + let captured = c.nodes[3].prepare_checkpoint(20, Uuid::new_v4())?; + assert!(matches!(close(&mut c, permit)?, Reply::Closed { .. })); + let prepared = captured.run()?; + let old_image = prepared.image(); + assert_eq!(old_image.bytes(Part::Payload), 8192); + assert_eq!( + old_image.chunk(Part::Payload, 0, 4096)?, + Bytes::from(vec![1; 4096]) + ); + drop(prepared); + for node in &mut c.nodes[3..] { + let prepared = node.prepare_checkpoint(20, Uuid::new_v4())?.run()?; + assert_eq!(prepared.image().bytes(Part::Payload), 0); + node.install_checkpoint(prepared)?; + } + c.reopen()?; + assert_eq!( + old_image.chunk(Part::Payload, 4096, 4096)?, + Bytes::from(vec![2; 4096]) + ); + let retry = Batch::new( + c.table, + 1, + Uuid::from_u128(94), + 2, + Bytes::from(vec![2; 4096]), + ); + c.nodes[3].propose(20, 810, retry.clone())?; + assert!(c.pump()?.completions.iter().any(|c| c.request == 810 + && c.outcome + == Outcome::Committed(Applied::Accepted { + index: original_index + }))); + assert_eq!( + c.nodes[3].status(20)?.accepted_records, + 0, + "retry resurrected payload" + ); + let mut changed = retry; + changed.body = Bytes::from(vec![9; 4096]); + changed.meta.digest = Sha256::digest(&changed.body).into(); + c.nodes[3].propose(20, 811, changed)?; + assert!(c + .pump()? + .completions + .iter() + .any(|c| c.request == 811 + && matches!(c.outcome, Outcome::Committed(Applied::Rejected { .. })))); + // Both record slots were reclaimed, while the producer sequence continued. + c.append(c.table, 3)?; + c.append(c.table, 4)?; + assert_eq!(c.nodes[3].status(20)?.accepted_records, 2); + for node in &c.nodes[3..] { + let state = &node.groups[&20].state; + state.validate_checkpoint(&c.layout, 20, &c.limits, state.applied)?; + assert!(node.read_accepted(20, original_index)?.is_none()); + } + Ok(()) +} + +#[test] +fn retirement_boundaries_cannot_fabricate_retry_receipts_or_skip_retained_sequences() -> Result<()> +{ + let (mut c, _, incarnation, predecessor) = fixture()?; + let permit = publish(&mut c, incarnation, predecessor)?; + c.append(c.table, 3)?; + close(&mut c, permit)?; + let state = c.nodes[3].groups[&20].state.clone(); + let key = (c.table, 1, Uuid::from_u128(94)); + for mutation in 0..8 { + let mut invalid = state.clone(); + match mutation { + 0 => { + invalid.retired.clear(); + } + 1 => { + invalid.producers.remove(&key); + } + 2 => invalid.retired.get_mut(&key).unwrap().sequence += 1, + 3 => invalid.retired.get_mut(&key).unwrap().index = state.applied, + 4 => { + invalid.closed.remove(&c.table); + } + 5 => { + invalid.accepted.retain(|_, meta| meta.stream != c.table); + } + 6 => invalid.producers.get_mut(&key).unwrap().1 -= 1, + 7 => { + invalid + .retired + .insert((Uuid::new_v4(), 1, Uuid::new_v4()), invalid.retired[&key]); + } + _ => unreachable!(), + } + assert!( + invalid + .validate_checkpoint(&c.layout, 20, &c.limits, invalid.applied) + .is_err(), + "mutation {mutation}" + ); + } + Ok(()) +} + #[test] fn closure_signature_scope_and_snapshot_fences_are_checked() -> Result<()> { let (mut c, pin, incarnation, predecessor) = fixture()?; diff --git a/crates/lance-context-wal/src/replication/state.rs b/crates/lance-context-wal/src/replication/state.rs index cba7382..7ec325a 100644 --- a/crates/lance-context-wal/src/replication/state.rs +++ b/crates/lance-context-wal/src/replication/state.rs @@ -53,6 +53,70 @@ fn validate_identity(layout: &Layout, group: u64, meta: &BatchMeta) -> Result<() ); Ok(()) } + +pub(super) type ProducerKey = (Uuid, u32, Uuid); +pub(super) const MAX_RETIRED_BYTES: usize = 64; + +/// The last sequence removed from a lane's payload prefix. Latest receipts +/// remain in `producers`, independently of whether their payload is retained. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(super) struct Retired { + pub sequence: u64, + pub index: u64, +} +pub(super) fn encode_retired(key: ProducerKey, retired: Retired) -> Result> { + // UUID's serde representation adds a length prefix. Store the two fixed + // arrays explicitly: 16 + 4 + 16 + 8 + 8 = 52 bytes per producer boundary. + let bytes = codec().serialize(&(key.0.as_bytes(), key.1, key.2.as_bytes(), retired))?; + ensure!( + bytes.len() <= MAX_RETIRED_BYTES, + "retired boundary byte limit" + ); + Ok(bytes) +} +pub(super) fn decode_retired(bytes: &[u8]) -> Result<(ProducerKey, Retired)> { + ensure!( + bytes.len() <= MAX_RETIRED_BYTES, + "retired boundary byte limit" + ); + let (table, bucket, producer, retired): ([u8; 16], u32, [u8; 16], Retired) = + codec().deserialize(bytes)?; + let key = (Uuid::from_bytes(table), bucket, Uuid::from_bytes(producer)); + ensure!( + encode_retired(key, retired)? == bytes, + "noncanonical retired boundary" + ); + Ok((key, retired)) +} + +#[cfg(test)] +mod retired_encoding_tests { + use super::*; + + #[test] + fn retired_boundary_has_a_fixed_52_byte_little_endian_frame() -> Result<()> { + let key = ( + Uuid::from_bytes([0x11; 16]), + 0x01020304, + Uuid::from_bytes([0x22; 16]), + ); + let retired = Retired { + sequence: 0x0102030405060708, + index: 9, + }; + let mut expected = vec![0x11; 16]; + expected.extend_from_slice(&[4, 3, 2, 1]); + expected.extend_from_slice(&[0x22; 16]); + expected.extend_from_slice(&[8, 7, 6, 5, 4, 3, 2, 1, 9, 0, 0, 0, 0, 0, 0, 0]); + assert_eq!(expected.len(), 52); + assert_eq!(encode_retired(key, retired)?, expected); + assert_eq!(decode_retired(&expected)?, (key, retired)); + assert!(decode_retired(&expected[..51]).is_err()); + expected.push(0); + assert!(decode_retired(&expected).is_err()); + Ok(()) + } +} pub(super) fn validate_entry( layout: &Layout, group: u64, @@ -83,6 +147,7 @@ pub(super) fn validate_entry( #[derive(Clone, Debug, Default, PartialEq, Eq)] pub(super) struct State { + pub retired: BTreeMap, pub closed: BTreeMap>, pub pins: BTreeMap>, pub catalog: publication::Catalog, @@ -113,9 +178,8 @@ impl State { Ok(Page { entries, through }) } - /// The image retains every accepted record. Producer tails must exactly - /// match those records, including sequence continuity within each lane. - /// Retirement will need a new format with independently retained tails. + /// Retained lane sequences continue from a durable retirement boundary. + /// Latest receipts survive even when their complete prefix has been retired. pub fn validate_checkpoint( &self, layout: &Layout, @@ -129,13 +193,54 @@ impl State { ); ensure!( self.accepted.len() as u64 <= limits.max_records - && self.producers.len() as u64 <= limits.max_producers, + && self.producers.len() as u64 <= limits.max_producers + && self.retired.len() <= self.producers.len(), "application checkpoint exceeds state limits" ); + self.catalog.validate(layout, group, cut, limits)?; + source::validate_state(self, layout, group, cut, limits)?; + for (key, (meta, index)) in &self.producers { + validate_identity(layout, group, meta)?; + meta.check_body_length( + usize::try_from(meta.body_bytes)?, + limits.max_record_bytes as usize, + )?; + ensure!( + *key == (meta.stream, meta.bucket, meta.producer) + && meta.sequence > 0 + && *index > 0 + && *index <= cut, + "invalid producer receipt identity or position" + ); + } + for (key, retired) in &self.retired { + let (latest, index) = self + .producers + .get(key) + .context("retired lane lacks latest receipt")?; + let closed = self + .closed + .get(&key.0) + .context("retired lane lacks source closure")?; + ensure!( + retired.sequence > 0 + && retired.sequence <= latest.sequence + && retired.index > 0 + && retired.index <= closed.permit.closure().through + && ((retired.sequence == latest.sequence && retired.index == *index) + || (retired.sequence < latest.sequence && retired.index < *index)), + "retired boundary disagrees with source closure or latest receipt" + ); + } let mut tails = BTreeMap::new(); for (index, meta) in &self.accepted { ensure!( - *index > 0 && *index <= cut, + *index > 0 + && *index <= cut + && self + .closed + .get(&meta.stream) + .is_none_or(|c| *index > c.permit.closure().through), "accepted index outside checkpoint" ); validate_identity(layout, group, meta)?; @@ -144,21 +249,32 @@ impl State { limits.max_record_bytes as usize, )?; let key = (meta.stream, meta.bucket, meta.producer); - let expected = tails - .get(&key) - .map_or(1, |(old, _): &(BatchMeta, u64)| old.sequence + 1); + ensure!( + self.producers.contains_key(&key), + "accepted lane lacks latest receipt" + ); + let expected = tails.get(&key).map_or_else( + || self.retired.get(&key).map_or(1, |r| r.sequence + 1), + |(old, _): &(BatchMeta, u64)| old.sequence + 1, + ); ensure!( meta.sequence == expected, "checkpoint producer sequence gap" ); tails.insert(key, (meta.clone(), *index)); } - ensure!( - tails == self.producers, - "checkpoint producer tails disagree" - ); - self.catalog.validate(layout, group, cut, limits)?; - source::validate_state(self, layout, group, cut, limits)?; + for (key, latest) in &self.producers { + ensure!( + match tails.get(key) { + Some(retained) => retained == latest, + None => self + .retired + .get(key) + .is_some_and(|r| r.sequence == latest.0.sequence && r.index == latest.1), + }, + "checkpoint producer tail has no retained or retired evidence" + ); + } Ok(()) } diff --git a/crates/lance-context-wal/tests/replication_process/source_closure.rs b/crates/lance-context-wal/tests/replication_process/source_closure.rs index ecaeca9..2cbf827 100644 --- a/crates/lance-context-wal/tests/replication_process/source_closure.rs +++ b/crates/lance-context-wal/tests/replication_process/source_closure.rs @@ -58,10 +58,19 @@ pub(super) fn close_and_restart( let before = c.completions.len(); c.completions.retain(|c| c.request != 600_002); assert_eq!(before - c.completions.len(), 1); - c.peers + let original_index = stale.selection().range.records.last().unwrap().0; + let Response::Image(_, _, payload_bytes) = c + .peers .get_mut(&source) .unwrap() - .call(Request::PrepareCheckpoint(20, Uuid::new_v4()))?; + .call(Request::PrepareCheckpoint(20, Uuid::new_v4()))? + else { + anyhow::bail!("source image missing") + }; + assert_eq!( + payload_bytes, 0, + "retired source still occupies the new image" + ); c.command(source, Request::InstallCheckpoint)?; c.pump(&[4, 5, 6])?; c.queue.clear(); @@ -110,11 +119,41 @@ pub(super) fn close_and_restart( }))) ); for id in 4..=6 { - verify_records( - c.peers.get_mut(&id).unwrap(), - 20, - std::slice::from_ref(input), - )?; + assert!(matches!( + c.peers + .get_mut(&id) + .unwrap() + .call(Request::Read(20, original_index))?, + Response::Record(None) + )); } + c.command(source, Request::Propose(20, 600_007, input.clone()))?; + c.pump(&[4, 5, 6])?; + assert_eq!( + c.outcome(600_007), + Some(Outcome::Committed(Applied::Accepted { + index: original_index + })) + ); + c.command( + source, + Request::Query( + 20, + 600_008, + Query::Producer { + table, + bucket: input.meta.bucket, + producer: input.meta.producer, + }, + ), + )?; + c.pump(&[4, 5, 6])?; + assert!( + matches!(c.read_outcome(600_008), Some(ReadOutcome::Observed { value: Value::Producer(Some((meta, index))), .. }) + if meta == input.meta && index == original_index) + ); + assert!( + matches!(c.peers.get_mut(&source).unwrap().call(Request::Page(20))?, Response::Page(page) if page.entries.is_empty()) + ); Ok(()) } diff --git a/docs/design/wal-manifest-authority.md b/docs/design/wal-manifest-authority.md index ae3b9b4..f3776ea 100644 --- a/docs/design/wal-manifest-authority.md +++ b/docs/design/wal-manifest-authority.md @@ -32,7 +32,7 @@ lookup, rather than serializing all retained history. Captured table views share the index and immutable records. There are at most 128 versions per table and a configurable node policy bound of 4,096 records by default. Capacity refuses new reservations while preserving exact retries. Application identity and checkpoint -format advance to v8 and include the index in bounds and integrity validation. +format advance to v9 and include the index in bounds and integrity validation. The real Lance test adapter writes unique staging manifests using the public `ManifestWriter`, then reserves them through six persistent TCP WAL processes. diff --git a/docs/design/wal-prepared-descriptors.md b/docs/design/wal-prepared-descriptors.md index 7df2aec..94fbe02 100644 --- a/docs/design/wal-prepared-descriptors.md +++ b/docs/design/wal-prepared-descriptors.md @@ -25,7 +25,7 @@ Captured table views share the prepared metadata through `Arc` and `Bytes`. Adding descriptors does not copy the plan's selected record list. The default per-plan descriptor budget is 256 KiB, with a 1 MiB protocol ceiling. Commands have separate type bounds, and checkpoint metadata accounts for both plans and -descriptors. The application/checkpoint format advances to v8; older identities +descriptors. The application/checkpoint format advances to v9; older identities and images are rejected rather than silently migrated. The `lance-integration` feature adds real Lance qualification dependencies. It diff --git a/docs/design/wal-publication-outcomes.md b/docs/design/wal-publication-outcomes.md index 02a62ea..a0aa89a 100644 --- a/docs/design/wal-publication-outcomes.md +++ b/docs/design/wal-publication-outcomes.md @@ -18,7 +18,7 @@ Source positions and the publication checkpoint remain unchanged, and another plan remains blocked. In particular, a catalog index is never used as a source group cursor. No source pin, payload, descriptor, manifest or image is deleted. -Application identity and checkpoint format advance to v8. Older identities are +Application identity and checkpoint format advance to v9. Older identities are rejected without in-place migration. Images account for the bounded outcome metadata and verify its exact selected manifest, pending plan, prepared digest and catalog index ordering. Outcome metadata is bounded by active plans. diff --git a/docs/design/wal-publication-plans.md b/docs/design/wal-publication-plans.md index ceb7845..0c9c793 100644 --- a/docs/design/wal-publication-plans.md +++ b/docs/design/wal-publication-plans.md @@ -37,7 +37,7 @@ the heap. Decoders reject noncanonical maps instead of silently accepting duplic keys normalized by the serializer. Source pins subsequently advance the optional application identity and checkpoint -format to v8. Existing v1/v2/v3/v4/v5/v6/v7 directories and images are rejected; there is no in-place migration +format to v9. Existing v1/v2/v3/v4/v5/v6/v7/v8 directories and images are rejected; there is no in-place migration or production rollout in this increment. Catalog-only and mixed data/catalog snapshots preserve plans, with semantic validation after checksum verification. diff --git a/docs/design/wal-publisher-authority.md b/docs/design/wal-publisher-authority.md index 56cdc88..1305d46 100644 --- a/docs/design/wal-publisher-authority.md +++ b/docs/design/wal-publisher-authority.md @@ -36,7 +36,7 @@ scope, epochs, grant positions, destination uniqueness and contracts, then repla the committed suffix. Metadata contains bounded, ordered table frames alongside the existing data records and producer tails. Payload packs still contain only opaque accepted data. The current replication identity and application image -format are v8, including [immutable publication plans](wal-publication-plans.md) +format are v9, including [immutable publication plans](wal-publication-plans.md) and [source pins](wal-source-pins.md). Earlier v1/v2/v3/v4/v5 application directories are rejected; this optional component does not provide an in-place migration or change the production ingestion path. diff --git a/docs/design/wal-source-closure.md b/docs/design/wal-source-closure.md index daa12ce..2ec2504 100644 --- a/docs/design/wal-source-closure.md +++ b/docs/design/wal-source-closure.md @@ -19,15 +19,15 @@ history. The compact closure retains no source record list or payload buffers. Subsequent pins must start at the saved source cursor and advance the catalog plan identity; retained history cannot be repinned from an old cursor. -Application identity and image format advance to v8, without in-place migration +Application identity and image format advance to v9, without in-place migration from older identities. Snapshot validation verifies closure signatures, scope, local index ordering and compatibility with any remaining active pins. -This increment retains accepted payloads, producer tails, catalog selections, -descriptors, manifests and image files. It does not yet advance the catalog to -its next plan, retire payload metadata, reclaim disk or enable production -ingestion. Those steps must preserve retry receipts and active image/read -ownership before they can establish bounded sustained ingestion. +Closure now also retires its exact accepted payload range as described in +[source retirement](wal-source-retirement.md), preserving producer receipts and +sequence boundaries. Catalog selections, descriptors, manifests and existing +image files remain retained. Catalog advancement, physical disk reclamation and +production ingestion are still separate work. Tests exercise closure after real Lance publication, dropped closure replies, source snapshots, full six-process restart, delayed pin rejection and exact diff --git a/docs/design/wal-source-pins.md b/docs/design/wal-source-pins.md index e0bec23..36fa954 100644 --- a/docs/design/wal-source-pins.md +++ b/docs/design/wal-source-pins.md @@ -34,7 +34,7 @@ Pins survive local checkpoints, received snapshots and retained-storage restart; checkpoint validation verifies both signatures and complete source prefixes. Command and image metadata lengths and the number of active pins are bounded. -The application/checkpoint format advances to v8 and rejects older identities +The application/checkpoint format advances to v9 and rejects older identities and images. There is no in-place migration in this optional component. This increment retains pins and payloads. It does not enable release, advance diff --git a/docs/design/wal-source-retirement.md b/docs/design/wal-source-retirement.md new file mode 100644 index 0000000..0c78589 --- /dev/null +++ b/docs/design/wal-source-retirement.md @@ -0,0 +1,38 @@ +# Source payload retirement with durable retry receipts + +Applying a verified source closure now retires only the exact records selected +by its durable pin. Each affected producer lane retains a compact sequence/index +boundary, while its latest accepted metadata and original ACK index remain +independent of payload retention. Retirement walks the bounded selection; it +does not scan all accepted records, decode Arrow or copy payload buffers. + +An exact latest-sequence retry returns the original accepted index after +retirement, snapshot and restart, without recreating its payload. Changed +content at that sequence remains rejected. New writes continue from the latest +producer sequence and can use the released accepted-record capacity. Records +for other tables and a suffix outside the closed selection remain retained. + +Application identity and image format advance to v9, without in-place migration +from older identities. A retired boundary must belong to an existing producer +receipt and a verified source closure. Retained records must continue from that +boundary without gaps and end at the latest receipt. A producer with no retained +records must have its latest sequence and original index exactly at its retired +boundary. Missing payloads do not justify arbitrary or fabricated receipt state. +Boundary capacity is bounded by producer capacity and adds no per-session state. + +New application images omit retired bodies and may have an empty payload pack +while preserving producer receipts. Existing captured images and in-flight +checkpoint builders retain their own data ownership. Tests exercise a builder +captured before closure but completed afterward, an older image read across +installation of an empty image and restart, restored retry behavior, interleaved +table and suffix preservation, and new writes after a full record cap is freed. +Real Lance six-process tests verify an empty new source payload image, full +restart, the original ACK on retry and actual published Lance payloads. + +This is logical retirement and smaller replacement images. Existing Raft-engine +files and older images are not physically collected yet. Producer identities are +also retained within their configured bound. Catalog progression, coordinated +manifest history retention, production service integration and sustained +throughput/disk bounds remain unfinished. + +Refs #378.