feat(wal): close source pins with durable publication fences

crates/lance-context-wal/src/replication.rs

@@ -93,7 +93,7 @@ impl Limits {
             "invalid replication limits"
         );
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v7",
+            "lance-context/replicated-append/v8",
             self,
         ))?)
     }
@@ -458,6 +458,14 @@ impl Node {
         self.propose(group, request, batch)
     }
 
+    /// Persist a catalog-authorized source closure and reject delayed pin
+    /// requests for the closed plan. Bodies remain retained until retirement.
+    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)?;
+        self.propose(group, request, batch)
+    }
+
     /// Local applied view, not a quorum read or proof that an owner is current.
     pub fn publisher(&self, table: Uuid) -> Result<Option<&publication::Table>> {
         self.healthy()?;

crates/lance-context-wal/src/replication/checkpoint.rs

@@ -14,8 +14,8 @@ use std::{
     path::PathBuf,
 };
 
-const FORMAT: u32 = 7;
-const MAGIC: &[u8; 8] = b"LCWALCP7";
+const FORMAT: u32 = 8;
+const MAGIC: &[u8; 8] = b"LCWALCP8";
 pub const CHUNK_BYTES: usize = 1024 * 1024;
 
 fn codec() -> impl Options {
@@ -73,6 +73,7 @@ struct Descriptor {
     producers: u64,
     tables: u64,
     pins: u64,
+    closed: u64,
     metadata_bytes: u64,
     payload_bytes: u64,
     metadata_digest: [u8; 32],
@@ -89,6 +90,7 @@ fn metadata_limit(limits: &Limits) -> u64 {
         + limits.max_tables.min(limits.max_active_plans) * (source::MAX_PIN_BYTES as u64 + 4)
         + limits.max_manifest_records * 2048
         + limits.max_tables * 256
+        + limits.max_tables * (source::MAX_CLOSED_BYTES as u64 + 4)
         + 1024
 }
 
@@ -103,6 +105,7 @@ impl Descriptor {
                 && self.producers <= self.records
                 && self.tables <= limits.max_tables
                 && self.pins <= limits.max_tables.min(limits.max_active_plans)
+                && self.closed <= limits.max_tables
                 && (group == publication::control_group(layout) || self.tables == 0)
                 && self.metadata_bytes <= metadata_limit(limits)
                 && self.payload_bytes <= self.records * limits.max_record_bytes,
@@ -449,6 +452,7 @@ pub(super) fn build(
         producers: state.producers.len() as u64,
         tables: state.catalog.tables.len() as u64,
         pins: state.pins.len() as u64,
+        closed: state.closed.len() as u64,
         metadata_bytes: 0,
         payload_bytes,
         metadata_digest: [0; 32],
@@ -462,6 +466,7 @@ pub(super) fn build(
     write_u64(&mut metadata, descriptor.producers)?;
     write_u64(&mut metadata, descriptor.tables)?;
     write_u64(&mut metadata, descriptor.pins)?;
+    write_u64(&mut metadata, descriptor.closed)?;
     for (index, meta) in &state.accepted {
         write_meta(&mut metadata, *index, meta)?;
     }
@@ -474,6 +479,9 @@ pub(super) fn build(
     for pin in state.pins.values() {
         write_frame(&mut metadata, &source::encode_pin(pin)?)?;
     }
+    for closed in state.closed.values() {
+        write_frame(&mut metadata, &source::encode_closed(closed)?)?;
+    }
     (descriptor.metadata_bytes, descriptor.metadata_digest) = metadata.finish()?;
     let mut pack = Writer::new(&paths[1], payload_bytes)?;
     for (index, meta) in &state.accepted {
@@ -521,7 +529,8 @@ pub(super) fn load(
         read_u64(&mut metadata)? == descriptor.records
             && read_u64(&mut metadata)? == descriptor.producers
             && read_u64(&mut metadata)? == descriptor.tables
-            && read_u64(&mut metadata)? == descriptor.pins,
+            && read_u64(&mut metadata)? == descriptor.pins
+            && read_u64(&mut metadata)? == descriptor.closed,
         "image metadata count mismatch"
     );
     let mut state = State {
@@ -573,6 +582,18 @@ pub(super) fn load(
         state.pins.insert(table, Arc::new(pin));
         previous = Some(table);
     }
+    let mut previous = None;
+    for _ in 0..descriptor.closed {
+        let bytes = read_dynamic_frame(&mut metadata, source::MAX_CLOSED_BYTES)?;
+        let closed = source::decode_closed(&bytes)?;
+        let table = closed.permit.closure().table;
+        ensure!(
+            previous.is_none_or(|old| old < table),
+            "unordered or duplicate source closure"
+        );
+        state.closed.insert(table, Arc::new(closed));
+        previous = Some(table);
+    }
     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

@@ -396,7 +396,17 @@ fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() ->
     let original = fs::read(&path)?;
     // Locate frames from their encoded lengths, without assuming a fixed header.
     let binding_len = u32::from_le_bytes(original[8..12].try_into()?) as usize;
-    let mut offset = 12 + binding_len + 4 * 8;
+    let mut header = &original[12 + binding_len..];
+    for count in [
+        state.accepted.len(),
+        state.producers.len(),
+        state.catalog.tables.len(),
+        state.pins.len(),
+        state.closed.len(),
+    ] {
+        assert_eq!(read_u64(&mut header)?, count as u64);
+    }
+    let mut offset = original.len() - header.len();
     for _ in 0..state.accepted.len() + state.producers.len() {
         offset += 8; // accepted index
         let len = u32::from_le_bytes(original[offset..offset + 4].try_into()?) as usize;

crates/lance-context-wal/src/replication/publication/outcome.rs

@@ -61,7 +61,7 @@ pub struct Published {
     pub manifest: Arc<ManifestRecord>,
 }
 impl Published {
-    pub(super) fn validate(&self, table: &Table, cut: u64) -> Result<()> {
+    pub(in crate::replication) fn validate(&self, table: &Table, cut: u64) -> Result<()> {
         let manifests = table
             .manifests
             .as_ref()

crates/lance-context-wal/src/replication/query.rs

@@ -8,6 +8,15 @@ mod tests;
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Query {
+    ClosePermit {
+        table: Uuid,
+        transaction: Uuid,
+        plan_index: u64,
+        source_group: u64,
+    },
+    SourceClosure {
+        table: Uuid,
+    },
     Manifest {
         table: Uuid,
         version: Option<u64>,
@@ -37,6 +46,8 @@ pub enum Query {
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Value {
+    ClosePermit(Option<Box<source::ClosurePermit>>),
+    SourceClosure(Option<Arc<source::Closed>>),
     Manifest(Option<Arc<publication::ManifestRecord>>),
     Publisher(Option<Box<publication::Table>>),
     Accepted(Page),
@@ -83,7 +94,13 @@ impl Query {
                     && version.is_none_or(|v| v > 0 && v < u64::MAX),
                 "invalid manifest query scope"
             ),
-            Self::Pin {
+            Self::ClosePermit {
+                table,
+                transaction,
+                plan_index,
+                source_group,
+            }
+            | Self::Pin {
                 table,
                 transaction,
                 plan_index,
@@ -97,7 +114,9 @@ impl Query {
                     && group == publication::control_group(layout),
                 "invalid catalog dependency query"
             ),
-            Self::SourcePin { table } => ensure!(!table.is_nil(), "nil source table"),
+            Self::SourceClosure { table } | Self::SourcePin { table } => {
+                ensure!(!table.is_nil(), "nil source table")
+            }
             Self::Publisher { table } => ensure!(
                 !table.is_nil() && group == publication::control_group(layout),
                 "invalid publisher query scope"
@@ -127,6 +146,37 @@ impl Query {
         signer: Option<&source::CatalogSigner>,
     ) -> Result<Value> {
         Ok(match self {
+            Self::SourceClosure { table } => Value::SourceClosure(state.closed.get(table).cloned()),
+            Self::ClosePermit {
+                table,
+                transaction,
+                plan_index,
+                source_group,
+            } => {
+                let selected = state.catalog.tables.get(table).filter(|table| {
+                    table.progress.published.is_some()
+                        && table.progress.pending.as_ref().is_some_and(|p| {
+                            p.index == *plan_index
+                                && p.plan.transaction == *transaction
+                                && p.plan.sources.contains_key(source_group)
+                        })
+                });
+                Value::ClosePermit(
+                    selected
+                        .map(|table| {
+                            source::ClosurePermit::issue(
+                                signer.context("catalog closure signer unavailable")?,
+                                cut.clone(),
+                                layout,
+                                limits,
+                                table,
+                                *source_group,
+                            )
+                            .map(Box::new)
+                        })
+                        .transpose()?,
+                )
+            }
             Self::Manifest { table, version } => Value::Manifest(
                 state
                     .catalog

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

@@ -6,6 +6,9 @@ use crate::{ingest_contract::IngestContract, record::RecordKind};
 use bincode::Options;
 use bytes::Bytes;
 use ring::signature::{Ed25519KeyPair, KeyPair, UnparsedPublicKey, ED25519};
+mod closure;
+pub(super) use closure::{decode_closed, encode_closed, MAX_CLOSED_BYTES};
+pub use closure::{Closed, Closure, ClosurePermit};
 
 #[cfg(test)]
 mod tests;
@@ -166,21 +169,25 @@ impl Permit {
 
     pub(super) fn batch(self, layout: &Layout, limits: &Limits) -> Result<Batch> {
         self.verify(layout, limits, self.selection.group)?;
-        let bytes = codec(MAX_BYTES).serialize(&self)?;
-        ensure!(bytes.len() <= MAX_BYTES, "oversized source pin command");
-        let mut batch = Batch::new(
+        let (table, transaction, plan_index) = (
             self.selection.table,
-            0,
             self.selection.transaction,
             self.selection.plan_index,
-            Bytes::from(bytes),
         );
+        let bytes = codec(MAX_BYTES).serialize(&Command::Pin(self))?;
+        ensure!(bytes.len() <= MAX_BYTES, "oversized source pin command");
+        let mut batch = Batch::new(table, 0, transaction, plan_index, Bytes::from(bytes));
         batch.meta.kind = RecordKind::Source;
         Ok(batch)
     }
 }
 
-pub(super) fn decode(bytes: &[u8]) -> Result<Permit> {
+#[derive(Serialize, Deserialize)]
+pub(super) enum Command {
+    Pin(Permit),
+    Close(ClosurePermit),
+}
+pub(super) fn decode(bytes: &[u8]) -> Result<Command> {
     ensure!(bytes.len() <= MAX_BYTES, "oversized source pin command");
     let permit = codec(MAX_BYTES).deserialize(bytes)?;
     ensure!(
@@ -196,14 +203,24 @@ pub(super) fn validate_record(
     meta: &BatchMeta,
     body: &[u8],
 ) -> Result<()> {
-    let permit = decode(body)?;
-    permit.verify(layout, limits, group)?;
-    let s = permit.selection();
+    let command = decode(body)?;
+    let (table, transaction, plan_index) = match command {
+        Command::Pin(permit) => {
+            permit.verify(layout, limits, group)?;
+            let s = permit.selection();
+            (s.table, s.transaction, s.plan_index)
+        }
+        Command::Close(permit) => {
+            permit.verify(layout, limits, group)?;
+            let c = permit.closure();
+            (c.table, c.transaction, c.plan_index)
+        }
+    };
     ensure!(
         meta.kind == RecordKind::Source
-            && meta.stream == s.table
-            && meta.producer == s.transaction
-            && meta.sequence == s.plan_index
+            && meta.stream == table
+            && meta.producer == transaction
+            && meta.sequence == plan_index
             && meta.bucket == 0,
         "source pin envelope mismatch"
     );
@@ -220,20 +237,28 @@ pub enum Rejection {
     Pending,
     Prefix,
     Capacity,
+    Closed,
 }
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Reply {
     Pinned { index: u64 },
     Rejected(Rejection),
+    Closed { index: u64 },
 }
 
-fn complete_prefix(accepted: &BTreeMap<u64, BatchMeta>, selection: &Selection, index: u64) -> bool {
+fn complete_prefix(state: &State, selection: &Selection, index: u64) -> bool {
     let range = &selection.range;
-    // No release/cursor advancement exists in this increment. The retained
-    // source prefix starts at zero until durable resolution is integrated.
-    range.after == 0
+    let previous = state
+        .closed
+        .get(&selection.table)
+        .map(|c| c.permit.closure());
+    range.after == previous.map_or(0, |c| c.through)
+        && previous.is_none_or(|c| {
+            selection.plan_index > c.plan_index && selection.read_version >= c.version
+        })
         && range.through < index
-        && accepted
+        && state
+            .accepted
             .range((
                 std::ops::Bound::Excluded(range.after),
                 std::ops::Bound::Included(range.through),
@@ -242,16 +267,34 @@ fn complete_prefix(accepted: &BTreeMap<u64, BatchMeta>, selection: &Selection, i
             .map(|(i, meta)| (*i, meta))
             .eq(range.records.iter().map(|(i, meta)| (*i, meta)))
 }
-pub(super) fn apply(state: &mut State, permit: Permit, index: u64, limits: &Limits) -> Reply {
+pub(super) fn apply(
+    state: &mut State,
+    command: Command,
+    index: u64,
+    limits: &Limits,
+) -> Result<Reply> {
+    match command {
+        Command::Pin(permit) => Ok(pin(state, permit, index, limits)),
+        Command::Close(permit) => closure::apply(state, permit, index, limits),
+    }
+}
+fn pin(state: &mut State, permit: Permit, index: u64, limits: &Limits) -> Reply {
     let selection = permit.selection();
+    if state
+        .closed
+        .get(&selection.table)
+        .is_some_and(|closed| selection.plan_index <= closed.permit.closure().plan_index)
+    {
+        return Reply::Rejected(Rejection::Closed);
+    }
     if let Some(old) = state.pins.get(&selection.table) {
         return if old.permit.selection() == selection {
             Reply::Pinned { index: old.index }
         } else {
             Reply::Rejected(Rejection::Pending)
         };
     }
-    if !complete_prefix(&state.accepted, selection, index) {
+    if !complete_prefix(state, selection, index) {
         return Reply::Rejected(Rejection::Prefix);
     }
     if state.pins.len() as u64 >= limits.max_tables.min(limits.max_active_plans) {
@@ -269,6 +312,7 @@ pub(super) fn validate_state(
     cut: u64,
     limits: &Limits,
 ) -> Result<()> {
+    closure::validate_state(state, layout, group, cut, limits)?;
     ensure!(
         state.pins.len() as u64 <= limits.max_tables.min(limits.max_active_plans),
         "source pin capacity"
@@ -279,7 +323,7 @@ pub(super) fn validate_state(
             *table == pin.permit.selection.table
                 && pin.index > 0
                 && pin.index <= cut
-                && complete_prefix(&state.accepted, pin.permit.selection(), pin.index),
+                && complete_prefix(state, pin.permit.selection(), pin.index),
             "invalid checkpoint source pin or prefix"
         );
     }