feat(wal): advance catalog cursors after source closure

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

@@ -521,6 +521,18 @@ impl Node {
         self.propose(publication::control_group(&self.layout), request, batch)
     }
 
+    /// Advance independent source cursors only after every selected source
+    /// proves the exact durable closure. Completion permits the next plan;
+    /// physical file reclamation still requires a durable application image.
+    pub fn advance_publication(
+        &mut self,
+        request: u64,
+        advancement: publication::AdvancePublication,
+    ) -> Result<()> {
+        let batch = advancement.batch(&self.layout, &self.limits)?;
+        self.propose(publication::control_group(&self.layout), request, batch)
+    }
+
     /// Verify a catalog quorum-read attestation and persist the exact complete
     /// table prefix in its source group. Completion, never enqueue, pins it.
     pub fn pin_source(&mut self, request: u64, permit: source::Permit) -> Result<()> {

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

@@ -90,7 +90,7 @@ fn metadata_limit(limits: &Limits) -> u64 {
             * (128 * 1024 + limits.max_descriptor_bytes + 4096)
         + 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 * (256 + publication::advance_metadata_bytes())
         + limits.max_tables * (source::MAX_CLOSED_BYTES as u64 + 4)
         + limits.max_producers * (state::MAX_RETIRED_BYTES as u64 + 4)
         + 1024

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

@@ -7,10 +7,13 @@ use super::*;
 use crate::{ingest_contract::IngestContract, record::RecordKind};
 use bincode::Options;
 use bytes::Bytes;
+mod advance;
 mod manifest;
 mod outcome;
 mod plan;
 mod prepared;
+mod table_codec;
+pub use advance::{AdvancePublication, Advanced};
 pub use manifest::{
     BootstrapManifest, ManifestIndex, ManifestObject, ManifestRecord, ReserveManifest,
     MAX_MANIFEST_VERSIONS,
@@ -28,7 +31,11 @@ pub(super) const MAX_TABLE_BYTES: usize = MAX_BASE_TABLE_BYTES
     + plan::MAX_PLAN_BYTES
     + prepared::MAX_PREPARE_BYTES
     + manifest::MAX_MANIFEST_INDEX_BYTES
-    + outcome::MAX_OUTCOME_BYTES;
+    + outcome::MAX_OUTCOME_BYTES
+    + advance::MAX_ADVANCED_BYTES;
+pub(super) const fn advance_metadata_bytes() -> u64 {
+    advance::MAX_ADVANCED_BYTES as u64
+}
 fn codec(limit: usize) -> impl Options {
     bincode::DefaultOptions::new()
         .with_fixint_encoding()
@@ -80,6 +87,7 @@ enum Command {
     BootstrapManifest(Box<BootstrapManifest>),
     ReserveManifest(Box<ReserveManifest>),
     ConfirmPublication(Box<ConfirmPublication>),
+    AdvancePublication(Box<AdvancePublication>),
 }
 impl Command {
     fn identity(&self) -> (Uuid, Uuid, u64) {
@@ -92,6 +100,9 @@ impl Command {
             Self::ConfirmPublication(p) => {
                 (p.authority.table, p.authority.owner, p.authority.epoch)
             }
+            Self::AdvancePublication(p) => {
+                (p.authority.table, p.authority.owner, p.authority.epoch)
+            }
         }
     }
     fn byte_limit(&self) -> usize {
@@ -103,6 +114,7 @@ impl Command {
                 manifest::MAX_MANIFEST_COMMAND_BYTES
             }
             Self::ConfirmPublication(_) => outcome::MAX_OUTCOME_BYTES,
+            Self::AdvancePublication(_) => advance::MAX_ADVANCE_BYTES,
         }
     }
     fn batch(self, layout: &Layout) -> Result<Batch> {
@@ -177,6 +189,7 @@ pub enum Rejection {
     ManifestIdentity,
     ManifestPredecessor,
     PublicationIdentity,
+    SourceClosure,
 }
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Reply {
@@ -187,6 +200,7 @@ pub enum Reply {
     ManifestReserved(Arc<ManifestRecord>),
     ManifestTaken(Arc<ManifestRecord>),
     Published(Arc<Published>),
+    Advanced(Arc<Advanced>),
 }
 
 #[derive(Serialize, Deserialize)]
@@ -266,6 +280,7 @@ pub(super) fn validate_record(
         Command::BootstrapManifest(p) => p.validate(layout)?,
         Command::ReserveManifest(p) => p.validate(layout)?,
         Command::ConfirmPublication(p) => p.validate(layout)?,
+        Command::AdvancePublication(p) => p.validate(layout, limits)?,
     }
     Ok(())
 }
@@ -288,6 +303,9 @@ impl Catalog {
             Command::ConfirmPublication(request) => {
                 return self.confirm_publication(*request, index)
             }
+            Command::AdvancePublication(request) => {
+                return self.advance_publication(*request, index)
+            }
             Command::Acquire(a) => *a,
         };
         let old = self.tables.get(&a.table);
@@ -392,14 +410,8 @@ impl Catalog {
     }
 }
 pub(super) fn encode_table(table: &Table) -> Result<Vec<u8>> {
-    Ok(codec(MAX_TABLE_BYTES).serialize(table)?)
+    table_codec::encode(table)
 }
 pub(super) fn decode_table(bytes: &[u8]) -> Result<Table> {
-    ensure!(bytes.len() <= MAX_TABLE_BYTES, "publisher table byte limit");
-    let table = codec(MAX_TABLE_BYTES).deserialize(bytes)?;
-    ensure!(
-        encode_table(&table)? == bytes,
-        "noncanonical publisher table"
-    );
-    Ok(table)
+    table_codec::decode(bytes)
 }

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

@@ -0,0 +1,183 @@
+//! Atomic source-cursor advancement after all selected source quorums close.
+use super::*;
+use crate::replication::source::{ClosedProof, Selection};
+
+#[cfg(test)]
+mod tests;
+
+pub(super) const MAX_ADVANCE_BYTES: usize = 64 * 4096 + 8192;
+pub(super) const MAX_ADVANCED_BYTES: usize = 8192;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct AdvancePublication {
+    pub authority: Authority,
+    pub published_index: u64,
+    pub proofs: BTreeMap<u64, ClosedProof>,
+}
+
+impl AdvancePublication {
+    pub(in crate::replication) fn batch(self, layout: &Layout, limits: &Limits) -> Result<Batch> {
+        self.validate(layout, limits)?;
+        Command::AdvancePublication(Box::new(self)).batch(layout)
+    }
+
+    pub(super) fn validate(&self, layout: &Layout, limits: &Limits) -> Result<()> {
+        self.authority.validate(layout)?;
+        ensure!(
+            self.published_index > 0
+                && self.published_index < u64::MAX
+                && !self.proofs.is_empty()
+                && self.proofs.len() <= layout.groups.len()
+                && self.proofs.len() <= 64,
+            "invalid source closure set"
+        );
+        for (group, proof) in &self.proofs {
+            proof.verify(layout, limits)?;
+            let closure = proof.closed().permit.closure();
+            ensure!(
+                *group == proof.source_cut().group
+                    && closure.table == self.authority.table
+                    && closure.published_index == self.published_index,
+                "source closure request scope mismatch"
+            );
+        }
+        Ok(())
+    }
+
+    fn digest(&self) -> Result<[u8; 32]> {
+        // Bind the entire sorted set, not refreshed attestation read cuts.
+        let closures: Vec<_> = self
+            .proofs
+            .iter()
+            .map(|(group, proof)| (*group, proof.closed().permit.closure()))
+            .collect();
+        Ok(Sha256::digest(
+            codec(MAX_ADVANCE_BYTES)
+                .serialize(&("lance-context/catalog-advanced-closures/v1", closures))?,
+        )
+        .into())
+    }
+}
+
+/// One bounded receipt per table. It shares the already selected manifest and
+/// retains no source payloads, input metadata or prepared fragment descriptors.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Advanced {
+    pub index: u64,
+    pub publication: Arc<Published>,
+    pub sources: BTreeMap<u64, u64>,
+    pub closure_digest: [u8; 32],
+}
+
+impl Advanced {
+    pub(super) fn validate(&self, table: &Table, cut: u64) -> Result<()> {
+        let manifests = table
+            .manifests
+            .as_ref()
+            .context("advancement without manifest index")?;
+        let published = &self.publication;
+        let selected = &published.manifest;
+        ensure!(
+            published.incarnation == manifests.incarnation
+                && manifests.versions.get(&selected.version) == Some(selected)
+                && selected.version > manifests.initial_version
+                && selected.version == table.progress.minimum_read_version
+                && published.index > selected.index
+                && self.index > published.index
+                && self.index <= cut
+                && self.index != table.authority.grant_index
+                && self.closure_digest != [0; 32]
+                && !self.sources.is_empty()
+                && self.sources.len() <= 64
+                && self.sources.iter().all(|(group, through)| {
+                    *through > 0
+                        && *through < u64::MAX
+                        && table.progress.checkpoint.get(group) == Some(through)
+                })
+                && table
+                    .progress
+                    .pending
+                    .as_ref()
+                    .is_none_or(|p| p.index > self.index),
+            "invalid catalog advancement receipt"
+        );
+        Ok(())
+    }
+}
+
+impl Catalog {
+    pub(super) fn advance_publication(&mut self, request: AdvancePublication, index: u64) -> Reply {
+        let Some(table) = self.tables.get(&request.authority.table) else {
+            return Reply::Rejected(Rejection::Authority);
+        };
+        let Ok(digest) = request.digest() else {
+            return Reply::Rejected(Rejection::SourceClosure);
+        };
+        if let Some(old) = &table.progress.advanced {
+            if old.publication.index == request.published_index && old.closure_digest == digest {
+                // A stale owner may resolve only the already completed exact
+                // operation. It cannot mutate a newer pending publication.
+                return Reply::Advanced(old.clone());
+            }
+        }
+        if table.authority != request.authority {
+            return Reply::Rejected(Rejection::Authority);
+        }
+        let (Some(pending), Some(published)) = (&table.progress.pending, &table.progress.published)
+        else {
+            return Reply::Rejected(Rejection::PublicationIdentity);
+        };
+        if published.index != request.published_index {
+            return Reply::Rejected(Rejection::PublicationIdentity);
+        }
+        if !request.proofs.keys().eq(pending.plan.sources.keys()) {
+            return Reply::Rejected(Rejection::SourceClosure);
+        }
+        for (group, range) in &pending.plan.sources {
+            let selection = Selection {
+                table: table.authority.table,
+                transaction: pending.plan.transaction,
+                plan_index: pending.index,
+                read_version: pending.plan.read_version,
+                group: *group,
+                ingest: table.ingest.clone(),
+                range: range.clone(),
+            };
+            let closure = request.proofs[group].closed().permit.closure();
+            if !closure
+                .matches_publication(&selection, published)
+                .unwrap_or(false)
+            {
+                return Reply::Rejected(Rejection::SourceClosure);
+            }
+        }
+        let advanced = Arc::new(Advanced {
+            index,
+            publication: published.clone(),
+            sources: pending
+                .plan
+                .sources
+                .iter()
+                .map(|(group, range)| (*group, range.through))
+                .collect(),
+            closure_digest: digest,
+        });
+        let progress = &mut self
+            .tables
+            .get_mut(&request.authority.table)
+            .expect("known table")
+            .progress;
+        progress.checkpoint.extend(
+            advanced
+                .sources
+                .iter()
+                .map(|(group, through)| (*group, *through)),
+        );
+        progress.minimum_read_version = advanced.publication.manifest.version;
+        progress.pending = None;
+        progress.prepared = None;
+        progress.published = None;
+        progress.advanced = Some(advanced.clone());
+        Reply::Advanced(advanced)
+    }
+}