feat(wal): persist publication outcomes from selected manifests

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

@@ -93,7 +93,7 @@ impl Limits {
             "invalid replication limits"
         );
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v6",
+            "lance-context/replicated-append/v7",
             self,
         ))?)
     }
@@ -439,6 +439,17 @@ impl Node {
         self.propose(publication::control_group(&self.layout), request, batch)
     }
 
+    /// Durably resolve the selected manifest against its exact pending plan.
+    /// This retains all source dependencies; it does not authorize reclamation.
+    pub fn confirm_publication(
+        &mut self,
+        request: u64,
+        confirmation: publication::ConfirmPublication,
+    ) -> Result<()> {
+        let batch = confirmation.batch(&self.layout)?;
+        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

@@ -14,8 +14,8 @@ use std::{
     path::PathBuf,
 };
 
-const FORMAT: u32 = 6;
-const MAGIC: &[u8; 8] = b"LCWALCP6";
+const FORMAT: u32 = 7;
+const MAGIC: &[u8; 8] = b"LCWALCP7";
 pub const CHUNK_BYTES: usize = 1024 * 1024;
 
 fn codec() -> impl Options {
@@ -85,7 +85,7 @@ fn metadata_limit(limits: &Limits) -> u64 {
     (limits.max_records + limits.max_producers) * 268
         + limits.max_tables * (publication::MAX_BASE_TABLE_BYTES as u64 + 4)
         + limits.max_tables.min(limits.max_active_plans)
-            * (128 * 1024 + limits.max_descriptor_bytes + 1024)
+            * (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

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

@@ -1,19 +1,21 @@
 //! Replicated table bindings and publisher epochs. These commands run through
 //! the same durable Raft path as appends; a local view is not an active lease.
 //! Immutable plans and prepared descriptors survive takeover. Source pins live
-//! in the source module; manifest outcomes and retirement remain separate work.
+//! in the source module; source closure and retirement remain separate work.
 //! No Lance manifest is changed here.
 use super::*;
 use crate::{ingest_contract::IngestContract, record::RecordKind};
 use bincode::Options;
 use bytes::Bytes;
 mod manifest;
+mod outcome;
 mod plan;
 mod prepared;
 pub use manifest::{
     BootstrapManifest, ManifestIndex, ManifestObject, ManifestRecord, ReserveManifest,
     MAX_MANIFEST_VERSIONS,
 };
+pub use outcome::{ConfirmPublication, Published};
 pub use plan::{PendingPlan, Plan, PlanRequest, Progress, SourceRange, MAX_INPUTS};
 pub use prepared::{Descriptors, PrepareRequest, Prepared, MAX_DESCRIPTOR_BYTES};
 #[cfg(test)]
@@ -25,7 +27,8 @@ pub(super) const MAX_BASE_TABLE_BYTES: usize = MAX_ACQUIRE_BYTES + 2048;
 pub(super) const MAX_TABLE_BYTES: usize = MAX_BASE_TABLE_BYTES
     + plan::MAX_PLAN_BYTES
     + prepared::MAX_PREPARE_BYTES
-    + manifest::MAX_MANIFEST_INDEX_BYTES;
+    + manifest::MAX_MANIFEST_INDEX_BYTES
+    + outcome::MAX_OUTCOME_BYTES;
 fn codec(limit: usize) -> impl Options {
     bincode::DefaultOptions::new()
         .with_fixint_encoding()
@@ -76,6 +79,7 @@ enum Command {
     Prepare(Box<PrepareRequest>),
     BootstrapManifest(Box<BootstrapManifest>),
     ReserveManifest(Box<ReserveManifest>),
+    ConfirmPublication(Box<ConfirmPublication>),
 }
 impl Command {
     fn identity(&self) -> (Uuid, Uuid, u64) {
@@ -85,6 +89,9 @@ impl Command {
             Self::Prepare(p) => (p.authority.table, p.authority.owner, p.authority.epoch),
             Self::BootstrapManifest(p) => (p.authority.table, p.authority.owner, p.authority.epoch),
             Self::ReserveManifest(p) => (p.authority.table, p.authority.owner, p.authority.epoch),
+            Self::ConfirmPublication(p) => {
+                (p.authority.table, p.authority.owner, p.authority.epoch)
+            }
         }
     }
     fn byte_limit(&self) -> usize {
@@ -95,13 +102,14 @@ impl Command {
             Self::BootstrapManifest(_) | Self::ReserveManifest(_) => {
                 manifest::MAX_MANIFEST_COMMAND_BYTES
             }
+            Self::ConfirmPublication(_) => outcome::MAX_OUTCOME_BYTES,
         }
     }
     fn batch(self, layout: &Layout) -> Result<Batch> {
         let (table, owner, sequence) = self.identity();
         let limit = self.byte_limit();
         let envelope = Envelope {
-            version: 4,
+            version: 5,
             cluster: layout.cluster,
             layout: layout.fingerprint()?,
             group: control_group(layout),
@@ -168,6 +176,7 @@ pub enum Rejection {
     PreparedConflict,
     ManifestIdentity,
     ManifestPredecessor,
+    PublicationIdentity,
 }
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Reply {
@@ -177,6 +186,7 @@ pub enum Reply {
     Prepared { index: u64 },
     ManifestReserved(Arc<ManifestRecord>),
     ManifestTaken(Arc<ManifestRecord>),
+    Published(Arc<Published>),
 }
 
 #[derive(Serialize, Deserialize)]
@@ -210,7 +220,7 @@ pub(super) fn decode(bytes: &[u8]) -> Result<Envelope> {
         "publisher command too large"
     );
     let envelope: Envelope = codec(MAX_COMMAND_BYTES).deserialize(bytes)?;
-    ensure!(envelope.version == 4, "unsupported publisher command");
+    ensure!(envelope.version == 5, "unsupported publisher command");
     ensure!(
         bytes.len() <= envelope.command.byte_limit(),
         "publisher command type byte limit"
@@ -255,6 +265,7 @@ pub(super) fn validate_record(
         Command::Prepare(p) => p.validate(layout, limits)?,
         Command::BootstrapManifest(p) => p.validate(layout)?,
         Command::ReserveManifest(p) => p.validate(layout)?,
+        Command::ConfirmPublication(p) => p.validate(layout)?,
     }
     Ok(())
 }
@@ -274,6 +285,9 @@ impl Catalog {
             Command::ReserveManifest(request) => {
                 return self.reserve_manifest(*request, index, limits)
             }
+            Command::ConfirmPublication(request) => {
+                return self.confirm_publication(*request, index)
+            }
             Command::Acquire(a) => *a,
         };
         let old = self.tables.get(&a.table);

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

@@ -3,7 +3,7 @@
 use super::*;
 
 #[cfg(test)]
-mod tests;
+pub(super) mod tests;
 
 pub const MAX_MANIFEST_VERSIONS: usize = 128;
 pub(super) const MAX_MANIFEST_BYTES: usize = 2048;

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

@@ -15,7 +15,9 @@ fn object(version: u64, staging: bool) -> ManifestObject {
     }
 }
 
-fn fixture(limits: &Limits) -> Result<(Catalog, BootstrapManifest, ReserveManifest)> {
+pub(in crate::replication::publication) fn fixture(
+    limits: &Limits,
+) -> Result<(Catalog, BootstrapManifest, ReserveManifest)> {
     let layout = crate::replication::tests::layout();
     let mut catalog = Catalog::default();
     let acquire = super::super::tests::input();

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

@@ -0,0 +1,152 @@
+//! Durable publication resolution from the catalog's selected manifest. Source
+//! selection and descriptors stay retained until a separate closure protocol.
+use super::*;
+
+#[cfg(test)]
+mod tests;
+
+pub(super) const MAX_OUTCOME_BYTES: usize = manifest::MAX_MANIFEST_BYTES + 128;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ConfirmPublication {
+    pub authority: Authority,
+    pub incarnation: Uuid,
+    pub transaction: Uuid,
+    pub plan_index: u64,
+    pub prepared_index: u64,
+    pub descriptor_digest: [u8; 32],
+    pub target: u64,
+    pub manifest_identity: Uuid,
+}
+impl ConfirmPublication {
+    pub(in crate::replication) fn batch(self, layout: &Layout) -> Result<Batch> {
+        self.validate(layout)?;
+        Command::ConfirmPublication(Box::new(self)).batch(layout)
+    }
+
+    pub(super) fn validate(&self, layout: &Layout) -> Result<()> {
+        self.authority.validate(layout)?;
+        ensure!(
+            !self.incarnation.is_nil()
+                && !self.transaction.is_nil()
+                && !self.manifest_identity.is_nil()
+                && self.plan_index > 0
+                && self.prepared_index > self.plan_index
+                && self.prepared_index < u64::MAX
+                && self.descriptor_digest != [0; 32]
+                && self.target > 1
+                && self.target < (1u64 << 63),
+            "invalid publication confirmation identity"
+        );
+        Ok(())
+    }
+
+    fn matches(&self, incarnation: Uuid, manifest: &ManifestRecord) -> bool {
+        self.incarnation == incarnation
+            && self.transaction == manifest.transaction
+            && self.plan_index == manifest.plan_index
+            && self.prepared_index == manifest.prepared_index
+            && self.descriptor_digest == manifest.descriptor_digest
+            && self.target == manifest.version
+            && self.manifest_identity == manifest.identity
+    }
+}
+
+/// One outcome per unresolved plan. This shares the immutable selected record;
+/// cloning a table view does not copy descriptors or any source payloads.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Published {
+    pub index: u64,
+    pub incarnation: Uuid,
+    pub manifest: Arc<ManifestRecord>,
+}
+impl Published {
+    pub(super) fn validate(&self, table: &Table, cut: u64) -> Result<()> {
+        let manifests = table
+            .manifests
+            .as_ref()
+            .context("outcome without manifest index")?;
+        let pending = table
+            .progress
+            .pending
+            .as_ref()
+            .context("outcome without pending selection")?;
+        let prepared = table
+            .progress
+            .prepared
+            .as_ref()
+            .context("outcome without prepared descriptors")?;
+        let selected = manifests
+            .versions
+            .get(&self.manifest.version)
+            .context("outcome manifest missing")?;
+        ensure!(
+            self.incarnation == manifests.incarnation
+                && self.manifest == *selected
+                && self.manifest.version == manifests.head
+                && self.manifest.version == pending.plan.target_version()?
+                && self.manifest.transaction == pending.plan.transaction
+                && self.manifest.plan_index == pending.index
+                && self.manifest.prepared_index == prepared.index
+                && self.manifest.descriptor_digest == prepared.descriptors.digest
+                && self.index > self.manifest.index
+                && self.index <= cut
+                && self.index != table.authority.grant_index,
+            "publication outcome disagrees with selected manifest or pending plan"
+        );
+        Ok(())
+    }
+}
+
+impl Catalog {
+    pub(super) fn confirm_publication(&mut self, request: ConfirmPublication, index: u64) -> Reply {
+        let Some(table) = self.tables.get(&request.authority.table) else {
+            return Reply::Rejected(Rejection::Authority);
+        };
+        // Exact resolution remains available after publisher takeover. It
+        // reports an existing outcome and grants no authority to change state.
+        if let Some(published) = &table.progress.published {
+            return if request.matches(published.incarnation, &published.manifest) {
+                Reply::Published(published.clone())
+            } else {
+                Reply::Rejected(Rejection::PublicationIdentity)
+            };
+        }
+        if request.authority != table.authority {
+            return Reply::Rejected(Rejection::Authority);
+        }
+        let Some(manifests) = &table.manifests else {
+            return Reply::Rejected(Rejection::ManifestIdentity);
+        };
+        let Some(selected) = manifests.versions.get(&request.target) else {
+            return Reply::Rejected(Rejection::ManifestIdentity);
+        };
+        let (Some(pending), Some(prepared)) = (&table.progress.pending, &table.progress.prepared)
+        else {
+            return Reply::Rejected(Rejection::PlanIdentity);
+        };
+        if !request.matches(manifests.incarnation, selected)
+            || selected.version != manifests.head
+            || pending.plan.target_version().ok() != Some(selected.version)
+            || selected.transaction != pending.plan.transaction
+            || selected.plan_index != pending.index
+            || selected.prepared_index != prepared.index
+            || selected.descriptor_digest != prepared.descriptors.digest
+        {
+            return Reply::Rejected(Rejection::PublicationIdentity);
+        }
+        let published = Arc::new(Published {
+            index,
+            incarnation: manifests.incarnation,
+            manifest: selected.clone(),
+        });
+        // Do not erase the selection, advance source cursors, or release pins.
+        // Closure must preserve independent source indices and durable fences.
+        self.tables
+            .get_mut(&request.authority.table)
+            .expect("known table")
+            .progress
+            .published = Some(published.clone());
+        Reply::Published(published)
+    }
+}