feat(wal): persist immutable publication plans

crates/lance-context-wal/Cargo.toml

@@ -8,7 +8,7 @@ description = "Independent Rust WAL experiments owned by lance-context"
 
 [features]
 client-journal = ["dep:anyhow", "dep:raft-engine", "dep:bytes", "dep:serde", "dep:serde_json", "dep:tokio", "tokio/sync"]
-replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog", "uuid/v4"]
+replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog", "uuid/v4", "serde/rc"]
 raft-log = ["placement", "dep:raft", "dep:raft-engine", "dep:raft-proto", "dep:protobuf"]
 vectored-io = ["dep:bytes", "dep:tokio"]
 runner = ["dep:anyhow", "dep:clap", "dep:libc", "dep:serde", "dep:serde_json", "dep:sha2"]

crates/lance-context-wal/README.md

@@ -176,6 +176,11 @@ publication prerequisite, not a production Lance writer or a manifest lease.
 catalog, producer-tail and accepted-page views after a safe read-index barrier.
 They share pending capacity with proposals and do not fetch bulk payloads.
 
+[Immutable publication plans](../../docs/design/wal-publication-plans.md) retain
+the exact source vector and Lance target through takeover, snapshots and restart.
+They share bounded metadata between views; source pins, outcome resolution and
+retirement remain separate integration steps.
+
 The optional `raft-log` feature exposes `raft_log::PhysicalLog` and its cloneable
 `GroupLog` handles. One raft-engine directory hosts all groups assigned to a node
 by the existing immutable `placement::Layout`. Opening binds the full layout,

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

