feat(wal): persist publisher bindings and ownership epochs

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", "dep:bincode", "dep:slog"]
+replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog"]
 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

@@ -168,6 +168,10 @@ cases; their numbers are not performance evidence.
 
 ## Durable Raft log storage
 
+The replication layer also provides [durable publisher bindings and epochs](../../docs/design/wal-publisher-authority.md)
+through its catalog group's normal consensus path. This is an incremental
+publication prerequisite, not a production Lance writer or a manifest lease.
+
 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

@@ -14,6 +14,7 @@
 
 pub mod checkpoint;
 mod images;
+pub mod publication;
 mod state;
 #[cfg(test)]
 mod tests;
@@ -40,6 +41,7 @@ use state::State;
 /// policy. These bounds do not account for queues in the caller's transport.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Limits {
+    pub max_tables: u64,
     pub max_record_bytes: u64,
     pub max_records: u64,
     pub max_producers: u64,
@@ -49,6 +51,7 @@ pub struct Limits {
 impl Default for Limits {
     fn default() -> Self {
         Self {
+            max_tables: 1024,
             max_record_bytes: 8 * 1024 * 1024,
             max_records: 100_000,
             max_producers: 16_384,
@@ -60,7 +63,8 @@ impl Default for Limits {
 impl Limits {
     fn identity(&self) -> Result<Vec<u8>> {
         ensure!(
-            (1..=8 * 1024 * 1024).contains(&self.max_record_bytes)
+            (1..=16_384).contains(&self.max_tables)
+                && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes)
                 && (1..=500_000).contains(&self.max_records)
                 && (1..=100_000).contains(&self.max_producers)
                 && (1..=4096).contains(&self.max_pending)
@@ -69,7 +73,7 @@ impl Limits {
             "invalid replication limits"
         );
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v1",
+            "lance-context/replicated-append/v2",
             self,
         ))?)
     }
@@ -88,6 +92,7 @@ pub struct Packet {
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum Applied {
+    Publisher(publication::Reply),
     /// A new append or exact latest retry. Index identifies the original append.
     Accepted {
         index: u64,
@@ -298,6 +303,26 @@ impl Node {
         })
     }
 
+    /// Queue a binding/epoch CAS in the one catalog group. Only a committed
+    /// Publisher completion from poll grants authority; no local cache shortcut.
+    pub fn acquire_publisher(&mut self, request: u64, acquire: publication::Acquire) -> Result<()> {
+        let batch = acquire.batch(&self.layout)?;
+        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()?;
+        Ok(self
+            .groups
+            .get(&publication::control_group(&self.layout))
+            .context("catalog group not hosted")?
+            .state
+            .catalog
+            .tables
+            .get(&table))
+    }
+
     /// Ok means queued locally, NOT durable or committed. A token may have only
     /// one unresolved request on this node. No retry result bypasses consensus.
     pub fn propose(&mut self, group: u64, request: u64, batch: Batch) -> Result<()> {

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

@@ -14,8 +14,8 @@ use std::{
     path::PathBuf,
 };
 
-const FORMAT: u32 = 1;
-const MAGIC: &[u8; 8] = b"LCWALCP1";
+const FORMAT: u32 = 2;
+const MAGIC: &[u8; 8] = b"LCWALCP2";
 pub const CHUNK_BYTES: usize = 1024 * 1024;
 
 fn codec() -> impl Options {
@@ -71,6 +71,7 @@ struct Descriptor {
     id: Uuid,
     records: u64,
     producers: u64,
+    tables: u64,
     metadata_bytes: u64,
     payload_bytes: u64,
     metadata_digest: [u8; 32],
@@ -80,7 +81,9 @@ struct Descriptor {
 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 + 1024
+    (limits.max_records + limits.max_producers) * 268
+        + limits.max_tables * (publication::MAX_BYTES as u64 + 4)
+        + 1024
 }
 
 impl Descriptor {
@@ -92,6 +95,8 @@ impl Descriptor {
                 && self.records <= limits.max_records
                 && self.producers <= limits.max_producers
                 && self.producers <= self.records
+                && self.tables <= 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,
             "application image format or bounds mismatch"
@@ -421,6 +426,7 @@ pub(super) fn build(
         id,
         records: state.accepted.len() as u64,
         producers: state.producers.len() as u64,
+        tables: state.catalog.tables.len() as u64,
         metadata_bytes: 0,
         payload_bytes,
         metadata_digest: [0; 32],
@@ -432,12 +438,16 @@ pub(super) fn build(
     write_frame(&mut metadata, &codec().serialize(&descriptor.binding)?)?;
     write_u64(&mut metadata, descriptor.records)?;
     write_u64(&mut metadata, descriptor.producers)?;
+    write_u64(&mut metadata, descriptor.tables)?;
     for (index, meta) in &state.accepted {
         write_meta(&mut metadata, *index, meta)?;
     }
     for (meta, index) in state.producers.values() {
         write_meta(&mut metadata, *index, meta)?;
     }
+    for table in state.catalog.tables.values() {
+        write_frame(&mut metadata, &publication::encode_table(table)?)?;
+    }
     (descriptor.metadata_bytes, descriptor.metadata_digest) = metadata.finish()?;
     let mut pack = Writer::new(&paths[1], payload_bytes)?;
     for (index, meta) in &state.accepted {
@@ -483,7 +493,8 @@ pub(super) fn load(
     );
     ensure!(
         read_u64(&mut metadata)? == descriptor.records
-            && read_u64(&mut metadata)? == descriptor.producers,
+            && read_u64(&mut metadata)? == descriptor.producers
+            && read_u64(&mut metadata)? == descriptor.tables,
         "image metadata count mismatch"
     );
     let mut state = State {
@@ -511,6 +522,18 @@ pub(super) fn load(
         state.producers.insert(key, (meta, index));
         previous = Some(key);
     }
+    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 id = table.authority.table;
+        ensure!(
+            previous.is_none_or(|old| old < id),
+            "unordered or duplicate image table"
+        );
+        state.catalog.tables.insert(id, table);
+        previous = Some(id);
+    }
     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

@@ -289,3 +289,118 @@ fn empty_application_cut_has_a_real_empty_pack_and_no_payload_fetch() -> Result<
     );
     Ok(())
 }
+
+#[test]
+fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() -> Result<()> {
+    use crate::ingest_contract::{IngestContract, WritePolicy};
+    let root = tempfile::tempdir()?;
+    let (mut state, bodies) = fixture()?;
+    for id in [200, 201] {
+        let batch = publication::Acquire {
+            table: Uuid::from_u128(id),
+            destination: publication::Destination {
+                namespace: Uuid::from_u128(202),
+                dataset: format!("tables/{id}"),
+            },
+            ingest: IngestContract {
+                tenant: "test".into(),
+                qualified_table: format!("table-{id}"),
+                schema: [7; 32],
+                policy: WritePolicy::ImmutableUniqueIds,
+                max_record_bytes: 4096,
+                max_rows: 32,
+            },
+            owner: Uuid::from_u128(203),
+            request: Uuid::from_u128(id + 10),
+            expected_epoch: 0,
+        }
+        .batch(&layout())?;
+        let entry = Entry {
+            index: state.applied + 1,
+            term: 2,
+            context: state::encode_meta(&batch.meta)?.into(),
+            data: batch.body,
+            ..Default::default()
+        };
+        state::validate_entry(&layout(), 10, &Limits::default(), &entry, true)?;
+        state.apply(&entry, &Limits::default())?;
+    }
+    let built = build(
+        root.path(),
+        &layout(),
+        &Limits::default(),
+        Cut {
+            group: 10,
+            index: state.applied,
+            term: 2,
+            id: Uuid::from_u128(220),
+        },
+        state.clone(),
+        |index, _| Ok(bodies[&index].clone()),
+    )?;
+    let snapshot = built.image.snapshot()?;
+    assert_eq!(
+        load(root.path(), &snapshot, &layout(), &Limits::default(), 10)?.state,
+        state
+    );
+    assert_eq!(built.image.bytes(Part::Payload), 8192);
+    for index in [2, 5] {
+        assert_eq!(
+            built.image.read_for_validation(index)?,
+            Some(bodies[&index].clone())
+        );
+    }
+    let path = built.image.descriptor.paths(root.path())[0].clone();
+    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 + 3 * 8;
+    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;
+        offset += 4 + len;
+    }
+    let first_len = u32::from_le_bytes(original[offset..offset + 4].try_into()?) as usize;
+    let second = offset + 4 + first_len;
+    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 {
+        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,
+        }
+        for table in &tables {
+            write_frame(&mut bytes, &publication::encode_table(table)?)?;
+        }
+        fs::write(&path, &bytes)?;
+        let mut descriptor = built.image.descriptor.clone();
+        descriptor.metadata_bytes = bytes.len() as u64;
+        descriptor.metadata_digest = Sha256::digest(&bytes).into();
+        assert!(load(
+            root.path(),
+            &descriptor.snapshot()?,
+            &layout(),
+            &Limits::default(),
+            10
+        )
+        .is_err());
+    }
+    fs::write(&path, original)?;
+    let target = tempfile::tempdir()?;
+    let mut receiver = Receiver::new(
+        target.path(),
+        &snapshot,
+        layout(),
+        Limits::default(),
+        10,
+        Uuid::from_u128(221),
+    )?;
+    copy(&built.image, &mut receiver)?;
+    assert_eq!(receiver.finish()?.state, state);
+    Ok(())
+}