feat(wal): bound producer state with reusable fenced slots

crates/lance-context-wal/README.md

@@ -176,6 +176,10 @@ 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.
 
+[Producer slots](../../docs/design/wal-producer-slots.md) provide opt-in bounded
+writer state on fresh groups. Durable generations fence old requests after a
+closed lane's published source has been retired and its slot reused.
+
 [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

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

@@ -49,6 +49,8 @@ pub enum RecordKind {
     MultiPublication,
     /// Per-group source pins, release and retention commands.
     Source,
+    /// Generation-fenced producer allocation, closure and revocation.
+    Producer,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]

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

@@ -16,6 +16,7 @@
 pub mod checkpoint;
 mod images;
 pub use images::{ImageCollection, ImageCollectionProgress};
+pub mod producer;
 pub mod publication;
 pub mod query;
 pub mod source;
@@ -63,6 +64,10 @@ pub struct Limits {
     pub max_producers: u64,
     pub max_pending: u64,
     pub max_uncommitted_bytes: u64,
+    /// Fixed slots per group; zero retains UUID-only semantics and disk identity.
+    /// Set only on fresh groups. Changing this policy requires migration.
+    #[serde(default)]
+    pub producer_slots: u32,
 }
 impl Default for Limits {
     fn default() -> Self {
@@ -79,6 +84,7 @@ impl Default for Limits {
             max_producers: 16_384,
             max_pending: 128,
             max_uncommitted_bytes: 64 * 1024 * 1024,
+            producer_slots: 0,
         }
     }
 }
@@ -106,11 +112,22 @@ impl Limits {
                 && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes)
                 && (1..=500_000).contains(&self.max_records)
                 && (1..=100_000).contains(&self.max_producers)
+                && self.producer_slots <= producer::MAX_SLOTS
+                && u64::from(self.producer_slots) <= self.max_producers
                 && (1..=4096).contains(&self.max_pending)
                 && self.max_uncommitted_bytes >= self.max_record_bytes.max(1024 * 1024)
                 && self.max_uncommitted_bytes <= 256 * 1024 * 1024,
             "invalid replication limits"
         );
+        if self.producer_slots > 0 {
+            let mut legacy = self.clone();
+            legacy.producer_slots = 0;
+            return Ok(bincode::serialize(&(
+                "lance-context/replicated-append/v11",
+                legacy.identity()?,
+                self.producer_slots,
+            ))?);
+        }
         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.
@@ -131,7 +148,20 @@ impl Limits {
         }
         Ok(bincode::serialize(&(
             "lance-context/replicated-append/v10",
-            self,
+            (
+                self.catalog_key,
+                &self.source_keys,
+                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,
+            ),
         ))?)
     }
 }
@@ -160,6 +190,9 @@ pub enum Applied {
     },
     Capacity,
     Source(source::Reply),
+    Producer(producer::Reply),
+    /// Stale generation, wrong lane binding or a closed lane attempting new data.
+    ProducerFenced,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -307,7 +340,7 @@ impl Node {
                 images.installed(&loaded.image)?;
                 (loaded.state, Some(loaded.image))
             } else {
-                (State::default(), None)
+                (State::with_producer_slots(limits.producer_slots), None)
             };
             while state.applied < committed {
                 let entries = store.entries(
@@ -613,6 +646,13 @@ impl Node {
             .get(&table))
     }
 
+    /// Replicate a trusted producer lifecycle command. Completion is released
+    /// only after durable quorum commit; an ambiguous open must retry its exact CAS.
+    pub fn producer_command(&mut self, request: u64, command: producer::Command) -> Result<()> {
+        let group = command.group;
+        self.propose(group, request, command.batch(&self.layout, &self.limits)?)
+    }
+
     /// 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,20 @@ use std::{
     path::PathBuf,
 };
 
