From 5827d5d0fb281083574b6e2d60cd462dee29e038 Mon Sep 17 00:00:00 2001 From: Beinan Date: Wed, 21 Oct 2026 03:02:08 +0000 Subject: [PATCH] feat(wal): advance catalog cursors after source closure --- crates/lance-context-wal/src/replication.rs | 12 + .../src/replication/checkpoint.rs | 2 +- .../src/replication/publication.rs | 30 +- .../src/replication/publication/advance.rs | 183 +++++++++ .../replication/publication/advance/tests.rs | 352 ++++++++++++++++++ .../src/replication/publication/outcome.rs | 6 + .../src/replication/publication/plan.rs | 6 + .../src/replication/publication/plan/tests.rs | 15 +- .../replication/publication/table_codec.rs | 82 ++++ .../src/replication/source/closure.rs | 21 ++ .../tests/replication_process.rs | 5 + .../replication_process/source_closure.rs | 88 +++++ docs/design/wal-catalog-advancement.md | 50 +++ 13 files changed, 839 insertions(+), 13 deletions(-) create mode 100644 crates/lance-context-wal/src/replication/publication/advance.rs create mode 100644 crates/lance-context-wal/src/replication/publication/advance/tests.rs create mode 100644 crates/lance-context-wal/src/replication/publication/table_codec.rs create mode 100644 docs/design/wal-catalog-advancement.md diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index e2858e0..2c1fb84 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -521,6 +521,18 @@ impl Node { self.propose(publication::control_group(&self.layout), request, batch) } + /// Advance independent source cursors only after every selected source + /// proves the exact durable closure. Completion permits the next plan; + /// physical file reclamation still requires a durable application image. + pub fn advance_publication( + &mut self, + request: u64, + advancement: publication::AdvancePublication, + ) -> Result<()> { + let batch = advancement.batch(&self.layout, &self.limits)?; + self.propose(publication::control_group(&self.layout), request, batch) + } + /// Verify a catalog quorum-read attestation and persist the exact complete /// table prefix in its source group. Completion, never enqueue, pins it. pub fn pin_source(&mut self, request: u64, permit: source::Permit) -> Result<()> { diff --git a/crates/lance-context-wal/src/replication/checkpoint.rs b/crates/lance-context-wal/src/replication/checkpoint.rs index 2a4f555..a9715b9 100644 --- a/crates/lance-context-wal/src/replication/checkpoint.rs +++ b/crates/lance-context-wal/src/replication/checkpoint.rs @@ -90,7 +90,7 @@ fn metadata_limit(limits: &Limits) -> u64 { * (128 * 1024 + limits.max_descriptor_bytes + 4096) + 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 * (256 + publication::advance_metadata_bytes()) + limits.max_tables * (source::MAX_CLOSED_BYTES as u64 + 4) + limits.max_producers * (state::MAX_RETIRED_BYTES as u64 + 4) + 1024 diff --git a/crates/lance-context-wal/src/replication/publication.rs b/crates/lance-context-wal/src/replication/publication.rs index 254613b..32132ed 100644 --- a/crates/lance-context-wal/src/replication/publication.rs +++ b/crates/lance-context-wal/src/replication/publication.rs @@ -7,10 +7,13 @@ use super::*; use crate::{ingest_contract::IngestContract, record::RecordKind}; use bincode::Options; use bytes::Bytes; +mod advance; mod manifest; mod outcome; mod plan; mod prepared; +mod table_codec; +pub use advance::{AdvancePublication, Advanced}; pub use manifest::{ BootstrapManifest, ManifestIndex, ManifestObject, ManifestRecord, ReserveManifest, MAX_MANIFEST_VERSIONS, @@ -28,7 +31,11 @@ pub(super) const MAX_TABLE_BYTES: usize = MAX_BASE_TABLE_BYTES + plan::MAX_PLAN_BYTES + prepared::MAX_PREPARE_BYTES + manifest::MAX_MANIFEST_INDEX_BYTES - + outcome::MAX_OUTCOME_BYTES; + + outcome::MAX_OUTCOME_BYTES + + advance::MAX_ADVANCED_BYTES; +pub(super) const fn advance_metadata_bytes() -> u64 { + advance::MAX_ADVANCED_BYTES as u64 +} fn codec(limit: usize) -> impl Options { bincode::DefaultOptions::new() .with_fixint_encoding() @@ -80,6 +87,7 @@ enum Command { BootstrapManifest(Box), ReserveManifest(Box), ConfirmPublication(Box), + AdvancePublication(Box), } impl Command { fn identity(&self) -> (Uuid, Uuid, u64) { @@ -92,6 +100,9 @@ impl Command { Self::ConfirmPublication(p) => { (p.authority.table, p.authority.owner, p.authority.epoch) } + Self::AdvancePublication(p) => { + (p.authority.table, p.authority.owner, p.authority.epoch) + } } } fn byte_limit(&self) -> usize { @@ -103,6 +114,7 @@ impl Command { manifest::MAX_MANIFEST_COMMAND_BYTES } Self::ConfirmPublication(_) => outcome::MAX_OUTCOME_BYTES, + Self::AdvancePublication(_) => advance::MAX_ADVANCE_BYTES, } } fn batch(self, layout: &Layout) -> Result { @@ -177,6 +189,7 @@ pub enum Rejection { ManifestIdentity, ManifestPredecessor, PublicationIdentity, + SourceClosure, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Reply { @@ -187,6 +200,7 @@ pub enum Reply { ManifestReserved(Arc), ManifestTaken(Arc), Published(Arc), + Advanced(Arc), } #[derive(Serialize, Deserialize)] @@ -266,6 +280,7 @@ pub(super) fn validate_record( Command::BootstrapManifest(p) => p.validate(layout)?, Command::ReserveManifest(p) => p.validate(layout)?, Command::ConfirmPublication(p) => p.validate(layout)?, + Command::AdvancePublication(p) => p.validate(layout, limits)?, } Ok(()) } @@ -288,6 +303,9 @@ impl Catalog { Command::ConfirmPublication(request) => { return self.confirm_publication(*request, index) } + Command::AdvancePublication(request) => { + return self.advance_publication(*request, index) + } Command::Acquire(a) => *a, }; let old = self.tables.get(&a.table); @@ -392,14 +410,8 @@ impl Catalog { } } pub(super) fn encode_table(table: &Table) -> Result> { - Ok(codec(MAX_TABLE_BYTES).serialize(table)?) + table_codec::encode(table) } pub(super) fn decode_table(bytes: &[u8]) -> Result { - 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) + table_codec::decode(bytes) } diff --git a/crates/lance-context-wal/src/replication/publication/advance.rs b/crates/lance-context-wal/src/replication/publication/advance.rs new file mode 100644 index 0000000..129d22d --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/advance.rs @@ -0,0 +1,183 @@ +//! Atomic source-cursor advancement after all selected source quorums close. +use super::*; +use crate::replication::source::{ClosedProof, Selection}; + +#[cfg(test)] +mod tests; + +pub(super) const MAX_ADVANCE_BYTES: usize = 64 * 4096 + 8192; +pub(super) const MAX_ADVANCED_BYTES: usize = 8192; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct AdvancePublication { + pub authority: Authority, + pub published_index: u64, + pub proofs: BTreeMap, +} + +impl AdvancePublication { + pub(in crate::replication) fn batch(self, layout: &Layout, limits: &Limits) -> Result { + self.validate(layout, limits)?; + Command::AdvancePublication(Box::new(self)).batch(layout) + } + + pub(super) fn validate(&self, layout: &Layout, limits: &Limits) -> Result<()> { + self.authority.validate(layout)?; + ensure!( + self.published_index > 0 + && self.published_index < u64::MAX + && !self.proofs.is_empty() + && self.proofs.len() <= layout.groups.len() + && self.proofs.len() <= 64, + "invalid source closure set" + ); + for (group, proof) in &self.proofs { + proof.verify(layout, limits)?; + let closure = proof.closed().permit.closure(); + ensure!( + *group == proof.source_cut().group + && closure.table == self.authority.table + && closure.published_index == self.published_index, + "source closure request scope mismatch" + ); + } + Ok(()) + } + + fn digest(&self) -> Result<[u8; 32]> { + // Bind the entire sorted set, not refreshed attestation read cuts. + let closures: Vec<_> = self + .proofs + .iter() + .map(|(group, proof)| (*group, proof.closed().permit.closure())) + .collect(); + Ok(Sha256::digest( + codec(MAX_ADVANCE_BYTES) + .serialize(&("lance-context/catalog-advanced-closures/v1", closures))?, + ) + .into()) + } +} + +/// One bounded receipt per table. It shares the already selected manifest and +/// retains no source payloads, input metadata or prepared fragment descriptors. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Advanced { + pub index: u64, + pub publication: Arc, + pub sources: BTreeMap, + pub closure_digest: [u8; 32], +} + +impl Advanced { + pub(super) fn validate(&self, table: &Table, cut: u64) -> Result<()> { + let manifests = table + .manifests + .as_ref() + .context("advancement without manifest index")?; + let published = &self.publication; + let selected = &published.manifest; + ensure!( + published.incarnation == manifests.incarnation + && manifests.versions.get(&selected.version) == Some(selected) + && selected.version > manifests.initial_version + && selected.version == table.progress.minimum_read_version + && published.index > selected.index + && self.index > published.index + && self.index <= cut + && self.index != table.authority.grant_index + && self.closure_digest != [0; 32] + && !self.sources.is_empty() + && self.sources.len() <= 64 + && self.sources.iter().all(|(group, through)| { + *through > 0 + && *through < u64::MAX + && table.progress.checkpoint.get(group) == Some(through) + }) + && table + .progress + .pending + .as_ref() + .is_none_or(|p| p.index > self.index), + "invalid catalog advancement receipt" + ); + Ok(()) + } +} + +impl Catalog { + pub(super) fn advance_publication(&mut self, request: AdvancePublication, index: u64) -> Reply { + let Some(table) = self.tables.get(&request.authority.table) else { + return Reply::Rejected(Rejection::Authority); + }; + let Ok(digest) = request.digest() else { + return Reply::Rejected(Rejection::SourceClosure); + }; + if let Some(old) = &table.progress.advanced { + if old.publication.index == request.published_index && old.closure_digest == digest { + // A stale owner may resolve only the already completed exact + // operation. It cannot mutate a newer pending publication. + return Reply::Advanced(old.clone()); + } + } + if table.authority != request.authority { + return Reply::Rejected(Rejection::Authority); + } + let (Some(pending), Some(published)) = (&table.progress.pending, &table.progress.published) + else { + return Reply::Rejected(Rejection::PublicationIdentity); + }; + if published.index != request.published_index { + return Reply::Rejected(Rejection::PublicationIdentity); + } + if !request.proofs.keys().eq(pending.plan.sources.keys()) { + return Reply::Rejected(Rejection::SourceClosure); + } + for (group, range) in &pending.plan.sources { + let selection = Selection { + table: table.authority.table, + transaction: pending.plan.transaction, + plan_index: pending.index, + read_version: pending.plan.read_version, + group: *group, + ingest: table.ingest.clone(), + range: range.clone(), + }; + let closure = request.proofs[group].closed().permit.closure(); + if !closure + .matches_publication(&selection, published) + .unwrap_or(false) + { + return Reply::Rejected(Rejection::SourceClosure); + } + } + let advanced = Arc::new(Advanced { + index, + publication: published.clone(), + sources: pending + .plan + .sources + .iter() + .map(|(group, range)| (*group, range.through)) + .collect(), + closure_digest: digest, + }); + let progress = &mut self + .tables + .get_mut(&request.authority.table) + .expect("known table") + .progress; + progress.checkpoint.extend( + advanced + .sources + .iter() + .map(|(group, through)| (*group, *through)), + ); + progress.minimum_read_version = advanced.publication.manifest.version; + progress.pending = None; + progress.prepared = None; + progress.published = None; + progress.advanced = Some(advanced.clone()); + Reply::Advanced(advanced) + } +} diff --git a/crates/lance-context-wal/src/replication/publication/advance/tests.rs b/crates/lance-context-wal/src/replication/publication/advance/tests.rs new file mode 100644 index 0000000..306cc38 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/advance/tests.rs @@ -0,0 +1,352 @@ +use super::*; +use crate::replication::{ + query::Cut, + source::{CatalogSigner, Closed, ClosurePermit, SourceSigner}, +}; +use ring::{rand::SystemRandom, signature::Ed25519KeyPair}; + +struct Fixture { + layout: Arc, + limits: Limits, + catalog: Catalog, + signer: CatalogSigner, + sources: BTreeMap, + confirmation: ConfirmPublication, +} + +fn fixture() -> Result { + let random = SystemRandom::new(); + let signer = + CatalogSigner::from_pkcs8(Ed25519KeyPair::generate_pkcs8(&random).unwrap().as_ref())?; + let layout = crate::replication::tests::layout(); + let mut sources = BTreeMap::new(); + for group in layout.groups.keys() { + sources.insert( + *group, + SourceSigner::from_pkcs8(Ed25519KeyPair::generate_pkcs8(&random).unwrap().as_ref())?, + ); + } + let limits = Limits { + catalog_key: Some(signer.public_key()), + source_keys: sources + .iter() + .map(|(group, signer)| (*group, signer.public_key())) + .collect(), + ..Default::default() + }; + let (mut catalog, _, reserve) = super::super::manifest::tests::fixture(&limits)?; + let table = catalog.tables.get_mut(&reserve.authority.table).unwrap(); + Arc::make_mut(table.progress.pending.as_mut().unwrap()) + .plan + .sources + .insert( + 10, + SourceRange { + after: 0, + through: 2, + records: vec![( + 2, + Batch::new( + table.authority.table, + 0, + Uuid::from_u128(900), + 1, + Bytes::from_static(b"source ten"), + ) + .meta, + )], + }, + ); + let Reply::ManifestReserved(selected) = catalog.reserve_manifest(reserve.clone(), 8, &limits) + else { + anyhow::bail!("manifest not selected") + }; + let confirmation = ConfirmPublication { + authority: reserve.authority, + incarnation: reserve.incarnation, + transaction: reserve.transaction, + plan_index: reserve.plan_index, + prepared_index: reserve.prepared_index, + descriptor_digest: reserve.descriptor_digest, + target: reserve.target, + manifest_identity: selected.identity, + }; + assert!(matches!( + catalog.confirm_publication(confirmation.clone(), 9), + Reply::Published(_) + )); + catalog.validate(&layout, 10, 9, &limits)?; + Ok(Fixture { + layout, + limits, + catalog, + signer, + sources, + confirmation, + }) +} + +impl Fixture { + fn table(&self) -> &Table { + &self.catalog.tables[&self.confirmation.authority.table] + } + + fn cut(&self, group: u64, index: u64) -> Result { + Ok(Cut { + cluster: self.layout.cluster, + layout: self.layout.fingerprint()?, + group, + term: 1, + quorum_index: index, + applied: index, + }) + } + + fn request_for(&self, table: &Table, source_cut: u64) -> Result { + let published = table.progress.published.as_ref().unwrap(); + let mut proofs = BTreeMap::new(); + for (group, range) in &table.progress.pending.as_ref().unwrap().plan.sources { + let permit = ClosurePermit::issue( + &self.signer, + self.cut(10, published.index)?, + &self.layout, + &self.limits, + table, + *group, + )?; + let pin_index = range.through.max(published.index) + 1; + let closed = Arc::new(Closed { + index: pin_index + 1, + pin_index, + permit, + }); + proofs.insert( + *group, + ClosedProof::issue( + &self.sources[group], + self.cut(*group, source_cut.max(closed.index))?, + &self.layout, + &self.limits, + closed, + )?, + ); + } + Ok(AdvancePublication { + authority: table.authority.clone(), + published_index: published.index, + proofs, + }) + } + + fn request(&self) -> Result { + self.request_for(self.table(), 23) + } + + fn apply(&mut self, request: AdvancePublication, index: u64) -> Result { + let batch = request.batch(&self.layout, &self.limits)?; + validate_record(&self.layout, 10, &self.limits, &batch.meta, &batch.body)?; + Ok(self + .catalog + .apply(decode(&batch.body)?, index, &self.limits)) + } +} + +#[test] +fn all_source_closures_advance_independent_cursors_and_free_next_plan() -> Result<()> { + let mut f = fixture()?; + let captured = f.table().clone(); + let request = f.request()?; + let Reply::Advanced(advanced) = f.apply(request.clone(), 30)? else { + anyhow::bail!("not advanced") + }; + assert_eq!(advanced.index, 30); + assert_eq!(advanced.sources, BTreeMap::from([(10, 2), (20, 20)])); + assert_eq!(f.table().progress.checkpoint, advanced.sources); + assert_eq!(f.table().progress.minimum_read_version, 8); + assert!(f.table().progress.pending.is_none()); + assert!(f.table().progress.prepared.is_none()); + assert!(f.table().progress.published.is_none()); + assert!(Arc::ptr_eq( + &advanced.publication, + captured.progress.published.as_ref().unwrap() + )); + assert!(captured.progress.checkpoint.is_empty()); + assert_eq!( + f.apply(request.clone(), 31)?, + Reply::Advanced(advanced.clone()) + ); + let refreshed = f.request_for(&captured, 100)?; + assert_ne!( + request.proofs[&20].source_cut(), + refreshed.proofs[&20].source_cut() + ); + assert_eq!(f.apply(refreshed, 32)?, Reply::Advanced(advanced.clone())); + + let mut takeover = super::super::tests::input(); + takeover.owner = Uuid::new_v4(); + takeover.request = Uuid::new_v4(); + takeover.expected_epoch = 1; + let Reply::Acquired(authority) = + f.catalog + .apply(decode(&takeover.batch(&f.layout)?.body)?, 33, &f.limits) + else { + anyhow::bail!("takeover") + }; + let mut next = captured.progress.pending.as_ref().unwrap().plan.clone(); + next.transaction = Uuid::new_v4(); + next.read_version = 8; + for (group, range) in &mut next.sources { + range.after = advanced.sources[group]; + range.through = if *group == 10 { 32 } else { 40 }; + range.records[0].0 = range.through; + range.records[0].1.sequence += 1; + } + let batch = PlanRequest { + authority, + plan: next, + } + .batch(&f.layout, &f.limits)?; + assert_eq!( + f.catalog.apply(decode(&batch.body)?, 34, &f.limits), + Reply::Planned { index: 34 } + ); + let pending = f.table().progress.pending.clone(); + assert_eq!( + f.apply(request, 35)?, + Reply::Advanced(advanced.clone()), + "old-owner exact completion survives takeover" + ); + assert_eq!(f.table().progress.pending, pending); + assert_eq!( + f.catalog.confirm_publication(f.confirmation.clone(), 36), + Reply::Published(advanced.publication.clone()) + ); + f.catalog.validate(&f.layout, 10, 36, &f.limits)?; + assert_eq!(decode_table(&encode_table(f.table())?)?, *f.table()); + Ok(()) +} + +#[test] +fn missing_or_wrong_valid_closure_cannot_partially_advance() -> Result<()> { + let mut f = fixture()?; + let before = f.catalog.clone(); + let request = f.request()?; + let mut missing = request.clone(); + missing.proofs.remove(&20); + assert_eq!( + f.apply(missing, 30)?, + Reply::Rejected(Rejection::SourceClosure) + ); + assert_eq!(f.catalog, before); + let mut wrong = f.table().clone(); + Arc::make_mut(wrong.progress.pending.as_mut().unwrap()) + .plan + .sources + .get_mut(&20) + .unwrap() + .through += 1; + let altered = f.request_for(&wrong, 30)?; + for proof in altered.proofs.values() { + proof.verify(&f.layout, &f.limits)?; + } + assert_eq!( + f.apply(altered, 31)?, + Reply::Rejected(Rejection::SourceClosure) + ); + assert_eq!(f.catalog, before); + let mut stale = request.clone(); + stale.authority.owner = Uuid::new_v4(); + assert_eq!(f.apply(stale, 32)?, Reply::Rejected(Rejection::Authority)); + assert_eq!(f.catalog, before); + let mut forged = request.clone(); + let mut wire = bincode::serialize(&forged.proofs[&20])?; + let tail = wire.len() - 1; + wire[tail] ^= 1; + forged.proofs.insert(20, bincode::deserialize(&wire)?); + assert!(f.apply(forged, 33).is_err()); + assert_eq!(f.catalog, before); + let mut mixed = request.clone(); + mixed.published_index += 1; + assert!(f.apply(mixed, 34).is_err()); + assert_eq!(f.catalog, before); + let mut misplaced = request.clone(); + misplaced.proofs.insert(10, request.proofs[&20].clone()); + assert!(f.apply(misplaced, 35).is_err()); + assert_eq!(f.catalog, before); + assert!(matches!(f.apply(request, 36)?, Reply::Advanced(_))); + f.catalog.validate(&f.layout, 10, 36, &f.limits)?; + Ok(()) +} + +#[test] +fn legacy_table_bytes_are_preserved_and_advancement_suffix_is_strict() -> Result<()> { + let mut f = fixture()?; + let table = f.table(); + // Independent encoding of the fields in the old on-disk table schema. + let old = codec(MAX_TABLE_BYTES).serialize(&( + &table.manifests, + &table.destination, + &table.ingest, + &table.authority, + table.request, + ( + &table.progress.checkpoint, + table.progress.minimum_read_version, + &table.progress.pending, + &table.progress.prepared, + &table.progress.published, + ), + ))?; + assert_eq!(encode_table(table)?, old); + assert_eq!(decode_table(&old)?, *table); + let request = f.request()?; + f.apply(request, 30)?; + let encoded = encode_table(f.table())?; + assert_eq!(decode_table(&encoded)?, *f.table()); + for removed in 1..=8 { + assert!(decode_table(&encoded[..encoded.len() - removed]).is_err()); + } + let mut trailing = encoded.clone(); + trailing.push(0); + assert!(decode_table(&trailing).is_err()); + let marker = encoded.windows(8).position(|w| w == b"LCWALAD1").unwrap(); + let mut unknown = encoded; + unknown[marker] ^= 1; + assert!(decode_table(&unknown).is_err()); + Ok(()) +} + +#[test] +fn checkpoint_rejects_corrupt_completed_receipts() -> Result<()> { + let mut f = fixture()?; + let request = f.request()?; + f.apply(request, 30)?; + for mutation in 0..8 { + let mut broken = f.catalog.clone(); + let table = broken + .tables + .get_mut(&f.confirmation.authority.table) + .unwrap(); + let advanced = Arc::make_mut(table.progress.advanced.as_mut().unwrap()); + match mutation { + 0 => advanced.index = advanced.publication.index, + 1 => advanced.index = 31, + 2 => advanced.closure_digest = [0; 32], + 3 => advanced.sources.clear(), + 4 => { + advanced.sources.insert(20, 21); + } + 5 => Arc::make_mut(&mut advanced.publication).incarnation = Uuid::new_v4(), + 6 => table.progress.minimum_read_version += 1, + 7 => { + table.progress.checkpoint.remove(&10); + } + _ => unreachable!(), + } + assert!( + broken.validate(&f.layout, 10, 30, &f.limits).is_err(), + "mutation {mutation}" + ); + } + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/publication/outcome.rs b/crates/lance-context-wal/src/replication/publication/outcome.rs index eee7885..3a1b83b 100644 --- a/crates/lance-context-wal/src/replication/publication/outcome.rs +++ b/crates/lance-context-wal/src/replication/publication/outcome.rs @@ -103,6 +103,12 @@ impl Catalog { let Some(table) = self.tables.get(&request.authority.table) else { return Reply::Rejected(Rejection::Authority); }; + if let Some(advanced) = &table.progress.advanced { + let published = &advanced.publication; + if request.matches(published.incarnation, &published.manifest) { + return Reply::Published(published.clone()); + } + } // Exact resolution remains available after publisher takeover. It // reports an existing outcome and grants no authority to change state. if let Some(published) = &table.progress.published { diff --git a/crates/lance-context-wal/src/replication/publication/plan.rs b/crates/lance-context-wal/src/replication/publication/plan.rs index 4e4284a..ddfd36f 100644 --- a/crates/lance-context-wal/src/replication/publication/plan.rs +++ b/crates/lance-context-wal/src/replication/publication/plan.rs @@ -146,6 +146,8 @@ pub struct Progress { pub prepared: Option>, /// Durable selected-manifest resolution; sources remain pinned for closure. pub published: Option>, + /// Latest completed advancement, retained for ambiguous exact retries. + pub advanced: Option>, } impl Default for Progress { fn default() -> Self { @@ -155,6 +157,7 @@ impl Default for Progress { pending: None, prepared: None, published: None, + advanced: None, } } } @@ -177,6 +180,9 @@ impl Progress { cut: u64, limits: &Limits, ) -> Result<()> { + if let Some(advanced) = &self.advanced { + advanced.validate(table, cut)?; + } if let Some(published) = &self.published { published.validate(table, cut)?; } diff --git a/crates/lance-context-wal/src/replication/publication/plan/tests.rs b/crates/lance-context-wal/src/replication/publication/plan/tests.rs index 44407a4..42a3447 100644 --- a/crates/lance-context-wal/src/replication/publication/plan/tests.rs +++ b/crates/lance-context-wal/src/replication/publication/plan/tests.rs @@ -353,8 +353,17 @@ fn wire_and_checkpoint_maps_cannot_hide_duplicate_source_groups() -> Result<()> 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()); + // Public query tables include the advancement field; checkpoint frames + // preserve the legacy prefix and carry advancement in a tagged suffix. + let query = codec(MAX_TABLE_BYTES).serialize(table)?; + let bad_query = replace_once(&query, &original, &duplicate); + assert_eq!( + codec(MAX_TABLE_BYTES).deserialize::
(&bad_query)?, + *table + ); + let checkpoint = encode_table(table)?; + assert_eq!(decode_table(&checkpoint)?, *table); + let bad_checkpoint = replace_once(&checkpoint, &original, &duplicate); + assert!(decode_table(&bad_checkpoint).is_err()); Ok(()) } diff --git a/crates/lance-context-wal/src/replication/publication/table_codec.rs b/crates/lance-context-wal/src/replication/publication/table_codec.rs new file mode 100644 index 0000000..16e4517 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/table_codec.rs @@ -0,0 +1,82 @@ +//! Preserve exact legacy table bytes until a table has its first advancement. +//! A bounded tagged suffix carries the new receipt in application checkpoints. +use super::*; +use std::io::Cursor; + +const EXTENSION: [u8; 8] = *b"LCWALAD1"; + +#[derive(Serialize, Deserialize)] +struct LegacyProgress { + checkpoint: BTreeMap, + minimum_read_version: u64, + pending: Option>, + prepared: Option>, + published: Option>, +} + +#[derive(Serialize, Deserialize)] +struct LegacyTable { + manifests: Option>, + destination: Destination, + ingest: IngestContract, + authority: Authority, + request: Uuid, + progress: LegacyProgress, +} + +pub(super) fn encode(table: &Table) -> Result> { + let progress = &table.progress; + let legacy = LegacyTable { + manifests: table.manifests.clone(), + destination: table.destination.clone(), + ingest: table.ingest.clone(), + authority: table.authority.clone(), + request: table.request, + progress: LegacyProgress { + checkpoint: progress.checkpoint.clone(), + minimum_read_version: progress.minimum_read_version, + pending: progress.pending.clone(), + prepared: progress.prepared.clone(), + published: progress.published.clone(), + }, + }; + let mut bytes = codec(MAX_TABLE_BYTES).serialize(&legacy)?; + if let Some(advanced) = &progress.advanced { + bytes.extend(codec(advance::MAX_ADVANCED_BYTES).serialize(&(EXTENSION, advanced))?); + } + ensure!(bytes.len() <= MAX_TABLE_BYTES, "publisher table byte limit"); + Ok(bytes) +} + +pub(super) fn decode(bytes: &[u8]) -> Result
{ + ensure!(bytes.len() <= MAX_TABLE_BYTES, "publisher table byte limit"); + let mut reader = Cursor::new(bytes); + let legacy: LegacyTable = codec(MAX_TABLE_BYTES) + .allow_trailing_bytes() + .deserialize_from(&mut reader)?; + let advanced = if reader.position() as usize == bytes.len() { + None + } else { + let (tag, advanced): ([u8; 8], Arc) = + codec(advance::MAX_ADVANCED_BYTES).deserialize(&bytes[reader.position() as usize..])?; + ensure!(tag == EXTENSION, "unknown table extension"); + Some(advanced) + }; + let table = Table { + manifests: legacy.manifests, + destination: legacy.destination, + ingest: legacy.ingest, + authority: legacy.authority, + request: legacy.request, + progress: Progress { + checkpoint: legacy.progress.checkpoint, + minimum_read_version: legacy.progress.minimum_read_version, + pending: legacy.progress.pending, + prepared: legacy.progress.prepared, + published: legacy.progress.published, + advanced, + }, + }; + ensure!(encode(&table)? == bytes, "noncanonical publisher table"); + Ok(table) +} diff --git a/crates/lance-context-wal/src/replication/source/closure.rs b/crates/lance-context-wal/src/replication/source/closure.rs index 8445d73..92f638a 100644 --- a/crates/lance-context-wal/src/replication/source/closure.rs +++ b/crates/lance-context-wal/src/replication/source/closure.rs @@ -21,6 +21,27 @@ pub struct Closure { pub published_index: u64, } +impl Closure { + pub(in crate::replication) fn matches_publication( + &self, + selection: &Selection, + published: &publication::Published, + ) -> Result { + Ok(self.table == selection.table + && self.transaction == selection.transaction + && self.plan_index == selection.plan_index + && self.group == selection.group + && self.after == selection.range.after + && self.through == selection.range.through + && self.selection_digest == selection_digest(selection)? + && self.incarnation == published.incarnation + && self.manifest_identity == published.manifest.identity + && self.version == published.manifest.version + && self.manifest_index == published.manifest.index + && self.published_index == published.index) + } +} + fn selection_digest(selection: &Selection) -> Result<[u8; 32]> { Ok(Sha256::digest( codec(MAX_BYTES).serialize(&("lance-context/source-selection/v1", selection))?, diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index a2d774f..e4d9dd2 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -62,6 +62,10 @@ impl WirePacket { } #[derive(Serialize, Deserialize)] enum Request { + AdvancePublication( + u64, + lance_context_wal::replication::publication::AdvancePublication, + ), CloseSource(u64, lance_context_wal::replication::source::ClosurePermit), ConfirmPublication( u64, @@ -145,6 +149,7 @@ fn run(worker: &mut Worker, request: Request) -> Result { match request { Request::CloseSource(token, permit) => node.close_source(token, permit)?, Request::ConfirmPublication(token, request) => node.confirm_publication(token, request)?, + Request::AdvancePublication(token, request) => node.advance_publication(token, request)?, Request::BootstrapManifest(token, request) => node.bootstrap_manifest(token, request)?, Request::ReserveManifest(token, request) => node.reserve_manifest(token, request)?, Request::Prepare(token, prepared) => node.prepare_publication(token, prepared)?, diff --git a/crates/lance-context-wal/tests/replication_process/source_closure.rs b/crates/lance-context-wal/tests/replication_process/source_closure.rs index 8cb86a3..ae4b3d4 100644 --- a/crates/lance-context-wal/tests/replication_process/source_closure.rs +++ b/crates/lance-context-wal/tests/replication_process/source_closure.rs @@ -1,6 +1,7 @@ //! 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::publication::AdvancePublication; use lance_context_wal::replication::source::{ClosedProof, Rejection as SourceRejection}; fn proof( @@ -198,5 +199,92 @@ pub(super) fn close_and_restart( assert!( matches!(c.peers.get_mut(&source).unwrap().call(Request::Page(20))?, Response::Page(page) if page.entries.is_empty()) ); + advance_and_restart(&mut c, table, *recovered_proof)?; + Ok(()) +} + +fn advance_and_restart(c: &mut Cluster, table: Uuid, proof: ClosedProof) -> Result<()> { + let catalog = c.elect(1, 10, &[1, 2, 3])?; + c.command( + catalog, + Request::Query(10, 610_000, Query::Publisher { table }), + )?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + value: Value::Publisher(Some(view)), + .. + }) = c.read_outcome(610_000) + else { + anyhow::bail!("catalog before advancement missing") + }; + let published = view.progress.published.context("publication missing")?; + let through = proof.closed().permit.closure().through; + let request = AdvancePublication { + authority: view.authority, + published_index: published.index, + proofs: BTreeMap::from([(20, proof)]), + }; + c.command( + catalog, + Request::AdvancePublication(610_001, request.clone()), + )?; + c.pump(&[1, 2, 3])?; + let Some(Outcome::Committed(Applied::Publisher(PublisherReply::Advanced(advanced)))) = c + .completions + .iter() + .find(|completion| completion.request == 610_001) + .map(|completion| completion.outcome.clone()) + else { + anyhow::bail!("catalog advancement missing") + }; + assert_eq!(advanced.publication, published); + assert_eq!(advanced.sources, BTreeMap::from([(20, through)])); + // Lose the completion, checkpoint the catalog, then restart every process. + let before = c.completions.len(); + c.completions + .retain(|completion| completion.request != 610_001); + assert_eq!(before - c.completions.len(), 1); + c.peers + .get_mut(&catalog) + .unwrap() + .call(Request::PrepareCheckpoint(10, Uuid::new_v4()))?; + c.command(catalog, Request::InstallCheckpoint)?; + c.pump(&[1, 2, 3])?; + c.queue.clear(); + c.peers.clear(); + for id in 1..=6 { + c.start(id)?; + } + let catalog = c.elect(1, 10, &[1, 2, 3])?; + c.command(catalog, Request::AdvancePublication(610_002, request))?; + c.command( + catalog, + Request::Query(10, 610_003, Query::Publisher { table }), + )?; + c.pump(&[1, 2, 3])?; + assert_eq!( + c.outcome(610_002), + Some(Outcome::Committed(Applied::Publisher( + PublisherReply::Advanced(advanced.clone()) + ))) + ); + let Some(ReadOutcome::Observed { + value: Value::Publisher(Some(view)), + .. + }) = c.read_outcome(610_003) + else { + anyhow::bail!("catalog advancement lost after restart") + }; + assert_eq!(view.progress.advanced, Some(advanced)); + assert_eq!( + view.progress.minimum_read_version, + published.manifest.version + ); + assert_eq!(view.progress.checkpoint, BTreeMap::from([(20, through)])); + assert!( + view.progress.pending.is_none() + && view.progress.prepared.is_none() + && view.progress.published.is_none() + ); Ok(()) } diff --git a/docs/design/wal-catalog-advancement.md b/docs/design/wal-catalog-advancement.md new file mode 100644 index 0000000..92d8fbf --- /dev/null +++ b/docs/design/wal-catalog-advancement.md @@ -0,0 +1,50 @@ +# Catalog advancement after source closure + +Tracking: [#378](https://github.com/lance-format/lance-context/issues/378). + +After a managed Lance publication, source groups can durably close their exact +pins and attest those closures. `AdvancePublication` moves the catalog past that +publication only when every selected source supplies its matching `ClosedProof`. +The command verifies both signatures, the table and publication identity, the +complete source set, each independent range, and the digest of the exact source +selection. A valid proof for a different selection is insufficient. Missing or +mismatched proofs leave the entire catalog unchanged. + +One catalog Raft command updates each source checkpoint to that source's +`through` position, raises the minimum read version to the published manifest, +and clears the pending plan, prepared descriptors and unresolved outcome. Source +indices are never collapsed to a maximum or compared across independent groups. +The next publication can then select new records. Source closure and catalog +advancement remain separate durable transitions, so a materializer can resume +after any intermediate failure. + +One bounded `Advanced` receipt per table retains the completed source-set digest, +source frontiers and shared publication record. It contains no payloads or +prepared fragment descriptors. Exact retries return the original completion even +after publisher takeover or while the next plan is pending. Refreshed proofs for +the same closures have the same request identity; their read cuts do not change +the receipt. Only the latest advancement receipt is retained. New state changes +still require the current publisher authority. Publication confirmation retries +can also resolve this latest completed outcome. + +Application checkpoint table frames retain their exact previous bytes until an +advancement exists. A bounded `LCWALAD1` suffix then carries the receipt. New code +reads the old frame and the extension; unknown, truncated and trailing suffix +data are rejected. The metadata budget includes one receipt per table. This is +forward migration support, not mixed-binary operation: upgrade all replicas +before submitting the new command, and do not downgrade an advanced directory. +The serialized public `Table` query value gains a field; arbitrary old external +bincode consumers are not compatible. Existing replication policy identities +and tables without advancement keep their previous checkpoint representation. + +Advancement does not delete log files or Lance objects. Physical collection still +requires a durable application checkpoint and the existing file-lifetime rules. +Catalog ownership, transport authentication and source-key protection remain +deployment responsibilities. + +Validation is in progress. Rust tests cover all-source atomicity, independent +source cursors, stale owners, refreshed proofs, missing or wrong signed proofs, +next-plan admission and strict legacy-frame compatibility. Real TCP child-process +tests extend actual Lance publication recovery with a lost advancement reply, +catalog checkpoint, all-replica restart and exact completion lookup. Passing +results and public PR qualification are required before this change is ready.