feat(wal): retire closed source payloads while retaining retry receipts

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)?;

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)?;

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);
     }

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 })
 }
 

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> {
+        Self::with_policy(max_active_plans, 100_000)
+    }
+    fn with_policy(max_active_plans: u64, max_records: u64) -> Result<Self> {
         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<u64> {
-        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<u64> {
+        for seq in 1..=background {
             self.append(Uuid::from_u128(999), seq)?;
         }
         self.append(self.table, 1)?;

crates/lance-context-wal/src/replication/source/tests/closed.rs

@@ -16,7 +16,13 @@ fn publisher(events: Events) -> Result<publication::Reply> {
 }
 
 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<Reply> {
 }
 
 #[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()?;