-const FORMAT: u32 = 9;
-const MAGIC: &[u8; 8] = b"LCWALCP9";
+fn format(limits: &Limits) -> u32 {
+    if limits.producer_slots == 0 {
+        9
+    } else {
+        10
+    }
+}
+fn magic(limits: &Limits) -> &'static [u8; 8] {
+    if limits.producer_slots == 0 {
+        b"LCWALCP9"
+    } else {
+        b"LCWALCPA"
+    }
+}
 pub const CHUNK_BYTES: usize = 1024 * 1024;
 
 fn codec() -> impl Options {
@@ -93,14 +105,15 @@ fn metadata_limit(limits: &Limits) -> u64 {
         + limits.max_tables * (256 + publication::advance_metadata_bytes())
         + limits.max_tables * (source::MAX_CLOSED_BYTES as u64 + 4)
         + limits.max_producers * (state::MAX_RETIRED_BYTES as u64 + 4)
+        + u64::from(limits.producer_slots) * (producer::MAX_CELL_BYTES as u64 + 4)
         + 1024
 }
 
 impl Descriptor {
     fn validate(&self, layout: &Layout, limits: &Limits, group: u64) -> Result<()> {
         self.binding.validate(layout, limits, group)?;
         ensure!(
-            self.format == FORMAT
+            self.format == format(limits)
                 && !self.id.is_nil()
                 && self.records <= limits.max_records
                 && self.producers <= limits.max_producers
@@ -460,7 +473,7 @@ pub(super) fn build(
     ensure!(!id.is_nil(), "nil application image id");
     let (locations, payload_bytes) = locations(&state)?;
     let mut descriptor = Descriptor {
-        format: FORMAT,
+        format: format(limits),
         binding,
         id,
         records: state.accepted.len() as u64,
@@ -476,7 +489,7 @@ pub(super) fn build(
     };
     let paths = descriptor.paths(directory);
     let mut metadata = Writer::new(&paths[0], metadata_limit(limits))?;
-    metadata.write_all(MAGIC)?;
+    metadata.write_all(magic(limits))?;
     write_frame(&mut metadata, &codec().serialize(&descriptor.binding)?)?;
     write_u64(&mut metadata, descriptor.records)?;
     write_u64(&mut metadata, descriptor.producers)?;
@@ -502,6 +515,14 @@ pub(super) fn build(
     for (key, retired) in &state.retired {
         write_frame(&mut metadata, &state::encode_retired(*key, *retired)?)?;
     }
+    // A flat, policy-sized cell table; decoding never trusts an allocation count
+    // supplied by the file. Zero-slot groups retain the exact legacy layout.
+    if limits.producer_slots > 0 {
+        write_u64(&mut metadata, u64::from(limits.producer_slots))?;
+        for cell in &state.producer_slots {
+            write_frame(&mut metadata, &producer::encode_cell(cell)?)?;
+        }
+    }
     (descriptor.metadata_bytes, descriptor.metadata_digest) = metadata.finish()?;
     let mut pack = Writer::new(&paths[1], payload_bytes)?;
     for (index, meta) in &state.accepted {
@@ -538,9 +559,12 @@ pub(super) fn load(
     let descriptor = Descriptor::decode(snapshot, layout, limits, group)?;
     let paths = descriptor.paths(directory);
     let mut metadata = Reader::new(&paths[0], descriptor.metadata_bytes)?;
-    let mut magic = [0; 8];
-    metadata.read_exact(&mut magic)?;
-    ensure!(&magic == MAGIC, "unsupported application image format");
+    let mut actual_magic = [0; 8];
+    metadata.read_exact(&mut actual_magic)?;
+    ensure!(
+        &actual_magic == magic(limits),
+        "unsupported application image format"
+    );
     let (binding, n) = read_frame::<512>(&mut metadata)?;
     ensure!(
         codec().deserialize::<Binding>(&binding[..n])? == descriptor.binding,
@@ -627,6 +651,18 @@ pub(super) fn load(
         state.retired.insert(key, retired);
         previous = Some(key);
     }
+    if limits.producer_slots > 0 {
+        ensure!(
+            read_u64(&mut metadata)? == u64::from(limits.producer_slots),
+            "image producer slot count mismatch"
+        );
+        for _ in 0..limits.producer_slots {
+            let (bytes, n) = read_frame::<{ producer::MAX_CELL_BYTES }>(&mut metadata)?;
+            state
+                .producer_slots
+                .push(producer::decode_cell(&bytes[..n])?);
+        }
+    }
     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)?;