feat(wal): attest committed source closures

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.
 
+[Source closure proofs](../../docs/design/wal-source-closure-proofs.md) attest that
+an exact catalog-authorized closure committed at its source. Optional per-group
+keys keep catalog private keys off source-only replicas; catalog checkpoint
+advancement and physical reclamation remain separate steps.
+
 [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](../../docs/design/wal-source-pins.md)

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

@@ -46,6 +46,11 @@ 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]>,
+    /// Optional per-group keys for quorum-confirmed source closure proofs.
+    /// A nonempty policy must cover every group with distinct keys, separate
+    /// from the catalog key. Installing it is an explicit policy migration.
+    #[serde(default)]
+    pub source_keys: BTreeMap<u64, [u8; 32]>,
     pub max_tables: u64,
     pub max_active_plans: u64,
     pub max_plan_source_bytes: u64,
@@ -61,6 +66,7 @@ impl Default for Limits {
     fn default() -> Self {
         Self {
             catalog_key: None,
+            source_keys: BTreeMap::new(),
             max_tables: 1024,
             max_active_plans: 128,
             max_plan_source_bytes: 64 * 1024 * 1024,
@@ -79,6 +85,17 @@ impl Limits {
         ensure!(
             (1..=16_384).contains(&self.max_tables)
                 && self.catalog_key.is_none_or(|key| key != [0; 32])
+                && self.source_keys.len() <= 64
+                && (self.source_keys.is_empty() || self.catalog_key.is_some())
+                && self.source_keys.iter().all(|(group, key)| {
+                    *group > 0 && *key != [0; 32] && Some(*key) != self.catalog_key
+                })
+                && self
+                    .source_keys
+                    .values()
+                    .collect::<std::collections::BTreeSet<_>>()
+                    .len()
+                    == self.source_keys.len()
                 && (1..=256).contains(&self.max_active_plans)
                 && (1..=256 * 1024 * 1024).contains(&self.max_plan_source_bytes)
                 && (1..=publication::MAX_DESCRIPTOR_BYTES as u64)
@@ -92,8 +109,26 @@ impl Limits {
                 && self.max_uncommitted_bytes <= 256 * 1024 * 1024,
             "invalid replication limits"
         );
+        if self.source_keys.is_empty() {
+            // Bincode structs and tuples encode their fields in order. Keep the
+            // exact old policy bytes when source attestations are not enabled.
+            return Ok(bincode::serialize(&(
+                "lance-context/replicated-append/v9",
+                self.catalog_key,
+                self.max_tables,
+                self.max_active_plans,
+                self.max_plan_source_bytes,
+                self.max_descriptor_bytes,
+                self.max_manifest_records,
+                self.max_record_bytes,
+                self.max_records,
+                self.max_producers,
+                self.max_pending,
+                self.max_uncommitted_bytes,
+            ))?);
+        }
         Ok(bincode::serialize(&(
-            "lance-context/replicated-append/v9",
+            "lance-context/replicated-append/v10",
             self,
         ))?)
     }
@@ -185,6 +220,7 @@ struct Group {
 /// drop it, reconcile outstanding requests and reopen the original directories.
 pub struct Node {
     signer: Option<Arc<source::CatalogSigner>>,
+    source_signers: BTreeMap<u64, Arc<source::SourceSigner>>,
     layout: Arc<Layout>,
     layout_hash: [u8; 32],
     policy: [u8; 32],
@@ -208,6 +244,10 @@ impl Node {
                 } else {
                     None
                 },
+                source_signers: self
+                    .source_signers
+                    .get(&id)
+                    .map_or_else(BTreeMap::new, |s| BTreeMap::from([(id, s.clone())])),
                 layout: self.layout.clone(),
                 layout_hash: self.layout_hash,
                 policy: self.policy,
@@ -223,6 +263,10 @@ impl Node {
     pub fn open(directory: &Path, layout: Arc<Layout>, id: u64, limits: Limits) -> Result<Self> {
         let layout_hash = layout.fingerprint()?;
         let identity = limits.identity()?;
+        ensure!(
+            limits.source_keys.is_empty() || limits.source_keys.keys().eq(layout.groups.keys()),
+            "source signing policy must cover exactly the configured groups"
+        );
         let policy = Sha256::digest(&identity).into();
         let log = PhysicalLog::open(directory, &layout, id)?;
         let images = images::Owner::new(directory)?;
@@ -284,6 +328,7 @@ impl Node {
         }
         Ok(Self {
             signer: None,
+            source_signers: BTreeMap::new(),
             layout,
             layout_hash,
             policy,
@@ -318,6 +363,23 @@ impl Node {
         Ok(())
     }
 
+    /// Provision this source group's key without exposing the catalog's private
+    /// key to source-only replicas. The persisted policy pins the public key.
+    pub fn set_source_signer(
+        &mut self,
+        group: u64,
+        signer: Arc<source::SourceSigner>,
+    ) -> Result<()> {
+        self.healthy()?;
+        ensure!(
+            self.groups.contains_key(&group)
+                && self.limits.source_keys.get(&group) == Some(&signer.public_key()),
+            "source signer does not match hosted scope and policy"
+        );
+        self.source_signers.insert(group, signer);
+        Ok(())
+    }
+
     pub fn tick(&mut self) -> Result<()> {
         self.healthy()?;
         for group in self.groups.values_mut() {
@@ -369,12 +431,21 @@ impl Node {
             "request token already pending"
         );
         query.validate(&self.layout, group)?;
-        if matches!(query, query::Query::Pin { .. }) {
+        if matches!(
+            query,
+            query::Query::Pin { .. } | query::Query::ClosePermit { .. }
+        ) {
             ensure!(
                 self.signer.is_some(),
                 "catalog dependency signer unavailable"
             );
         }
+        if matches!(query, query::Query::ClosedProof { .. }) {
+            ensure!(
+                self.source_signers.contains_key(&group),
+                "source 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.
@@ -668,7 +739,10 @@ impl Node {
                         cut,
                         &self.layout,
                         &self.limits,
-                        self.signer.as_deref(),
+                        source::QuerySigners {
+                            catalog: self.signer.as_deref(),
+                            source: self.source_signers.get(id).map(Arc::as_ref),
+                        },
                     )
                 },
             )?);

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

@@ -42,6 +42,9 @@ pub enum Query {
     SourcePin {
         table: Uuid,
     },
+    ClosedProof {
+        table: Uuid,
+    },
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -55,6 +58,7 @@ pub enum Value {
     CursorAhead { applied: u64 },
     Pin(Option<Box<source::Permit>>),
     SourcePin(Option<Arc<source::Pin>>),
+    ClosedProof(Option<Box<source::ClosedProof>>),
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -114,7 +118,9 @@ impl Query {
                     && group == publication::control_group(layout),
                 "invalid catalog dependency query"
             ),
-            Self::SourceClosure { table } | Self::SourcePin { table } => {
+            Self::SourceClosure { table }
+            | Self::SourcePin { table }
+            | Self::ClosedProof { table } => {
                 ensure!(!table.is_nil(), "nil source table")
             }
             Self::Publisher { table } => ensure!(
@@ -143,9 +149,25 @@ impl Query {
         cut: &Cut,
         layout: &Layout,
         limits: &Limits,
-        signer: Option<&source::CatalogSigner>,
+        signers: source::QuerySigners<'_>,
     ) -> Result<Value> {
         Ok(match self {
+            Self::ClosedProof { table } => Value::ClosedProof(
+                state
+                    .closed
+                    .get(table)
+                    .map(|closed| {
+                        source::ClosedProof::issue(
+                            signers.source.context("source signer unavailable")?,
+                            cut.clone(),
+                            layout,
+                            limits,
+                            closed.clone(),
+                        )
+                        .map(Box::new)
+                    })
+                    .transpose()?,
+            ),
             Self::SourceClosure { table } => Value::SourceClosure(state.closed.get(table).cloned()),
             Self::ClosePermit {
                 table,
@@ -165,7 +187,9 @@ impl Query {
                     selected
                         .map(|table| {
                             source::ClosurePermit::issue(
-                                signer.context("catalog closure signer unavailable")?,
+                                signers
+                                    .catalog
+                                    .context("catalog closure signer unavailable")?,
                                 cut.clone(),
                                 layout,
                                 limits,
@@ -203,7 +227,9 @@ impl Query {
                     selected
                         .map(|table| {
                             source::Permit::issue(
-                                signer.context("catalog dependency signer unavailable")?,
+                                signers
+                                    .catalog
+                                    .context("catalog dependency signer unavailable")?,
                                 cut.clone(),
                                 layout,
                                 limits,

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

@@ -39,7 +39,10 @@ fn evaluate(query: &Query, cut: &Cut, state: &State) -> Result<Value> {
         cut,
         &super::super::tests::layout(),
         &Limits::default(),
-        None,
+        source::QuerySigners {
+            catalog: None,
+            source: None,
+        },
     )
 }
 

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

@@ -7,8 +7,15 @@ use bincode::Options;
 use bytes::Bytes;
 use ring::signature::{Ed25519KeyPair, KeyPair, UnparsedPublicKey, ED25519};
 mod closure;
+mod proof;
 pub(super) use closure::{decode_closed, encode_closed, MAX_CLOSED_BYTES};
 pub use closure::{Closed, Closure, ClosurePermit};
+pub use proof::{ClosedProof, SourceSigner};
+
+pub(super) struct QuerySigners<'a> {
+    pub catalog: Option<&'a CatalogSigner>,
+    pub source: Option<&'a SourceSigner>,
+}
 
 #[cfg(test)]
 mod tests;

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

@@ -0,0 +1,107 @@
+//! A source quorum's durable closure, distinct from a catalog permit to close.
+use super::*;
+
+const DOMAIN: &str = "lance-context/source-closed-proof/v1";
+const MAX_PROOF_BYTES: usize = 4096;
+
+/// Private signing key shared only by the replicas of one source group.
+/// This has no public signing operation and does not contain the catalog key.
+pub struct SourceSigner(Ed25519KeyPair);
+impl SourceSigner {
+    pub fn from_pkcs8(bytes: &[u8]) -> Result<Self> {
+        Ok(Self(Ed25519KeyPair::from_pkcs8(bytes).map_err(|_| {
+            anyhow::anyhow!("invalid source signing key")
+        })?))
+    }
+
+    pub fn public_key(&self) -> [u8; 32] {
+        self.0
+            .public_key()
+            .as_ref()
+            .try_into()
+            .expect("Ed25519 key")
+    }
+}
+
+/// Issued only from a successful source ReadIndex completion. It attests the
+/// exact persisted closure and its original source index. It is not a lease,
+/// publisher authority or permission to delete Lance objects or log files.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ClosedProof {
+    policy: [u8; 32],
+    cut: query::Cut,
+    closed: Arc<Closed>,
+    signature: ([u8; 32], [u8; 32]),
+}
+impl ClosedProof {
+    pub fn source_cut(&self) -> &query::Cut {
+        &self.cut
+    }
+
+    pub fn closed(&self) -> &Closed {
+        &self.closed
+    }
+
+    fn signed_bytes(&self) -> Result<Vec<u8>> {
+        Ok(codec(MAX_PROOF_BYTES).serialize(&(DOMAIN, self.policy, &self.cut, &self.closed))?)
+    }
+
+    pub(in crate::replication) fn issue(
+        signer: &SourceSigner,
+        cut: query::Cut,
+        layout: &Layout,
+        limits: &Limits,
+        closed: Arc<Closed>,
+    ) -> Result<Self> {
+        ensure!(
+            limits.source_keys.get(&cut.group) == Some(&signer.public_key()),
+            "source signing policy mismatch"
+        );
+        let mut proof = Self {
+            policy: Sha256::digest(limits.identity()?).into(),
+            cut,
+            closed,
+            signature: ([0; 32], [0; 32]),
+        };
+        let signature = signer.0.sign(&proof.signed_bytes()?);
+        proof.signature = (
+            signature.as_ref()[..32].try_into()?,
+            signature.as_ref()[32..].try_into()?,
+        );
+        proof.verify(layout, limits)?;
+        Ok(proof)
+    }
+
+    /// Authenticate the scoped statement. A consumer must additionally match
+    /// its exact expected table, plan and publication before advancing state.
+    pub fn verify(&self, layout: &Layout, limits: &Limits) -> Result<()> {
+        let key = limits
+            .source_keys
+            .get(&self.cut.group)
+            .context("source verification key unavailable")?;
+        ensure!(
+            self.policy == <[u8; 32]>::from(Sha256::digest(limits.identity()?)),
+            "foreign source proof policy"
+        );
+        let mut signature = [0; 64];
+        signature[..32].copy_from_slice(&self.signature.0);
+        signature[32..].copy_from_slice(&self.signature.1);
+        UnparsedPublicKey::new(&ED25519, key)
+            .verify(&self.signed_bytes()?, &signature)
+            .map_err(|_| anyhow::anyhow!("invalid source closure proof"))?;
+        self.closed.permit.verify(layout, limits, self.cut.group)?;
+        ensure!(
+            self.cut.cluster == layout.cluster
+                && self.cut.layout == layout.fingerprint()?
+                && self.cut.term > 0
+                && self.cut.quorum_index > 0
+                && self.cut.quorum_index <= self.cut.applied
+                && self.cut.applied < u64::MAX
+                && self.closed.index <= self.cut.applied
+                && self.closed.index > self.closed.pin_index
+                && self.closed.pin_index > self.closed.permit.closure().through,
+            "invalid source closure proof scope or cut"
+        );
+        Ok(())
+    }
+}