diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index a57944a..44ced10 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/v6", + "lance-context/replicated-append/v7", self, ))?) } @@ -439,6 +439,17 @@ impl Node { self.propose(publication::control_group(&self.layout), request, batch) } + /// Durably resolve the selected manifest against its exact pending plan. + /// This retains all source dependencies; it does not authorize reclamation. + pub fn confirm_publication( + &mut self, + request: u64, + confirmation: publication::ConfirmPublication, + ) -> Result<()> { + let batch = confirmation.batch(&self.layout)?; + 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 4ea4b87..d0a9736 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 = 6; -const MAGIC: &[u8; 8] = b"LCWALCP6"; +const FORMAT: u32 = 7; +const MAGIC: &[u8; 8] = b"LCWALCP7"; pub const CHUNK_BYTES: usize = 1024 * 1024; fn codec() -> impl Options { @@ -85,7 +85,7 @@ fn metadata_limit(limits: &Limits) -> u64 { (limits.max_records + limits.max_producers) * 268 + limits.max_tables * (publication::MAX_BASE_TABLE_BYTES as u64 + 4) + limits.max_tables.min(limits.max_active_plans) - * (128 * 1024 + limits.max_descriptor_bytes + 1024) + * (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 diff --git a/crates/lance-context-wal/src/replication/publication.rs b/crates/lance-context-wal/src/replication/publication.rs index 6da1783..254613b 100644 --- a/crates/lance-context-wal/src/replication/publication.rs +++ b/crates/lance-context-wal/src/replication/publication.rs @@ -1,19 +1,21 @@ //! 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. //! Immutable plans and prepared descriptors survive takeover. Source pins live -//! in the source module; manifest outcomes and retirement remain separate work. +//! in the source module; source closure 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 manifest; +mod outcome; mod plan; mod prepared; pub use manifest::{ BootstrapManifest, ManifestIndex, ManifestObject, ManifestRecord, ReserveManifest, MAX_MANIFEST_VERSIONS, }; +pub use outcome::{ConfirmPublication, Published}; pub use plan::{PendingPlan, Plan, PlanRequest, Progress, SourceRange, MAX_INPUTS}; pub use prepared::{Descriptors, PrepareRequest, Prepared, MAX_DESCRIPTOR_BYTES}; #[cfg(test)] @@ -25,7 +27,8 @@ 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 + prepared::MAX_PREPARE_BYTES - + manifest::MAX_MANIFEST_INDEX_BYTES; + + manifest::MAX_MANIFEST_INDEX_BYTES + + outcome::MAX_OUTCOME_BYTES; fn codec(limit: usize) -> impl Options { bincode::DefaultOptions::new() .with_fixint_encoding() @@ -76,6 +79,7 @@ enum Command { Prepare(Box), BootstrapManifest(Box), ReserveManifest(Box), + ConfirmPublication(Box), } impl Command { fn identity(&self) -> (Uuid, Uuid, u64) { @@ -85,6 +89,9 @@ impl Command { Self::Prepare(p) => (p.authority.table, p.authority.owner, p.authority.epoch), Self::BootstrapManifest(p) => (p.authority.table, p.authority.owner, p.authority.epoch), Self::ReserveManifest(p) => (p.authority.table, p.authority.owner, p.authority.epoch), + Self::ConfirmPublication(p) => { + (p.authority.table, p.authority.owner, p.authority.epoch) + } } } fn byte_limit(&self) -> usize { @@ -95,13 +102,14 @@ impl Command { Self::BootstrapManifest(_) | Self::ReserveManifest(_) => { manifest::MAX_MANIFEST_COMMAND_BYTES } + Self::ConfirmPublication(_) => outcome::MAX_OUTCOME_BYTES, } } fn batch(self, layout: &Layout) -> Result { let (table, owner, sequence) = self.identity(); let limit = self.byte_limit(); let envelope = Envelope { - version: 4, + version: 5, cluster: layout.cluster, layout: layout.fingerprint()?, group: control_group(layout), @@ -168,6 +176,7 @@ pub enum Rejection { PreparedConflict, ManifestIdentity, ManifestPredecessor, + PublicationIdentity, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Reply { @@ -177,6 +186,7 @@ pub enum Reply { Prepared { index: u64 }, ManifestReserved(Arc), ManifestTaken(Arc), + Published(Arc), } #[derive(Serialize, Deserialize)] @@ -210,7 +220,7 @@ pub(super) fn decode(bytes: &[u8]) -> Result { "publisher command too large" ); let envelope: Envelope = codec(MAX_COMMAND_BYTES).deserialize(bytes)?; - ensure!(envelope.version == 4, "unsupported publisher command"); + ensure!(envelope.version == 5, "unsupported publisher command"); ensure!( bytes.len() <= envelope.command.byte_limit(), "publisher command type byte limit" @@ -255,6 +265,7 @@ pub(super) fn validate_record( Command::Prepare(p) => p.validate(layout, limits)?, Command::BootstrapManifest(p) => p.validate(layout)?, Command::ReserveManifest(p) => p.validate(layout)?, + Command::ConfirmPublication(p) => p.validate(layout)?, } Ok(()) } @@ -274,6 +285,9 @@ impl Catalog { Command::ReserveManifest(request) => { return self.reserve_manifest(*request, index, limits) } + Command::ConfirmPublication(request) => { + return self.confirm_publication(*request, index) + } Command::Acquire(a) => *a, }; let old = self.tables.get(&a.table); diff --git a/crates/lance-context-wal/src/replication/publication/manifest.rs b/crates/lance-context-wal/src/replication/publication/manifest.rs index 8bc20f4..c3fb149 100644 --- a/crates/lance-context-wal/src/replication/publication/manifest.rs +++ b/crates/lance-context-wal/src/replication/publication/manifest.rs @@ -3,7 +3,7 @@ use super::*; #[cfg(test)] -mod tests; +pub(super) mod tests; pub const MAX_MANIFEST_VERSIONS: usize = 128; pub(super) const MAX_MANIFEST_BYTES: usize = 2048; diff --git a/crates/lance-context-wal/src/replication/publication/manifest/tests.rs b/crates/lance-context-wal/src/replication/publication/manifest/tests.rs index f6fc318..a9c73db 100644 --- a/crates/lance-context-wal/src/replication/publication/manifest/tests.rs +++ b/crates/lance-context-wal/src/replication/publication/manifest/tests.rs @@ -15,7 +15,9 @@ fn object(version: u64, staging: bool) -> ManifestObject { } } -fn fixture(limits: &Limits) -> Result<(Catalog, BootstrapManifest, ReserveManifest)> { +pub(in crate::replication::publication) fn fixture( + limits: &Limits, +) -> Result<(Catalog, BootstrapManifest, ReserveManifest)> { let layout = crate::replication::tests::layout(); let mut catalog = Catalog::default(); let acquire = super::super::tests::input(); diff --git a/crates/lance-context-wal/src/replication/publication/outcome.rs b/crates/lance-context-wal/src/replication/publication/outcome.rs new file mode 100644 index 0000000..9e734c2 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/outcome.rs @@ -0,0 +1,152 @@ +//! Durable publication resolution from the catalog's selected manifest. Source +//! selection and descriptors stay retained until a separate closure protocol. +use super::*; + +#[cfg(test)] +mod tests; + +pub(super) const MAX_OUTCOME_BYTES: usize = manifest::MAX_MANIFEST_BYTES + 128; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ConfirmPublication { + pub authority: Authority, + pub incarnation: Uuid, + pub transaction: Uuid, + pub plan_index: u64, + pub prepared_index: u64, + pub descriptor_digest: [u8; 32], + pub target: u64, + pub manifest_identity: Uuid, +} +impl ConfirmPublication { + pub(in crate::replication) fn batch(self, layout: &Layout) -> Result { + self.validate(layout)?; + Command::ConfirmPublication(Box::new(self)).batch(layout) + } + + pub(super) fn validate(&self, layout: &Layout) -> Result<()> { + self.authority.validate(layout)?; + ensure!( + !self.incarnation.is_nil() + && !self.transaction.is_nil() + && !self.manifest_identity.is_nil() + && self.plan_index > 0 + && self.prepared_index > self.plan_index + && self.prepared_index < u64::MAX + && self.descriptor_digest != [0; 32] + && self.target > 1 + && self.target < (1u64 << 63), + "invalid publication confirmation identity" + ); + Ok(()) + } + + fn matches(&self, incarnation: Uuid, manifest: &ManifestRecord) -> bool { + self.incarnation == incarnation + && self.transaction == manifest.transaction + && self.plan_index == manifest.plan_index + && self.prepared_index == manifest.prepared_index + && self.descriptor_digest == manifest.descriptor_digest + && self.target == manifest.version + && self.manifest_identity == manifest.identity + } +} + +/// One outcome per unresolved plan. This shares the immutable selected record; +/// cloning a table view does not copy descriptors or any source payloads. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Published { + pub index: u64, + pub incarnation: Uuid, + pub manifest: Arc, +} +impl Published { + pub(super) fn validate(&self, table: &Table, cut: u64) -> Result<()> { + let manifests = table + .manifests + .as_ref() + .context("outcome without manifest index")?; + let pending = table + .progress + .pending + .as_ref() + .context("outcome without pending selection")?; + let prepared = table + .progress + .prepared + .as_ref() + .context("outcome without prepared descriptors")?; + let selected = manifests + .versions + .get(&self.manifest.version) + .context("outcome manifest missing")?; + ensure!( + self.incarnation == manifests.incarnation + && self.manifest == *selected + && self.manifest.version == manifests.head + && self.manifest.version == pending.plan.target_version()? + && self.manifest.transaction == pending.plan.transaction + && self.manifest.plan_index == pending.index + && self.manifest.prepared_index == prepared.index + && self.manifest.descriptor_digest == prepared.descriptors.digest + && self.index > self.manifest.index + && self.index <= cut + && self.index != table.authority.grant_index, + "publication outcome disagrees with selected manifest or pending plan" + ); + Ok(()) + } +} + +impl Catalog { + pub(super) fn confirm_publication(&mut self, request: ConfirmPublication, index: u64) -> Reply { + let Some(table) = self.tables.get(&request.authority.table) else { + return Reply::Rejected(Rejection::Authority); + }; + // 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 { + return if request.matches(published.incarnation, &published.manifest) { + Reply::Published(published.clone()) + } else { + Reply::Rejected(Rejection::PublicationIdentity) + }; + } + if request.authority != table.authority { + return Reply::Rejected(Rejection::Authority); + } + let Some(manifests) = &table.manifests else { + return Reply::Rejected(Rejection::ManifestIdentity); + }; + let Some(selected) = manifests.versions.get(&request.target) else { + return Reply::Rejected(Rejection::ManifestIdentity); + }; + let (Some(pending), Some(prepared)) = (&table.progress.pending, &table.progress.prepared) + else { + return Reply::Rejected(Rejection::PlanIdentity); + }; + if !request.matches(manifests.incarnation, selected) + || selected.version != manifests.head + || pending.plan.target_version().ok() != Some(selected.version) + || selected.transaction != pending.plan.transaction + || selected.plan_index != pending.index + || selected.prepared_index != prepared.index + || selected.descriptor_digest != prepared.descriptors.digest + { + return Reply::Rejected(Rejection::PublicationIdentity); + } + let published = Arc::new(Published { + index, + incarnation: manifests.incarnation, + manifest: selected.clone(), + }); + // Do not erase the selection, advance source cursors, or release pins. + // Closure must preserve independent source indices and durable fences. + self.tables + .get_mut(&request.authority.table) + .expect("known table") + .progress + .published = Some(published.clone()); + Reply::Published(published) + } +} diff --git a/crates/lance-context-wal/src/replication/publication/outcome/tests.rs b/crates/lance-context-wal/src/replication/publication/outcome/tests.rs new file mode 100644 index 0000000..151f6dd --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/outcome/tests.rs @@ -0,0 +1,232 @@ +use super::*; + +fn fixture() -> Result<(Catalog, ConfirmPublication)> { + let limits = Limits::default(); + let (mut catalog, _, reserve) = super::super::manifest::tests::fixture(&limits)?; + let Reply::ManifestReserved(selected) = catalog.reserve_manifest(reserve.clone(), 8, &limits) + else { + anyhow::bail!("manifest not selected"); + }; + Ok(( + catalog, + 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, + }, + )) +} + +fn confirm(catalog: &mut Catalog, request: ConfirmPublication, index: u64) -> Result { + let layout = crate::replication::tests::layout(); + let limits = Limits::default(); + let batch = request.batch(&layout)?; + validate_record(&layout, 10, &limits, &batch.meta, &batch.body)?; + Ok(catalog.apply(decode(&batch.body)?, index, &limits)) +} + +fn takeover(catalog: &mut Catalog, index: u64) -> Result { + let layout = crate::replication::tests::layout(); + let mut request = super::super::tests::input(); + request.owner = Uuid::from_u128(500); + request.request = Uuid::from_u128(501); + request.expected_epoch = 1; + let Reply::Acquired(authority) = catalog.apply( + decode(&request.batch(&layout)?.body)?, + index, + &Limits::default(), + ) else { + anyhow::bail!("takeover failed") + }; + Ok(authority) +} + +#[test] +fn published_resolution_is_immutable_shared_and_preserves_source_dependencies() -> Result<()> { + let (mut catalog, request) = fixture()?; + let captured = catalog.tables[&request.authority.table].clone(); + let Reply::Published(published) = confirm(&mut catalog, request.clone(), 9)? else { + anyhow::bail!("publication confirmation missing"); + }; + assert_eq!(published.index, 9); + let table = &catalog.tables[&request.authority.table]; + assert!(Arc::ptr_eq( + &published.manifest, + &table.manifests.as_ref().unwrap().versions[&8] + )); + assert!(Arc::ptr_eq( + table.progress.pending.as_ref().unwrap(), + captured.progress.pending.as_ref().unwrap() + )); + assert!(Arc::ptr_eq( + table.progress.prepared.as_ref().unwrap(), + captured.progress.prepared.as_ref().unwrap() + )); + assert!(captured.progress.published.is_none()); + assert!(table.progress.checkpoint.is_empty()); + assert_eq!(table.progress.minimum_read_version, 1); + // A source cursor is independent of this catalog's confirmation index. + assert_eq!( + table.progress.pending.as_ref().unwrap().plan.sources[&20].through, + 20 + ); + let new_authority = takeover(&mut catalog, 10)?; + assert_eq!( + confirm(&mut catalog, request.clone(), 11)?, + Reply::Published(published.clone()) + ); + let mut retry = request.clone(); + retry.authority = new_authority.clone(); + assert_eq!( + confirm(&mut catalog, retry, 12)?, + Reply::Published(published.clone()) + ); + assert!(Arc::ptr_eq( + catalog.tables[&request.authority.table] + .progress + .published + .as_ref() + .unwrap(), + &published + )); + let mut next = captured.progress.pending.as_ref().unwrap().plan.clone(); + next.transaction = Uuid::new_v4(); + next.read_version = 8; + let layout = crate::replication::tests::layout(); + let limits = Limits::default(); + let batch = PlanRequest { + authority: new_authority, + plan: next, + } + .batch(&layout, &limits)?; + assert_eq!( + catalog.apply(decode(&batch.body)?, 13, &limits), + Reply::Rejected(Rejection::PendingPlan) + ); + catalog.validate(&layout, 10, 13, &limits)?; + Ok(()) +} + +#[test] +fn confirmation_requires_current_owner_and_exact_selected_manifest() -> Result<()> { + let (catalog, request) = fixture()?; + for mutation in 0..9 { + let mut altered = request.clone(); + let expected = match mutation { + 0 => { + altered.authority.owner = Uuid::new_v4(); + Rejection::Authority + } + 1 => { + altered.incarnation = Uuid::new_v4(); + Rejection::PublicationIdentity + } + 2 => { + altered.manifest_identity = Uuid::new_v4(); + Rejection::PublicationIdentity + } + 3 => { + altered.transaction = Uuid::new_v4(); + Rejection::PublicationIdentity + } + 4 => { + altered.plan_index -= 1; + Rejection::PublicationIdentity + } + 5 => { + altered.prepared_index += 1; + Rejection::PublicationIdentity + } + 6 => { + altered.descriptor_digest[0] ^= 1; + Rejection::PublicationIdentity + } + 7 => { + altered.target += 1; + Rejection::ManifestIdentity + } + 8 => { + altered.authority.table = Uuid::new_v4(); + Rejection::Authority + } + _ => unreachable!(), + }; + let mut changed = catalog.clone(); + assert_eq!( + confirm(&mut changed, altered.clone(), 9)?, + Reply::Rejected(expected) + ); + assert_eq!(changed, catalog); + confirm(&mut changed, request.clone(), 10)?; + // Old-owner exact outcome lookup is allowed; another identity never is. + if mutation != 0 { + let before = changed.clone(); + assert!(matches!( + confirm(&mut changed, altered, 11)?, + Reply::Rejected(_) + )); + assert_eq!(before, changed); + } + } + let mut changed = catalog.clone(); + let authority = takeover(&mut changed, 9)?; + assert_eq!( + confirm(&mut changed, request.clone(), 10)?, + Reply::Rejected(Rejection::Authority) + ); + let mut new = request.clone(); + new.authority = authority; + assert!(matches!( + confirm(&mut changed, new, 11)?, + Reply::Published(_) + )); + let limits = Limits::default(); + let (mut unselected, _, _) = super::super::manifest::tests::fixture(&limits)?; + let before = unselected.clone(); + assert_eq!( + confirm(&mut unselected, request, 9)?, + Reply::Rejected(Rejection::ManifestIdentity) + ); + assert_eq!(unselected, before); + Ok(()) +} + +#[test] +fn checkpoint_rejects_damaged_outcomes_or_missing_retained_dependencies() -> Result<()> { + let (mut catalog, request) = fixture()?; + confirm(&mut catalog, request.clone(), 9)?; + let layout = crate::replication::tests::layout(); + let limits = Limits::default(); + catalog.validate(&layout, 10, 9, &limits)?; + let table = &catalog.tables[&request.authority.table]; + assert_eq!(decode_table(&encode_table(table)?)?, *table); + for mutation in 0..9 { + let mut changed = catalog.clone(); + let table = changed.tables.get_mut(&request.authority.table).unwrap(); + let outcome = Arc::make_mut(table.progress.published.as_mut().unwrap()); + match mutation { + 0 => outcome.index = outcome.manifest.index, + 1 => outcome.index = 10, + 2 => outcome.index = table.authority.grant_index, + 3 => outcome.incarnation = Uuid::new_v4(), + 4 => Arc::make_mut(&mut outcome.manifest).identity = Uuid::new_v4(), + 5 => table.progress.pending = None, + 6 => table.progress.prepared = None, + 7 => table.manifests = None, + 8 => { + table.progress.checkpoint.insert(20, 20); + } + _ => unreachable!(), + } + assert!( + changed.validate(&layout, 10, 9, &limits).is_err(), + "mutation {mutation}" + ); + } + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/publication/plan.rs b/crates/lance-context-wal/src/replication/publication/plan.rs index dcb966d..4e4284a 100644 --- a/crates/lance-context-wal/src/replication/publication/plan.rs +++ b/crates/lance-context-wal/src/replication/publication/plan.rs @@ -144,6 +144,8 @@ pub struct Progress { pub pending: Option>, /// Materialization result shares ownership independently of the selection. pub prepared: Option>, + /// Durable selected-manifest resolution; sources remain pinned for closure. + pub published: Option>, } impl Default for Progress { fn default() -> Self { @@ -152,6 +154,7 @@ impl Default for Progress { minimum_read_version: 1, pending: None, prepared: None, + published: None, } } } @@ -174,6 +177,9 @@ impl Progress { cut: u64, limits: &Limits, ) -> Result<()> { + if let Some(published) = &self.published { + published.validate(table, cut)?; + } ensure!( self.minimum_read_version > 0 && self.checkpoint.len() <= layout.groups.len() diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index 4adc9eb..f9d1cfe 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 { + ConfirmPublication( + u64, + lance_context_wal::replication::publication::ConfirmPublication, + ), BootstrapManifest( u64, lance_context_wal::replication::publication::BootstrapManifest, @@ -138,6 +142,7 @@ fn part(payload: bool) -> Part { fn run(worker: &mut Worker, request: Request) -> Result { let node = &mut worker.node; match request { + 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)?, Request::Prepare(token, prepared) => node.prepare_publication(token, prepared)?, 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 b3f8afa..f68f0da 100644 --- a/crates/lance-context-wal/tests/replication_process/managed_manifest.rs +++ b/crates/lance-context-wal/tests/replication_process/managed_manifest.rs @@ -2,7 +2,8 @@ //! This qualifies the durable index, not production transport or coordinated GC. use super::*; use lance_context_wal::replication::publication::{ - Authority, BootstrapManifest, ManifestObject, ManifestRecord, Prepared, ReserveManifest, + Authority, BootstrapManifest, ConfirmPublication, ManifestObject, ManifestRecord, Prepared, + ReserveManifest, }; use object_store::path::Path as ObjectPath; use std::sync::{ @@ -330,9 +331,30 @@ pub(super) async fn publish_and_recover(c: Cluster, attempt: Attempt, mode: Mode } let selected = handler.read(Some(2))?.context("missing committed target")?; assert!(selected.object.path.contains(".manifest-")); + let confirmation = ConfirmPublication { + authority: handler.scope.authority.clone(), + incarnation: handler.scope.incarnation, + transaction: selected.transaction, + plan_index: selected.plan_index, + prepared_index: selected.prepared_index, + descriptor_digest: selected.descriptor_digest, + target: selected.version, + manifest_identity: selected.identity, + }; { let mut c = cluster.lock().unwrap(); let leader = c.elect(1, 10, &[1, 2, 3])?; + c.command( + leader, + Request::ConfirmPublication(500_000, confirmation.clone()), + )?; + c.pump(&[1, 2, 3])?; + // Lose the confirmation reply before any caller consumes it. Recovery + // must discover the outcome from durable state, including its index. + let before = c.completions.len(); + c.completions + .retain(|completion| completion.request != 500_000); + assert_eq!(before - c.completions.len(), 1); c.peers .get_mut(&leader) .unwrap() @@ -353,6 +375,54 @@ pub(super) async fn publish_and_recover(c: Cluster, attempt: Attempt, mode: Mode } } assert_eq!(handler.read(Some(2))?.as_deref(), Some(selected.as_ref())); + { + let mut c = cluster.lock().unwrap(); + let leader = c.elect(1, 10, &[1, 2, 3])?; + c.command( + leader, + Request::Query( + 10, + 500_001, + Query::Publisher { + table: confirmation.authority.table, + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + value: Value::Publisher(Some(view)), + .. + }) = c.read_outcome(500_001) + else { + anyhow::bail!("publication outcome query missing"); + }; + let published = view + .progress + .published + .context("durable outcome missing after restart")?; + assert_eq!(published.manifest, selected); + assert!(published.index > selected.index); + assert_eq!( + view.progress + .pending + .context("source selection lost")? + .index, + selected.plan_index + ); + assert_eq!( + view.progress.prepared.context("descriptors lost")?, + attempt.prepared + ); + assert!(view.progress.checkpoint.is_empty()); + c.command(leader, Request::ConfirmPublication(500_002, confirmation))?; + c.pump(&[1, 2, 3])?; + assert_eq!( + c.outcome(500_002), + Some(Outcome::Committed(Applied::Publisher( + PublisherReply::Published(published) + ))) + ); + } // 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/docs/design/wal-manifest-authority.md b/docs/design/wal-manifest-authority.md index 784cec7..a21d5f5 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 v6 and include the index in bounds and integrity validation. +format advance to v7 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 795f1bd..b4e7e47 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 v6; older identities +descriptors. The application/checkpoint format advances to v7; 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 new file mode 100644 index 0000000..631f351 --- /dev/null +++ b/docs/design/wal-publication-outcomes.md @@ -0,0 +1,39 @@ +# Durable publication outcomes + +The replicated catalog can now confirm that its selected immutable manifest +belongs to the exact pending plan and prepared descriptors. This records one +`Published` outcome through the catalog quorum. It provides the durable input +for source closure; closure and reclamation are not enabled by this increment. + +`Node::confirm_publication` binds the dataset incarnation, transaction, plan and +preparation indices, descriptor digest, fixed target version and selected +manifest identity. The first confirmation requires the current publisher owner. +A mismatched or missing manifest cannot be converted into a successful outcome. +An identical retry resolves the original outcome and confirmation index even +after publisher takeover. This resolution grants no new write authority. + +The outcome shares the immutable manifest record through `Arc`. Pending source +selection and prepared descriptor buffers retain their independent ownership. +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 +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. + +Validation includes rejected forged identities, stale-owner first confirmation, +exact retries after takeover, retained source dependencies, damaged checkpoint +state, and six-process tests with real Lance files. Those process tests discard +the confirmation reply, snapshot the catalog, restart all replicas and resolve +the same outcome before checking the actual transaction and full payloads. +These tests do not establish production integration or sustained throughput. + +The next transition must durably fence closed source plans before releasing +their pins, preserve producer retry receipts after payload retirement, and retain +the selection until all source closures can be recovered. Physical collection +also needs active image/read ownership protection. Manifest history retention +and a production managed commit handler remain separate requirements. + +Refs #378. diff --git a/docs/design/wal-publication-plans.md b/docs/design/wal-publication-plans.md index 7358a38..f470f99 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 v6. Existing v1/v2/v3/v4/v5 directories and images are rejected; there is no in-place migration +format to v7. Existing v1/v2/v3/v4/v5/v6 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 5017360..f27c136 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 v6, including [immutable publication plans](wal-publication-plans.md) +format are v7, 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-pins.md b/docs/design/wal-source-pins.md index cbd8b3d..e7aa604 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 v6 and rejects older identities +The application/checkpoint format advances to v7 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