feat(wal): persist received Raft snapshots atomically

.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

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<Self> {
+        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<Vec<u8>> {
@@ -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::<EntryIndex>(self.group, entries)?;

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::<EntryIndex>(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

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 {