diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index eeb528c0..fc8c2447 100644 --- a/.github/workflows/rust-test.yml +++ b/.github/workflows/rust-test.yml @@ -75,7 +75,7 @@ jobs: - name: WAL durable Raft storage tests run: | cargo clippy -p lance-context-wal --features raft-log --all-targets -- -D warnings - cargo test -p lance-context-wal --features raft-log --lib --test raft_log --test raft_log_cache_identity + cargo test -p lance-context-wal --features raft-log --lib --test raft_log --test raft_log_cache_identity --test raft_log_checkpoint - name: Client journal durability and process recovery run: | cargo clippy -p lance-context-wal --features client-journal --all-targets -- -D warnings diff --git a/crates/lance-context-wal/src/raft_log.rs b/crates/lance-context-wal/src/raft_log.rs index e8145871..46b5d335 100644 --- a/crates/lance-context-wal/src/raft_log.rs +++ b/crates/lance-context-wal/src/raft_log.rs @@ -7,10 +7,11 @@ //! raft-rs's message, persistence, commit and application ordering requirements. //! In particular, do not acknowledge a producer merely because `persist` returned. //! -//! This first storage increment retains the complete log. Snapshot installation, -//! compaction, membership changes, replica replacement, admission and transport -//! belong to subsequent integration. Its versioned directories cannot open the -//! earlier experimental formats. Only the owning process may modify the files. +//! An application may checkpoint a validated, durable committed cut. The small +//! opaque descriptor and logical compaction cross one physical sync barrier. +//! Application snapshot files, remote installation, physical file reclamation, +//! membership changes and replica replacement require higher-level integration. +//! Only the owning process may modify the files. use std::collections::{BTreeMap, BTreeSet}; use std::path::Path; @@ -25,6 +26,7 @@ use raft_proto::eraftpb::{ConfState, Entry, EntryType, HardState, Snapshot}; use crate::placement::Layout; +mod checkpoint; #[cfg(test)] mod fault_tests; @@ -37,6 +39,8 @@ const IDENTITY: &[u8] = b"identity"; const HARD: &[u8] = b"hard"; const CONF: &[u8] = b"conf"; const TERMS: &[u8] = b"terms"; +const SNAPSHOT: &[u8] = b"snapshot"; +const MAX_SNAPSHOT_BYTES: usize = 4096; // raft-engine preallocates entry descriptors for the requested index interval, // before applying max_size. Never pass an unbounded interval to that API. const READ_CHUNK_ENTRIES: u64 = 256; @@ -147,16 +151,23 @@ struct GroupState { hard: HardState, conf: ConfState, terms: Vec, + snapshot: Snapshot, } impl GroupState { fn last(&self) -> u64 { - self.terms.last().map_or(0, |range| range.last) + self.terms + .last() + .map_or(self.boundary(), |range| range.last) + } + + fn boundary(&self) -> u64 { + self.snapshot.get_metadata().index } fn term(&self, index: u64) -> Option { - if index == 0 { - return Some(0); + if index == self.boundary() { + return Some(self.snapshot.get_metadata().term); } self.terms .get(self.terms.partition_point(|range| range.last < index)) @@ -165,19 +176,31 @@ impl GroupState { } fn validate(&self) -> Result<()> { - let (mut previous_last, mut previous_term) = (0, 0); + let metadata = self.snapshot.get_metadata(); + if self.boundary() == 0 { + ensure!( + self.snapshot == Snapshot::default(), + "invalid empty checkpoint" + ); + } else { + checkpoint::validate(&self.snapshot, &self.conf)?; + } + let (mut previous_last, mut previous_term) = (metadata.index, metadata.term); for range in &self.terms { ensure!( range.first == previous_last + 1 && range.last >= range.first && range.last < u64::MAX - && range.term > previous_term, + && range.term > 0 + && (range.term > previous_term + || (previous_last == self.boundary() && range.term == previous_term)), "invalid durable term index" ); (previous_last, previous_term) = (range.last, range.term); } ensure!( - self.hard.commit <= self.last() + self.hard.commit >= self.boundary() + && self.hard.commit <= self.last() && self.hard.term >= previous_term && (self.hard.term > 0 || self.hard.vote == 0) && (self.hard.vote == 0 || self.conf.voters.contains(&self.hard.vote)), @@ -325,12 +348,31 @@ impl PhysicalLog { .get(group, TERMS) .context("missing term index")?, )?, + snapshot: match node.engine.get(group, SNAPSHOT) { + Some(bytes) => { + ensure!( + bytes.len() <= MAX_SNAPSHOT_BYTES, + "oversized durable snapshot" + ); + let application = node + .engine + .get(group, b"application") + .context("unbound checkpoint")?; + ensure!( + !application.is_empty() && application.len() <= 4096, + "invalid checkpoint application binding" + ); + Snapshot::parse_from_bytes(&bytes)? + } + None => Snapshot::default(), + }, }; ensure!(state.conf == expected_conf, "static membership changed"); state.validate()?; ensure!( - node.engine.last_index(group).unwrap_or(0) == state.last() - && node.engine.first_index(group) == (state.last() > 0).then_some(1), + node.engine.last_index(group) == state.terms.last().map(|range| range.last) + && node.engine.first_index(group) + == state.terms.first().map(|range| range.first), "entry bounds differ from durable term index" ); groups.insert( @@ -477,6 +519,7 @@ impl GroupLog { hard: next_hard.clone(), conf: state.conf.clone(), terms, + snapshot: state.snapshot.clone(), }; next.validate()?; let mut batch = LogBatch::default(); @@ -516,7 +559,7 @@ impl Storage for GroupLog { _: GetEntriesContext, ) -> raft::Result> { let state = self.read().map_err(storage_error)?; - if low < 1 { + if low <= state.boundary() { return Err(StorageError::Compacted.into()); } if low > high || high > state.last() + 1 { @@ -559,23 +602,26 @@ impl Storage for GroupLog { } fn term(&self, index: u64) -> raft::Result { - self.read() - .map_err(storage_error)? - .term(index) - .ok_or(StorageError::Unavailable.into()) + let state = self.read().map_err(storage_error)?; + if index < state.boundary() { + return Err(StorageError::Compacted.into()); + } + state.term(index).ok_or(StorageError::Unavailable.into()) } fn first_index(&self) -> raft::Result { - self.node.check().map_err(storage_error)?; - Ok(1) + Ok(self.read().map_err(storage_error)?.boundary() + 1) } fn last_index(&self) -> raft::Result { Ok(self.read().map_err(storage_error)?.last()) } - fn snapshot(&self, _: u64, _: u64) -> raft::Result { - self.node.check().map_err(storage_error)?; - Err(StorageError::SnapshotTemporarilyUnavailable.into()) + fn snapshot(&self, request_index: u64, _: u64) -> raft::Result { + let state = self.read().map_err(storage_error)?; + if state.boundary() == 0 || state.boundary() < request_index { + return Err(StorageError::SnapshotTemporarilyUnavailable.into()); + } + Ok(state.snapshot.clone()) } } diff --git a/crates/lance-context-wal/src/raft_log/checkpoint.rs b/crates/lance-context-wal/src/raft_log/checkpoint.rs new file mode 100644 index 00000000..e5b700dd --- /dev/null +++ b/crates/lance-context-wal/src/raft_log/checkpoint.rs @@ -0,0 +1,86 @@ +use super::*; + +pub(super) fn validate(snapshot: &Snapshot, conf: &ConfState) -> Result<()> { + let metadata = snapshot.get_metadata(); + ensure!( + metadata.index > 0 + && metadata.index < u64::MAX + && metadata.term > 0 + && metadata.get_conf_state() == conf + && !snapshot.data.is_empty() + && snapshot.compute_size() as usize <= MAX_SNAPSHOT_BYTES, + "invalid checkpoint cut, static membership or bounded descriptor" + ); + Ok(()) +} + +impl GroupLog { + /// Durably replace a committed prefix with an application checkpoint. + /// + /// The application MUST first validate the descriptor's scope and complete + /// state at this exact applied index, fsync every referenced immutable file + /// and its directory, and retain dependencies of both the old and new + /// descriptors on an uncertain write. This layer treats Snapshot.data as + /// opaque; it cannot establish application recovery or Lance publication. + /// Its entire encoded Snapshot is limited to 4 KiB, for a small state or a + /// descriptor rather than bulk payload transfer. + /// + /// This synchronously installs the descriptor, truncation boundary and term + /// index in the same engine batch as logical compaction. It preserves the + /// 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. + pub fn checkpoint(&self, snapshot: &Snapshot) -> Result<()> { + let mut state = self + .state + .write() + .map_err(|_| anyhow::anyhow!("group state lock poisoned"))?; + self.node.check()?; + validate(snapshot, &state.conf)?; + ensure!( + self.node.engine.get(self.group, b"application").is_some(), + "bind the application before checkpointing" + ); + let metadata = snapshot.get_metadata(); + ensure!( + metadata.index >= state.boundary() + && metadata.index <= state.hard.commit + && state.term(metadata.index) == Some(metadata.term), + "checkpoint is not a retained durable committed cut" + ); + if metadata.index == state.boundary() { + ensure!( + *snapshot == state.snapshot, + "checkpoint changed at the same cut" + ); + return self.node.check(); + } + let mut terms = state.terms.clone(); + terms.retain(|range| range.last > metadata.index); + if let Some(first) = terms.first_mut() { + first.first = metadata.index + 1; + } + let next = GroupState { + hard: state.hard.clone(), + conf: state.conf.clone(), + terms, + snapshot: snapshot.clone(), + }; + next.validate()?; + let mut batch = LogBatch::default(); + batch.put_message(self.group, SNAPSHOT.to_vec(), snapshot)?; + batch.put(self.group, TERMS.to_vec(), encode_terms(&next.terms)?)?; + // Engine::compact_to writes without sync and suppresses errors. Keep + // the command and its recovery metadata under our shared failure fence. + batch.add_command( + self.group, + raft_engine::Command::Compact { + index: metadata.index + 1, + }, + ); + self.node.write(&mut batch)?; + *state = next; + Ok(()) + } +} diff --git a/crates/lance-context-wal/src/raft_log/fault_tests.rs b/crates/lance-context-wal/src/raft_log/fault_tests.rs index 3032de0a..81af6ccc 100644 --- a/crates/lance-context-wal/src/raft_log/fault_tests.rs +++ b/crates/lance-context-wal/src/raft_log/fault_tests.rs @@ -190,6 +190,77 @@ fn rotation_sync_panic_cannot_ack_another_group() -> Result<()> { sync_failure_cannot_ack_another_group(1) } +#[test] +fn checkpoint_sync_failure_fences_all_handles_and_recovers_one_complete_cut() -> Result<()> { + let layout = Layout { + cluster: uuid::Uuid::from_u128(3), + nodes: (1..=3) + .map(|id| (id, format!("wal-{id}.example.invalid:7800"))) + .collect(), + groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![1, 2, 3])]), + buckets: vec![10, 20], + }; + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + let log = PhysicalLog::open(&path, &layout, 1)?; + let group = log.group(10)?; + group.bind_application(b"checkpoint-fault/v1")?; + let entries: Vec<_> = (1..=3) + .map(|index| Entry { + index, + term: 1, + data: vec![index as u8; 4096].into(), + ..Default::default() + }) + .collect(); + group.persist( + &entries, + Some(&HardState { + term: 1, + vote: 1, + commit: 3, + ..Default::default() + }), + )?; + let mut snapshot = Snapshot::default(); + snapshot.mut_metadata().index = 2; + snapshot.mut_metadata().term = 1; + snapshot + .mut_metadata() + .set_conf_state(group.initial_state()?.conf_state); + snapshot.data = vec![2; 32].into(); + group.node.fault.fail_sync.store(true, Ordering::Release); + assert!(group.checkpoint(&snapshot).is_err()); + assert!(group.snapshot(0, 2).is_err()); + assert!(group.first_index().is_err()); + assert!(log.group(20)?.persist(&[], None).is_err()); + group.node.fault.fail_sync.store(false, Ordering::Release); + drop((group, log)); + let log = PhysicalLog::open(&path, &layout, 1)?; + let group = log.group(10)?; + assert_eq!(group.initial_state()?.hard_state.commit, 3); + match group.first_index()? { + 1 => assert_eq!( + group.entries(1, 4, None, GetEntriesContext::empty(false))?, + entries + ), + 3 => { + assert_eq!(group.snapshot(2, 2)?, snapshot); + assert_eq!(group.term(2)?, 1); + assert_eq!( + group.entries(3, 4, None, GetEntriesContext::empty(false))?, + entries[2..] + ); + } + index => panic!("partial checkpoint boundary {index}"), + } + assert_eq!( + log.group(20)?.initial_state()?.hard_state, + HardState::default() + ); + Ok(()) +} + fn sync_failure_cannot_ack_another_group(file_size: u64) -> Result<()> { let layout = Layout { cluster: uuid::Uuid::from_u128(1), diff --git a/crates/lance-context-wal/tests/raft_log_checkpoint.rs b/crates/lance-context-wal/tests/raft_log_checkpoint.rs new file mode 100644 index 00000000..503a70fb --- /dev/null +++ b/crates/lance-context-wal/tests/raft_log_checkpoint.rs @@ -0,0 +1,321 @@ +#![cfg(feature = "raft-log")] + +use std::collections::BTreeMap; + +use anyhow::Result; +use lance_context_wal::{ + placement::Layout, + raft_log::{GroupLog, PhysicalLog}, +}; +use protobuf::Message; +use raft::{GetEntriesContext, Storage, StorageError}; +use raft_proto::eraftpb::{Entry, HardState, Snapshot}; +use uuid::Uuid; + +fn layout() -> Layout { + Layout { + cluster: Uuid::from_u128(22), + nodes: (1..=3) + .map(|n| (n, format!("wal-{n}.example.invalid:7800"))) + .collect(), + groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![1, 2, 3])]), + buckets: vec![10, 20], + } +} + +fn entry(index: u64, term: u64) -> Entry { + Entry { + index, + term, + data: vec![index as u8; 8192].into(), + ..Default::default() + } +} + +fn hard(term: u64, commit: u64) -> HardState { + HardState { + term, + commit, + vote: 1, + ..Default::default() + } +} + +fn snapshot(log: &GroupLog, index: u64, term: u64) -> Result { + let mut snapshot = Snapshot::default(); + snapshot.mut_metadata().index = index; + snapshot.mut_metadata().term = term; + snapshot + .mut_metadata() + .set_conf_state(log.initial_state()?.conf_state); + // Self-contained toy application state; production descriptors additionally + // need validated immutable files and their own scope/recovery contracts. + snapshot.data = vec![index as u8; 32].into(); + Ok(snapshot) +} + +fn read(log: &GroupLog, low: u64, high: u64) -> raft::Result> { + log.entries(low, high, None, GetEntriesContext::empty(false)) +} + +#[test] +fn checkpoint_preserves_suffix_vote_and_other_group_across_reopen() -> Result<()> { + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + let original: Vec<_> = (1..=6) + .map(|i| entry(i, if i < 3 { 1 } else { 2 })) + .collect(); + let saved; + { + let log = PhysicalLog::open(&path, &layout(), 1)?; + let group = log.group(10)?; + group.bind_application(b"checkpoint-test/v1")?; + group.persist(&original, Some(&hard(2, 5)))?; + log.group(20)?.persist(&[entry(1, 7)], Some(&hard(7, 1)))?; + saved = snapshot(&group, 3, 2)?; + group.checkpoint(&saved)?; + group.checkpoint(&saved)?; + assert_eq!(group.first_index()?, 4); + assert_eq!(group.last_index()?, 6); + assert_eq!(group.term(3)?, 2); + assert!(matches!( + group.term(2), + Err(raft::Error::Store(StorageError::Compacted)) + )); + assert!(matches!( + read(&group, 3, 4), + Err(raft::Error::Store(StorageError::Compacted)) + )); + assert_eq!(read(&group, 4, 7)?, original[3..]); + assert_eq!(group.snapshot(3, 2)?, saved); + assert!(matches!( + group.snapshot(4, 2), + Err(raft::Error::Store( + StorageError::SnapshotTemporarilyUnavailable + )) + )); + assert_eq!(group.initial_state()?.hard_state, hard(2, 5)); + } + { + let log = PhysicalLog::open(&path, &layout(), 1)?; + let group = log.group(10)?; + group.bind_application(b"checkpoint-test/v1")?; + assert_eq!(group.snapshot(0, 3)?, saved); + assert_eq!(read(&group, 4, 7)?, original[3..]); + assert_eq!(read(&log.group(20)?, 1, 2)?, vec![entry(1, 7)]); + assert_eq!(log.group(20)?.first_index()?, 1); + // A retained committed entry remains immutable; an uncommitted suffix + // still permits replacement in a later term. + assert!(group.persist(&[entry(5, 3)], Some(&hard(3, 5))).is_err()); + group.persist(&[entry(6, 3), entry(7, 3)], Some(&hard(3, 7)))?; + assert_eq!( + read(&group, 4, 8)?, + vec![entry(4, 2), entry(5, 2), entry(6, 3), entry(7, 3)] + ); + let next = snapshot(&group, 7, 3)?; + group.checkpoint(&next)?; + assert_eq!((group.first_index()?, group.last_index()?), (8, 7)); + assert!(read(&group, 8, 8)?.is_empty()); + } + let log = PhysicalLog::open(&path, &layout(), 1)?; + let group = log.group(10)?; + assert_eq!( + (group.first_index()?, group.last_index()?, group.term(7)?), + (8, 7, 3) + ); + assert_eq!(group.initial_state()?.hard_state, hard(3, 7)); + group.persist(&[entry(8, 3)], Some(&hard(3, 8)))?; + assert_eq!(read(&group, 8, 9)?, vec![entry(8, 3)]); + let config = raft::Config { + id: 1, + applied: 7, + ..Default::default() + }; + let mut raw = raft::RawNode::new( + &config, + group, + &slog::Logger::root(slog::Discard, slog::o!()), + )?; + let mut ready = raw.ready(); + assert_eq!(ready.take_committed_entries(), vec![entry(8, 3)]); + Ok(()) +} + +#[test] +fn invalid_or_changed_checkpoint_never_changes_usable_log() -> Result<()> { + let directory = tempfile::tempdir()?; + let log = PhysicalLog::open(&directory.path().join("log"), &layout(), 1)?; + let group = log.group(10)?; + group.persist(&[entry(1, 1), entry(2, 2), entry(3, 2)], Some(&hard(2, 2)))?; + let good = snapshot(&group, 2, 2)?; + assert!( + group.checkpoint(&good).is_err(), + "application must be bound before its first entry" + ); + // Use a separately bound group to exercise every invalid candidate without + // mutating or adopting the unbound group's existing log. + let group = log.group(20)?; + group.bind_application(b"checkpoint-test/v1")?; + group.persist(&[entry(1, 1), entry(2, 2), entry(3, 2)], Some(&hard(2, 2)))?; + let good = snapshot(&group, 2, 2)?; + let mut cases = vec![ + snapshot(&group, 0, 0)?, + snapshot(&group, 3, 2)?, + snapshot(&group, 2, 1)?, + snapshot(&group, u64::MAX, 2)?, + ]; + let mut wrong = good.clone(); + wrong.mut_metadata().mut_conf_state().voters = vec![1, 2, 4]; + cases.push(wrong); + let mut wrong = good.clone(); + wrong.data = vec![0; 4096].into(); + cases.push(wrong); + let mut wrong = good.clone(); + wrong.data.clear(); + cases.push(wrong); + for invalid in cases { + assert!(group.checkpoint(&invalid).is_err()); + assert_eq!(group.first_index()?, 1); + assert_eq!( + read(&group, 1, 4)?, + vec![entry(1, 1), entry(2, 2), entry(3, 2)] + ); + } + group.checkpoint(&good)?; + let mut changed = good.clone(); + changed.data = vec![99].into(); + assert!(group.checkpoint(&changed).is_err()); + assert!(group.checkpoint(&snapshot(&group, 1, 1)?).is_err()); + assert_eq!(group.snapshot(0, 2)?, good); + assert_eq!(read(&group, 3, 4)?, vec![entry(3, 2)]); + Ok(()) +} + +#[test] +fn missing_corrupt_or_inconsistent_checkpoint_fails_reopen() -> Result<()> { + for corruption in 0..5 { + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + let saved; + { + let log = PhysicalLog::open(&path, &layout(), 1)?; + let group = log.group(10)?; + group.bind_application(b"checkpoint-test/v1")?; + group.persist(&[entry(1, 1), entry(2, 2), entry(3, 2)], Some(&hard(2, 3)))?; + saved = snapshot(&group, 2, 2)?; + group.checkpoint(&saved)?; + } + { + let engine = raft_engine::Engine::open(raft_engine::Config { + dir: path.to_str().unwrap().into(), + ..Default::default() + })?; + let mut batch = raft_engine::LogBatch::default(); + match corruption { + 0 => batch.delete(10, b"snapshot".to_vec()), + 1 => batch.put(10, b"snapshot".to_vec(), vec![255])?, + 2 => { + let mut invalid = saved; + invalid.mut_metadata().index = 3; + batch.put(10, b"snapshot".to_vec(), invalid.write_to_bytes()?)?; + } + 3 => batch.put(10, b"snapshot".to_vec(), vec![0; 4097])?, + 4 => batch.delete(10, b"application".to_vec()), + _ => unreachable!(), + } + engine.write(&mut batch, true)?; + } + assert!( + PhysicalLog::open(&path, &layout(), 1).is_err(), + "accepted corruption {corruption}" + ); + } + Ok(()) +} + +#[cfg(unix)] +mod process { + use super::*; + use std::{ + process::{Child, Command, Stdio}, + time::{Duration, Instant}, + }; + + struct Worker(Child); + impl Drop for Worker { + fn drop(&mut self) { + if !matches!(self.0.try_wait(), Ok(Some(_))) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } + } + } + + #[test] + #[ignore = "subprocess fixture invoked by checkpoint_survives_sigkill"] + fn checkpoint_worker() -> Result<()> { + let directory = std::path::PathBuf::from(std::env::var("CONTEXT_CHECKPOINT_WORKER_ROOT")?); + let after = std::env::var("CONTEXT_CHECKPOINT_WORKER_PHASE")? == "after"; + let log = PhysicalLog::open(&directory.join("log"), &layout(), 1)?; + let group = log.group(10)?; + group.bind_application(b"checkpoint-test/v1")?; + group.persist(&[entry(1, 1), entry(2, 2), entry(3, 2)], Some(&hard(2, 3)))?; + log.group(20)?.persist(&[entry(1, 5)], Some(&hard(5, 1)))?; + if after { + group.checkpoint(&snapshot(&group, 2, 2)?)?; + } + std::fs::write(directory.join("ready"), b"ready")?; + std::thread::sleep(Duration::from_secs(30)); + anyhow::bail!("parent did not stop checkpoint worker") + } + + #[test] + fn checkpoint_survives_sigkill() -> Result<()> { + use std::os::unix::process::ExitStatusExt; + for phase in ["before", "after"] { + let directory = tempfile::tempdir()?; + let stderr = directory.path().join("stderr"); + let mut child = Worker( + Command::new(std::env::current_exe()?) + .args([ + "--ignored", + "--exact", + "process::checkpoint_worker", + "--nocapture", + ]) + .env("CONTEXT_CHECKPOINT_WORKER_ROOT", directory.path()) + .env("CONTEXT_CHECKPOINT_WORKER_PHASE", phase) + .stdout(Stdio::null()) + .stderr(std::fs::File::create(&stderr)?) + .spawn()?, + ); + let deadline = Instant::now() + Duration::from_secs(20); + while !directory.path().join("ready").exists() { + anyhow::ensure!( + child.0.try_wait()?.is_none() && Instant::now() < deadline, + "checkpoint worker did not reach {phase}: {}", + std::fs::read_to_string(&stderr)? + ); + std::thread::sleep(Duration::from_millis(10)); + } + child.0.kill()?; + assert_eq!(child.0.wait()?.signal(), Some(9)); + let log = PhysicalLog::open(&directory.path().join("log"), &layout(), 1)?; + let group = log.group(10)?; + assert_eq!(group.initial_state()?.hard_state, hard(2, 3)); + assert_eq!(read(&log.group(20)?, 1, 2)?, vec![entry(1, 5)]); + if phase == "after" { + assert_eq!(group.first_index()?, 3); + assert_eq!(group.snapshot(2, 2)?, snapshot(&group, 2, 2)?); + assert_eq!(read(&group, 3, 4)?, vec![entry(3, 2)]); + } else { + assert_eq!(group.first_index()?, 1); + assert_eq!( + read(&group, 1, 4)?, + vec![entry(1, 1), entry(2, 2), entry(3, 2)] + ); + } + } + Ok(()) + } +} diff --git a/docs/design/raft-log-checkpoints.md b/docs/design/raft-log-checkpoints.md new file mode 100644 index 00000000..563da123 --- /dev/null +++ b/docs/design/raft-log-checkpoints.md @@ -0,0 +1,46 @@ +# Durable local Raft checkpoint boundary + +Refs https://github.com/lance-format/lance-context/issues/378. + +The optional `raft-log` storage supports a local application checkpoint at a +durable committed cut. `GroupLog::checkpoint` persists its bounded Raft Snapshot, +the retained term ranges and logical prefix compaction in one engine batch, +then waits for the existing shared physical sync barrier. A checkpoint does not +change HardState or remove any suffix entries, including uncommitted entries. +Only an identical descriptor can be retried at the same cut. + +The caller must first validate the complete application state at that exact +applied cut. If Snapshot.data references files, it must validate their scope, +contents and dependencies, fsync those immutable files and their directory, and +retain both previous and new dependencies after an uncertain write. The storage +layer deliberately treats Snapshot.data as opaque. The entire encoded Snapshot +is limited to 4 KiB; bulk payloads belong in separately managed files. This API +does not verify Lance publication or authorize deletion of those files. + +Recovery checks the snapshot membership and boundary against HardState, retained +term ranges and the physical engine's entry bounds. The compacted boundary's term +remains readable; earlier terms and entries return `Compacted`. An empty suffix +still has the snapshot's index as its logical last index. A newer requested +snapshot returns `SnapshotTemporarilyUnavailable`. Logs without a checkpoint +continue to start at index 1. An application binding must precede checkpointing. +Missing, corrupt or inconsistent checkpoint metadata fails recovery. + +Compaction uses an explicit `Command::Compact` in the same synced batch as its +metadata. The pinned engine's convenience `compact_to` method is insufficient: +it does not sync and does not return failures. Shared engine failures invalidate +all group handles before any subsequent successful operation. + +This increment supplies the storage boundary only. The current +`replication::Node` still retains its complete log and does not invoke this API. +Application state/payload packs, remote snapshot installation and transfer, +reader/transfer pins, physical reclamation, admission during maintenance and +Lance publication remain subsequent integration work. No production ingestion +path or retention default changes. + +Tests cover a committed cut inside a term, preserved votes and suffixes, an +empty suffix followed by append, a sibling group, repeated checkpoints, RawNode +replay from the retained suffix, invalid descriptors and corrupt recovery +metadata. Real child processes are killed before and after installation; an +injected filesystem sync failure checks the shared failure fence and complete +recovery boundary. These are storage correctness tests, not power-loss testing, +snapshot network recovery or a throughput benchmark.