From 4d939bff3eb150d3ef2fd7708d20f63dc50d5c92 Mon Sep 17 00:00:00 2001 From: Beinan Date: Tue, 20 Oct 2026 18:28:06 +0000 Subject: [PATCH] feat(wal): close source pins with durable publication fences Refs #378. --- crates/lance-context-wal/src/replication.rs | 10 +- .../src/replication/checkpoint.rs | 27 +- .../src/replication/checkpoint/tests.rs | 12 +- .../src/replication/publication/outcome.rs | 2 +- .../src/replication/query.rs | 54 +++- .../src/replication/source.rs | 84 ++++-- .../src/replication/source/closure.rs | 272 ++++++++++++++++++ .../src/replication/source/tests.rs | 1 + .../src/replication/source/tests/closed.rs | 246 ++++++++++++++++ .../src/replication/state.rs | 3 +- .../tests/replication_process.rs | 2 + .../replication_process/managed_manifest.rs | 8 + .../replication_process/source_closure.rs | 120 ++++++++ docs/design/wal-manifest-authority.md | 2 +- docs/design/wal-prepared-descriptors.md | 2 +- docs/design/wal-publication-outcomes.md | 2 +- docs/design/wal-publication-plans.md | 2 +- docs/design/wal-publisher-authority.md | 2 +- docs/design/wal-source-closure.md | 37 +++ docs/design/wal-source-pins.md | 2 +- 20 files changed, 855 insertions(+), 35 deletions(-) create mode 100644 crates/lance-context-wal/src/replication/source/closure.rs create mode 100644 crates/lance-context-wal/src/replication/source/tests/closed.rs create mode 100644 crates/lance-context-wal/tests/replication_process/source_closure.rs create mode 100644 docs/design/wal-source-closure.md diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 44ced10..61eced9 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -93,7 +93,7 @@ impl Limits { "invalid replication limits" ); Ok(bincode::serialize(&( - "lance-context/replicated-append/v7", + "lance-context/replicated-append/v8", self, ))?) } @@ -458,6 +458,14 @@ impl Node { self.propose(group, request, batch) } + /// Persist a catalog-authorized source closure and reject delayed pin + /// requests for the closed plan. Bodies remain retained until retirement. + pub fn close_source(&mut self, request: u64, permit: source::ClosurePermit) -> Result<()> { + let group = permit.closure().group; + let batch = permit.batch(&self.layout, &self.limits)?; + self.propose(group, 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 d0a9736..cd358df 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 = 7; -const MAGIC: &[u8; 8] = b"LCWALCP7"; +const FORMAT: u32 = 8; +const MAGIC: &[u8; 8] = b"LCWALCP8"; pub const CHUNK_BYTES: usize = 1024 * 1024; fn codec() -> impl Options { @@ -73,6 +73,7 @@ struct Descriptor { producers: u64, tables: u64, pins: u64, + closed: u64, metadata_bytes: u64, payload_bytes: u64, metadata_digest: [u8; 32], @@ -89,6 +90,7 @@ fn metadata_limit(limits: &Limits) -> u64 { + limits.max_tables.min(limits.max_active_plans) * (source::MAX_PIN_BYTES as u64 + 4) + limits.max_manifest_records * 2048 + limits.max_tables * 256 + + limits.max_tables * (source::MAX_CLOSED_BYTES as u64 + 4) + 1024 } @@ -103,6 +105,7 @@ impl Descriptor { && self.producers <= self.records && self.tables <= limits.max_tables && self.pins <= limits.max_tables.min(limits.max_active_plans) + && self.closed <= 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, @@ -449,6 +452,7 @@ pub(super) fn build( producers: state.producers.len() as u64, tables: state.catalog.tables.len() as u64, pins: state.pins.len() as u64, + closed: state.closed.len() as u64, metadata_bytes: 0, payload_bytes, metadata_digest: [0; 32], @@ -462,6 +466,7 @@ pub(super) fn build( write_u64(&mut metadata, descriptor.producers)?; write_u64(&mut metadata, descriptor.tables)?; write_u64(&mut metadata, descriptor.pins)?; + write_u64(&mut metadata, descriptor.closed)?; for (index, meta) in &state.accepted { write_meta(&mut metadata, *index, meta)?; } @@ -474,6 +479,9 @@ pub(super) fn build( for pin in state.pins.values() { write_frame(&mut metadata, &source::encode_pin(pin)?)?; } + for closed in state.closed.values() { + write_frame(&mut metadata, &source::encode_closed(closed)?)?; + } (descriptor.metadata_bytes, descriptor.metadata_digest) = metadata.finish()?; let mut pack = Writer::new(&paths[1], payload_bytes)?; for (index, meta) in &state.accepted { @@ -521,7 +529,8 @@ pub(super) fn load( read_u64(&mut metadata)? == descriptor.records && read_u64(&mut metadata)? == descriptor.producers && read_u64(&mut metadata)? == descriptor.tables - && read_u64(&mut metadata)? == descriptor.pins, + && read_u64(&mut metadata)? == descriptor.pins + && read_u64(&mut metadata)? == descriptor.closed, "image metadata count mismatch" ); let mut state = State { @@ -573,6 +582,18 @@ pub(super) fn load( state.pins.insert(table, Arc::new(pin)); previous = Some(table); } + let mut previous = None; + for _ in 0..descriptor.closed { + let bytes = read_dynamic_frame(&mut metadata, source::MAX_CLOSED_BYTES)?; + let closed = source::decode_closed(&bytes)?; + let table = closed.permit.closure().table; + ensure!( + previous.is_none_or(|old| old < table), + "unordered or duplicate source closure" + ); + state.closed.insert(table, Arc::new(closed)); + previous = Some(table); + } 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 1a35474..7acac8e 100644 --- a/crates/lance-context-wal/src/replication/checkpoint/tests.rs +++ b/crates/lance-context-wal/src/replication/checkpoint/tests.rs @@ -396,7 +396,17 @@ fn mixed_catalog_image_preserves_payloads_and_rejects_corrupt_table_frames() -> 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 + 4 * 8; + let mut header = &original[12 + binding_len..]; + for count in [ + state.accepted.len(), + state.producers.len(), + state.catalog.tables.len(), + state.pins.len(), + state.closed.len(), + ] { + assert_eq!(read_u64(&mut header)?, count as u64); + } + let mut offset = original.len() - header.len(); 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; diff --git a/crates/lance-context-wal/src/replication/publication/outcome.rs b/crates/lance-context-wal/src/replication/publication/outcome.rs index 9e734c2..eee7885 100644 --- a/crates/lance-context-wal/src/replication/publication/outcome.rs +++ b/crates/lance-context-wal/src/replication/publication/outcome.rs @@ -61,7 +61,7 @@ pub struct Published { pub manifest: Arc, } impl Published { - pub(super) fn validate(&self, table: &Table, cut: u64) -> Result<()> { + pub(in crate::replication) fn validate(&self, table: &Table, cut: u64) -> Result<()> { let manifests = table .manifests .as_ref() diff --git a/crates/lance-context-wal/src/replication/query.rs b/crates/lance-context-wal/src/replication/query.rs index ea3bf41..bdc24f3 100644 --- a/crates/lance-context-wal/src/replication/query.rs +++ b/crates/lance-context-wal/src/replication/query.rs @@ -8,6 +8,15 @@ mod tests; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Query { + ClosePermit { + table: Uuid, + transaction: Uuid, + plan_index: u64, + source_group: u64, + }, + SourceClosure { + table: Uuid, + }, Manifest { table: Uuid, version: Option, @@ -37,6 +46,8 @@ pub enum Query { #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Value { + ClosePermit(Option>), + SourceClosure(Option>), Manifest(Option>), Publisher(Option>), Accepted(Page), @@ -83,7 +94,13 @@ impl Query { && version.is_none_or(|v| v > 0 && v < u64::MAX), "invalid manifest query scope" ), - Self::Pin { + Self::ClosePermit { + table, + transaction, + plan_index, + source_group, + } + | Self::Pin { table, transaction, plan_index, @@ -97,7 +114,9 @@ impl Query { && group == publication::control_group(layout), "invalid catalog dependency query" ), - Self::SourcePin { table } => ensure!(!table.is_nil(), "nil source table"), + Self::SourceClosure { table } | Self::SourcePin { table } => { + ensure!(!table.is_nil(), "nil source table") + } Self::Publisher { table } => ensure!( !table.is_nil() && group == publication::control_group(layout), "invalid publisher query scope" @@ -127,6 +146,37 @@ impl Query { signer: Option<&source::CatalogSigner>, ) -> Result { Ok(match self { + Self::SourceClosure { table } => Value::SourceClosure(state.closed.get(table).cloned()), + Self::ClosePermit { + table, + transaction, + plan_index, + source_group, + } => { + let selected = state.catalog.tables.get(table).filter(|table| { + table.progress.published.is_some() + && table.progress.pending.as_ref().is_some_and(|p| { + p.index == *plan_index + && p.plan.transaction == *transaction + && p.plan.sources.contains_key(source_group) + }) + }); + Value::ClosePermit( + selected + .map(|table| { + source::ClosurePermit::issue( + signer.context("catalog closure signer unavailable")?, + cut.clone(), + layout, + limits, + table, + *source_group, + ) + .map(Box::new) + }) + .transpose()?, + ) + } Self::Manifest { table, version } => Value::Manifest( state .catalog diff --git a/crates/lance-context-wal/src/replication/source.rs b/crates/lance-context-wal/src/replication/source.rs index 63e881e..cb54e4d 100644 --- a/crates/lance-context-wal/src/replication/source.rs +++ b/crates/lance-context-wal/src/replication/source.rs @@ -6,6 +6,9 @@ use crate::{ingest_contract::IngestContract, record::RecordKind}; use bincode::Options; use bytes::Bytes; use ring::signature::{Ed25519KeyPair, KeyPair, UnparsedPublicKey, ED25519}; +mod closure; +pub(super) use closure::{decode_closed, encode_closed, MAX_CLOSED_BYTES}; +pub use closure::{Closed, Closure, ClosurePermit}; #[cfg(test)] mod tests; @@ -166,21 +169,25 @@ impl Permit { pub(super) fn batch(self, layout: &Layout, limits: &Limits) -> Result { self.verify(layout, limits, self.selection.group)?; - let bytes = codec(MAX_BYTES).serialize(&self)?; - ensure!(bytes.len() <= MAX_BYTES, "oversized source pin command"); - let mut batch = Batch::new( + let (table, transaction, plan_index) = ( self.selection.table, - 0, self.selection.transaction, self.selection.plan_index, - Bytes::from(bytes), ); + let bytes = codec(MAX_BYTES).serialize(&Command::Pin(self))?; + ensure!(bytes.len() <= MAX_BYTES, "oversized source pin command"); + let mut batch = Batch::new(table, 0, transaction, plan_index, Bytes::from(bytes)); batch.meta.kind = RecordKind::Source; Ok(batch) } } -pub(super) fn decode(bytes: &[u8]) -> Result { +#[derive(Serialize, Deserialize)] +pub(super) enum Command { + Pin(Permit), + Close(ClosurePermit), +} +pub(super) fn decode(bytes: &[u8]) -> Result { ensure!(bytes.len() <= MAX_BYTES, "oversized source pin command"); let permit = codec(MAX_BYTES).deserialize(bytes)?; ensure!( @@ -196,14 +203,24 @@ pub(super) fn validate_record( meta: &BatchMeta, body: &[u8], ) -> Result<()> { - let permit = decode(body)?; - permit.verify(layout, limits, group)?; - let s = permit.selection(); + let command = decode(body)?; + let (table, transaction, plan_index) = match command { + Command::Pin(permit) => { + permit.verify(layout, limits, group)?; + let s = permit.selection(); + (s.table, s.transaction, s.plan_index) + } + Command::Close(permit) => { + permit.verify(layout, limits, group)?; + let c = permit.closure(); + (c.table, c.transaction, c.plan_index) + } + }; ensure!( meta.kind == RecordKind::Source - && meta.stream == s.table - && meta.producer == s.transaction - && meta.sequence == s.plan_index + && meta.stream == table + && meta.producer == transaction + && meta.sequence == plan_index && meta.bucket == 0, "source pin envelope mismatch" ); @@ -220,20 +237,28 @@ pub enum Rejection { Pending, Prefix, Capacity, + Closed, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Reply { Pinned { index: u64 }, Rejected(Rejection), + Closed { index: u64 }, } -fn complete_prefix(accepted: &BTreeMap, selection: &Selection, index: u64) -> bool { +fn complete_prefix(state: &State, selection: &Selection, index: u64) -> bool { let range = &selection.range; - // No release/cursor advancement exists in this increment. The retained - // source prefix starts at zero until durable resolution is integrated. - range.after == 0 + let previous = state + .closed + .get(&selection.table) + .map(|c| c.permit.closure()); + range.after == previous.map_or(0, |c| c.through) + && previous.is_none_or(|c| { + selection.plan_index > c.plan_index && selection.read_version >= c.version + }) && range.through < index - && accepted + && state + .accepted .range(( std::ops::Bound::Excluded(range.after), std::ops::Bound::Included(range.through), @@ -242,8 +267,26 @@ fn complete_prefix(accepted: &BTreeMap, selection: &Selection, i .map(|(i, meta)| (*i, meta)) .eq(range.records.iter().map(|(i, meta)| (*i, meta))) } -pub(super) fn apply(state: &mut State, permit: Permit, index: u64, limits: &Limits) -> Reply { +pub(super) fn apply( + state: &mut State, + command: Command, + index: u64, + limits: &Limits, +) -> Result { + match command { + Command::Pin(permit) => Ok(pin(state, permit, index, limits)), + Command::Close(permit) => closure::apply(state, permit, index, limits), + } +} +fn pin(state: &mut State, permit: Permit, index: u64, limits: &Limits) -> Reply { let selection = permit.selection(); + if state + .closed + .get(&selection.table) + .is_some_and(|closed| selection.plan_index <= closed.permit.closure().plan_index) + { + return Reply::Rejected(Rejection::Closed); + } if let Some(old) = state.pins.get(&selection.table) { return if old.permit.selection() == selection { Reply::Pinned { index: old.index } @@ -251,7 +294,7 @@ pub(super) fn apply(state: &mut State, permit: Permit, index: u64, limits: &Limi Reply::Rejected(Rejection::Pending) }; } - if !complete_prefix(&state.accepted, selection, index) { + if !complete_prefix(state, selection, index) { return Reply::Rejected(Rejection::Prefix); } if state.pins.len() as u64 >= limits.max_tables.min(limits.max_active_plans) { @@ -269,6 +312,7 @@ pub(super) fn validate_state( cut: u64, limits: &Limits, ) -> Result<()> { + closure::validate_state(state, layout, group, cut, limits)?; ensure!( state.pins.len() as u64 <= limits.max_tables.min(limits.max_active_plans), "source pin capacity" @@ -279,7 +323,7 @@ pub(super) fn validate_state( *table == pin.permit.selection.table && pin.index > 0 && pin.index <= cut - && complete_prefix(&state.accepted, pin.permit.selection(), pin.index), + && complete_prefix(state, pin.permit.selection(), pin.index), "invalid checkpoint source pin or prefix" ); } diff --git a/crates/lance-context-wal/src/replication/source/closure.rs b/crates/lance-context-wal/src/replication/source/closure.rs new file mode 100644 index 0000000..7891316 --- /dev/null +++ b/crates/lance-context-wal/src/replication/source/closure.rs @@ -0,0 +1,272 @@ +//! Catalog-authorized closure of an exact source pin. Closure persists a +//! monotonic fence before releasing the pin. Payload retirement is separate. +use super::*; + +const DOMAIN: &str = "lance-context/catalog-source-closure/v1"; +pub(in crate::replication) const MAX_CLOSED_BYTES: usize = 2048; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Closure { + pub table: Uuid, + pub transaction: Uuid, + pub plan_index: u64, + pub group: u64, + pub after: u64, + pub through: u64, + pub selection_digest: [u8; 32], + pub incarnation: Uuid, + pub manifest_identity: Uuid, + pub version: u64, + pub manifest_index: u64, + pub published_index: u64, +} + +fn selection_digest(selection: &Selection) -> Result<[u8; 32]> { + Ok(Sha256::digest( + codec(MAX_BYTES).serialize(&("lance-context/source-selection/v1", selection))?, + ) + .into()) +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ClosurePermit { + policy: [u8; 32], + cut: query::Cut, + closure: Closure, + signature: ([u8; 32], [u8; 32]), +} +impl ClosurePermit { + pub fn closure(&self) -> &Closure { + &self.closure + } + pub fn catalog_cut(&self) -> &query::Cut { + &self.cut + } + fn signed_bytes(&self) -> Result> { + Ok(codec(MAX_CLOSED_BYTES).serialize(&(DOMAIN, self.policy, &self.cut, &self.closure))?) + } + pub(in crate::replication) fn issue( + signer: &CatalogSigner, + cut: query::Cut, + layout: &Layout, + limits: &Limits, + table: &publication::Table, + group: u64, + ) -> Result { + ensure!( + limits.catalog_key == Some(signer.public_key()), + "catalog signing policy mismatch" + ); + let published = table + .progress + .published + .as_ref() + .context("publication unresolved")?; + published.validate(table, cut.applied)?; + let pending = table + .progress + .pending + .as_ref() + .context("source selection missing")?; + let range = pending + .plan + .sources + .get(&group) + .context("source not selected")?; + let selection = Selection { + table: table.authority.table, + transaction: pending.plan.transaction, + plan_index: pending.index, + read_version: pending.plan.read_version, + group, + ingest: table.ingest.clone(), + range: range.clone(), + }; + let mut permit = Self { + policy: Sha256::digest(limits.identity()?).into(), + cut, + closure: Closure { + table: selection.table, + transaction: selection.transaction, + plan_index: selection.plan_index, + group, + after: range.after, + through: range.through, + selection_digest: selection_digest(&selection)?, + incarnation: published.incarnation, + manifest_identity: published.manifest.identity, + version: published.manifest.version, + manifest_index: published.manifest.index, + published_index: published.index, + }, + signature: ([0; 32], [0; 32]), + }; + let signature = signer.0.sign(&permit.signed_bytes()?); + permit.signature = ( + signature.as_ref()[..32].try_into()?, + signature.as_ref()[32..].try_into()?, + ); + permit.verify(layout, limits, group)?; + Ok(permit) + } + pub(super) fn verify(&self, layout: &Layout, limits: &Limits, group: u64) -> Result<()> { + let key = limits + .catalog_key + .context("closure needs catalog verification key")?; + ensure!( + self.policy == <[u8; 32]>::from(Sha256::digest(limits.identity()?)), + "foreign closure policy" + ); + let mut signature = [0; 64]; + signature[..32].copy_from_slice(&self.signature.0); + signature[32..].copy_from_slice(&self.signature.1); + UnparsedPublicKey::new(&ED25519, key) + .verify(&self.signed_bytes()?, &signature) + .map_err(|_| anyhow::anyhow!("invalid catalog closure attestation"))?; + let c = &self.closure; + ensure!( + self.cut.cluster == layout.cluster + && self.cut.layout == layout.fingerprint()? + && self.cut.group == publication::control_group(layout) + && self.cut.term > 0 + && self.cut.quorum_index > 0 + && self.cut.quorum_index <= self.cut.applied + && self.cut.applied < u64::MAX + && c.plan_index > 0 + && c.plan_index < c.manifest_index + && c.manifest_index < c.published_index + && c.published_index <= self.cut.applied + && !c.table.is_nil() + && !c.transaction.is_nil() + && !c.incarnation.is_nil() + && !c.manifest_identity.is_nil() + && c.selection_digest != [0; 32] + && c.group == group + && layout.groups.contains_key(&group) + && c.after < c.through + && c.through < u64::MAX + && c.version > 1 + && c.version < (1u64 << 63) + && (group != self.cut.group || c.through < c.plan_index), + "invalid source closure scope or cut" + ); + Ok(()) + } + pub(in crate::replication) fn batch(self, layout: &Layout, limits: &Limits) -> Result { + self.verify(layout, limits, self.closure.group)?; + let c = &self.closure; + let mut batch = Batch::new( + c.table, + 0, + c.transaction, + c.plan_index, + Bytes::from(codec(MAX_BYTES).serialize(&Command::Close(self))?), + ); + batch.meta.kind = RecordKind::Source; + Ok(batch) + } +} + +/// Bounded by registered tables, not active pin slots. The signed statement is +/// compact and never retains the selected source metadata or payload buffers. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Closed { + pub index: u64, + pub pin_index: u64, + pub permit: ClosurePermit, +} + +pub(super) fn apply( + state: &mut State, + permit: ClosurePermit, + index: u64, + limits: &Limits, +) -> Result { + let c = &permit.closure; + if let Some(old) = state.closed.get(&c.table) { + if old.permit.closure == *c { + return Ok(Reply::Closed { index: old.index }); + } + if c.plan_index <= old.permit.closure.plan_index { + return Ok(Reply::Rejected(Rejection::Closed)); + } + if c.after != old.permit.closure.through + || c.version <= old.permit.closure.version + || c.incarnation != old.permit.closure.incarnation + { + return Ok(Reply::Rejected(Rejection::Prefix)); + } + } else if c.after != 0 { + return Ok(Reply::Rejected(Rejection::Prefix)); + } + let Some(pin) = state.pins.get(&c.table) else { + return Ok(Reply::Rejected(Rejection::Pending)); + }; + let selection = pin.permit.selection(); + if c.selection_digest != selection_digest(selection)? + || c.table != selection.table + || c.transaction != selection.transaction + || c.plan_index != selection.plan_index + || c.group != selection.group + || c.after != selection.range.after + || c.through != selection.range.through + || c.version != selection.read_version + 1 + { + return Ok(Reply::Rejected(Rejection::Prefix)); + } + if !state.closed.contains_key(&c.table) && state.closed.len() as u64 >= limits.max_tables { + return Ok(Reply::Rejected(Rejection::Capacity)); + } + let table = c.table; + let pin_index = pin.index; + state.closed.insert( + table, + Arc::new(Closed { + index, + pin_index, + permit, + }), + ); + state.pins.remove(&table); + // Accepted bodies and producer tails remain intact in this increment. + Ok(Reply::Closed { index }) +} + +pub(super) fn validate_state( + state: &State, + layout: &Layout, + group: u64, + cut: u64, + limits: &Limits, +) -> Result<()> { + ensure!( + state.closed.len() as u64 <= limits.max_tables, + "closed-source table capacity" + ); + for (table, closed) in &state.closed { + closed.permit.verify(layout, limits, group)?; + ensure!( + *table == closed.permit.closure.table + && closed.index <= cut + && closed.index > closed.pin_index + && closed.pin_index > closed.permit.closure.through, + "invalid closed-source checkpoint position" + ); + } + Ok(()) +} +pub(in crate::replication) fn encode_closed(closed: &Closed) -> Result> { + Ok(codec(MAX_CLOSED_BYTES).serialize(closed)?) +} +pub(in crate::replication) fn decode_closed(bytes: &[u8]) -> Result { + ensure!( + bytes.len() <= MAX_CLOSED_BYTES, + "oversized closed-source frame" + ); + let closed = codec(MAX_CLOSED_BYTES).deserialize(bytes)?; + ensure!( + encode_closed(&closed)? == bytes, + "noncanonical closed-source frame" + ); + Ok(closed) +} diff --git a/crates/lance-context-wal/src/replication/source/tests.rs b/crates/lance-context-wal/src/replication/source/tests.rs index 95a671f..2ca42f5 100644 --- a/crates/lance-context-wal/src/replication/source/tests.rs +++ b/crates/lance-context-wal/src/replication/source/tests.rs @@ -2,6 +2,7 @@ use super::*; use crate::ingest_contract::WritePolicy; use publication::{Acquire, Authority, Destination, Plan, PlanRequest, SourceRange}; use query::{Outcome as ReadOutcome, Query, Value}; +mod closed; struct Cluster { nodes: Vec, diff --git a/crates/lance-context-wal/src/replication/source/tests/closed.rs b/crates/lance-context-wal/src/replication/source/tests/closed.rs new file mode 100644 index 0000000..9bbee99 --- /dev/null +++ b/crates/lance-context-wal/src/replication/source/tests/closed.rs @@ -0,0 +1,246 @@ +use super::*; +use publication::{ + BootstrapManifest, ConfirmPublication, Descriptors, ManifestObject, PrepareRequest, + ReserveManifest, +}; + +fn publisher(events: Events) -> Result { + events + .completions + .into_iter() + .find_map(|c| match c.outcome { + Outcome::Committed(Applied::Publisher(reply)) => Some(reply), + _ => None, + }) + .context("publisher completion missing") +} + +fn fixture() -> Result<(Cluster, Permit, Uuid, Uuid)> { + let mut c = Cluster::new()?; + let authority = c.acquire(0)?; + let incarnation = Uuid::from_u128(800); + c.nodes[0].bootstrap_manifest( + 800, + BootstrapManifest { + authority, + incarnation, + request: Uuid::from_u128(801), + version: 8, + object: ManifestObject { + path: "_versions/8.manifest".into(), + bytes: 8192, + v2_names: false, + }, + transaction: Uuid::from_u128(802), + }, + )?; + let publication::Reply::ManifestReserved(base) = publisher(c.pump()?)? else { + anyhow::bail!("bootstrap missing") + }; + let plan = c.plan(|_| {})?; + let permit = c.permit(plan)?; + assert!(matches!(c.pin(permit.clone())?, Reply::Pinned { .. })); + Ok((c, permit, incarnation, base.identity)) +} + +fn closure_permit(c: &mut Cluster, plan: u64) -> Result> { + c.nodes[0].query( + 10, + 803, + Query::ClosePermit { + table: c.table, + transaction: c.transaction, + plan_index: plan, + source_group: 20, + }, + )?; + c.pump()? + .reads + .into_iter() + .find_map(|read| match read.outcome { + ReadOutcome::Observed { + value: Value::ClosePermit(permit), + .. + } => Some(permit.map(|p| *p)), + _ => None, + }) + .context("closure query missing") +} + +fn publish(c: &mut Cluster, incarnation: Uuid, predecessor: Uuid) -> Result { + let table = c.nodes[0].publisher(c.table)?.unwrap().clone(); + let pending = table.progress.pending.unwrap(); + let descriptors = Descriptors::lance_fragments(Bytes::from_static(b"unit-test-descriptors"))?; + c.nodes[0].prepare_publication( + 804, + PrepareRequest { + authority: table.authority.clone(), + transaction: c.transaction, + plan_index: pending.index, + descriptors: descriptors.clone(), + }, + )?; + let publication::Reply::Prepared { index } = publisher(c.pump()?)? else { + anyhow::bail!("prepare missing") + }; + c.nodes[0].reserve_manifest( + 805, + ReserveManifest { + authority: table.authority.clone(), + incarnation, + transaction: c.transaction, + plan_index: pending.index, + prepared_index: index, + descriptor_digest: descriptors.digest, + predecessor, + target: 9, + object: ManifestObject { + path: format!("_versions/9.manifest-{}", Uuid::new_v4()), + bytes: 8192, + v2_names: false, + }, + }, + )?; + let publication::Reply::ManifestReserved(selected) = publisher(c.pump()?)? else { + anyhow::bail!("reservation missing") + }; + assert!( + closure_permit(c, pending.index)?.is_none(), + "selection alone is not a closure outcome" + ); + c.nodes[0].confirm_publication( + 806, + ConfirmPublication { + authority: table.authority, + incarnation, + transaction: c.transaction, + plan_index: pending.index, + prepared_index: index, + descriptor_digest: descriptors.digest, + target: 9, + manifest_identity: selected.identity, + }, + )?; + assert!(matches!( + publisher(c.pump()?)?, + publication::Reply::Published(_) + )); + closure_permit(c, pending.index)?.context("published closure permit missing") +} + +fn close(c: &mut Cluster, permit: ClosurePermit) -> Result { + c.nodes[3].close_source(807, permit)?; + c.pump()? + .completions + .into_iter() + .find_map(|completion| match completion.outcome { + Outcome::Committed(Applied::Source(reply)) => Some(reply), + _ => None, + }) + .context("source closure missing") +} + +#[test] +fn durable_closure_releases_pin_rejects_delayed_permits_and_preserves_data() -> Result<()> { + let (mut c, pin, incarnation, predecessor) = fixture()?; + assert!(closure_permit(&mut c, pin.selection().plan_index)?.is_none()); + let permit = publish(&mut c, incarnation, predecessor)?; + c.append(c.table, 3)?; + let before = c.nodes[3].groups[&20].state.clone(); + let result = close(&mut c, permit.clone())?; + assert!(matches!(result, Reply::Closed { .. })); + assert_eq!(c.pin(pin.clone())?, Reply::Rejected(Rejection::Closed)); + assert_eq!(close(&mut c, permit.clone())?, result); + let state = &c.nodes[3].groups[&20].state; + assert!(state.pins.is_empty()); + assert_eq!( + state.closed.len(), + 1, + "raw streams must not allocate closure history" + ); + assert_eq!(state.accepted, before.accepted); + assert_eq!(state.producers, before.producers); + assert_eq!( + state.closed[&c.table].pin_index, + before.pins[&c.table].index + ); + assert_eq!( + state.closed[&c.table].permit.closure().through, + pin.selection().range.through + ); + for n in 0..6 { + let group = if n < 3 { 10 } else { 20 }; + let image = c.nodes[n] + .prepare_checkpoint(group, Uuid::new_v4())? + .run()?; + c.nodes[n].install_checkpoint(image)?; + } + c.reopen()?; + c.acquire(1)?; + let refreshed = closure_permit(&mut c, pin.selection().plan_index)?.unwrap(); + assert_ne!(permit.catalog_cut(), refreshed.catalog_cut()); + assert_eq!(close(&mut c, refreshed)?, result); + assert_eq!(c.pin(pin.clone())?, Reply::Rejected(Rejection::Closed)); + for node in &c.nodes[3..] { + let state = &node.groups[&20].state; + assert!(state.pins.is_empty()); + state.validate_checkpoint(&c.layout, 20, &c.limits, state.applied)?; + for (index, meta) in &before.accepted { + let record = node + .read_accepted(20, *index)? + .context("retained source disappeared")?; + assert_eq!(&record.meta, meta); + assert_eq!(record.body, Bytes::from(vec![meta.sequence as u8; 4096])); + } + } + Ok(()) +} + +#[test] +fn closure_signature_scope_and_snapshot_fences_are_checked() -> Result<()> { + let (mut c, pin, incarnation, predecessor) = fixture()?; + let permit = publish(&mut c, incarnation, predecessor)?; + let before = c.nodes[3].groups[&20].raw.raft.raft_log.last_index(); + for field in ["through", "plan_index", "group"] { + let mut value = serde_json::to_value(&permit)?; + let slot = &mut value["closure"][field]; + *slot = serde_json::json!(slot.as_u64().unwrap() + 1); + assert!(c.nodes[3] + .close_source(808, serde_json::from_value(value)?) + .is_err()); + } + let mut bytes = bincode::serialize(&permit)?; + *bytes.last_mut().unwrap() ^= 1; + assert!(c.nodes[3] + .close_source(808, bincode::deserialize(&bytes)?) + .is_err()); + assert_eq!( + c.nodes[3].groups[&20].raw.raft.raft_log.last_index(), + before + ); + assert!(matches!(close(&mut c, permit)?, Reply::Closed { .. })); + let state = c.nodes[3].groups[&20].state.clone(); + for mutation in 0..4 { + let mut invalid = state.clone(); + let closed = Arc::make_mut(invalid.closed.get_mut(&c.table).unwrap()); + match mutation { + 0 => closed.index = state.applied + 1, + 1 => closed.pin_index = closed.index, + 2 => closed.pin_index = closed.permit.closure().through, + 3 => { + invalid.pins.insert( + c.table, + Arc::new(Pin { + index: before, + permit: pin.clone(), + }), + ); + } + _ => unreachable!(), + } + assert!(invalid + .validate_checkpoint(&c.layout, 20, &c.limits, invalid.applied) + .is_err()); + } + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/state.rs b/crates/lance-context-wal/src/replication/state.rs index 427e3a7..cba7382 100644 --- a/crates/lance-context-wal/src/replication/state.rs +++ b/crates/lance-context-wal/src/replication/state.rs @@ -83,6 +83,7 @@ pub(super) fn validate_entry( #[derive(Clone, Debug, Default, PartialEq, Eq)] pub(super) struct State { + pub closed: BTreeMap>, pub pins: BTreeMap>, pub catalog: publication::Catalog, pub applied: u64, @@ -177,7 +178,7 @@ impl State { source::decode(&entry.data)?, entry.index, limits, - )))); + )?))); } if meta.kind == RecordKind::MultiPublication { return Ok(Some(Applied::Publisher(self.catalog.apply( diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index f9d1cfe..df9915d 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 { + CloseSource(u64, lance_context_wal::replication::source::ClosurePermit), ConfirmPublication( u64, lance_context_wal::replication::publication::ConfirmPublication, @@ -142,6 +143,7 @@ fn part(payload: bool) -> Part { fn run(worker: &mut Worker, request: Request) -> Result { let node = &mut worker.node; match request { + Request::CloseSource(token, permit) => node.close_source(token, permit)?, Request::ConfirmPublication(token, request) => node.confirm_publication(token, request)?, Request::BootstrapManifest(token, request) => node.bootstrap_manifest(token, request)?, Request::ReserveManifest(token, request) => node.reserve_manifest(token, request)?, 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 f68f0da..46e66c5 100644 --- a/crates/lance-context-wal/tests/replication_process/managed_manifest.rs +++ b/crates/lance-context-wal/tests/replication_process/managed_manifest.rs @@ -1,6 +1,8 @@ //! Actual Lance writer/reader adapter over the six-process catalog fixture. //! This qualifies the durable index, not production transport or coordinated GC. use super::*; +#[path = "source_closure.rs"] +mod source_closure; use lance_context_wal::replication::publication::{ Authority, BootstrapManifest, ConfirmPublication, ManifestObject, ManifestRecord, Prepared, ReserveManifest, @@ -423,6 +425,12 @@ pub(super) async fn publish_and_recover(c: Cluster, attempt: Attempt, mode: Mode ))) ); } + source_closure::close_and_restart( + &cluster, + handler.scope.authority.table, + &selected, + &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/source_closure.rs b/crates/lance-context-wal/tests/replication_process/source_closure.rs new file mode 100644 index 0000000..ecaeca9 --- /dev/null +++ b/crates/lance-context-wal/tests/replication_process/source_closure.rs @@ -0,0 +1,120 @@ +//! Real quorum closure after a real Lance publication, with a lost reply and +//! full process restart before attempting to replay the original pin request. +use super::*; +use lance_context_wal::replication::source::Rejection as SourceRejection; + +pub(super) fn close_and_restart( + cluster: &Arc>, + table: Uuid, + published: &ManifestRecord, + input: &Batch, +) -> Result<()> { + let mut c = cluster.lock().unwrap(); + let catalog = c.elect(1, 10, &[1, 2, 3])?; + let source = c.elect(4, 20, &[4, 5, 6])?; + c.command( + catalog, + Request::Query( + 10, + 600_000, + Query::Pin { + table, + transaction: published.transaction, + plan_index: published.plan_index, + source_group: 20, + }, + ), + )?; + c.command( + catalog, + Request::Query( + 10, + 600_001, + Query::ClosePermit { + table, + transaction: published.transaction, + plan_index: published.plan_index, + source_group: 20, + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + value: Value::Pin(Some(stale)), + .. + }) = c.read_outcome(600_000) + else { + anyhow::bail!("original pin retry permit missing"); + }; + let Some(ReadOutcome::Observed { + value: Value::ClosePermit(Some(permit)), + .. + }) = c.read_outcome(600_001) + else { + anyhow::bail!("closure authorization missing"); + }; + c.command(source, Request::CloseSource(600_002, *permit.clone()))?; + c.pump(&[4, 5, 6])?; + let before = c.completions.len(); + c.completions.retain(|c| c.request != 600_002); + assert_eq!(before - c.completions.len(), 1); + c.peers + .get_mut(&source) + .unwrap() + .call(Request::PrepareCheckpoint(20, Uuid::new_v4()))?; + c.command(source, Request::InstallCheckpoint)?; + c.pump(&[4, 5, 6])?; + c.queue.clear(); + c.peers.clear(); + for id in 1..=6 { + c.start(id)?; + } + let source = c.elect(4, 20, &[4, 5, 6])?; + c.command( + source, + Request::Query(20, 600_003, Query::SourceClosure { table }), + )?; + c.command( + source, + Request::Query(20, 600_004, Query::SourcePin { table }), + )?; + c.pump(&[4, 5, 6])?; + let Some(ReadOutcome::Observed { + value: Value::SourceClosure(Some(closed)), + .. + }) = c.read_outcome(600_003) + else { + anyhow::bail!("closed-source fence lost on restart"); + }; + assert_eq!(closed.permit.closure(), permit.closure()); + assert!(matches!( + c.read_outcome(600_004), + Some(ReadOutcome::Observed { + value: Value::SourcePin(None), + .. + }) + )); + c.command(source, Request::Pin(600_005, *stale))?; + c.command(source, Request::CloseSource(600_006, *permit))?; + c.pump(&[4, 5, 6])?; + assert_eq!( + c.outcome(600_005), + Some(Outcome::Committed(Applied::Source(SourceReply::Rejected( + SourceRejection::Closed + )))) + ); + assert_eq!( + c.outcome(600_006), + Some(Outcome::Committed(Applied::Source(SourceReply::Closed { + index: closed.index + }))) + ); + for id in 4..=6 { + verify_records( + c.peers.get_mut(&id).unwrap(), + 20, + std::slice::from_ref(input), + )?; + } + Ok(()) +} diff --git a/docs/design/wal-manifest-authority.md b/docs/design/wal-manifest-authority.md index a21d5f5..ae3b9b4 100644 --- a/docs/design/wal-manifest-authority.md +++ b/docs/design/wal-manifest-authority.md @@ -32,7 +32,7 @@ lookup, rather than serializing all retained history. Captured table views share the index and immutable records. There are at most 128 versions per table and a configurable node policy bound of 4,096 records by default. Capacity refuses new reservations while preserving exact retries. Application identity and checkpoint -format advance to v7 and include the index in bounds and integrity validation. +format advance to v8 and include the index in bounds and integrity validation. The real Lance test adapter writes unique staging manifests using the public `ManifestWriter`, then reserves them through six persistent TCP WAL processes. diff --git a/docs/design/wal-prepared-descriptors.md b/docs/design/wal-prepared-descriptors.md index b4e7e47..7df2aec 100644 --- a/docs/design/wal-prepared-descriptors.md +++ b/docs/design/wal-prepared-descriptors.md @@ -25,7 +25,7 @@ Captured table views share the prepared metadata through `Arc` and `Bytes`. Adding descriptors does not copy the plan's selected record list. The default per-plan descriptor budget is 256 KiB, with a 1 MiB protocol ceiling. Commands have separate type bounds, and checkpoint metadata accounts for both plans and -descriptors. The application/checkpoint format advances to v7; older identities +descriptors. The application/checkpoint format advances to v8; older identities and images are rejected rather than silently migrated. The `lance-integration` feature adds real Lance qualification dependencies. It diff --git a/docs/design/wal-publication-outcomes.md b/docs/design/wal-publication-outcomes.md index 631f351..02a62ea 100644 --- a/docs/design/wal-publication-outcomes.md +++ b/docs/design/wal-publication-outcomes.md @@ -18,7 +18,7 @@ Source positions and the publication checkpoint remain unchanged, and another plan remains blocked. In particular, a catalog index is never used as a source group cursor. No source pin, payload, descriptor, manifest or image is deleted. -Application identity and checkpoint format advance to v7. Older identities are +Application identity and checkpoint format advance to v8. Older identities are rejected without in-place migration. Images account for the bounded outcome metadata and verify its exact selected manifest, pending plan, prepared digest and catalog index ordering. Outcome metadata is bounded by active plans. diff --git a/docs/design/wal-publication-plans.md b/docs/design/wal-publication-plans.md index f470f99..ceb7845 100644 --- a/docs/design/wal-publication-plans.md +++ b/docs/design/wal-publication-plans.md @@ -37,7 +37,7 @@ the heap. Decoders reject noncanonical maps instead of silently accepting duplic keys normalized by the serializer. Source pins subsequently advance the optional application identity and checkpoint -format to v7. Existing v1/v2/v3/v4/v5/v6 directories and images are rejected; there is no in-place migration +format to v8. Existing v1/v2/v3/v4/v5/v6/v7 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. diff --git a/docs/design/wal-publisher-authority.md b/docs/design/wal-publisher-authority.md index f27c136..56cdc88 100644 --- a/docs/design/wal-publisher-authority.md +++ b/docs/design/wal-publisher-authority.md @@ -36,7 +36,7 @@ scope, epochs, grant positions, destination uniqueness and contracts, then repla 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. The current replication identity and application image -format are v7, including [immutable publication plans](wal-publication-plans.md) +format are v8, including [immutable publication plans](wal-publication-plans.md) and [source pins](wal-source-pins.md). Earlier v1/v2/v3/v4/v5 application directories are rejected; this optional component does not provide an in-place migration or change the production ingestion path. diff --git a/docs/design/wal-source-closure.md b/docs/design/wal-source-closure.md new file mode 100644 index 0000000..daa12ce --- /dev/null +++ b/docs/design/wal-source-closure.md @@ -0,0 +1,37 @@ +# Durable source closure + +A catalog quorum read can issue a signed closure only after the exact pending +publication has a durable `Published` outcome. The compact statement binds the +table, transaction, source range and selection digest to the selected manifest +identity and dataset incarnation. Source replicas need only the public catalog +key. An old isolated catalog cannot issue a fresh quorum attestation. + +`Node::close_source` replicates that statement in the source group. Apply checks +the exact durable pin, persists a closed-plan fence and source cursor, and then +releases the pin. Delayed pin permits for the closed plan cannot recreate it. +The latest exact closure retry returns the original source closure index, even +when its catalog attestation was refreshed after publisher takeover. Catalog +and source indices remain independent. + +Closed-source records are bounded by registered-table capacity, separately from +active pin capacity. Raw unregistered stream writes cannot allocate closure +history. The compact closure retains no source record list or payload buffers. +Subsequent pins must start at the saved source cursor and advance the catalog +plan identity; retained history cannot be repinned from an old cursor. + +Application identity and image format advance to v8, without in-place migration +from older identities. Snapshot validation verifies closure signatures, scope, +local index ordering and compatibility with any remaining active pins. + +This increment retains accepted payloads, producer tails, catalog selections, +descriptors, manifests and image files. It does not yet advance the catalog to +its next plan, retire payload metadata, reclaim disk or enable production +ingestion. Those steps must preserve retry receipts and active image/read +ownership before they can establish bounded sustained ingestion. + +Tests exercise closure after real Lance publication, dropped closure replies, +source snapshots, full six-process restart, delayed pin rejection and exact +closure retry. Source state tests also cover signature tampering, invalid +checkpoint fences and preservation of data outside the selected range. + +Refs #378. diff --git a/docs/design/wal-source-pins.md b/docs/design/wal-source-pins.md index e7aa604..e0bec23 100644 --- a/docs/design/wal-source-pins.md +++ b/docs/design/wal-source-pins.md @@ -34,7 +34,7 @@ Pins survive local checkpoints, received snapshots and retained-storage restart; checkpoint validation verifies both signatures and complete source prefixes. Command and image metadata lengths and the number of active pins are bounded. -The application/checkpoint format advances to v7 and rejects older identities +The application/checkpoint format advances to v8 and rejects older identities and images. There is no in-place migration in this optional component. This increment retains pins and payloads. It does not enable release, advance