diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index fc8c244..68cba13 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 --test raft_log_checkpoint + cargo test -p lance-context-wal --features raft-log --lib --test raft_log --test raft_log_cache_identity --test raft_log_checkpoint --test raft_log_snapshot_install - 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 46b5d33..faf8e22 100644 --- a/crates/lance-context-wal/src/raft_log.rs +++ b/crates/lance-context-wal/src/raft_log.rs @@ -9,8 +9,10 @@ //! //! 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. +//! A received Ready snapshot can atomically replace the log and persist its +//! suffix and HardState. Application snapshot files/transfer, driver integration, +//! physical reclamation, membership changes and replica replacement remain +//! higher-level responsibilities. //! Only the owning process may modify the files. use std::collections::{BTreeMap, BTreeSet}; @@ -208,6 +210,64 @@ impl GroupState { ); Ok(()) } + fn appended(&self, entries: &[Entry], hard: Option<&HardState>) -> Result { + let next_hard = hard.unwrap_or(&self.hard); + ensure!( + next_hard.term >= self.hard.term + && next_hard.commit >= self.hard.commit + && (next_hard.term != self.hard.term + || self.hard.vote == 0 + || next_hard.vote == self.hard.vote), + "HardState term, vote or commit regression" + ); + let mut terms = self.terms.clone(); + if let Some(first) = entries.first() { + ensure!( + first.index > self.hard.commit && first.index <= self.last() + 1, + "entries overwrite committed data or leave a gap" + ); + let mut previous_index = first.index - 1; + let mut previous_term = self + .term(previous_index) + .context("missing preceding term")?; + for entry in entries { + ensure!( + entry.index == previous_index + 1 + && entry.index < u64::MAX + && entry.term > 0 + && entry.term >= previous_term + && entry.term <= next_hard.term + && entry.get_entry_type() == EntryType::EntryNormal, + "noncontiguous entries, invalid term or unsupported membership change" + ); + previous_index = entry.index; + previous_term = entry.term; + } + terms.retain(|range| range.first < first.index); + if let Some(last) = terms.last_mut() { + last.last = last.last.min(first.index - 1); + } + for entry in entries { + if let Some(last) = terms.last_mut().filter(|range| range.term == entry.term) { + last.last = entry.index; + } else { + terms.push(TermRange { + first: entry.index, + last: entry.index, + term: entry.term, + }); + } + } + } + let next = GroupState { + hard: next_hard.clone(), + conf: self.conf.clone(), + terms, + snapshot: self.snapshot.clone(), + }; + next.validate()?; + Ok(next) + } } fn encode_terms(terms: &[TermRange]) -> Result> { @@ -467,61 +527,7 @@ impl GroupLog { .write() .map_err(|_| anyhow::anyhow!("group state lock poisoned"))?; self.node.check()?; - let next_hard = hard.unwrap_or(&state.hard); - ensure!( - next_hard.term >= state.hard.term - && next_hard.commit >= state.hard.commit - && (next_hard.term != state.hard.term - || state.hard.vote == 0 - || next_hard.vote == state.hard.vote), - "HardState term, vote or commit regression" - ); - let mut terms = state.terms.clone(); - if let Some(first) = entries.first() { - ensure!( - first.index > state.hard.commit && first.index <= state.last() + 1, - "entries overwrite committed data or leave a gap" - ); - let mut previous_index = first.index - 1; - let mut previous_term = state - .term(previous_index) - .context("missing preceding term")?; - for entry in entries { - ensure!( - entry.index == previous_index + 1 - && entry.index < u64::MAX - && entry.term > 0 - && entry.term >= previous_term - && entry.term <= next_hard.term - && entry.get_entry_type() == EntryType::EntryNormal, - "noncontiguous entries, invalid term or unsupported membership change" - ); - previous_index = entry.index; - previous_term = entry.term; - } - terms.retain(|range| range.first < first.index); - if let Some(last) = terms.last_mut() { - last.last = last.last.min(first.index - 1); - } - for entry in entries { - if let Some(last) = terms.last_mut().filter(|range| range.term == entry.term) { - last.last = entry.index; - } else { - terms.push(TermRange { - first: entry.index, - last: entry.index, - term: entry.term, - }); - } - } - } - let next = GroupState { - hard: next_hard.clone(), - conf: state.conf.clone(), - terms, - snapshot: state.snapshot.clone(), - }; - next.validate()?; + let next = state.appended(entries, hard)?; let mut batch = LogBatch::default(); if !entries.is_empty() { batch.add_entries::(self.group, entries)?; diff --git a/crates/lance-context-wal/src/raft_log/checkpoint.rs b/crates/lance-context-wal/src/raft_log/checkpoint.rs index e5b700d..0c7b7a1 100644 --- a/crates/lance-context-wal/src/raft_log/checkpoint.rs +++ b/crates/lance-context-wal/src/raft_log/checkpoint.rs @@ -15,6 +15,77 @@ pub(super) fn validate(snapshot: &Snapshot, conf: &ConfState) -> Result<()> { } impl GroupLog { + /// Atomically persist a received Raft Ready snapshot, suffix and HardState. + /// + /// The owner must validate and fsync the complete application snapshot and + /// every referenced file/directory BEFORE passing it to Raft. This method + /// supplies only the storage transition after Raft accepts that snapshot; + /// it cannot authenticate a sender or validate opaque application state. + /// Do not release snapshot/append responses or apply the suffix before it + /// returns successfully. Membership must remain the bound static membership. + /// + /// A received snapshot advances beyond the previous durable commit and + /// replaces ALL prior entries, including a divergent uncommitted suffix. + /// Entries from the same Ready start immediately after its cut. HardState, + /// application binding, snapshot descriptor and suffix cross one shared + /// sync barrier. An uncertain write fences the physical engine; retain old + /// and new application dependencies until recovery resolves the outcome. + /// Repeated/stale installs are rejected; ordinary later Ready updates use + /// `persist`, and a local committed cut uses `checkpoint` instead. + pub fn install_snapshot( + &self, + snapshot: &Snapshot, + entries: &[Entry], + hard: &HardState, + ) -> Result<()> { + let mut state = self + .state + .write() + .map_err(|_| anyhow::anyhow!("group state lock poisoned"))?; + self.node.check()?; + validate(snapshot, &state.conf)?; + let application = self + .node + .engine + .get(self.group, b"application") + .context("bind the application before receiving a snapshot")?; + ensure!( + !application.is_empty() && application.len() <= 4096, + "invalid snapshot application binding" + ); + let metadata = snapshot.get_metadata(); + ensure!( + metadata.index > state.hard.commit + && metadata.term + >= state + .term(state.hard.commit) + .context("missing committed term")?, + "received snapshot does not advance the committed history" + ); + let base = GroupState { + hard: state.hard.clone(), + conf: state.conf.clone(), + terms: Vec::new(), + snapshot: snapshot.clone(), + }; + let next = base.appended(entries, Some(hard))?; + let mut batch = LogBatch::default(); + // Clean also removes this group's key/value metadata. Restore every + // owned key in this SAME batch; the node/layout identity is in group 0. + batch.add_command(self.group, raft_engine::Command::Clean); + batch.put(self.group, b"application".to_vec(), application)?; + batch.put_message(self.group, CONF.to_vec(), &next.conf)?; + batch.put_message(self.group, HARD.to_vec(), &next.hard)?; + batch.put_message(self.group, SNAPSHOT.to_vec(), &next.snapshot)?; + batch.put(self.group, TERMS.to_vec(), encode_terms(&next.terms)?)?; + if !entries.is_empty() { + batch.add_entries::(self.group, entries)?; + } + self.node.write(&mut batch)?; + *state = next; + Ok(()) + } + /// Durably replace a committed prefix with an application checkpoint. /// /// The application MUST first validate the descriptor's scope and complete 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 81af6cc..23d0c41 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,100 @@ fn rotation_sync_panic_cannot_ack_another_group() -> Result<()> { sync_failure_cannot_ack_another_group(1) } +#[test] +fn snapshot_install_sync_failure_recovers_all_old_or_all_new_state() -> Result<()> { + let layout = Layout { + cluster: uuid::Uuid::from_u128(24), + 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"snapshot-install-fault/v1")?; + let entry = |index, term| Entry { + index, + term, + data: vec![(index + term) as u8; 4096].into(), + ..Default::default() + }; + let old: Vec<_> = (1..=5).map(|i| entry(i, 2)).collect(); + let old_hard = HardState { + term: 2, + vote: 1, + commit: 2, + ..Default::default() + }; + group.persist(&old, Some(&old_hard))?; + let sibling = log.group(20)?; + let sibling_hard = HardState { + term: 7, + vote: 2, + commit: 1, + ..Default::default() + }; + sibling.persist(&[entry(1, 7)], Some(&sibling_hard))?; + let mut snapshot = Snapshot::default(); + snapshot.mut_metadata().index = 4; + snapshot.mut_metadata().term = 3; + snapshot + .mut_metadata() + .set_conf_state(group.initial_state()?.conf_state); + snapshot.data = b"snapshot-install-fault/descriptor".to_vec().into(); + let suffix = vec![entry(5, 3), entry(6, 3)]; + let new_hard = HardState { + term: 3, + vote: 2, + commit: 5, + ..Default::default() + }; + group.node.fault.fail_once.store(true, Ordering::Release); + assert!(group + .install_snapshot(&snapshot, &suffix, &new_hard) + .is_err()); + assert!(group.initial_state().is_err()); + assert!(group.snapshot(0, 2).is_err()); + assert!(sibling.persist(&[], None).is_err()); + assert!(sibling.initial_state().is_err()); + drop((group, sibling, log)); + let log = PhysicalLog::open(&path, &layout, 1)?; + let group = log.group(10)?; + group.bind_application(b"snapshot-install-fault/v1")?; + match group.first_index()? { + 1 => { + assert_eq!(group.initial_state()?.hard_state, old_hard); + assert_eq!(group.last_index()?, 5); + assert_eq!( + group.entries(1, 6, None, GetEntriesContext::empty(false))?, + old + ); + assert!(group.snapshot(0, 2).is_err()); + } + 5 => { + assert_eq!(group.initial_state()?.hard_state, new_hard); + assert_eq!(group.last_index()?, 6); + assert_eq!(group.term(4)?, 3); + assert_eq!(group.snapshot(4, 2)?, snapshot); + assert_eq!( + group.entries(5, 7, None, GetEntriesContext::empty(false))?, + suffix + ); + } + index => panic!("partial snapshot install boundary {index}"), + } + let sibling = log.group(20)?; + assert_eq!(sibling.initial_state()?.hard_state, sibling_hard); + assert_eq!( + sibling.entries(1, 2, None, GetEntriesContext::empty(false))?, + vec![entry(1, 7)] + ); + Ok(()) +} + #[test] fn checkpoint_sync_failure_fences_all_handles_and_recovers_one_complete_cut() -> Result<()> { let layout = Layout { diff --git a/crates/lance-context-wal/tests/raft_log_snapshot_install.rs b/crates/lance-context-wal/tests/raft_log_snapshot_install.rs new file mode 100644 index 0000000..df0471f --- /dev/null +++ b/crates/lance-context-wal/tests/raft_log_snapshot_install.rs @@ -0,0 +1,311 @@ +#![cfg(feature = "raft-log")] + +use std::{collections::BTreeMap, path::Path}; + +use anyhow::{Context, Result}; +use lance_context_wal::{ + placement::Layout, + raft_log::{GroupLog, PhysicalLog}, +}; +use raft::{GetEntriesContext, Storage, StorageError}; +use raft_proto::eraftpb::{Entry, HardState, Message, MessageType, Snapshot}; +use uuid::Uuid; + +fn layout() -> Layout { + Layout { + cluster: Uuid::from_u128(23), + 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 * 17 + term) as u8; 8192].into(), + ..Default::default() + } +} +fn hard(term: u64, commit: u64, vote: u64) -> HardState { + HardState { + term, + commit, + vote, + ..Default::default() + } +} +fn read(log: &GroupLog, low: u64, high: u64) -> raft::Result> { + log.entries(low, high, None, GetEntriesContext::empty(false)) +} +fn original() -> Vec { + (1..=7) + .map(|i| entry(i, if i == 1 { 1 } else { 2 })) + .collect() +} +fn seed(path: &Path) -> Result { + let log = PhysicalLog::open(path, &layout(), 2)?; + let group = log.group(10)?; + group.bind_application(b"snapshot-install-test/v1")?; + group.persist(&original(), Some(&hard(2, 1, 2)))?; + log.group(20)? + .persist(&[entry(1, 7)], Some(&hard(7, 1, 2)))?; + Ok(log) +} +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); + snapshot.data = b"storage-contract/application-descriptor".to_vec().into(); + Ok(snapshot) +} +fn verify_installed(log: &PhysicalLog, snapshot: &Snapshot) -> Result<()> { + let group = log.group(10)?; + group.bind_application(b"snapshot-install-test/v1")?; + assert!(group.bind_application(b"different-application").is_err()); + assert_eq!(group.snapshot(0, 1)?, *snapshot); + assert_eq!(group.initial_state()?.hard_state, hard(3, 5, 0)); + assert_eq!( + (group.first_index()?, group.last_index()?, group.term(4)?), + (5, 6, 3) + ); + assert_eq!(read(&group, 5, 7)?, vec![entry(5, 3), entry(6, 3)]); + assert!(matches!( + group.term(3), + Err(raft::Error::Store(StorageError::Compacted)) + )); + assert!(matches!( + group.term(7), + Err(raft::Error::Store(StorageError::Unavailable)) + )); + assert_eq!(read(&log.group(20)?, 1, 2)?, vec![entry(1, 7)]); + assert_eq!(log.group(20)?.initial_state()?.hard_state, hard(7, 1, 2)); + Ok(()) +} + +#[test] +fn received_ready_replaces_divergent_tail_and_recovers_committed_suffix() -> Result<()> { + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + let saved; + { + let log = seed(&path)?; + let group = log.group(10)?; + let mut raw = raft::RawNode::new( + &raft::Config { + id: 2, + applied: 1, + ..Default::default() + }, + group.clone(), + &slog::Logger::root(slog::Discard, slog::o!()), + )?; + saved = snapshot(&group, 4, 3)?; + let mut received = Message { + from: 1, + to: 2, + term: 3, + ..Default::default() + }; + received.set_msg_type(MessageType::MsgSnapshot); + received.set_snapshot(saved.clone()); + raw.step(received)?; + let mut suffix = Message { + from: 1, + to: 2, + term: 3, + index: 4, + log_term: 3, + commit: 5, + ..Default::default() + }; + suffix.set_msg_type(MessageType::MsgAppend); + suffix.set_entries(vec![entry(5, 3), entry(6, 3)].into()); + raw.step(suffix)?; + let mut ready = raw.ready(); + assert_eq!(ready.snapshot(), &saved); + assert_eq!(ready.entries(), &[entry(5, 3), entry(6, 3)]); + assert_eq!(read(&group, 1, 8)?, original()); + group.install_snapshot( + ready.snapshot(), + ready.entries(), + ready.hs().context("missing Ready HardState")?, + )?; + verify_installed(&log, &saved)?; + // Release/apply Ready only after the complete storage transition. + let mut applied = ready.take_committed_entries(); + let mut light = raw.advance(ready); + if light.commit_index().is_some() { + group.persist(&[], Some(&raw.raft.hard_state()))?; + } + applied.extend(light.take_committed_entries()); + raw.advance_apply(); + assert_eq!(applied, vec![entry(5, 3)]); + } + let log = PhysicalLog::open(&path, &layout(), 2)?; + verify_installed(&log, &saved)?; + let group = log.group(10)?; + let mut raw = raft::RawNode::new( + &raft::Config { + id: 2, + applied: 4, + ..Default::default() + }, + group.clone(), + &slog::Logger::root(slog::Discard, slog::o!()), + )?; + assert_eq!(raw.ready().take_committed_entries(), vec![entry(5, 3)]); + drop(raw); + group.persist(&[entry(7, 3)], Some(&hard(3, 7, 0)))?; + let next = snapshot(&group, 9, 4)?; + group.install_snapshot(&next, &[], &hard(4, 9, 0))?; + assert_eq!((group.first_index()?, group.last_index()?), (10, 9)); + drop((group, log)); + let log = PhysicalLog::open(&path, &layout(), 2)?; + let group = log.group(10)?; + assert_eq!(group.snapshot(9, 1)?, next); + assert!(read(&group, 10, 10)?.is_empty()); + group.persist(&[entry(10, 4)], Some(&hard(4, 10, 0)))?; + assert_eq!(read(&group, 10, 11)?, vec![entry(10, 4)]); + Ok(()) +} + +#[test] +fn invalid_install_preserves_prior_log_and_vote() -> Result<()> { + let directory = tempfile::tempdir()?; + let log = seed(&directory.path().join("log"))?; + let group = log.group(10)?; + let good = snapshot(&group, 4, 3)?; + let suffix = vec![entry(5, 3), entry(6, 3)]; + let mut cases = vec![ + (snapshot(&group, 1, 1)?, suffix.clone(), hard(3, 5, 0)), + (good.clone(), suffix.clone(), hard(3, 3, 0)), + (good.clone(), suffix.clone(), hard(3, 7, 0)), + (good.clone(), suffix.clone(), hard(3, 5, 4)), + (good.clone(), vec![entry(6, 3)], hard(3, 6, 0)), + (good.clone(), vec![entry(4, 3)], hard(3, 4, 0)), + (good.clone(), vec![entry(5, 2)], hard(3, 5, 0)), + (good.clone(), vec![entry(5, 4)], hard(3, 5, 0)), + (snapshot(&group, 4, 2)?, vec![], hard(2, 4, 1)), + (snapshot(&group, 4, 1)?, vec![], hard(1, 4, 1)), + ]; + for corruption in 0..4 { + let mut invalid = good.clone(); + match corruption { + 0 => invalid.mut_metadata().mut_conf_state().voters = vec![1, 2, 4], + 1 => invalid.data.clear(), + 2 => invalid.data = vec![0; 4096].into(), + _ => invalid.mut_metadata().index = u64::MAX, + } + cases.push((invalid, suffix.clone(), hard(3, 5, 0))); + } + for (snapshot, entries, next_hard) in cases { + assert!(group + .install_snapshot(&snapshot, &entries, &next_hard) + .is_err()); + assert_eq!(group.initial_state()?.hard_state, hard(2, 1, 2)); + assert_eq!(read(&group, 1, 8)?, original()); + } + let unbound = log.group(20)?; + assert!(unbound + .install_snapshot(&snapshot(&unbound, 4, 8)?, &[], &hard(8, 4, 0)) + .is_err()); + group.install_snapshot(&good, &suffix, &hard(3, 5, 0))?; + assert!(group + .install_snapshot(&good, &suffix, &hard(3, 5, 0)) + .is_err()); + verify_installed(&log, &good)?; + let directory = tempfile::tempdir()?; + let log = seed(&directory.path().join("log"))?; + let group = log.group(10)?; + group.persist(&[], Some(&hard(2, 3, 2)))?; + assert!(group + .install_snapshot(&snapshot(&group, 4, 1)?, &[], &hard(3, 4, 0)) + .is_err()); + assert_eq!(group.initial_state()?.hard_state, hard(2, 3, 2)); + assert_eq!(read(&group, 1, 8)?, original()); + 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 install_survives_sigkill"] + fn install_worker() -> Result<()> { + let directory = std::path::PathBuf::from(std::env::var("CONTEXT_INSTALL_TEST_ROOT")?); + let log = seed(&directory.join("log"))?; + let group = log.group(10)?; + if std::env::var("CONTEXT_INSTALL_TEST_PHASE")? == "after" { + group.install_snapshot( + &snapshot(&group, 4, 3)?, + &[entry(5, 3), entry(6, 3)], + &hard(3, 5, 0), + )?; + } + std::fs::write(directory.join("ready"), b"ready")?; + std::thread::sleep(Duration::from_secs(30)); + anyhow::bail!("parent did not stop snapshot worker") + } + #[test] + fn install_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::install_worker", + "--nocapture", + ]) + .env("CONTEXT_INSTALL_TEST_ROOT", directory.path()) + .env("CONTEXT_INSTALL_TEST_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, + "snapshot 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(), 2)?; + let group = log.group(10)?; + if phase == "after" { + verify_installed(&log, &snapshot(&group, 4, 3)?)?; + } else { + assert_eq!(group.initial_state()?.hard_state, hard(2, 1, 2)); + assert_eq!(read(&group, 1, 8)?, original()); + assert_eq!(read(&log.group(20)?, 1, 2)?, vec![entry(1, 7)]); + } + } + Ok(()) + } +} diff --git a/docs/design/raft-log-checkpoints.md b/docs/design/raft-log-checkpoints.md index 563da12..1b24d0c 100644 --- a/docs/design/raft-log-checkpoints.md +++ b/docs/design/raft-log-checkpoints.md @@ -1,4 +1,4 @@ -# Durable local Raft checkpoint boundary +# Durable Raft checkpoint storage Refs https://github.com/lance-format/lance-context/issues/378. @@ -30,9 +30,25 @@ 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. +`GroupLog::install_snapshot` supplies the different transition needed after Raft +accepts a received snapshot. Its cut must advance beyond the previous durable +commit. It replaces the entire prior log, including any divergent uncommitted +tail, and installs the new descriptor, terms, HardState and optional suffix in +one synced batch. Suffix entries begin immediately after the new cut. The +snapshot cannot change static membership, move committed history backwards or +change a vote within the same term. Repeated/stale installs are rejected. + +The engine's `Clean` command removes group metadata as well as entries. The +install batch restores the immutable application binding and every owned group +metadata key; the separate node/layout identity and sibling groups stay intact. +Before giving a snapshot to Raft, the caller must validate and durably stage its +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 does not invoke this API. -Application state/payload packs, remote snapshot installation and transfer, +`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 Lance publication remain subsequent integration work. No production ingestion path or retention default changes. @@ -44,3 +60,10 @@ 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. + +Received-snapshot tests additionally exercise actual RawNode Ready output with +a new suffix, replacement below the old uncommitted last index, an empty suffix, +application-binding preservation, stale/invalid installs, process kills and +injected sync failure. Recovery must expose the complete old state or complete +new state. These tests do not authorize bootstrapping a lost disk under an old +voter identity; replica replacement still needs a separate membership protocol.