feat(wal): persist local Raft checkpoint boundaries

.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

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<TermRange>,
+    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<u64> {
-        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<Vec<Entry>> {
         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<u64> {
-        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<u64> {
-        self.node.check().map_err(storage_error)?;
-        Ok(1)
+        Ok(self.read().map_err(storage_error)?.boundary() + 1)
     }
 
     fn last_index(&self) -> raft::Result<u64> {
         Ok(self.read().map_err(storage_error)?.last())
     }
 
-    fn snapshot(&self, _: u64, _: u64) -> raft::Result<Snapshot> {
-        self.node.check().map_err(storage_error)?;
-        Err(StorageError::SnapshotTemporarilyUnavailable.into())
+    fn snapshot(&self, request_index: u64, _: u64) -> raft::Result<Snapshot> {
+        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())
     }
 }

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(())
+    }
+}

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),