diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index 85d32b2..5683ac7 100644 --- a/crates/lance-context-wal/README.md +++ b/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 diff --git a/crates/lance-context-wal/src/record.rs b/crates/lance-context-wal/src/record.rs index 123f264..430a923 100644 --- a/crates/lance-context-wal/src/record.rs +++ b/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)] diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 668f6bc..ce540c5 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/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<()> { diff --git a/crates/lance-context-wal/src/replication/checkpoint.rs b/crates/lance-context-wal/src/replication/checkpoint.rs index bf7a32c..6cd64db 100644 --- a/crates/lance-context-wal/src/replication/checkpoint.rs +++ b/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,6 +105,7 @@ 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 } @@ -100,7 +113,7 @@ 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[..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)?; diff --git a/crates/lance-context-wal/src/replication/producer.rs b/crates/lance-context-wal/src/replication/producer.rs new file mode 100644 index 0000000..12a5c6d --- /dev/null +++ b/crates/lance-context-wal/src/replication/producer.rs @@ -0,0 +1,417 @@ +//! Bounded producer identities with durable generation fences. +//! +//! Configure slots when creating a group. Reuse requires a committed close and +//! retirement of its latest source record. Neither a timeout nor a new UUID can +//! discard an uncertain write. Control commands require a trusted caller; scope +//! validation is not authorization. In particular, revocation is administrative. + +use super::*; +use crate::record::RecordKind; +use bincode::Options; +use bytes::Bytes; + +pub const MAX_SLOTS: u32 = 16_384; +pub(super) const MAX_CELL_BYTES: usize = 96; +pub(super) const MAX_COMMAND_BYTES: usize = 256; +const DOMAIN: u32 = 0x57414c50; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Command { + pub cluster: Uuid, + pub layout: [u8; 32], + pub group: u64, + pub table: Uuid, + pub slot: u32, + pub action: Action, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Action { + /// Persist this exact tuple before sending. An ambiguous reply never + /// authorizes selecting another slot, generation or operation UUID. + Open { + expected_generation: u64, + operation: Uuid, + bucket: u32, + }, + Close { + generation: u64, + expected_last: Option, + }, + /// Fences an abandoned writer while retaining its latest receipt and data. + Revoke { generation: u64 }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Last { + pub sequence: u64, + pub index: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Reply { + Opened { + producer: Uuid, + generation: u64, + closed: bool, + }, + Closed { + last: Option, + }, + LastChanged { + actual: Option, + }, + Fenced, + Busy, + Exhausted, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Lease { + pub table: Uuid, + pub bucket: u32, + pub operation: Uuid, + pub closed: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct View { + pub capacity: u32, + pub slot: u32, + pub generation: u64, + pub lease: Option, + pub last: Option, + /// Local state at the query's quorum cut; an open still needs its CAS. + pub reusable: bool, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub(super) struct Cell { + pub generation: u64, + pub lease: Option, +} + +fn codec(limit: usize) -> impl Options { + bincode::DefaultOptions::new() + .with_fixint_encoding() + .with_little_endian() + .with_limit(limit as u64) + .reject_trailing_bytes() +} + +/// This UUID-shaped value is meaningful only with its cluster/layout/group. +pub fn producer_id(slot: u32, generation: u64) -> Uuid { + Uuid::from_u128(((DOMAIN as u128) << 96) | ((slot as u128) << 64) | generation as u128) +} + +fn handle(id: Uuid) -> Option<(usize, u64)> { + let bits = id.as_u128(); + ((bits >> 96) as u32 == DOMAIN).then_some(((bits >> 64) as u32 as usize, bits as u64)) +} + +impl Command { + fn validate(&self, layout: &Layout, group: u64, limits: &Limits) -> Result<()> { + ensure!( + self.cluster == layout.cluster + && self.layout == layout.fingerprint()? + && self.group == group + && layout.groups.contains_key(&group) + && !self.table.is_nil() + && self.slot < limits.producer_slots, + "invalid producer command scope or slot" + ); + match &self.action { + Action::Open { + operation, bucket, .. + } => ensure!( + !operation.is_nil() && layout.group_for_bucket(*bucket)? == group, + "invalid producer allocation identity or route" + ), + Action::Close { + generation, + expected_last, + } => ensure!( + *generation > 0 + && expected_last.is_none_or(|last| last.sequence > 0 + && last.sequence < u64::MAX + && last.index > 0 + && last.index < u64::MAX), + "invalid producer close" + ), + Action::Revoke { generation } => ensure!(*generation > 0, "invalid producer revoke"), + } + Ok(()) + } + + pub(super) fn batch(self, layout: &Layout, limits: &Limits) -> Result { + self.validate(layout, self.group, limits)?; + let body = codec(MAX_COMMAND_BYTES).serialize(&self)?; + ensure!( + body.len() <= MAX_COMMAND_BYTES, + "oversized producer command" + ); + let mut batch = Batch::new( + self.table, + self.slot, + producer_id(self.slot, 0), + 1, + Bytes::from(body), + ); + batch.meta.kind = RecordKind::Producer; + Ok(batch) + } +} + +pub(super) fn decode(bytes: &[u8]) -> Result { + ensure!( + bytes.len() <= MAX_COMMAND_BYTES, + "oversized producer command" + ); + let command = codec(MAX_COMMAND_BYTES).deserialize(bytes)?; + ensure!( + codec(MAX_COMMAND_BYTES).serialize(&command)? == bytes, + "noncanonical producer command" + ); + Ok(command) +} + +pub(super) fn validate_record( + layout: &Layout, + group: u64, + limits: &Limits, + meta: &BatchMeta, + body: &[u8], +) -> Result<()> { + let command = decode(body)?; + command.validate(layout, group, limits)?; + ensure!( + meta.kind == RecordKind::Producer + && meta.stream == command.table + && meta.bucket == command.slot + && meta.producer == producer_id(command.slot, 0) + && meta.sequence == 1, + "producer command envelope mismatch" + ); + Ok(()) +} + +pub(super) fn encode_cell(cell: &Cell) -> Result> { + Ok(codec(MAX_CELL_BYTES).serialize(cell)?) +} +pub(super) fn decode_cell(bytes: &[u8]) -> Result { + ensure!(bytes.len() <= MAX_CELL_BYTES, "oversized producer cell"); + let cell = codec(MAX_CELL_BYTES).deserialize(bytes)?; + ensure!(encode_cell(&cell)? == bytes, "noncanonical producer cell"); + Ok(cell) +} + +impl State { + pub(super) fn with_producer_slots(capacity: u32) -> Self { + Self { + producer_slots: vec![Cell::default(); capacity as usize], + ..Self::default() + } + } + + fn producer_key(slot: u32, cell: &Cell) -> Option { + cell.lease.as_ref().map(|lease| { + ( + lease.table, + lease.bucket, + producer_id(slot, cell.generation), + ) + }) + } + + fn producer_last(&self, slot: u32, cell: &Cell) -> Option { + self.producers + .get(&Self::producer_key(slot, cell)?) + .map(|(meta, index)| Last { + sequence: meta.sequence, + index: *index, + }) + } + + fn producer_reusable(&self, slot: u32, cell: &Cell) -> bool { + if cell.generation == u64::MAX { + return false; + } + let Some(lease) = &cell.lease else { + return cell.generation == 0; + }; + if !lease.closed { + return false; + } + let Some(last) = self.producer_last(slot, cell) else { + return true; + }; + let key = Self::producer_key(slot, cell).expect("lease present"); + // Closure alone is not sufficient: require the exact latest lane tail + // to have been retired by the existing signed source-closure transition. + self.retired + .get(&key) + .is_some_and(|retired| retired.sequence == last.sequence && retired.index == last.index) + } + + pub(super) fn producer_view(&self, slot: u32) -> Option { + let cell = self.producer_slots.get(slot as usize)?; + Some(View { + capacity: self.producer_slots.len() as u32, + slot, + generation: cell.generation, + lease: cell.lease.clone(), + last: self.producer_last(slot, cell), + reusable: self.producer_reusable(slot, cell), + }) + } + + pub(super) fn producer_admission(&self, meta: &BatchMeta, allow_closed: bool) -> bool { + if self.producer_slots.is_empty() { + return true; + } + let Some((slot, generation)) = handle(meta.producer) else { + return false; + }; + self.producer_slots.get(slot).is_some_and(|cell| { + cell.generation == generation + && generation > 0 + && cell.lease.as_ref().is_some_and(|lease| { + lease.table == meta.stream + && lease.bucket == meta.bucket + && (allow_closed || !lease.closed) + }) + }) + } + + pub(super) fn validate_producers( + &self, + layout: &Layout, + group: u64, + limits: &Limits, + ) -> Result<()> { + ensure!( + self.producer_slots.len() == limits.producer_slots as usize, + "producer slot capacity mismatch" + ); + if self.producer_slots.is_empty() { + return Ok(()); + } + ensure!( + self.producers.len() <= self.producer_slots.len(), + "producer receipts exceed slot capacity" + ); + for cell in &self.producer_slots { + match &cell.lease { + None => ensure!(cell.generation == 0, "missing producer generation lease"), + Some(lease) => ensure!( + cell.generation > 0 + && !lease.table.is_nil() + && !lease.operation.is_nil() + && layout.group_for_bucket(lease.bucket)? == group, + "invalid producer lease" + ), + } + } + ensure!( + self.producers + .values() + .all(|(meta, _)| self.producer_admission(meta, true)), + "producer receipt has stale slot scope" + ); + Ok(()) + } + + pub(super) fn apply_producer(&mut self, command: Command) -> Reply { + let Some(cell) = self.producer_slots.get(command.slot as usize).cloned() else { + return Reply::Fenced; + }; + let slot = command.slot; + let last = self.producer_last(slot, &cell); + match command.action { + Action::Open { + expected_generation, + operation, + bucket, + } => { + let Some(generation) = expected_generation.checked_add(1) else { + return Reply::Exhausted; + }; + if cell.generation == generation { + return match cell.lease { + Some(lease) + if lease.table == command.table + && lease.bucket == bucket + && lease.operation == operation => + { + Reply::Opened { + producer: producer_id(slot, generation), + generation, + closed: lease.closed, + } + } + _ => Reply::Fenced, + }; + } + if cell.generation != expected_generation { + return Reply::Fenced; + } + if !self.producer_reusable(slot, &cell) { + return Reply::Busy; + } + if let Some(key) = Self::producer_key(slot, &cell) { + self.producers.remove(&key); + self.retired.remove(&key); + } + self.producer_slots[slot as usize] = Cell { + generation, + lease: Some(Lease { + table: command.table, + bucket, + operation, + closed: false, + }), + }; + Reply::Opened { + producer: producer_id(slot, generation), + generation, + closed: false, + } + } + Action::Close { + generation, + expected_last, + } => { + if cell.generation != generation + || cell.lease.as_ref().is_none_or(|l| l.table != command.table) + { + return Reply::Fenced; + } + if last != expected_last { + return Reply::LastChanged { actual: last }; + } + self.producer_slots[slot as usize] + .lease + .as_mut() + .unwrap() + .closed = true; + Reply::Closed { last } + } + Action::Revoke { generation } => { + if cell.generation != generation + || cell.lease.as_ref().is_none_or(|l| l.table != command.table) + { + return Reply::Fenced; + } + self.producer_slots[slot as usize] + .lease + .as_mut() + .unwrap() + .closed = true; + Reply::Closed { last } + } + } + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/lance-context-wal/src/replication/producer/tests.rs b/crates/lance-context-wal/src/replication/producer/tests.rs new file mode 100644 index 0000000..13a15fa --- /dev/null +++ b/crates/lance-context-wal/src/replication/producer/tests.rs @@ -0,0 +1,320 @@ +use super::*; + +fn layout() -> Arc { + crate::replication::tests::layout() +} +fn limits() -> Limits { + Limits { + producer_slots: 1, + ..Default::default() + } +} +fn command(action: Action) -> Command { + let layout = layout(); + Command { + cluster: layout.cluster, + layout: layout.fingerprint().unwrap(), + group: 10, + table: Uuid::from_u128(42), + slot: 0, + action, + } +} +fn open(expected_generation: u64) -> Command { + command(Action::Open { + expected_generation, + operation: Uuid::from_u128(100 + expected_generation as u128), + bucket: 0, + }) +} +fn apply(state: &mut State, batch: Batch) -> Result { + let entry = Entry { + index: state.applied + 1, + term: 1, + data: batch.body, + context: state::encode_meta(&batch.meta)?.into(), + ..Default::default() + }; + state::validate_entry(&layout(), 10, &limits(), &entry, true)?; + Ok(state.apply(&entry, &limits())?.unwrap()) +} +fn control(state: &mut State, command: Command) -> Result { + let Applied::Producer(reply) = apply(state, command.batch(&layout(), &limits())?)? else { + anyhow::bail!("missing producer outcome") + }; + Ok(reply) +} +fn data(generation: u64, sequence: u64) -> Batch { + Batch::new( + Uuid::from_u128(42), + 0, + producer_id(0, generation), + sequence, + Bytes::from(vec![sequence as u8; 128]), + ) +} + +#[test] +fn close_requires_exact_tail_and_reuse_requires_retirement() -> Result<()> { + let mut state = State::with_producer_slots(1); + assert!(matches!( + control(&mut state, open(0))?, + Reply::Opened { generation: 1, .. } + )); + let first = data(1, 1); + let Applied::Accepted { index } = apply(&mut state, first.clone())? else { + panic!("not accepted") + }; + let last = Last { sequence: 1, index }; + assert_eq!( + control( + &mut state, + command(Action::Close { + generation: 1, + expected_last: None + }) + )?, + Reply::LastChanged { actual: Some(last) } + ); + assert_eq!(control(&mut state, open(1))?, Reply::Busy); + assert_eq!( + control( + &mut state, + command(Action::Close { + generation: 1, + expected_last: Some(last) + }) + )?, + Reply::Closed { last: Some(last) } + ); + assert_eq!( + apply(&mut state, first.clone())?, + Applied::Accepted { index } + ); + assert_eq!(apply(&mut state, data(1, 2))?, Applied::ProducerFenced); + assert_eq!(control(&mut state, open(1))?, Reply::Busy); + // State-machine boundary test; the real Lance/process fixture separately + // exercises the signed publication/closure transition that creates this tail. + let key = (first.meta.stream, first.meta.bucket, first.meta.producer); + state.accepted.remove(&index); + state + .retired + .insert(key, state::Retired { sequence: 1, index }); + assert!(state.producer_view(0).unwrap().reusable); + assert!(matches!( + control(&mut state, open(1))?, + Reply::Opened { generation: 2, .. } + )); + assert!(state.producers.is_empty() && state.retired.is_empty()); + assert_eq!(apply(&mut state, first)?, Applied::ProducerFenced); + assert_eq!(control(&mut state, open(0))?, Reply::Fenced); + assert_eq!( + control(&mut state, command(Action::Revoke { generation: 1 }))?, + Reply::Fenced + ); + assert!(matches!( + apply(&mut state, data(2, 1))?, + Applied::Accepted { .. } + )); + Ok(()) +} + +#[test] +fn allocation_retries_revocation_scope_and_exhaustion_are_explicit() -> Result<()> { + let mut state = State::with_producer_slots(1); + let opened = control(&mut state, open(0))?; + assert_eq!(control(&mut state, open(0))?, opened); + let mut competing = open(0); + if let Action::Open { operation, .. } = &mut competing.action { + *operation = Uuid::new_v4(); + } + assert_eq!(control(&mut state, competing)?, Reply::Fenced); + let mut wrong = command(Action::Revoke { generation: 1 }); + wrong.table = Uuid::new_v4(); + assert_eq!(control(&mut state, wrong)?, Reply::Fenced); + assert_eq!( + control(&mut state, command(Action::Revoke { generation: 1 }))?, + Reply::Closed { last: None } + ); + assert!(matches!( + control(&mut state, open(0))?, + Reply::Opened { closed: true, .. } + )); + assert!(matches!( + control(&mut state, open(1))?, + Reply::Opened { generation: 2, .. } + )); + let mut wrong = data(2, 1); + wrong.meta.stream = Uuid::new_v4(); + assert_eq!(apply(&mut state, wrong)?, Applied::ProducerFenced); + state.producer_slots[0].generation = u64::MAX; + state.producer_slots[0].lease.as_mut().unwrap().closed = true; + assert_eq!(control(&mut state, open(u64::MAX))?, Reply::Exhausted); + assert_eq!(state.producer_slots[0].generation, u64::MAX); + assert!(!state.producer_view(0).unwrap().reusable); + Ok(()) +} + +#[test] +fn checkpoint_preserves_slot_generation_without_receipt_history() -> Result<()> { + let directory = tempfile::tempdir()?; + let limits = limits(); + let layout = layout(); + let mut state = State::with_producer_slots(1); + for generation in 0..40 { + assert!(matches!( + control(&mut state, open(generation))?, + Reply::Opened { .. } + )); + control( + &mut state, + command(Action::Revoke { + generation: generation + 1, + }), + )?; + } + state.validate_checkpoint(&layout, 10, &limits, state.applied)?; + let built = checkpoint::build( + directory.path(), + &layout, + &limits, + checkpoint::Cut { + group: 10, + index: state.applied, + term: 1, + id: Uuid::new_v4(), + }, + state.clone(), + |_, _| anyhow::bail!("no retained payload"), + )?; + let snapshot = built.image.snapshot()?; + let restored = checkpoint::load(directory.path(), &snapshot, &layout, &limits, 10)?; + assert_eq!(restored.state, state); + assert_eq!(restored.image.bytes(checkpoint::Part::Payload), 0); + assert!(restored.image.bytes(checkpoint::Part::Metadata) < 1024); + assert!( + checkpoint::load(directory.path(), &snapshot, &layout, &Limits::default(), 10).is_err() + ); + let mut restored = restored.state; + assert_eq!(control(&mut restored, open(0))?, Reply::Fenced); + assert!(matches!( + control(&mut restored, open(40))?, + Reply::Opened { generation: 41, .. } + )); + Ok(()) +} + +#[test] +fn command_and_checkpoint_cell_decoders_reject_unbounded_or_foreign_input() -> Result<()> { + let batch = open(0).batch(&layout(), &limits())?; + for length in 0..batch.body.len() { + assert!(decode(&batch.body[..length]).is_err()); + } + let mut trailing = batch.body.to_vec(); + trailing.push(0); + assert!(decode(&trailing).is_err()); + assert!(decode(&vec![0; MAX_COMMAND_BYTES + 1]).is_err()); + for mutate in [0, 1, 2, 3] { + let mut request = open(0); + match mutate { + 0 => request.cluster = Uuid::new_v4(), + 1 => request.layout[0] ^= 1, + 2 => request.group = 20, + _ => request.slot = 1, + } + assert!(request.batch(&layout(), &limits()).is_err()); + } + assert!(open(0).batch(&layout(), &Limits::default()).is_err()); + let cell = Cell { + generation: 1, + lease: Some(Lease { + table: Uuid::new_v4(), + bucket: 0, + operation: Uuid::new_v4(), + closed: true, + }), + }; + let encoded = encode_cell(&cell)?; + assert!(encoded.len() <= MAX_CELL_BYTES); + for length in 0..encoded.len() { + assert!(decode_cell(&encoded[..length]).is_err()); + } + assert_eq!(decode_cell(&encoded)?, cell); + let mut bad = State::with_producer_slots(1); + bad.producer_slots[0] = cell; + bad.producer_slots[0].generation = 0; + assert!(bad.validate_producers(&layout(), 10, &limits()).is_err()); + assert!(Limits { + producer_slots: MAX_SLOTS + 1, + ..Default::default() + } + .identity() + .is_err()); + assert!(Limits { + producer_slots: 2, + max_producers: 1, + ..Default::default() + } + .identity() + .is_err()); + Ok(()) +} + +#[test] +fn zero_slot_policy_keeps_v10_bytes_and_enabled_policy_cannot_reopen_legacy() -> Result<()> { + // Independent pre-slot struct: an accidental new field in the old bincode + // identity would invalidate existing journals and signed closure permits. + #[derive(Serialize)] + struct OldLimits { + catalog_key: Option<[u8; 32]>, + source_keys: BTreeMap, + max_tables: u64, + max_active_plans: u64, + max_plan_source_bytes: u64, + max_descriptor_bytes: u64, + max_manifest_records: u64, + max_record_bytes: u64, + max_records: u64, + max_producers: u64, + max_pending: u64, + max_uncommitted_bytes: u64, + } + let limits = Limits { + catalog_key: Some([1; 32]), + source_keys: BTreeMap::from([(10, [2; 32])]), + ..Default::default() + }; + let old = OldLimits { + catalog_key: limits.catalog_key, + source_keys: limits.source_keys.clone(), + max_tables: limits.max_tables, + max_active_plans: limits.max_active_plans, + max_plan_source_bytes: limits.max_plan_source_bytes, + max_descriptor_bytes: limits.max_descriptor_bytes, + max_manifest_records: limits.max_manifest_records, + max_record_bytes: limits.max_record_bytes, + max_records: limits.max_records, + max_producers: limits.max_producers, + max_pending: limits.max_pending, + max_uncommitted_bytes: limits.max_uncommitted_bytes, + }; + assert_eq!( + limits.identity()?, + bincode::serialize(&("lance-context/replicated-append/v10", old))? + ); + let directory = tempfile::tempdir()?; + drop(Node::open( + directory.path(), + layout(), + 1, + Limits::default(), + )?); + assert!(Node::open(directory.path(), layout(), 1, super::tests::limits()).is_err()); + drop(Node::open( + directory.path(), + layout(), + 1, + Limits::default(), + )?); + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/query.rs b/crates/lance-context-wal/src/replication/query.rs index 484f89c..db398f7 100644 --- a/crates/lance-context-wal/src/replication/query.rs +++ b/crates/lance-context-wal/src/replication/query.rs @@ -45,6 +45,9 @@ pub enum Query { ClosedProof { table: Uuid, }, + ProducerSlot { + slot: u32, + }, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -64,6 +67,7 @@ pub enum Value { /// A historical position is outside catalog retention. This is neither a /// missing-object result nor proof that a supplied staging object lost. ManifestRetired(publication::HistoryBoundary), + ProducerSlot(Option), } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -97,6 +101,10 @@ pub struct Completion { impl Query { pub(super) fn validate(&self, layout: &Layout, group: u64) -> Result<()> { match self { + Self::ProducerSlot { slot } => ensure!( + *slot < producer::MAX_SLOTS, + "producer slot query out of bounds" + ), Self::Manifest { table, version } => ensure!( !table.is_nil() && group == publication::control_group(layout) @@ -157,6 +165,7 @@ impl Query { signers: source::QuerySigners<'_>, ) -> Result { Ok(match self { + Self::ProducerSlot { slot } => Value::ProducerSlot(state.producer_view(*slot)), Self::ClosedProof { table } => Value::ClosedProof( state .closed diff --git a/crates/lance-context-wal/src/replication/state.rs b/crates/lance-context-wal/src/replication/state.rs index 7ec325a..75589d6 100644 --- a/crates/lance-context-wal/src/replication/state.rs +++ b/crates/lance-context-wal/src/replication/state.rs @@ -25,6 +25,10 @@ pub(super) fn validate_meta( body: &[u8], digest: bool, ) -> Result<()> { + if meta.kind == RecordKind::Producer { + meta.validate_payload(body, producer::MAX_COMMAND_BYTES)?; + return producer::validate_record(layout, group, limits, meta, body); + } if meta.kind == RecordKind::Source { meta.validate_payload(body, source::MAX_BYTES)?; return source::validate_record(layout, group, limits, meta, body); @@ -147,6 +151,7 @@ pub(super) fn validate_entry( #[derive(Clone, Debug, Default, PartialEq, Eq)] pub(super) struct State { + pub producer_slots: Vec, pub retired: BTreeMap, pub closed: BTreeMap>, pub pins: BTreeMap>, @@ -197,6 +202,7 @@ impl State { && self.retired.len() <= self.producers.len(), "application checkpoint exceeds state limits" ); + self.validate_producers(layout, group, limits)?; self.catalog.validate(layout, group, cut, limits)?; source::validate_state(self, layout, group, cut, limits)?; for (key, (meta, index)) in &self.producers { @@ -303,6 +309,14 @@ impl State { limits, )))); } + if meta.kind == RecordKind::Producer { + return Ok(Some(Applied::Producer( + self.apply_producer(producer::decode(&entry.data)?), + ))); + } + if !self.producer_admission(&meta, true) { + return Ok(Some(Applied::ProducerFenced)); + } let key = (meta.stream, meta.bucket, meta.producer); let previous = self.producers.get(&key); if let Some((old, index)) = previous { @@ -310,6 +324,9 @@ impl State { return Ok(Some(Applied::Accepted { index: *index })); } } + if !self.producer_admission(&meta, false) { + return Ok(Some(Applied::ProducerFenced)); + } let expected_sequence = previous.map_or(1, |(old, _)| old.sequence + 1); if meta.sequence != expected_sequence { return Ok(Some(Applied::Rejected { expected_sequence })); diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index 7f28b0d..a6872eb 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -62,6 +62,7 @@ impl WirePacket { } #[derive(Serialize, Deserialize)] enum Request { + Producer(u64, lance_context_wal::replication::producer::Command), RetireManifestHistory( u64, lance_context_wal::replication::publication::RetireManifestHistory, @@ -158,6 +159,7 @@ fn run(worker: &mut Worker, request: Request) -> Result { Request::RetireManifestHistory(token, request) => { node.retire_manifest_history(token, request)? } + Request::Producer(token, command) => node.producer_command(token, command)?, Request::CloseSource(token, permit) => node.close_source(token, permit)?, Request::ConfirmPublication(token, request) => node.confirm_publication(token, request)?, Request::AdvancePublication(token, request) => node.advance_publication(token, request)?, @@ -433,6 +435,13 @@ impl Cluster { Self::with_layout(layout, None) } fn with_layout(layout: Layout, catalog_key: Option<&[u8]>) -> Result { + Self::with_producer_slots(layout, catalog_key, 0) + } + fn with_producer_slots( + layout: Layout, + catalog_key: Option<&[u8]>, + producer_slots: u32, + ) -> Result { let root = tempfile::tempdir()?; if let Some(key) = catalog_key { let signer = CatalogSigner::from_pkcs8(key)?; @@ -456,6 +465,7 @@ impl Cluster { serde_json::to_vec(&Limits { catalog_key: Some(signer.public_key()), source_keys, + producer_slots, ..Default::default() })?, )?; @@ -463,6 +473,15 @@ impl Cluster { std::fs::write(root.path().join(format!("catalog-key-{id}.pkcs8")), key)?; } } + if catalog_key.is_none() && producer_slots > 0 { + std::fs::write( + root.path().join("limits.json"), + serde_json::to_vec(&Limits { + producer_slots, + ..Default::default() + })?, + )?; + } std::fs::write( root.path().join("layout.json"), serde_json::to_vec(&layout)?, @@ -1691,3 +1710,6 @@ fn disjoint_source_pins_survive_transfer_takeover_and_all_process_restart() -> R #[cfg(feature = "lance-integration")] #[path = "replication_process/lance_publication.rs"] mod lance_publication; + +#[path = "replication_process/producer_slots.rs"] +mod producer_slots; diff --git a/crates/lance-context-wal/tests/replication_process/lance_publication.rs b/crates/lance-context-wal/tests/replication_process/lance_publication.rs index 21b32b6..bcbef73 100644 --- a/crates/lance-context-wal/tests/replication_process/lance_publication.rs +++ b/crates/lance-context-wal/tests/replication_process/lance_publication.rs @@ -102,7 +102,8 @@ async fn recovered_publication(occupied: bool, managed: Option) - groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![4, 5, 6])]), buckets: vec![10, 20], }; - let mut c = Cluster::with_layout(layout, Some(key.as_ref()))?; + let slots = u32::from(managed == Some(managed::Mode::ProducerSlots)); + let mut c = Cluster::with_producer_slots(layout, Some(key.as_ref()), slots)?; let uri = c .root .path() @@ -130,7 +131,12 @@ async fn recovered_publication(occupied: bool, managed: Option) - encoder.finish()?; } let table = Uuid::from_u128(902); - let input = Batch::new(table, 1, Uuid::from_u128(903), 1, Bytes::from(wire)); + let producer = if slots > 0 { + managed::producer_lifecycle::open(&mut c, source, table)? + } else { + Uuid::from_u128(903) + }; + let input = Batch::new(table, 1, producer, 1, Bytes::from(wire)); c.command(source, Request::Propose(20, 2000, input.clone()))?; c.pump(&[4, 5, 6])?; let Some(Outcome::Committed(Applied::Accepted { @@ -139,6 +145,14 @@ async fn recovered_publication(occupied: bool, managed: Option) - else { anyhow::bail!("WAL append missing"); }; + if slots > 0 { + managed::producer_lifecycle::close_before_publication( + &mut c, + source, + &input, + source_index, + )?; + } let mut acquire = Acquire { table, destination: Destination { @@ -418,3 +432,9 @@ async fn catalog_manifest_retention_preserves_real_lance_after_history_limit_and ) -> Result<()> { recovered_publication(false, Some(managed::Mode::Retention)).await } + +#[tokio::test] +async fn producer_slot_reuse_preserves_real_lance_and_fences_old_writes_after_restart() -> Result<()> +{ + recovered_publication(false, Some(managed::Mode::ProducerSlots)).await +} diff --git a/crates/lance-context-wal/tests/replication_process/managed_manifest.rs b/crates/lance-context-wal/tests/replication_process/managed_manifest.rs index 3c7d383..caec3a6 100644 --- a/crates/lance-context-wal/tests/replication_process/managed_manifest.rs +++ b/crates/lance-context-wal/tests/replication_process/managed_manifest.rs @@ -3,6 +3,8 @@ use super::*; #[path = "manifest_retention.rs"] mod manifest_retention; +#[path = "producer_lifecycle.rs"] +pub(super) mod producer_lifecycle; #[path = "source_closure.rs"] mod source_closure; use lance_context_wal::replication::publication::{ @@ -71,6 +73,7 @@ pub(super) enum Mode { LostReply, Takeover, Retention, + ProducerSlots, } #[derive(Default)] @@ -434,6 +437,9 @@ pub(super) async fn publish_and_recover(c: Cluster, attempt: Attempt, mode: Mode &selected, &attempt.input, )?; + if mode == Mode::ProducerSlots { + producer_lifecycle::reuse_after_retirement(&cluster, &attempt.input)?; + } // Both version lookup and latest lookup are strict catalog reads. A raw // URI reader cannot discover this unique path and is not a managed reader. let latest = lance::dataset::builder::DatasetBuilder::from_uri(&attempt.uri) diff --git a/crates/lance-context-wal/tests/replication_process/producer_lifecycle.rs b/crates/lance-context-wal/tests/replication_process/producer_lifecycle.rs new file mode 100644 index 0000000..344fb98 --- /dev/null +++ b/crates/lance-context-wal/tests/replication_process/producer_lifecycle.rs @@ -0,0 +1,238 @@ +//! Exercise lifecycle boundaries through actual Raft processes and the existing +//! Lance publication/closure fixture. No synthetic retirement proof is inserted. +use super::*; +use lance_context_wal::replication::producer::{self, Action, Command, Last, Reply}; + +fn command(c: &Cluster, table: Uuid, action: Action) -> Result { + let layout: Layout = + serde_json::from_slice(&std::fs::read(c.root.path().join("layout.json"))?)?; + Ok(Command { + cluster: layout.cluster, + layout: layout.fingerprint()?, + group: 20, + table, + slot: 0, + action, + }) +} +fn allocate(c: &Cluster, table: Uuid, expected_generation: u64) -> Result { + command( + c, + table, + Action::Open { + expected_generation, + operation: Uuid::from_u128(800_000 + expected_generation as u128), + bucket: 1, + }, + ) +} +fn control(c: &mut Cluster, leader: u64, token: u64, command: Command) -> Result { + c.command(leader, Request::Producer(token, command))?; + c.pump(&[4, 5, 6])?; + let Some(Outcome::Committed(Applied::Producer(reply))) = c.outcome(token) else { + anyhow::bail!("producer command did not commit") + }; + Ok(reply) +} +fn view(c: &mut Cluster, leader: u64, token: u64) -> Result { + c.command( + leader, + Request::Query(20, token, Query::ProducerSlot { slot: 0 }), + )?; + c.pump(&[4, 5, 6])?; + let Some(ReadOutcome::Observed { + value: Value::ProducerSlot(Some(view)), + .. + }) = c.read_outcome(token) + else { + anyhow::bail!("slot quorum view missing") + }; + Ok(view) +} + +pub(in crate::lance_publication) fn open( + c: &mut Cluster, + leader: u64, + table: Uuid, +) -> Result { + let request = allocate(c, table, 0)?; + // Drop a committed allocation completion, then retry the saved CAS exactly. + c.command(leader, Request::Producer(700_000, request.clone()))?; + c.pump(&[4, 5, 6])?; + assert!(c.outcome(700_000).is_some()); + let reply = control(c, leader, 700_001, request)?; + assert_eq!( + reply, + Reply::Opened { + producer: producer::producer_id(0, 1), + generation: 1, + closed: false + } + ); + assert_eq!(view(c, leader, 700_002)?.generation, 1); + Ok(producer::producer_id(0, 1)) +} + +pub(in crate::lance_publication) fn close_before_publication( + c: &mut Cluster, + leader: u64, + input: &Batch, + index: u64, +) -> Result<()> { + let table = input.meta.stream; + let last = Last { sequence: 1, index }; + let stale = command( + c, + table, + Action::Close { + generation: 1, + expected_last: None, + }, + )?; + assert_eq!( + control(c, leader, 700_010, stale)?, + Reply::LastChanged { actual: Some(last) } + ); + let close = command( + c, + table, + Action::Close { + generation: 1, + expected_last: Some(last), + }, + )?; + assert_eq!( + control(c, leader, 700_011, close)?, + Reply::Closed { last: Some(last) } + ); + let next = allocate(c, table, 1)?; + assert_eq!(control(c, leader, 700_012, next)?, Reply::Busy); + let mut delayed = input.clone(); + delayed.meta.sequence = 2; + c.command(leader, Request::Propose(20, 700_013, delayed))?; + c.pump(&[4, 5, 6])?; + assert_eq!( + c.outcome(700_013), + Some(Outcome::Committed(Applied::ProducerFenced)) + ); + let current = view(c, leader, 700_014)?; + assert!(current.lease.unwrap().closed && !current.reusable); + Ok(()) +} + +pub(super) fn reuse_after_retirement(cluster: &Arc>, input: &Batch) -> Result<()> { + let mut c = cluster.lock().unwrap(); + let table = input.meta.stream; + let leader = c.elect(4, 20, &[4, 5, 6])?; + let before = view(&mut c, leader, 700_020)?; + assert_eq!(before.generation, 1); + assert!(before.lease.unwrap().closed && before.reusable); + let request = allocate(&c, table, 1)?; + assert_eq!( + control(&mut c, leader, 700_021, request.clone())?, + Reply::Opened { + producer: producer::producer_id(0, 2), + generation: 2, + closed: false + } + ); + c.command(leader, Request::Propose(20, 700_022, input.clone()))?; + c.pump(&[4, 5, 6])?; + assert_eq!( + c.outcome(700_022), + Some(Outcome::Committed(Applied::ProducerFenced)) + ); + let old = allocate(&c, table, 0)?; + assert_eq!(control(&mut c, leader, 700_023, old)?, Reply::Fenced); + let revoke = command(&c, table, Action::Revoke { generation: 1 })?; + assert_eq!(control(&mut c, leader, 700_024, revoke)?, Reply::Fenced); + let mut next = input.clone(); + next.meta.producer = producer::producer_id(0, 2); + c.command(leader, Request::Propose(20, 700_025, next.clone()))?; + c.pump(&[4, 5, 6])?; + let Some(Outcome::Committed(Applied::Accepted { index })) = c.outcome(700_025) else { + anyhow::bail!("new generation append missing") + }; + // Kill the active leader. Resolve the exact new write on the surviving quorum. + c.peers.remove(&leader); + c.queue.clear(); + let survivors: Vec<_> = (4..=6).filter(|id| *id != leader).collect(); + let replacement = c.elect(survivors[0], 20, &survivors)?; + c.command(replacement, Request::Propose(20, 700_026, next.clone()))?; + c.pump(&survivors)?; + assert_eq!( + c.outcome(700_026), + Some(Outcome::Committed(Applied::Accepted { index })) + ); + c.start(leader)?; + c.elect(replacement, 20, &[4, 5, 6])?; + for id in 4..=6 { + let Response::Image(_, _, bytes) = c + .peers + .get_mut(&id) + .unwrap() + .call(Request::PrepareCheckpoint(20, Uuid::new_v4()))? + else { + anyhow::bail!("slot image missing") + }; + assert_eq!(bytes, next.body.len() as u64); + c.command(id, Request::InstallCheckpoint)?; + c.pump(&[4, 5, 6])?; + } + c.peers.clear(); + c.queue.clear(); + for id in 1..=6 { + c.start(id)?; + } + let leader = c.elect(4, 20, &[4, 5, 6])?; + assert_eq!( + control(&mut c, leader, 700_027, request)?, + Reply::Opened { + producer: producer::producer_id(0, 2), + generation: 2, + closed: false + } + ); + c.command(leader, Request::Propose(20, 700_028, input.clone()))?; + c.command(leader, Request::Propose(20, 700_029, next.clone()))?; + c.pump(&[4, 5, 6])?; + assert_eq!( + c.outcome(700_028), + Some(Outcome::Committed(Applied::ProducerFenced)) + ); + assert_eq!( + c.outcome(700_029), + Some(Outcome::Committed(Applied::Accepted { index })) + ); + let current = view(&mut c, leader, 700_030)?; + assert_eq!(current.capacity, 1); + assert_eq!(current.last, Some(Last { sequence: 1, index })); + c.command( + leader, + Request::Query( + 20, + 700_031, + Query::Producer { + table, + bucket: 1, + producer: input.meta.producer, + }, + ), + )?; + c.pump(&[4, 5, 6])?; + assert!(matches!( + c.read_outcome(700_031), + Some(ReadOutcome::Observed { + value: Value::Producer(None), + .. + }) + )); + for id in 4..=6 { + verify_records( + c.peers.get_mut(&id).unwrap(), + 20, + std::slice::from_ref(&next), + )?; + } + Ok(()) +} diff --git a/crates/lance-context-wal/tests/replication_process/producer_slots.rs b/crates/lance-context-wal/tests/replication_process/producer_slots.rs new file mode 100644 index 0000000..01dadb6 --- /dev/null +++ b/crates/lance-context-wal/tests/replication_process/producer_slots.rs @@ -0,0 +1,164 @@ +use super::*; +use lance_context_wal::replication::producer::{self, Action, Command as ProducerCommand, Reply}; + +#[test] +fn producer_generations_survive_interrupted_snapshot_and_all_process_restart() -> Result<()> { + let layout = Layout { + cluster: Uuid::from_u128(808), + nodes: (1..=3) + .map(|id| (id, format!("wal-{id}.example.invalid:7800"))) + .collect(), + groups: BTreeMap::from([(10, vec![1, 2, 3])]), + buckets: vec![10], + }; + let command = |action| ProducerCommand { + cluster: layout.cluster, + layout: layout.fingerprint().unwrap(), + group: 10, + table: Uuid::from_u128(809), + slot: 0, + action, + }; + let open = |generation| { + command(Action::Open { + expected_generation: generation, + operation: Uuid::from_u128(810 + generation as u128), + bucket: 0, + }) + }; + let mut c = Cluster::with_producer_slots(layout.clone(), None, 1)?; + let leader = c.elect(1, 10, &[1, 2])?; + for generation in 0..8 { + let token = 1000 + generation * 2; + c.command(leader, Request::Producer(token, open(generation)))?; + c.pump(&[1, 2])?; + assert_eq!( + c.outcome(token), + Some(Outcome::Committed(Applied::Producer(Reply::Opened { + producer: producer::producer_id(0, generation + 1), + generation: generation + 1, + closed: false + }))) + ); + c.command( + leader, + Request::Producer( + token + 1, + command(Action::Revoke { + generation: generation + 1, + }), + ), + )?; + c.pump(&[1, 2])?; + assert_eq!( + c.outcome(token + 1), + Some(Outcome::Committed(Applied::Producer(Reply::Closed { + last: None + }))) + ); + } + let Response::Status(cut) = c + .peers + .get_mut(&leader) + .unwrap() + .call(Request::Status(10))? + else { + anyhow::bail!("missing cut") + }; + assert!(matches!( + c.peers + .get_mut(&leader) + .unwrap() + .call(Request::PrepareCheckpoint(10, Uuid::new_v4()))?, + Response::Image(_, _, 0) + )); + c.command(leader, Request::InstallCheckpoint)?; + c.pump(&[1, 2])?; + c.interrupt_next_snapshot = true; + for _ in 0..30 { + c.command(leader, Request::Tick)?; + c.pump(&[1, 2, 3])?; + if !c.peers.contains_key(&3) { + break; + } + } + assert!( + !c.peers.contains_key(&3), + "snapshot interruption was not reached" + ); + c.start(3)?; + for _ in 0..30 { + c.command(leader, Request::Tick)?; + c.pump(&[1, 2, 3])?; + let Response::Status(status) = c.peers.get_mut(&3).unwrap().call(Request::Status(10))? + else { + anyhow::bail!("missing follower status") + }; + if status.applied >= cut.applied { + break; + } + } + assert!(c.transfers >= 2, "snapshot retry not exercised"); + c.peers.clear(); + c.queue.clear(); + for id in 1..=3 { + c.start(id)?; + } + let leader = c.elect(3, 10, &[1, 2, 3])?; + c.command( + leader, + Request::Query(10, 2000, Query::ProducerSlot { slot: 0 }), + )?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + value: Value::ProducerSlot(Some(view)), + .. + }) = c.read_outcome(2000) + else { + anyhow::bail!("slot lost on restart") + }; + assert_eq!(view.generation, 8); + assert!(view.reusable); + c.command(leader, Request::Producer(2001, open(0)))?; + c.pump(&[1, 2, 3])?; + assert_eq!( + c.outcome(2001), + Some(Outcome::Committed(Applied::Producer(Reply::Fenced))) + ); + c.command(leader, Request::Producer(2002, open(7)))?; + c.pump(&[leader])?; + assert!( + c.outcome(2002).is_none(), + "isolated leader returned an allocation retry from cache" + ); + for _ in 0..10 { + for id in 1..=3 { + c.command(id, Request::Tick)?; + } + c.pump(&[1, 2, 3])?; + if c.completions + .iter() + .any(|completion| completion.request == 2002) + { + break; + } + } + assert_eq!( + c.outcome(2002), + Some(Outcome::Committed(Applied::Producer(Reply::Opened { + producer: producer::producer_id(0, 8), + generation: 8, + closed: true + }))) + ); + c.command(leader, Request::Producer(2003, open(8)))?; + c.pump(&[1, 2, 3])?; + assert!(matches!( + c.outcome(2003), + Some(Outcome::Committed(Applied::Producer(Reply::Opened { + generation: 9, + .. + }))) + )); + Ok(()) +} diff --git a/docs/design/wal-producer-slots.md b/docs/design/wal-producer-slots.md new file mode 100644 index 0000000..e2ea7ad --- /dev/null +++ b/docs/design/wal-producer-slots.md @@ -0,0 +1,56 @@ +# Bounded producer slots + +The optional replicated WAL can use a fixed producer slot table instead of +retaining a UUID for every historical writer. Set `Limits::producer_slots` when +creating fresh groups. Zero keeps the existing UUID-based behavior. Capacity is +at most 16,384 cells per group and cannot exceed `max_producers`; this is a count +bound, not a measured memory budget. Session keys still determine bucket routing. +Producer lanes are reusable sequencing identities, not business sessions or TCP +connections. This is part of [the ingestion work in #378](https://github.com/lance-format/lance-context/issues/378). + +`Node::producer_command` submits an allocation, close or administrative revocation +through the ordinary durable Raft path. A command carries cluster, layout, group, +table and slot scope. An allocation additionally supplies the expected generation, +bucket and a saved operation UUID. Its exact retry returns the same generation; +competing or stale allocations cannot silently pick another slot. A caller must +persist this intent before sending it. `Query::ProducerSlot` provides a bounded +quorum view for choosing a candidate, but the eventual allocation still needs the +generation comparison in the committed state machine. +An allocation retry can resolve an already-used lane. `Opened` does not reset its +sequence or prove it is unused; recovering clients must retain their original +sequence/pending-body journal and reconcile the quorum tail before sending data. + +A close supplies the current generation and exact latest sequence/index. If an +append committed but its reply was lost, a close using an older tail returns the +actual tail without closing the lane. Once closed, new appends are fenced, while +the exact retained latest append can still return its original receipt. A trusted +administrative revoke also closes a generation but preserves its unresolved tail. +There is no timer-based eviction. + +Reusing a slot requires both closure and retirement of its exact latest record by +the existing catalog-authorized source closure. Publication alone, a local timeout +or a closed flag does not establish retirement. Reuse removes the old receipt and +retirement marker and increments the cell's durable generation. Old append, open, +close and revoke requests are then fenced. Generation exhaustion never wraps. +An explicit fencing result does not mean the old write never committed and must +not cause automatic replay under a new identity. Historical receipt lookup beyond +slot reuse requires a separate contract and index. + +Checkpoint metadata stores a flat, bounded cell table including unused and closed +cells. The reader checks the immutable policy's count before allocating cells and +validates each bounded frame, lane binding and retained receipt. Snapshot transfer +uses the same self-contained image. Zero-slot groups retain policy identity v9/v10 +and image format 9; enabled groups use policy identity v11 and image format 10. +Opening an existing UUID-only directory with slots enabled is rejected. Mixed +versions and in-place migration are not implemented. + +Scope checks do not authenticate clients or authorize revocation. Production +transport, administrative authorization, SDK journal integration and migration +remain separate work. This change adds no Arrow decode on a replica and does not +claim a throughput improvement or qualify sustained producer churn. + +Validation includes state-machine bounds and stale-tail rejection, checkpoint +round trips, an interrupted snapshot transfer between real Raft processes, and a +real Lance publication/retirement fixture. The latter closes a producer before +publication, rejects premature reuse, then reuses it after signed source closure +and verifies old-request fencing, new-generation leader failover and full restart.