@@ -43,6 +43,8 @@ use state::State;
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Limits {
     pub max_tables: u64,
+    pub max_active_plans: u64,
+    pub max_plan_source_bytes: u64,
     pub max_record_bytes: u64,
     pub max_records: u64,
     pub max_producers: u64,
@@ -53,6 +55,8 @@ impl Default for Limits {
     fn default() -> Self {
         Self {
             max_tables: 1024,
+            max_active_plans: 128,
+            max_plan_source_bytes: 64 * 1024 * 1024,
             max_record_bytes: 8 * 1024 * 1024,
             max_records: 100_000,
             max_producers: 16_384,
@@ -65,6 +69,8 @@ impl Limits {
     fn identity(&self) -> Result<Vec<u8>> {
         ensure!(
             (1..=16_384).contains(&self.max_tables)
+                && (1..=256).contains(&self.max_active_plans)
+                && (1..=256 * 1024 * 1024).contains(&self.max_plan_source_bytes)
                 && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes)
                 && (1..=500_000).contains(&self.max_records)
                 && (1..=100_000).contains(&self.max_producers)
@@ -74,7 +80,7 @@ impl Limits {
             "invalid replication limits"
         );
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v2",
+            "lance-context/replicated-append/v3",
             self,
         ))?)
     }
@@ -352,6 +358,13 @@ impl Node {
         self.propose(publication::control_group(&self.layout), request, batch)
     }
 
+    /// Persist an immutable publication plan through the catalog quorum. The
+    /// plan is not a source pin or permission to commit an external manifest.
+    pub fn plan_publication(&mut self, request: u64, plan: publication::PlanRequest) -> Result<()> {
+        let batch = plan.batch(&self.layout, &self.limits)?;
+        self.propose(publication::control_group(&self.layout), 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 = 2;
-const MAGIC: &[u8; 8] = b"LCWALCP2";
+const FORMAT: u32 = 3;
+const MAGIC: &[u8; 8] = b"LCWALCP3";
 pub const CHUNK_BYTES: usize = 1024 * 1024;
 
 fn codec() -> impl Options {
@@ -82,7 +82,8 @@ fn metadata_limit(limits: &Limits) -> u64 {
     // A framed record needs at most 8 + 4 + 256 bytes, plus a bounded header.
     // limits.identity() validates all policy maxima before this is used.
     (limits.max_records + limits.max_producers) * 268
-        + limits.max_tables * (publication::MAX_BYTES as u64 + 4)
+        + limits.max_tables * (publication::MAX_BASE_TABLE_BYTES as u64 + 4)
+        + limits.max_tables.min(limits.max_active_plans) * publication::MAX_COMMAND_BYTES as u64
         + 1024
 }
 
@@ -371,6 +372,20 @@ fn read_frame<const N: usize>(input: &mut impl Read) -> Result<([u8; N], usize)>
     input.read_exact(&mut bytes[..length])?;
     Ok((bytes, length))
 }
+fn read_table_frame(input: &mut impl Read) -> Result<Vec<u8>> {
+    let mut length = [0; 4];
+    input.read_exact(&mut length)?;
+    let length = u32::from_le_bytes(length) as usize;
+    ensure!(
+        length > 0 && length <= publication::MAX_TABLE_BYTES,
+        "oversized publisher image frame"
+    );
+    // Plan metadata is larger than a record header. Allocate its actual bounded
+    // length on the heap rather than putting the maximum frame on the stack.
+    let mut bytes = vec![0; length];
+    input.read_exact(&mut bytes)?;
+    Ok(bytes)
+}
 fn write_meta(out: &mut impl Write, index: u64, meta: &BatchMeta) -> Result<()> {
     write_u64(out, index)?;
     write_frame(out, &state::encode_meta(meta)?)
@@ -524,8 +539,8 @@ pub(super) fn load(
     }
     let mut previous = None;
     for _ in 0..descriptor.tables {
-        let (bytes, n) = read_frame::<{ publication::MAX_BYTES }>(&mut metadata)?;
-        let table = publication::decode_table(&bytes[..n])?;
+        let bytes = read_table_frame(&mut metadata)?;
+        let table = publication::decode_table(&bytes)?;
         let id = table.authority.table;
         ensure!(
             previous.is_none_or(|old| old < id),

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

@@ -238,7 +238,7 @@ fn metadata_counts_frames_trailing_bytes_and_exact_producer_tails_are_validated(
     try_bytes(&count)?;
     fs::write(&path, original)?;
     let (state, _) = fixture()?;
-    for mutation in 0..4 {
+    for mutation in 0..7 {
         let mut invalid = state.clone();
         match mutation {
             0 => invalid.producers.clear(),
@@ -325,6 +325,48 @@ fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() ->
         state::validate_entry(&layout(), 10, &Limits::default(), &entry, true)?;
         state.apply(&entry, &Limits::default())?;
     }
+    let table = &state.catalog.tables[&Uuid::from_u128(200)];
+    let mut records = Vec::new();
+    for index in 1..=publication::MAX_INPUTS as u64 {
+        records.push((
+            index,
+            Batch::new(
+                table.authority.table,
+                1,
+                Uuid::from_u128(230),
+                index,
+                Bytes::from_static(b"source"),
+            )
+            .meta,
+        ));
+    }
+    let plan = publication::PlanRequest {
+        authority: table.authority.clone(),
+        plan: publication::Plan {
+            transaction: Uuid::from_u128(231),
+            read_version: 7,
+            sources: BTreeMap::from([(
+                20,
+                publication::SourceRange {
+                    after: 0,
+                    through: 1000,
+                    records,
+                },
+            )]),
+            next_group: 10,
+        },
+    }
+    .batch(&layout(), &Limits::default())?;
+    assert!(plan.body.len() > publication::MAX_ACQUIRE_BYTES);
+    let entry = Entry {
+        index: state.applied + 1,
+        term: 2,
+        context: state::encode_meta(&plan.meta)?.into(),
+        data: plan.body,
+        ..Default::default()
+    };
+    state::validate_entry(&layout(), 10, &Limits::default(), &entry, true)?;
+    state.apply(&entry, &Limits::default())?;
     let built = build(
         root.path(),
         &layout(),
@@ -365,14 +407,36 @@ fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() ->
     let second_len = u32::from_le_bytes(original[second..second + 4].try_into()?) as usize;
     let first_table = publication::decode_table(&original[offset + 4..second])?;
     let second_table = publication::decode_table(&original[second + 4..second + 4 + second_len])?;
-    for mutation in 0..4 {
+    for mutation in 0..7 {
         let mut bytes = original[..offset].to_vec();
         let mut tables = [first_table.clone(), second_table.clone()];
         match mutation {
             0 => tables.swap(0, 1),
             1 => tables[1] = tables[0].clone(),
             2 => tables[1].destination = tables[0].destination.clone(),
-            _ => tables[1].authority.grant_index = state.applied + 1,
+            3 => tables[1].authority.grant_index = state.applied + 1,
+            4 => {
+                Arc::make_mut(tables[0].progress.pending.as_mut().unwrap()).index =
+                    state.applied + 1
+            }
+            5 => {
+                Arc::make_mut(tables[0].progress.pending.as_mut().unwrap())
+                    .plan
+                    .sources
+                    .get_mut(&20)
+                    .unwrap()
+                    .after = 1
+            }
+            _ => {
+                Arc::make_mut(tables[0].progress.pending.as_mut().unwrap())
+                    .plan
+                    .sources
+                    .get_mut(&20)
+                    .unwrap()
+                    .records[0]
+                    .1
+                    .bucket = 0
+            }
         }
         for table in &tables {
             write_frame(&mut bytes, &publication::encode_table(table)?)?;