From 3412b6c366ac86785c1c65c529733bd664bf466d Mon Sep 17 00:00:00 2001 From: Beinan Date: Tue, 20 Oct 2026 18:11:02 +0000 Subject: [PATCH] feat(wal): select immutable manifests through the catalog quorum Refs #378. --- crates/lance-context-wal/src/replication.rs | 27 +- .../src/replication/checkpoint.rs | 6 +- .../src/replication/publication.rs | 44 +- .../src/replication/publication/manifest.rs | 377 +++++++++++++++ .../replication/publication/manifest/tests.rs | 295 ++++++++++++ .../src/replication/publication/plan.rs | 4 + .../src/replication/query.rs | 20 + .../tests/replication_process.rs | 10 + .../replication_process/lance_publication.rs | 47 +- .../replication_process/managed_manifest.rs | 433 ++++++++++++++++++ docs/design/wal-manifest-authority.md | 52 +++ docs/design/wal-prepared-descriptors.md | 2 +- docs/design/wal-publication-plans.md | 2 +- docs/design/wal-publisher-authority.md | 4 +- docs/design/wal-source-pins.md | 2 +- 15 files changed, 1310 insertions(+), 15 deletions(-) create mode 100644 crates/lance-context-wal/src/replication/publication/manifest.rs create mode 100644 crates/lance-context-wal/src/replication/publication/manifest/tests.rs create mode 100644 crates/lance-context-wal/tests/replication_process/managed_manifest.rs create mode 100644 docs/design/wal-manifest-authority.md diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 6b9121a..a57944a 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -50,6 +50,7 @@ pub struct Limits { pub max_active_plans: u64, pub max_plan_source_bytes: u64, pub max_descriptor_bytes: u64, + pub max_manifest_records: u64, pub max_record_bytes: u64, pub max_records: u64, pub max_producers: u64, @@ -64,6 +65,7 @@ impl Default for Limits { max_active_plans: 128, max_plan_source_bytes: 64 * 1024 * 1024, max_descriptor_bytes: 256 * 1024, + max_manifest_records: 4096, max_record_bytes: 8 * 1024 * 1024, max_records: 100_000, max_producers: 16_384, @@ -81,6 +83,7 @@ impl Limits { && (1..=256 * 1024 * 1024).contains(&self.max_plan_source_bytes) && (1..=publication::MAX_DESCRIPTOR_BYTES as u64) .contains(&self.max_descriptor_bytes) + && (1..=16_384).contains(&self.max_manifest_records) && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes) && (1..=500_000).contains(&self.max_records) && (1..=100_000).contains(&self.max_producers) @@ -90,7 +93,7 @@ impl Limits { "invalid replication limits" ); Ok(bincode::serialize(&( - "lance-context/replicated-append/v5", + "lance-context/replicated-append/v6", self, ))?) } @@ -414,6 +417,28 @@ impl Node { self.propose(publication::control_group(&self.layout), request, batch) } + /// Adopt an immutable starting manifest after the caller has verified it + /// and excluded unmanaged writers/cleaners for this dataset incarnation. + pub fn bootstrap_manifest( + &mut self, + request: u64, + bootstrap: publication::BootstrapManifest, + ) -> Result<()> { + let batch = bootstrap.batch(&self.layout)?; + self.propose(publication::control_group(&self.layout), request, batch) + } + + /// Select the unique manifest path for the exact immutable prepared plan. + /// An enqueue is not publication; wait for durable quorum completion. + pub fn reserve_manifest( + &mut self, + request: u64, + reservation: publication::ReserveManifest, + ) -> Result<()> { + let batch = reservation.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 ae9103b..4ea4b87 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 = 5; -const MAGIC: &[u8; 8] = b"LCWALCP5"; +const FORMAT: u32 = 6; +const MAGIC: &[u8; 8] = b"LCWALCP6"; pub const CHUNK_BYTES: usize = 1024 * 1024; fn codec() -> impl Options { @@ -87,6 +87,8 @@ fn metadata_limit(limits: &Limits) -> u64 { + limits.max_tables.min(limits.max_active_plans) * (128 * 1024 + limits.max_descriptor_bytes + 1024) + limits.max_tables.min(limits.max_active_plans) * (source::MAX_PIN_BYTES as u64 + 4) + + limits.max_manifest_records * 2048 + + limits.max_tables * 256 + 1024 } diff --git a/crates/lance-context-wal/src/replication/publication.rs b/crates/lance-context-wal/src/replication/publication.rs index c1a5aae..6da1783 100644 --- a/crates/lance-context-wal/src/replication/publication.rs +++ b/crates/lance-context-wal/src/replication/publication.rs @@ -7,8 +7,13 @@ use super::*; use crate::{ingest_contract::IngestContract, record::RecordKind}; use bincode::Options; use bytes::Bytes; +mod manifest; mod plan; mod prepared; +pub use manifest::{ + BootstrapManifest, ManifestIndex, ManifestObject, ManifestRecord, ReserveManifest, + MAX_MANIFEST_VERSIONS, +}; pub use plan::{PendingPlan, Plan, PlanRequest, Progress, SourceRange, MAX_INPUTS}; pub use prepared::{Descriptors, PrepareRequest, Prepared, MAX_DESCRIPTOR_BYTES}; #[cfg(test)] @@ -17,8 +22,10 @@ mod tests; pub(super) const MAX_ACQUIRE_BYTES: usize = 8192; pub(super) const MAX_COMMAND_BYTES: usize = prepared::MAX_PREPARE_BYTES; pub(super) const MAX_BASE_TABLE_BYTES: usize = MAX_ACQUIRE_BYTES + 2048; -pub(super) const MAX_TABLE_BYTES: usize = - MAX_BASE_TABLE_BYTES + plan::MAX_PLAN_BYTES + prepared::MAX_PREPARE_BYTES; +pub(super) const MAX_TABLE_BYTES: usize = MAX_BASE_TABLE_BYTES + + plan::MAX_PLAN_BYTES + + prepared::MAX_PREPARE_BYTES + + manifest::MAX_MANIFEST_INDEX_BYTES; fn codec(limit: usize) -> impl Options { bincode::DefaultOptions::new() .with_fixint_encoding() @@ -67,6 +74,8 @@ enum Command { Acquire(Box), Plan(Box), Prepare(Box), + BootstrapManifest(Box), + ReserveManifest(Box), } impl Command { fn identity(&self) -> (Uuid, Uuid, u64) { @@ -74,6 +83,8 @@ impl Command { Self::Acquire(a) => (a.table, a.owner, a.expected_epoch + 1), Self::Plan(p) => (p.authority.table, p.authority.owner, p.authority.epoch), 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), } } fn byte_limit(&self) -> usize { @@ -81,13 +92,16 @@ impl Command { Self::Acquire(_) => MAX_ACQUIRE_BYTES, Self::Plan(_) => plan::MAX_PLAN_BYTES, Self::Prepare(_) => prepared::MAX_PREPARE_BYTES, + Self::BootstrapManifest(_) | Self::ReserveManifest(_) => { + manifest::MAX_MANIFEST_COMMAND_BYTES + } } } fn batch(self, layout: &Layout) -> Result { let (table, owner, sequence) = self.identity(); let limit = self.byte_limit(); let envelope = Envelope { - version: 3, + version: 4, cluster: layout.cluster, layout: layout.fingerprint()?, group: control_group(layout), @@ -132,6 +146,7 @@ impl Authority { #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Table { + pub manifests: Option>, pub destination: Destination, pub ingest: IngestContract, pub authority: Authority, @@ -151,6 +166,8 @@ pub enum Rejection { SourcePosition, PlanIdentity, PreparedConflict, + ManifestIdentity, + ManifestPredecessor, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Reply { @@ -158,6 +175,8 @@ pub enum Reply { Planned { index: u64 }, Rejected(Rejection), Prepared { index: u64 }, + ManifestReserved(Arc), + ManifestTaken(Arc), } #[derive(Serialize, Deserialize)] @@ -191,7 +210,7 @@ pub(super) fn decode(bytes: &[u8]) -> Result { "publisher command too large" ); let envelope: Envelope = codec(MAX_COMMAND_BYTES).deserialize(bytes)?; - ensure!(envelope.version == 3, "unsupported publisher command"); + ensure!(envelope.version == 4, "unsupported publisher command"); ensure!( bytes.len() <= envelope.command.byte_limit(), "publisher command type byte limit" @@ -234,6 +253,8 @@ pub(super) fn validate_record( ), Command::Plan(p) => p.validate(layout, limits)?, Command::Prepare(p) => p.validate(layout, limits)?, + Command::BootstrapManifest(p) => p.validate(layout)?, + Command::ReserveManifest(p) => p.validate(layout)?, } Ok(()) } @@ -247,6 +268,12 @@ impl Catalog { let a = match e.command { Command::Plan(p) => return self.plan(*p, index, limits), Command::Prepare(request) => return self.prepare(*request, index), + Command::BootstrapManifest(request) => { + return self.bootstrap_manifest(*request, index, limits) + } + Command::ReserveManifest(request) => { + return self.reserve_manifest(*request, index, limits) + } Command::Acquire(a) => *a, }; let old = self.tables.get(&a.table); @@ -286,9 +313,11 @@ impl Catalog { grant_index: index, }; let progress = old.map(|table| table.progress.clone()).unwrap_or_default(); + let manifests = old.and_then(|table| table.manifests.clone()); self.tables.insert( a.table, Table { + manifests, destination: a.destination, ingest: a.ingest, authority: authority.clone(), @@ -299,6 +328,10 @@ impl Catalog { Reply::Acquired(authority) } pub fn validate(&self, layout: &Layout, group: u64, cut: u64, limits: &Limits) -> Result<()> { + ensure!( + self.manifest_records() as u64 <= limits.max_manifest_records, + "manifest record capacity" + ); ensure!( self.tables.len() as u64 <= limits.max_tables, "publisher catalog exceeds limit" @@ -320,6 +353,9 @@ impl Catalog { for (id, table) in &self.tables { validate_binding(&table.destination, &table.ingest)?; table.progress.validate(layout, table, cut, limits)?; + if let Some(manifests) = &table.manifests { + manifests.validate(table, cut)?; + } let a = &table.authority; ensure!( !id.is_nil() diff --git a/crates/lance-context-wal/src/replication/publication/manifest.rs b/crates/lance-context-wal/src/replication/publication/manifest.rs new file mode 100644 index 0000000..8bc20f4 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/manifest.rs @@ -0,0 +1,377 @@ +//! A bounded, durable index of immutable manifests. Only a catalog commit may +//! select a staged path. No canonical-path repair, fallback import or GC occurs. +use super::*; + +#[cfg(test)] +mod tests; + +pub const MAX_MANIFEST_VERSIONS: usize = 128; +pub(super) const MAX_MANIFEST_BYTES: usize = 2048; +pub(super) const MAX_MANIFEST_COMMAND_BYTES: usize = 4096; +pub(super) const MAX_MANIFEST_INDEX_BYTES: usize = MAX_MANIFEST_VERSIONS * MAX_MANIFEST_BYTES + 256; + +/// A path relative to the registered dataset, never a credential-bearing URI. +/// The materializer verifies the actual immutable object before registration. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ManifestObject { + pub path: String, + pub bytes: u64, + pub v2_names: bool, +} +impl ManifestObject { + pub fn validate(&self, version: u64) -> Result<()> { + ensure!( + version > 0 && version < (1u64 << 63), + "invalid attached manifest version" + ); + let name = self + .path + .strip_prefix("_versions/") + .context("manifest path scope")?; + ensure!( + !name.is_empty() + && self.path.len() <= 1024 + && name + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b"-_.".contains(&b)) + && !name.contains("..") + && self.bytes > 0 + && self.bytes <= 64 * 1024 * 1024, + "invalid manifest path or size" + ); + let number = if self.v2_names { + format!("{:020}", u64::MAX - version) + } else { + version.to_string() + }; + let canonical = format!("{number}.manifest"); + // Lance's conditioned handler appends a UUID to the canonical basename. + ensure!( + name == canonical + || name + .strip_prefix(&(canonical + "-")) + .is_some_and(|id| { Uuid::parse_str(id).is_ok_and(|id| !id.is_nil()) }), + "manifest path does not identify its fixed version" + ); + Ok(()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ManifestRecord { + pub version: u64, + /// Logical identity minted by the durable catalog, not an object-store ETag. + pub identity: Uuid, + pub object: ManifestObject, + pub transaction: Uuid, + pub index: u64, + pub plan_index: u64, + pub prepared_index: u64, + pub descriptor_digest: [u8; 32], +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ManifestIndex { + pub incarnation: Uuid, + pub bootstrap_request: Uuid, + pub initial_version: u64, + pub head: u64, + /// Captured views and updates share the immutable records. + pub versions: BTreeMap>, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct BootstrapManifest { + pub authority: Authority, + pub incarnation: Uuid, + pub request: Uuid, + pub version: u64, + pub object: ManifestObject, + pub transaction: Uuid, +} +impl BootstrapManifest { + pub(in crate::replication) fn batch(self, layout: &Layout) -> Result { + self.validate(layout)?; + Command::BootstrapManifest(Box::new(self)).batch(layout) + } + pub(super) fn validate(&self, layout: &Layout) -> Result<()> { + self.authority.validate(layout)?; + ensure!( + !self.incarnation.is_nil() && !self.request.is_nil() && !self.transaction.is_nil(), + "invalid manifest bootstrap identity" + ); + self.object.validate(self.version) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ReserveManifest { + pub authority: Authority, + pub incarnation: Uuid, + pub transaction: Uuid, + pub plan_index: u64, + pub prepared_index: u64, + pub descriptor_digest: [u8; 32], + pub predecessor: Uuid, + pub target: u64, + pub object: ManifestObject, +} +impl ReserveManifest { + pub(in crate::replication) fn batch(self, layout: &Layout) -> Result { + self.validate(layout)?; + Command::ReserveManifest(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.predecessor.is_nil() + && self.plan_index > 0 + && self.prepared_index > self.plan_index + && self.prepared_index < u64::MAX + && self.descriptor_digest != [0; 32], + "invalid manifest reservation identity" + ); + self.object.validate(self.target)?; + ensure!( + self.object.path.contains(".manifest-"), + "reservation needs a unique staging path" + ); + Ok(()) + } +} + +fn identity(authority: &Authority, incarnation: Uuid, version: u64, index: u64) -> Uuid { + let mut hash = Sha256::new(); + hash.update(b"lance-context/manifest-identity/v1"); + hash.update(authority.cluster.as_bytes()); + hash.update(authority.layout); + hash.update(authority.table.as_bytes()); + hash.update(incarnation.as_bytes()); + hash.update(version.to_le_bytes()); + hash.update(index.to_le_bytes()); + Uuid::from_bytes(hash.finalize()[..16].try_into().expect("digest size")) +} + +impl ManifestIndex { + pub(super) fn validate(&self, table: &Table, cut: u64) -> Result<()> { + ensure!( + !self.incarnation.is_nil() + && !self.bootstrap_request.is_nil() + && self.initial_version > 0 + && self.head < u64::MAX + && self.head >= self.initial_version + && self.versions.len() <= MAX_MANIFEST_VERSIONS + && self.versions.len() as u64 == self.head - self.initial_version + 1, + "invalid manifest index bounds" + ); + let mut previous_index = 0; + for (offset, (version, record)) in self.versions.iter().enumerate() { + record.object.validate(*version)?; + ensure!( + *version == self.initial_version + offset as u64 + && record.version == *version + && !record.transaction.is_nil() + && record.index > previous_index + && record.index <= cut + && record.index != table.authority.grant_index + && record.identity + == identity(&table.authority, self.incarnation, *version, record.index), + "invalid manifest record identity or order" + ); + if offset == 0 { + ensure!( + record.plan_index == 0 + && record.prepared_index == 0 + && record.descriptor_digest == [0; 32], + "bootstrap carries publication metadata" + ); + } else { + ensure!( + record.plan_index > 0 + && record.prepared_index > record.plan_index + && record.index > record.prepared_index + && record.descriptor_digest != [0; 32] + && record.object.path.contains(".manifest-"), + "invalid prepared manifest record" + ); + } + previous_index = record.index; + } + if let Some(pending) = &table.progress.pending { + ensure!( + pending.plan.read_version >= self.initial_version + && self.head >= pending.plan.read_version + && self.head <= pending.plan.target_version()?, + "managed plan does not name the current predecessor or target" + ); + if let Some(record) = self.versions.get(&pending.plan.target_version()?) { + let prepared = table + .progress + .prepared + .as_ref() + .context("manifest without descriptors")?; + ensure!( + record.transaction == pending.plan.transaction + && record.plan_index == pending.index + && record.prepared_index == prepared.index + && record.descriptor_digest == prepared.descriptors.digest, + "manifest disagrees with immutable prepared plan" + ); + } + } + Ok(()) + } +} + +impl Catalog { + pub(super) fn manifest_records(&self) -> usize { + self.tables + .values() + .filter_map(|t| t.manifests.as_ref()) + .map(|m| m.versions.len()) + .sum() + } + + pub(super) fn bootstrap_manifest( + &mut self, + request: BootstrapManifest, + index: u64, + limits: &Limits, + ) -> Reply { + let Some(table) = self.tables.get(&request.authority.table) else { + return Reply::Rejected(Rejection::Authority); + }; + if table.authority != request.authority { + return Reply::Rejected(Rejection::Authority); + } + if let Some(manifests) = &table.manifests { + let first = &manifests.versions[&manifests.initial_version]; + return if manifests.incarnation == request.incarnation + && manifests.bootstrap_request == request.request + && first.version == request.version + && first.object == request.object + && first.transaction == request.transaction + { + Reply::ManifestReserved(first.clone()) + } else { + Reply::Rejected(Rejection::ManifestIdentity) + }; + } + if table.progress.pending.is_some() { + return Reply::Rejected(Rejection::PendingPlan); + } + if self.manifest_records() as u64 >= limits.max_manifest_records { + return Reply::Rejected(Rejection::Capacity); + } + let record = Arc::new(ManifestRecord { + version: request.version, + identity: identity( + &request.authority, + request.incarnation, + request.version, + index, + ), + object: request.object, + transaction: request.transaction, + index, + plan_index: 0, + prepared_index: 0, + descriptor_digest: [0; 32], + }); + self.tables + .get_mut(&request.authority.table) + .expect("known table") + .manifests = Some(Arc::new(ManifestIndex { + incarnation: request.incarnation, + bootstrap_request: request.request, + initial_version: request.version, + head: request.version, + versions: BTreeMap::from([(request.version, record.clone())]), + })); + Reply::ManifestReserved(record) + } + + pub(super) fn reserve_manifest( + &mut self, + request: ReserveManifest, + index: u64, + limits: &Limits, + ) -> Reply { + let Some(table) = self.tables.get(&request.authority.table) else { + return Reply::Rejected(Rejection::Authority); + }; + let Some(manifests) = &table.manifests else { + return Reply::Rejected(Rejection::ManifestIdentity); + }; + if manifests.incarnation != request.incarnation { + return Reply::Rejected(Rejection::ManifestIdentity); + } + // Resolve an already chosen path before a new authority/predecessor + // check. A lost successful reply must never let the caller delete the + // selected object as a newly refused staging upload after takeover. + if let Some(record) = manifests.versions.get(&request.target) { + return if record.transaction == request.transaction + && record.plan_index == request.plan_index + && record.prepared_index == request.prepared_index + && record.descriptor_digest == request.descriptor_digest + && record.object == request.object + { + Reply::ManifestReserved(record.clone()) + } else { + Reply::ManifestTaken(record.clone()) + }; + } + if table.authority != request.authority { + return Reply::Rejected(Rejection::Authority); + } + let Some(pending) = &table.progress.pending else { + return Reply::Rejected(Rejection::PlanIdentity); + }; + let Some(prepared) = &table.progress.prepared else { + return Reply::Rejected(Rejection::PlanIdentity); + }; + if pending.index != request.plan_index + || pending.plan.transaction != request.transaction + || prepared.index != request.prepared_index + || prepared.descriptors.digest != request.descriptor_digest + || pending.plan.target_version().ok() != Some(request.target) + { + return Reply::Rejected(Rejection::PlanIdentity); + } + if manifests.head != pending.plan.read_version + || manifests.versions[&manifests.head].identity != request.predecessor + { + return Reply::Rejected(Rejection::ManifestPredecessor); + } + if manifests.versions.len() >= MAX_MANIFEST_VERSIONS + || self.manifest_records() as u64 >= limits.max_manifest_records + { + return Reply::Rejected(Rejection::Capacity); + } + let record = Arc::new(ManifestRecord { + version: request.target, + identity: identity( + &request.authority, + request.incarnation, + request.target, + index, + ), + object: request.object, + transaction: request.transaction, + index, + plan_index: request.plan_index, + prepared_index: request.prepared_index, + descriptor_digest: request.descriptor_digest, + }); + let table = self + .tables + .get_mut(&request.authority.table) + .expect("known table"); + let manifests = Arc::make_mut(table.manifests.as_mut().expect("managed table")); + manifests.head = request.target; + manifests.versions.insert(request.target, record.clone()); + Reply::ManifestReserved(record) + } +} diff --git a/crates/lance-context-wal/src/replication/publication/manifest/tests.rs b/crates/lance-context-wal/src/replication/publication/manifest/tests.rs new file mode 100644 index 0000000..f6fc318 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/manifest/tests.rs @@ -0,0 +1,295 @@ +use super::*; + +fn object(version: u64, staging: bool) -> ManifestObject { + ManifestObject { + path: format!( + "_versions/{version}.manifest{}", + if staging { + format!("-{}", Uuid::new_v4()) + } else { + String::new() + } + ), + bytes: 8192, + v2_names: false, + } +} + +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(); + let Reply::Acquired(authority) = + catalog.apply(decode(&acquire.clone().batch(&layout)?.body)?, 4, limits) + else { + anyhow::bail!("missing authority"); + }; + let bootstrap = BootstrapManifest { + authority: authority.clone(), + incarnation: Uuid::from_u128(400), + request: Uuid::from_u128(401), + version: 7, + object: object(7, false), + transaction: Uuid::from_u128(402), + }; + let Reply::ManifestReserved(first) = + catalog.apply(decode(&bootstrap.clone().batch(&layout)?.body)?, 5, limits) + else { + anyhow::bail!("missing bootstrap"); + }; + let plan = PlanRequest { + authority: authority.clone(), + plan: Plan { + transaction: Uuid::from_u128(403), + read_version: 7, + next_group: 10, + sources: BTreeMap::from([( + 20, + SourceRange { + after: 0, + through: 20, + records: vec![( + 20, + Batch::new( + acquire.table, + 1, + Uuid::from_u128(404), + 1, + Bytes::from_static(b"opaque"), + ) + .meta, + )], + }, + )]), + }, + }; + assert_eq!( + catalog.apply( + decode(&plan.clone().batch(&layout, limits)?.body)?, + 6, + limits + ), + Reply::Planned { index: 6 } + ); + let descriptors = Descriptors::lance_fragments(Bytes::from_static(b"fragments"))?; + let prepare = PrepareRequest { + authority: authority.clone(), + transaction: plan.plan.transaction, + plan_index: 6, + descriptors: descriptors.clone(), + }; + assert_eq!( + catalog.apply(decode(&prepare.batch(&layout, limits)?.body)?, 7, limits), + Reply::Prepared { index: 7 } + ); + let reserve = ReserveManifest { + authority, + incarnation: bootstrap.incarnation, + transaction: plan.plan.transaction, + plan_index: 6, + prepared_index: 7, + descriptor_digest: descriptors.digest, + predecessor: first.identity, + target: 8, + object: object(8, true), + }; + Ok((catalog, bootstrap, reserve)) +} + +fn reserve( + catalog: &mut Catalog, + request: ReserveManifest, + index: u64, + limits: &Limits, +) -> Result { + let layout = crate::replication::tests::layout(); + let batch = request.batch(&layout)?; + validate_record(&layout, 10, limits, &batch.meta, &batch.body)?; + Ok(catalog.apply(decode(&batch.body)?, index, limits)) +} + +#[test] +fn immutable_manifest_identity_survives_takeover_and_resolves_lost_success() -> Result<()> { + let limits = Limits::default(); + let layout = crate::replication::tests::layout(); + let (mut catalog, bootstrap, request) = fixture(&limits)?; + let captured = catalog.tables[&request.authority.table].clone(); + let Reply::ManifestReserved(first) = reserve(&mut catalog, request.clone(), 8, &limits)? else { + anyhow::bail!("reservation missing"); + }; + assert_eq!(first.index, 8); + assert_eq!(captured.manifests.as_ref().unwrap().head, 7); + assert_eq!( + catalog.tables[&request.authority.table] + .manifests + .as_ref() + .unwrap() + .head, + 8 + ); + assert!(Arc::ptr_eq( + &captured.manifests.as_ref().unwrap().versions[&7], + &catalog.tables[&request.authority.table] + .manifests + .as_ref() + .unwrap() + .versions[&7] + )); + let mut acquire = super::super::tests::input(); + acquire.owner = Uuid::from_u128(405); + acquire.request = Uuid::from_u128(406); + acquire.expected_epoch = 1; + let Reply::Acquired(authority) = + catalog.apply(decode(&acquire.batch(&layout)?.body)?, 9, &limits) + else { + anyhow::bail!("takeover missing"); + }; + // Resolve the old successful call, even with its old authority: reporting + // a refusal here could delete the already selected staging object. + assert_eq!( + reserve(&mut catalog, request.clone(), 10, &limits)?, + Reply::ManifestReserved(first.clone()) + ); + let mut retry = request.clone(); + retry.authority = authority; + retry.object = object(8, true); + assert_eq!( + reserve(&mut catalog, retry, 11, &limits)?, + Reply::ManifestTaken(first.clone()) + ); + assert_eq!( + catalog.tables[&request.authority.table] + .manifests + .as_ref() + .unwrap() + .versions + .len(), + 2 + ); + let mut wrong = request.clone(); + wrong.incarnation = Uuid::new_v4(); + assert_eq!( + reserve(&mut catalog, wrong, 12, &limits)?, + Reply::Rejected(Rejection::ManifestIdentity) + ); + catalog.validate(&layout, 10, 12, &limits)?; + let encoded = encode_table(&catalog.tables[&request.authority.table])?; + assert_eq!( + decode_table(&encoded)?, + catalog.tables[&request.authority.table] + ); + assert!(catalog.tables[&request.authority.table] + .progress + .checkpoint + .is_empty()); + for mutation in 0..6 { + let mut invalid = catalog.clone(); + let table = invalid.tables.get_mut(&request.authority.table).unwrap(); + let manifests = Arc::make_mut(table.manifests.as_mut().unwrap()); + match mutation { + 0 => manifests.head += 1, + 1 => { + manifests.versions.remove(&7); + } + 2 => Arc::make_mut(manifests.versions.get_mut(&8).unwrap()).identity = Uuid::new_v4(), + 3 => Arc::make_mut(manifests.versions.get_mut(&8).unwrap()).descriptor_digest[0] ^= 1, + 4 => Arc::make_mut(manifests.versions.get_mut(&8).unwrap()).index = 7, + 5 => manifests.incarnation = Uuid::new_v4(), + _ => unreachable!(), + } + assert!(invalid.validate(&layout, 10, 12, &limits).is_err()); + } + assert_eq!( + bootstrap.version, + captured.manifests.unwrap().initial_version + ); + Ok(()) +} + +#[test] +fn reservation_requires_exact_prepared_plan_current_owner_and_head() -> Result<()> { + let limits = Limits::default(); + let (catalog, _, request) = fixture(&limits)?; + for mutation in 0..8 { + let mut changed = request.clone(); + let expected = match mutation { + 0 => { + changed.authority.owner = Uuid::new_v4(); + Rejection::Authority + } + 1 => { + changed.predecessor = Uuid::new_v4(); + Rejection::ManifestPredecessor + } + 2 => { + changed.transaction = Uuid::new_v4(); + Rejection::PlanIdentity + } + 3 => { + changed.plan_index -= 1; + Rejection::PlanIdentity + } + 4 => { + changed.prepared_index += 1; + Rejection::PlanIdentity + } + 5 => { + changed.descriptor_digest[0] ^= 1; + Rejection::PlanIdentity + } + 6 => { + changed.target = 9; + changed.object = object(9, true); + Rejection::PlanIdentity + } + 7 => { + changed.incarnation = Uuid::new_v4(); + Rejection::ManifestIdentity + } + _ => unreachable!(), + }; + let mut current = catalog.clone(); + assert_eq!( + reserve(&mut current, changed, 8, &limits)?, + Reply::Rejected(expected) + ); + assert_eq!(current, catalog); + } + Ok(()) +} + +#[test] +fn bounded_capacity_preserves_exact_retries_and_paths_cannot_cross_versions() -> Result<()> { + let limits = Limits { + max_manifest_records: 1, + ..Default::default() + }; + let (mut catalog, bootstrap, request) = fixture(&limits)?; + assert_eq!( + reserve(&mut catalog, request, 8, &limits)?, + Reply::Rejected(Rejection::Capacity) + ); + let layout = crate::replication::tests::layout(); + let Reply::ManifestReserved(record) = + catalog.apply(decode(&bootstrap.clone().batch(&layout)?.body)?, 9, &limits) + else { + anyhow::bail!("retry blocked at capacity"); + }; + assert_eq!(record.index, 5); + for path in [ + "_versions/8.manifest", + "_versions/../7.manifest", + "other/7.manifest", + "_versions/7.manifest-not-uuid", + "_versions/7.manifest?secret=x", + ] { + let mut bad = bootstrap.clone(); + bad.object.path = path.into(); + assert!(bad.batch(&layout).is_err()); + } + let mut v2 = bootstrap; + v2.object.v2_names = true; + v2.object.path = format!("_versions/{:020}.manifest", u64::MAX - 7); + v2.batch(&layout)?; + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/publication/plan.rs b/crates/lance-context-wal/src/replication/publication/plan.rs index 5f88d1f..dcb966d 100644 --- a/crates/lance-context-wal/src/replication/publication/plan.rs +++ b/crates/lance-context-wal/src/replication/publication/plan.rs @@ -236,6 +236,10 @@ impl Catalog { }; } if !table.progress.matches(&request.plan, &table.ingest) + || table + .manifests + .as_ref() + .is_some_and(|m| m.head != request.plan.read_version) || request .plan .sources diff --git a/crates/lance-context-wal/src/replication/query.rs b/crates/lance-context-wal/src/replication/query.rs index 8c15d26..ea3bf41 100644 --- a/crates/lance-context-wal/src/replication/query.rs +++ b/crates/lance-context-wal/src/replication/query.rs @@ -8,6 +8,10 @@ mod tests; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Query { + Manifest { + table: Uuid, + version: Option, + }, Publisher { table: Uuid, }, @@ -33,6 +37,7 @@ pub enum Query { #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Value { + Manifest(Option>), Publisher(Option>), Accepted(Page), Producer(Option<(BatchMeta, u64)>), @@ -72,6 +77,12 @@ pub struct Completion { impl Query { pub(super) fn validate(&self, layout: &Layout, group: u64) -> Result<()> { match self { + Self::Manifest { table, version } => ensure!( + !table.is_nil() + && group == publication::control_group(layout) + && version.is_none_or(|v| v > 0 && v < u64::MAX), + "invalid manifest query scope" + ), Self::Pin { table, transaction, @@ -116,6 +127,15 @@ impl Query { signer: Option<&source::CatalogSigner>, ) -> Result { Ok(match self { + Self::Manifest { table, version } => Value::Manifest( + state + .catalog + .tables + .get(table) + .and_then(|t| t.manifests.as_ref()) + .and_then(|m| m.versions.get(&version.unwrap_or(m.head))) + .cloned(), + ), Self::Pin { table, transaction, diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index 2764fbf..4adc9eb 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -62,6 +62,14 @@ impl WirePacket { } #[derive(Serialize, Deserialize)] enum Request { + BootstrapManifest( + u64, + lance_context_wal::replication::publication::BootstrapManifest, + ), + ReserveManifest( + u64, + lance_context_wal::replication::publication::ReserveManifest, + ), Prepare(u64, PrepareRequest), Pin(u64, Permit), Query(u64, u64, Query), @@ -130,6 +138,8 @@ fn part(payload: bool) -> Part { fn run(worker: &mut Worker, request: Request) -> Result { let node = &mut worker.node; match 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)?, Request::Pin(token, permit) => node.pin_source(token, permit)?, Request::Query(group, token, query) => node.query(group, token, query)?, diff --git a/crates/lance-context-wal/tests/replication_process/lance_publication.rs b/crates/lance-context-wal/tests/replication_process/lance_publication.rs index 090fdd2..6f7b105 100644 --- a/crates/lance-context-wal/tests/replication_process/lance_publication.rs +++ b/crates/lance-context-wal/tests/replication_process/lance_publication.rs @@ -1,6 +1,8 @@ //! Real Lance files and six persistent WAL processes. Collection is disabled; //! this qualifies descriptor recovery, not the external commit/GC fence. use super::*; +#[path = "managed_manifest.rs"] +mod managed; use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator, StringArray}; use arrow_schema::{DataType, Field, Schema}; use futures::TryStreamExt; @@ -39,6 +41,9 @@ fn batch(ids: &[i64]) -> Result { } async fn verify(uri: &str, expected: &[i64]) -> Result<()> { let dataset = Dataset::open(uri).await?; + verify_dataset(dataset, expected).await +} +async fn verify_dataset(dataset: Dataset, expected: &[i64]) -> Result<()> { let batches: Vec<_> = dataset .scan() .try_into_stream() @@ -86,7 +91,7 @@ impl CommitHandler for FixedTarget { } } -async fn recovered_publication(occupied: bool) -> Result<()> { +async fn recovered_publication(occupied: bool, managed: Option) -> Result<()> { let key = ring::signature::Ed25519KeyPair::generate_pkcs8(&ring::rand::SystemRandom::new()) .map_err(|_| anyhow::anyhow!("test key generation failed"))?; let layout = Layout { @@ -161,6 +166,11 @@ async fn recovered_publication(occupied: bool) -> Result<()> { else { anyhow::bail!("authority missing"); }; + let bootstrap = if managed.is_some() { + Some(managed::bootstrap(&mut c, catalog, &base, &authority).await?) + } else { + None + }; let transaction = Uuid::from_u128(907); c.command( catalog, @@ -313,6 +323,22 @@ async fn recovered_publication(occupied: bool) -> Result<()> { let decoded: Vec = serde_json::from_slice(&saved.descriptors.bytes)?; let mut tx = Transaction::new(1, Operation::Append { fragments: decoded }, None); tx.uuid = saved.transaction.to_string(); + if let Some(mode) = managed { + return managed::publish_and_recover( + c, + managed::Attempt { + uri, + base, + tx, + prepared: saved, + authority: view.authority, + bootstrap: bootstrap.context("missing managed bootstrap")?, + input, + }, + mode, + ) + .await; + } let result = CommitBuilder::new(Arc::new(base)) .with_commit_handler(Arc::new(FixedTarget(2))) .with_max_retries(0) @@ -364,10 +390,25 @@ async fn recovered_publication(occupied: bool) -> Result<()> { #[tokio::test] async fn recovered_descriptors_publish_exact_lance_payload_after_all_process_restart() -> Result<()> { - recovered_publication(false).await + recovered_publication(false, None).await } #[tokio::test] async fn recovered_descriptors_do_not_retarget_an_occupied_lance_version() -> Result<()> { - recovered_publication(true).await + recovered_publication(true, None).await +} + +#[tokio::test] +async fn catalog_manifest_authority_recovers_unique_lance_path_after_full_restart() -> Result<()> { + recovered_publication(false, Some(managed::Mode::Normal)).await +} + +#[tokio::test] +async fn catalog_manifest_authority_resolves_lost_commit_reply_after_full_restart() -> Result<()> { + recovered_publication(false, Some(managed::Mode::LostReply)).await +} + +#[tokio::test] +async fn catalog_rejects_old_owner_paused_after_real_manifest_upload() -> Result<()> { + recovered_publication(false, Some(managed::Mode::Takeover)).await } diff --git a/crates/lance-context-wal/tests/replication_process/managed_manifest.rs b/crates/lance-context-wal/tests/replication_process/managed_manifest.rs new file mode 100644 index 0000000..b3f8afa --- /dev/null +++ b/crates/lance-context-wal/tests/replication_process/managed_manifest.rs @@ -0,0 +1,433 @@ +//! Actual Lance writer/reader adapter over the six-process catalog fixture. +//! 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, +}; +use object_store::path::Path as ObjectPath; +use std::sync::{ + atomic::{AtomicBool, AtomicU64, Ordering}, + Mutex, +}; + +pub(super) async fn bootstrap( + c: &mut Cluster, + leader: u64, + dataset: &Dataset, + authority: &Authority, +) -> Result<(Uuid, Arc)> { + let location = dataset.manifest_location(); + let filename = location.path.filename().context("manifest filename")?; + let incarnation = Uuid::from_u128(930); + let genesis = dataset + .read_transaction_by_version(1) + .await? + .context("actual genesis transaction")?; + c.command( + leader, + Request::BootstrapManifest( + 2009, + BootstrapManifest { + authority: authority.clone(), + incarnation, + request: Uuid::from_u128(931), + version: 1, + object: ManifestObject { + path: format!("_versions/{filename}"), + bytes: location.size.context("known written size")?, + v2_names: location.naming_scheme == ManifestNamingScheme::V2, + }, + transaction: Uuid::parse_str(&genesis.uuid)?, + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(Outcome::Committed(Applied::Publisher(PublisherReply::ManifestReserved(record)))) = + c.outcome(2009) + else { + anyhow::bail!("manifest bootstrap missing"); + }; + Ok((incarnation, record)) +} + +pub(super) struct Attempt { + pub uri: String, + pub base: Dataset, + pub tx: Transaction, + pub prepared: Arc, + pub authority: Authority, + pub bootstrap: (Uuid, Arc), + pub input: Batch, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +pub(super) enum Mode { + Normal, + LostReply, + Takeover, +} + +#[derive(Default)] +struct Pause { + uploaded: tokio::sync::Notify, + resume: tokio::sync::Notify, + path: Mutex>, +} + +struct Managed { + cluster: Arc>, + base: ObjectPath, + scope: ReserveManifest, + tokens: AtomicU64, + lose_reply: AtomicBool, + pause: Option>, +} +impl std::fmt::Debug for Managed { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("ManagedCatalogFixture") + } +} +fn error(error: impl std::fmt::Display) -> lance::Error { + lance::Error::io(error.to_string()) +} +impl Managed { + fn check_base(&self, base: &ObjectPath) -> lance::Result<()> { + if *base != self.base { + return Err(lance::Error::invalid_input("foreign managed dataset")); + } + Ok(()) + } + fn read(&self, version: Option) -> lance::Result>> { + let mut c = self.cluster.lock().unwrap(); + let leader = c.elect(1, 10, &[1, 2, 3]).map_err(error)?; + let token = self.tokens.fetch_add(1, Ordering::Relaxed); + c.command( + leader, + Request::Query( + 10, + token, + Query::Manifest { + table: self.scope.authority.table, + version, + }, + ), + ) + .map_err(error)?; + c.pump(&[1, 2, 3]).map_err(error)?; + match c.read_outcome(token) { + Some(ReadOutcome::Observed { + value: Value::Manifest(record), + .. + }) => Ok(record), + other => Err(error(format!( + "manifest quorum read unavailable: {other:?}" + ))), + } + } + fn location(&self, record: &ManifestRecord) -> ManifestLocation { + ManifestLocation { + version: record.version, + path: ObjectPath::parse(format!("{}/{}", self.base, record.object.path)) + .expect("validated relative manifest path"), + size: Some(record.object.bytes), + naming_scheme: if record.object.v2_names { + ManifestNamingScheme::V2 + } else { + ManifestNamingScheme::V1 + }, + e_tag: None, + identity: Some(record.identity.to_string()), + } + } +} + +#[async_trait::async_trait] +#[allow(clippy::too_many_arguments)] +impl CommitHandler for Managed { + async fn resolve_latest_location( + &self, + base: &ObjectPath, + _: &lance_io::object_store::ObjectStore, + ) -> lance::Result { + self.check_base(base)?; + self.read(None)? + .map(|r| self.location(&r)) + .ok_or_else(|| lance::Error::not_found("managed head")) + } + async fn resolve_version_location( + &self, + base: &ObjectPath, + version: u64, + _: &dyn object_store::ObjectStore, + ) -> lance::Result { + self.check_base(base)?; + self.read(Some(version))? + .map(|r| self.location(&r)) + .ok_or_else(|| lance::Error::not_found(format!("managed version {version}"))) + } + async fn version_exists( + &self, + base: &ObjectPath, + version: u64, + _: &dyn object_store::ObjectStore, + _: ManifestNamingScheme, + ) -> lance::Result { + self.check_base(base)?; + Ok(self.read(Some(version))?.is_some()) + } + async fn commit( + &self, + manifest: &mut Manifest, + indices: Option>, + base: &ObjectPath, + store: &lance_io::object_store::ObjectStore, + writer: ManifestWriter, + naming: ManifestNamingScheme, + transaction: Option, + ) -> std::result::Result { + self.check_base(base)?; + if manifest.version != self.scope.target + || transaction + .as_ref() + .is_none_or(|tx| tx.as_pb().uuid != self.scope.transaction.to_string()) + { + return Err(CommitError::CommitConflict); + } + let path = ObjectPath::parse(format!( + "{}-{}", + naming.manifest_path(base, manifest.version), + Uuid::new_v4() + )) + .map_err(error)?; + let written = writer(store, manifest, indices, &path, transaction).await?; + if let Some(pause) = &self.pause { + *pause.path.lock().unwrap() = Some(path.clone()); + pause.uploaded.notify_one(); + pause.resume.notified().await; + } + let mut request = self.scope.clone(); + request.object = ManifestObject { + path: format!("_versions/{}", path.filename().expect("staging filename")), + bytes: written.size as u64, + v2_names: naming == ManifestNamingScheme::V2, + }; + let outcome = { + let mut c = self.cluster.lock().unwrap(); + let leader = c.elect(1, 10, &[1, 2, 3]).map_err(error)?; + let token = self.tokens.fetch_add(1, Ordering::Relaxed); + c.command(leader, Request::ReserveManifest(token, request)) + .map_err(error)?; + c.pump(&[1, 2, 3]).map_err(error)?; + c.outcome(token) + }; + let Some(Outcome::Committed(Applied::Publisher(PublisherReply::ManifestReserved(record)))) = + outcome + else { + // Keep staging on every unresolved/error outcome in this test. + return Err(CommitError::OtherError(error(format!( + "reservation unavailable: {outcome:?}" + )))); + }; + if self.lose_reply.swap(false, Ordering::SeqCst) { + return Err(CommitError::OtherError(error( + "injected lost reply after durable catalog commit", + ))); + } + Ok(self.location(&record)) + } + async fn forget_version(&self, _: &ObjectPath, _: u64, _: &str) -> lance::Result<()> { + Err(lance::Error::not_supported( + "coordinated collection is not enabled", + )) + } +} + +pub(super) async fn publish_and_recover(c: Cluster, attempt: Attempt, mode: Mode) -> Result<()> { + let base_path = attempt + .base + .manifest_location() + .path + .as_ref() + .rsplit_once("/_versions/") + .context("dataset base path")? + .0; + let base_path = ObjectPath::parse(base_path)?; + let cluster = Arc::new(Mutex::new(c)); + let pause = (mode == Mode::Takeover).then(|| Arc::new(Pause::default())); + let mut handler = Arc::new(Managed { + cluster: cluster.clone(), + base: base_path, + scope: ReserveManifest { + authority: attempt.authority.clone(), + incarnation: attempt.bootstrap.0, + transaction: attempt.prepared.transaction, + plan_index: attempt.prepared.plan_index, + prepared_index: attempt.prepared.index, + descriptor_digest: attempt.prepared.descriptors.digest, + predecessor: attempt.bootstrap.1.identity, + target: 2, + object: attempt.bootstrap.1.object.clone(), + }, + tokens: AtomicU64::new(6000), + lose_reply: AtomicBool::new(mode == Mode::LostReply), + pause: pause.clone(), + }); + if let Some(pause) = pause { + let old = tokio::spawn(commit( + Arc::new(attempt.base), + attempt.tx.clone(), + handler.clone(), + )); + tokio::time::timeout(Duration::from_secs(20), pause.uploaded.notified()).await?; + let old_path = pause + .path + .lock() + .unwrap() + .clone() + .context("paused staging path")?; + assert!(Path::new(&attempt.uri) + .join("_versions") + .join(old_path.filename().unwrap()) + .is_file()); + let authority = takeover(&handler)?; + pause.resume.notify_one(); + let old_result = tokio::time::timeout(Duration::from_secs(20), old).await??; + assert!( + old_result.is_err(), + "old owner selected a path after takeover" + ); + assert!(handler.read(Some(2))?.is_none()); + assert_eq!(handler.read(None)?.unwrap().version, 1); + let mut scope = handler.scope.clone(); + scope.authority = authority; + handler = Arc::new(Managed { + cluster: cluster.clone(), + base: handler.base.clone(), + scope, + tokens: AtomicU64::new(10000), + lose_reply: AtomicBool::new(false), + pause: None, + }); + let base = lance::dataset::builder::DatasetBuilder::from_uri(&attempt.uri) + .with_commit_handler(handler.clone()) + .load() + .await?; + commit(Arc::new(base), attempt.tx, handler.clone()).await?; + assert_ne!( + handler.location(&handler.read(Some(2))?.unwrap()).path, + old_path + ); + } else { + let result = commit(Arc::new(attempt.base), attempt.tx, handler.clone()).await; + if mode == Mode::LostReply { + assert!( + result.is_err(), + "lost response should remain ambiguous to the caller" + ); + } else { + result?; + } + } + let selected = handler.read(Some(2))?.context("missing committed target")?; + assert!(selected.object.path.contains(".manifest-")); + { + let mut c = cluster.lock().unwrap(); + let leader = c.elect(1, 10, &[1, 2, 3])?; + c.peers + .get_mut(&leader) + .unwrap() + .call(Request::PrepareCheckpoint(10, Uuid::new_v4()))?; + c.command(leader, Request::InstallCheckpoint)?; + c.pump(&[1, 2, 3])?; + c.queue.clear(); + c.peers.clear(); + for id in 1..=6 { + c.start(id)?; + } + for id in 4..=6 { + verify_records( + c.peers.get_mut(&id).unwrap(), + 20, + std::slice::from_ref(&attempt.input), + )?; + } + } + assert_eq!(handler.read(Some(2))?.as_deref(), Some(selected.as_ref())); + // 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) + .with_commit_handler(handler.clone()) + .load() + .await?; + assert_eq!(latest.version().version, 2); + assert_eq!( + latest.manifest_location().identity, + Some(selected.identity.to_string()) + ); + let transaction = latest + .read_transaction_by_version(2) + .await? + .context("missing actual transaction")?; + assert_eq!(transaction.uuid, attempt.prepared.transaction.to_string()); + verify_dataset(latest, &[0, 1, 2, 3, 4]).await?; + assert!(handler.read(Some(3))?.is_none()); + assert_eq!(handler.read(None)?.unwrap().version, 2); + Ok(()) +} + +async fn commit( + base: Arc, + tx: Transaction, + handler: Arc, +) -> lance::Result { + CommitBuilder::new(base) + .with_commit_handler(handler) + .with_max_retries(0) + .with_skip_auto_cleanup(true) + .execute(tx) + .await +} + +fn takeover(handler: &Managed) -> Result { + let mut c = handler.cluster.lock().unwrap(); + let leader = c.elect(1, 10, &[1, 2, 3])?; + c.command( + leader, + Request::Query( + 10, + 100_000, + Query::Publisher { + table: handler.scope.authority.table, + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + value: Value::Publisher(Some(view)), + .. + }) = c.read_outcome(100_000) + else { + anyhow::bail!("takeover view missing"); + }; + c.command( + leader, + Request::Acquire( + 100_001, + Acquire { + table: view.authority.table, + destination: view.destination, + ingest: view.ingest, + owner: Uuid::from_u128(940), + request: Uuid::from_u128(941), + expected_epoch: view.authority.epoch, + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(authority)))) = + c.outcome(100_001) + else { + anyhow::bail!("takeover did not commit"); + }; + Ok(authority) +} diff --git a/docs/design/wal-manifest-authority.md b/docs/design/wal-manifest-authority.md new file mode 100644 index 0000000..784cec7 --- /dev/null +++ b/docs/design/wal-manifest-authority.md @@ -0,0 +1,52 @@ +# Replicated manifest selection + +Refs #378. + +A publisher epoch checked before a blob write cannot revoke a write already in +flight. The optional catalog manifest index instead selects a unique immutable +staging path through the same durable quorum as the publication plan. Managed +readers resolve that path from the catalog; they must not import a missing row +from object-store listing or a canonical version path. + +`bootstrap_manifest` adopts one verified immutable manifest for an exclusively +managed dataset incarnation. The caller must verify its actual transaction and +exclude unmanaged writers and cleaners before adoption. This call does not +establish those external preconditions or validate object bytes on replicas. + +`reserve_manifest` requires the current owner, incarnation, exact pending plan, +prepared descriptor index/digest, fixed target and current head identity. The +head comparison is stronger than merely finding an older retained predecessor. +The quorum chooses one path and advances the head in the same state transition. +An ETag is not used as logical version identity. Staging paths identify their +attached version; canonical paths cannot be selected for new publications. + +An exact retry of an already selected object resolves its original record even +after publisher takeover. This does not grant a new write. Reporting that old +successful request as newly refused could cause its caller to delete a committed +staging object. A different object for an occupied target returns the chosen +record; it never replaces it. Unknown RPC outcomes require quorum readback and +retention, not a retargeted transaction. + +`Query::Manifest` returns one reference-counted record, including for latest +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. + +The real Lance test adapter writes unique staging manifests using the public +`ManifestWriter`, then reserves them through six persistent TCP WAL processes. +Its reader resolves latest and explicit versions strictly through quorum +queries. Tests cover snapshot/restart and a lost response after catalog commit, +and inspect actual Lance transaction UUIDs and complete payloads. A third test +pauses an old owner after its real manifest upload, commits a publisher takeover, +then resumes the old reservation. The old owner must leave the target vacant; +only the new owner may select a path. This adapter +is a test transport bridge, not the production ingestion service. + +No index row, manifest, descriptor or source pin is collected in this increment. +Outcome advancement, source closure, active-reader protection, bounded orphan +handling and coordinated cleanup remain required. Production adapters must also +override version enumeration, detached-version behavior and maintenance APIs; +the test adapter is deliberately limited to publication and direct reads. +URI-only readers and unmanaged writers do not satisfy this managed contract. diff --git a/docs/design/wal-prepared-descriptors.md b/docs/design/wal-prepared-descriptors.md index b47b5da..795f1bd 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 v5; older identities +descriptors. The application/checkpoint format advances to v6; 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-plans.md b/docs/design/wal-publication-plans.md index fe844ed..7358a38 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 v5. Existing v1/v2/v3/v4 directories and images are rejected; there is no in-place migration +format to v6. Existing v1/v2/v3/v4/v5 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 8f917b4..5017360 100644 --- a/docs/design/wal-publisher-authority.md +++ b/docs/design/wal-publisher-authority.md @@ -36,9 +36,9 @@ 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 v5, including [immutable publication plans](wal-publication-plans.md) +format are v6, including [immutable publication plans](wal-publication-plans.md) and [source pins](wal-source-pins.md). -Earlier v1/v2/v3/v4 application directories are rejected; this optional component does +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. `Node::publisher` exposes only a local applied view. It is not a quorum read, diff --git a/docs/design/wal-source-pins.md b/docs/design/wal-source-pins.md index 3673768..cbd8b3d 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 v5 and rejects older identities +The application/checkpoint format advances to v6 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