From 07d9a3b1bb24995154fe0f4703aa7b06caf99681 Mon Sep 17 00:00:00 2001 From: Beinan Date: Tue, 20 Oct 2026 17:49:37 +0000 Subject: [PATCH] feat(wal): persist immutable materialization descriptors Refs #378. --- .github/workflows/rust-test.yml | 2 + Cargo.lock | 6 + crates/lance-context-wal/Cargo.toml | 8 + crates/lance-context-wal/src/replication.rs | 17 +- .../src/replication/checkpoint.rs | 7 +- .../src/replication/publication.rs | 44 ++- .../src/replication/publication/plan.rs | 32 +- .../src/replication/publication/prepared.rs | 109 +++++ .../replication/publication/prepared/tests.rs | 205 ++++++++++ .../tests/replication_process.rs | 126 +++++- .../replication_process/lance_publication.rs | 373 ++++++++++++++++++ docs/design/wal-prepared-descriptors.md | 45 +++ docs/design/wal-publication-plans.md | 2 +- docs/design/wal-publisher-authority.md | 4 +- docs/design/wal-source-pins.md | 2 +- 15 files changed, 950 insertions(+), 32 deletions(-) create mode 100644 crates/lance-context-wal/src/replication/publication/prepared.rs create mode 100644 crates/lance-context-wal/src/replication/publication/prepared/tests.rs create mode 100644 crates/lance-context-wal/tests/replication_process/lance_publication.rs create mode 100644 docs/design/wal-prepared-descriptors.md diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index 68cba13..ea9f216 100644 --- a/.github/workflows/rust-test.yml +++ b/.github/workflows/rust-test.yml @@ -84,6 +84,8 @@ jobs: run: | cargo clippy -p lance-context-wal --features replication --all-targets -- -D warnings cargo test -p lance-context-wal --features replication --lib --test replication_process + - name: Prepared WAL descriptors with real Lance files and process recovery + run: cargo test -p lance-context-wal --features lance-integration --test replication_process lance_publication - name: WAL resource observation in a container cgroup namespace run: | cargo test -p lance-context-wal --features runner --test resource --no-run --message-format=json > "$RUNNER_TEMP/wal-resource-build.jsonl" diff --git a/Cargo.lock b/Cargo.lock index 08145f5..546928e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4219,11 +4219,17 @@ dependencies = [ "arrow-buffer", "arrow-ipc", "arrow-schema", + "async-trait", "bincode", "bytes", "clap", + "futures", "hdrhistogram", + "lance", + "lance-io", + "lance-table", "libc", + "object_store 0.14.2", "protobuf", "raft", "raft-engine", diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 7f23285..a74ac46 100644 --- a/crates/lance-context-wal/Cargo.toml +++ b/crates/lance-context-wal/Cargo.toml @@ -7,6 +7,8 @@ publish = false description = "Independent Rust WAL experiments owned by lance-context" [features] +# Real Lance materialization/recovery qualification; normal WAL replicas do not link Lance. +lance-integration = ["replication", "ipc", "arrow-contract", "dep:lance", "dep:lance-table", "dep:lance-io", "dep:object_store", "dep:async-trait", "dep:futures", "dep:serde_json"] client-journal = ["dep:anyhow", "dep:raft-engine", "dep:bytes", "dep:serde", "dep:serde_json", "dep:tokio", "tokio/sync"] replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog", "dep:ring", "uuid/v4", "serde/rc"] raft-log = ["placement", "dep:raft", "dep:raft-engine", "dep:raft-proto", "dep:protobuf"] @@ -21,6 +23,12 @@ ipc = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "de lab = ["ipc", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "raft-engine/internals", "dep:raft-proto", "dep:serde_json"] [dependencies] +lance = { version = "13.0.0", optional = true } +lance-table = { version = "13.0.0", optional = true } +lance-io = { version = "13.0.0", optional = true } +object_store = { version = "0.14.2", optional = true } +async-trait = { version = "0.1.92", optional = true } +futures = { version = "0.3", optional = true } ring = { version = "0.17", optional = true } bincode = { version = "1.3", optional = true } slog = { version = "2", optional = true } diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 7415dd4..6b9121a 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -49,6 +49,7 @@ pub struct Limits { pub max_tables: u64, pub max_active_plans: u64, pub max_plan_source_bytes: u64, + pub max_descriptor_bytes: u64, pub max_record_bytes: u64, pub max_records: u64, pub max_producers: u64, @@ -62,6 +63,7 @@ impl Default for Limits { max_tables: 1024, max_active_plans: 128, max_plan_source_bytes: 64 * 1024 * 1024, + max_descriptor_bytes: 256 * 1024, max_record_bytes: 8 * 1024 * 1024, max_records: 100_000, max_producers: 16_384, @@ -77,6 +79,8 @@ impl Limits { && self.catalog_key.is_none_or(|key| key != [0; 32]) && (1..=256).contains(&self.max_active_plans) && (1..=256 * 1024 * 1024).contains(&self.max_plan_source_bytes) + && (1..=publication::MAX_DESCRIPTOR_BYTES as u64) + .contains(&self.max_descriptor_bytes) && (1..=8 * 1024 * 1024).contains(&self.max_record_bytes) && (1..=500_000).contains(&self.max_records) && (1..=100_000).contains(&self.max_producers) @@ -86,7 +90,7 @@ impl Limits { "invalid replication limits" ); Ok(bincode::serialize(&( - "lance-context/replicated-append/v4", + "lance-context/replicated-append/v5", self, ))?) } @@ -399,6 +403,17 @@ impl Node { self.propose(publication::control_group(&self.layout), request, batch) } + /// Persist one immutable descriptor set for an exact pending plan. The + /// caller must separately validate source pins and coordinate Lance commits. + pub fn prepare_publication( + &mut self, + request: u64, + prepared: publication::PrepareRequest, + ) -> Result<()> { + let batch = prepared.batch(&self.layout, &self.limits)?; + self.propose(publication::control_group(&self.layout), request, batch) + } + /// Verify a catalog quorum-read attestation and persist the exact complete /// table prefix in its source group. Completion, never enqueue, pins it. pub fn pin_source(&mut self, request: u64, permit: source::Permit) -> Result<()> { diff --git a/crates/lance-context-wal/src/replication/checkpoint.rs b/crates/lance-context-wal/src/replication/checkpoint.rs index bb4abc6..ae9103b 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 = 4; -const MAGIC: &[u8; 8] = b"LCWALCP4"; +const FORMAT: u32 = 5; +const MAGIC: &[u8; 8] = b"LCWALCP5"; pub const CHUNK_BYTES: usize = 1024 * 1024; fn codec() -> impl Options { @@ -84,7 +84,8 @@ fn metadata_limit(limits: &Limits) -> u64 { // limits.identity() validates all policy maxima before this is used. (limits.max_records + limits.max_producers) * 268 + limits.max_tables * (publication::MAX_BASE_TABLE_BYTES as u64 + 4) - + limits.max_tables.min(limits.max_active_plans) * publication::MAX_COMMAND_BYTES as 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) + 1024 } diff --git a/crates/lance-context-wal/src/replication/publication.rs b/crates/lance-context-wal/src/replication/publication.rs index 8987f33..c1a5aae 100644 --- a/crates/lance-context-wal/src/replication/publication.rs +++ b/crates/lance-context-wal/src/replication/publication.rs @@ -1,20 +1,24 @@ //! Replicated table bindings and publisher epochs. These commands run through //! the same durable Raft path as appends; a local view is not an active lease. -//! Immutable publication plans survive takeover. Source pins, manifest outcomes -//! and retirement remain separate work. No Lance manifest is changed here. +//! Immutable plans and prepared descriptors survive takeover. Source pins live +//! in the source module; manifest outcomes and retirement remain separate work. +//! No Lance manifest is changed here. use super::*; use crate::{ingest_contract::IngestContract, record::RecordKind}; use bincode::Options; use bytes::Bytes; mod plan; +mod prepared; pub use plan::{PendingPlan, Plan, PlanRequest, Progress, SourceRange, MAX_INPUTS}; +pub use prepared::{Descriptors, PrepareRequest, Prepared, MAX_DESCRIPTOR_BYTES}; #[cfg(test)] mod tests; pub(super) const MAX_ACQUIRE_BYTES: usize = 8192; -pub(super) const MAX_COMMAND_BYTES: usize = plan::MAX_PLAN_BYTES; +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; +pub(super) const MAX_TABLE_BYTES: usize = + MAX_BASE_TABLE_BYTES + plan::MAX_PLAN_BYTES + prepared::MAX_PREPARE_BYTES; fn codec(limit: usize) -> impl Options { bincode::DefaultOptions::new() .with_fixint_encoding() @@ -62,25 +66,28 @@ impl Acquire { enum Command { Acquire(Box), Plan(Box), + Prepare(Box), } impl Command { fn identity(&self) -> (Uuid, Uuid, u64) { match self { 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), } } fn byte_limit(&self) -> usize { match self { Self::Acquire(_) => MAX_ACQUIRE_BYTES, - Self::Plan(_) => MAX_COMMAND_BYTES, + Self::Plan(_) => plan::MAX_PLAN_BYTES, + Self::Prepare(_) => prepared::MAX_PREPARE_BYTES, } } fn batch(self, layout: &Layout) -> Result { let (table, owner, sequence) = self.identity(); let limit = self.byte_limit(); let envelope = Envelope { - version: 2, + version: 3, cluster: layout.cluster, layout: layout.fingerprint()?, group: control_group(layout), @@ -105,6 +112,24 @@ pub struct Authority { pub grant_index: u64, } +impl Authority { + fn validate(&self, layout: &Layout) -> Result<()> { + let a = self; + ensure!( + a.cluster == layout.cluster + && a.layout == layout.fingerprint()? + && a.group == control_group(layout) + && !a.owner.is_nil() + && a.epoch > 0 + && a.epoch <= a.grant_index + && a.grant_index < u64::MAX, + "invalid publication authority scope" + ); + ensure!(!a.table.is_nil(), "nil authority table"); + Ok(()) + } +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct Table { pub destination: Destination, @@ -124,12 +149,15 @@ pub enum Rejection { Authority, PendingPlan, SourcePosition, + PlanIdentity, + PreparedConflict, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum Reply { Acquired(Authority), Planned { index: u64 }, Rejected(Rejection), + Prepared { index: u64 }, } #[derive(Serialize, Deserialize)] @@ -163,7 +191,7 @@ pub(super) fn decode(bytes: &[u8]) -> Result { "publisher command too large" ); let envelope: Envelope = codec(MAX_COMMAND_BYTES).deserialize(bytes)?; - ensure!(envelope.version == 2, "unsupported publisher command"); + ensure!(envelope.version == 3, "unsupported publisher command"); ensure!( bytes.len() <= envelope.command.byte_limit(), "publisher command type byte limit" @@ -205,6 +233,7 @@ pub(super) fn validate_record( "unserviceable ingestion limit" ), Command::Plan(p) => p.validate(layout, limits)?, + Command::Prepare(p) => p.validate(layout, limits)?, } Ok(()) } @@ -217,6 +246,7 @@ impl Catalog { pub fn apply(&mut self, e: Envelope, index: u64, limits: &Limits) -> Reply { let a = match e.command { Command::Plan(p) => return self.plan(*p, index, limits), + Command::Prepare(request) => return self.prepare(*request, index), Command::Acquire(a) => *a, }; let old = self.tables.get(&a.table); diff --git a/crates/lance-context-wal/src/replication/publication/plan.rs b/crates/lance-context-wal/src/replication/publication/plan.rs index f8a713e..5f88d1f 100644 --- a/crates/lance-context-wal/src/replication/publication/plan.rs +++ b/crates/lance-context-wal/src/replication/publication/plan.rs @@ -124,18 +124,8 @@ impl PlanRequest { Command::Plan(Box::new(self)).batch(layout) } pub(super) fn validate(&self, layout: &Layout, limits: &Limits) -> Result<()> { - let a = &self.authority; - ensure!( - a.cluster == layout.cluster - && a.layout == layout.fingerprint()? - && a.group == control_group(layout) - && !a.owner.is_nil() - && a.epoch > 0 - && a.epoch <= a.grant_index - && a.grant_index < u64::MAX, - "invalid publication authority scope" - ); - self.plan.validate(layout, a.table, limits) + self.authority.validate(layout)?; + self.plan.validate(layout, self.authority.table, limits) } } @@ -152,6 +142,8 @@ pub struct Progress { pub minimum_read_version: u64, /// Cloning a table view shares the immutable metadata; it never clones bodies. pub pending: Option>, + /// Materialization result shares ownership independently of the selection. + pub prepared: Option>, } impl Default for Progress { fn default() -> Self { @@ -159,6 +151,7 @@ impl Default for Progress { checkpoint: BTreeMap::new(), minimum_read_version: 1, pending: None, + prepared: None, } } } @@ -189,6 +182,21 @@ impl Progress { }), "invalid publication progress" ); + if let Some(prepared) = &self.prepared { + let pending = self + .pending + .as_ref() + .context("prepared result without plan")?; + prepared.descriptors.validate(limits.max_descriptor_bytes)?; + ensure!( + prepared.plan_index == pending.index + && prepared.transaction == pending.plan.transaction + && prepared.index > pending.index + && prepared.index <= cut + && prepared.index != table.authority.grant_index, + "invalid prepared checkpoint identity or index" + ); + } if let Some(pending) = &self.pending { pending .plan diff --git a/crates/lance-context-wal/src/replication/publication/prepared.rs b/crates/lance-context-wal/src/replication/publication/prepared.rs new file mode 100644 index 0000000..16c23d5 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/prepared.rs @@ -0,0 +1,109 @@ +//! One immutable materialization result per pending plan. Recording descriptors +//! does not authorize a Lance commit or certify that remote source pins exist. +use super::*; + +#[cfg(test)] +mod tests; + +pub const MAX_DESCRIPTOR_BYTES: usize = 1024 * 1024; +pub(super) const MAX_PREPARE_BYTES: usize = MAX_DESCRIPTOR_BYTES + 1024; + +/// Versioned materializer metadata. Replicas validate length and digest, without +/// decoding Lance fragments or Arrow payloads. Only materializers interpret it. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Descriptors { + pub format: u32, + pub bytes: Bytes, + pub digest: [u8; 32], +} +impl Descriptors { + /// JSON representation of Lance 13 `Vec`, interpreted by the + /// materializer. This constructor does not validate the fragment semantics. + pub fn lance_fragments(bytes: Bytes) -> Result { + let descriptors = Self { + format: 1, + digest: Sha256::digest(&bytes).into(), + bytes, + }; + descriptors.validate(MAX_DESCRIPTOR_BYTES as u64)?; + Ok(descriptors) + } + + pub fn validate(&self, maximum: u64) -> Result<()> { + ensure!( + self.format == 1 + && !self.bytes.is_empty() + && self.bytes.len() <= MAX_DESCRIPTOR_BYTES + && self.bytes.len() as u64 <= maximum + && self.digest == <[u8; 32]>::from(Sha256::digest(&self.bytes)), + "invalid prepared descriptor format, length or digest" + ); + Ok(()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct PrepareRequest { + pub authority: Authority, + pub transaction: Uuid, + pub plan_index: u64, + pub descriptors: Descriptors, +} +impl PrepareRequest { + pub(in crate::replication) fn batch(self, layout: &Layout, limits: &Limits) -> Result { + self.validate(layout, limits)?; + Command::Prepare(Box::new(self)).batch(layout) + } + + pub(super) fn validate(&self, layout: &Layout, limits: &Limits) -> Result<()> { + self.authority.validate(layout)?; + ensure!( + !self.transaction.is_nil() && self.plan_index > 0 && self.plan_index < u64::MAX, + "invalid prepared plan identity" + ); + self.descriptors.validate(limits.max_descriptor_bytes) + } +} + +/// Shared by captured views and retained across publisher takeover. Its index +/// is the original durable preparation entry, not the latest retry entry. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Prepared { + pub index: u64, + pub transaction: Uuid, + pub plan_index: u64, + pub descriptors: Descriptors, +} + +impl Catalog { + pub(super) fn prepare(&mut self, request: PrepareRequest, index: u64) -> Reply { + let Some(table) = self.tables.get_mut(&request.authority.table) else { + return Reply::Rejected(Rejection::Authority); + }; + if table.authority != request.authority { + return Reply::Rejected(Rejection::Authority); + } + let Some(pending) = &table.progress.pending else { + return Reply::Rejected(Rejection::PlanIdentity); + }; + if pending.index != request.plan_index || pending.plan.transaction != request.transaction { + return Reply::Rejected(Rejection::PlanIdentity); + } + if let Some(prepared) = &table.progress.prepared { + return if prepared.descriptors == request.descriptors { + Reply::Prepared { + index: prepared.index, + } + } else { + Reply::Rejected(Rejection::PreparedConflict) + }; + } + table.progress.prepared = Some(Arc::new(Prepared { + index, + transaction: request.transaction, + plan_index: request.plan_index, + descriptors: request.descriptors, + })); + Reply::Prepared { index } + } +} diff --git a/crates/lance-context-wal/src/replication/publication/prepared/tests.rs b/crates/lance-context-wal/src/replication/publication/prepared/tests.rs new file mode 100644 index 0000000..5c28aa4 --- /dev/null +++ b/crates/lance-context-wal/src/replication/publication/prepared/tests.rs @@ -0,0 +1,205 @@ +use super::*; + +fn fixture() -> Result<(Catalog, PlanRequest)> { + let layout = crate::replication::tests::layout(); + let limits = Limits::default(); + let mut catalog = Catalog::default(); + let a = super::super::tests::input(); + let batch = a.clone().batch(&layout)?; + let Reply::Acquired(authority) = catalog.apply(decode(&batch.body)?, 4, &limits) else { + anyhow::bail!("missing authority"); + }; + let request = PlanRequest { + authority, + plan: Plan { + transaction: Uuid::from_u128(91), + read_version: 7, + next_group: 10, + sources: BTreeMap::from([( + 20, + SourceRange { + after: 0, + through: 1000, + records: vec![( + 999, + Batch::new( + a.table, + 1, + Uuid::from_u128(90), + 1, + Bytes::from_static(b"data"), + ) + .meta, + )], + }, + )]), + }, + }; + Ok((catalog, request)) +} + +fn apply( + catalog: &mut Catalog, + request: PrepareRequest, + index: u64, + limits: &Limits, +) -> Result { + let layout = crate::replication::tests::layout(); + let batch = request.batch(&layout, limits)?; + validate_record(&layout, 10, limits, &batch.meta, &batch.body)?; + Ok(catalog.apply(decode(&batch.body)?, index, limits)) +} + +#[test] +fn immutable_descriptors_keep_original_index_and_shared_bytes_through_takeover() -> Result<()> { + let (mut catalog, plan) = fixture()?; + let layout = crate::replication::tests::layout(); + let limits = Limits { + max_active_plans: 1, + ..Default::default() + }; + let mut request = PrepareRequest { + authority: plan.authority.clone(), + transaction: plan.plan.transaction, + plan_index: 5, + descriptors: Descriptors::lance_fragments(Bytes::from_static(b"fragment-descriptors"))?, + }; + assert_eq!( + apply(&mut catalog, request.clone(), 5, &limits)?, + Reply::Rejected(Rejection::PlanIdentity) + ); + let batch = plan.clone().batch(&layout, &limits)?; + assert_eq!( + catalog.apply(decode(&batch.body)?, 5, &limits), + Reply::Planned { index: 5 } + ); + let before = catalog.tables[&plan.authority.table].clone(); + for mutation in 0..2 { + let mut wrong = request.clone(); + if mutation == 0 { + wrong.plan_index += 1; + } else { + wrong.transaction = Uuid::new_v4(); + } + assert_eq!( + apply(&mut catalog, wrong, 6, &limits)?, + Reply::Rejected(Rejection::PlanIdentity) + ); + } + assert_eq!( + apply(&mut catalog, request.clone(), 6, &limits)?, + Reply::Prepared { index: 6 } + ); + let captured = catalog.tables[&plan.authority.table].clone(); + let saved = captured.progress.prepared.as_ref().unwrap(); + assert!(before.progress.prepared.is_none()); + assert!(Arc::ptr_eq( + before.progress.pending.as_ref().unwrap(), + captured.progress.pending.as_ref().unwrap() + )); + assert!(Arc::ptr_eq( + saved, + catalog.tables[&plan.authority.table] + .progress + .prepared + .as_ref() + .unwrap() + )); + assert_eq!( + apply(&mut catalog, request.clone(), 7, &limits)?, + Reply::Prepared { index: 6 } + ); + let mut changed = request.clone(); + changed.descriptors = Descriptors::lance_fragments(Bytes::from_static(b"different-files"))?; + assert_eq!( + apply(&mut catalog, changed, 8, &limits)?, + Reply::Rejected(Rejection::PreparedConflict) + ); + let mut acquire = super::super::tests::input(); + acquire.owner = Uuid::from_u128(93); + acquire.request = Uuid::from_u128(94); + acquire.expected_epoch = 1; + let batch = acquire.batch(&layout)?; + let Reply::Acquired(authority) = catalog.apply(decode(&batch.body)?, 9, &limits) else { + anyhow::bail!("takeover blocked by active-plan capacity"); + }; + assert_eq!( + apply(&mut catalog, request.clone(), 10, &limits)?, + Reply::Rejected(Rejection::Authority) + ); + request.authority = authority; + assert_eq!( + apply(&mut catalog, request, 11, &limits)?, + Reply::Prepared { index: 6 } + ); + let recovered = decode_table(&encode_table(&catalog.tables[&plan.authority.table])?)?; + assert_eq!(recovered.progress.prepared.as_deref(), Some(saved.as_ref())); + catalog.validate(&layout, 10, 11, &limits)?; + for mutation in 0..5 { + let mut invalid = catalog.clone(); + let progress = &mut invalid + .tables + .get_mut(&plan.authority.table) + .unwrap() + .progress; + let prepared = Arc::make_mut(progress.prepared.as_mut().unwrap()); + match mutation { + 0 => prepared.index = prepared.plan_index, + 1 => prepared.plan_index += 1, + 2 => prepared.transaction = Uuid::new_v4(), + 3 => prepared.descriptors.digest[0] ^= 1, + 4 => progress.pending = None, + _ => unreachable!(), + } + assert!(invalid.validate(&layout, 10, 11, &limits).is_err()); + } + assert_eq!(captured.authority.epoch, 1); + assert!(catalog.tables[&plan.authority.table] + .progress + .checkpoint + .is_empty()); + Ok(()) +} + +#[test] +fn descriptor_budget_digest_and_wire_type_limits_apply_before_admission() -> Result<()> { + let (_, plan) = fixture()?; + let layout = crate::replication::tests::layout(); + let limits = Limits { + max_descriptor_bytes: 512, + ..Default::default() + }; + let request = PrepareRequest { + authority: plan.authority, + transaction: plan.plan.transaction, + plan_index: 5, + descriptors: Descriptors::lance_fragments(Bytes::from(vec![7; 512]))?, + }; + request.clone().batch(&layout, &limits)?; + for mutation in 0..6 { + let mut invalid = request.clone(); + match mutation { + 0 => invalid.descriptors = Descriptors::lance_fragments(Bytes::from(vec![7; 513]))?, + 1 => invalid.descriptors.bytes = Bytes::new(), + 2 => invalid.descriptors.digest[0] ^= 1, + 3 => invalid.descriptors.format = 2, + 4 => invalid.authority.layout[0] ^= 1, + 5 => invalid.plan_index = 0, + _ => unreachable!(), + } + assert!(invalid.clone().batch(&layout, &limits).is_err()); + let batch = Command::Prepare(Box::new(invalid)).batch(&layout)?; + assert!(validate_record(&layout, 10, &limits, &batch.meta, &batch.body).is_err()); + } + assert!(Descriptors::lance_fragments(Bytes::from(vec![0; MAX_DESCRIPTOR_BYTES + 1])).is_err()); + let mut large = request; + large.descriptors = Descriptors::lance_fragments(Bytes::from(vec![7; MAX_DESCRIPTOR_BYTES]))?; + large.batch( + &layout, + &Limits { + max_descriptor_bytes: MAX_DESCRIPTOR_BYTES as u64, + ..Default::default() + }, + )?; + Ok(()) +} diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index f9017e5..2764fbf 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -9,8 +9,8 @@ use lance_context_wal::{ replication::{ checkpoint::{Image, Part}, publication::{ - Acquire, Destination, Plan, PlanRequest, Rejection as PublisherRejection, - Reply as PublisherReply, SourceRange, Table, + Acquire, Descriptors, Destination, Plan, PlanRequest, PrepareRequest, + Rejection as PublisherRejection, Reply as PublisherReply, SourceRange, Table, }, query::{Completion as ReadCompletion, Outcome as ReadOutcome, Query, Value}, source::{CatalogSigner, Permit, Reply as SourceReply}, @@ -62,6 +62,7 @@ impl WirePacket { } #[derive(Serialize, Deserialize)] enum Request { + Prepare(u64, PrepareRequest), Pin(u64, Permit), Query(u64, u64, Query), Plan(u64, PlanRequest), @@ -129,6 +130,7 @@ fn part(payload: bool) -> Part { fn run(worker: &mut Worker, request: Request) -> Result { let node = &mut worker.node; match 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)?, Request::Plan(token, plan) => node.plan_publication(token, plan)?, @@ -414,6 +416,7 @@ impl Cluster { Request::Query(group, token, _) => format!("query group {group}, token {token}"), Request::Acquire(token, _) => format!("acquire token {token}"), Request::Plan(token, _) => format!("plan token {token}"), + Request::Prepare(token, _) => format!("prepare token {token}"), Request::Pin(token, _) => format!("pin token {token}"), _ => "replication control".to_string(), }; @@ -1295,7 +1298,7 @@ fn disjoint_source_pins_survive_transfer_takeover_and_all_process_restart() -> R Request::Plan( 1021, PlanRequest { - authority, + authority: authority.clone(), plan: Plan { transaction, read_version: 7, @@ -1349,6 +1352,59 @@ fn disjoint_source_pins_survive_transfer_takeover_and_all_process_restart() -> R else { anyhow::bail!("source pin missing"); }; + drop(c.peers.remove(&3)); + let descriptors = Descriptors::lance_fragments(Bytes::from(vec![19; 192 * 1024]))?; + let prepared = PrepareRequest { + authority, + transaction, + plan_index, + descriptors: descriptors.clone(), + }; + c.command(catalog, Request::Prepare(1031, prepared.clone()))?; + c.pump(&[1])?; + assert!( + c.outcome(1031).is_none(), + "preparation preceded durable quorum" + ); + // The isolated pump dropped the first Append packets. Healing the link + // needs a heartbeat to resume replication, not just an empty queue drain. + for _ in 0..30 { + c.command(catalog, Request::Tick)?; + c.pump(&[1, 2])?; + if c.completions + .iter() + .any(|completion| completion.request == 1031) + { + break; + } + } + let prepared_outcome = c.outcome(1031); + let Some(Outcome::Committed(Applied::Publisher(PublisherReply::Prepared { + index: prepared_index, + }))) = prepared_outcome + else { + anyhow::bail!("prepared descriptors missing: {prepared_outcome:?}"); + }; + c.peers + .get_mut(&catalog) + .unwrap() + .call(Request::PrepareCheckpoint(10, Uuid::from_u128(820)))?; + c.command(catalog, Request::InstallCheckpoint)?; + c.pump(&[1, 2])?; + let before_catalog_transfer = c.transfers; + c.start(3)?; + for _ in 0..40 { + c.command(catalog, Request::Tick)?; + c.pump(&[1, 2, 3])?; + if c.transfers > before_catalog_transfer { + break; + } + } + assert!( + c.transfers > before_catalog_transfer, + "descriptors never crossed a catalog snapshot transfer" + ); + let before_source_transfer = c.transfers; c.peers .get_mut(&4) .unwrap() @@ -1359,12 +1415,12 @@ fn disjoint_source_pins_survive_transfer_takeover_and_all_process_restart() -> R for _ in 0..40 { c.command(4, Request::Tick)?; c.pump(&[4, 5, 6])?; - if c.transfers > 0 { + if c.transfers > before_source_transfer { break; } } assert!( - c.transfers > 0, + c.transfers > before_source_transfer, "source pin never crossed a snapshot transfer" ); drop(c.peers.remove(&4)); @@ -1382,6 +1438,34 @@ fn disjoint_source_pins_survive_transfer_takeover_and_all_process_restart() -> R PublisherReply::Acquired(_) ))) )); + let Response::Publisher(Some(taken_over)) = c + .peers + .get_mut(&catalog) + .unwrap() + .call(Request::Publisher(table))? + else { + anyhow::bail!("takeover table missing"); + }; + let mut retry = prepared.clone(); + retry.authority = taken_over.authority; + c.command(catalog, Request::Prepare(1043, retry.clone()))?; + c.pump(&[2, 3])?; + assert_eq!( + c.outcome(1043), + Some(Outcome::Committed(Applied::Publisher( + PublisherReply::Prepared { + index: prepared_index + } + ))) + ); + c.command(catalog, Request::Prepare(1044, prepared))?; + c.pump(&[2, 3])?; + assert_eq!( + c.outcome(1044), + Some(Outcome::Committed(Applied::Publisher( + PublisherReply::Rejected(PublisherRejection::Authority) + ))) + ); c.command(catalog, Request::Query(10, 1041, query.clone()))?; c.pump(&[2, 3])?; let Some(ReadOutcome::Observed { @@ -1434,8 +1518,40 @@ fn disjoint_source_pins_survive_transfer_takeover_and_all_process_restart() -> R }; assert_eq!(pin.index, index); assert_eq!(pin.permit.selection().plan_index, plan_index); + for id in 1..=3 { + let Response::Publisher(Some(recovered)) = c + .peers + .get_mut(&id) + .unwrap() + .call(Request::Publisher(table))? + else { + anyhow::bail!("restarted descriptor table missing"); + }; + let saved = recovered + .progress + .prepared + .context("restart lost descriptors")?; + assert_eq!(saved.index, prepared_index); + assert_eq!(saved.plan_index, plan_index); + assert_eq!(saved.transaction, transaction); + assert_eq!(saved.descriptors, descriptors); + } + c.command(catalog, Request::Prepare(1053, retry))?; + c.pump(&[1, 2, 3])?; + assert_eq!( + c.outcome(1053), + Some(Outcome::Committed(Applied::Publisher( + PublisherReply::Prepared { + index: prepared_index + } + ))) + ); for id in 4..=6 { verify_records(c.peers.get_mut(&id).unwrap(), 20, &rows)?; } Ok(()) } + +#[cfg(feature = "lance-integration")] +#[path = "replication_process/lance_publication.rs"] +mod lance_publication; diff --git a/crates/lance-context-wal/tests/replication_process/lance_publication.rs b/crates/lance-context-wal/tests/replication_process/lance_publication.rs new file mode 100644 index 0000000..090fdd2 --- /dev/null +++ b/crates/lance-context-wal/tests/replication_process/lance_publication.rs @@ -0,0 +1,373 @@ +//! Real Lance files and six persistent WAL processes. Collection is disabled; +//! this qualifies descriptor recovery, not the external commit/GC fence. +use super::*; +use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator, StringArray}; +use arrow_schema::{DataType, Field, Schema}; +use futures::TryStreamExt; +use lance::{ + dataset::{ + transaction::{Operation, Transaction}, + CommitBuilder, InsertBuilder, WriteMode, WriteParams, + }, + Dataset, +}; +use lance_context_wal::ingest_contract::{IngestContract, WritePolicy}; +use lance_table::{ + format::{Fragment, IndexMetadata, Manifest, Transaction as ManifestTransaction}, + io::commit::{ + CommitError, CommitHandler, ConditionalPutCommitHandler, ManifestLocation, + ManifestNamingScheme, ManifestWriter, + }, +}; + +fn payload(id: i64) -> String { + format!("{id}:{}", "source-bytes".repeat(4096)) +} +fn batch(ids: &[i64]) -> Result { + Ok(RecordBatch::try_new( + Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("payload", DataType::Utf8, false), + ])), + vec![ + Arc::new(Int64Array::from(ids.to_vec())), + Arc::new(StringArray::from( + ids.iter().map(|id| payload(*id)).collect::>(), + )), + ], + )?) +} +async fn verify(uri: &str, expected: &[i64]) -> Result<()> { + let dataset = Dataset::open(uri).await?; + let batches: Vec<_> = dataset + .scan() + .try_into_stream() + .await? + .try_collect() + .await?; + let mut rows = BTreeMap::new(); + for b in batches { + let ids = b.column(0).as_any().downcast_ref::().unwrap(); + let values = b.column(1).as_any().downcast_ref::().unwrap(); + for i in 0..b.num_rows() { + assert!(rows + .insert(ids.value(i), values.value(i).to_owned()) + .is_none()); + } + } + assert_eq!( + rows, + expected.iter().map(|id| (*id, payload(*id))).collect() + ); + Ok(()) +} + +#[derive(Debug)] +struct FixedTarget(u64); +#[async_trait::async_trait] +#[allow(clippy::too_many_arguments)] +impl CommitHandler for FixedTarget { + async fn commit( + &self, + manifest: &mut Manifest, + indices: Option>, + path: &object_store::path::Path, + store: &lance_io::object_store::ObjectStore, + writer: ManifestWriter, + naming: ManifestNamingScheme, + transaction: Option, + ) -> std::result::Result { + if manifest.version != self.0 { + return Err(CommitError::CommitConflict); + } + ConditionalPutCommitHandler + .commit(manifest, indices, path, store, writer, naming, transaction) + .await + } +} + +async fn recovered_publication(occupied: bool) -> Result<()> { + let key = ring::signature::Ed25519KeyPair::generate_pkcs8(&ring::rand::SystemRandom::new()) + .map_err(|_| anyhow::anyhow!("test key generation failed"))?; + let layout = Layout { + cluster: Uuid::from_u128(901), + nodes: (1..=6) + .map(|id| (id, format!("wal-{id}.example.invalid:7800"))) + .collect(), + groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![4, 5, 6])]), + buckets: vec![10, 20], + }; + let mut c = Cluster::with_layout(layout, Some(key.as_ref()))?; + let uri = c + .root + .path() + .join("dataset.lance") + .to_str() + .unwrap() + .to_owned(); + let genesis = batch(&[0])?; + let base = Dataset::write( + RecordBatchIterator::new(vec![Ok(genesis.clone())], genesis.schema()), + &uri, + Some(WriteParams { + commit_handler: Some(Arc::new(ConditionalPutCommitHandler)), + ..Default::default() + }), + ) + .await?; + let catalog = c.elect(1, 10, &[1, 2, 3])?; + let source = c.elect(4, 20, &[4, 5, 6])?; + let rows = batch(&[1, 2, 3, 4])?; + let mut wire = Vec::new(); + { + let mut encoder = arrow_ipc::writer::FileWriter::try_new(&mut wire, &rows.schema())?; + encoder.write(&rows)?; + encoder.finish()?; + } + let table = Uuid::from_u128(902); + let input = Batch::new(table, 1, Uuid::from_u128(903), 1, Bytes::from(wire)); + c.command(source, Request::Propose(20, 2000, input.clone()))?; + c.pump(&[4, 5, 6])?; + let Some(Outcome::Committed(Applied::Accepted { + index: source_index, + })) = c.outcome(2000) + else { + anyhow::bail!("WAL append missing"); + }; + let mut acquire = Acquire { + table, + destination: Destination { + namespace: Uuid::from_u128(904), + dataset: "tables/recovered".into(), + }, + ingest: IngestContract { + tenant: "test".into(), + qualified_table: "recovered".into(), + schema: lance_context_wal::ingest_contract::arrow_schema_fingerprint( + rows.schema().as_ref(), + )?, + policy: WritePolicy::ImmutableUniqueIds, + max_record_bytes: 1024 * 1024, + max_rows: 1024, + }, + owner: Uuid::from_u128(905), + request: Uuid::from_u128(906), + expected_epoch: 0, + }; + c.command(catalog, Request::Acquire(2001, acquire.clone()))?; + c.pump(&[1, 2, 3])?; + let Some(Outcome::Committed(Applied::Publisher(PublisherReply::Acquired(authority)))) = + c.outcome(2001) + else { + anyhow::bail!("authority missing"); + }; + let transaction = Uuid::from_u128(907); + c.command( + catalog, + Request::Plan( + 2002, + PlanRequest { + authority: authority.clone(), + plan: Plan { + transaction, + read_version: 1, + next_group: 10, + sources: BTreeMap::from([( + 20, + SourceRange { + after: 0, + through: source_index, + records: vec![(source_index, input.meta.clone())], + }, + )]), + }, + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(Outcome::Committed(Applied::Publisher(PublisherReply::Planned { index: plan_index }))) = + c.outcome(2002) + else { + anyhow::bail!("plan missing"); + }; + c.command( + catalog, + Request::Query( + 10, + 2003, + Query::Pin { + table, + transaction, + plan_index, + source_group: 20, + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + value: Value::Pin(Some(permit)), + .. + }) = c.read_outcome(2003) + else { + anyhow::bail!("attestation missing"); + }; + c.command(source, Request::Pin(2004, *permit))?; + c.pump(&[4, 5, 6])?; + assert!(matches!( + c.outcome(2004), + Some(Outcome::Committed(Applied::Source( + SourceReply::Pinned { .. } + ))) + )); + let Response::Record(Some(recovered_input)) = c + .peers + .get_mut(&source) + .unwrap() + .call(Request::Read(20, source_index))? + else { + anyhow::bail!("source payload missing"); + }; + assert_eq!(recovered_input, input); + let decoded = lance_context_wal::ipc::decode(recovered_input.body, rows.schema())?; + let staged = InsertBuilder::new(Arc::new(base)) + .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)?))?; + c.command( + catalog, + Request::Prepare( + 2005, + PrepareRequest { + authority, + transaction, + plan_index, + descriptors: descriptors.clone(), + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(Outcome::Committed(Applied::Publisher(PublisherReply::Prepared { + index: prepared_index, + }))) = c.outcome(2005) + else { + anyhow::bail!("preparation missing"); + }; + for (node, group) in [(catalog, 10), (source, 20)] { + c.peers + .get_mut(&node) + .unwrap() + .call(Request::PrepareCheckpoint(group, Uuid::new_v4()))?; + c.command(node, Request::InstallCheckpoint)?; + c.pump(&[1, 2, 3, 4, 5, 6])?; + } + // No publisher object or in-memory fragment vector supplies the retry. + drop(fragments); + c.queue.clear(); + c.peers.clear(); + for id in 1..=6 { + c.start(id)?; + } + let catalog = c.elect(2, 10, &[1, 2, 3])?; + acquire.owner = Uuid::from_u128(908); + acquire.request = Uuid::from_u128(909); + acquire.expected_epoch = 1; + c.command(catalog, Request::Acquire(2010, acquire))?; + c.pump(&[1, 2, 3])?; + c.command( + catalog, + Request::Query(10, 2011, Query::Publisher { table }), + )?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + value: Value::Publisher(Some(view)), + .. + }) = c.read_outcome(2011) + else { + anyhow::bail!("recovered plan missing"); + }; + let saved = view + .progress + .prepared + .context("recovery lost descriptors")?; + assert_eq!(saved.index, prepared_index); + assert_eq!(saved.plan_index, plan_index); + assert_eq!(saved.descriptors, descriptors); + let base = Dataset::open(&uri).await?; + if occupied { + InsertBuilder::new(Arc::new(base.clone())) + .with_params(&WriteParams { + mode: WriteMode::Append, + ..Default::default() + }) + .execute(vec![batch(&[99])?]) + .await?; + } + saved.descriptors.validate(1024 * 1024)?; + 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(); + let result = CommitBuilder::new(Arc::new(base)) + .with_commit_handler(Arc::new(FixedTarget(2))) + .with_max_retries(0) + .with_skip_auto_cleanup(true) + .execute(tx) + .await; + if occupied { + assert!(result.is_err(), "changed target must not rebase this plan"); + } else { + result?; + } + // Treat the response as lost: recover again and inspect the actual immutable + // target. No catalog resolution or source release is implemented here. + c.queue.clear(); + c.peers.clear(); + for id in 1..=6 { + c.start(id)?; + } + let latest = Dataset::open(&uri).await?; + assert_eq!(latest.version().version, 2); + let committed = latest + .read_transaction_by_version(2) + .await? + .context("missing actual transaction")?; + assert_eq!(committed.uuid == transaction.to_string(), !occupied); + verify(&uri, if occupied { &[0, 99] } else { &[0, 1, 2, 3, 4] }).await?; + for id in 1..=3 { + let Response::Publisher(Some(view)) = c + .peers + .get_mut(&id) + .unwrap() + .call(Request::Publisher(table))? + else { + anyhow::bail!("lost pending plan"); + }; + assert!(view.progress.checkpoint.is_empty()); + assert_eq!(view.progress.prepared.unwrap().descriptors, descriptors); + } + for id in 4..=6 { + verify_records( + c.peers.get_mut(&id).unwrap(), + 20, + std::slice::from_ref(&input), + )?; + } + Ok(()) +} + +#[tokio::test] +async fn recovered_descriptors_publish_exact_lance_payload_after_all_process_restart() -> Result<()> +{ + recovered_publication(false).await +} + +#[tokio::test] +async fn recovered_descriptors_do_not_retarget_an_occupied_lance_version() -> Result<()> { + recovered_publication(true).await +} diff --git a/docs/design/wal-prepared-descriptors.md b/docs/design/wal-prepared-descriptors.md new file mode 100644 index 0000000..b47b5da --- /dev/null +++ b/docs/design/wal-prepared-descriptors.md @@ -0,0 +1,45 @@ +# Durable materialization descriptors + +Refs #378. + +After source materialization, a publisher must recover the same immutable files +and fixed transaction target across retries and ownership changes. Rewriting +files during every ambiguous retry wastes I/O and can change which files a +transaction means. `Node::prepare_publication` records one descriptor set for an +exact pending `(table, transaction, plan_index)` through the catalog quorum. + +The current owner can register descriptors once. An exact retry returns the +original preparation index; different bytes are rejected. A stale owner is +rejected even when its bytes match. Takeover retains both the selected source +ranges and the chosen descriptor bytes. It does not reset the preparation slot. +Preparation and retries need no additional active-plan slot. + +Descriptors have a format identifier, bounded bytes and SHA-256 digest. Format 1 +is the materializer's JSON representation of Lance 13 fragments. WAL replicas +check the representation's length and digest without deserializing fragments or +Arrow bodies. The caller remains responsible for validating fragment semantics +and source dependencies. A stored descriptor is not proof that remote sources +have pinned their data, an external commit authorization, or a visibility ACK. + +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 +and images are rejected rather than silently migrated. + +The `lance-integration` feature adds real Lance qualification dependencies. It +is not required by ordinary WAL replicas. Process tests write Arrow IPC to a +source quorum, pin its selection, decode the recovered source, write real Lance +files, and record their descriptors in the catalog. After restarting all six +processes, they recover the descriptors and attempt the same fixed target. +A second writer occupying that target must not cause the recovered plan to +rebase to another version. Full payload reads and actual transaction identities +are checked independently of the commit response. + +These tests retain manifests and data files and disable automatic cleanup. +External manifest authority, durable outcome resolution, source closure and +collection remain necessary before enabling publication-driven retirement. +The fixed-target test handler is not a production GC fence; the delayed-write +cleanup race in the publication-GC regressions still requires coordinated +version identity and admission state. diff --git a/docs/design/wal-publication-plans.md b/docs/design/wal-publication-plans.md index d8f1822..fe844ed 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 v4. Existing v1/v2/v3 directories and images are rejected; there is no in-place migration +format to v5. Existing v1/v2/v3/v4 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 107ab5c..8f917b4 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 v4, including [immutable publication plans](wal-publication-plans.md) +format are v5, including [immutable publication plans](wal-publication-plans.md) and [source pins](wal-source-pins.md). -Earlier v1/v2/v3 application directories are rejected; this optional component does +Earlier v1/v2/v3/v4 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 7c00e15..3673768 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 v4 and rejects older identities +The application/checkpoint format advances to v5 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