From bdeaa3f1cee40e42a90d908c6bc259d7b991910c Mon Sep 17 00:00:00 2001 From: Beinan Date: Wed, 21 Oct 2026 06:29:38 +0000 Subject: [PATCH] feat(wal): retire bounded manifest history --- crates/lance-context-wal/src/replication.rs | 12 + .../src/replication/publication.rs | 21 +- .../replication/publication/advance/tests.rs | 2 + .../advance/tests/manifest_retention.rs | 299 ++++++++++++++ .../src/replication/publication/manifest.rs | 19 +- .../publication/manifest/retention.rs | 110 ++++++ .../src/replication/query.rs | 27 +- .../tests/replication_process.rs | 7 + .../replication_process/lance_publication.rs | 6 + .../replication_process/managed_manifest.rs | 6 + .../replication_process/manifest_retention.rs | 370 ++++++++++++++++++ docs/design/wal-manifest-authority.md | 35 +- 12 files changed, 897 insertions(+), 17 deletions(-) create mode 100644 crates/lance-context-wal/src/replication/publication/advance/tests/manifest_retention.rs create mode 100644 crates/lance-context-wal/src/replication/publication/manifest/retention.rs create mode 100644 crates/lance-context-wal/tests/replication_process/manifest_retention.rs diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 9f62ed87..668f6bcc 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -548,6 +548,18 @@ impl Node { self.propose(publication::control_group(&self.layout), request, batch) } + /// Retire old catalog directory records while preserving the bootstrap, + /// current head and active publication dependencies. Wait for a committed + /// result. This is not permission to delete immutable Lance objects. + pub fn retire_manifest_history( + &mut self, + request: u64, + retirement: publication::RetireManifestHistory, + ) -> Result<()> { + let batch = retirement.batch(&self.layout)?; + 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( diff --git a/crates/lance-context-wal/src/replication/publication.rs b/crates/lance-context-wal/src/replication/publication.rs index 32132edf..80c4387a 100644 --- a/crates/lance-context-wal/src/replication/publication.rs +++ b/crates/lance-context-wal/src/replication/publication.rs @@ -15,8 +15,8 @@ mod prepared; mod table_codec; pub use advance::{AdvancePublication, Advanced}; pub use manifest::{ - BootstrapManifest, ManifestIndex, ManifestObject, ManifestRecord, ReserveManifest, - MAX_MANIFEST_VERSIONS, + BootstrapManifest, HistoryBoundary, ManifestIndex, ManifestObject, ManifestRecord, + ReserveManifest, RetireManifestHistory, MAX_MANIFEST_VERSIONS, }; pub use outcome::{ConfirmPublication, Published}; pub use plan::{PendingPlan, Plan, PlanRequest, Progress, SourceRange, MAX_INPUTS}; @@ -88,6 +88,7 @@ enum Command { ReserveManifest(Box), ConfirmPublication(Box), AdvancePublication(Box), + RetireManifestHistory(Box), } impl Command { fn identity(&self) -> (Uuid, Uuid, u64) { @@ -103,6 +104,9 @@ impl Command { Self::AdvancePublication(p) => { (p.authority.table, p.authority.owner, p.authority.epoch) } + Self::RetireManifestHistory(p) => { + (p.authority.table, p.authority.owner, p.authority.epoch) + } } } fn byte_limit(&self) -> usize { @@ -110,9 +114,9 @@ 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 - } + Self::BootstrapManifest(_) + | Self::ReserveManifest(_) + | Self::RetireManifestHistory(_) => manifest::MAX_MANIFEST_COMMAND_BYTES, Self::ConfirmPublication(_) => outcome::MAX_OUTCOME_BYTES, Self::AdvancePublication(_) => advance::MAX_ADVANCE_BYTES, } @@ -190,6 +194,7 @@ pub enum Rejection { ManifestPredecessor, PublicationIdentity, SourceClosure, + ManifestHistoryBoundary, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Reply { @@ -201,6 +206,8 @@ pub enum Reply { ManifestTaken(Arc), Published(Arc), Advanced(Arc), + ManifestHistory(HistoryBoundary), + ManifestRetired(HistoryBoundary), } #[derive(Serialize, Deserialize)] @@ -279,6 +286,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::RetireManifestHistory(p) => p.validate(layout)?, Command::ConfirmPublication(p) => p.validate(layout)?, Command::AdvancePublication(p) => p.validate(layout, limits)?, } @@ -306,6 +314,9 @@ impl Catalog { Command::AdvancePublication(request) => { return self.advance_publication(*request, index) } + Command::RetireManifestHistory(request) => { + return self.retire_manifest_history(*request) + } Command::Acquire(a) => *a, }; let old = self.tables.get(&a.table); diff --git a/crates/lance-context-wal/src/replication/publication/advance/tests.rs b/crates/lance-context-wal/src/replication/publication/advance/tests.rs index 306cc389..9c05d732 100644 --- a/crates/lance-context-wal/src/replication/publication/advance/tests.rs +++ b/crates/lance-context-wal/src/replication/publication/advance/tests.rs @@ -5,6 +5,8 @@ use crate::replication::{ }; use ring::{rand::SystemRandom, signature::Ed25519KeyPair}; +mod manifest_retention; + struct Fixture { layout: Arc, limits: Limits, diff --git a/crates/lance-context-wal/src/replication/publication/advance/tests/manifest_retention.rs b/crates/lance-context-wal/src/replication/publication/advance/tests/manifest_retention.rs new file mode 100644 index 00000000..5e05bae0 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/advance/tests/manifest_retention.rs @@ -0,0 +1,299 @@ +use super::*; +use crate::replication::query::{Query, Value}; + +fn apply_batch(f: &mut Fixture, batch: Batch, index: &mut u64) -> Result { + *index += 1; + validate_record(&f.layout, 10, &f.limits, &batch.meta, &batch.body)?; + Ok(f.catalog.apply(decode(&batch.body)?, *index, &f.limits)) +} + +fn retirement(f: &Fixture, retain_from: u64) -> RetireManifestHistory { + RetireManifestHistory { + authority: f.table().authority.clone(), + incarnation: f.table().manifests.as_ref().unwrap().incarnation, + retain_from, + } +} + +fn next_manifest(f: &mut Fixture, template: &Plan, index: &mut u64) -> Result { + let authority = f.table().authority.clone(); + let manifests = f.table().manifests.as_ref().unwrap(); + let predecessor = manifests.versions[&manifests.head].identity; + let incarnation = manifests.incarnation; + let mut plan = template.clone(); + plan.transaction = Uuid::new_v4(); + plan.read_version = manifests.head; + for (group, range) in &mut plan.sources { + range.after = f.table().progress.checkpoint[group]; + range.through = range.after + 1; + range.records[0].0 = range.through; + range.records[0].1.sequence = plan.read_version; + } + let batch = PlanRequest { + authority: authority.clone(), + plan: plan.clone(), + } + .batch(&f.layout, &f.limits)?; + assert_eq!( + apply_batch(f, batch, index)?, + Reply::Planned { index: *index } + ); + let plan_index = *index; + let descriptors = Descriptors::lance_fragments(Bytes::from_static(b"bounded-fixture"))?; + let batch = PrepareRequest { + authority: authority.clone(), + transaction: plan.transaction, + plan_index, + descriptors: descriptors.clone(), + } + .batch(&f.layout, &f.limits)?; + assert_eq!( + apply_batch(f, batch, index)?, + Reply::Prepared { index: *index } + ); + let target = plan.target_version()?; + Ok(ReserveManifest { + authority, + incarnation, + transaction: plan.transaction, + plan_index, + prepared_index: *index, + descriptor_digest: descriptors.digest, + predecessor, + target, + object: ManifestObject { + path: format!("_versions/{target}.manifest-{}", Uuid::new_v4()), + bytes: 4096, + v2_names: false, + }, + }) +} + +fn confirm_and_advance( + f: &mut Fixture, + record: Arc, + index: &mut u64, +) -> Result<()> { + let request = ConfirmPublication { + authority: f.table().authority.clone(), + incarnation: f.table().manifests.as_ref().unwrap().incarnation, + transaction: record.transaction, + plan_index: record.plan_index, + prepared_index: record.prepared_index, + descriptor_digest: record.descriptor_digest, + target: record.version, + manifest_identity: record.identity, + }; + let batch = request.batch(&f.layout)?; + assert!(matches!(apply_batch(f, batch, index)?, Reply::Published(_))); + let request = f.request()?; + *index += 1; + assert!(matches!(f.apply(request, *index)?, Reply::Advanced(_))); + Ok(()) +} + +#[test] +fn manifest_history_stays_bounded_across_three_hundred_publications_and_restore() -> Result<()> { + let mut f = fixture()?; + let template = f.table().progress.pending.as_ref().unwrap().plan.clone(); + let mut index = 30; + let request = f.request()?; + assert!(matches!(f.apply(request, index)?, Reply::Advanced(_))); + let bootstrap = f.table().manifests.as_ref().unwrap().versions[&7].clone(); + let mut old = None; + let mut captured = None; + for round in 0..300u64 { + let request = next_manifest(&mut f, &template, &mut index)?; + let batch = request.clone().batch(&f.layout)?; + let Reply::ManifestReserved(record) = apply_batch(&mut f, batch, &mut index)? else { + anyhow::bail!("manifest reservation failed") + }; + confirm_and_advance(&mut f, record, &mut index)?; + if round == 0 { + old = Some(request); + captured = Some(f.table().clone()); + } + let head = f.table().manifests.as_ref().unwrap().head; + if head >= 24 && round.is_multiple_of(8) { + let batch = retirement(&f, head - 7).batch(&f.layout)?; + assert!(matches!( + apply_batch(&mut f, batch, &mut index)?, + Reply::ManifestHistory(_) + )); + let bytes = encode_table(f.table())?; + let restored = decode_table(&bytes)?; + assert_eq!(encode_table(&restored)?, bytes); + f.catalog.tables.insert(restored.authority.table, restored); + } + let manifests = f.table().manifests.as_ref().unwrap(); + assert!(manifests.versions.len() <= 25); + assert_eq!(manifests.versions[&7], bootstrap); + f.catalog.validate(&f.layout, 10, index, &f.limits)?; + } + let old = old.unwrap(); + let captured = captured.unwrap(); + assert!(captured + .manifests + .as_ref() + .unwrap() + .versions + .contains_key(&old.target)); + let boundary = f.table().manifests.as_ref().unwrap().history_boundary(); + assert!(boundary.is_retired(old.target)); + assert!(!boundary.is_retired(7)); + let mut corrupt = f.catalog.clone(); + let manifests = Arc::make_mut( + corrupt + .tables + .get_mut(&f.table().authority.table) + .unwrap() + .manifests + .as_mut() + .unwrap(), + ); + manifests.versions.remove(&(boundary.retained_from + 1)); + assert!( + corrupt.validate(&f.layout, 10, index, &f.limits).is_err(), + "a hole inside the retained suffix must not become another retirement boundary" + ); + let before = f.catalog.clone(); + let batch = old.batch(&f.layout)?; + assert_eq!( + apply_batch(&mut f, batch, &mut index)?, + Reply::ManifestRetired(boundary.clone()) + ); + assert_eq!( + f.catalog, before, + "old reservation must not create another manifest" + ); + let state = State { + applied: index, + catalog: f.catalog.clone(), + ..Default::default() + }; + for (version, expected) in [ + (Some(9), Value::ManifestRetired(boundary)), + (Some(7), Value::Manifest(Some(bootstrap.clone()))), + (Some(6), Value::Manifest(None)), + ] { + assert_eq!( + Query::Manifest { + table: f.table().authority.table, + version + } + .evaluate( + &state, + &f.cut(10, index)?, + &f.layout, + &f.limits, + source::QuerySigners { + catalog: None, + source: None + }, + )?, + expected + ); + } + let bootstrap_retry = BootstrapManifest { + authority: f.table().authority.clone(), + incarnation: f.table().manifests.as_ref().unwrap().incarnation, + request: f.table().manifests.as_ref().unwrap().bootstrap_request, + version: bootstrap.version, + object: bootstrap.object.clone(), + transaction: bootstrap.transaction, + } + .batch(&f.layout)?; + assert_eq!( + apply_batch(&mut f, bootstrap_retry, &mut index)?, + Reply::ManifestReserved(bootstrap) + ); + Ok(()) +} + +#[test] +fn manifest_retirement_preserves_pending_predecessor_and_rejects_stale_authority() -> Result<()> { + let mut f = fixture()?; + let mut index = 20; + let before = f.catalog.clone(); + for (retain_from, expected) in [ + (6, Rejection::ManifestHistoryBoundary), + (9, Rejection::ManifestHistoryBoundary), + (8, Rejection::PendingPlan), + ] { + let batch = retirement(&f, retain_from).batch(&f.layout)?; + assert_eq!( + apply_batch(&mut f, batch, &mut index)?, + Reply::Rejected(expected) + ); + assert_eq!(f.catalog, before); + } + let request = f.request()?; + index += 1; + f.apply(request, index)?; + let mut stale = retirement(&f, 8); + stale.authority.owner = Uuid::new_v4(); + let batch = stale.batch(&f.layout)?; + assert_eq!( + apply_batch(&mut f, batch, &mut index)?, + Reply::Rejected(Rejection::Authority) + ); + let mut foreign = retirement(&f, 8); + foreign.incarnation = Uuid::new_v4(); + let batch = foreign.batch(&f.layout)?; + assert_eq!( + apply_batch(&mut f, batch, &mut index)?, + Reply::Rejected(Rejection::ManifestIdentity) + ); + let batch = retirement(&f, 8).batch(&f.layout)?; + assert!(matches!( + apply_batch(&mut f, batch, &mut index)?, + Reply::ManifestHistory(_) + )); + f.catalog.validate(&f.layout, 10, index, &f.limits)?; + Ok(()) +} + +#[test] +fn manifest_capacity_is_released_without_losing_the_pending_plan_or_retry_boundary() -> Result<()> { + let mut f = fixture()?; + f.limits.max_manifest_records = 4; + let template = f.table().progress.pending.as_ref().unwrap().plan.clone(); + let mut index = 30; + let request = f.request()?; + f.apply(request, index)?; + for _ in 0..2 { + let request = next_manifest(&mut f, &template, &mut index)?; + let batch = request.batch(&f.layout)?; + let Reply::ManifestReserved(record) = apply_batch(&mut f, batch, &mut index)? else { + anyhow::bail!("reservation missing") + }; + confirm_and_advance(&mut f, record, &mut index)?; + } + let request = next_manifest(&mut f, &template, &mut index)?; + let batch = request.clone().batch(&f.layout)?; + assert_eq!( + apply_batch(&mut f, batch, &mut index)?, + Reply::Rejected(Rejection::Capacity) + ); + let pending = f.table().progress.pending.clone(); + let batch = retirement(&f, 10).batch(&f.layout)?; + assert!(matches!( + apply_batch(&mut f, batch, &mut index)?, + Reply::ManifestHistory(_) + )); + assert_eq!(f.table().progress.pending, pending); + let boundary = f.table().manifests.as_ref().unwrap().history_boundary(); + assert!(boundary.is_retired(8)); + let batch = retirement(&f, 8).batch(&f.layout)?; + assert_eq!( + apply_batch(&mut f, batch, &mut index)?, + Reply::ManifestHistory(boundary) + ); + let batch = request.batch(&f.layout)?; + let Reply::ManifestReserved(record) = apply_batch(&mut f, batch, &mut index)? else { + anyhow::bail!("pending plan cannot resume after retirement") + }; + confirm_and_advance(&mut f, record, &mut index)?; + f.catalog.validate(&f.layout, 10, index, &f.limits)?; + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/publication/manifest.rs b/crates/lance-context-wal/src/replication/publication/manifest.rs index c3fb149a..1d1c95a4 100644 --- a/crates/lance-context-wal/src/replication/publication/manifest.rs +++ b/crates/lance-context-wal/src/replication/publication/manifest.rs @@ -2,6 +2,9 @@ //! select a staged path. No canonical-path repair, fallback import or GC occurs. use super::*; +mod retention; +pub use retention::{HistoryBoundary, RetireManifestHistory}; + #[cfg(test)] pub(super) mod tests; @@ -163,14 +166,17 @@ impl ManifestIndex { && 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, + && self.versions.first_key_value().map(|(v, _)| *v) == Some(self.initial_version) + && self.versions.last_key_value().map(|(v, _)| *v) == Some(self.head), "invalid manifest index bounds" ); let mut previous_index = 0; + let mut previous_version = self.initial_version; for (offset, (version, record)) in self.versions.iter().enumerate() { record.object.validate(*version)?; ensure!( - *version == self.initial_version + offset as u64 + (offset == 0 || *version > previous_version) + && (offset <= 1 || *version == previous_version + 1) && record.version == *version && !record.transaction.is_nil() && record.index > previous_index @@ -198,10 +204,12 @@ impl ManifestIndex { ); } previous_index = record.index; + previous_version = *version; } if let Some(pending) = &table.progress.pending { ensure!( pending.plan.read_version >= self.initial_version + && self.versions.contains_key(&pending.plan.read_version) && self.head >= pending.plan.read_version && self.head <= pending.plan.target_version()?, "managed plan does not name the current predecessor or target" @@ -308,6 +316,13 @@ impl Catalog { if manifests.incarnation != request.incarnation { return Reply::Rejected(Rejection::ManifestIdentity); } + let boundary = manifests.history_boundary(); + if boundary.is_retired(request.target) { + // Absence below the retained suffix is not evidence that this + // caller's immutable object was never selected. It cannot be deleted + // on the strength of this reply, including after publisher takeover. + return Reply::ManifestRetired(boundary); + } // 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. diff --git a/crates/lance-context-wal/src/replication/publication/manifest/retention.rs b/crates/lance-context-wal/src/replication/publication/manifest/retention.rs new file mode 100644 index 00000000..654ff849 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/manifest/retention.rs @@ -0,0 +1,110 @@ +//! Explicit catalog-history retirement. Immutable objects are not deleted here. +use super::*; +use std::ops::Bound::{Excluded, Unbounded}; + +/// The bootstrap record remains available for exact adoption retries. Other +/// retained versions form one contiguous suffix starting at `retained_from`. +/// A retired position does not prove that any supplied object was once selected. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct HistoryBoundary { + pub incarnation: Uuid, + pub bootstrap_version: u64, + pub retained_from: u64, +} + +impl HistoryBoundary { + pub fn is_retired(&self, version: u64) -> bool { + version > self.bootstrap_version && version < self.retained_from + } +} + +impl ManifestIndex { + pub fn history_boundary(&self) -> HistoryBoundary { + HistoryBoundary { + incarnation: self.incarnation, + bootstrap_version: self.initial_version, + retained_from: self + .versions + .range((Excluded(self.initial_version), Unbounded)) + .next() + .map_or(self.initial_version, |(version, _)| *version), + } + } +} + +/// Retire directory records strictly below `retain_from`, except the bootstrap +/// anchor. The current head, active publication dependencies and latest durable +/// advancement receipt must remain resolvable. This grants no object GC permit. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RetireManifestHistory { + pub authority: Authority, + pub incarnation: Uuid, + pub retain_from: u64, +} + +impl RetireManifestHistory { + pub(in crate::replication) fn batch(self, layout: &Layout) -> Result { + self.validate(layout)?; + Command::RetireManifestHistory(Box::new(self)).batch(layout) + } + + pub(in crate::replication::publication) fn validate(&self, layout: &Layout) -> Result<()> { + self.authority.validate(layout)?; + ensure!( + !self.incarnation.is_nil() && self.retain_from > 0 && self.retain_from < (1u64 << 63), + "invalid manifest history retirement scope" + ); + Ok(()) + } +} + +impl Catalog { + pub(in crate::replication::publication) fn retire_manifest_history( + &mut self, + request: RetireManifestHistory, + ) -> 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); + } + let Some(manifests) = &table.manifests else { + return Reply::Rejected(Rejection::ManifestIdentity); + }; + if manifests.incarnation != request.incarnation { + return Reply::Rejected(Rejection::ManifestIdentity); + } + if request.retain_from < manifests.initial_version || request.retain_from > manifests.head { + return Reply::Rejected(Rejection::ManifestHistoryBoundary); + } + let protected = table + .progress + .pending + .iter() + .map(|p| p.plan.read_version) + .chain(table.progress.published.iter().map(|p| p.manifest.version)) + .chain( + table + .progress + .advanced + .iter() + .map(|a| a.publication.manifest.version), + ); + if protected.into_iter().any(|v| request.retain_from > v) { + return Reply::Rejected(Rejection::PendingPlan); + } + // At most MAX_MANIFEST_VERSIONS records are visited. Captured immutable + // views keep their Arc records; the directory has no external I/O here. + let table = self + .tables + .get_mut(&request.authority.table) + .expect("known table"); + let manifests = Arc::make_mut(table.manifests.as_mut().expect("managed table")); + let bootstrap = manifests.initial_version; + manifests + .versions + .retain(|v, _| *v == bootstrap || *v >= request.retain_from); + Reply::ManifestHistory(manifests.history_boundary()) + } +} diff --git a/crates/lance-context-wal/src/replication/query.rs b/crates/lance-context-wal/src/replication/query.rs index 51fa3f5b..484f89c2 100644 --- a/crates/lance-context-wal/src/replication/query.rs +++ b/crates/lance-context-wal/src/replication/query.rs @@ -55,10 +55,15 @@ pub enum Value { Publisher(Option>), Accepted(Page), Producer(Option<(BatchMeta, u64)>), - CursorAhead { applied: u64 }, + CursorAhead { + applied: u64, + }, Pin(Option>), SourcePin(Option>), ClosedProof(Option>), + /// A historical position is outside catalog retention. This is neither a + /// missing-object result nor proof that a supplied staging object lost. + ManifestRetired(publication::HistoryBoundary), } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -201,15 +206,23 @@ impl Query { .transpose()?, ) } - Self::Manifest { table, version } => Value::Manifest( - state + Self::Manifest { table, version } => { + let manifests = state .catalog .tables .get(table) - .and_then(|t| t.manifests.as_ref()) - .and_then(|m| m.versions.get(&version.unwrap_or(m.head))) - .cloned(), - ), + .and_then(|t| t.manifests.as_ref()); + match manifests { + Some(m) if version.is_some_and(|v| m.history_boundary().is_retired(v)) => { + Value::ManifestRetired(m.history_boundary()) + } + _ => Value::Manifest( + manifests + .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 9a918c72..7f28b0d1 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 { + RetireManifestHistory( + u64, + lance_context_wal::replication::publication::RetireManifestHistory, + ), AdvancePublication( u64, lance_context_wal::replication::publication::AdvancePublication, @@ -151,6 +155,9 @@ fn part(payload: bool) -> Part { fn run(worker: &mut Worker, request: Request) -> Result { let node = &mut worker.node; match request { + Request::RetireManifestHistory(token, request) => { + node.retire_manifest_history(token, 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)?, 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 6f7b105b..21b32b6e 100644 --- a/crates/lance-context-wal/tests/replication_process/lance_publication.rs +++ b/crates/lance-context-wal/tests/replication_process/lance_publication.rs @@ -412,3 +412,9 @@ async fn catalog_manifest_authority_resolves_lost_commit_reply_after_full_restar async fn catalog_rejects_old_owner_paused_after_real_manifest_upload() -> Result<()> { recovered_publication(false, Some(managed::Mode::Takeover)).await } + +#[tokio::test] +async fn catalog_manifest_retention_preserves_real_lance_after_history_limit_and_restart( +) -> Result<()> { + recovered_publication(false, Some(managed::Mode::Retention)).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 index 46e66c53..3c7d383d 100644 --- a/crates/lance-context-wal/tests/replication_process/managed_manifest.rs +++ b/crates/lance-context-wal/tests/replication_process/managed_manifest.rs @@ -1,6 +1,8 @@ //! Actual Lance writer/reader adapter over the six-process catalog fixture. //! This qualifies the durable index, not production transport or coordinated GC. use super::*; +#[path = "manifest_retention.rs"] +mod manifest_retention; #[path = "source_closure.rs"] mod source_closure; use lance_context_wal::replication::publication::{ @@ -68,6 +70,7 @@ pub(super) enum Mode { Normal, LostReply, Takeover, + Retention, } #[derive(Default)] @@ -450,6 +453,9 @@ pub(super) async fn publish_and_recover(c: Cluster, attempt: Attempt, mode: Mode verify_dataset(latest, &[0, 1, 2, 3, 4]).await?; assert!(handler.read(Some(3))?.is_none()); assert_eq!(handler.read(None)?.unwrap().version, 2); + if mode == Mode::Retention { + manifest_retention::qualify(&attempt.uri, handler).await?; + } Ok(()) } diff --git a/crates/lance-context-wal/tests/replication_process/manifest_retention.rs b/crates/lance-context-wal/tests/replication_process/manifest_retention.rs new file mode 100644 index 00000000..09e61fc5 --- /dev/null +++ b/crates/lance-context-wal/tests/replication_process/manifest_retention.rs @@ -0,0 +1,370 @@ +//! More real Lance publications than the catalog history cap, with bounded +//! retained metadata, a lost commit reply, takeover and six-process restart. +use super::*; +use lance_context_wal::replication::publication::{AdvancePublication, RetireManifestHistory}; + +const CATALOG: &[u64] = &[1, 2, 3]; +const SOURCE: &[u64] = &[4, 5, 6]; +const LAST_VERSION: u64 = 162; + +struct Run { + handler: Arc, + token: u64, +} + +impl Run { + fn command(&mut self, group: u64, request: impl FnOnce(u64) -> Request) -> Result { + self.token += 1; + let mut c = self.handler.cluster.lock().unwrap(); + let members = if group == 10 { CATALOG } else { SOURCE }; + let leader = c.elect(members[0], group, members)?; + c.command(leader, request(self.token))?; + c.pump(members)?; + c.outcome(self.token).context("missing command completion") + } + + fn query(&mut self, group: u64, query: Query) -> Result { + self.token += 1; + let mut c = self.handler.cluster.lock().unwrap(); + let members = if group == 10 { CATALOG } else { SOURCE }; + let leader = c.elect(members[0], group, members)?; + c.command(leader, Request::Query(group, self.token, query))?; + c.pump(members)?; + let Some(ReadOutcome::Observed { value, .. }) = c.read_outcome(self.token) else { + anyhow::bail!("missing quorum observation"); + }; + Ok(value) + } + + fn view(&mut self) -> Result { + let table = self.handler.scope.authority.table; + let Value::Publisher(Some(view)) = self.query(10, Query::Publisher { table })? else { + anyhow::bail!("missing publisher view"); + }; + Ok(*view) + } + + fn restart(&self) -> Result<()> { + let mut c = self.handler.cluster.lock().unwrap(); + // Every member must reopen from its own installed image and durable log. + for id in 1..=6 { + let group = if id <= 3 { 10 } else { 20 }; + c.peers + .get_mut(&id) + .unwrap() + .call(Request::PrepareCheckpoint(group, Uuid::new_v4()))?; + c.command(id, Request::InstallCheckpoint)?; + } + c.pump(&[1, 2, 3, 4, 5, 6])?; + c.queue.clear(); + c.peers.clear(); + c.completions.clear(); + c.reads.clear(); + for id in 1..=6 { + c.start(id)?; + } + Ok(()) + } + + fn confirm_and_advance(&mut self, record: &ManifestRecord) -> Result<()> { + let authority = self.handler.scope.authority.clone(); + let table = authority.table; + let confirmation = ConfirmPublication { + authority: authority.clone(), + incarnation: self.handler.scope.incarnation, + transaction: record.transaction, + plan_index: record.plan_index, + prepared_index: record.prepared_index, + descriptor_digest: record.descriptor_digest, + target: record.version, + manifest_identity: record.identity, + }; + let Outcome::Committed(Applied::Publisher(PublisherReply::Published(published))) = + self.command(10, |token| Request::ConfirmPublication(token, confirmation))? + else { + anyhow::bail!("publication confirmation missing"); + }; + let Value::ClosePermit(Some(permit)) = self.query( + 10, + Query::ClosePermit { + table, + transaction: record.transaction, + plan_index: record.plan_index, + source_group: 20, + }, + )? + else { + anyhow::bail!("source closure permit missing"); + }; + assert!(matches!( + self.command(20, |token| Request::CloseSource(token, *permit))?, + Outcome::Committed(Applied::Source(SourceReply::Closed { .. })) + )); + let Value::ClosedProof(Some(proof)) = self.query(20, Query::ClosedProof { table })? else { + anyhow::bail!("durable source closure proof missing"); + }; + let request = AdvancePublication { + authority, + published_index: published.index, + proofs: BTreeMap::from([(20, *proof)]), + }; + let Outcome::Committed(Applied::Publisher(PublisherReply::Advanced(advanced))) = + self.command(10, |token| Request::AdvancePublication(token, request))? + else { + anyhow::bail!("catalog advancement missing"); + }; + assert_eq!(advanced.publication.manifest.as_ref(), record); + Ok(()) + } +} + +pub(super) async fn qualify(uri: &str, handler: Arc) -> Result<()> { + let mut run = Run { + handler, + token: 1_000_000, + }; + let bootstrap = run.handler.read(Some(1))?.context("bootstrap missing")?; + let mut old_reservation = None; + for version in 3..=LAST_VERSION { + if version == 90 { + let authority = takeover(&run.handler)?; + let mut scope = run.handler.scope.clone(); + scope.authority = authority; + run.handler = Arc::new(Managed { + cluster: run.handler.cluster.clone(), + base: run.handler.base.clone(), + scope, + tokens: AtomicU64::new(9_000_000), + lose_reply: AtomicBool::new(false), + pause: None, + }); + } + let view = run.view()?; + let authority = view.authority; + let table = authority.table; + let predecessor = run.handler.read(None)?.context("head missing")?; + assert_eq!(predecessor.version, version - 1); + let rows = batch(&[version as i64 + 2])?; + let mut wire = Vec::new(); + { + let mut encoder = arrow_ipc::writer::FileWriter::try_new(&mut wire, &rows.schema())?; + encoder.write(&rows)?; + encoder.finish()?; + } + let input = Batch::new( + table, + 1, + Uuid::from_u128(10_000 + version as u128), + 1, + Bytes::from(wire), + ); + let Outcome::Committed(Applied::Accepted { index }) = + run.command(20, |token| Request::Propose(20, token, input.clone()))? + else { + anyhow::bail!("source append missing at version {version}"); + }; + let transaction = Uuid::new_v4(); + let plan = PlanRequest { + authority: authority.clone(), + plan: Plan { + transaction, + read_version: version - 1, + next_group: 10, + sources: BTreeMap::from([( + 20, + SourceRange { + after: view.progress.checkpoint[&20], + through: index, + records: vec![(index, input.meta.clone())], + }, + )]), + }, + }; + let Outcome::Committed(Applied::Publisher(PublisherReply::Planned { index: plan_index })) = + run.command(10, |token| Request::Plan(token, plan))? + else { + anyhow::bail!("plan missing at version {version}"); + }; + let Value::Pin(Some(permit)) = run.query( + 10, + Query::Pin { + table, + transaction, + plan_index, + source_group: 20, + }, + )? + else { + anyhow::bail!("source pin permit missing"); + }; + assert!(matches!( + run.command(20, |token| Request::Pin(token, *permit))?, + Outcome::Committed(Applied::Source(SourceReply::Pinned { .. })) + )); + let recovered = { + let mut c = run.handler.cluster.lock().unwrap(); + let source = c.elect(4, 20, SOURCE)?; + let Response::Record(Some(record)) = c + .peers + .get_mut(&source) + .unwrap() + .call(Request::Read(20, index))? + else { + anyhow::bail!("pinned source body missing"); + }; + record + }; + assert_eq!(recovered, input); + let decoded = lance_context_wal::ipc::decode(recovered.body, rows.schema())?; + let base = Arc::new( + lance::dataset::builder::DatasetBuilder::from_uri(uri) + .with_commit_handler(run.handler.clone()) + .load() + .await?, + ); + let staged = InsertBuilder::new(base.clone()) + .with_params(&WriteParams { + mode: WriteMode::Append, + ..Default::default() + }) + .execute_uncommitted(decoded) + .await?; + let Operation::Append { fragments } = staged.operation else { + anyhow::bail!("not append fragments"); + }; + let descriptors = + Descriptors::lance_fragments(Bytes::from(serde_json::to_vec(&fragments)?))?; + let prepare = PrepareRequest { + authority: authority.clone(), + transaction, + plan_index, + descriptors: descriptors.clone(), + }; + let Outcome::Committed(Applied::Publisher(PublisherReply::Prepared { + index: prepared_index, + })) = run.command(10, |token| Request::Prepare(token, prepare))? + else { + anyhow::bail!("preparation missing"); + }; + let mut tx = Transaction::new(version - 1, Operation::Append { fragments }, None); + tx.uuid = transaction.to_string(); + run.handler = Arc::new(Managed { + cluster: run.handler.cluster.clone(), + base: run.handler.base.clone(), + scope: ReserveManifest { + authority, + incarnation: run.handler.scope.incarnation, + transaction, + plan_index, + prepared_index, + descriptor_digest: descriptors.digest, + predecessor: predecessor.identity, + target: version, + object: predecessor.object.clone(), + }, + tokens: AtomicU64::new(10_000_000 + version * 1000), + lose_reply: AtomicBool::new(version == 64), + pause: None, + }); + let committed = commit(base, tx, run.handler.clone()).await; + if version == 64 { + assert!(committed.is_err(), "injected lost reply was not observed"); + run.restart()?; + } else { + committed?; + } + let selected = run + .handler + .read(Some(version))? + .context("durable selection missing")?; + assert_eq!(selected.transaction, transaction); + if version == 3 { + let mut request = run.handler.scope.clone(); + request.object = selected.object.clone(); + old_reservation = Some(( + request, + std::fs::read(Path::new(uri).join(&selected.object.path))?, + )); + } + run.confirm_and_advance(&selected)?; + if version >= 16 && version % 8 == 0 { + let retirement = RetireManifestHistory { + authority: run.handler.scope.authority.clone(), + incarnation: run.handler.scope.incarnation, + retain_from: version - 7, + }; + let Outcome::Committed(Applied::Publisher(PublisherReply::ManifestHistory(boundary))) = + run.command(10, |token| { + Request::RetireManifestHistory(token, retirement) + })? + else { + anyhow::bail!("history retirement missing"); + }; + assert_eq!(boundary.retained_from, version - 7); + } + let view = run.view()?; + let manifests = view.manifests.context("manifest directory missing")?; + assert!(manifests.versions.len() <= 16); + assert_eq!(manifests.head, version); + assert_eq!(manifests.versions[&1], bootstrap); + assert!( + view.progress.pending.is_none() + && view.progress.prepared.is_none() + && view.progress.published.is_none() + ); + } + run.restart()?; + let (old, bytes) = old_reservation.context("original version-three request missing")?; + assert_ne!(old.authority, run.handler.scope.authority); + let old_path = Path::new(uri).join(&old.object.path); + let Outcome::Committed(Applied::Publisher(PublisherReply::ManifestRetired(boundary))) = + run.command(10, |token| Request::ReserveManifest(token, old))? + else { + anyhow::bail!("retired retry must resolve before stale authority rejection"); + }; + assert_eq!(boundary.retained_from, 153); + let table = run.handler.scope.authority.table; + assert_eq!( + run.query( + 10, + Query::Manifest { + table, + version: Some(3) + } + )?, + Value::ManifestRetired(boundary) + ); + assert_eq!( + std::fs::read(old_path)?, + bytes, + "metadata retirement must not delete or replace immutable objects" + ); + assert_eq!(run.handler.read(Some(1))?.as_ref(), Some(&bootstrap)); + let current = run.handler.read(None)?.context("head lost after restart")?; + assert_eq!(current.version, LAST_VERSION); + { + let mut c = run.handler.cluster.lock().unwrap(); + for id in 1..=3 { + let Response::Publisher(Some(view)) = c + .peers + .get_mut(&id) + .unwrap() + .call(Request::Publisher(table))? + else { + anyhow::bail!("replica catalog missing"); + }; + let manifests = view + .manifests + .context("replica manifest directory missing")?; + assert_eq!(manifests.versions.len(), 11); + assert_eq!(manifests.versions[&LAST_VERSION], current); + } + } + let latest = lance::dataset::builder::DatasetBuilder::from_uri(uri) + .with_commit_handler(run.handler) + .load() + .await?; + assert_eq!(latest.version().version, LAST_VERSION); + verify_dataset(latest, &(0..=LAST_VERSION as i64 + 2).collect::>()).await?; + Ok(()) +} diff --git a/docs/design/wal-manifest-authority.md b/docs/design/wal-manifest-authority.md index f3776eac..9b2e00cb 100644 --- a/docs/design/wal-manifest-authority.md +++ b/docs/design/wal-manifest-authority.md @@ -44,9 +44,38 @@ 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 +History retirement is explicit and opt-in. `retire_manifest_history` requires +the current authority and incarnation. It retains the bootstrap anchor and a +contiguous suffix starting at `retain_from`, including the current head. Pending +plans, confirmed publications and the latest durable advancement receipt protect +their versions from retirement. Advancing the boundary permits further +publications within the existing bounded index; it does not raise capacity. + +Queries and reservation retries for retired positions return `ManifestRetired` +with the history boundary, including retries from an old publisher epoch. This +result does not establish whether a supplied staging object was selected. Keep +that object until a separate coordinated cleanup protocol authorizes deletion. +The bootstrap remains available for exact adoption retries, and a captured +immutable index view retains its own records. Metadata retirement grants no +permission to remove manifests, transactions or Lance data files. Applications +must accept the loss of direct catalog lookup for retired historical versions +before enabling this policy. + +All replicas must support the new command before retirement is enabled. The +serialized index shape is unchanged, but older readers may reject sparse history +because they require consecutive versions from bootstrap through head. This is +not a mixed-version rollout guarantee. + +The process fixture continues actual Lance appends past the 128-version bound, +retires history in batches, loses a committed manifest reply before a complete +restart, and changes publisher authority. It verifies complete latest payloads +after checkpoint/restart of all six processes, bounded history on each catalog +replica, and a retired old-owner retry without removing its immutable object. +This uses local files and the test TCP bridge; it does not qualify a production +adapter, object-store garbage collection or volume loss. + +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.