feat(wal): attest catalog selections and retain source pins

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", "serde/rc"]
+replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog", "dep:ring", "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"]
@@ -21,6 +21,7 @@ ipc = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "de
 lab = ["ipc", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "raft-engine/internals", "dep:raft-proto", "dep:serde_json"]
 
 [dependencies]
+ring = { version = "0.17", optional = true }
 bincode = { version = "1.3", optional = true }
 slog = { version = "2", optional = true }
 tokio = { version = "1", features = ["io-util"], optional = true }

crates/lance-context-wal/README.md

@@ -178,8 +178,10 @@ 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.
+They share bounded metadata between views. [Source pins](../../docs/design/wal-source-pins.md)
+verify signed catalog quorum observations against complete source prefixes and
+survive snapshots and restart. Outcome resolution and retirement remain separate
+integration steps; neither a plan nor a pin permits an external Lance commit.
 
 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

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

@@ -16,6 +16,7 @@ pub mod checkpoint;
 mod images;
 pub mod publication;
 pub mod query;
+pub mod source;
 mod state;
 #[cfg(test)]
 mod tests;
@@ -42,6 +43,9 @@ 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 {
+    /// Public Ed25519 key for catalog dependency attestations. Key changes are
+    /// application-policy migrations; private key bytes are never persisted here.
+    pub catalog_key: Option<[u8; 32]>,
     pub max_tables: u64,
     pub max_active_plans: u64,
     pub max_plan_source_bytes: u64,
@@ -54,6 +58,7 @@ pub struct Limits {
 impl Default for Limits {
     fn default() -> Self {
         Self {
+            catalog_key: None,
             max_tables: 1024,
             max_active_plans: 128,
             max_plan_source_bytes: 64 * 1024 * 1024,
@@ -69,6 +74,7 @@ impl Limits {
     fn identity(&self) -> Result<Vec<u8>> {
         ensure!(
             (1..=16_384).contains(&self.max_tables)
+                && self.catalog_key.is_none_or(|key| key != [0; 32])
                 && (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)
@@ -80,7 +86,7 @@ impl Limits {
             "invalid replication limits"
         );
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v3",
+            "lance-context/replicated-append/v4",
             self,
         ))?)
     }
@@ -109,6 +115,7 @@ pub enum Applied {
         expected_sequence: u64,
     },
     Capacity,
+    Source(source::Reply),
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -170,6 +177,7 @@ struct Group {
 /// or retain storage handles outside it. An internal/disk error fences this node;
 /// drop it, reconcile outstanding requests and reopen the original directories.
 pub struct Node {
+    signer: Option<Arc<source::CatalogSigner>>,
     layout: Arc<Layout>,
     layout_hash: [u8; 32],
     policy: [u8; 32],
@@ -188,6 +196,11 @@ impl Node {
         self.groups
             .into_iter()
             .map(|(id, group)| Self {
+                signer: if id == publication::control_group(&self.layout) {
+                    self.signer.clone()
+                } else {
+                    None
+                },
                 layout: self.layout.clone(),
                 layout_hash: self.layout_hash,
                 policy: self.policy,
@@ -263,6 +276,7 @@ impl Node {
             );
         }
         Ok(Self {
+            signer: None,
             layout,
             layout_hash,
             policy,
@@ -283,6 +297,20 @@ impl Node {
         Ok(())
     }
 
+    /// Install the catalog's private dependency-attestation key. The persisted
+    /// policy pins its public key. Source-only replicas need only the public key.
+    pub fn set_catalog_signer(&mut self, signer: Arc<source::CatalogSigner>) -> Result<()> {
+        self.healthy()?;
+        ensure!(
+            self.groups
+                .contains_key(&publication::control_group(&self.layout))
+                && self.limits.catalog_key == Some(signer.public_key()),
+            "catalog signer does not match hosted scope and policy"
+        );
+        self.signer = Some(signer);
+        Ok(())
+    }
+
     pub fn tick(&mut self) -> Result<()> {
         self.healthy()?;
         for group in self.groups.values_mut() {
@@ -334,6 +362,12 @@ impl Node {
             "request token already pending"
         );
         query.validate(&self.layout, group)?;
+        if matches!(query, query::Query::Pin { .. }) {
+            ensure!(
+                self.signer.is_some(),
+                "catalog dependency signer unavailable"
+            );
+        }
         let g = self.groups.get_mut(&group).context("unhosted group")?;
         ensure!(g.raw.raft.state == StateRole::Leader, "not leader");
         // raft-rs 0.7 silently drops ReadIndex before this condition holds.
@@ -365,6 +399,14 @@ impl Node {
         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<()> {
+        let group = permit.selection().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()?;
@@ -560,6 +602,15 @@ impl Node {
                 },
                 &group.state,
                 group.raw.raft.state == StateRole::Leader,
+                |query, cut, state| {
+                    query.evaluate(
+                        state,
+                        cut,
+                        &self.layout,
+                        &self.limits,
+                        self.signer.as_deref(),
+                    )
+                },
             )?);
             events
                 .messages

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

@@ -14,8 +14,8 @@ use std::{
     path::PathBuf,
 };
 
-const FORMAT: u32 = 3;
-const MAGIC: &[u8; 8] = b"LCWALCP3";
+const FORMAT: u32 = 4;
+const MAGIC: &[u8; 8] = b"LCWALCP4";
 pub const CHUNK_BYTES: usize = 1024 * 1024;
 
 fn codec() -> impl Options {
@@ -72,6 +72,7 @@ struct Descriptor {
     records: u64,
     producers: u64,
     tables: u64,
+    pins: u64,
     metadata_bytes: u64,
     payload_bytes: u64,
     metadata_digest: [u8; 32],
@@ -84,6 +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) * publication::MAX_COMMAND_BYTES as u64
+        + limits.max_tables.min(limits.max_active_plans) * (source::MAX_PIN_BYTES as u64 + 4)
         + 1024
 }
 
@@ -97,6 +99,7 @@ impl Descriptor {
                 && self.producers <= limits.max_producers
                 && self.producers <= self.records
                 && self.tables <= limits.max_tables
+                && self.pins <= limits.max_tables.min(limits.max_active_plans)
                 && (group == publication::control_group(layout) || self.tables == 0)
                 && self.metadata_bytes <= metadata_limit(limits)
                 && self.payload_bytes <= self.records * limits.max_record_bytes,
@@ -372,12 +375,12 @@ 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>> {
+fn read_dynamic_frame(input: &mut impl Read, maximum: usize) -> 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,
+        length > 0 && length <= maximum,
         "oversized publisher image frame"
     );
     // Plan metadata is larger than a record header. Allocate its actual bounded
@@ -442,6 +445,7 @@ pub(super) fn build(
         records: state.accepted.len() as u64,
         producers: state.producers.len() as u64,
         tables: state.catalog.tables.len() as u64,
+        pins: state.pins.len() as u64,
         metadata_bytes: 0,
         payload_bytes,
         metadata_digest: [0; 32],
@@ -454,6 +458,7 @@ pub(super) fn build(
     write_u64(&mut metadata, descriptor.records)?;
     write_u64(&mut metadata, descriptor.producers)?;
     write_u64(&mut metadata, descriptor.tables)?;
+    write_u64(&mut metadata, descriptor.pins)?;
     for (index, meta) in &state.accepted {
         write_meta(&mut metadata, *index, meta)?;
     }
@@ -463,6 +468,9 @@ pub(super) fn build(
     for table in state.catalog.tables.values() {
         write_frame(&mut metadata, &publication::encode_table(table)?)?;
     }
+    for pin in state.pins.values() {
+        write_frame(&mut metadata, &source::encode_pin(pin)?)?;
+    }
     (descriptor.metadata_bytes, descriptor.metadata_digest) = metadata.finish()?;
     let mut pack = Writer::new(&paths[1], payload_bytes)?;
     for (index, meta) in &state.accepted {
@@ -509,7 +517,8 @@ pub(super) fn load(
     ensure!(
         read_u64(&mut metadata)? == descriptor.records
             && read_u64(&mut metadata)? == descriptor.producers
-            && read_u64(&mut metadata)? == descriptor.tables,
+            && read_u64(&mut metadata)? == descriptor.tables
+            && read_u64(&mut metadata)? == descriptor.pins,
         "image metadata count mismatch"
     );
     let mut state = State {
@@ -539,7 +548,7 @@ pub(super) fn load(
     }
     let mut previous = None;
     for _ in 0..descriptor.tables {
-        let bytes = read_table_frame(&mut metadata)?;
+        let bytes = read_dynamic_frame(&mut metadata, publication::MAX_TABLE_BYTES)?;
         let table = publication::decode_table(&bytes)?;
         let id = table.authority.table;
         ensure!(
@@ -549,6 +558,18 @@ pub(super) fn load(
         state.catalog.tables.insert(id, table);
         previous = Some(id);
     }
+    let mut previous = None;
+    for _ in 0..descriptor.pins {
+        let bytes = read_dynamic_frame(&mut metadata, source::MAX_PIN_BYTES)?;
+        let pin = source::decode_pin(&bytes)?;
+        let table = pin.permit.selection().table;
+        ensure!(
+            previous.is_none_or(|old| old < table),
+            "unordered or duplicate source pin"
+        );
+        state.pins.insert(table, Arc::new(pin));
+        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,7 @@ 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 + 3 * 8;
+    let mut offset = 12 + binding_len + 4 * 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;

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

@@ -14,6 +14,49 @@ pub struct SourceRange {
     pub through: u64,
     pub records: Vec<(u64, BatchMeta)>,
 }
+impl SourceRange {
+    pub(in crate::replication) fn validate(
+        &self,
+        layout: &Layout,
+        table: Uuid,
+        group: u64,
+        limits: &Limits,
+    ) -> Result<()> {
+        ensure!(
+            layout.groups.contains_key(&group)
+                && self.after < self.through
+                && self.through < u64::MAX
+                && !self.records.is_empty()
+                && self.records.len() <= MAX_INPUTS
+                && self.records.windows(2).all(|w| w[0].0 < w[1].0),
+            "invalid publication source range"
+        );
+        let mut bytes = 0u64;
+        for (index, meta) in &self.records {
+            ensure!(
+                *index > self.after
+                    && *index <= self.through
+                    && meta.stream == table
+                    && meta.kind == RecordKind::Data
+                    && !meta.producer.is_nil()
+                    && meta.sequence > 0
+                    && meta.sequence < u64::MAX
+                    && meta.body_bytes > 0
+                    && meta.body_bytes <= limits.max_record_bytes
+                    && layout.group_for_bucket(meta.bucket)? == group,
+                "invalid publication source metadata"
+            );
+            bytes = bytes
+                .checked_add(meta.body_bytes)
+                .context("source byte sum overflow")?;
+        }
+        ensure!(
+            bytes <= limits.max_plan_source_bytes,
+            "publication source byte limit"
+        );
+        Ok(())
+    }
+}
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Plan {
@@ -59,30 +102,8 @@ impl Plan {
             count = count
                 .checked_add(range.records.len())
                 .context("plan input count overflow")?;
-            ensure!(
-                count <= MAX_INPUTS
-                    && layout.groups.contains_key(group)
-                    && range.after < range.through
-                    && range.through < u64::MAX
-                    && !range.records.is_empty()
-                    && range.records.windows(2).all(|w| w[0].0 < w[1].0),
-                "invalid publication source range"
-            );
-            for (index, meta) in &range.records {
-                ensure!(
-                    *index > range.after
-                        && *index <= range.through
-                        && meta.stream == table
-                        && meta.kind == RecordKind::Data
-                        && !meta.producer.is_nil()
-                        && meta.sequence > 0
-                        && meta.sequence < u64::MAX
-                        && meta.body_bytes > 0
-                        && meta.body_bytes <= limits.max_record_bytes
-                        && layout.group_for_bucket(meta.bucket)? == *group,
-                    "invalid publication source metadata"
-                );
-            }
+            ensure!(count <= MAX_INPUTS, "publication input count limit");
+            range.validate(layout, table, *group, limits)?;
         }
         ensure!(
             self.source_bytes()? <= limits.max_plan_source_bytes,