feat(wal): add durable quorum append and recovery

.github/workflows/rust-test.yml

@@ -76,6 +76,10 @@ jobs:
         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
+      - name: Replicated WAL quorum and process recovery
+        run: |
+          cargo clippy -p lance-context-wal --features replication --all-targets -- -D warnings
+          cargo test -p lance-context-wal --features replication --lib --test replication_process
       - name: WAL resource observation in a container cgroup namespace
         run: |
           cargo test -p lance-context-wal --features runner --test resource --no-run --message-format=json > "$RUNNER_TEMP/wal-resource-build.jsonl"

crates/lance-context-wal/Cargo.toml

@@ -7,6 +7,7 @@ publish = false
 description = "Independent Rust WAL experiments owned by lance-context"
 
 [features]
+replication = ["raft-log", "record", "dep:bincode", "dep:slog"]
 raft-log = ["placement", "dep:raft", "dep:raft-engine", "dep:raft-proto", "dep:protobuf"]
 vectored-io = ["dep:bytes", "dep:tokio"]
 runner = ["dep:anyhow", "dep:clap", "dep:libc", "dep:serde", "dep:serde_json", "dep:sha2"]
@@ -19,6 +20,8 @@ ipc = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "de
 lab = ["ipc", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "raft-engine/internals", "dep:raft-proto", "dep:serde_json"]
 
 [dependencies]
+bincode = { version = "1.3", optional = true }
+slog = { version = "2", optional = true }
 tokio = { version = "1", features = ["io-util"], optional = true }
 anyhow = { version = "1", optional = true }
 arrow-array = { version = "58", optional = true }

crates/lance-context-wal/src/lib.rs

@@ -11,6 +11,9 @@ pub mod placement;
 #[cfg(feature = "raft-log")]
 pub mod raft_log;
 
+#[cfg(feature = "replication")]
+pub mod replication;
+
 #[cfg(feature = "admission")]
 pub mod admission;
 

crates/lance-context-wal/src/raft_log.rs

@@ -367,6 +367,37 @@ pub struct GroupLog {
 }
 
 impl GroupLog {
+    #[cfg(all(test, feature = "replication"))]
+    pub(crate) fn fail_next_sync_for_test(&self) {
+        self.node.fault.fail_sync_for_test();
+    }
+
+    /// Bind an immutable application format/policy before its first entry.
+    /// Reopen must supply identical bytes; an unidentified existing log cannot
+    /// be silently interpreted as a different state machine.
+    pub fn bind_application(&self, identity: &[u8]) -> Result<()> {
+        ensure!(
+            !identity.is_empty() && identity.len() <= 4096,
+            "invalid application identity"
+        );
+        let state = self
+            .state
+            .write()
+            .map_err(|_| anyhow::anyhow!("group state lock poisoned"))?;
+        self.node.check()?;
+        if let Some(saved) = self.node.engine.get(self.group, b"application") {
+            ensure!(saved == identity, "application format or policy changed");
+            return self.node.check();
+        }
+        ensure!(
+            state.last() == 0,
+            "cannot identify an existing application log"
+        );
+        let mut batch = LogBatch::default();
+        batch.put(self.group, b"application".to_vec(), identity.to_vec())?;
+        self.node.write(&mut batch)
+    }
+
     fn read(&self) -> Result<RwLockReadGuard<'_, GroupState>> {
         self.node.check()?;
         let state = self

crates/lance-context-wal/src/raft_log/fault_tests.rs

@@ -15,11 +15,19 @@ type FileHandle = <DefaultFileSystem as FileSystem>::Handle;
 #[derive(Default)]
 pub(super) struct FaultControl {
     fail_sync: AtomicBool,
+    fail_once: AtomicBool,
     sync_calls: AtomicU64,
     entered: Mutex<Option<mpsc::SyncSender<()>>>,
     release: Mutex<Option<mpsc::Receiver<()>>>,
 }
 
+impl FaultControl {
+    #[cfg(feature = "replication")]
+    pub(super) fn fail_sync_for_test(&self) {
+        self.fail_once.store(true, Ordering::Release);
+    }
+}
+
 pub(super) struct FaultHandle {
     inner: Arc<FileHandle>,
     control: Arc<FaultControl>,
@@ -36,7 +44,9 @@ impl Handle for FaultHandle {
 
     fn sync(&self) -> io::Result<()> {
         self.control.sync_calls.fetch_add(1, Ordering::Relaxed);
-        if self.control.fail_sync.load(Ordering::Acquire) {
+        if self.control.fail_once.swap(false, Ordering::AcqRel)
+            || self.control.fail_sync.load(Ordering::Acquire)
+        {
             if let Some(entered) = self.control.entered.lock().unwrap().take() {
                 entered.send(()).unwrap();
             }