From 5ff4f4e5f53e1d44ec8f4225aac7be92520b03a9 Mon Sep 17 00:00:00 2001 From: Beinan Date: Tue, 20 Oct 2026 01:49:45 +0000 Subject: [PATCH 1/2] feat(wal): add shared durable raft log storage Refs #378 --- .github/workflows/rust-test.yml | 4 + Cargo.lock | 60 ++ crates/lance-context-wal/Cargo.toml | 5 + crates/lance-context-wal/README.md | 57 ++ crates/lance-context-wal/src/lib.rs | 3 + crates/lance-context-wal/src/raft_log.rs | 548 +++++++++++++++ .../src/raft_log/fault_tests.rs | 276 ++++++++ crates/lance-context-wal/tests/raft_log.rs | 635 ++++++++++++++++++ .../tests/raft_log_cache_identity.rs | 132 ++++ 9 files changed, 1720 insertions(+) create mode 100644 crates/lance-context-wal/src/raft_log.rs create mode 100644 crates/lance-context-wal/src/raft_log/fault_tests.rs create mode 100644 crates/lance-context-wal/tests/raft_log.rs create mode 100644 crates/lance-context-wal/tests/raft_log_cache_identity.rs diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index 7159c5b..e83b3c5 100644 --- a/.github/workflows/rust-test.yml +++ b/.github/workflows/rust-test.yml @@ -72,6 +72,10 @@ jobs: cargo test -p lance-context-wal --features arrow-contract,serde_json/preserve_order --lib ingest_contract::arrow - name: WAL experiment process lifecycle tests run: cargo test -p lance-context-wal --features runner --test runner --bin context-wal-resource + - 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 - 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" diff --git a/Cargo.lock b/Cargo.lock index 28adf49..72ecd6c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2388,6 +2388,15 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "erased-serde" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c138974f9d5e7fe373eb04df7cae98833802ae4b11c24ac7039a21d5af4b26c" +dependencies = [ + "serde", +] + [[package]] name = "errno" version = "0.3.14" @@ -2740,6 +2749,15 @@ dependencies = [ "slab", ] +[[package]] +name = "fxhash" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c31b6d751ae2c7f11320402d34e41349dd1016f8d5d45e48c4312bc8625af50c" +dependencies = [ + "byteorder", +] + [[package]] name = "gearhash" version = "0.1.3" @@ -2940,6 +2958,17 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "getset" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cf442baaabe4213ce7d1239afc26c039180b6456da2cededa316ae2c8a77a77" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "git-version" version = "0.3.9" @@ -4195,11 +4224,14 @@ dependencies = [ "clap", "hdrhistogram", "libc", + "protobuf", + "raft", "raft-engine", "raft-proto", "serde", "serde_json", "sha2 0.10.9", + "slog", "tempfile", "tokio", "uuid", @@ -6436,6 +6468,22 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09" +[[package]] +name = "raft" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f12688b23a649902762d4c11d854d73c49c9b93138f2de16403ef9f571ad5bae" +dependencies = [ + "bytes", + "fxhash", + "getset", + "protobuf", + "raft-proto", + "rand 0.8.6", + "slog", + "thiserror 1.0.69", +] + [[package]] name = "raft-engine" version = "0.4.2" @@ -7529,6 +7577,18 @@ version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" +[[package]] +name = "slog" +version = "2.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b3b8565691b22d2bdfc066426ed48f837fc0c5f2c8cad8d9718f7f99d6995c1" +dependencies = [ + "anyhow", + "erased-serde", + "rustversion", + "serde_core", +] + [[package]] name = "smallvec" version = "1.15.1" diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 6de7398..61a9c4f 100644 --- a/crates/lance-context-wal/Cargo.toml +++ b/crates/lance-context-wal/Cargo.toml @@ -7,6 +7,7 @@ publish = false description = "Independent Rust WAL experiments owned by lance-context" [features] +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"] contract = ["dep:serde", "dep:sha2"] @@ -30,12 +31,16 @@ hdrhistogram = { version = "7", optional = true } libc = { version = "0.2", optional = true } raft-engine = { path = "../../vendor/raft-engine", version = "=0.4.2", default-features = false, optional = true } raft-proto = { version = "=0.7.0", optional = true } +raft = { version = "=0.7.0", default-features = false, features = ["protobuf-codec"], optional = true } +protobuf = { version = "2", features = ["bytes"], optional = true } serde_json = { version = "1", optional = true } serde = { version = "1", features = ["derive"], optional = true } sha2 = { version = "0.10", optional = true } uuid = { version = "1", features = ["serde"], optional = true } [dev-dependencies] +libc = "0.2" +slog = "2" tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util"] } bincode = "1.3" arrow-ipc = { version = "58", features = ["zstd"] } diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index 8166273..26a3c6d 100644 --- a/crates/lance-context-wal/README.md +++ b/crates/lance-context-wal/README.md @@ -154,3 +154,60 @@ durability boundary from an empty batch without checking the engine's behavior. For comparisons, preserve the source commit, binary hash, release/debug mode, command, filesystem/storage placement and JSON. CI runs tiny correctness smoke cases; their numbers are not performance evidence. + +## Durable Raft log storage + +The optional `raft-log` feature exposes `raft_log::PhysicalLog` and its cloneable +`GroupLog` handles. One raft-engine directory hosts all groups assigned to a node +by the existing immutable `placement::Layout`. Opening binds the full layout, +node identity and each group's three voters before returning a handle. A changed +identity/layout, a foreign nonempty engine or inconsistent durable metadata is +rejected. The directory format is separate from earlier experiment binaries. + +`GroupLog` implements raft-rs `Storage`. `persist(entries, hard_state)` puts the +entries, term ranges and optional term/vote/commit state in one engine batch, +then waits for a physical sync barrier. Different groups can append concurrently +and share a later barrier. The barrier is a nonempty engine write: the pinned +engine's empty `sync()` call does not sync. Syncing is isolated from concurrent +native engine writes because that engine can panic on sync failure. A write +error or an unwinding write panic invalidates every handle on the physical node; +stop its driver and reopen all handles to resolve the uncertain durable state. +Validation errors before engine submission leave the log usable. + +Uncommitted suffixes may be replaced. Committed entries, votes in the same term, +and monotonic term/commit state are protected. Term lookup uses small cached +ranges rather than reading and decoding entry bodies. Range reads request at +most 256 engine entry descriptors at a time instead of allocating descriptors +for an entire history before applying the byte limit. Returned entries obey the +Raft byte limit, including its required first-entry exception. At a chunk boundary, +one extra entry may be decoded before determining that it does not fit; callers +still need an explicit maximum record size and memory admission. This storage +layer does not decode Arrow, and it does not provide end-to-end zero-copy I/O. + +A successful local `persist` is not a quorum ACK. The subsequent Ready driver +must order durable writes, outgoing messages, commits and state-machine apply +correctly. This increment supports static membership and retains the complete +log; snapshots, compaction, replacement, transport, producer deduplication and +Lance publication are separate integration steps tracked by +[#378](https://github.com/lance-format/lance-context/issues/378). It does not switch +production ingestion. Storage assumes exclusive ownership of the directory and +that successful filesystem syncs survive the supported failures. + +```bash +cargo test -p lance-context-wal --features raft-log --lib --test raft_log --test raft_log_cache_identity +cargo clippy -p lance-context-wal --features raft-log --all-targets -- -D warnings +``` + +Tests use actual raft-engine files for shared-group recovery, conflict replacement, +vote/commit protection and byte-limited reads. Independent-log cases verify cache +isolation for simultaneous engines, dropped handles and a recreated directory; +they require the vendored engine correction described in +`vendor/raft-engine/LANCE_CONTEXT_PATCH.md`. Linux subprocess cases exercise +SIGKILL after a synced prefix (including segment rotation) and a real write +failure imposed only on the child process. Unit tests use a test-only filesystem +wrapper around real files to count sync calls and inject a blocked failing sync +while another group writes, both at the explicit barrier and during rotation. +That wrapper is absent from ordinary builds. A RawNode test drives persistence, +elections, minority isolation and restart with deterministic in-process message +delivery. These tests establish component contracts; they do not establish +power-loss recovery or distributed quorum durability. diff --git a/crates/lance-context-wal/src/lib.rs b/crates/lance-context-wal/src/lib.rs index c76e49f..39ffe25 100644 --- a/crates/lance-context-wal/src/lib.rs +++ b/crates/lance-context-wal/src/lib.rs @@ -8,6 +8,9 @@ #[cfg(feature = "placement")] pub mod placement; +#[cfg(feature = "raft-log")] +pub mod raft_log; + #[cfg(feature = "admission")] pub mod admission; diff --git a/crates/lance-context-wal/src/raft_log.rs b/crates/lance-context-wal/src/raft_log.rs new file mode 100644 index 0000000..06ede7e --- /dev/null +++ b/crates/lance-context-wal/src/raft_log.rs @@ -0,0 +1,548 @@ +//! Synchronous durable log storage for static session-sharded Raft groups. +//! +//! One physical engine serves every group assigned to a node. A successful +//! `persist` writes entries, the term index and HardState in one engine batch, +//! then waits for a shared physical sync barrier. +//! This is a storage boundary, not a quorum ACK: a Ready driver must still obey +//! 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. + +use std::collections::{BTreeMap, BTreeSet}; +use std::path::Path; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard}; + +use anyhow::{ensure, Context, Result}; +use protobuf::Message; +use raft::{GetEntriesContext, RaftState, Storage, StorageError}; +use raft_engine::{Engine, LogBatch, MessageExt}; +use raft_proto::eraftpb::{ConfState, Entry, EntryType, HardState, Snapshot}; + +use crate::placement::Layout; + +#[cfg(test)] +mod fault_tests; + +#[cfg(not(test))] +type LogEngine = Engine; +#[cfg(test)] +type LogEngine = Engine; + +const IDENTITY: &[u8] = b"identity"; +const HARD: &[u8] = b"hard"; +const CONF: &[u8] = b"conf"; +const TERMS: &[u8] = b"terms"; +// 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; + +struct EntryIndex; +impl MessageExt for EntryIndex { + type Entry = Entry; + + fn index(entry: &Entry) -> u64 { + entry.index + } +} + +struct NodeEngine { + engine: LogEngine, + failed: AtomicBool, + append_gate: RwLock<()>, + sync_gate: Mutex<()>, + synced: AtomicU64, + #[cfg(test)] + fault: Arc, +} + +impl NodeEngine { + fn check(&self) -> Result<()> { + ensure!( + !self.failed.load(Ordering::Acquire), + "uncertain engine write: drop every handle and reopen the node log" + ); + Ok(()) + } + + fn write(&self, batch: &mut LogBatch) -> Result<()> { + let result = self.append_and_sync(batch); + if result.is_err() { + self.failed.store(true, Ordering::Release); + } + result + } + + fn append_and_sync(&self, batch: &mut LogBatch) -> Result<()> { + let observed; + { + let _append = self + .append_gate + .read() + .map_err(|_| anyhow::anyhow!("append gate poisoned"))?; + self.check()?; + self.engine_write(batch, false)?; + // Capture while holding the read gate. A later generation can only + // be published by a sync that began after this append completed. + observed = self.synced.load(Ordering::Acquire); + } + let _sync = self + .sync_gate + .lock() + .map_err(|_| anyhow::anyhow!("sync gate poisoned"))?; + self.check()?; + if self.synced.load(Ordering::Acquire) == observed { + // raft-engine 0.4.2 panics on sync failure. Isolate the syncing write + // from native write coalescing: another group must not return a + // successful result while the native group leader is unwinding. + let _exclusive = self + .append_gate + .write() + .map_err(|_| anyhow::anyhow!("append gate poisoned"))?; + self.check()?; + let next = observed + .checked_add(1) + .context("sync generation exhausted")?; + let mut marker = LogBatch::default(); + marker.put(0, b"sync".to_vec(), next.to_le_bytes().to_vec())?; + // Engine::sync() is a no-op in the pinned version: its empty batch + // returns before sync. A nonempty marker is required here. + self.engine_write(&mut marker, true)?; + self.synced.store(next, Ordering::Release); + } + self.check() + } + + fn engine_write(&self, batch: &mut LogBatch, sync: bool) -> Result<()> { + // The caller retains append_gate until failure is published. No other + // group can complete a successful sync barrier after an uncertain write. + match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + self.engine.write(batch, sync) + })) { + Ok(Ok(_)) => Ok(()), + Ok(Err(error)) => { + self.failed.store(true, Ordering::Release); + Err(error).context("uncertain engine write; node log must be reopened") + } + Err(_) => { + self.failed.store(true, Ordering::Release); + anyhow::bail!("engine write panicked; drop every handle and reopen the node log") + } + } + } +} + +#[derive(Clone, Copy)] +struct TermRange { + first: u64, + last: u64, + term: u64, +} + +struct GroupState { + hard: HardState, + conf: ConfState, + terms: Vec, +} + +impl GroupState { + fn last(&self) -> u64 { + self.terms.last().map_or(0, |range| range.last) + } + + fn term(&self, index: u64) -> Option { + if index == 0 { + return Some(0); + } + self.terms + .get(self.terms.partition_point(|range| range.last < index)) + .filter(|range| range.first <= index) + .map(|range| range.term) + } + + fn validate(&self) -> Result<()> { + let (mut previous_last, mut previous_term) = (0, 0); + for range in &self.terms { + ensure!( + range.first == previous_last + 1 + && range.last >= range.first + && range.last < u64::MAX + && range.term > previous_term, + "invalid durable term index" + ); + (previous_last, previous_term) = (range.last, range.term); + } + ensure!( + 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)), + "HardState does not match the durable log or membership" + ); + Ok(()) + } +} + +fn encode_terms(terms: &[TermRange]) -> Result> { + let capacity = terms.len().checked_mul(24).context("term index overflow")?; + let mut bytes = Vec::with_capacity(capacity); + for range in terms { + for value in [range.first, range.last, range.term] { + bytes.extend_from_slice(&value.to_le_bytes()); + } + } + Ok(bytes) +} + +fn decode_terms(bytes: &[u8]) -> Result> { + ensure!( + bytes.len().is_multiple_of(24), + "truncated durable term index" + ); + Ok(bytes + .chunks_exact(24) + .map(|bytes| TermRange { + first: u64::from_le_bytes(bytes[..8].try_into().unwrap()), + last: u64::from_le_bytes(bytes[8..16].try_into().unwrap()), + term: u64::from_le_bytes(bytes[16..].try_into().unwrap()), + }) + .collect()) +} + +/// A node's exclusively opened physical engine and its configured group handles. +/// +/// The layout, node identity and initial membership are bound in one synced batch +/// before any handle is returned. Reopen requires the same complete layout. +pub struct PhysicalLog { + groups: BTreeMap, +} + +impl PhysicalLog { + pub fn open(directory: &Path, layout: &Layout, node: u64) -> Result { + Self::open_with_file_size(directory, layout, node, 64 * 1024 * 1024) + } + + fn open_with_file_size( + directory: &Path, + layout: &Layout, + node: u64, + file_size: u64, + ) -> Result { + layout.validate()?; + ensure!(layout.nodes.contains_key(&node), "unknown local node"); + if !directory.exists() { + std::fs::create_dir(directory)?; + let parent = directory + .parent() + .filter(|path| !path.as_os_str().is_empty()); + std::fs::File::open(parent.unwrap_or_else(|| Path::new(".")))?.sync_all()?; + } + let directory = directory.to_str().context("log directory is not UTF-8")?; + let config = raft_engine::Config { + dir: directory.to_owned(), + batch_compression_threshold: raft_engine::ReadableSize(0), + target_file_size: raft_engine::ReadableSize(file_size), + enable_log_recycle: false, + ..Default::default() + }; + #[cfg(not(test))] + let engine = Engine::open(config)?; + #[cfg(test)] + let (engine, fault) = { + let fs = Arc::new(fault_tests::FaultFileSystem::default()); + let control = fs.control.clone(); + (Engine::open_with_file_system(config, fs)?, control) + }; + let node_id = node; + let node = Arc::new(NodeEngine { + engine, + failed: AtomicBool::new(false), + append_gate: RwLock::new(()), + sync_gate: Mutex::new(()), + synced: AtomicU64::new(0), + #[cfg(test)] + fault, + }); + let identity = serde_json::to_vec(&("lance-context/raft-log/v1", node_id, layout))?; + let assigned: BTreeMap<_, _> = layout + .groups + .iter() + .filter(|(_, voters)| voters.contains(&node_id)) + .map(|(group, voters)| (*group, voters)) + .collect(); + if let Some(saved) = node.engine.get(0, IDENTITY) { + ensure!( + saved == identity, + "node identity, layout or log format changed" + ); + } else { + ensure!( + node.engine.is_empty(), + "cannot adopt a nonempty unbound engine" + ); + let mut batch = LogBatch::default(); + batch.put(0, IDENTITY.to_vec(), identity)?; + for (group, voters) in &assigned { + let mut conf = ConfState::default(); + conf.set_voters((*voters).clone()); + batch.put_message(*group, CONF.to_vec(), &conf)?; + batch.put_message(*group, HARD.to_vec(), &HardState::default())?; + batch.put(*group, TERMS.to_vec(), Vec::new())?; + } + node.write(&mut batch)?; + } + let expected: BTreeSet<_> = assigned.keys().copied().chain([0]).collect(); + ensure!( + node.engine + .raft_groups() + .into_iter() + .collect::>() + == expected, + "durable group inventory differs from the layout" + ); + let mut groups = BTreeMap::new(); + for (group, voters) in assigned { + let mut expected_conf = ConfState::default(); + expected_conf.set_voters(voters.clone()); + let state = GroupState { + hard: node + .engine + .get_message(group, HARD)? + .context("missing HardState")?, + conf: node + .engine + .get_message(group, CONF)? + .context("missing ConfState")?, + terms: decode_terms( + &node + .engine + .get(group, TERMS) + .context("missing term index")?, + )?, + }; + 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), + "entry bounds differ from durable term index" + ); + groups.insert( + group, + GroupLog { + node: node.clone(), + group, + state: Arc::new(RwLock::new(state)), + }, + ); + } + Ok(Self { groups }) + } + + pub fn group(&self, group: u64) -> Result { + self.groups + .get(&group) + .cloned() + .context("group is not hosted on this node") + } +} + +/// Cloneable raft-rs `Storage` view of one group in a shared engine. +/// +/// Clones share a lock: readers cannot observe entries with an old term index +/// while a write is being synced. Different groups can submit engine writes +/// concurrently. Term queries inspect only the small cached range index; they +/// never decode a large entry payload just to compare log terms. +#[derive(Clone)] +pub struct GroupLog { + node: Arc, + group: u64, + state: Arc>, +} + +impl GroupLog { + fn read(&self) -> Result> { + self.node.check()?; + let state = self + .state + .read() + .map_err(|_| anyhow::anyhow!("group state lock poisoned"))?; + self.node.check()?; + Ok(state) + } + + /// Atomically sync a Ready's entries and optional HardState. + /// + /// The entries may replace an uncommitted suffix, but cannot alter a committed + /// prefix, leave a gap, change a vote in the same term or move commit backwards. + /// Validation errors leave this handle usable. An engine write error has an + /// uncertain outcome and invalidates every handle for this physical engine; + /// stop the driver and reopen rather than blindly repeating that Ready. + /// An empty entries/HardState update is a no-op, not a flush barrier. + pub fn persist(&self, entries: &[Entry], hard: Option<&HardState>) -> Result<()> { + if entries.is_empty() && hard.is_none() { + return self.node.check(); + } + let mut state = self + .state + .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, + }; + next.validate()?; + let mut batch = LogBatch::default(); + if !entries.is_empty() { + batch.add_entries::(self.group, entries)?; + batch.put(self.group, TERMS.to_vec(), encode_terms(&next.terms)?)?; + } + if let Some(hard) = hard { + batch.put_message(self.group, HARD.to_vec(), hard)?; + } + if !batch.is_empty() { + self.node.write(&mut batch)?; + } + *state = next; + Ok(()) + } +} + +fn storage_error(error: impl std::fmt::Display) -> raft::Error { + StorageError::Other(Box::new(std::io::Error::other(error.to_string()))).into() +} + +impl Storage for GroupLog { + fn initial_state(&self) -> raft::Result { + let state = self.read().map_err(storage_error)?; + Ok(RaftState { + hard_state: state.hard.clone(), + conf_state: state.conf.clone(), + }) + } + + fn entries( + &self, + low: u64, + high: u64, + max_size: impl Into>, + _: GetEntriesContext, + ) -> raft::Result> { + let state = self.read().map_err(storage_error)?; + if low < 1 { + return Err(StorageError::Compacted.into()); + } + if low > high || high > state.last() + 1 { + return Err(StorageError::Unavailable.into()); + } + let limit = max_size.into().unwrap_or(u64::MAX); + let (mut cursor, mut bytes) = (low, 0u64); + let mut entries = Vec::new(); + while cursor < high { + let end = cursor.saturating_add(READ_CHUNK_ENTRIES).min(high); + let start = entries.len(); + let remaining = usize::try_from(limit.saturating_sub(bytes)).unwrap_or(usize::MAX); + self.node + .engine + .fetch_entries_to::( + self.group, + cursor, + end, + Some(remaining), + &mut entries, + ) + .map_err(storage_error)?; + if entries.len() == start { + return Err(StorageError::Unavailable.into()); + } + for i in start..entries.len() { + let size = u64::from(entries[i].compute_size()); + if i > 0 && size > limit.saturating_sub(bytes) { + entries.truncate(i); + return Ok(entries); + } + bytes = bytes.saturating_add(size); + } + cursor = entries.last().unwrap().index + 1; + if bytes >= limit || cursor < end { + break; + } + } + Ok(entries) + } + + fn term(&self, index: u64) -> raft::Result { + self.read() + .map_err(storage_error)? + .term(index) + .ok_or(StorageError::Unavailable.into()) + } + + fn first_index(&self) -> raft::Result { + self.node.check().map_err(storage_error)?; + Ok(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()) + } +} diff --git a/crates/lance-context-wal/src/raft_log/fault_tests.rs b/crates/lance-context-wal/src/raft_log/fault_tests.rs new file mode 100644 index 0000000..bd1d81b --- /dev/null +++ b/crates/lance-context-wal/src/raft_log/fault_tests.rs @@ -0,0 +1,276 @@ +//! Test-only filesystem adapter: real files with a controlled sync error. No +//! fault adapter or injection flag is included in ordinary library/binary builds. +use std::io; +use std::path::Path; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{mpsc, Arc, Mutex}; +use std::time::Duration; + +use raft_engine::env::{DefaultFileSystem, FileSystem, Handle, Permission}; + +use super::*; + +type FileHandle = ::Handle; + +#[derive(Default)] +pub(super) struct FaultControl { + fail_sync: AtomicBool, + sync_calls: AtomicU64, + entered: Mutex>>, + release: Mutex>>, +} + +pub(super) struct FaultHandle { + inner: Arc, + control: Arc, +} + +impl Handle for FaultHandle { + fn truncate(&self, offset: usize) -> io::Result<()> { + self.inner.truncate(offset) + } + + fn file_size(&self) -> io::Result { + self.inner.file_size() + } + + fn sync(&self) -> io::Result<()> { + self.control.sync_calls.fetch_add(1, Ordering::Relaxed); + if self.control.fail_sync.load(Ordering::Acquire) { + if let Some(entered) = self.control.entered.lock().unwrap().take() { + entered.send(()).unwrap(); + } + if let Some(release) = self.control.release.lock().unwrap().take() { + release.recv_timeout(Duration::from_secs(10)).unwrap(); + } + return Err(io::Error::other("injected file sync failure")); + } + self.inner.sync() + } +} + +#[test] +fn concurrent_appends_share_one_real_sync() -> Result<()> { + let layout = Layout { + cluster: uuid::Uuid::from_u128(2), + 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 log = PhysicalLog::open(&directory.path().join("log"), &layout, 1)?; + let node = log.group(10)?.node.clone(); + let before = node.fault.sync_calls.load(Ordering::Relaxed); + let gate = node.sync_gate.lock().unwrap(); + let writers: Vec<_> = [10, 20] + .into_iter() + .map(|id| { + let group = log.group(id).unwrap(); + std::thread::spawn(move || { + group.persist( + &[Entry { + index: 1, + term: 1, + data: vec![id as u8; 4096].into(), + ..Default::default() + }], + Some(&HardState { + term: 1, + vote: 1, + commit: 1, + ..Default::default() + }), + ) + }) + }) + .collect(); + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while node.engine.last_index(10) != Some(1) || node.engine.last_index(20) != Some(1) { + assert!( + std::time::Instant::now() < deadline, + "both appends must reach the real engine before sync" + ); + std::thread::sleep(Duration::from_millis(1)); + } + assert_eq!(node.fault.sync_calls.load(Ordering::Relaxed), before); + drop(gate); + for writer in writers { + writer.join().unwrap()?; + } + assert_eq!(node.fault.sync_calls.load(Ordering::Relaxed) - before, 1); + for id in [10, 20] { + let group = log.group(id)?; + assert_eq!(group.initial_state()?.hard_state.commit, 1); + assert_eq!( + group.entries(1, 2, None, GetEntriesContext::empty(false))?[0] + .data + .as_ref(), + vec![id as u8; 4096] + ); + } + // An append submitted after that barrier needs a new physical sync. + log.group(10)?.persist( + &[], + Some(&HardState { + term: 2, + vote: 2, + commit: 1, + ..Default::default() + }), + )?; + assert_eq!(node.fault.sync_calls.load(Ordering::Relaxed) - before, 2); + Ok(()) +} + +#[derive(Default)] +pub(super) struct FaultFileSystem { + pub(super) control: Arc, +} + +impl FaultFileSystem { + fn wrap(&self, inner: FileHandle) -> FaultHandle { + FaultHandle { + inner: Arc::new(inner), + control: self.control.clone(), + } + } +} + +impl FileSystem for FaultFileSystem { + type Handle = FaultHandle; + type Reader = ::Reader; + type Writer = ::Writer; + + fn create>(&self, path: P) -> io::Result { + Ok(self.wrap(DefaultFileSystem.create(path)?)) + } + + fn open>(&self, path: P, permission: Permission) -> io::Result { + Ok(self.wrap(DefaultFileSystem.open(path, permission)?)) + } + + fn delete>(&self, path: P) -> io::Result<()> { + DefaultFileSystem.delete(path) + } + + fn rename>(&self, source: P, destination: P) -> io::Result<()> { + DefaultFileSystem.rename(source, destination) + } + + fn new_reader(&self, handle: Arc) -> io::Result { + DefaultFileSystem.new_reader(handle.inner.clone()) + } + + fn new_writer(&self, handle: Arc) -> io::Result { + DefaultFileSystem.new_writer(handle.inner.clone()) + } +} + +#[test] +fn sync_panic_cannot_ack_another_group() -> Result<()> { + sync_failure_cannot_ack_another_group(64 * 1024 * 1024) +} + +#[test] +fn rotation_sync_panic_cannot_ack_another_group() -> Result<()> { + // Rotation also calls Handle::sync, even for an engine write(sync=false). + // Force rotation at every append to exercise that different panic boundary. + sync_failure_cannot_ack_another_group(1) +} + +fn sync_failure_cannot_ack_another_group(file_size: u64) -> Result<()> { + let layout = Layout { + cluster: uuid::Uuid::from_u128(1), + 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_with_file_size(&path, &layout, 1, file_size)?; + let first = log.group(10)?; + let second = log.group(20)?; + let prefix = Entry { + index: 1, + term: 1, + data: vec![7; 4096].into(), + ..Default::default() + }; + first.persist( + std::slice::from_ref(&prefix), + Some(&HardState { + term: 1, + vote: 1, + commit: 1, + ..Default::default() + }), + )?; + + let control = first.node.fault.clone(); + let (entered_tx, entered_rx) = mpsc::sync_channel(1); + let (release_tx, release_rx) = mpsc::sync_channel(1); + *control.entered.lock().unwrap() = Some(entered_tx); + *control.release.lock().unwrap() = Some(release_rx); + control.fail_sync.store(true, Ordering::Release); + let writer = first.clone(); + let a = std::thread::spawn(move || { + writer.persist( + &[Entry { + index: 2, + term: 1, + data: vec![8; 4096].into(), + ..Default::default() + }], + Some(&HardState { + term: 1, + vote: 1, + commit: 2, + ..Default::default() + }), + ) + }); + entered_rx.recv_timeout(Duration::from_secs(10))?; + let (finished_tx, finished_rx) = mpsc::sync_channel(1); + let (started_tx, started_rx) = mpsc::sync_channel(1); + let writer = second.clone(); + let b = std::thread::spawn(move || { + started_tx.send(()).unwrap(); + let result = writer.persist( + &[], + Some(&HardState { + term: 1, + vote: 1, + commit: 0, + ..Default::default() + }), + ); + finished_tx.send(result).unwrap(); + }); + started_rx.recv_timeout(Duration::from_secs(10))?; + assert!(finished_rx.recv_timeout(Duration::from_millis(50)).is_err()); + release_tx.send(())?; + assert!( + a.join().unwrap().is_err(), + "sync panic must not become a successful persist" + ); + assert!(finished_rx.recv_timeout(Duration::from_secs(10))?.is_err()); + b.join().unwrap(); + assert!(first.initial_state().is_err()); + assert!(second.initial_state().is_err()); + assert!(first.persist(&[], None).is_err()); + control.fail_sync.store(false, Ordering::Release); + drop((first, second, log)); + let reopened = PhysicalLog::open(&path, &layout, 1)?; + let group = reopened.group(10)?; + assert!(group.initial_state()?.hard_state.commit >= 1); + assert_eq!( + group.entries(1, 2, None, GetEntriesContext::empty(false))?, + vec![prefix] + ); + assert_eq!(reopened.group(20)?.initial_state()?.hard_state.term, 0); + Ok(()) +} diff --git a/crates/lance-context-wal/tests/raft_log.rs b/crates/lance-context-wal/tests/raft_log.rs new file mode 100644 index 0000000..c755af1 --- /dev/null +++ b/crates/lance-context-wal/tests/raft_log.rs @@ -0,0 +1,635 @@ +#![cfg(feature = "raft-log")] + +use std::collections::BTreeMap; + +use anyhow::Result; +use lance_context_wal::placement::Layout; +use lance_context_wal::raft_log::{GroupLog, PhysicalLog}; +use protobuf::Message; +use raft::{GetEntriesContext, Storage}; +use raft_proto::eraftpb::{Entry, EntryType, HardState}; +use uuid::Uuid; + +fn layout() -> Layout { + Layout { + cluster: Uuid::from_u128(1), + 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], + } +} + +fn hard(term: u64, vote: u64, commit: u64) -> HardState { + HardState { + term, + vote, + commit, + ..Default::default() + } +} + +fn entry(index: u64, term: u64, body: Vec) -> Entry { + Entry { + index, + term, + data: body.into(), + ..Default::default() + } +} + +fn read(log: &GroupLog, low: u64, high: u64, limit: Option) -> raft::Result> { + log.entries(low, high, limit, GetEntriesContext::empty(false)) +} + +#[test] +fn recovers_two_groups_and_rejects_layout_changes() -> Result<()> { + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + let entries = vec![ + entry(1, 1, vec![1; 4096]), + entry(2, 3, vec![2; 8192]), + entry(3, 3, vec![3; 2048]), + ]; + { + let log = PhysicalLog::open(&path, &layout(), 1)?; + assert!(log.group(20)?.persist(&[], Some(&hard(0, 1, 0))).is_err()); + log.group(10)?.persist(&entries, Some(&hard(3, 2, 2)))?; + log.group(20)? + .persist(&[entry(1, 8, vec![9; 4096])], Some(&hard(8, 3, 1)))?; + assert!(log.group(99).is_err()); + assert!( + PhysicalLog::open(&path, &layout(), 1).is_err(), + "exclusive physical engine lock" + ); + } + { + let log = PhysicalLog::open(&path, &layout(), 1)?; + let group = log.group(10)?; + assert_eq!(group.initial_state()?.hard_state, hard(3, 2, 2)); + assert_eq!(group.initial_state()?.conf_state.voters, vec![1, 2, 3]); + assert_eq!(read(&group, 1, 4, None)?, entries); + assert_eq!( + ( + group.term(0)?, + group.term(1)?, + group.term(2)?, + group.term(3)? + ), + (0, 1, 3, 3) + ); + assert_eq!(log.group(20)?.initial_state()?.hard_state, hard(8, 3, 1)); + assert_eq!( + read(&log.group(20)?, 1, 2, None)?[0].data.as_ref(), + vec![9; 4096] + ); + } + let mut changed = layout(); + changed + .nodes + .insert(2, "changed.example.invalid:7800".into()); + assert!(PhysicalLog::open(&path, &changed, 1).is_err()); + assert!(PhysicalLog::open(&path, &layout(), 2).is_err()); + let log = PhysicalLog::open(&path, &layout(), 1)?; + assert_eq!(read(&log.group(10)?, 1, 4, None)?, entries); + Ok(()) +} + +#[test] +fn foreign_engine_and_invalid_metadata_are_rejected_without_rebinding() -> Result<()> { + fn engine(path: &std::path::Path) -> Result { + Ok(raft_engine::Engine::open(raft_engine::Config { + dir: path.to_str().unwrap().to_owned(), + ..Default::default() + })?) + } + let directory = tempfile::tempdir()?; + let foreign = directory.path().join("foreign"); + { + let engine = engine(&foreign)?; + let mut batch = raft_engine::LogBatch::default(); + batch.put(99, b"foreign".to_vec(), b"retain-me".to_vec())?; + engine.write(&mut batch, true)?; + } + assert!(PhysicalLog::open(&foreign, &layout(), 1).is_err()); + { + let engine = engine(&foreign)?; + assert_eq!(engine.get(99, b"foreign"), Some(b"retain-me".to_vec())); + assert!(engine.get(0, b"identity").is_none()); + } + let invalid = directory.path().join("invalid"); + { + let log = PhysicalLog::open(&invalid, &layout(), 1)?; + log.group(10)? + .persist(&[entry(1, 1, vec![9; 4096])], Some(&hard(1, 1, 1)))?; + } + { + let engine = engine(&invalid)?; + let mut batch = raft_engine::LogBatch::default(); + batch.put(10, b"terms".to_vec(), b"invalid".to_vec())?; + engine.write(&mut batch, true)?; + } + assert!(PhysicalLog::open(&invalid, &layout(), 1).is_err()); + let engine = engine(&invalid)?; + assert_eq!(engine.last_index(10), Some(1)); + assert_eq!( + engine.get_message::(10, b"hard")?, + Some(hard(1, 1, 1)) + ); + assert_eq!(engine.get(10, b"terms"), Some(b"invalid".to_vec())); + Ok(()) +} + +#[test] +fn replaces_only_uncommitted_suffix_and_preserves_vote() -> Result<()> { + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + let expected = vec![ + entry(1, 1, vec![1]), + entry(2, 1, vec![2]), + entry(3, 2, vec![33]), + entry(4, 2, vec![44]), + ]; + { + let log = PhysicalLog::open(&path, &layout(), 1)?; + let group = log.group(10)?; + let initial: Vec<_> = (1..=5) + .map(|index| entry(index, 1, vec![index as u8])) + .collect(); + group.persist(&initial, Some(&hard(1, 1, 2)))?; + group.persist(&expected[2..], Some(&hard(2, 2, 3)))?; + assert_eq!(group.last_index()?, 4); + assert!(group.term(5).is_err()); + for invalid in [ + hard(1, 1, 3), + hard(2, 3, 3), + hard(2, 0, 3), + hard(2, 2, 2), + hard(2, 2, 5), + ] { + assert!(group.persist(&[], Some(&invalid)).is_err()); + } + for invalid in [ + vec![entry(3, 2, vec![99])], + vec![entry(6, 2, vec![])], + vec![entry(5, 2, vec![]), entry(7, 2, vec![])], + vec![entry(5, 1, vec![])], + vec![entry(5, 3, vec![])], + ] { + assert!(group.persist(&invalid, None).is_err()); + } + let mut membership = entry(5, 2, vec![]); + membership.set_entry_type(EntryType::EntryConfChange); + assert!(group.persist(&[membership], None).is_err()); + assert_eq!(read(&group, 1, 5, None)?, expected); + assert_eq!(group.initial_state()?.hard_state, hard(2, 2, 3)); + } + let log = PhysicalLog::open(&path, &layout(), 1)?; + assert_eq!(read(&log.group(10)?, 1, 5, None)?, expected); + assert_eq!(log.group(10)?.initial_state()?.hard_state, hard(2, 2, 3)); + Ok(()) +} + +#[test] +fn read_size_and_chunk_boundaries_obey_raft_storage_contract() -> Result<()> { + let directory = tempfile::tempdir()?; + let log = PhysicalLog::open(&directory.path().join("log"), &layout(), 1)?; + let group = log.group(10)?; + let entries: Vec<_> = (1..=300) + .map(|index| { + entry( + index, + 1, + vec![index as u8; if index == 257 { 4 * 1024 * 1024 } else { 1024 }], + ) + }) + .collect(); + group.persist(&entries, Some(&hard(1, 1, 300)))?; + assert_eq!(read(&group, 1, 301, Some(0))?, entries[..1]); + assert_eq!(read(&group, 257, 301, Some(1))?, entries[256..257]); + let prefix_size: u64 = entries[..256] + .iter() + .map(|entry| u64::from(entry.compute_size())) + .sum(); + assert_eq!( + read(&group, 1, 301, Some(prefix_size + 10))?, + entries[..256] + ); + assert_eq!(read(&group, 1, 301, None)?, entries); + assert!(read(&group, 10, 10, Some(0))?.is_empty()); + assert!(read(&group, 0, 1, None).is_err()); + assert!(read(&group, 1, 302, None).is_err()); + assert!(read(&group, 10, 9, None).is_err()); + assert!(group.snapshot(1, 2).is_err()); + Ok(()) +} + +#[test] +fn concurrent_groups_share_engine_without_sharing_positions() -> Result<()> { + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + { + let log = PhysicalLog::open(&path, &layout(), 1)?; + let writers: Vec<_> = [10, 20] + .into_iter() + .map(|id| { + let group = log.group(id).unwrap(); + std::thread::spawn(move || -> Result<()> { + for index in 1..=32 { + group.persist( + &[entry(index, id, vec![id as u8; 8192])], + Some(&hard(id, 1, index)), + )?; + } + Ok(()) + }) + }) + .collect(); + for writer in writers { + writer.join().unwrap()?; + } + } + let log = PhysicalLog::open(&path, &layout(), 1)?; + for id in [10, 20] { + let group = log.group(id)?; + assert_eq!(group.initial_state()?.hard_state, hard(id, 1, 32)); + let actual = read(&group, 1, 33, None)?; + assert_eq!(actual.len(), 32); + assert!(actual + .iter() + .all(|entry| entry.term == id && entry.data.as_ref() == vec![id as u8; 8192])); + } + Ok(()) +} + +#[test] +fn raft_ready_persistence_preserves_committed_data_through_election_and_restart() -> Result<()> { + use raft::{Config, RawNode, StateRole}; + use std::path::Path; + + struct Replica { + raw: RawNode, + log: GroupLog, + applied: Vec, + _engine: PhysicalLog, + } + fn open(root: &Path, id: u64) -> Result { + let engine = PhysicalLog::open(&root.join(format!("node-{id}")), &layout(), id)?; + let log = engine.group(10)?; + let config = Config { + id, + election_tick: 10, + heartbeat_tick: 1, + ..Default::default() + }; + let logger = slog::Logger::root(slog::Discard, slog::o!()); + let raw = RawNode::new(&config, log.clone(), &logger)?; + Ok(Replica { + raw, + log, + applied: Vec::new(), + _engine: engine, + }) + } + fn apply(replica: &mut Replica, committed: Vec) { + for entry in committed { + assert_eq!( + entry.index, + replica.applied.last().map_or(1, |old| old.index + 1) + ); + replica.applied.push(entry); + } + } + // Manually initiated elections and deterministic in-process message delivery; + // only the storage/Ready contract is under test, not failure-detector timing. + // Each replica has real independent engine files. This is not a network or + // multiprocess throughput/failover benchmark. + fn drive(replicas: &mut BTreeMap) -> Result<()> { + for _ in 0..1000 { + let mut changed = false; + let mut messages = Vec::new(); + for replica in replicas.values_mut() { + if !replica.raw.has_ready() { + continue; + } + changed = true; + let mut ready = replica.raw.ready(); + assert_eq!(ready.snapshot().get_metadata().index, 0); + replica.log.persist(ready.entries(), ready.hs())?; + messages.extend(ready.take_messages()); + messages.extend(ready.take_persisted_messages()); + apply(replica, ready.take_committed_entries()); + let mut light = replica.raw.advance_append(ready); + if light.commit_index().is_some() { + replica + .log + .persist(&[], Some(&replica.raw.raft.hard_state()))?; + } + messages.extend(light.take_messages()); + apply(replica, light.take_committed_entries()); + replica.raw.advance_apply(); + } + for message in messages { + if let Some(replica) = replicas.get_mut(&message.to) { + replica.raw.step(message)?; + } + } + if !changed { + return Ok(()); + } + } + anyhow::bail!("Ready delivery did not quiesce") + } + fn durable_data(replica: &Replica) -> Result>> { + Ok(read(&replica.log, 1, replica.log.last_index()? + 1, None)? + .into_iter() + .filter(|entry| !entry.data.is_empty()) + .map(|entry| entry.data.to_vec()) + .collect()) + } + let root = tempfile::tempdir()?; + let mut replicas: BTreeMap<_, _> = (1..=3) + .map(|id| Ok((id, open(root.path(), id)?))) + .collect::>()?; + replicas.get_mut(&1).unwrap().raw.campaign()?; + drive(&mut replicas)?; + assert_eq!(replicas[&1].raw.raft.state, StateRole::Leader); + let first = vec![11; 8192]; + let second = vec![22; 16384]; + replicas + .get_mut(&1) + .unwrap() + .raw + .propose(Vec::new(), first.clone())?; + drive(&mut replicas)?; + assert!(replicas.values().all(|replica| replica + .applied + .iter() + .any(|entry| entry.data.as_ref() == first))); + replicas.remove(&1); + replicas.get_mut(&2).unwrap().raw.campaign()?; + drive(&mut replicas)?; + assert_eq!(replicas[&2].raw.raft.state, StateRole::Leader); + replicas + .get_mut(&2) + .unwrap() + .raw + .propose(Vec::new(), second.clone())?; + drive(&mut replicas)?; + replicas.insert(1, open(root.path(), 1)?); + replicas.get_mut(&2).unwrap().raw.tick(); + drive(&mut replicas)?; + let expected = vec![first, second]; + for replica in replicas.values() { + assert_eq!(durable_data(replica)?, expected); + assert_eq!( + replica.log.initial_state()?.hard_state.commit, + replica.log.last_index()? + ); + } + let committed = replicas[&2].log.initial_state()?.hard_state.commit; + replicas.remove(&1); + replicas.remove(&3); + replicas + .get_mut(&2) + .unwrap() + .raw + .propose(Vec::new(), vec![99; 4096])?; + drive(&mut replicas)?; + assert_eq!( + replicas[&2].log.initial_state()?.hard_state.commit, + committed + ); + assert!( + replicas[&2].log.last_index()? > committed, + "local persistence is not quorum commit" + ); + drop(replicas); + let mut replicas: BTreeMap<_, _> = (1..=3) + .map(|id| Ok((id, open(root.path(), id)?))) + .collect::>()?; + replicas.get_mut(&1).unwrap().raw.campaign()?; + drive(&mut replicas)?; + assert_eq!(replicas[&1].raw.raft.state, StateRole::Leader); + for replica in replicas.values() { + assert_eq!(durable_data(replica)?, expected); + assert_eq!( + replica.log.initial_state()?.hard_state.commit, + replica.log.last_index()? + ); + } + let third = vec![33; 4096]; + replicas + .get_mut(&1) + .unwrap() + .raw + .propose(Vec::new(), third.clone())?; + drive(&mut replicas)?; + let mut expected = expected; + expected.push(third); + for replica in replicas.values() { + assert_eq!(durable_data(replica)?, expected); + assert_eq!( + replica.log.initial_state()?.hard_state.commit, + replica.log.last_index()? + ); + let applied: Vec<_> = replica + .applied + .iter() + .filter(|entry| !entry.data.is_empty()) + .map(|entry| entry.data.to_vec()) + .collect(); + assert_eq!(applied, expected); + } + Ok(()) +} + +#[cfg(target_os = "linux")] +mod process { + use super::*; + use std::io::Write; + use std::process::{Child, Command, ExitStatus, Stdio}; + use std::time::{Duration, Instant}; + + struct Worker(Child); + + impl Drop for Worker { + fn drop(&mut self) { + // Even a failed assertion must not leave the disposable worker alive. + if !matches!(self.0.try_wait(), Ok(Some(_))) { + let _ = self.0.kill(); + } + let _ = self.0.wait(); + } + } + + impl Worker { + fn wait(&mut self, deadline: Instant) -> Result { + loop { + if let Some(status) = self.0.try_wait()? { + return Ok(status); + } + anyhow::ensure!(Instant::now() < deadline, "worker exit timed out"); + std::thread::sleep(Duration::from_millis(10)); + } + } + } + + fn prefix_len(mode: &str) -> u64 { + if mode == "rotate-kill" { + 272 + } else { + 8 + } + } + + fn body(index: u64) -> Vec { + (0..256 * 1024) + .map(|offset| (offset as u64).wrapping_mul(17).wrapping_add(index) as u8) + .collect() + } + + #[test] + #[ignore = "invoked in an isolated subprocess by the crash and write-error tests"] + fn journal_worker() -> Result<()> { + let path = std::env::var("CONTEXT_WAL_LOG_TEST_DIRECTORY")?; + let mode = std::env::var("CONTEXT_WAL_LOG_TEST_MODE")?; + let log = PhysicalLog::open(Path::new(&path), &layout(), 1)?; + let group = log.group(10)?; + for index in 1..=prefix_len(&mode) { + group.persist(&[entry(index, 1, body(index))], Some(&hard(1, 1, index)))?; + } + println!("DURABLE_PREFIX_READY"); + std::io::stdout().flush()?; + if mode == "kill" || mode == "rotate-kill" { + std::thread::sleep(std::time::Duration::from_secs(60)); + anyhow::bail!("parent did not kill worker"); + } + assert_eq!(mode, "write-error"); + // Confined to this disposable child. Force an actual file write failure + // after a synced prefix; restore the limit before closing the engine. + let mut saved = libc::rlimit { + rlim_cur: 0, + rlim_max: 0, + }; + unsafe { + assert_eq!(libc::getrlimit(libc::RLIMIT_FSIZE, &mut saved), 0); + libc::signal(libc::SIGXFSZ, libc::SIG_IGN); + let limit = libc::rlimit { + rlim_cur: 0, + rlim_max: saved.rlim_max, + }; + assert_eq!(libc::setrlimit(libc::RLIMIT_FSIZE, &limit), 0); + } + let result = group.persist(&[entry(9, 1, body(9))], Some(&hard(1, 1, 9))); + unsafe { + assert_eq!(libc::setrlimit(libc::RLIMIT_FSIZE, &saved), 0); + } + assert!( + result.is_err(), + "file size limit must fail the actual engine write" + ); + assert!(group.term(1).is_err()); + assert!(group.initial_state().is_err()); + assert!(log.group(20)?.persist(&[], Some(&hard(1, 1, 0))).is_err()); + assert!(log.group(20)?.first_index().is_err()); + println!("ENGINE_HANDLES_FENCED"); + Ok(()) + } + + use std::path::Path; + + fn run_child(mode: &str) -> Result<()> { + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + let stdout = directory.path().join("worker.stdout"); + let stderr = directory.path().join("worker.stderr"); + // Regular files avoid pipe backpressure and unbounded blocking reads. + let mut child = Worker( + Command::new(std::env::current_exe()?) + .args([ + "--exact", + "process::journal_worker", + "--ignored", + "--nocapture", + ]) + .env("CONTEXT_WAL_LOG_TEST_DIRECTORY", &path) + .env("CONTEXT_WAL_LOG_TEST_MODE", mode) + .stdout(Stdio::from(std::fs::File::create(&stdout)?)) + .stderr(Stdio::from(std::fs::File::create(&stderr)?)) + .spawn()?, + ); + let deadline = Instant::now() + Duration::from_secs(60); + loop { + let status = child.0.try_wait()?; + if std::fs::read_to_string(&stdout)?.contains("DURABLE_PREFIX_READY") { + break; + } + anyhow::ensure!( + status.is_none() && Instant::now() < deadline, + "worker failed to report synced prefix: {status:?}; {}", + std::fs::read_to_string(&stderr)? + ); + std::thread::sleep(Duration::from_millis(10)); + } + if mode == "kill" || mode == "rotate-kill" { + child.0.kill()?; + use std::os::unix::process::ExitStatusExt; + assert_eq!(child.wait(deadline)?.signal(), Some(9)); + } else { + assert!( + child.wait(deadline)?.success(), + "{}", + std::fs::read_to_string(&stderr)? + ); + assert!(std::fs::read_to_string(&stdout)?.contains("ENGINE_HANDLES_FENCED")); + } + if mode == "rotate-kill" { + let segments = std::fs::read_dir(&path)? + .map(|entry| { + Ok(entry? + .path() + .extension() + .is_some_and(|ext| ext == "raftlog")) + }) + .collect::>>()?; + assert!(segments.into_iter().filter(|is_log| *is_log).count() > 1); + } + let log = PhysicalLog::open(&path, &layout(), 1)?; + let group = log.group(10)?; + let state = group.initial_state()?.hard_state; + assert_eq!((state.term, state.vote), (1, 1)); + let uncertain_suffix = u64::from(mode == "write-error"); + assert!( + state.commit >= prefix_len(mode) && state.commit <= prefix_len(mode) + uncertain_suffix + ); + assert_eq!(group.last_index()?, state.commit); + for start in (1..=state.commit).step_by(16) { + let end = (start + 16).min(state.commit + 1); + let entries = read(&group, start, end, None)?; + assert_eq!(entries.len() as u64, end - start); + for (index, entry) in (start..end).zip(entries) { + assert_eq!(entry.index, index); + assert_eq!(entry.data.as_ref(), body(entry.index)); + assert_eq!(group.term(entry.index)?, 1); + } + } + Ok(()) + } + + #[test] + fn acknowledged_prefix_survives_sigkill() -> Result<()> { + run_child("kill") + } + + #[test] + fn acknowledged_prefix_survives_rotation_and_sigkill() -> Result<()> { + run_child("rotate-kill") + } + + #[test] + fn write_error_fences_all_groups_and_reopen_preserves_synced_prefix() -> Result<()> { + run_child("write-error") + } +} diff --git a/crates/lance-context-wal/tests/raft_log_cache_identity.rs b/crates/lance-context-wal/tests/raft_log_cache_identity.rs new file mode 100644 index 0000000..ab21090 --- /dev/null +++ b/crates/lance-context-wal/tests/raft_log_cache_identity.rs @@ -0,0 +1,132 @@ +#![cfg(feature = "raft-log")] + +//! Independent physical logs must never reuse another engine's cached payloads. +use std::collections::BTreeMap; + +use anyhow::{ensure, Result}; +use lance_context_wal::placement::Layout; +use lance_context_wal::raft_log::{GroupLog, PhysicalLog}; +use raft::{GetEntriesContext, Storage}; +use raft_proto::eraftpb::{Entry, HardState}; +use uuid::Uuid; + +fn layout() -> Layout { + Layout { + cluster: Uuid::from_u128(1), + nodes: (1..=3) + .map(|id| (id, format!("wal-{id}.example.invalid:7800"))) + .collect(), + groups: BTreeMap::from([(10, vec![1, 2, 3])]), + buckets: vec![10], + } +} + +fn append(group: &GroupLog, byte: u8) -> Result> { + // Identical serialized sizes and write ordering make block handles collide + // across independent engines. The full payload must identify its owner. + let entries: Vec<_> = (1..=3) + .map(|index| Entry { + index, + term: 1, + data: vec![byte; 4096].into(), + ..Default::default() + }) + .collect(); + group.persist( + &entries, + Some(&HardState { + term: 1, + vote: 1, + commit: 3, + ..Default::default() + }), + )?; + Ok(entries) +} + +fn read(group: &GroupLog) -> Result> { + Ok(group.entries(1, 4, None, GetEntriesContext::empty(false))?) +} + +fn verify(group: &GroupLog, expected: &[Entry]) -> Result<()> { + let actual = read(group)?; + ensure!(actual.len() == expected.len(), "unexpected entry count"); + for (actual, expected) in actual.iter().zip(expected) { + ensure!( + actual.index == expected.index && actual.term == expected.term, + "unexpected entry index or term" + ); + ensure!( + actual.data == expected.data, + "wrong physical log payload at index {}: expected byte {:?}, actual byte {:?}", + actual.index, + expected.data.first(), + actual.data.first() + ); + ensure!(actual == expected, "unexpected entry metadata"); + } + Ok(()) +} + +#[test] +fn independent_physical_logs_return_their_own_payloads_on_one_thread() -> Result<()> { + let directory = tempfile::tempdir()?; + let a = PhysicalLog::open(&directory.path().join("a"), &layout(), 1)?; + let b = PhysicalLog::open(&directory.path().join("b"), &layout(), 2)?; + let group_a = a.group(10)?; + let group_b = b.group(10)?; + let expected_a = append(&group_a, 41)?; + let expected_b = append(&group_b, 42)?; + + // Fresh-thread controls establish that each actual file contains its own + // data before exercising alternating cache ownership on the caller thread. + for (group, expected) in [ + (group_a.clone(), expected_a.clone()), + (group_b.clone(), expected_b.clone()), + ] { + std::thread::spawn(move || -> Result<()> { verify(&group, &expected) }) + .join() + .expect("control reader panicked")?; + } + for _ in 0..3 { + verify(&group_a, &expected_a)?; + verify(&group_b, &expected_b)?; + } + Ok(()) +} + +#[test] +fn dropping_every_handle_does_not_alias_a_later_physical_log() -> Result<()> { + let directory = tempfile::tempdir()?; + { + let log = PhysicalLog::open(&directory.path().join("a"), &layout(), 1)?; + let group = log.group(10)?; + let expected = append(&group, 51)?; + verify(&group, &expected)?; + } + let log = PhysicalLog::open(&directory.path().join("b"), &layout(), 1)?; + let group = log.group(10)?; + let expected = append(&group, 52)?; + verify(&group, &expected)?; + Ok(()) +} + +#[test] +fn a_recreated_directory_does_not_reuse_the_prior_cache_identity() -> Result<()> { + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + { + let log = PhysicalLog::open(&path, &layout(), 1)?; + let group = log.group(10)?; + let expected = append(&group, 61)?; + verify(&group, &expected)?; + } + // Only this test's TempDir is removed. Reuse of the exact pathname rules out + // a path-only cache key as a sufficient dependency correction. + std::fs::remove_dir_all(&path)?; + let log = PhysicalLog::open(&path, &layout(), 1)?; + let group = log.group(10)?; + let expected = append(&group, 62)?; + verify(&group, &expected)?; + Ok(()) +} From 5603e4da01ebc78b8b2693d83727a9ea97166201 Mon Sep 17 00:00:00 2001 From: Beinan Date: Tue, 20 Oct 2026 02:44:29 +0000 Subject: [PATCH 2/2] fix(wal): use typed chunks for durable term decoding --- crates/lance-context-wal/src/raft_log.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/lance-context-wal/src/raft_log.rs b/crates/lance-context-wal/src/raft_log.rs index 06ede7e..6011c0f 100644 --- a/crates/lance-context-wal/src/raft_log.rs +++ b/crates/lance-context-wal/src/raft_log.rs @@ -204,7 +204,9 @@ fn decode_terms(bytes: &[u8]) -> Result> { "truncated durable term index" ); Ok(bytes - .chunks_exact(24) + .as_chunks::<24>() + .0 + .iter() .map(|bytes| TermRange { first: u64::from_le_bytes(bytes[..8].try_into().unwrap()), last: u64::from_le_bytes(bytes[8..16].try_into().unwrap()),