diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 6708405..3867197 100644 --- a/crates/lance-context-wal/Cargo.toml +++ b/crates/lance-context-wal/Cargo.toml @@ -8,7 +8,7 @@ description = "Independent Rust WAL experiments owned by lance-context" [features] client-journal = ["dep:anyhow", "dep:raft-engine", "dep:bytes", "dep:serde", "dep:serde_json", "dep:tokio", "tokio/sync"] -replication = ["raft-log", "record", "dep:bincode", "dep:slog"] +replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog"] raft-log = ["placement", "dep:raft", "dep:raft-engine", "dep:raft-proto", "dep:protobuf"] vectored-io = ["dep:bytes", "dep:tokio"] runner = ["dep:anyhow", "dep:clap", "dep:libc", "dep:serde", "dep:serde_json", "dep:sha2"] diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index 1615289..b9b057a 100644 --- a/crates/lance-context-wal/README.md +++ b/crates/lance-context-wal/README.md @@ -168,6 +168,10 @@ cases; their numbers are not performance evidence. ## Durable Raft log storage +The replication layer also provides [durable publisher bindings and epochs](../../docs/design/wal-publisher-authority.md) +through its catalog group's normal consensus path. This is an incremental +publication prerequisite, not a production Lance writer or a manifest lease. + The optional `raft-log` feature exposes `raft_log::PhysicalLog` and its cloneable `GroupLog` handles. One raft-engine directory hosts all groups assigned to a node by the existing immutable `placement::Layout`. Opening binds the full layout, diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 3e3e762..04b8eca 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -14,6 +14,7 @@ pub mod checkpoint; mod images; +pub mod publication; mod state; #[cfg(test)] mod tests; @@ -40,6 +41,7 @@ use state::State; /// policy. These bounds do not account for queues in the caller's transport. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Limits { + pub max_tables: u64, pub max_record_bytes: u64, pub max_records: u64, pub max_producers: u64, @@ -49,6 +51,7 @@ pub struct Limits { impl Default for Limits { fn default() -> Self { Self { + max_tables: 1024, max_record_bytes: 8 * 1024 * 1024, max_records: 100_000, max_producers: 16_384, @@ -60,7 +63,8 @@ impl Default for Limits { impl Limits { fn identity(&self) -> Result> { ensure!( - (1..=8 * 1024 * 1024).contains(&self.max_record_bytes) + (1..=16_384).contains(&self.max_tables) + && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes) && (1..=500_000).contains(&self.max_records) && (1..=100_000).contains(&self.max_producers) && (1..=4096).contains(&self.max_pending) @@ -69,7 +73,7 @@ impl Limits { "invalid replication limits" ); Ok(bincode::serialize(&( - "lance-context/replicated-append/v1", + "lance-context/replicated-append/v2", self, ))?) } @@ -88,6 +92,7 @@ pub struct Packet { #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Applied { + Publisher(publication::Reply), /// A new append or exact latest retry. Index identifies the original append. Accepted { index: u64, @@ -298,6 +303,26 @@ impl Node { }) } + /// Queue a binding/epoch CAS in the one catalog group. Only a committed + /// Publisher completion from poll grants authority; no local cache shortcut. + pub fn acquire_publisher(&mut self, request: u64, acquire: publication::Acquire) -> Result<()> { + let batch = acquire.batch(&self.layout)?; + self.propose(publication::control_group(&self.layout), request, batch) + } + + /// Local applied view, not a quorum read or proof that an owner is current. + pub fn publisher(&self, table: Uuid) -> Result> { + self.healthy()?; + Ok(self + .groups + .get(&publication::control_group(&self.layout)) + .context("catalog group not hosted")? + .state + .catalog + .tables + .get(&table)) + } + /// Ok means queued locally, NOT durable or committed. A token may have only /// one unresolved request on this node. No retry result bypasses consensus. pub fn propose(&mut self, group: u64, request: u64, batch: Batch) -> Result<()> { diff --git a/crates/lance-context-wal/src/replication/checkpoint.rs b/crates/lance-context-wal/src/replication/checkpoint.rs index cc2df77..3934c65 100644 --- a/crates/lance-context-wal/src/replication/checkpoint.rs +++ b/crates/lance-context-wal/src/replication/checkpoint.rs @@ -14,8 +14,8 @@ use std::{ path::PathBuf, }; -const FORMAT: u32 = 1; -const MAGIC: &[u8; 8] = b"LCWALCP1"; +const FORMAT: u32 = 2; +const MAGIC: &[u8; 8] = b"LCWALCP2"; pub const CHUNK_BYTES: usize = 1024 * 1024; fn codec() -> impl Options { @@ -71,6 +71,7 @@ struct Descriptor { id: Uuid, records: u64, producers: u64, + tables: u64, metadata_bytes: u64, payload_bytes: u64, metadata_digest: [u8; 32], @@ -80,7 +81,9 @@ struct Descriptor { fn metadata_limit(limits: &Limits) -> u64 { // A framed record needs at most 8 + 4 + 256 bytes, plus a bounded header. // limits.identity() validates all policy maxima before this is used. - (limits.max_records + limits.max_producers) * 268 + 1024 + (limits.max_records + limits.max_producers) * 268 + + limits.max_tables * (publication::MAX_BYTES as u64 + 4) + + 1024 } impl Descriptor { @@ -92,6 +95,8 @@ impl Descriptor { && self.records <= limits.max_records && self.producers <= limits.max_producers && self.producers <= self.records + && self.tables <= limits.max_tables + && (group == publication::control_group(layout) || self.tables == 0) && self.metadata_bytes <= metadata_limit(limits) && self.payload_bytes <= self.records * limits.max_record_bytes, "application image format or bounds mismatch" @@ -421,6 +426,7 @@ pub(super) fn build( id, records: state.accepted.len() as u64, producers: state.producers.len() as u64, + tables: state.catalog.tables.len() as u64, metadata_bytes: 0, payload_bytes, metadata_digest: [0; 32], @@ -432,12 +438,16 @@ pub(super) fn build( write_frame(&mut metadata, &codec().serialize(&descriptor.binding)?)?; write_u64(&mut metadata, descriptor.records)?; write_u64(&mut metadata, descriptor.producers)?; + write_u64(&mut metadata, descriptor.tables)?; for (index, meta) in &state.accepted { write_meta(&mut metadata, *index, meta)?; } for (meta, index) in state.producers.values() { write_meta(&mut metadata, *index, meta)?; } + for table in state.catalog.tables.values() { + write_frame(&mut metadata, &publication::encode_table(table)?)?; + } (descriptor.metadata_bytes, descriptor.metadata_digest) = metadata.finish()?; let mut pack = Writer::new(&paths[1], payload_bytes)?; for (index, meta) in &state.accepted { @@ -483,7 +493,8 @@ pub(super) fn load( ); ensure!( read_u64(&mut metadata)? == descriptor.records - && read_u64(&mut metadata)? == descriptor.producers, + && read_u64(&mut metadata)? == descriptor.producers + && read_u64(&mut metadata)? == descriptor.tables, "image metadata count mismatch" ); let mut state = State { @@ -511,6 +522,18 @@ pub(super) fn load( state.producers.insert(key, (meta, index)); previous = Some(key); } + let mut previous = None; + for _ in 0..descriptor.tables { + let (bytes, n) = read_frame::<{ publication::MAX_BYTES }>(&mut metadata)?; + let table = publication::decode_table(&bytes[..n])?; + let id = table.authority.table; + ensure!( + previous.is_none_or(|old| old < id), + "unordered or duplicate image table" + ); + state.catalog.tables.insert(id, table); + previous = Some(id); + } let metadata_file = metadata.finish(descriptor.metadata_bytes, descriptor.metadata_digest)?; state.validate_checkpoint(layout, group, limits, descriptor.binding.index)?; let (locations, payload_bytes) = locations(&state)?; diff --git a/crates/lance-context-wal/src/replication/checkpoint/tests.rs b/crates/lance-context-wal/src/replication/checkpoint/tests.rs index d8eb101..42c6eb2 100644 --- a/crates/lance-context-wal/src/replication/checkpoint/tests.rs +++ b/crates/lance-context-wal/src/replication/checkpoint/tests.rs @@ -289,3 +289,118 @@ fn empty_application_cut_has_a_real_empty_pack_and_no_payload_fetch() -> Result< ); Ok(()) } + +#[test] +fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() -> Result<()> { + use crate::ingest_contract::{IngestContract, WritePolicy}; + let root = tempfile::tempdir()?; + let (mut state, bodies) = fixture()?; + for id in [200, 201] { + let batch = publication::Acquire { + table: Uuid::from_u128(id), + destination: publication::Destination { + namespace: Uuid::from_u128(202), + dataset: format!("tables/{id}"), + }, + ingest: IngestContract { + tenant: "test".into(), + qualified_table: format!("table-{id}"), + schema: [7; 32], + policy: WritePolicy::ImmutableUniqueIds, + max_record_bytes: 4096, + max_rows: 32, + }, + owner: Uuid::from_u128(203), + request: Uuid::from_u128(id + 10), + expected_epoch: 0, + } + .batch(&layout())?; + let entry = Entry { + index: state.applied + 1, + term: 2, + context: state::encode_meta(&batch.meta)?.into(), + data: batch.body, + ..Default::default() + }; + state::validate_entry(&layout(), 10, &Limits::default(), &entry, true)?; + state.apply(&entry, &Limits::default())?; + } + let built = build( + root.path(), + &layout(), + &Limits::default(), + Cut { + group: 10, + index: state.applied, + term: 2, + id: Uuid::from_u128(220), + }, + state.clone(), + |index, _| Ok(bodies[&index].clone()), + )?; + let snapshot = built.image.snapshot()?; + assert_eq!( + load(root.path(), &snapshot, &layout(), &Limits::default(), 10)?.state, + state + ); + assert_eq!(built.image.bytes(Part::Payload), 8192); + for index in [2, 5] { + assert_eq!( + built.image.read_for_validation(index)?, + Some(bodies[&index].clone()) + ); + } + let path = built.image.descriptor.paths(root.path())[0].clone(); + let original = fs::read(&path)?; + // Locate frames from their encoded lengths, without assuming a fixed header. + let binding_len = u32::from_le_bytes(original[8..12].try_into()?) as usize; + let mut offset = 12 + binding_len + 3 * 8; + for _ in 0..state.accepted.len() + state.producers.len() { + offset += 8; // accepted index + let len = u32::from_le_bytes(original[offset..offset + 4].try_into()?) as usize; + offset += 4 + len; + } + let first_len = u32::from_le_bytes(original[offset..offset + 4].try_into()?) as usize; + let second = offset + 4 + first_len; + let second_len = u32::from_le_bytes(original[second..second + 4].try_into()?) as usize; + let first_table = publication::decode_table(&original[offset + 4..second])?; + let second_table = publication::decode_table(&original[second + 4..second + 4 + second_len])?; + for mutation in 0..4 { + let mut bytes = original[..offset].to_vec(); + let mut tables = [first_table.clone(), second_table.clone()]; + match mutation { + 0 => tables.swap(0, 1), + 1 => tables[1] = tables[0].clone(), + 2 => tables[1].destination = tables[0].destination.clone(), + _ => tables[1].authority.grant_index = state.applied + 1, + } + for table in &tables { + write_frame(&mut bytes, &publication::encode_table(table)?)?; + } + fs::write(&path, &bytes)?; + let mut descriptor = built.image.descriptor.clone(); + descriptor.metadata_bytes = bytes.len() as u64; + descriptor.metadata_digest = Sha256::digest(&bytes).into(); + assert!(load( + root.path(), + &descriptor.snapshot()?, + &layout(), + &Limits::default(), + 10 + ) + .is_err()); + } + fs::write(&path, original)?; + let target = tempfile::tempdir()?; + let mut receiver = Receiver::new( + target.path(), + &snapshot, + layout(), + Limits::default(), + 10, + Uuid::from_u128(221), + )?; + copy(&built.image, &mut receiver)?; + assert_eq!(receiver.finish()?.state, state); + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/publication.rs b/crates/lance-context-wal/src/replication/publication.rs new file mode 100644 index 0000000..2f14c87 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication.rs @@ -0,0 +1,254 @@ +//! Replicated table bindings and publisher epochs. These commands run through +//! the same durable Raft path as appends; a local view is not an active lease. +//! Publication plans, manifest outcomes and source retirement build on these +//! epochs in subsequent changes. No Lance manifest is changed by this module. +use super::*; +use crate::{ingest_contract::IngestContract, record::RecordKind}; +use bincode::Options; +use bytes::Bytes; +#[cfg(test)] +mod tests; + +pub(super) const MAX_BYTES: usize = 8192; +fn codec() -> impl Options { + bincode::DefaultOptions::new() + .with_fixint_encoding() + .with_little_endian() + .with_limit(MAX_BYTES as u64) + .reject_trailing_bytes() +} + +/// The authorized catalog supplies one stable storage namespace and a canonical +/// dataset key within it. Raw aliases or client-provided URIs are not identities. +/// This representation does not discover aliases or authenticate the caller. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +pub struct Destination { + pub namespace: Uuid, + pub dataset: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Acquire { + pub table: Uuid, + pub destination: Destination, + pub ingest: IngestContract, + pub owner: Uuid, + /// Stable across an ambiguous retry; use a fresh UUID for a new takeover. + pub request: Uuid, + pub expected_epoch: u64, +} +impl Acquire { + fn validate(&self) -> Result<()> { + ensure!( + !self.table.is_nil() && !self.owner.is_nil() && !self.request.is_nil(), + "nil publisher identity" + ); + validate_binding(&self.destination, &self.ingest)?; + ensure!(self.expected_epoch < u64::MAX, "publisher epoch exhausted"); + Ok(()) + } + pub(super) fn batch(self, layout: &Layout) -> Result { + self.validate()?; + let envelope = Envelope { + version: 1, + cluster: layout.cluster, + layout: layout.fingerprint()?, + group: control_group(layout), + acquire: self, + }; + let bytes = codec().serialize(&envelope)?; + ensure!(bytes.len() <= MAX_BYTES, "publisher command too large"); + let mut batch = Batch::new( + envelope.acquire.table, + 0, + envelope.acquire.owner, + envelope.acquire.expected_epoch + 1, + Bytes::from(bytes), + ); + batch.meta.kind = RecordKind::MultiPublication; + Ok(batch) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Authority { + pub cluster: Uuid, + pub layout: [u8; 32], + pub group: u64, + pub table: Uuid, + pub owner: Uuid, + pub epoch: u64, + pub grant_index: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Table { + pub destination: Destination, + pub ingest: IngestContract, + pub authority: Authority, + pub request: Uuid, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Rejection { + BindingConflict, + DestinationBound, + Epoch { current: u64 }, + RequestReused, + Capacity, +} +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Reply { + Acquired(Authority), + Rejected(Rejection), +} + +#[derive(Serialize, Deserialize)] +pub(super) struct Envelope { + version: u32, + cluster: Uuid, + layout: [u8; 32], + group: u64, + acquire: Acquire, +} +pub(super) fn control_group(layout: &Layout) -> u64 { + *layout + .groups + .first_key_value() + .expect("validated nonempty layout") + .0 +} +fn validate_binding(destination: &Destination, ingest: &IngestContract) -> Result<()> { + ensure!( + !destination.namespace.is_nil() + && !destination.dataset.is_empty() + && destination.dataset.len() <= 4096, + "invalid canonical destination" + ); + ingest.validate()?; + Ok(()) +} +pub(super) fn decode(bytes: &[u8]) -> Result { + ensure!(bytes.len() <= MAX_BYTES, "publisher command too large"); + let envelope: Envelope = codec().deserialize(bytes)?; + ensure!(envelope.version == 1, "unsupported publisher command"); + envelope.acquire.validate()?; + Ok(envelope) +} +pub(super) fn validate_record( + layout: &Layout, + group: u64, + limits: &Limits, + meta: &BatchMeta, + body: &[u8], +) -> Result<()> { + let e = decode(body)?; + ensure!( + group == control_group(layout) + && e.group == group + && e.cluster == layout.cluster + && e.layout == layout.fingerprint()? + && meta.kind == RecordKind::MultiPublication + && meta.stream == e.acquire.table + && meta.bucket == 0 + && meta.producer == e.acquire.owner + && meta.sequence == e.acquire.expected_epoch + 1 + && u64::from(e.acquire.ingest.max_record_bytes) <= limits.max_record_bytes, + "publisher scope/envelope or ingestion limit mismatch" + ); + Ok(()) +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub(super) struct Catalog { + pub tables: BTreeMap, +} +impl Catalog { + pub fn apply(&mut self, e: Envelope, index: u64, max_tables: u64) -> Reply { + let a = e.acquire; + let old = self.tables.get(&a.table); + if let Some(old) = old { + if old.destination != a.destination || old.ingest != a.ingest { + return Reply::Rejected(Rejection::BindingConflict); + } + if old.request == a.request { + return if old.authority.owner == a.owner + && old.authority.epoch == a.expected_epoch + 1 + { + Reply::Acquired(old.authority.clone()) + } else { + Reply::Rejected(Rejection::RequestReused) + }; + } + } + let current = old.map_or(0, |old| old.authority.epoch); + if a.expected_epoch != current { + return Reply::Rejected(Rejection::Epoch { current }); + } + if old.is_none() { + if self.tables.values().any(|t| t.destination == a.destination) { + return Reply::Rejected(Rejection::DestinationBound); + } + if self.tables.len() as u64 >= max_tables { + return Reply::Rejected(Rejection::Capacity); + } + } + let authority = Authority { + cluster: e.cluster, + layout: e.layout, + group: e.group, + table: a.table, + owner: a.owner, + epoch: current + 1, + grant_index: index, + }; + self.tables.insert( + a.table, + Table { + destination: a.destination, + ingest: a.ingest, + authority: authority.clone(), + request: a.request, + }, + ); + Reply::Acquired(authority) + } + pub fn validate(&self, layout: &Layout, group: u64, cut: u64, limits: &Limits) -> Result<()> { + ensure!( + self.tables.len() as u64 <= limits.max_tables, + "publisher catalog exceeds limit" + ); + ensure!( + group == control_group(layout) || self.tables.is_empty(), + "catalog in source group" + ); + let fingerprint = layout.fingerprint()?; + let mut destinations = std::collections::BTreeSet::new(); + for (id, table) in &self.tables { + validate_binding(&table.destination, &table.ingest)?; + let a = &table.authority; + ensure!( + !id.is_nil() + && *id == a.table + && !a.owner.is_nil() + && !table.request.is_nil() + && a.cluster == layout.cluster + && a.layout == fingerprint + && a.group == group + && a.epoch > 0 + && a.epoch <= a.grant_index + && a.grant_index <= cut + && u64::from(table.ingest.max_record_bytes) <= limits.max_record_bytes + && destinations.insert(&table.destination), + "invalid publisher checkpoint state" + ); + } + Ok(()) + } +} +pub(super) fn encode_table(table: &Table) -> Result> { + Ok(codec().serialize(table)?) +} +pub(super) fn decode_table(bytes: &[u8]) -> Result { + Ok(codec().deserialize(bytes)?) +} diff --git a/crates/lance-context-wal/src/replication/publication/tests.rs b/crates/lance-context-wal/src/replication/publication/tests.rs new file mode 100644 index 0000000..e459b55 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/tests.rs @@ -0,0 +1,191 @@ +use super::*; +use crate::ingest_contract::WritePolicy; + +fn input() -> Acquire { + Acquire { + table: Uuid::from_u128(10), + destination: Destination { + namespace: Uuid::from_u128(11), + dataset: "canonical/table".into(), + }, + ingest: IngestContract { + tenant: "tenant".into(), + qualified_table: "table".into(), + schema: [3; 32], + policy: WritePolicy::ImmutableUniqueIds, + max_record_bytes: 1024 * 1024, + max_rows: 1024, + }, + owner: Uuid::from_u128(12), + request: Uuid::from_u128(13), + expected_epoch: 0, + } +} +fn apply(catalog: &mut Catalog, acquire: Acquire, index: u64, capacity: u64) -> Result { + let batch = acquire.batch(&super::super::tests::layout())?; + Ok(catalog.apply(decode(&batch.body)?, index, capacity)) +} + +#[test] +fn competing_acquisitions_bind_once_and_retry_only_the_exact_current_grant() -> Result<()> { + let mut catalog = Catalog::default(); + let mut acquire = input(); + let first = apply(&mut catalog, acquire.clone(), 5, 1)?; + assert_eq!(apply(&mut catalog, acquire.clone(), 6, 1)?, first); + let mut changed = acquire.clone(); + changed.ingest.schema[0] ^= 1; + assert_eq!( + apply(&mut catalog, changed, 7, 1)?, + Reply::Rejected(Rejection::BindingConflict) + ); + changed = acquire.clone(); + changed.expected_epoch = 1; + assert_eq!( + apply(&mut catalog, changed, 8, 1)?, + Reply::Rejected(Rejection::RequestReused) + ); + changed = acquire.clone(); + changed.owner = Uuid::from_u128(20); + changed.request = Uuid::from_u128(21); + assert_eq!( + apply(&mut catalog, changed.clone(), 9, 1)?, + Reply::Rejected(Rejection::Epoch { current: 1 }) + ); + changed.expected_epoch = 1; + let second = apply(&mut catalog, changed.clone(), 10, 1)?; + assert!(matches!(&second, Reply::Acquired(a) if a.epoch == 2 && a.grant_index == 10)); + assert_eq!(apply(&mut catalog, changed, 11, 1)?, second); + assert_eq!( + apply(&mut catalog, acquire.clone(), 12, 1)?, + Reply::Rejected(Rejection::Epoch { current: 2 }) + ); + acquire.table = Uuid::from_u128(22); + assert_eq!( + apply(&mut catalog, acquire.clone(), 13, 8)?, + Reply::Rejected(Rejection::DestinationBound) + ); + acquire.destination.namespace = Uuid::from_u128(23); + assert_eq!( + apply(&mut catalog, acquire.clone(), 14, 1)?, + Reply::Rejected(Rejection::Capacity) + ); + assert!(matches!( + apply(&mut catalog, acquire, 15, 8)?, + Reply::Acquired(_) + )); + assert_eq!(catalog.tables.len(), 2); + Ok(()) +} + +#[test] +fn bounded_decode_rejects_wrong_scopes_domains_and_unserviceable_ingest_limits() -> Result<()> { + let layout = super::super::tests::layout(); + let limits = Limits::default(); + let mut batch = input().batch(&layout)?; + validate_record(&layout, 10, &limits, &batch.meta, &batch.body)?; + assert!(validate_record(&layout, 20, &limits, &batch.meta, &batch.body).is_err()); + let mut foreign = (*layout).clone(); + foreign.cluster = Uuid::from_u128(100); + assert!(validate_record(&foreign, 10, &limits, &batch.meta, &batch.body).is_err()); + batch.meta.stream = Uuid::from_u128(101); + assert!(validate_record(&layout, 10, &limits, &batch.meta, &batch.body).is_err()); + batch = input().batch(&layout)?; + let mut trailing = batch.body.to_vec(); + trailing.push(0); + assert!(decode(&trailing).is_err()); + assert!(decode(&vec![0; MAX_BYTES + 1]).is_err()); + let mut oversized = input(); + oversized.destination.dataset = "x".repeat(4097); + assert!(oversized.batch(&layout).is_err()); + let mut exhausted = input(); + exhausted.expected_epoch = u64::MAX; + assert!(exhausted.batch(&layout).is_err()); + let mut contract = input(); + contract.ingest.max_record_bytes = 16 * 1024 * 1024; + batch = contract.batch(&layout)?; + assert!(validate_record(&layout, 10, &limits, &batch.meta, &batch.body).is_err()); + Ok(()) +} + +#[test] +fn checkpoint_catalog_requires_exact_scope_ordering_identity_and_unique_destinations() -> Result<()> +{ + let layout = super::super::tests::layout(); + let limits = Limits::default(); + let mut catalog = Catalog::default(); + let acquire = input(); + apply(&mut catalog, acquire.clone(), 4, 8)?; + catalog.validate(&layout, 10, 4, &limits)?; + assert!(catalog.validate(&layout, 20, 4, &limits).is_err()); + assert!(catalog.validate(&layout, 10, 3, &limits).is_err()); + let original = catalog.tables[&acquire.table].clone(); + assert_eq!(decode_table(&encode_table(&original)?)?, original); + for mutate in [ + (|t: &mut Table| t.authority.epoch = 5) as fn(&mut Table), + |t| t.authority.grant_index = 0, + |t| t.authority.owner = Uuid::nil(), + |t| t.authority.layout[0] ^= 1, + |t| t.request = Uuid::nil(), + |t| t.ingest.schema = [0; 32], + ] { + let mut table = original.clone(); + mutate(&mut table); + let changed = Catalog { + tables: BTreeMap::from([(acquire.table, table)]), + }; + assert!(changed.validate(&layout, 10, 4, &limits).is_err()); + } + let other = Uuid::from_u128(400); + let mut duplicate = original; + duplicate.authority.table = other; + catalog.tables.insert(other, duplicate); + assert!(catalog.validate(&layout, 10, 4, &limits).is_err()); + Ok(()) +} + +#[test] +fn failed_follower_sync_cannot_release_a_publisher_grant() -> Result<()> { + let root = tempfile::tempdir()?; + let layout = super::super::tests::layout(); + let mut nodes = (1..=3) + .map(|id| { + Node::open( + &root.path().join(id.to_string()), + layout.clone(), + id, + Limits::default(), + ) + }) + .collect::>>()?; + nodes[0].campaign(10)?; + // Elect with real persistent logs, then cut delivery to the third replica. + for _ in 0..100 { + let mut packets = Vec::new(); + for node in &mut nodes { + packets.extend(node.poll()?.messages); + } + if packets.is_empty() { + break; + } + for packet in packets { + nodes[packet.message.to as usize - 1].step(packet)?; + } + } + nodes[0].acquire_publisher(1, input())?; + let events = nodes[0].poll()?; + assert!(events.completions.is_empty()); + assert!(nodes[0].publisher(input().table)?.is_none()); + let mut delivered = 0; + for packet in events.messages.into_iter().filter(|p| p.message.to == 2) { + delivered += 1; + nodes[1].step(packet)?; + } + assert!(delivered > 0, "publisher proposal never reached follower"); + nodes[1].groups[&10].raw.store().fail_next_sync_for_test(); + assert!(nodes[1].poll().is_err()); + assert!(nodes[1].publisher(input().table).is_err()); + assert!(nodes[1].status(20).is_err(), "shared engine not fenced"); + assert!(nodes[0].poll()?.completions.is_empty()); + assert!(nodes[0].publisher(input().table)?.is_none()); + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/state.rs b/crates/lance-context-wal/src/replication/state.rs index 276d510..abf95f9 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::MultiPublication { + meta.validate_payload(body, publication::MAX_BYTES)?; + return publication::validate_record(layout, group, limits, meta, body); + } validate_identity(layout, group, meta)?; if digest { meta.validate_payload(body, limits.max_record_bytes as usize)?; @@ -75,12 +79,13 @@ pub(super) fn validate_entry( #[derive(Clone, Debug, Default, PartialEq, Eq)] pub(super) struct State { + pub catalog: publication::Catalog, pub applied: u64, pub accepted: BTreeMap, pub producers: BTreeMap<(Uuid, u32, Uuid), (BatchMeta, u64)>, } impl State { - /// The v1 image retains every accepted record. Producer tails must exactly + /// The image retains every accepted record. Producer tails must exactly /// match those records, including sequence continuity within each lane. /// Retirement will need a new format with independently retained tails. pub fn validate_checkpoint( @@ -124,6 +129,7 @@ impl State { tails == self.producers, "checkpoint producer tails disagree" ); + self.catalog.validate(layout, group, cut, limits)?; Ok(()) } @@ -137,6 +143,13 @@ impl State { return Ok(None); } let meta = decode_meta(&entry.context)?; + if meta.kind == RecordKind::MultiPublication { + return Ok(Some(Applied::Publisher(self.catalog.apply( + publication::decode(&entry.data)?, + entry.index, + limits.max_tables, + )))); + } let key = (meta.stream, meta.bucket, meta.producer); let previous = self.producers.get(&key); if let Some((old, index)) = previous { diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index 7d49f1b..6c91232 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -8,6 +8,9 @@ use lance_context_wal::{ record::Batch, replication::{ checkpoint::{Image, Part}, + publication::{ + Acquire, Destination, Rejection as PublisherRejection, Reply as PublisherReply, Table, + }, Applied, Completion, Events, Limits, Node, Outcome, Packet, Page, PreparedCheckpoint, SnapshotReceiver, Status, }, @@ -56,6 +59,8 @@ impl WirePacket { } #[derive(Serialize, Deserialize)] enum Request { + Acquire(u64, Acquire), + Publisher(Uuid), Campaign(u64), Propose(u64, u64, Batch), Step(WirePacket), @@ -74,6 +79,7 @@ enum Request { } #[derive(Serialize, Deserialize)] enum Response { + Publisher(Option
), Events(Vec, Vec), Status(Status), Record(Option), @@ -117,6 +123,10 @@ fn part(payload: bool) -> Part { fn run(worker: &mut Worker, request: Request) -> Result { let node = &mut worker.node; match request { + Request::Acquire(token, acquire) => node.acquire_publisher(token, acquire)?, + Request::Publisher(table) => { + return Ok(Response::Publisher(node.publisher(table)?.cloned())) + } Request::Campaign(group) => node.campaign(group)?, Request::Propose(group, token, batch) => node.propose(group, token, batch)?, Request::Step(packet) => node.step(packet.decode()?)?, @@ -759,3 +769,144 @@ fn durable_quorum_retry_leader_loss_suffix_replacement_and_full_restart() -> Res println!("verified real RF3 processes: durable majority ACK, latest retry, isolated old leader, replaced uncommitted suffix, two groups and full process restart with exact payloads"); Ok(()) } + +#[test] +fn publisher_epochs_require_quorum_and_survive_snapshot_transfer_and_restart() -> Result<()> { + use lance_context_wal::ingest_contract::{IngestContract, WritePolicy}; + let mut c = Cluster::new()?; + let mut acquire = Acquire { + table: Uuid::from_u128(700), + destination: Destination { + namespace: Uuid::from_u128(701), + dataset: "tables/example".into(), + }, + ingest: IngestContract { + tenant: "test".into(), + qualified_table: "example".into(), + schema: [9; 32], + policy: WritePolicy::ImmutableUniqueIds, + max_record_bytes: 1024 * 1024, + max_rows: 1024, + }, + owner: Uuid::from_u128(702), + request: Uuid::from_u128(703), + expected_epoch: 0, + }; + c.command(1, Request::Campaign(10))?; + c.pump(&[1, 2, 3])?; + c.command(1, Request::Acquire(700, acquire.clone()))?; + assert!( + c.outcome(700).is_none(), + "local acquisition released authority without quorum" + ); + c.pump(&[1, 2])?; + let first = c.outcome(700).context("missing committed authority")?; + let Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(first_authority))) = &first + else { + anyhow::bail!("unexpected acquisition {first:?}") + }; + assert_eq!(first_authority.epoch, 1); + drop(c.peers.remove(&1)); + assert_eq!(c.elect(2, 10, &[2, 3])?, 2); + c.command(2, Request::Acquire(701, acquire.clone()))?; + c.pump(&[2, 3])?; + assert_eq!(c.outcome(701), Some(first)); + + // A competing owner wins only at the exact epoch. The binding is immutable. + acquire.owner = Uuid::from_u128(704); + acquire.request = Uuid::from_u128(705); + acquire.expected_epoch = 1; + c.command(2, Request::Acquire(702, acquire.clone()))?; + c.pump(&[2, 3])?; + let second = c.outcome(702).context("takeover did not commit")?; + assert!( + matches!(&second, Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(a))) if a.epoch == 2) + ); + let image = c + .peers + .get_mut(&2) + .unwrap() + .call(Request::PrepareCheckpoint(10, Uuid::from_u128(710)))?; + assert!( + matches!(image, Response::Image(_, _, 0)), + "catalog unexpectedly retained control payloads as data" + ); + c.command(2, Request::InstallCheckpoint)?; + c.pump(&[2, 3])?; + c.start(1)?; + for _ in 0..30 { + c.command(2, Request::Tick)?; + c.pump(&[1, 2, 3])?; + if c.transfers > 0 { + break; + } + } + assert!( + c.transfers > 0, + "lagging owner did not receive an application image" + ); + let Response::Publisher(Some(table)) = c + .peers + .get_mut(&1) + .unwrap() + .call(Request::Publisher(acquire.table))? + else { + anyhow::bail!("snapshot omitted publisher") + }; + assert_eq!(table.authority.epoch, 2); + assert_eq!(table.destination, acquire.destination); + assert_eq!(table.ingest, acquire.ingest); + + c.command(2, Request::Acquire(703, acquire.clone()))?; + c.pump(&[2])?; + assert!( + c.outcome(703).is_none(), + "isolated leader answered an authority retry from cache" + ); + let leader = c.elect(1, 10, &[1, 3])?; + let stale = acquire.clone(); + acquire.owner = Uuid::from_u128(706); + acquire.request = Uuid::from_u128(707); + acquire.expected_epoch = 2; + c.command(leader, Request::Acquire(704, acquire.clone()))?; + c.pump(&[1, 3])?; + let latest = c.outcome(704).context("third epoch unavailable")?; + assert!( + matches!(&latest, Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(a))) if a.epoch == 3) + ); + c.command(leader, Request::Acquire(705, stale))?; + c.pump(&[1, 3])?; + assert_eq!( + c.outcome(705), + Some(Outcome::Committed(Applied::Publisher( + PublisherReply::Rejected(PublisherRejection::Epoch { current: 3 }) + ))) + ); + for _ in 0..4 { + c.command(leader, Request::Tick)?; + c.pump(&[1, 2, 3])?; + } + c.queue.clear(); + c.peers.clear(); + for id in 1..=3 { + c.start(id)?; + } + for peer in c.peers.values_mut() { + let Response::Publisher(Some(table)) = peer.call(Request::Publisher(acquire.table))? else { + anyhow::bail!("restart omitted publisher") + }; + assert_eq!(table.authority.epoch, 3); + assert_eq!(table.authority.owner, acquire.owner); + assert_eq!(table.ingest, acquire.ingest); + verify_records(peer, 10, &[])?; + } + let leader = c.elect(1, 10, &[1, 2, 3])?; + c.command(leader, Request::Acquire(706, acquire))?; + c.pump(&[1, 2, 3])?; + assert_eq!( + c.outcome(706), + Some(latest), + "restart changed the committed grant index" + ); + Ok(()) +} diff --git a/docs/design/wal-publisher-authority.md b/docs/design/wal-publisher-authority.md new file mode 100644 index 0000000..f605c09 --- /dev/null +++ b/docs/design/wal-publisher-authority.md @@ -0,0 +1,56 @@ +# Durable table bindings and publisher epochs + +Refs #378. + +A publisher restart must recover the same table generation and distinguish its +authority from a previous owner's. The optional replication component now puts +table registration and expected-epoch acquisition through the catalog group's +normal Ready/commit/apply path. The lowest configured group is the one catalog +group for the layout. Independent source groups cannot register separate owners +for the same destination. + +`Node::acquire_publisher` queues an `Acquire` command. Only +`Outcome::Committed(Applied::Publisher(Reply::Acquired(...)))` grants its authority; +the authority includes cluster, layout, group, table, owner, epoch and original +grant index. An exact retry of the current grant still goes through consensus and +returns that same index. An isolated former leader cannot complete it from local +state. A new request must compare the exact current epoch before advancing it. +Older grants do not have unlimited retained request history. + +Registration binds the table UUID to a canonical destination and complete +`IngestContract`. Both remain immutable across takeovers. Destinations pair a +stable storage-namespace UUID with the catalog's canonical dataset key. A second +table cannot claim an already bound destination. Caller authentication and alias +canonicalization belong to the authorized catalog integration; raw client URIs +are not storage identities. The contract's record ceiling must fit the configured +replication record limit. + +Commands have an 8 KiB decoding limit. The immutable replication policy bounds +registered tables (default 1,024, maximum 16,384). Reaching the limit rejects new +bindings while preserving retries and takeovers of existing tables. Control +commands are not accepted data records or producer-lane entries. They do not +increase data-page counts or keep their old command bodies in application packs. + +Application images preserve the complete catalog at the applied cut, validate +scope, epochs, grant positions, destination uniqueness and contracts, then replay +the committed suffix. Metadata contains bounded, ordered table frames alongside +the existing data records and producer tails. Payload packs still contain only +opaque accepted data. This changes the experimental replication identity to v2 +and application image format to v2. Earlier v1 application directories are +rejected; this change does not provide an in-place migration or change the +existing production ingestion path. + +`Node::publisher` exposes only a local applied view. It is not a quorum read, +lease, authorization check or proof of Lance visibility. A granted epoch alone +does not stop an old owner from writing outside this protocol. Subsequent work +must check it on durable publication plans, verify source dependencies, commit a +pinned Lance target, resolve its immutable outcome and authorize retirement from +that outcome. Table closing/rebinding, table-bound data admission and integration +with the actual context catalog remain separate requirements. + +Tests cover competing acquisitions, immutable bindings, namespace collisions, +capacity, malformed scopes, schema/size limits and checkpoint validation. A real +three-process TCP test covers the quorum boundary, leader loss, exact grant retry, +snapshot transfer to a lagging replica, isolated-owner retry, takeover and a full +restart with the same grant index. Existing opaque payload/snapshot tests remain +part of the regression suite.