diff --git a/crates/lance-context-wal/src/raft_log/checkpoint.rs b/crates/lance-context-wal/src/raft_log/checkpoint.rs index 0c7b7a1..9ed9564 100644 --- a/crates/lance-context-wal/src/raft_log/checkpoint.rs +++ b/crates/lance-context-wal/src/raft_log/checkpoint.rs @@ -101,7 +101,7 @@ impl GroupLog { /// committed/uncommitted suffix and HardState. The same cut may only be /// retried with the identical descriptor. Remote snapshot installation and /// physical segment deletion are separate operations, not supplied here. - /// The current replication::Node does not invoke this interface. + /// replication::Node installs its validated application images here. pub fn checkpoint(&self, snapshot: &Snapshot) -> Result<()> { let mut state = self .state diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 918a18b..3e3e762 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -6,11 +6,14 @@ //! commit. An isolated former leader cannot answer retries from local state. //! //! The owner schedules ticks, bounded network I/O and calls on one thread. This -//! increment retains the full log and bounded metadata: it is not the production -//! service, a snapshot/retention implementation, or a bounded disk allocator. +//! component supports explicit application checkpoints and prepared snapshot +//! transfer. It retains bounded metadata and all image files: it is not the +//! production service, a physical retention policy, or a bounded disk allocator. //! Message scope checks are not authentication. Do not expose it to untrusted //! transport or enable ingestion without the remaining admission/lifecycle work. +pub mod checkpoint; +mod images; mod state; #[cfg(test)] mod tests; @@ -30,6 +33,7 @@ use crate::{ raft_log::{GroupLog, PhysicalLog}, record::{Batch, BatchMeta}, }; +pub use images::{CheckpointTask, PreparedCheckpoint, SnapshotReceiver}; use state::State; /// Immutable, replicated-state bounds. All peers and reopens must use the same @@ -143,6 +147,8 @@ struct Group { state: State, // Index AND term prevents suffix replacement from completing another input. pending: BTreeMap<(u64, u64), (u64, BatchMeta)>, + image: Option>, + incoming: Option, } /// Owns all Raft groups in one shared physical engine. Do not call concurrently @@ -155,6 +161,7 @@ pub struct Node { limits: Limits, id: u64, groups: BTreeMap, + images: Arc, failed: bool, } @@ -172,6 +179,7 @@ impl Node { limits: self.limits.clone(), id: self.id, groups: BTreeMap::from([(id, group)]), + images: self.images.clone(), failed: self.failed, }) .collect() @@ -182,6 +190,7 @@ impl Node { let identity = limits.identity()?; let policy = Sha256::digest(&identity).into(); let log = PhysicalLog::open(directory, &layout, id)?; + let images = images::Owner::new(directory)?; let mut groups = BTreeMap::new(); let logger = slog::Logger::root(slog::Discard, slog::o!()); for (group, voters) in &layout.groups { @@ -191,7 +200,13 @@ impl Node { let store = log.group(*group)?; store.bind_application(&identity)?; let committed = store.initial_state()?.hard_state.commit; - let mut state = State::default(); + let (mut state, image) = if store.first_index()? > 1 { + let snapshot = store.snapshot(store.first_index()? - 1, id)?; + let loaded = checkpoint::load(directory, &snapshot, &layout, &limits, *group)?; + (loaded.state, Some(loaded.image)) + } else { + (State::default(), None) + }; while state.applied < committed { let entries = store.entries( state.applied + 1, @@ -225,6 +240,8 @@ impl Node { raw, state, pending: BTreeMap::new(), + image, + incoming: None, }, ); } @@ -235,6 +252,7 @@ impl Node { limits, id, groups, + images, failed: false, }) } @@ -327,8 +345,7 @@ impl Node { Ok(()) } - pub fn step(&mut self, packet: Packet) -> Result<()> { - self.healthy()?; + fn validate_packet(&self, packet: &Packet) -> Result<()> { ensure!( packet.cluster == self.layout.cluster && packet.layout == self.layout_hash @@ -340,11 +357,23 @@ impl Node { .groups .get(&packet.group) .context("unknown group")?; - let message = packet.message; + let message = &packet.message; ensure!( message.to == self.id && message.from != self.id && voters.contains(&message.from), "invalid Raft peer route" ); + ensure!(self.groups.contains_key(&packet.group), "unhosted group"); + ensure!( + message.compute_size() <= 16 * 1024 * 1024, + "oversized message" + ); + Ok(()) + } + + pub fn step(&mut self, packet: Packet) -> Result<()> { + self.healthy()?; + self.validate_packet(&packet)?; + let message = packet.message; ensure!( matches!( message.get_msg_type(), @@ -360,8 +389,8 @@ impl Node { "unsupported peer message" ); ensure!( - message.compute_size() <= 16 * 1024 * 1024 && message.snapshot.is_none(), - "oversized message or unsupported snapshot" + message.snapshot.is_none(), + "snapshot requires receive_snapshot and step_snapshot" ); for entry in &message.entries { state::validate_entry(&self.layout, packet.group, &self.limits, entry, true)?; @@ -391,8 +420,11 @@ impl Node { let mut outbound = Vec::new(); while group.raw.has_ready() { let mut ready = group.raw.ready(); - ensure!(ready.snapshot().is_empty(), "snapshot integration required"); - group.raw.store().persist(ready.entries(), ready.hs())?; + if ready.snapshot().is_empty() { + group.raw.store().persist(ready.entries(), ready.hs())?; + } else { + images::install_received(group, &ready, *id, &mut events)?; + } outbound.extend(ready.take_messages()); outbound.extend(ready.take_persisted_messages()); apply( @@ -483,21 +515,11 @@ impl Node { let Some(meta) = group.state.accepted.get(&index) else { return Ok(None); }; - let mut entries = group.raw.store().entries( - index, - index + 1, - None, - raft::GetEntriesContext::empty(false), - )?; - let entry = entries.pop().context("missing accepted payload")?; - ensure!( - &state::decode_meta(&entry.context)? == meta, - "accepted metadata mismatch" - ); - meta.validate_payload(&entry.data, self.limits.max_record_bytes as usize)?; + let body = images::read_body(group.raw.store(), group.image.as_deref(), index, meta)?; + meta.validate_payload(&body, self.limits.max_record_bytes as usize)?; Ok(Some(Batch { meta: meta.clone(), - body: entry.data, + body, })) } } diff --git a/crates/lance-context-wal/src/replication/checkpoint.rs b/crates/lance-context-wal/src/replication/checkpoint.rs new file mode 100644 index 0000000..cc2df77 --- /dev/null +++ b/crates/lance-context-wal/src/replication/checkpoint.rs @@ -0,0 +1,628 @@ +//! Immutable application images and bounded file transfer. Payloads stay opaque; +//! building and loading an image never decode Arrow or concatenate its bodies. +//! File I/O belongs on a bounded worker, outside the Raft owner. Only a later +//! durable Raft transition can install a prepared image. Files are retained on +//! errors; physical collection and publication-aware retirement are separate. + +use super::*; +use bincode::Options; +use bytes::Bytes; +use raft_proto::eraftpb::{ConfState, Snapshot}; +use std::{ + fs::{File, OpenOptions}, + io::{BufReader, BufWriter, Read, Write}, + path::PathBuf, +}; + +const FORMAT: u32 = 1; +const MAGIC: &[u8; 8] = b"LCWALCP1"; +pub const CHUNK_BYTES: usize = 1024 * 1024; + +fn codec() -> impl Options { + bincode::DefaultOptions::new() + .with_fixint_encoding() + .with_little_endian() + .with_limit(4096) + .reject_trailing_bytes() +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct Binding { + cluster: Uuid, + layout: [u8; 32], + policy: [u8; 32], + group: u64, + index: u64, + term: u64, + voters: [u64; 3], +} + +impl Binding { + fn new(layout: &Layout, limits: &Limits, group: u64, index: u64, term: u64) -> Result { + let layout_hash = layout.fingerprint()?; + ensure!( + index > 0 && index < u64::MAX && term > 0, + "invalid image cut" + ); + Ok(Self { + cluster: layout.cluster, + layout: layout_hash, + policy: Sha256::digest(limits.identity()?).into(), + group, + index, + term, + voters: layout.groups.get(&group).context("unknown image group")?[..].try_into()?, + }) + } + + fn validate(&self, layout: &Layout, limits: &Limits, group: u64) -> Result<()> { + ensure!( + *self == Self::new(layout, limits, group, self.index, self.term)?, + "foreign application image scope, policy or membership" + ); + Ok(()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct Descriptor { + format: u32, + binding: Binding, + id: Uuid, + records: u64, + producers: u64, + metadata_bytes: u64, + payload_bytes: u64, + metadata_digest: [u8; 32], + payload_digest: [u8; 32], +} + +fn metadata_limit(limits: &Limits) -> u64 { + // A framed record needs at most 8 + 4 + 256 bytes, plus a bounded header. + // limits.identity() validates all policy maxima before this is used. + (limits.max_records + limits.max_producers) * 268 + 1024 +} + +impl Descriptor { + fn validate(&self, layout: &Layout, limits: &Limits, group: u64) -> Result<()> { + self.binding.validate(layout, limits, group)?; + ensure!( + self.format == FORMAT + && !self.id.is_nil() + && self.records <= limits.max_records + && self.producers <= limits.max_producers + && self.producers <= self.records + && self.metadata_bytes <= metadata_limit(limits) + && self.payload_bytes <= self.records * limits.max_record_bytes, + "application image format or bounds mismatch" + ); + Ok(()) + } + + fn snapshot(&self) -> Result { + let mut snapshot = Snapshot::default(); + let metadata = snapshot.mut_metadata(); + metadata.index = self.binding.index; + metadata.term = self.binding.term; + metadata.set_conf_state(ConfState { + voters: self.binding.voters.to_vec(), + ..Default::default() + }); + snapshot.data = codec().serialize(self)?.into(); + ensure!( + snapshot.compute_size() <= 4096, + "oversized image descriptor" + ); + Ok(snapshot) + } + + fn decode(snapshot: &Snapshot, layout: &Layout, limits: &Limits, group: u64) -> Result { + ensure!( + snapshot.compute_size() <= 4096, + "oversized image descriptor" + ); + let descriptor: Self = codec().deserialize(&snapshot.data)?; + descriptor.validate(layout, limits, group)?; + ensure!( + descriptor.snapshot()? == *snapshot, + "image descriptor and Raft cut disagree" + ); + Ok(descriptor) + } + + fn paths(&self, directory: &Path) -> [PathBuf; 2] { + let name = format!("context-application-{}-{}", self.binding.group, self.id); + [ + directory.join(format!("{name}.state")), + directory.join(format!("{name}.pack")), + ] + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Part { + Metadata, + Payload, +} +impl Part { + fn slot(self) -> usize { + match self { + Self::Metadata => 0, + Self::Payload => 1, + } + } +} + +#[derive(Clone, Debug)] +struct Location { + offset: u64, + bytes: usize, + digest: [u8; 32], +} + +/// Open immutable files keep an outgoing transfer or reader attached to its +/// captured image. The current increment does not delete any image files. +pub struct Image { + descriptor: Descriptor, + files: [File; 2], + #[cfg(not(unix))] + paths: [PathBuf; 2], + locations: BTreeMap, +} +impl std::fmt::Debug for Image { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Image") + .field("descriptor", &self.descriptor) + .finish_non_exhaustive() + } +} + +impl Image { + pub fn snapshot(&self) -> Result { + self.descriptor.snapshot() + } + + pub fn bytes(&self, part: Part) -> u64 { + match part { + Part::Metadata => self.descriptor.metadata_bytes, + Part::Payload => self.descriptor.payload_bytes, + } + } + + /// Return at most one MiB from a captured image. Transport authentication, + /// deadlines and node-wide network/disk admission belong to the caller. + pub fn chunk(&self, part: Part, offset: u64, bytes: usize) -> Result { + ensure!( + (1..=CHUNK_BYTES).contains(&bytes) + && offset + .checked_add(bytes as u64) + .is_some_and(|end| end <= self.bytes(part)), + "invalid image chunk range" + ); + self.read_at(part, offset, bytes) + } + + fn read_at(&self, part: Part, offset: u64, bytes: usize) -> Result { + let slot = part.slot(); + ensure!( + self.files[slot].metadata()?.len() == self.bytes(part), + "image file size changed" + ); + let mut body = vec![0; bytes]; + #[cfg(unix)] + { + use std::os::unix::fs::FileExt; + self.files[slot].read_exact_at(&mut body, offset)?; + } + #[cfg(not(unix))] + { + use std::io::{Seek, SeekFrom}; + // Independently opened cursors avoid try_clone's shared seek offset. + let mut file = File::open(&self.paths[slot])?; + file.seek(SeekFrom::Start(offset))?; + file.read_exact(&mut body)?; + } + Ok(body.into()) + } + + /// The build/read owner validates this body against its captured BatchMeta. + /// Keep hashing at that common boundary, once for pack and log inputs alike. + pub(super) fn read_for_validation(&self, index: u64) -> Result> { + let Some(location) = self.locations.get(&index) else { + return Ok(None); + }; + Ok(Some(self.read_at( + Part::Payload, + location.offset, + location.bytes, + )?)) + } +} + +pub(super) struct Loaded { + pub image: Arc, + pub state: State, +} + +pub(super) struct Cut { + pub group: u64, + pub index: u64, + pub term: u64, + pub id: Uuid, +} + +struct Writer { + file: BufWriter, + digest: Sha256, + bytes: u64, + limit: u64, + failed: bool, +} +impl Writer { + fn new(path: &Path, limit: u64) -> Result { + Ok(Self { + file: BufWriter::new(OpenOptions::new().create_new(true).write(true).open(path)?), + digest: Sha256::new(), + bytes: 0, + limit, + failed: false, + }) + } + fn finish(mut self) -> Result<(u64, [u8; 32])> { + ensure!(!self.failed, "image writer failed; abandon this candidate"); + self.file.flush()?; + self.file.get_ref().sync_all()?; + Ok((self.bytes, self.digest.finalize().into())) + } +} +impl Write for Writer { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + if self.failed { + return Err(std::io::Error::other("image writer failed")); + } + if (bytes.len() as u64) > self.limit - self.bytes { + return Err(std::io::Error::other("image file exceeds bound")); + } + let n = match self.file.write(bytes) { + Ok(n) => n, + Err(error) => { + self.failed = true; + return Err(error); + } + }; + self.bytes += n as u64; + self.digest.update(&bytes[..n]); + Ok(n) + } + fn flush(&mut self) -> std::io::Result<()> { + let result = self.file.flush(); + if result.is_err() { + self.failed = true; + } + result + } +} + +struct Reader { + file: BufReader, + digest: Sha256, + bytes: u64, +} +impl Reader { + fn new(path: &Path, expected: u64) -> Result { + let file = File::open(path)?; + ensure!( + file.metadata()?.len() == expected, + "image file length mismatch" + ); + Ok(Self { + file: BufReader::new(file), + digest: Sha256::new(), + bytes: 0, + }) + } + fn finish(mut self, bytes: u64, digest: [u8; 32]) -> Result { + let mut extra = [0]; + ensure!( + self.read(&mut extra)? == 0 && self.bytes == bytes, + "trailing image bytes" + ); + ensure!( + <[u8; 32]>::from(self.digest.finalize()) == digest, + "image file checksum mismatch" + ); + Ok(self.file.into_inner()) + } +} +impl Read for Reader { + fn read(&mut self, bytes: &mut [u8]) -> std::io::Result { + let n = self.file.read(bytes)?; + self.bytes += n as u64; + self.digest.update(&bytes[..n]); + Ok(n) + } +} + +fn write_u64(out: &mut impl Write, value: u64) -> Result<()> { + out.write_all(&value.to_le_bytes())?; + Ok(()) +} +fn read_u64(input: &mut impl Read) -> Result { + let mut bytes = [0; 8]; + input.read_exact(&mut bytes)?; + Ok(u64::from_le_bytes(bytes)) +} +fn write_frame(out: &mut impl Write, bytes: &[u8]) -> Result<()> { + out.write_all(&u32::try_from(bytes.len())?.to_le_bytes())?; + out.write_all(bytes)?; + Ok(()) +} +fn read_frame(input: &mut impl Read) -> Result<([u8; N], usize)> { + let mut length = [0; 4]; + input.read_exact(&mut length)?; + let length = u32::from_le_bytes(length) as usize; + ensure!(length > 0 && length <= N, "oversized image metadata frame"); + let mut bytes = [0; N]; + input.read_exact(&mut bytes[..length])?; + Ok((bytes, length)) +} +fn write_meta(out: &mut impl Write, index: u64, meta: &BatchMeta) -> Result<()> { + write_u64(out, index)?; + write_frame(out, &state::encode_meta(meta)?) +} +fn read_meta(input: &mut impl Read) -> Result<(u64, BatchMeta)> { + let index = read_u64(input)?; + let (bytes, n) = read_frame::<256>(input)?; + Ok((index, state::decode_meta(&bytes[..n])?)) +} + +fn locations(state: &State) -> Result<(BTreeMap, u64)> { + let mut offset = 0u64; + let mut locations = BTreeMap::new(); + for (index, meta) in &state.accepted { + locations.insert( + *index, + Location { + offset, + bytes: usize::try_from(meta.body_bytes)?, + digest: meta.digest, + }, + ); + offset = offset + .checked_add(meta.body_bytes) + .context("image payload size overflow")?; + } + Ok((locations, offset)) +} + +/// Inputs must be a captured, immutable applied cut. The owner retains its log +/// and prior image while this worker reads one bounded body at a time. +pub(super) fn build( + directory: &Path, + layout: &Layout, + limits: &Limits, + cut: Cut, + state: State, + mut read: impl FnMut(u64, &BatchMeta) -> Result, +) -> Result { + let Cut { + group, + index, + term, + id, + } = cut; + let binding = Binding::new(layout, limits, group, index, term)?; + state.validate_checkpoint(layout, group, limits, binding.index)?; + ensure!(!id.is_nil(), "nil application image id"); + let (locations, payload_bytes) = locations(&state)?; + let mut descriptor = Descriptor { + format: FORMAT, + binding, + id, + records: state.accepted.len() as u64, + producers: state.producers.len() as u64, + metadata_bytes: 0, + payload_bytes, + metadata_digest: [0; 32], + payload_digest: [0; 32], + }; + let paths = descriptor.paths(directory); + let mut metadata = Writer::new(&paths[0], metadata_limit(limits))?; + metadata.write_all(MAGIC)?; + write_frame(&mut metadata, &codec().serialize(&descriptor.binding)?)?; + write_u64(&mut metadata, descriptor.records)?; + write_u64(&mut metadata, descriptor.producers)?; + for (index, meta) in &state.accepted { + write_meta(&mut metadata, *index, meta)?; + } + for (meta, index) in state.producers.values() { + write_meta(&mut metadata, *index, meta)?; + } + (descriptor.metadata_bytes, descriptor.metadata_digest) = metadata.finish()?; + let mut pack = Writer::new(&paths[1], payload_bytes)?; + for (index, meta) in &state.accepted { + let body = read(*index, meta)?; + meta.validate_payload(&body, limits.max_record_bytes as usize)?; + pack.write_all(&body)?; + } + let (bytes, digest) = pack.finish()?; + ensure!(bytes == payload_bytes, "image pack length mismatch"); + descriptor.payload_digest = digest; + descriptor.validate(layout, limits, group)?; + File::open(directory)?.sync_all()?; + let files = [File::open(&paths[0])?, File::open(&paths[1])?]; + Ok(Loaded { + image: Arc::new(Image { + descriptor, + files, + locations, + #[cfg(not(unix))] + paths, + }), + state, + }) +} + +pub(super) fn load( + directory: &Path, + snapshot: &Snapshot, + layout: &Layout, + limits: &Limits, + group: u64, +) -> Result { + let descriptor = Descriptor::decode(snapshot, layout, limits, group)?; + let paths = descriptor.paths(directory); + let mut metadata = Reader::new(&paths[0], descriptor.metadata_bytes)?; + let mut magic = [0; 8]; + metadata.read_exact(&mut magic)?; + ensure!(&magic == MAGIC, "unsupported application image format"); + let (binding, n) = read_frame::<512>(&mut metadata)?; + ensure!( + codec().deserialize::(&binding[..n])? == descriptor.binding, + "image metadata binding mismatch" + ); + ensure!( + read_u64(&mut metadata)? == descriptor.records + && read_u64(&mut metadata)? == descriptor.producers, + "image metadata count mismatch" + ); + let mut state = State { + applied: descriptor.binding.index, + ..Default::default() + }; + let mut previous = 0; + for _ in 0..descriptor.records { + let (index, meta) = read_meta(&mut metadata)?; + ensure!( + index > previous, + "unordered or duplicate accepted image index" + ); + state.accepted.insert(index, meta); + previous = index; + } + let mut previous = None; + for _ in 0..descriptor.producers { + let (index, meta) = read_meta(&mut metadata)?; + let key = (meta.stream, meta.bucket, meta.producer); + ensure!( + previous.is_none_or(|old| old < key), + "unordered or duplicate image producer" + ); + state.producers.insert(key, (meta, index)); + previous = Some(key); + } + let metadata_file = metadata.finish(descriptor.metadata_bytes, descriptor.metadata_digest)?; + state.validate_checkpoint(layout, group, limits, descriptor.binding.index)?; + let (locations, payload_bytes) = locations(&state)?; + ensure!( + payload_bytes == descriptor.payload_bytes, + "image record lengths disagree with pack" + ); + let mut pack = Reader::new(&paths[1], payload_bytes)?; + let mut scratch = [0; 64 * 1024]; + for location in locations.values() { + let mut remaining = location.bytes; + let mut hash = Sha256::new(); + while remaining > 0 { + let n = remaining.min(scratch.len()); + pack.read_exact(&mut scratch[..n])?; + hash.update(&scratch[..n]); + remaining -= n; + } + ensure!( + <[u8; 32]>::from(hash.finalize()) == location.digest, + "image record checksum mismatch" + ); + } + let pack_file = pack.finish(descriptor.payload_bytes, descriptor.payload_digest)?; + Ok(Loaded { + image: Arc::new(Image { + descriptor, + files: [metadata_file, pack_file], + locations, + #[cfg(not(unix))] + paths, + }), + state, + }) +} + +/// One sequential receiver on a bounded blocking worker. No received bytes are +/// visible to Raft before finish validates and syncs both complete files. +pub(super) struct Receiver { + directory: PathBuf, + descriptor: Descriptor, + layout: Arc, + limits: Limits, + writers: [Writer; 2], +} +impl Receiver { + pub fn new( + directory: &Path, + snapshot: &Snapshot, + layout: Arc, + limits: Limits, + group: u64, + local_id: Uuid, + ) -> Result { + let mut descriptor = Descriptor::decode(snapshot, &layout, &limits, group)?; + ensure!( + !local_id.is_nil() && local_id != descriptor.id, + "receiver needs a fresh local image id" + ); + descriptor.id = local_id; + let paths = descriptor.paths(directory); + let writers = [ + Writer::new(&paths[0], descriptor.metadata_bytes)?, + Writer::new(&paths[1], descriptor.payload_bytes)?, + ]; + Ok(Self { + directory: directory.to_path_buf(), + descriptor, + layout, + limits, + writers, + }) + } + + pub fn write(&mut self, part: Part, offset: u64, bytes: &[u8]) -> Result<()> { + ensure!( + !bytes.is_empty() && bytes.len() <= CHUNK_BYTES, + "invalid received image chunk size" + ); + let writer = &mut self.writers[part.slot()]; + ensure!(offset == writer.bytes, "noncontiguous received image chunk"); + writer.write_all(bytes)?; + Ok(()) + } + + pub fn finish(self) -> Result { + let expected = [ + ( + self.descriptor.metadata_bytes, + self.descriptor.metadata_digest, + ), + ( + self.descriptor.payload_bytes, + self.descriptor.payload_digest, + ), + ]; + for (writer, expected) in self.writers.into_iter().zip(expected) { + ensure!( + writer.finish()? == expected, + "incomplete or corrupt received image" + ); + } + File::open(&self.directory)?.sync_all()?; + load( + &self.directory, + &self.descriptor.snapshot()?, + &self.layout, + &self.limits, + self.descriptor.binding.group, + ) + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/lance-context-wal/src/replication/checkpoint/tests.rs b/crates/lance-context-wal/src/replication/checkpoint/tests.rs new file mode 100644 index 0000000..d8eb101 --- /dev/null +++ b/crates/lance-context-wal/src/replication/checkpoint/tests.rs @@ -0,0 +1,291 @@ +use super::*; +use std::fs; + +fn layout() -> Arc { + super::super::tests::layout() +} + +fn fixture() -> Result<(State, BTreeMap)> { + let mut state = State::default(); + let mut bodies = BTreeMap::new(); + // Blank, exact retry and rejected entries leave gaps in accepted indexes. + for (index, sequence) in [(1, 0), (2, 1), (3, 1), (4, 3), (5, 2), (6, 0)] { + let mut entry = Entry { + index, + term: 2, + ..Default::default() + }; + if sequence > 0 { + let batch = super::super::tests::batch(sequence); + entry.context = state::encode_meta(&batch.meta)?.into(); + entry.data = batch.body.clone(); + bodies.insert(index, batch.body); + } + state.apply(&entry, &Limits::default())?; + } + Ok((state, bodies)) +} + +fn create(directory: &Path, id: u128) -> Result { + let (state, bodies) = fixture()?; + build( + directory, + &layout(), + &Limits::default(), + Cut { + group: 10, + index: state.applied, + term: 2, + id: Uuid::from_u128(id), + }, + state, + |index, _| Ok(bodies[&index].clone()), + ) +} + +fn copy(source: &Image, receiver: &mut Receiver) -> Result<()> { + for part in [Part::Metadata, Part::Payload] { + let mut offset = 0; + while offset < source.bytes(part) { + let bytes = source.chunk( + part, + offset, + (source.bytes(part) - offset).min(997) as usize, + )?; + receiver.write(part, offset, &bytes)?; + offset += bytes.len() as u64; + } + } + Ok(()) +} + +#[test] +fn exact_state_payloads_and_streamed_transfer_without_changing_source_image() -> Result<()> { + let source = tempfile::tempdir()?; + let target = tempfile::tempdir()?; + let built = create(source.path(), 101)?; + let snapshot = built.image.snapshot()?; + let restored = load(source.path(), &snapshot, &layout(), &Limits::default(), 10)?; + assert_eq!(built.state, restored.state); + assert_eq!( + restored.state.accepted.keys().copied().collect::>(), + [2, 5] + ); + assert_eq!(restored.state.producers.values().next().unwrap().1, 5); + let mut receiver = Receiver::new( + target.path(), + &snapshot, + layout(), + Limits::default(), + 10, + Uuid::from_u128(102), + )?; + copy(&built.image, &mut receiver)?; + let received = receiver.finish()?; + assert_eq!(received.state, restored.state); + assert_ne!(received.image.descriptor.id, built.image.descriptor.id); + assert_eq!(built.image.snapshot()?, snapshot); + for index in [2, 5] { + assert_eq!( + restored.image.read_for_validation(index)?, + received.image.read_for_validation(index)? + ); + assert_eq!( + received.image.read_for_validation(index)?.unwrap(), + fixture()?.1[&index] + ); + } + assert!(received.image.read_for_validation(3)?.is_none()); + assert!(built.image.chunk(Part::Payload, 0, 0).is_err()); + assert!(built + .image + .chunk(Part::Payload, 0, CHUNK_BYTES + 1) + .is_err()); + assert!(built.image.chunk(Part::Payload, u64::MAX, 1).is_err()); + assert!( + create(source.path(), 101).is_err(), + "existing immutable files overwritten" + ); + assert_eq!( + load(source.path(), &snapshot, &layout(), &Limits::default(), 10)?.state, + restored.state + ); + Ok(()) +} + +#[test] +fn received_images_reject_partial_corrupt_and_out_of_order_bytes() -> Result<()> { + let source = tempfile::tempdir()?; + let target = tempfile::tempdir()?; + let built = create(source.path(), 111)?; + let snapshot = built.image.snapshot()?; + let receive = |id| { + Receiver::new( + target.path(), + &snapshot, + layout(), + Limits::default(), + 10, + Uuid::from_u128(id), + ) + }; + let mut partial = receive(112)?; + let first = built.image.chunk(Part::Metadata, 0, 8)?; + assert!(partial.write(Part::Metadata, 1, &first).is_err()); + partial.write(Part::Metadata, 0, &first)?; + assert!( + partial.finish().is_err(), + "partial state became installable" + ); + let mut bad = receive(113)?; + for part in [Part::Metadata, Part::Payload] { + let mut bytes = built + .image + .chunk(part, 0, built.image.bytes(part) as usize)? + .to_vec(); + if part == Part::Payload { + bytes[0] ^= 1; + } + bad.write(part, 0, &bytes)?; + } + assert!( + bad.finish().is_err(), + "wrong payload hash became installable" + ); + let mut good = receive(114)?; + copy(&built.image, &mut good)?; + assert_eq!(good.finish()?.state, built.state); + assert!( + receive(114).is_err(), + "interrupted or existing files overwritten" + ); + Ok(()) +} + +#[test] +fn installed_files_fail_closed_on_corruption_missing_parts_and_foreign_binding() -> Result<()> { + let root = tempfile::tempdir()?; + let built = create(root.path(), 121)?; + let snapshot = built.image.snapshot()?; + let policy = Limits { + max_records: 999, + ..Default::default() + }; + assert!(load(root.path(), &snapshot, &layout(), &policy, 10).is_err()); + assert!(load(root.path(), &snapshot, &layout(), &Limits::default(), 20).is_err()); + let mut cut = snapshot.clone(); + cut.mut_metadata().index += 1; + assert!(load(root.path(), &cut, &layout(), &Limits::default(), 10).is_err()); + let paths = built.image.descriptor.paths(root.path()); + let original = fs::read(&paths[1])?; + let mut corrupt = original.clone(); + corrupt[0] ^= 1; + fs::write(&paths[1], &corrupt)?; + assert!(built.state.accepted[&2] + .validate_payload( + &built.image.read_for_validation(2)?.unwrap(), + 8 * 1024 * 1024 + ) + .is_err()); + assert!(load(root.path(), &snapshot, &layout(), &Limits::default(), 10).is_err()); + // Even a matching outer digest cannot hide a record/metadata mismatch. + let mut descriptor = built.image.descriptor.clone(); + descriptor.payload_digest = Sha256::digest(&corrupt).into(); + assert!(load( + root.path(), + &descriptor.snapshot()?, + &layout(), + &Limits::default(), + 10 + ) + .is_err()); + fs::write(&paths[1], original)?; + fs::remove_file(&paths[0])?; + assert!(load(root.path(), &snapshot, &layout(), &Limits::default(), 10).is_err()); + Ok(()) +} + +#[test] +fn metadata_counts_frames_trailing_bytes_and_exact_producer_tails_are_validated() -> Result<()> { + let root = tempfile::tempdir()?; + let built = create(root.path(), 131)?; + let path = built.image.descriptor.paths(root.path())[0].clone(); + let original = fs::read(&path)?; + let try_bytes = |bytes: &[u8]| -> Result<()> { + fs::write(&path, bytes)?; + let mut descriptor = built.image.descriptor.clone(); + descriptor.metadata_bytes = bytes.len() as u64; + descriptor.metadata_digest = Sha256::digest(bytes).into(); + assert!(load( + root.path(), + &descriptor.snapshot()?, + &layout(), + &Limits::default(), + 10 + ) + .is_err()); + Ok(()) + }; + let mut trailing = original.clone(); + trailing.push(0); + try_bytes(&trailing)?; + let mut frame = original.clone(); + frame[8..12].copy_from_slice(&u32::MAX.to_le_bytes()); + try_bytes(&frame)?; + let binding_bytes = u32::from_le_bytes(original[8..12].try_into()?) as usize; + let mut count = original.clone(); + count[12 + binding_bytes..20 + binding_bytes].copy_from_slice(&u64::MAX.to_le_bytes()); + try_bytes(&count)?; + fs::write(&path, original)?; + let (state, _) = fixture()?; + for mutation in 0..4 { + let mut invalid = state.clone(); + match mutation { + 0 => invalid.producers.clear(), + 1 => invalid.producers.values_mut().next().unwrap().1 = 2, + 2 => invalid.accepted.get_mut(&5).unwrap().sequence = 3, + _ => { + invalid.accepted.insert(7, invalid.accepted[&5].clone()); + } + } + assert!(invalid + .validate_checkpoint(&layout(), 10, &Limits::default(), 6) + .is_err()); + } + Ok(()) +} + +#[test] +fn empty_application_cut_has_a_real_empty_pack_and_no_payload_fetch() -> Result<()> { + let root = tempfile::tempdir()?; + let state = State { + applied: 1, + ..Default::default() + }; + let built = build( + root.path(), + &layout(), + &Limits::default(), + Cut { + group: 10, + index: 1, + term: 1, + id: Uuid::from_u128(141), + }, + state.clone(), + |_, _| anyhow::bail!("unexpected payload fetch"), + )?; + assert_eq!(built.image.bytes(Part::Payload), 0); + assert_eq!( + load( + root.path(), + &built.image.snapshot()?, + &layout(), + &Limits::default(), + 10 + )? + .state, + state + ); + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/images.rs b/crates/lance-context-wal/src/replication/images.rs new file mode 100644 index 0000000..5caf350 --- /dev/null +++ b/crates/lance-context-wal/src/replication/images.rs @@ -0,0 +1,328 @@ +use super::*; +use checkpoint::{Cut, Image, Loaded, Part}; +use raft_proto::eraftpb::Snapshot; +use std::{ + path::PathBuf, + sync::atomic::{AtomicBool, Ordering}, +}; + +pub(super) struct Owner { + directory: PathBuf, + busy: AtomicBool, +} +impl Owner { + pub fn new(directory: &Path) -> Result> { + Ok(Arc::new(Self { + directory: directory.canonicalize()?, + busy: AtomicBool::new(false), + })) + } + fn acquire(self: &Arc) -> Result { + self.busy + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .map_err(|_| anyhow::anyhow!("node image builder/receiver busy"))?; + Ok(Permit(self.clone())) + } +} +struct Permit(Arc); +impl Drop for Permit { + fn drop(&mut self) { + self.0.busy.store(false, Ordering::Release); + } +} + +/// Captures a committed application cut without copying payload bodies. Run it +/// on a bounded blocking worker; the node may keep polling and appending. One +/// build/receive permit spans every group sharing this physical node. It also +/// prevents another checkpoint from compacting this task's captured log input. +pub struct CheckpointTask { + permit: Permit, + store: GroupLog, + previous: Option>, + layout: Arc, + limits: Limits, + cut: Cut, + state: State, +} +impl CheckpointTask { + pub fn run(self) -> Result { + let Self { + permit, + store, + previous, + layout, + limits, + cut, + state, + } = self; + let group = cut.group; + let loaded = checkpoint::build( + &permit.0.directory, + &layout, + &limits, + cut, + state, + |index, meta| read_body(&store, previous.as_deref(), index, meta), + )?; + Ok(PreparedCheckpoint { + permit, + group, + loaded, + received: None, + }) + } +} + +/// Complete, validated and synced files. This value cannot be constructed from +/// unverified descriptors. Holding it reserves the node's image worker slot; +/// dropping it abandons the candidate and retains its files as orphans. +pub struct PreparedCheckpoint { + permit: Permit, + group: u64, + loaded: Loaded, + received: Option, +} +impl PreparedCheckpoint { + pub fn image(&self) -> Arc { + self.loaded.image.clone() + } +} + +/// Bounded transfer state. Socket I/O stays outside the Raft owner; move this +/// object to the admitted worker and feed exact, sequential file chunks. +pub struct SnapshotReceiver { + permit: Permit, + packet: Packet, + receiver: checkpoint::Receiver, +} +impl SnapshotReceiver { + pub fn write(&mut self, part: Part, offset: u64, bytes: &[u8]) -> Result<()> { + self.receiver.write(part, offset, bytes) + } + pub fn finish(self) -> Result { + Ok(PreparedCheckpoint { + permit: self.permit, + group: self.packet.group, + loaded: self.receiver.finish()?, + received: Some(self.packet), + }) + } +} + +pub(super) fn read_body( + store: &GroupLog, + image: Option<&Image>, + index: u64, + meta: &BatchMeta, +) -> Result { + if let Some(image) = image { + if let Some(body) = image.read_for_validation(index)? { + return Ok(body); + } + } + let mut entries = store.entries( + index, + index + 1, + None, + raft::GetEntriesContext::empty(false), + )?; + let entry = entries.pop().context("missing application image input")?; + ensure!(entry.index == index, "image input index mismatch"); + ensure!( + &state::decode_meta(&entry.context)? == meta, + "accepted metadata mismatch" + ); + Ok(entry.data) +} + +impl Node { + pub fn prepare_checkpoint(&self, group: u64, id: Uuid) -> Result { + self.healthy()?; + ensure!(!id.is_nil(), "nil application image id"); + let g = self.groups.get(&group).context("unhosted group")?; + let cut = Cut { + group, + id, + index: g.state.applied, + term: g.raw.store().term(g.state.applied)?, + }; + ensure!( + cut.index >= g.raw.store().first_index()?, + "no new committed state to checkpoint" + ); + Ok(CheckpointTask { + permit: self.images.acquire()?, + store: g.raw.store().clone(), + previous: g.image.clone(), + layout: self.layout.clone(), + limits: self.limits.clone(), + cut, + state: g.state.clone(), + }) + } + + /// Install a completed local build. Inputs after the captured cut remain in + /// the log. The full image is now available for a lagging follower's transfer; + /// callers must service MsgSnapshot before enabling periodic compaction. + /// This logically compacts only; it does not bound retained disk space. + pub fn install_checkpoint(&mut self, prepared: PreparedCheckpoint) -> Result<()> { + self.healthy()?; + ensure!( + Arc::ptr_eq(&prepared.permit.0, &self.images), + "foreign node image worker" + ); + ensure!( + prepared.received.is_none(), + "received image needs step_snapshot" + ); + let g = self + .groups + .get_mut(&prepared.group) + .context("unhosted group")?; + let snapshot = prepared.loaded.image.snapshot()?; + ensure!( + prepared.loaded.state.applied <= g.state.applied, + "checkpoint exceeds applied state" + ); + if let Err(error) = g.raw.store().checkpoint(&snapshot) { + self.failed = true; + return Err(error); + } + g.image = Some(prepared.loaded.image); + Ok(()) + } + + /// Capture immediately after poll, before installing another checkpoint. + /// Keep the returned Arc until the outgoing transfer finishes. A transport + /// can discard an obsolete message and report failure so Raft retries. + pub fn snapshot_image(&self, group: u64, snapshot: &Snapshot) -> Result> { + self.healthy()?; + let image = self + .groups + .get(&group) + .context("unhosted group")? + .image + .as_ref() + .context("no installed application image")?; + ensure!( + image.snapshot()? == *snapshot, + "outgoing snapshot is obsolete" + ); + Ok(image.clone()) + } + + pub fn receive_snapshot(&self, packet: Packet, local_id: Uuid) -> Result { + self.healthy()?; + self.validate_packet(&packet)?; + ensure!( + packet.message.get_msg_type() == MessageType::MsgSnapshot + && packet.message.entries.is_empty() + && packet.message.has_snapshot(), + "expected snapshot-only message" + ); + let permit = self.images.acquire()?; + let receiver = checkpoint::Receiver::new( + &self.images.directory, + packet.message.get_snapshot(), + self.layout.clone(), + self.limits.clone(), + packet.group, + local_id, + )?; + Ok(SnapshotReceiver { + permit, + packet, + receiver, + }) + } + + /// Only a complete image can reach RawNode. It may now be stale and ignored; + /// otherwise poll persists the received snapshot, HardState and new suffix + /// atomically before releasing the follower response. Keep polling during + /// download; no Ready is retained across a network or file wait. + pub fn step_snapshot(&mut self, mut prepared: PreparedCheckpoint) -> Result<()> { + self.healthy()?; + ensure!( + Arc::ptr_eq(&prepared.permit.0, &self.images), + "foreign node image worker" + ); + let mut packet = prepared + .received + .take() + .context("local image needs install_checkpoint")?; + self.validate_packet(&packet)?; + let snapshot = prepared.loaded.image.snapshot()?; + packet.message.set_snapshot(snapshot.clone()); + let g = self + .groups + .get_mut(&prepared.group) + .context("unhosted group")?; + ensure!( + g.incoming.is_none(), + "poll the previous received snapshot first" + ); + g.raw.step(packet.message)?; + if g.raw.raft.raft_log.unstable.snapshot.as_ref() == Some(&snapshot) { + g.incoming = Some(prepared); + } + Ok(()) + } + + pub fn report_snapshot(&mut self, group: u64, to: u64, delivered: bool) -> Result<()> { + self.healthy()?; + ensure!( + to != self.id + && self + .layout + .groups + .get(&group) + .context("unknown group")? + .contains(&to), + "invalid snapshot destination" + ); + self.groups + .get_mut(&group) + .context("unhosted group")? + .raw + .report_snapshot( + to, + if delivered { + raft::SnapshotStatus::Finish + } else { + raft::SnapshotStatus::Failure + }, + ); + Ok(()) + } +} + +pub(super) fn install_received( + g: &mut Group, + ready: &raft::Ready, + id: u64, + events: &mut Events, +) -> Result<()> { + let prepared = g + .incoming + .take() + .context("received Raft snapshot has no prepared application image")?; + ensure!( + prepared.loaded.image.snapshot()? == *ready.snapshot(), + "prepared image differs from Ready" + ); + g.raw.store().install_snapshot( + ready.snapshot(), + ready.entries(), + ready.hs().context("received snapshot has no HardState")?, + )?; + g.state = prepared.loaded.state; + g.image = Some(prepared.loaded.image); + for (_, (request, _)) in std::mem::take(&mut g.pending) { + events.completions.push(Completion { + group: id, + request, + outcome: Outcome::Unknown, + }); + } + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/state.rs b/crates/lance-context-wal/src/replication/state.rs index 451fbf7..276d510 100644 --- a/crates/lance-context-wal/src/replication/state.rs +++ b/crates/lance-context-wal/src/replication/state.rs @@ -25,6 +25,16 @@ pub(super) fn validate_meta( body: &[u8], digest: bool, ) -> Result<()> { + validate_identity(layout, group, meta)?; + if digest { + meta.validate_payload(body, limits.max_record_bytes as usize)?; + } else { + meta.check_body_length(body.len(), limits.max_record_bytes as usize)?; + } + Ok(()) +} + +fn validate_identity(layout: &Layout, group: u64, meta: &BatchMeta) -> Result<()> { ensure!( meta.kind == RecordKind::Data && !meta.stream.is_nil() @@ -33,11 +43,6 @@ pub(super) fn validate_meta( && layout.group_for_bucket(meta.bucket)? == group, "invalid data identity, record kind or group placement" ); - if digest { - meta.validate_payload(body, limits.max_record_bytes as usize)?; - } else { - meta.check_body_length(body.len(), limits.max_record_bytes as usize)?; - } Ok(()) } pub(super) fn validate_entry( @@ -68,13 +73,60 @@ pub(super) fn validate_entry( ) } -#[derive(Default)] +#[derive(Clone, Debug, Default, PartialEq, Eq)] pub(super) struct State { pub applied: u64, pub accepted: BTreeMap, pub producers: BTreeMap<(Uuid, u32, Uuid), (BatchMeta, u64)>, } impl State { + /// The v1 image retains every accepted record. Producer tails must exactly + /// match those records, including sequence continuity within each lane. + /// Retirement will need a new format with independently retained tails. + pub fn validate_checkpoint( + &self, + layout: &Layout, + group: u64, + limits: &Limits, + cut: u64, + ) -> Result<()> { + ensure!( + cut > 0 && cut < u64::MAX && self.applied == cut, + "invalid application checkpoint cut" + ); + ensure!( + self.accepted.len() as u64 <= limits.max_records + && self.producers.len() as u64 <= limits.max_producers, + "application checkpoint exceeds state limits" + ); + let mut tails = BTreeMap::new(); + for (index, meta) in &self.accepted { + ensure!( + *index > 0 && *index <= cut, + "accepted index outside checkpoint" + ); + validate_identity(layout, group, meta)?; + meta.check_body_length( + usize::try_from(meta.body_bytes)?, + limits.max_record_bytes as usize, + )?; + let key = (meta.stream, meta.bucket, meta.producer); + let expected = tails + .get(&key) + .map_or(1, |(old, _): &(BatchMeta, u64)| old.sequence + 1); + ensure!( + meta.sequence == expected, + "checkpoint producer sequence gap" + ); + tails.insert(key, (meta.clone(), *index)); + } + ensure!( + tails == self.producers, + "checkpoint producer tails disagree" + ); + Ok(()) + } + pub fn apply(&mut self, entry: &Entry, limits: &Limits) -> Result> { ensure!( entry.index == self.applied + 1, diff --git a/crates/lance-context-wal/src/replication/tests.rs b/crates/lance-context-wal/src/replication/tests.rs index ac877f6..06f20c9 100644 --- a/crates/lance-context-wal/src/replication/tests.rs +++ b/crates/lance-context-wal/src/replication/tests.rs @@ -1,7 +1,7 @@ use super::*; use bytes::Bytes; -fn layout() -> Arc { +pub(super) fn layout() -> Arc { Arc::new(Layout { cluster: Uuid::from_u128(1), nodes: (1..=3) @@ -12,7 +12,7 @@ fn layout() -> Arc { }) } -fn batch(sequence: u64) -> Batch { +pub(super) fn batch(sequence: u64) -> Batch { Batch::new( Uuid::from_u128(2), 0, @@ -41,6 +41,271 @@ fn pump(nodes: &mut [Node]) -> Result> { anyhow::bail!("message pump did not settle") } +fn pump_images( + nodes: &mut [Node], + allowed: &[u64], + next_image: &mut u128, +) -> Result<(Vec, usize)> { + let mut completions = Vec::new(); + let mut transfers = 0; + for _ in 0..100 { + let mut messages = Vec::new(); + for node in nodes.iter_mut().filter(|n| allowed.contains(&n.id)) { + let events = node.poll()?; + completions.extend(events.completions); + for packet in events.messages { + let image = if packet.message.get_msg_type() == MessageType::MsgSnapshot { + Some(node.snapshot_image(packet.group, packet.message.get_snapshot())?) + } else { + None + }; + messages.push((packet, image)); + } + } + if messages.is_empty() { + return Ok((completions, transfers)); + } + for (packet, image) in messages { + let target = packet.message.to; + if !allowed.contains(&target) { + if image.is_some() { + nodes[packet.message.from as usize - 1].report_snapshot( + packet.group, + target, + false, + )?; + } + continue; + } + let node = &mut nodes[target as usize - 1]; + if let Some(image) = image { + *next_image += 1; + let mut receiver = node.receive_snapshot(packet, Uuid::from_u128(*next_image))?; + for part in [checkpoint::Part::Metadata, checkpoint::Part::Payload] { + let mut offset = 0; + while offset < image.bytes(part) { + let bytes = image.chunk( + part, + offset, + (image.bytes(part) - offset).min(1024) as usize, + )?; + receiver.write(part, offset, &bytes)?; + offset += bytes.len() as u64; + } + } + node.step_snapshot(receiver.finish()?)?; + transfers += 1; + } else { + node.step(packet)?; + } + } + } + anyhow::bail!("image message pump did not settle") +} + +#[test] +fn application_images_keep_payloads_tails_suffix_and_lagging_peer_recovery() -> Result<()> { + let directory = tempfile::tempdir()?; + let open = || { + (1..=3) + .map(|id| { + Node::open( + &directory.path().join(id.to_string()), + layout(), + id, + Limits::default(), + ) + }) + .collect::>>() + }; + let mut nodes = open()?; + nodes[0].campaign(10)?; + nodes[0].campaign(20)?; + pump(&mut nodes)?; + let mut next_image = 1000; + nodes[0].propose(10, 1, batch(1))?; + let (events, _) = pump_images(&mut nodes, &[1, 2], &mut next_image)?; + let first = events.into_iter().find(|c| c.request == 1).unwrap().outcome; + // Capture, then let ordinary append/commit proceed before its worker runs. + let task = nodes[0].prepare_checkpoint(10, Uuid::from_u128(100))?; + assert!( + nodes[0] + .prepare_checkpoint(20, Uuid::from_u128(101)) + .is_err(), + "shared worker admission ignored" + ); + nodes[0].propose(10, 2, batch(2))?; + let (_, transfers) = pump_images(&mut nodes, &[1, 2], &mut next_image)?; + assert_eq!(transfers, 0); + let prepared = std::thread::spawn(move || task.run()).join().unwrap()?; + let old_image = prepared.image(); + let cut = old_image.snapshot()?.get_metadata().index; + nodes[0].install_checkpoint(prepared)?; + assert_eq!(nodes[0].groups[&10].raw.store().first_index()?, cut + 1); + assert!( + nodes[0].groups[&10].state.applied > cut, + "suffix was not exercised" + ); + assert_eq!(nodes[0].groups[&10].state.accepted.len(), 2); + let latest = nodes[0] + .prepare_checkpoint(10, Uuid::from_u128(102))? + .run()?; + nodes[0].install_checkpoint(latest)?; + assert!( + nodes[0] + .prepare_checkpoint(10, Uuid::from_u128(103)) + .is_err(), + "same cut rebuilt with a different identity" + ); + assert!( + !old_image.chunk(checkpoint::Part::Payload, 0, 1)?.is_empty(), + "captured old transfer lost its files" + ); + let mut transfers = 0; + for _ in 0..10 { + nodes[0].tick()?; + transfers += pump_images(&mut nodes, &[1, 2, 3], &mut next_image)?.1; + if nodes[2].status(10)?.accepted_records == 2 { + break; + } + } + assert!( + transfers > 0, + "lagging peer recovered without testing snapshot transfer" + ); + assert!(nodes[2].groups[&10].raw.store().first_index()? > 1); + assert_eq!(nodes[2].groups[&10].state, nodes[0].groups[&10].state); + drop(nodes); + let mut nodes = open()?; + for node in &nodes { + for ((index, _), sequence) in node.accepted(10, 0, 128)?.entries.into_iter().zip(1..=2) { + assert_eq!(node.read_accepted(10, index)?, Some(batch(sequence))); + } + assert_eq!(node.accepted(10, 0, 128)?.entries.len(), 2); + } + nodes[0].campaign(10)?; + pump_images(&mut nodes, &[1, 2, 3], &mut next_image)?; + // Latest retry survives packing; an older sequence is still rejected. + nodes[0].propose(10, 3, batch(2))?; + let events = pump_images(&mut nodes, &[1, 2, 3], &mut next_image)?.0; + assert!(matches!( + events[0].outcome, + Outcome::Committed(Applied::Accepted { .. }) + )); + nodes[0].propose(10, 4, batch(1))?; + let events = pump_images(&mut nodes, &[1, 2, 3], &mut next_image)?.0; + assert_eq!( + events[0].outcome, + Outcome::Committed(Applied::Rejected { + expected_sequence: 3 + }) + ); + assert!(matches!( + first, + Outcome::Committed(Applied::Accepted { .. }) + )); + nodes[0].propose(10, 5, batch(3))?; + pump_images(&mut nodes, &[1, 2, 3], &mut next_image)?; + assert_eq!(nodes[2].status(10)?.accepted_records, 3); + Ok(()) +} + +#[test] +fn application_snapshot_sync_failure_fences_sibling_groups_and_recovers_complete_files( +) -> Result<()> { + let directory = tempfile::tempdir()?; + let mut nodes = (1..=3) + .map(|id| { + Node::open( + &directory.path().join(id.to_string()), + layout(), + id, + Limits::default(), + ) + }) + .collect::>>()?; + nodes[0].campaign(10)?; + pump(&mut nodes)?; + nodes[0].propose(10, 1, batch(1))?; + pump(&mut nodes)?; + let prepared = nodes[0] + .prepare_checkpoint(10, Uuid::from_u128(201))? + .run()?; + let accepted = nodes[0].accepted(10, 0, 128)?; + nodes[0].groups[&10].raw.store().fail_next_sync_for_test(); + assert!(nodes[0].install_checkpoint(prepared).is_err()); + assert!(nodes[0].status(20).is_err()); + assert!(nodes[0].poll().is_err()); + drop(nodes); + let node = Node::open(&directory.path().join("1"), layout(), 1, Limits::default())?; + assert_eq!(node.accepted(10, 0, 128)?, accepted); + assert_eq!( + node.read_accepted(10, accepted.entries[0].0)?, + Some(batch(1)) + ); + Ok(()) +} + +#[test] +fn received_application_snapshot_sync_failure_releases_no_response() -> Result<()> { + let directory = tempfile::tempdir()?; + let mut nodes = (1..=3) + .map(|id| { + Node::open( + &directory.path().join(id.to_string()), + layout(), + id, + Limits::default(), + ) + }) + .collect::>>()?; + nodes[0].campaign(10)?; + pump(&mut nodes)?; + nodes[0].propose(10, 1, batch(1))?; + let mut image_id = 10_000; + pump_images(&mut nodes, &[1, 2], &mut image_id)?; + let prepared = nodes[0] + .prepare_checkpoint(10, Uuid::from_u128(301))? + .run()?; + let image = prepared.image(); + nodes[0].install_checkpoint(prepared)?; + let mut message = Message { + from: 1, + to: 3, + term: nodes[0].status(10)?.term, + ..Default::default() + }; + message.set_msg_type(MessageType::MsgSnapshot); + message.set_snapshot(image.snapshot()?); + let packet = Packet { + cluster: nodes[0].layout.cluster, + layout: nodes[0].layout_hash, + policy: nodes[0].policy, + group: 10, + message, + }; + let mut receiver = nodes[2].receive_snapshot(packet, Uuid::from_u128(302))?; + for part in [checkpoint::Part::Metadata, checkpoint::Part::Payload] { + let body = image.chunk(part, 0, image.bytes(part) as usize)?; + receiver.write(part, 0, &body)?; + } + nodes[2].step_snapshot(receiver.finish()?)?; + nodes[2].groups[&10].raw.store().fail_next_sync_for_test(); + assert!( + nodes[2].poll().is_err(), + "snapshot response escaped a failed sync" + ); + assert!(nodes[2].status(20).is_err()); + drop(nodes); + let recovered = Node::open(&directory.path().join("3"), layout(), 3, Limits::default())?; + let entries = recovered.accepted(10, 0, 128)?.entries; + assert!(entries.len() <= 1); + if let Some((index, _)) = entries.first() { + assert_eq!(recovered.read_accepted(10, *index)?, Some(batch(1))); + } + Ok(()) +} + #[test] fn no_quorum_ack_before_peer_persistence_and_proposal_keeps_body_allocation() -> Result<()> { let directory = tempfile::tempdir()?; diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index 3b91948..7d49f1b 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -6,7 +6,11 @@ use bytes::Bytes; use lance_context_wal::{ placement::Layout, record::Batch, - replication::{Applied, Completion, Events, Limits, Node, Outcome, Packet, Page, Status}, + replication::{ + checkpoint::{Image, Part}, + Applied, Completion, Events, Limits, Node, Outcome, Packet, Page, PreparedCheckpoint, + SnapshotReceiver, Status, + }, }; use protobuf::Message as _; use serde::{de::DeserializeOwned, Deserialize, Serialize}; @@ -59,6 +63,14 @@ enum Request { Status(u64), Read(u64, u64), Page(u64), + PrepareCheckpoint(u64, Uuid), + InstallCheckpoint, + ImageInfo(Vec), + ImageChunk(Vec, bool, u64, usize), + BeginSnapshot(WirePacket, Uuid), + ReceiveChunk(bool, u64, Bytes), + FinishSnapshot, + ReportSnapshot(u64, u64, bool), } #[derive(Serialize, Deserialize)] enum Response { @@ -67,6 +79,9 @@ enum Response { Record(Option), Page(Page), Error(String), + Image(Vec, u64, u64), + Chunk(Bytes), + Ok, } fn send(stream: &mut TcpStream, value: &T) -> Result<()> { let bytes = bincode::serialize(value)?; @@ -84,7 +99,23 @@ fn receive(stream: &mut TcpStream) -> Result { stream.read_exact(&mut bytes)?; Ok(bincode::deserialize(&bytes)?) } -fn run(node: &mut Node, request: Request) -> Result { +struct Worker { + node: Node, + prepared: Option, + receiver: Option, + images: BTreeMap, Arc>, +} + +fn part(payload: bool) -> Part { + if payload { + Part::Payload + } else { + Part::Metadata + } +} + +fn run(worker: &mut Worker, request: Request) -> Result { + let node = &mut worker.node; match request { Request::Campaign(group) => node.campaign(group)?, Request::Propose(group, token, batch) => node.propose(group, token, batch)?, @@ -95,11 +126,78 @@ fn run(node: &mut Node, request: Request) -> Result { return Ok(Response::Record(node.read_accepted(group, index)?)) } Request::Page(group) => return Ok(Response::Page(node.accepted(group, 0, 128)?)), + Request::PrepareCheckpoint(group, id) => { + ensure!(worker.prepared.is_none(), "existing prepared image"); + let prepared = node.prepare_checkpoint(group, id)?.run()?; + let image = prepared.image(); + let snapshot = image.snapshot()?.write_to_bytes()?; + let response = Response::Image( + snapshot.clone(), + image.bytes(Part::Metadata), + image.bytes(Part::Payload), + ); + worker.images.insert(snapshot, image); + worker.prepared = Some(prepared); + return Ok(response); + } + Request::InstallCheckpoint => { + node.install_checkpoint(worker.prepared.take().context("no prepared image")?)? + } + Request::ImageInfo(snapshot) => { + let image = worker + .images + .get(&snapshot) + .context("no captured transfer image")?; + return Ok(Response::Image( + snapshot, + image.bytes(Part::Metadata), + image.bytes(Part::Payload), + )); + } + Request::ImageChunk(snapshot, payload, offset, bytes) => { + let image = worker + .images + .get(&snapshot) + .context("no captured transfer image")?; + return Ok(Response::Chunk(image.chunk( + part(payload), + offset, + bytes, + )?)); + } + Request::BeginSnapshot(packet, id) => { + ensure!(worker.receiver.is_none(), "existing receiver"); + worker.receiver = Some(node.receive_snapshot(packet.decode()?, id)?); + return Ok(Response::Ok); + } + Request::ReceiveChunk(payload, offset, bytes) => { + worker.receiver.as_mut().context("no receiver")?.write( + part(payload), + offset, + &bytes, + )?; + return Ok(Response::Ok); + } + Request::FinishSnapshot => { + node.step_snapshot(worker.receiver.take().context("no receiver")?.finish()?)? + } + Request::ReportSnapshot(group, to, delivered) => { + node.report_snapshot(group, to, delivered)? + } } let Events { messages, completions, } = node.poll()?; + for packet in &messages { + if packet.message.get_msg_type() == raft_proto::eraftpb::MessageType::MsgSnapshot { + let snapshot = packet.message.get_snapshot(); + worker.images.insert( + snapshot.write_to_bytes()?, + node.snapshot_image(packet.group, snapshot)?, + ); + } + } Ok(Response::Events( messages .into_iter() @@ -118,7 +216,7 @@ fn replication_peer_fixture() -> Result<()> { let root = PathBuf::from(root); let id = std::env::var("CONTEXT_REPLICATION_NODE")?.parse()?; let listener = TcpListener::bind("127.0.0.1:0")?; - let mut node = Node::open( + let node = Node::open( &root.join(format!("data-{id}")), Arc::new(serde_json::from_slice(&std::fs::read( root.join("layout.json"), @@ -126,6 +224,12 @@ fn replication_peer_fixture() -> Result<()> { id, Limits::default(), )?; + let mut worker = Worker { + node, + prepared: None, + receiver: None, + images: BTreeMap::new(), + }; let address = PathBuf::from(std::env::var("CONTEXT_REPLICATION_ADDRESS_FILE")?); let temporary = address.with_extension("tmp"); std::fs::write(&temporary, listener.local_addr()?.to_string())?; @@ -137,7 +241,7 @@ fn replication_peer_fixture() -> Result<()> { loop { let request = receive(&mut stream)?; let response = - run(&mut node, request).unwrap_or_else(|error| Response::Error(format!("{error:#}"))); + run(&mut worker, request).unwrap_or_else(|error| Response::Error(format!("{error:#}"))); send(&mut stream, &response)?; } } @@ -210,6 +314,8 @@ struct Cluster { incarnations: BTreeMap, queue: VecDeque, completions: Vec, + transfers: u64, + interrupt_next_snapshot: bool, } impl Cluster { fn new() -> Result { @@ -232,6 +338,8 @@ impl Cluster { incarnations: BTreeMap::new(), queue: VecDeque::new(), completions: Vec::new(), + transfers: 0, + interrupt_next_snapshot: false, }; for id in 1..=3 { this.start(id)?; @@ -272,15 +380,97 @@ impl Cluster { if allowed.contains(&packet.message.from) && allowed.contains(&packet.message.to) && self.peers.contains_key(&packet.message.to) + { + if packet.message.get_msg_type() == raft_proto::eraftpb::MessageType::MsgSnapshot { + self.transfer(packet)?; + } else { + self.command( + packet.message.to, + Request::Step(WirePacket::encode(packet)?), + )?; + } + } else if packet.message.get_msg_type() == raft_proto::eraftpb::MessageType::MsgSnapshot + && allowed.contains(&packet.message.from) + && self.peers.contains_key(&packet.message.from) { self.command( - packet.message.to, - Request::Step(WirePacket::encode(packet)?), + packet.message.from, + Request::ReportSnapshot(packet.group, packet.message.to, false), )?; } } anyhow::bail!("test message queue did not settle") } + + fn transfer(&mut self, packet: Packet) -> Result<()> { + let from = packet.message.from; + let to = packet.message.to; + let group = packet.group; + let snapshot = packet.message.get_snapshot().write_to_bytes()?; + let Response::Image(_, metadata, payload) = self + .peers + .get_mut(&from) + .context("missing image source")? + .call(Request::ImageInfo(snapshot.clone()))? + else { + anyhow::bail!("wrong image info"); + }; + self.transfers += 1; + let response = self + .peers + .get_mut(&to) + .unwrap() + .call(Request::BeginSnapshot( + WirePacket::encode(packet)?, + Uuid::from_u128(10_000 + u128::from(self.transfers)), + ))?; + ensure!( + matches!(response, Response::Ok), + "snapshot receiver did not start" + ); + for (is_payload, total) in [(false, metadata), (true, payload)] { + let mut offset = 0; + while offset < total { + let size = if self.interrupt_next_snapshot { + 8 + } else { + (total - offset).min(64 * 1024) as usize + }; + let Response::Chunk(bytes) = + self.peers + .get_mut(&from) + .unwrap() + .call(Request::ImageChunk( + snapshot.clone(), + is_payload, + offset, + size, + ))? + else { + anyhow::bail!("wrong image chunk"); + }; + ensure!(bytes.len() == size, "short transfer chunk"); + let response = self + .peers + .get_mut(&to) + .unwrap() + .call(Request::ReceiveChunk(is_payload, offset, bytes))?; + ensure!(matches!(response, Response::Ok), "snapshot chunk rejected"); + if self.interrupt_next_snapshot { + self.interrupt_next_snapshot = false; + // SIGKILL leaves only an incomplete, uninstalled image. The + // restart below must use the original durable committed state. + drop(self.peers.remove(&to)); + self.command(from, Request::ReportSnapshot(group, to, false))?; + return Ok(()); + } + offset += size as u64; + } + } + self.command(to, Request::FinishSnapshot)?; + self.command(from, Request::ReportSnapshot(group, to, true))?; + Ok(()) + } fn outcome(&mut self, token: u64) -> Option { self.completions .iter() @@ -332,6 +522,136 @@ fn input(bucket: u32, sequence: u64, fill: u8) -> Batch { ) } +fn verify_records(peer: &mut Peer, group: u64, expected: &[Batch]) -> Result<()> { + let Response::Page(page) = peer.call(Request::Page(group))? else { + anyhow::bail!("wrong page"); + }; + assert_eq!(page.entries.len(), expected.len()); + for ((index, meta), expected) in page.entries.into_iter().zip(expected) { + assert_eq!(&meta, &expected.meta); + let Response::Record(Some(actual)) = peer.call(Request::Read(group, index))? else { + anyhow::bail!("missing payload"); + }; + assert_eq!(&actual, expected); + } + Ok(()) +} + +#[test] +fn application_snapshot_transfer_interruption_and_full_process_restart() -> Result<()> { + let mut c = Cluster::new()?; + let first = input(0, 1, 31); + let second = input(0, 2, 32); + let sibling = input(1, 1, 33); + c.command(1, Request::Campaign(10))?; + c.command(1, Request::Campaign(20))?; + c.pump(&[1, 2, 3])?; + c.command(1, Request::Propose(20, 100, sibling.clone()))?; + c.pump(&[1, 2, 3])?; + c.command(1, Request::Propose(10, 101, first.clone()))?; + c.pump(&[1, 2])?; + assert!(matches!( + c.outcome(101), + Some(Outcome::Committed(Applied::Accepted { .. })) + )); + let response = c + .peers + .get_mut(&1) + .unwrap() + .call(Request::PrepareCheckpoint(10, Uuid::from_u128(501)))?; + assert!(matches!(response, Response::Image(_, _, _))); + // Kill after complete file sync but BEFORE the Raft descriptor transition. + drop(c.peers.remove(&1)); + c.start(1)?; + verify_records( + c.peers.get_mut(&1).unwrap(), + 10, + std::slice::from_ref(&first), + )?; + let leader = c.elect(1, 10, &[1, 2])?; + let response = c + .peers + .get_mut(&leader) + .unwrap() + .call(Request::PrepareCheckpoint(10, Uuid::from_u128(502)))?; + assert!(matches!(response, Response::Image(_, _, _))); + c.command(leader, Request::InstallCheckpoint)?; + c.pump(&[1, 2])?; + // Keep a real suffix after the packed cut so restart/transfer must use both. + c.command(leader, Request::Propose(10, 102, second.clone()))?; + c.pump(&[1, 2])?; + let second_ack = c.outcome(102).context("second append had no ACK")?; + assert!(matches!( + second_ack, + Outcome::Committed(Applied::Accepted { .. }) + )); + c.interrupt_next_snapshot = true; + for _ in 0..30 { + c.command(leader, Request::Tick)?; + c.pump(&[1, 2, 3])?; + if !c.peers.contains_key(&3) { + break; + } + } + assert!( + !c.peers.contains_key(&3), + "partial snapshot kill was not reached" + ); + assert_eq!(c.transfers, 1); + c.start(3)?; + verify_records(c.peers.get_mut(&3).unwrap(), 10, &[])?; + verify_records( + c.peers.get_mut(&3).unwrap(), + 20, + std::slice::from_ref(&sibling), + )?; + for _ in 0..30 { + c.command(leader, Request::Tick)?; + c.pump(&[1, 2, 3])?; + let Response::Status(status) = c.peers.get_mut(&3).unwrap().call(Request::Status(10))? + else { + anyhow::bail!("wrong status"); + }; + if status.accepted_records == 2 { + break; + } + } + assert!(c.transfers >= 2, "snapshot retry was not exercised"); + for peer in c.peers.values_mut() { + verify_records(peer, 10, &[first.clone(), second.clone()])?; + verify_records(peer, 20, std::slice::from_ref(&sibling))?; + } + // Kill all peers after descriptor installation; preserve original volumes. + c.queue.clear(); + c.peers.clear(); + for id in 1..=3 { + c.start(id)?; + } + for peer in c.peers.values_mut() { + verify_records(peer, 10, &[first.clone(), second.clone()])?; + verify_records(peer, 20, std::slice::from_ref(&sibling))?; + } + let leader = c.elect(3, 10, &[1, 2, 3])?; + c.command(leader, Request::Propose(10, 103, second.clone()))?; + c.pump(&[1, 2, 3])?; + assert_eq!( + c.outcome(103), + Some(second_ack), + "snapshot lost the original retry receipt" + ); + let third = input(0, 3, 34); + c.command(leader, Request::Propose(10, 104, third.clone()))?; + c.pump(&[1, 2, 3])?; + assert!(matches!( + c.outcome(104), + Some(Outcome::Committed(Applied::Accepted { .. })) + )); + for peer in c.peers.values_mut() { + verify_records(peer, 10, &[first.clone(), second.clone(), third.clone()])?; + } + Ok(()) +} + #[test] fn durable_quorum_retry_leader_loss_suffix_replacement_and_full_restart() -> Result<()> { let mut c = Cluster::new()?; diff --git a/docs/design/raft-log-checkpoints.md b/docs/design/raft-log-checkpoints.md index 1b24d0c..e518161 100644 --- a/docs/design/raft-log-checkpoints.md +++ b/docs/design/raft-log-checkpoints.md @@ -46,10 +46,10 @@ complete application state. After receiving Ready, it must wait for this sync before releasing responses or applying suffix entries. On an uncertain write, retain both sets of application dependencies and recover the physical log. -This increment supplies the storage boundary only. The current -`replication::Node` still retains its complete log and invokes neither snapshot -API. Application state/payload packs, snapshot transfer and driver integration, -reader/transfer pins, physical reclamation, admission during maintenance and +These APIs supply the storage boundary. The `replication::Node` integration +uses validated [application images](replicated-wal-images.md) before invoking +them and supports explicit transfer of those immutable files. Production +transport scheduling, collector pins, physical reclamation, disk admission and Lance publication remain subsequent integration work. No production ingestion path or retention default changes. diff --git a/docs/design/replicated-wal-append.md b/docs/design/replicated-wal-append.md index c94c51c..add1798 100644 --- a/docs/design/replicated-wal-append.md +++ b/docs/design/replicated-wal-append.md @@ -62,8 +62,9 @@ answering directly from local state. A later quorum-read optimization must preserve that boundary. Pending local requests bind both log index and term; replacement of an uncommitted suffix cannot complete a different request. -Reopen reconstructs accepted metadata and latest producer state by replaying -only the durable committed prefix. The uncommitted suffix remains available to +Reopen validates an installed application image, then reconstructs accepted +metadata and latest producer state by replaying its durable committed suffix. +Without an image, replay starts at index one. The uncommitted suffix remains available to Raft for reconciliation. Accepted pages and payload reads expose a **local durable applied view**, not a quorum read, authenticated receipt or proof of Lance visibility. Consumers must preserve the group/layout scope. Their use by a fenced @@ -82,12 +83,17 @@ rejected, so a lagging replica cannot make the consumer move backward. The immutable policy bounds record size, accepted metadata, tracked producers, pending local requests and Raft's uncommitted-data budget. Capacity rejection is itself deterministic after commit; an exact latest retry can still succeed when -metadata capacity is full. Historical accepted metadata and all raw log entries -are retained in this increment, including rejected or duplicate proposals. +metadata capacity is full. Historical accepted metadata remains retained. +Explicit application checkpoints can compact the logical log prefix while +keeping accepted payloads readable in an immutable image. Image files are retained. These are not bounded physical disk usage or an unlimited-duration retention policy. Transport must also bound frame decoding and queued messages. -Producer slot reuse, authenticated ownership/revoke, snapshot transfer, replica +See [application images](replicated-wal-images.md) for the explicit build, +transfer and installation contract. The caller must implement the bounded +transport/worker scheduling before enabling periodic checkpoints. + +Producer slot reuse, authenticated ownership/revoke, dynamic replica replacement, disk credits, log reclamation and fenced Lance publication remain subsequent increments. This feature is not a deployment-ready WAL service and does not establish the complete ingestion goal or a performance result. diff --git a/docs/design/replicated-wal-images.md b/docs/design/replicated-wal-images.md new file mode 100644 index 0000000..7e48850 --- /dev/null +++ b/docs/design/replicated-wal-images.md @@ -0,0 +1,102 @@ +# Application images for replicated WAL recovery + +Refs https://github.com/lance-format/lance-context/issues/378. + +The optional replication component can explicitly checkpoint a committed applied +cut while retaining its accepted payloads and exact producer retry state. A +lagging member of the same static group can receive that image and then replay +the suffix. No existing production writer or periodic maintenance setting changes. + +## Build and install + +`Node::prepare_checkpoint(group, id)` captures bounded metadata, the committed +cut and its term, plus handles to the original log and previous immutable image. +It does not copy payloads. Run the resulting `CheckpointTask` on a blocking +worker, while continuing to tick and poll the node. A single permit shared by +all groups on the physical node limits concurrent image builders/receivers and +prevents a second checkpoint from compacting the task's captured inputs. The +permit remains held until the prepared candidate is installed or dropped. + +The worker streams metadata and one bounded record at a time into immutable +files created with `create_new`. It validates each body against its captured +metadata, syncs both files and their directory, and returns a non-forgeable +`PreparedCheckpoint`. File names use only a group number and caller-supplied +fresh UUID; a failed attempt cannot overwrite an installed image or orphan. + +`Node::install_checkpoint` validates ownership and durably installs the local +descriptor through the shared Raft log checkpoint boundary. The cut may precede +the node's current applied index: intervening committed and uncommitted entries +remain in the log. `read_accepted` uses packed bytes for accepted prefix entries +and ordinary log reads for suffix entries, validating their body digest through +the same boundary. A failed installation fences the node. Retain all old/new +dependencies until reopen resolves an uncertain engine write. + +## Transfer before Raft acceptance + +When `poll` emits `MsgSnapshot`, immediately capture its `Image` using +`snapshot_image` before another local checkpoint replaces the current image. +Keep this handle for the transfer. A later obsolete message can be discarded and +reported with `report_snapshot(..., false)` so Raft retries. Ordinary `step` +continues to reject snapshot messages. + +At the recipient, `receive_snapshot` checks the packet's static scope, policy, +route and bounded descriptor before creating a receiver with a fresh local UUID. +Send the two files through sequential chunks of at most one MiB. File I/O and +network waits stay outside the Raft owner; the owner can keep processing ordinary +traffic during the transfer. Supply transport authentication, deadlines and +node-wide byte/disk admission separately. Scope equality is not authentication. + +`SnapshotReceiver::finish` requires exact lengths and hashes, validates metadata, +record hashes and producer tails, and syncs files and their directory. Only then +does `step_snapshot` pass the locally named descriptor to RawNode. Raft can ignore +an image that became stale during download. If it accepts the image, `poll` +atomically persists the descriptor, replacement suffix and HardState before +releasing messages, swaps application state, and applies later committed entries. +Old unresolved local requests return `Unknown` and require exact retries. + +A lost volume must not reuse an old Raft voter identity with empty state. This +static-membership transfer API recovers lagging retained replicas; allocating a +new voter identity and committing its membership transition remain separate work. + +## Recovery and format bounds + +The versioned descriptor binds cluster, layout fingerprint, application policy, +group, static voters, index and term, file identity, counts, lengths and hashes. +The entire Raft Snapshot fits in four KiB. Its metadata file repeats the binding +and frames each record independently. Counts are validated before iteration; +frames have a fixed small maximum and never deserialize a payload-sized vector. +Recovery streams payload verification through a 64 KiB scratch buffer. + +Format v1 retains all accepted metadata and stores exact producer tails. It +checks that each producer's accepted sequences start at one and are contiguous, +and that its saved tail is precisely the latest accepted index and metadata. +Blank, duplicate and rejected log entries may leave gaps between accepted indexes. +Missing files, wrong policy, duplicate indexes, unexpected lengths, trailing +bytes or mismatched checksums fail recovery; orphan images never become fallback +state. Reopen loads only the descriptor selected by durable Raft storage. + +This representation intentionally precedes publication-aware retirement. Retiring +accepted records will require a new version that validates independently retained +producer tails and publication/source state. Full images currently copy retained +payloads again; they are not incremental checkpoints. Reusing immutable segments, +budgeting rewrite headroom, tracking collector pins and safe physical collection +remain necessary for sustained throughput and bounded disk usage. No Arrow +decoding occurs in the image path, but pack construction and file reads still +copy bytes; this is not an end-to-end zero-copy claim. + +## Verification + +Run the replication library and real-process suite with: + +```bash +cargo test --locked -p lance-context-wal --features replication --lib --test replication_process +``` + +The suite exercises append progress after a captured cut, exact prefix/suffix +payload recovery, producer retries after packing, shared builder admission, +checksum/scope/length failures, a sibling group and sync-failure fencing. Actual +Rust child processes transfer bounded chunks over TCP; the parent kills a process +after file sync but before installation, interrupts a follower's transfer with +SIGKILL, retries using fresh file identities, and kills/restarts all peers after +successful installation. These tests must pass before this increment is shipped; +they do not measure power-loss durability or production throughput.