diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 62d614fe..cf0ac64a 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", "contract", "dep:bincode", "dep:slog", "uuid/v4"] +replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog", "uuid/v4", "serde/rc"] raft-log = ["placement", "dep:raft", "dep:raft-engine", "dep:raft-proto", "dep:protobuf"] vectored-io = ["dep:bytes", "dep:tokio"] runner = ["dep:anyhow", "dep:clap", "dep:libc", "dep:serde", "dep:serde_json", "dep:sha2"] diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index fb652721..13dfa7c1 100644 --- a/crates/lance-context-wal/README.md +++ b/crates/lance-context-wal/README.md @@ -176,6 +176,11 @@ publication prerequisite, not a production Lance writer or a manifest lease. catalog, producer-tail and accepted-page views after a safe read-index barrier. They share pending capacity with proposals and do not fetch bulk payloads. +[Immutable publication plans](../../docs/design/wal-publication-plans.md) retain +the exact source vector and Lance target through takeover, snapshots and restart. +They share bounded metadata between views; source pins, outcome resolution and +retirement remain separate integration steps. + 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 da9fc735..bea1a9d8 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -43,6 +43,8 @@ use state::State; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Limits { pub max_tables: u64, + pub max_active_plans: u64, + pub max_plan_source_bytes: u64, pub max_record_bytes: u64, pub max_records: u64, pub max_producers: u64, @@ -53,6 +55,8 @@ impl Default for Limits { fn default() -> Self { Self { max_tables: 1024, + max_active_plans: 128, + max_plan_source_bytes: 64 * 1024 * 1024, max_record_bytes: 8 * 1024 * 1024, max_records: 100_000, max_producers: 16_384, @@ -65,6 +69,8 @@ impl Limits { fn identity(&self) -> Result> { ensure!( (1..=16_384).contains(&self.max_tables) + && (1..=256).contains(&self.max_active_plans) + && (1..=256 * 1024 * 1024).contains(&self.max_plan_source_bytes) && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes) && (1..=500_000).contains(&self.max_records) && (1..=100_000).contains(&self.max_producers) @@ -74,7 +80,7 @@ impl Limits { "invalid replication limits" ); Ok(bincode::serialize(&( - "lance-context/replicated-append/v2", + "lance-context/replicated-append/v3", self, ))?) } @@ -352,6 +358,13 @@ impl Node { self.propose(publication::control_group(&self.layout), request, batch) } + /// Persist an immutable publication plan through the catalog quorum. The + /// plan is not a source pin or permission to commit an external manifest. + pub fn plan_publication(&mut self, request: u64, plan: publication::PlanRequest) -> Result<()> { + let batch = plan.batch(&self.layout, &self.limits)?; + 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()?; diff --git a/crates/lance-context-wal/src/replication/checkpoint.rs b/crates/lance-context-wal/src/replication/checkpoint.rs index 3934c659..bf49ac76 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 = 2; -const MAGIC: &[u8; 8] = b"LCWALCP2"; +const FORMAT: u32 = 3; +const MAGIC: &[u8; 8] = b"LCWALCP3"; pub const CHUNK_BYTES: usize = 1024 * 1024; fn codec() -> impl Options { @@ -82,7 +82,8 @@ 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 - + limits.max_tables * (publication::MAX_BYTES as u64 + 4) + + limits.max_tables * (publication::MAX_BASE_TABLE_BYTES as u64 + 4) + + limits.max_tables.min(limits.max_active_plans) * publication::MAX_COMMAND_BYTES as u64 + 1024 } @@ -371,6 +372,20 @@ fn read_frame(input: &mut impl Read) -> Result<([u8; N], usize)> input.read_exact(&mut bytes[..length])?; Ok((bytes, length)) } +fn read_table_frame(input: &mut impl Read) -> Result> { + let mut length = [0; 4]; + input.read_exact(&mut length)?; + let length = u32::from_le_bytes(length) as usize; + ensure!( + length > 0 && length <= publication::MAX_TABLE_BYTES, + "oversized publisher image frame" + ); + // Plan metadata is larger than a record header. Allocate its actual bounded + // length on the heap rather than putting the maximum frame on the stack. + let mut bytes = vec![0; length]; + input.read_exact(&mut bytes)?; + Ok(bytes) +} fn write_meta(out: &mut impl Write, index: u64, meta: &BatchMeta) -> Result<()> { write_u64(out, index)?; write_frame(out, &state::encode_meta(meta)?) @@ -524,8 +539,8 @@ pub(super) fn load( } 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 bytes = read_table_frame(&mut metadata)?; + let table = publication::decode_table(&bytes)?; let id = table.authority.table; ensure!( previous.is_none_or(|old| old < id), diff --git a/crates/lance-context-wal/src/replication/checkpoint/tests.rs b/crates/lance-context-wal/src/replication/checkpoint/tests.rs index 42c6eb2a..aa912d88 100644 --- a/crates/lance-context-wal/src/replication/checkpoint/tests.rs +++ b/crates/lance-context-wal/src/replication/checkpoint/tests.rs @@ -238,7 +238,7 @@ fn metadata_counts_frames_trailing_bytes_and_exact_producer_tails_are_validated( try_bytes(&count)?; fs::write(&path, original)?; let (state, _) = fixture()?; - for mutation in 0..4 { + for mutation in 0..7 { let mut invalid = state.clone(); match mutation { 0 => invalid.producers.clear(), @@ -325,6 +325,48 @@ fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() -> state::validate_entry(&layout(), 10, &Limits::default(), &entry, true)?; state.apply(&entry, &Limits::default())?; } + let table = &state.catalog.tables[&Uuid::from_u128(200)]; + let mut records = Vec::new(); + for index in 1..=publication::MAX_INPUTS as u64 { + records.push(( + index, + Batch::new( + table.authority.table, + 1, + Uuid::from_u128(230), + index, + Bytes::from_static(b"source"), + ) + .meta, + )); + } + let plan = publication::PlanRequest { + authority: table.authority.clone(), + plan: publication::Plan { + transaction: Uuid::from_u128(231), + read_version: 7, + sources: BTreeMap::from([( + 20, + publication::SourceRange { + after: 0, + through: 1000, + records, + }, + )]), + next_group: 10, + }, + } + .batch(&layout(), &Limits::default())?; + assert!(plan.body.len() > publication::MAX_ACQUIRE_BYTES); + let entry = Entry { + index: state.applied + 1, + term: 2, + context: state::encode_meta(&plan.meta)?.into(), + data: plan.body, + ..Default::default() + }; + state::validate_entry(&layout(), 10, &Limits::default(), &entry, true)?; + state.apply(&entry, &Limits::default())?; let built = build( root.path(), &layout(), @@ -365,14 +407,36 @@ fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() -> 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 { + for mutation in 0..7 { 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, + 3 => tables[1].authority.grant_index = state.applied + 1, + 4 => { + Arc::make_mut(tables[0].progress.pending.as_mut().unwrap()).index = + state.applied + 1 + } + 5 => { + Arc::make_mut(tables[0].progress.pending.as_mut().unwrap()) + .plan + .sources + .get_mut(&20) + .unwrap() + .after = 1 + } + _ => { + Arc::make_mut(tables[0].progress.pending.as_mut().unwrap()) + .plan + .sources + .get_mut(&20) + .unwrap() + .records[0] + .1 + .bucket = 0 + } } for table in &tables { write_frame(&mut bytes, &publication::encode_table(table)?)?; diff --git a/crates/lance-context-wal/src/replication/publication.rs b/crates/lance-context-wal/src/replication/publication.rs index 2f14c871..8987f33e 100644 --- a/crates/lance-context-wal/src/replication/publication.rs +++ b/crates/lance-context-wal/src/replication/publication.rs @@ -1,20 +1,25 @@ //! 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. +//! Immutable publication plans survive takeover. Source pins, manifest outcomes +//! and retirement remain separate work. No Lance manifest is changed here. use super::*; use crate::{ingest_contract::IngestContract, record::RecordKind}; use bincode::Options; use bytes::Bytes; +mod plan; +pub use plan::{PendingPlan, Plan, PlanRequest, Progress, SourceRange, MAX_INPUTS}; #[cfg(test)] mod tests; -pub(super) const MAX_BYTES: usize = 8192; -fn codec() -> impl Options { +pub(super) const MAX_ACQUIRE_BYTES: usize = 8192; +pub(super) const MAX_COMMAND_BYTES: usize = plan::MAX_PLAN_BYTES; +pub(super) const MAX_BASE_TABLE_BYTES: usize = MAX_ACQUIRE_BYTES + 2048; +pub(super) const MAX_TABLE_BYTES: usize = MAX_BASE_TABLE_BYTES + plan::MAX_PLAN_BYTES; +fn codec(limit: usize) -> impl Options { bincode::DefaultOptions::new() .with_fixint_encoding() .with_little_endian() - .with_limit(MAX_BYTES as u64) + .with_limit(limit as u64) .reject_trailing_bytes() } @@ -49,22 +54,41 @@ impl Acquire { } pub(super) fn batch(self, layout: &Layout) -> Result { self.validate()?; + Command::Acquire(Box::new(self)).batch(layout) + } +} + +#[derive(Serialize, Deserialize)] +enum Command { + Acquire(Box), + Plan(Box), +} +impl Command { + fn identity(&self) -> (Uuid, Uuid, u64) { + match self { + Self::Acquire(a) => (a.table, a.owner, a.expected_epoch + 1), + Self::Plan(p) => (p.authority.table, p.authority.owner, p.authority.epoch), + } + } + fn byte_limit(&self) -> usize { + match self { + Self::Acquire(_) => MAX_ACQUIRE_BYTES, + Self::Plan(_) => MAX_COMMAND_BYTES, + } + } + fn batch(self, layout: &Layout) -> Result { + let (table, owner, sequence) = self.identity(); + let limit = self.byte_limit(); let envelope = Envelope { - version: 1, + version: 2, cluster: layout.cluster, layout: layout.fingerprint()?, group: control_group(layout), - acquire: self, + command: 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), - ); + let bytes = codec(limit).serialize(&envelope)?; + ensure!(bytes.len() <= limit, "publisher command too large"); + let mut batch = Batch::new(table, 0, owner, sequence, Bytes::from(bytes)); batch.meta.kind = RecordKind::MultiPublication; Ok(batch) } @@ -87,6 +111,7 @@ pub struct Table { pub ingest: IngestContract, pub authority: Authority, pub request: Uuid, + pub progress: Progress, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -96,10 +121,14 @@ pub enum Rejection { Epoch { current: u64 }, RequestReused, Capacity, + Authority, + PendingPlan, + SourcePosition, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Reply { Acquired(Authority), + Planned { index: u64 }, Rejected(Rejection), } @@ -109,7 +138,7 @@ pub(super) struct Envelope { cluster: Uuid, layout: [u8; 32], group: u64, - acquire: Acquire, + command: Command, } pub(super) fn control_group(layout: &Layout) -> u64 { *layout @@ -129,10 +158,23 @@ fn validate_binding(destination: &Destination, ingest: &IngestContract) -> Resul 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()?; + ensure!( + bytes.len() <= MAX_COMMAND_BYTES, + "publisher command too large" + ); + let envelope: Envelope = codec(MAX_COMMAND_BYTES).deserialize(bytes)?; + ensure!(envelope.version == 2, "unsupported publisher command"); + ensure!( + bytes.len() <= envelope.command.byte_limit(), + "publisher command type byte limit" + ); + if let Command::Acquire(a) = &envelope.command { + a.validate()?; + } + ensure!( + codec(MAX_COMMAND_BYTES).serialize(&envelope)? == bytes, + "noncanonical publisher command" + ); Ok(envelope) } pub(super) fn validate_record( @@ -143,19 +185,27 @@ pub(super) fn validate_record( body: &[u8], ) -> Result<()> { let e = decode(body)?; + let (table, owner, sequence) = e.command.identity(); 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.stream == 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, + && meta.producer == owner + && meta.sequence == sequence, "publisher scope/envelope or ingestion limit mismatch" ); + match e.command { + Command::Acquire(a) => ensure!( + u64::from(a.ingest.max_record_bytes) <= limits.max_record_bytes + && u64::from(a.ingest.max_record_bytes) <= limits.max_plan_source_bytes, + "unserviceable ingestion limit" + ), + Command::Plan(p) => p.validate(layout, limits)?, + } Ok(()) } @@ -164,8 +214,11 @@ 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; + pub fn apply(&mut self, e: Envelope, index: u64, limits: &Limits) -> Reply { + let a = match e.command { + Command::Plan(p) => return self.plan(*p, index, limits), + Command::Acquire(a) => *a, + }; let old = self.tables.get(&a.table); if let Some(old) = old { if old.destination != a.destination || old.ingest != a.ingest { @@ -189,7 +242,7 @@ impl Catalog { if self.tables.values().any(|t| t.destination == a.destination) { return Reply::Rejected(Rejection::DestinationBound); } - if self.tables.len() as u64 >= max_tables { + if self.tables.len() as u64 >= limits.max_tables { return Reply::Rejected(Rejection::Capacity); } } @@ -202,6 +255,7 @@ impl Catalog { epoch: current + 1, grant_index: index, }; + let progress = old.map(|table| table.progress.clone()).unwrap_or_default(); self.tables.insert( a.table, Table { @@ -209,6 +263,7 @@ impl Catalog { ingest: a.ingest, authority: authority.clone(), request: a.request, + progress, }, ); Reply::Acquired(authority) @@ -218,6 +273,14 @@ impl Catalog { self.tables.len() as u64 <= limits.max_tables, "publisher catalog exceeds limit" ); + ensure!( + self.tables + .values() + .filter(|table| table.progress.pending.is_some()) + .count() as u64 + <= limits.max_active_plans, + "active publication plan limit" + ); ensure!( group == control_group(layout) || self.tables.is_empty(), "catalog in source group" @@ -226,6 +289,7 @@ impl Catalog { let mut destinations = std::collections::BTreeSet::new(); for (id, table) in &self.tables { validate_binding(&table.destination, &table.ingest)?; + table.progress.validate(layout, table, cut, limits)?; let a = &table.authority; ensure!( !id.is_nil() @@ -239,6 +303,7 @@ impl Catalog { && a.epoch <= a.grant_index && a.grant_index <= cut && u64::from(table.ingest.max_record_bytes) <= limits.max_record_bytes + && u64::from(table.ingest.max_record_bytes) <= limits.max_plan_source_bytes && destinations.insert(&table.destination), "invalid publisher checkpoint state" ); @@ -247,8 +312,14 @@ impl Catalog { } } pub(super) fn encode_table(table: &Table) -> Result> { - Ok(codec().serialize(table)?) + Ok(codec(MAX_TABLE_BYTES).serialize(table)?) } pub(super) fn decode_table(bytes: &[u8]) -> Result { - Ok(codec().deserialize(bytes)?) + ensure!(bytes.len() <= MAX_TABLE_BYTES, "publisher table byte limit"); + let table = codec(MAX_TABLE_BYTES).deserialize(bytes)?; + ensure!( + encode_table(&table)? == bytes, + "noncanonical publisher table" + ); + Ok(table) } diff --git a/crates/lance-context-wal/src/replication/publication/plan.rs b/crates/lance-context-wal/src/replication/publication/plan.rs new file mode 100644 index 00000000..dfac4f67 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/plan.rs @@ -0,0 +1,237 @@ +//! Immutable selection of source records and a fixed Lance transaction target. +//! Source groups must subsequently verify their complete prefixes before pinning. +use super::*; + +#[cfg(test)] +mod tests; + +pub const MAX_INPUTS: usize = 512; +pub(super) const MAX_PLAN_BYTES: usize = 128 * 1024; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SourceRange { + pub after: u64, + pub through: u64, + pub records: Vec<(u64, BatchMeta)>, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Plan { + pub transaction: Uuid, + pub read_version: u64, + pub sources: BTreeMap, + /// Resume fair source selection from this group after publication. + pub next_group: u64, +} +impl Plan { + pub fn target_version(&self) -> Result { + ensure!( + self.read_version > 0, + "publication requires an existing dataset" + ); + self.read_version + .checked_add(1) + .context("Lance version exhausted") + } + + pub fn source_bytes(&self) -> Result { + self.sources + .values() + .flat_map(|range| &range.records) + .try_fold(0u64, |sum, (_, meta)| { + sum.checked_add(meta.body_bytes) + .context("source byte sum overflow") + }) + } + + pub(super) fn validate(&self, layout: &Layout, table: Uuid, limits: &Limits) -> Result<()> { + self.target_version()?; + ensure!( + !self.transaction.is_nil() + && !table.is_nil() + && !self.sources.is_empty() + && self.sources.len() <= layout.groups.len() + && layout.groups.contains_key(&self.next_group), + "invalid publication plan identity or group selection" + ); + let mut count = 0usize; + for (group, range) in &self.sources { + count = count + .checked_add(range.records.len()) + .context("plan input count overflow")?; + ensure!( + count <= MAX_INPUTS + && layout.groups.contains_key(group) + && range.after < range.through + && range.through < u64::MAX + && !range.records.is_empty() + && range.records.windows(2).all(|w| w[0].0 < w[1].0), + "invalid publication source range" + ); + for (index, meta) in &range.records { + ensure!( + *index > range.after + && *index <= range.through + && meta.stream == table + && meta.kind == RecordKind::Data + && !meta.producer.is_nil() + && meta.sequence > 0 + && meta.sequence < u64::MAX + && meta.body_bytes > 0 + && meta.body_bytes <= limits.max_record_bytes + && layout.group_for_bucket(meta.bucket)? == *group, + "invalid publication source metadata" + ); + } + } + ensure!( + self.source_bytes()? <= limits.max_plan_source_bytes, + "publication source byte limit" + ); + Ok(()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PlanRequest { + pub authority: Authority, + pub plan: Plan, +} +impl PlanRequest { + pub(in crate::replication) fn batch(self, layout: &Layout, limits: &Limits) -> Result { + self.validate(layout, limits)?; + Command::Plan(Box::new(self)).batch(layout) + } + pub(super) fn validate(&self, layout: &Layout, limits: &Limits) -> Result<()> { + let a = &self.authority; + ensure!( + a.cluster == layout.cluster + && a.layout == layout.fingerprint()? + && a.group == control_group(layout) + && !a.owner.is_nil() + && a.epoch > 0 + && a.epoch <= a.grant_index + && a.grant_index < u64::MAX, + "invalid publication authority scope" + ); + self.plan.validate(layout, a.table, limits) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PendingPlan { + pub index: u64, + pub plan: Plan, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Progress { + /// Independent source-group positions; never collapse these to one maximum. + pub checkpoint: BTreeMap, + pub minimum_read_version: u64, + /// Cloning a table view shares the immutable metadata; it never clones bodies. + pub pending: Option>, +} +impl Default for Progress { + fn default() -> Self { + Self { + checkpoint: BTreeMap::new(), + minimum_read_version: 1, + pending: None, + } + } +} +impl Progress { + pub(super) fn matches(&self, plan: &Plan, ingest: &IngestContract) -> bool { + plan.read_version >= self.minimum_read_version + && plan.sources.iter().all(|(group, range)| { + range.after == self.checkpoint.get(group).copied().unwrap_or(0) + && range + .records + .iter() + .all(|(_, meta)| meta.body_bytes <= u64::from(ingest.max_record_bytes)) + }) + } + + pub(super) fn validate( + &self, + layout: &Layout, + table: &Table, + cut: u64, + limits: &Limits, + ) -> Result<()> { + ensure!( + self.minimum_read_version > 0 + && self.checkpoint.len() <= layout.groups.len() + && self.checkpoint.iter().all(|(group, position)| { + layout.groups.contains_key(group) && *position > 0 && *position < u64::MAX + }), + "invalid publication progress" + ); + if let Some(pending) = &self.pending { + pending + .plan + .validate(layout, table.authority.table, limits)?; + ensure!( + pending.index > 0 + && pending.index <= cut + && pending.index != table.authority.grant_index + && pending + .plan + .sources + .get(&table.authority.group) + .is_none_or(|range| range.through < pending.index) + && self.matches(&pending.plan, &table.ingest), + "invalid pending publication plan" + ); + } + Ok(()) + } +} + +impl Catalog { + pub(super) fn plan(&mut self, request: PlanRequest, index: u64, limits: &Limits) -> Reply { + let Some(table) = self.tables.get(&request.authority.table) else { + return Reply::Rejected(Rejection::Authority); + }; + if table.authority != request.authority { + return Reply::Rejected(Rejection::Authority); + } + if let Some(pending) = &table.progress.pending { + return if pending.plan == request.plan { + Reply::Planned { + index: pending.index, + } + } else { + Reply::Rejected(Rejection::PendingPlan) + }; + } + if !table.progress.matches(&request.plan, &table.ingest) + || request + .plan + .sources + .get(&table.authority.group) + .is_some_and(|range| range.through >= index) + { + return Reply::Rejected(Rejection::SourcePosition); + } + if self + .tables + .values() + .filter(|t| t.progress.pending.is_some()) + .count() as u64 + >= limits.max_active_plans + { + return Reply::Rejected(Rejection::Capacity); + } + self.tables + .get_mut(&request.authority.table) + .expect("known table") + .progress + .pending = Some(Arc::new(PendingPlan { + index, + plan: request.plan, + })); + Reply::Planned { index } + } +} diff --git a/crates/lance-context-wal/src/replication/publication/plan/tests.rs b/crates/lance-context-wal/src/replication/publication/plan/tests.rs new file mode 100644 index 00000000..44407a4d --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/plan/tests.rs @@ -0,0 +1,360 @@ +use super::*; + +fn layout() -> Arc { + crate::replication::tests::layout() +} +fn acquire() -> Acquire { + super::super::tests::input() +} +fn catalog() -> Result<(Catalog, PlanRequest)> { + let mut catalog = Catalog::default(); + let a = acquire(); + let batch = a.clone().batch(&layout())?; + let Reply::Acquired(authority) = catalog.apply(decode(&batch.body)?, 4, &Limits::default()) + else { + anyhow::bail!("missing authority"); + }; + let meta = Batch::new( + a.table, + 1, + Uuid::from_u128(90), + 1, + Bytes::from_static(b"record"), + ) + .meta; + let request = PlanRequest { + authority, + plan: Plan { + transaction: Uuid::from_u128(91), + read_version: 7, + sources: BTreeMap::from([( + 20, + SourceRange { + after: 0, + through: 1000, + records: vec![(999, meta)], + }, + )]), + next_group: 10, + }, + }; + Ok((catalog, request)) +} +fn apply( + catalog: &mut Catalog, + request: PlanRequest, + index: u64, + limits: &Limits, +) -> Result { + let batch = request.batch(&layout(), limits)?; + validate_record(&layout(), 10, limits, &batch.meta, &batch.body)?; + Ok(catalog.apply(decode(&batch.body)?, index, limits)) +} + +#[test] +fn exact_plan_and_shared_view_survive_takeover_without_releasing_capacity() -> Result<()> { + let (mut catalog, mut request) = catalog()?; + let limits = Limits { + max_active_plans: 1, + ..Default::default() + }; + assert_eq!( + apply(&mut catalog, request.clone(), 5, &limits)?, + Reply::Planned { index: 5 } + ); + assert_eq!( + apply(&mut catalog, request.clone(), 6, &limits)?, + Reply::Planned { index: 5 } + ); + let captured = catalog.tables[&request.authority.table].clone(); + let pending = captured.progress.pending.as_ref().unwrap(); + assert_eq!(pending.plan.target_version()?, 8); + assert!(Arc::ptr_eq( + pending, + catalog.tables[&request.authority.table] + .progress + .pending + .as_ref() + .unwrap() + )); + let mut changed = request.clone(); + changed.plan.transaction = Uuid::from_u128(92); + assert_eq!( + apply(&mut catalog, changed, 7, &limits)?, + Reply::Rejected(Rejection::PendingPlan) + ); + + let mut a = acquire(); + a.owner = Uuid::from_u128(93); + a.request = Uuid::from_u128(94); + a.expected_epoch = 1; + let batch = a.batch(&layout())?; + let Reply::Acquired(next) = catalog.apply(decode(&batch.body)?, 8, &limits) else { + anyhow::bail!("takeover blocked by plan capacity"); + }; + assert_eq!( + apply(&mut catalog, request.clone(), 9, &limits)?, + Reply::Rejected(Rejection::Authority) + ); + request.authority = next; + assert_eq!( + apply(&mut catalog, request.clone(), 10, &limits)?, + Reply::Planned { index: 5 } + ); + assert!(Arc::ptr_eq( + pending, + catalog.tables[&request.authority.table] + .progress + .pending + .as_ref() + .unwrap() + )); + assert_eq!(captured.authority.epoch, 1); + catalog.validate(&layout(), 10, 10, &limits)?; + + let mut other = acquire(); + other.table = Uuid::from_u128(95); + other.destination.dataset = "another/table".into(); + let batch = other.batch(&layout())?; + let Reply::Acquired(other) = catalog.apply(decode(&batch.body)?, 11, &limits) else { + anyhow::bail!("another table could not bind"); + }; + request.authority = other.clone(); + request.plan.sources.get_mut(&20).unwrap().records[0] + .1 + .stream = other.table; + assert_eq!( + apply(&mut catalog, request, 12, &limits)?, + Reply::Rejected(Rejection::Capacity) + ); + assert_eq!( + catalog.tables[&captured.authority.table] + .progress + .checkpoint + .len(), + 0 + ); + assert_eq!( + catalog.tables[&captured.authority.table] + .progress + .minimum_read_version, + 1 + ); + Ok(()) +} + +#[test] +fn source_positions_are_per_group_and_plans_cannot_skip_publication_checkpoints() -> Result<()> { + let (mut catalog, mut request) = catalog()?; + let limits = Limits::default(); + request.plan.sources.get_mut(&20).unwrap().after = 998; + assert_eq!( + apply(&mut catalog, request.clone(), 5, &limits)?, + Reply::Rejected(Rejection::SourcePosition) + ); + request.plan.sources.get_mut(&20).unwrap().after = 0; + let mut local = request.clone(); + let mut range = local.plan.sources.remove(&20).unwrap(); + range.records[0].1.bucket = 0; + local.plan.sources.insert(10, range); + assert_eq!( + apply(&mut catalog, local, 6, &limits)?, + Reply::Rejected(Rejection::SourcePosition) + ); + // The same numeric position is valid in an independent source group. A + // catalog index must never be used as a global source high-water mark. + assert_eq!( + apply(&mut catalog, request, 7, &limits)?, + Reply::Planned { index: 7 } + ); + catalog.validate(&layout(), 10, 7, &limits)?; + Ok(()) +} + +#[test] +fn plan_bounds_reject_bad_sources_before_admission_and_allow_large_metadata() -> Result<()> { + let (mut catalog, request) = catalog()?; + let limits = Limits::default(); + for mutate in [ + (|r: &mut PlanRequest| r.plan.transaction = Uuid::nil()) as fn(&mut PlanRequest), + |r| r.plan.read_version = 0, + |r| r.plan.read_version = u64::MAX, + |r| r.plan.next_group = 99, + |r| r.authority.layout[0] ^= 1, + |r| r.plan.sources.get_mut(&20).unwrap().through = u64::MAX, + |r| r.plan.sources.get_mut(&20).unwrap().records[0].0 = 0, + |r| r.plan.sources.get_mut(&20).unwrap().records[0].1.bucket = 0, + |r| r.plan.sources.get_mut(&20).unwrap().records[0].1.stream = Uuid::from_u128(100), + |r| r.plan.sources.get_mut(&20).unwrap().records[0].1.sequence = 0, + |r| r.plan.sources.get_mut(&20).unwrap().records[0].1.body_bytes = 0, + |r| { + let range = r.plan.sources.get_mut(&20).unwrap(); + range.records.push(range.records[0].clone()); + }, + ] { + let mut invalid = request.clone(); + mutate(&mut invalid); + assert!(invalid.clone().batch(&layout(), &limits).is_err()); + // A hand-crafted wire request must also fail validation at the replica. + let batch = Command::Plan(Box::new(invalid)).batch(&layout())?; + assert!(validate_record(&layout(), 10, &limits, &batch.meta, &batch.body).is_err()); + } + let tiny = Limits { + max_plan_source_bytes: 5, + ..limits.clone() + }; + assert!(request.clone().batch(&layout(), &tiny).is_err()); + let mut aggregate = request.clone(); + let range = aggregate.plan.sources.get_mut(&20).unwrap(); + range.records[0].1.body_bytes = 4; + range.records.push(( + 1000, + BatchMeta { + sequence: 2, + ..range.records[0].1.clone() + }, + )); + assert!(aggregate + .batch( + &layout(), + &Limits { + max_plan_source_bytes: 6, + ..limits.clone() + } + ) + .is_err()); + let mut large = request; + let range = large.plan.sources.get_mut(&20).unwrap(); + let meta = range.records[0].1.clone(); + range.records = (1..=MAX_INPUTS as u64) + .map(|index| { + ( + index, + BatchMeta { + sequence: index, + ..meta.clone() + }, + ) + }) + .collect(); + let batch = large.clone().batch(&layout(), &limits)?; + assert!(batch.body.len() > MAX_ACQUIRE_BYTES); + assert_eq!( + apply(&mut catalog, large.clone(), 5, &limits)?, + Reply::Planned { index: 5 } + ); + let table = &catalog.tables[&large.authority.table]; + assert_eq!(decode_table(&encode_table(table)?)?, *table); + large + .plan + .sources + .get_mut(&20) + .unwrap() + .records + .push((513, meta)); + assert!(large.batch(&layout(), &limits).is_err()); + Ok(()) +} + +#[test] +fn pending_plan_checkpoint_validation_checks_inner_semantics() -> Result<()> { + let (mut catalog, request) = catalog()?; + let limits = Limits::default(); + apply(&mut catalog, request.clone(), 5, &limits)?; + let saved = catalog.tables[&request.authority.table].clone(); + for mutate in [ + (|p: &mut PendingPlan| p.index = 0) as fn(&mut PendingPlan), + |p| p.index = 4, + |p| p.index = 6, + |p| p.plan.read_version = u64::MAX, + |p| p.plan.sources.get_mut(&20).unwrap().after = 1, + |p| p.plan.sources.get_mut(&20).unwrap().records[0].1.body_bytes = 2 * 1024 * 1024, + ] { + let mut table = saved.clone(); + mutate(Arc::make_mut(table.progress.pending.as_mut().unwrap())); + catalog.tables.insert(request.authority.table, table); + assert!(catalog.validate(&layout(), 10, 5, &limits).is_err()); + } + assert_eq!(saved.progress.pending.as_ref().unwrap().index, 5); + catalog + .tables + .insert(request.authority.table, saved.clone()); + let mut other = saved; + other.authority.table = Uuid::from_u128(101); + other.destination.dataset = "other/table".into(); + Arc::make_mut(other.progress.pending.as_mut().unwrap()) + .plan + .sources + .get_mut(&20) + .unwrap() + .records[0] + .1 + .stream = other.authority.table; + catalog.tables.insert(other.authority.table, other); + catalog.validate(&layout(), 10, 5, &limits)?; + assert!(catalog + .validate( + &layout(), + 10, + 5, + &Limits { + max_active_plans: 1, + ..limits + } + ) + .is_err()); + Ok(()) +} + +#[test] +fn wire_and_checkpoint_maps_cannot_hide_duplicate_source_groups() -> Result<()> { + #[derive(Serialize)] + struct WirePlan<'a> { + transaction: Uuid, + read_version: u64, + sources: Vec<(u64, &'a SourceRange)>, + next_group: u64, + } + fn replace_once(body: &[u8], old: &[u8], new: &[u8]) -> Vec { + let positions: Vec<_> = body + .windows(old.len()) + .enumerate() + .filter_map(|(i, bytes)| (bytes == old).then_some(i)) + .collect(); + assert_eq!(positions.len(), 1); + let start = positions[0]; + [ + body[..start].to_vec(), + new.to_vec(), + body[start + old.len()..].to_vec(), + ] + .concat() + } + let (mut catalog, request) = catalog()?; + let limits = Limits::default(); + let original = codec(MAX_COMMAND_BYTES).serialize(&request.plan)?; + let range = &request.plan.sources[&20]; + let duplicate = codec(MAX_COMMAND_BYTES).serialize(&WirePlan { + transaction: request.plan.transaction, + read_version: request.plan.read_version, + sources: vec![(20, range), (20, range)], + next_group: request.plan.next_group, + })?; + let batch = request.clone().batch(&layout(), &limits)?; + let bad = replace_once(&batch.body, &original, &duplicate); + let decoded: Envelope = codec(MAX_COMMAND_BYTES).deserialize(&bad)?; + let Command::Plan(normalized) = decoded.command else { + anyhow::bail!("wrong command") + }; + assert_eq!( + *normalized, request, + "bincode did not normalize the malformed map" + ); + assert!(decode(&bad).is_err()); + apply(&mut catalog, request.clone(), 5, &limits)?; + let table = &catalog.tables[&request.authority.table]; + let bad = replace_once(&encode_table(table)?, &original, &duplicate); + assert_eq!(codec(MAX_TABLE_BYTES).deserialize::
(&bad)?, *table); + assert!(decode_table(&bad).is_err()); + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/publication/tests.rs b/crates/lance-context-wal/src/replication/publication/tests.rs index e459b555..4188131b 100644 --- a/crates/lance-context-wal/src/replication/publication/tests.rs +++ b/crates/lance-context-wal/src/replication/publication/tests.rs @@ -1,7 +1,7 @@ use super::*; use crate::ingest_contract::WritePolicy; -fn input() -> Acquire { +pub(super) fn input() -> Acquire { Acquire { table: Uuid::from_u128(10), destination: Destination { @@ -23,7 +23,14 @@ fn input() -> Acquire { } 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)) + Ok(catalog.apply( + decode(&batch.body)?, + index, + &Limits { + max_tables: capacity, + ..Default::default() + }, + )) } #[test] @@ -93,7 +100,7 @@ fn bounded_decode_rejects_wrong_scopes_domains_and_unserviceable_ingest_limits() let mut trailing = batch.body.to_vec(); trailing.push(0); assert!(decode(&trailing).is_err()); - assert!(decode(&vec![0; MAX_BYTES + 1]).is_err()); + assert!(decode(&vec![0; MAX_COMMAND_BYTES + 1]).is_err()); let mut oversized = input(); oversized.destination.dataset = "x".repeat(4097); assert!(oversized.batch(&layout).is_err()); @@ -143,6 +150,36 @@ fn checkpoint_catalog_requires_exact_scope_ordering_identity_and_unique_destinat Ok(()) } +#[test] +fn registered_records_must_fit_the_publication_source_budget() -> Result<()> { + let layout = super::super::tests::layout(); + let limits = Limits { + max_plan_source_bytes: 512 * 1024, + ..Default::default() + }; + limits.identity()?; // A small publication budget is valid for smaller contracts. + let mut acquire = input(); + let batch = acquire.clone().batch(&layout)?; + assert!( + validate_record(&layout, 10, &limits, &batch.meta, &batch.body).is_err(), + "registered contract can admit a record no publication plan can select" + ); + let mut catalog = Catalog::default(); + apply(&mut catalog, acquire.clone(), 4, 1)?; + assert!( + catalog.validate(&layout, 10, 4, &limits).is_err(), + "checkpoint restored an unserviceable ingestion contract" + ); + + acquire.ingest.max_record_bytes = limits.max_plan_source_bytes as u32; + let batch = acquire.clone().batch(&layout)?; + validate_record(&layout, 10, &limits, &batch.meta, &batch.body)?; + let mut catalog = Catalog::default(); + apply(&mut catalog, acquire, 4, 1)?; + catalog.validate(&layout, 10, 4, &limits)?; + Ok(()) +} + #[test] fn failed_follower_sync_cannot_release_a_publisher_grant() -> Result<()> { let root = tempfile::tempdir()?; diff --git a/crates/lance-context-wal/src/replication/state.rs b/crates/lance-context-wal/src/replication/state.rs index 08140ddb..8d9d9c86 100644 --- a/crates/lance-context-wal/src/replication/state.rs +++ b/crates/lance-context-wal/src/replication/state.rs @@ -26,7 +26,7 @@ pub(super) fn validate_meta( digest: bool, ) -> Result<()> { if meta.kind == RecordKind::MultiPublication { - meta.validate_payload(body, publication::MAX_BYTES)?; + meta.validate_payload(body, publication::MAX_COMMAND_BYTES)?; return publication::validate_record(layout, group, limits, meta, body); } validate_identity(layout, group, meta)?; @@ -169,7 +169,7 @@ impl State { return Ok(Some(Applied::Publisher(self.catalog.apply( publication::decode(&entry.data)?, entry.index, - limits.max_tables, + limits, )))); } let key = (meta.stream, meta.bucket, meta.producer); diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index c6a28df5..705d1215 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -9,7 +9,8 @@ use lance_context_wal::{ replication::{ checkpoint::{Image, Part}, publication::{ - Acquire, Destination, Rejection as PublisherRejection, Reply as PublisherReply, Table, + Acquire, Destination, Plan, PlanRequest, Rejection as PublisherRejection, + Reply as PublisherReply, SourceRange, Table, }, query::{Completion as ReadCompletion, Outcome as ReadOutcome, Query, Value}, Applied, Completion, Events, Limits, Node, Outcome, Packet, Page, PreparedCheckpoint, @@ -61,6 +62,7 @@ impl WirePacket { #[derive(Serialize, Deserialize)] enum Request { Query(u64, u64, Query), + Plan(u64, PlanRequest), Acquire(u64, Acquire), Publisher(Uuid), Campaign(u64), @@ -126,6 +128,7 @@ fn run(worker: &mut Worker, request: Request) -> Result { let node = &mut worker.node; match request { Request::Query(group, token, query) => node.query(group, token, query)?, + Request::Plan(token, plan) => node.plan_publication(token, plan)?, Request::Acquire(token, acquire) => node.acquire_publisher(token, acquire)?, Request::Publisher(table) => { return Ok(Response::Publisher(node.publisher(table)?.cloned())) @@ -375,11 +378,19 @@ impl Cluster { Ok(()) } fn command(&mut self, id: u64, request: Request) -> Result<()> { + let context = match &request { + Request::Propose(group, token, _) => format!("propose group {group}, token {token}"), + Request::Query(group, token, _) => format!("query group {group}, token {token}"), + Request::Acquire(token, _) => format!("acquire token {token}"), + Request::Plan(token, _) => format!("plan token {token}"), + _ => "replication control".to_string(), + }; let Response::Events(packets, completions, reads) = self .peers .get_mut(&id) .context("missing peer")? - .call(request)? + .call(request) + .with_context(|| format!("peer {id}: {context}"))? else { anyhow::bail!("wrong reply"); }; @@ -836,6 +847,50 @@ fn publisher_epochs_require_quorum_and_survive_snapshot_transfer_and_restart() - assert!( matches!(&second, Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(a))) if a.epoch == 2) ); + let Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(authority))) = &second + else { + anyhow::bail!("missing takeover authority"); + }; + // Catalog election ticks all hosted groups. The source group may already + // have another leader; a campaign request alone does not transfer its lease. + let source_leader = c.elect(2, 20, &[2, 3])?; + let mut source = input(1, 1, 99); + source.meta.stream = acquire.table; + c.command(source_leader, Request::Propose(20, 710, source.clone()))?; + c.pump(&[2, 3])?; + let Some(Outcome::Committed(Applied::Accepted { + index: source_index, + })) = c.outcome(710) + else { + anyhow::bail!("source did not commit"); + }; + let mut plan = PlanRequest { + authority: authority.clone(), + plan: Plan { + transaction: Uuid::from_u128(720), + read_version: 7, + sources: BTreeMap::from([( + 20, + SourceRange { + after: 0, + through: source_index, + records: vec![(source_index, source.meta.clone())], + }, + )]), + next_group: 10, + }, + }; + c.command(2, Request::Plan(711, plan.clone()))?; + assert!( + c.outcome(711).is_none(), + "local plan completed without quorum" + ); + c.pump(&[2, 3])?; + let Some(Outcome::Committed(Applied::Publisher(PublisherReply::Planned { index: plan_index }))) = + c.outcome(711) + else { + anyhow::bail!("plan did not commit"); + }; let image = c .peers .get_mut(&2) @@ -870,6 +925,21 @@ fn publisher_epochs_require_quorum_and_survive_snapshot_transfer_and_restart() - assert_eq!(table.authority.epoch, 2); assert_eq!(table.destination, acquire.destination); assert_eq!(table.ingest, acquire.ingest); + let saved = table + .progress + .pending + .as_ref() + .context("snapshot lost plan")?; + assert_eq!(saved.index, plan_index); + assert_eq!(saved.plan, plan.plan); + assert_eq!(saved.plan.target_version()?, 8); + + c.command(2, Request::Plan(712, plan.clone()))?; + c.pump(&[2])?; + assert!( + c.outcome(712).is_none(), + "isolated leader answered a plan retry from cache" + ); c.command( 2, @@ -910,6 +980,19 @@ fn publisher_epochs_require_quorum_and_survive_snapshot_transfer_and_restart() - assert!( matches!(&latest, Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(a))) if a.epoch == 3) ); + c.command(leader, Request::Plan(713, plan.clone()))?; + c.pump(&[1, 3])?; + assert_eq!( + c.outcome(713), + Some(Outcome::Committed(Applied::Publisher( + PublisherReply::Rejected(PublisherRejection::Authority) + ))) + ); + let Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(authority))) = &latest + else { + anyhow::bail!("missing third authority"); + }; + plan.authority = authority.clone(); c.command(leader, Request::Acquire(705, stale))?; c.pump(&[1, 3])?; assert_eq!( @@ -934,7 +1017,15 @@ fn publisher_epochs_require_quorum_and_survive_snapshot_transfer_and_restart() - assert_eq!(table.authority.epoch, 3); assert_eq!(table.authority.owner, acquire.owner); assert_eq!(table.ingest, acquire.ingest); + let recovered = table + .progress + .pending + .as_ref() + .context("restart lost plan")?; + assert_eq!(recovered.index, plan_index); + assert_eq!(recovered.plan, plan.plan); verify_records(peer, 10, &[])?; + verify_records(peer, 20, std::slice::from_ref(&source))?; } let leader = c.elect(1, 10, &[1, 2, 3])?; c.command(leader, Request::Acquire(706, acquire))?; @@ -944,6 +1035,14 @@ fn publisher_epochs_require_quorum_and_survive_snapshot_transfer_and_restart() - Some(latest), "restart changed the committed grant index" ); + c.command(leader, Request::Plan(714, plan))?; + c.pump(&[1, 2, 3])?; + assert_eq!( + c.outcome(714), + Some(Outcome::Committed(Applied::Publisher( + PublisherReply::Planned { index: plan_index } + ))) + ); assert_eq!( captured.authority.epoch, 2, "later takeover mutated a captured read" diff --git a/docs/design/wal-publication-plans.md b/docs/design/wal-publication-plans.md new file mode 100644 index 00000000..3a8b37e3 --- /dev/null +++ b/docs/design/wal-publication-plans.md @@ -0,0 +1,57 @@ +# Durable publication plans + +Refs #378. + +A publisher takeover must recover the same transaction and source selection that +the previous owner was preparing. The optional replicated WAL now accepts an +immutable `PlanRequest` through its catalog group. Only a durable quorum completion +returns the plan's original log index. Exact retries use consensus; an isolated +leader cannot answer them from its local catalog. + +A plan contains a transaction UUID, an existing Lance read version, the fixed +target `read_version + 1`, and an ordered vector of source groups. Each source has +its own `(after, through]` range and selected record metadata. Independent group +positions are never compared as a global index. The catalog checks the current +publisher authority at apply, the table's per-group published cursors and the +ingestion record-size limit. A source range in the catalog group must precede the +plan entry itself. Other groups may have larger numeric positions. + +Takeover changes authority while retaining the immutable pending plan and progress. +The new owner can retry that exact plan and recover its original index. Replacing +it with a different transaction, source selection or target is rejected. One plan +is retained per table; the configured global active-plan limit does not prevent +an existing plan's retry or an ownership takeover. Cloned table/checkpoint views +share the immutable plan with `Arc`; payload bodies do not live in the catalog. + +The bounds are separate: acquisition commands retain their 8 KiB limit, plan +commands have a 128 KiB limit and at most 512 selected records, and the policy +bounds their aggregate encoded source bytes before payload fetch. Defaults allow +128 active plans and 64 MiB of selected source bytes per plan. These limits bound +metadata and selected input. Each registered ingestion contract's maximum record +must fit the publication source-byte budget so an accepted record can be selected. +The budget may be smaller than the global WAL record ceiling for smaller contracts. +These limits do not reserve encoder memory or transport +queues. Checkpoint metadata capacity accounts separately for table bindings and +active plans. Large metadata frames allocate only their bounded actual length on +the heap. Decoders reject noncanonical maps instead of silently accepting duplicate +keys normalized by the serializer. + +This changes the optional application identity and checkpoint format to v3. +Existing v1/v2 directories and images are rejected; there is no in-place migration +or production rollout in this increment. Catalog-only and mixed data/catalog +snapshots preserve plans, with semantic validation after checksum verification. + +A stored plan is not a source pin, proof of complete source selection or permission +to commit a Lance manifest. Source groups must still validate their full selected +prefixes through the subsequent pin protocol. Descriptor CAS, exact-target Lance +outcome recovery, resolution and cleanup commands remain subsequent work; this +increment intentionally retains pending plans and advances no published cursor. +No source payload can be retired on the strength of this plan alone. + +Validation includes real TCP replicas that persist a source append and a catalog +plan, lose leaders, transfer a catalog-only snapshot, take over ownership and +restart every process. The restarted owner retries the original plan index and +the source payload remains exact. Mixed snapshots exercise plans larger than the +old acquisition limit and reject corrupt inner plan state even when outer digests +are recomputed. Tests also cover shared view ownership, metadata/source-byte and +active-plan bounds, malformed maps and stale publisher authority. diff --git a/docs/design/wal-publisher-authority.md b/docs/design/wal-publisher-authority.md index f605c091..2f34f5e7 100644 --- a/docs/design/wal-publisher-authority.md +++ b/docs/design/wal-publisher-authority.md @@ -25,7 +25,7 @@ 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 +Acquisition commands have an 8 KiB size 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 @@ -35,15 +35,15 @@ 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. +opaque accepted data. The current replication identity and application image +format are v3, including [immutable publication plans](wal-publication-plans.md). +Earlier v1/v2 application directories are rejected; this optional component does +not provide an in-place migration or change the 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 +does not stop an old owner from writing outside this protocol. Durable publication +plans check the current authority at apply. Subsequent work must 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.