From 93c0301c3e58b6a3bf888a3af6e9a5a9bdfa6464 Mon Sep 17 00:00:00 2001 From: Beinan Date: Tue, 20 Oct 2026 08:45:44 +0000 Subject: [PATCH] feat(wal): add bounded durable client state with group commit --- .github/workflows/rust-test.yml | 4 + crates/lance-context-wal/Cargo.toml | 1 + crates/lance-context-wal/README.md | 11 + .../lance-context-wal/src/client_journal.rs | 418 ++++++++++++++++++ .../src/client_journal/tests.rs | 292 ++++++++++++ .../src/client_journal/tests/rewrite.rs | 134 ++++++ crates/lance-context-wal/src/lib.rs | 4 + docs/design/client-journal.md | 89 ++++ 8 files changed, 953 insertions(+) create mode 100644 crates/lance-context-wal/src/client_journal.rs create mode 100644 crates/lance-context-wal/src/client_journal/tests.rs create mode 100644 crates/lance-context-wal/src/client_journal/tests/rewrite.rs create mode 100644 docs/design/client-journal.md diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index 04da6cf0..eeb528c0 100644 --- a/.github/workflows/rust-test.yml +++ b/.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: Client journal durability and process recovery + run: | + cargo clippy -p lance-context-wal --features client-journal --all-targets -- -D warnings + cargo test -p lance-context-wal --features client-journal --lib client_journal - name: Replicated WAL quorum and process recovery run: | cargo clippy -p lance-context-wal --features replication --all-targets -- -D warnings diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 21333b55..67084053 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] +client-journal = ["dep:anyhow", "dep:raft-engine", "dep:bytes", "dep:serde", "dep:serde_json", "dep:tokio", "tokio/sync"] 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"] diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index 26a3c6de..1615289b 100644 --- a/crates/lance-context-wal/README.md +++ b/crates/lance-context-wal/README.md @@ -7,6 +7,17 @@ Implementation and validation are tracked in [#378](https://github.com/lance-format/lance-context/issues/378). Related PRs should include `Refs #378`; individual steps should leave the tracking issue open. +The optional `client-journal` feature stores recoverable client state in one +local engine and shares durable barriers across writers. It returns latest +values for a fixed set of keys, with immutable scope/settings binding, bounded +admission, reclamation and a shared failure fence. Callers still own producer +ordering, exact retry identity and recovery validation; a local completion does +not establish a remote quorum ACK. See [the API and recovery contract](../../docs/design/client-journal.md). + +```bash +cargo test -p lance-context-wal --features client-journal --lib client_journal +``` + The independent `contract` feature defines ingestion routing keys, the schema fingerprint, immutable-ID policy and record byte/row limits. Validation checks the representation after decoding; callers must durably bind and compare the diff --git a/crates/lance-context-wal/src/client_journal.rs b/crates/lance-context-wal/src/client_journal.rs new file mode 100644 index 00000000..ac5293e0 --- /dev/null +++ b/crates/lance-context-wal/src/client_journal.rs @@ -0,0 +1,418 @@ +//! Local, single-owner recoverable client state with a shared durable barrier. +//! This is not a replicated WAL or an ownership allocator. Payload parts stay +//! shared until the disk worker encodes one owned engine value. Engine encoding, +//! checksums and recovery still copy; this does not promise zero-copy storage. +use anyhow::{ensure, Context, Result}; +use bytes::Bytes; +use raft_engine::{env::FileSystem, Engine, LogBatch, ReadableSize}; +use serde::{Deserialize, Serialize}; +use std::path::Path; +use std::sync::{ + atomic::{AtomicBool, AtomicU64, Ordering}, + Arc, +}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; +use tokio::sync::{mpsc, oneshot, OwnedSemaphorePermit, Semaphore}; + +const GROUP: u64 = 1; +const OVERHEAD: usize = 128; +const BINDING: &[u8] = b"client-journal-v1"; + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Settings { + pub linger_ms: u64, + pub batch_bytes: u32, + pub queue_bytes: u32, + pub state_bytes: u32, + pub file_bytes: u32, + pub log_bytes: u64, +} +impl Default for Settings { + fn default() -> Self { + Self { + linger_ms: 1, + batch_bytes: 16 << 20, + queue_bytes: 32 << 20, + state_bytes: 128 << 20, + file_bytes: 8 << 20, + log_bytes: 512 << 20, + } + } +} +impl Settings { + pub fn validate(&self) -> Result<()> { + ensure!(self.linger_ms <= 25, "journal linger exceeds 25 ms"); + ensure!( + self.batch_bytes > OVERHEAD as u32 + && self.batch_bytes <= self.queue_bytes + && self.queue_bytes <= 256 << 20 + && self.state_bytes > 0 + && self.state_bytes <= 256 << 20 + && self.file_bytes >= 64 << 10 + && self.file_bytes <= 64 << 20, + "invalid journal memory/segment bounds" + ); + ensure!( + self.log_bytes + >= self.reserve() + self.state_bytes as u64 + 2 * self.batch_bytes as u64, + "journal disk budget lacks rewrite and append headroom" + ); + Ok(()) + } + fn reserve(&self) -> u64 { + // Rewrite live values plus two segment tails. This is conservative + // admission headroom, not a filesystem quota or power-loss guarantee. + 2 * self.state_bytes as u64 + 4 * self.file_bytes as u64 + } + fn engine(&self, path: &Path) -> raft_engine::Config { + raft_engine::Config { + dir: path.to_string_lossy().into_owned(), + target_file_size: ReadableSize(self.file_bytes as u64), + purge_threshold: ReadableSize(2 * self.file_bytes as u64), + purge_rewrite_threshold: Some(ReadableSize(self.file_bytes as u64)), + batch_compression_threshold: ReadableSize(0), + enable_log_recycle: false, + ..Default::default() + } + } +} + +#[derive(Default)] +struct Counters { + groups: AtomicU64, + requests: AtomicU64, + bytes: AtomicU64, + max_group_requests: AtomicU64, + write_ns: AtomicU64, + maintenance_calls: AtomicU64, + maintenance_ns: AtomicU64, + queue_ns: AtomicU64, + fenced: AtomicBool, +} +#[derive(Debug, Serialize)] +pub struct Statistics { + pub groups: u64, + pub requests: u64, + pub bytes: u64, + pub max_group_requests: u64, + pub write_seconds: f64, + pub maintenance_calls: u64, + pub maintenance_seconds: f64, + pub queue_seconds: f64, + pub fenced: bool, +} +fn nanos(start: Instant) -> u64 { + start.elapsed().as_nanos().min(u64::MAX as u128) as u64 +} +struct Request { + key: u32, + parts: [Bytes; 2], + _permit: OwnedSemaphorePermit, + reply: oneshot::Sender>, +} +impl Request { + fn bytes(&self) -> usize { + self.parts.iter().map(Bytes::len).sum::() + OVERHEAD + } +} +struct Inner { + sender: Option>, + thread: Option>, + queue: Arc, + counters: Arc, + batch_bytes: usize, + keys: u32, + // Keep a failed engine alive until every caller has observed the fence and + // the owner drops its handles. Never destroy it during mutation unwinding. + engine_guard: Option>, +} +impl Drop for Inner { + fn drop(&mut self) { + self.sender.take(); + // Join before releasing caller ownership. A cancelled put can already + // be durable; its original intent remains the only safe retry identity. + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + self.engine_guard.take(); + } +} +#[derive(Clone)] +pub struct Journal(Arc); + +impl Journal { + /// Open under the caller's single-owner lock. The engine also locks its own + /// directory. Recovery returns latest values only; callers must validate + /// their complete protocol state before issuing any network requests. + pub fn open( + path: &Path, + scope: &[u8], + keys: u32, + settings: Settings, + ) -> Result<(Self, Vec>)> { + settings.validate()?; + // Require the parent to exist: syncing only the immediate parent cannot + // make a recursively created ancestor chain durable. A bare relative + // directory name has an empty parent, meaning the current directory. + let parent = path.parent().context("journal parent missing")?; + let parent = if parent.as_os_str().is_empty() { + Path::new(".") + } else { + parent + }; + let parent = std::fs::File::open(parent).context("journal parent must already exist")?; + let engine = Engine::open(settings.engine(path))?; + // Persist creation of the engine directory in its existing parent too. + std::fs::File::open(path)?.sync_all()?; + parent.sync_all()?; + Self::with_engine(engine, scope, keys, settings) + } + + fn with_engine( + engine: Engine, + scope: &[u8], + keys: u32, + settings: Settings, + ) -> Result<(Self, Vec>)> { + settings.validate()?; + ensure!( + (1..=16384).contains(&keys) && scope.len() <= 8192, + "invalid journal binding" + ); + let binding = serde_json::to_vec(&(1u32, scope, keys, &settings))?; + if let Some(saved) = engine.get(0, BINDING) { + ensure!(saved == binding, "client journal scope/settings mismatch"); + } else { + ensure!(engine.is_empty(), "unbound nonempty client journal"); + let mut batch = LogBatch::default(); + batch.put(0, BINDING.to_vec(), binding)?; + engine.write(&mut batch, true)?; + } + let recovered: Vec<_> = (0..keys) + .map(|key| engine.get(GROUP, &key.to_be_bytes()).map(Bytes::from)) + .collect(); + let lengths: Vec<_> = recovered + .iter() + .map(|v| v.as_ref().map_or(0, Bytes::len)) + .collect(); + ensure!( + lengths.iter().sum::() <= settings.state_bytes as usize, + "recovered client state exceeds bound" + ); + let (sender, receiver) = mpsc::channel(1024); + let counters = Arc::new(Counters::default()); + let queue = Arc::new(Semaphore::new(settings.queue_bytes as usize)); + let worker_counters = counters.clone(); + let worker_queue = queue.clone(); + let batch_bytes = settings.batch_bytes as usize; + let engine = Arc::new(engine); + let engine_guard = engine.clone(); + let thread = std::thread::Builder::new() + .name("client-journal".into()) + .spawn(move || { + // Includes unwinding panics from filesystem hooks. No pending sender + // can mistake an abandoned reply for durable completion. + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + worker(&engine, receiver, lengths, settings, &worker_counters) + })); + if !matches!(result, Ok(Ok(()))) { + worker_counters.fenced.store(true, Ordering::Release); + } + worker_queue.close(); + })?; + Ok(( + Self(Arc::new(Inner { + sender: Some(sender), + thread: Some(thread), + queue, + counters, + batch_bytes, + keys, + engine_guard: Some(Box::new(engine_guard)), + })), + recovered, + )) + } + + pub async fn put(&self, key: u32, prefix: Bytes, body: Bytes) -> Result<()> { + let size = prefix + .len() + .checked_add(body.len()) + .and_then(|n| n.checked_add(OVERHEAD)) + .context("journal value overflow")?; + ensure!( + key < self.0.keys && size <= self.0.batch_bytes, + "journal key/value exceeds bounds" + ); + ensure!( + !self.0.counters.fenced.load(Ordering::Acquire), + "client journal fenced" + ); + let start = Instant::now(); + let permit = self + .0 + .queue + .clone() + .acquire_many_owned(size as u32) + .await + .context("client journal unavailable")?; + let (reply, receive) = oneshot::channel(); + self.0 + .sender + .as_ref() + .unwrap() + .send(Request { + key, + parts: [prefix, body], + _permit: permit, + reply, + }) + .await + .context("client journal worker stopped")?; + self.0 + .counters + .queue_ns + .fetch_add(nanos(start), Ordering::Relaxed); + receive + .await + .context("client journal outcome unknown; reopen before recovery")? + .map_err(anyhow::Error::msg) + } + + pub fn statistics(&self) -> Statistics { + let c = &self.0.counters; + let get = |a: &AtomicU64| a.load(Ordering::Relaxed); + Statistics { + groups: get(&c.groups), + requests: get(&c.requests), + bytes: get(&c.bytes), + max_group_requests: get(&c.max_group_requests), + write_seconds: get(&c.write_ns) as f64 / 1e9, + maintenance_calls: get(&c.maintenance_calls), + maintenance_seconds: get(&c.maintenance_ns) as f64 / 1e9, + queue_seconds: get(&c.queue_ns) as f64 / 1e9, + fenced: c.fenced.load(Ordering::Acquire), + } + } +} + +fn footprint(path: &Path) -> Result { + let mut bytes = 0u64; + for file in std::fs::read_dir(path)? { + let metadata = file?.metadata()?; + ensure!(metadata.is_file(), "unexpected journal directory entry"); + #[cfg(unix)] + let allocated = { + use std::os::unix::fs::MetadataExt; + metadata.blocks().saturating_mul(512) + }; + #[cfg(not(unix))] + let allocated = metadata.len(); + bytes = bytes + .checked_add(metadata.len().max(allocated)) + .context("disk accounting overflow")?; + } + Ok(bytes) +} + +fn worker( + engine: &Engine, + mut receive: mpsc::Receiver, + mut lengths: Vec, + settings: Settings, + counters: &Counters, +) -> Result<()> { + let mut deferred = None; + while let Some(first) = deferred.take().or_else(|| receive.blocking_recv()) { + let mut bytes = first.bytes(); + let mut requests = vec![first]; + // Bound intentional batching delay; queue pressure can add further wait. + if settings.linger_ms > 0 { + std::thread::sleep(Duration::from_millis(settings.linger_ms)); + } + while requests.len() < 1024 { + let Ok(next) = receive.try_recv() else { + break; + }; + if bytes + next.bytes() > settings.batch_bytes as usize { + deferred = Some(next); + break; + } + bytes += next.bytes(); + requests.push(next); + } + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| -> Result<()> { + let physical = footprint(Path::new(engine.path()))?; + if physical >= 2 * settings.file_bytes as u64 { + let start = Instant::now(); + engine.purge_expired_files()?; + counters + .maintenance_ns + .fetch_add(nanos(start), Ordering::Relaxed); + counters.maintenance_calls.fetch_add(1, Ordering::Relaxed); + } + ensure!( + footprint(Path::new(engine.path()))? + bytes as u64 + <= settings.log_bytes - settings.reserve(), + "client journal disk admission closed" + ); + let mut total: usize = lengths.iter().sum(); + let mut batch = LogBatch::default(); + for request in &requests { + let length = request.bytes() - OVERHEAD; + let old = &mut lengths[request.key as usize]; + total = total - *old + length; + ensure!( + total <= settings.state_bytes as usize, + "client journal state budget exceeded" + ); + *old = length; + let mut value = Vec::with_capacity(length); + for part in &request.parts { + value.extend_from_slice(part); + } + batch.put(GROUP, request.key.to_be_bytes().to_vec(), value)?; + } + let start = Instant::now(); + // Never use Engine::sync(): pinned 0.4.2's empty sync is a no-op. + engine.write(&mut batch, true)?; + counters.write_ns.fetch_add(nanos(start), Ordering::Relaxed); + Ok(()) + })) + .unwrap_or_else(|_| { + Err(anyhow::anyhow!( + "client journal mutation panicked; reopen required" + )) + }); + if result.is_ok() { + counters.groups.fetch_add(1, Ordering::Relaxed); + counters + .requests + .fetch_add(requests.len() as u64, Ordering::Relaxed); + counters.bytes.fetch_add( + (bytes - OVERHEAD * requests.len()) as u64, + Ordering::Relaxed, + ); + counters + .max_group_requests + .fetch_max(requests.len() as u64, Ordering::Relaxed); + } else { + counters.fenced.store(true, Ordering::Release); + } + for request in requests { + let _ = request.reply.send( + result + .as_ref() + .map(|_| ()) + .map_err(|e| format!("client journal fenced: {e:#}")), + ); + } + result?; + } + Ok(()) +} + +#[cfg(test)] +mod tests; diff --git a/crates/lance-context-wal/src/client_journal/tests.rs b/crates/lance-context-wal/src/client_journal/tests.rs new file mode 100644 index 00000000..9bda6b92 --- /dev/null +++ b/crates/lance-context-wal/src/client_journal/tests.rs @@ -0,0 +1,292 @@ +use super::*; +use raft_engine::env::{DefaultFileSystem, Handle, Permission}; +use std::io; +use std::sync::{mpsc as blocking, Mutex}; +mod rewrite; + +fn settings() -> Settings { + Settings { + linger_ms: 2, + batch_bytes: 1 << 20, + queue_bytes: 2 << 20, + state_bytes: 4 << 20, + file_bytes: 1 << 20, + log_bytes: 32 << 20, + } +} + +#[test] +fn disk_admission_keeps_room_for_live_state_and_rewrite() { + let mut config = settings(); + config.log_bytes = config.reserve() + 2 * config.batch_bytes as u64; + assert!(config.validate().is_err()); + config.log_bytes += config.state_bytes as u64; + assert!(config.validate().is_ok()); +} + +#[test] +fn relative_directory_reopens_and_missing_parent_does_not_create_ancestors() -> Result<()> { + // Avoid changing the process-wide current directory while tests run. + let dir = tempfile::Builder::new() + .prefix(".client-journal-relative-") + .tempdir_in(".")?; + let relative = Path::new(dir.path().file_name().unwrap()); + let (journal, _) = Journal::open(relative, b"relative", 1, settings())?; + drop(journal); + let (journal, _) = Journal::open(relative, b"relative", 1, settings())?; + drop(journal); + + let root = tempfile::tempdir()?; + let missing = root.path().join("missing"); + assert!(Journal::open(&missing.join("journal"), b"scope", 1, settings()).is_err()); + assert!(!missing.exists()); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_replacements_reclaim_and_reopen_exact_latest_values() -> Result<()> { + let dir = tempfile::tempdir()?; + let config = settings(); + let (journal, recovered) = Journal::open(dir.path(), b"scope", 4, config.clone())?; + assert!(recovered.iter().all(Option::is_none)); + let mut jobs = tokio::task::JoinSet::new(); + for key in 0..4 { + let writer = journal.clone(); + jobs.spawn(async move { + for round in 0..128u8 { + writer + .put( + key, + Bytes::from(vec![key as u8, round]), + Bytes::from(vec![round; 65536]), + ) + .await?; + } + Ok::<_, anyhow::Error>(()) + }); + } + while let Some(result) = jobs.join_next().await { + result??; + } + let stats = journal.statistics(); + assert_eq!(stats.requests, 512); + assert!(stats.groups < 512 && stats.max_group_requests > 1); + assert!(stats.maintenance_calls > 0 && !stats.fenced); + assert!(footprint(dir.path())? < config.log_bytes - config.reserve()); + drop(journal); + let (journal, values) = Journal::open(dir.path(), b"scope", 4, config.clone())?; + for (key, value) in values.into_iter().enumerate() { + let value = value.unwrap(); + assert_eq!(&value[..2], &[key as u8, 127]); + assert_eq!(&value[2..], vec![127; 65536]); + } + assert!(Journal::open(dir.path(), b"scope", 4, config.clone()).is_err()); + drop(journal); + assert!(Journal::open(dir.path(), b"other", 4, config.clone()).is_err()); + assert!(Journal::open(dir.path(), b"scope", 3, config).is_err()); + Ok(()) +} + +type RealHandle = ::Handle; +#[derive(Default)] +struct Control { + fail: AtomicBool, + calls: AtomicU64, + entered: Mutex>>, + release: Mutex>>, + rewrite_pause: Mutex>, +} +struct FaultHandle { + inner: Arc, + control: Arc, + rewrite: bool, +} +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<()> { + if self.rewrite { + let pause = self.control.rewrite_pause.lock().unwrap().take(); + if let Some((marker, after_sync)) = pause { + if after_sync { + self.inner.sync()?; + } + std::fs::write(marker, b"actual rewrite sync reached")?; + loop { + std::thread::park(); + } + } + } + self.control.calls.fetch_add(1, Ordering::Relaxed); + if let Some(tx) = self.control.entered.lock().unwrap().take() { + tx.send(()).unwrap(); + } + if let Some(rx) = self.control.release.lock().unwrap().take() { + rx.recv_timeout(Duration::from_secs(10)).unwrap(); + } + if self.control.fail.load(Ordering::Acquire) { + return Err(io::Error::other("injected client sync failure")); + } + self.inner.sync() + } +} +#[derive(Default)] +struct FaultFs { + control: Arc, +} +impl FileSystem for FaultFs { + type Handle = FaultHandle; + type Reader = ::Reader; + type Writer = ::Writer; + fn create>(&self, path: P) -> io::Result { + let rewrite = path.as_ref().extension().is_some_and(|e| e == "rewrite"); + Ok(FaultHandle { + inner: Arc::new(DefaultFileSystem.create(path)?), + control: self.control.clone(), + rewrite, + }) + } + fn open>(&self, path: P, permission: Permission) -> io::Result { + let rewrite = path.as_ref().extension().is_some_and(|e| e == "rewrite"); + Ok(FaultHandle { + inner: Arc::new(DefaultFileSystem.open(path, permission)?), + control: self.control.clone(), + rewrite, + }) + } + fn delete>(&self, path: P) -> io::Result<()> { + DefaultFileSystem.delete(path) + } + fn rename>(&self, src: P, dst: P) -> io::Result<()> { + DefaultFileSystem.rename(src, dst) + } + fn new_reader(&self, h: Arc) -> io::Result { + DefaultFileSystem.new_reader(h.inner.clone()) + } + fn new_writer(&self, h: Arc) -> io::Result { + DefaultFileSystem.new_writer(h.inner.clone()) + } +} +type Controlled = ( + Journal, + Arc, + blocking::Receiver<()>, + blocking::SyncSender<()>, +); +fn controlled(dir: &Path) -> Result { + let fs = Arc::new(FaultFs::default()); + let control = fs.control.clone(); + let config = Settings { + linger_ms: 25, + ..settings() + }; + let engine = Engine::open_with(config.engine(dir), fs, vec![])?; + let (journal, _) = Journal::with_engine(engine, b"scope", 2, config)?; + let (entered, rx) = blocking::sync_channel(1); + let (tx, release) = blocking::sync_channel(1); + *control.entered.lock().unwrap() = Some(entered); + *control.release.lock().unwrap() = Some(release); + Ok((journal, control, rx, tx)) +} +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn no_waiter_completes_before_sync_and_failure_fences_all_handles() -> Result<()> { + let dir = tempfile::tempdir()?; + let (journal, control, entered, release) = controlled(dir.path())?; + control.fail.store(true, Ordering::Release); + let mut jobs = tokio::task::JoinSet::new(); + for key in 0..2 { + let j = journal.clone(); + jobs.spawn(async move { + j.put( + key, + Bytes::from_static(b"intent"), + Bytes::from(vec![7; 4096]), + ) + .await + }); + } + tokio::task::spawn_blocking(move || entered.recv_timeout(Duration::from_secs(10))).await??; + assert!( + jobs.try_join_next().is_none(), + "success before actual filesystem sync" + ); + release.send(())?; + while let Some(result) = jobs.join_next().await { + assert!(result?.is_err()); + } + let calls = control.calls.load(Ordering::Relaxed); + assert!(journal.statistics().fenced); + assert!(journal + .put(0, Bytes::new(), Bytes::from_static(b"next")) + .await + .is_err()); + assert_eq!(control.calls.load(Ordering::Relaxed), calls); + assert_eq!(journal.statistics().requests, 0); + // Pinned engine destruction also syncs. Clear only the test filesystem + // injection before dropping the permanently fenced journal, never reuse it. + control.fail.store(false, Ordering::Release); + drop(journal); + Ok(()) +} +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn cancellation_keeps_the_original_durable_outcome_recoverable() -> Result<()> { + let dir = tempfile::tempdir()?; + let (journal, _control, entered, release) = controlled(dir.path())?; + let j = journal.clone(); + let task = tokio::spawn(async move { + j.put( + 0, + Bytes::from_static(b"intent"), + Bytes::from_static(b"exact-body"), + ) + .await + }); + tokio::task::spawn_blocking(move || entered.recv_timeout(Duration::from_secs(10))).await??; + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + release.send(())?; + journal + .put(1, Bytes::from_static(b"barrier"), Bytes::new()) + .await?; + drop(journal); + let (_reopened, values) = Journal::open( + dir.path(), + b"scope", + 2, + Settings { + linger_ms: 25, + ..settings() + }, + )?; + assert_eq!(values[0].as_deref(), Some(b"intentexact-body".as_slice())); + Ok(()) +} + +#[tokio::test] +async fn state_admission_and_oversized_values_do_not_acknowledge() -> Result<()> { + let dir = tempfile::tempdir()?; + let config = Settings { + state_bytes: 1024, + ..settings() + }; + let (journal, _) = Journal::open(dir.path(), b"scope", 2, config)?; + assert!(journal + .put(0, Bytes::new(), Bytes::from(vec![3; 1 << 20])) + .await + .is_err()); + assert!(journal.put(2, Bytes::new(), Bytes::new()).await.is_err()); + journal + .put(0, Bytes::new(), Bytes::from(vec![1; 800])) + .await?; + assert!(journal + .put(1, Bytes::new(), Bytes::from(vec![2; 800])) + .await + .is_err()); + assert!(journal.statistics().fenced); + assert_eq!(journal.statistics().requests, 1); + Ok(()) +} diff --git a/crates/lance-context-wal/src/client_journal/tests/rewrite.rs b/crates/lance-context-wal/src/client_journal/tests/rewrite.rs new file mode 100644 index 00000000..4eedf735 --- /dev/null +++ b/crates/lance-context-wal/src/client_journal/tests/rewrite.rs @@ -0,0 +1,134 @@ +//! Kill a real journal process at the actual Rewrite queue's sync boundary. +use super::*; +use std::io::Write; +use std::process::{Child, Command, Stdio}; + +fn config() -> Settings { + Settings { + file_bytes: 64 << 10, + ..settings() + } +} +fn hot(round: u64) -> Bytes { + let mut bytes = round.to_le_bytes().to_vec(); + bytes.extend(vec![(round % 251) as u8; 32768]); + Bytes::from(bytes) +} +fn record_ack(root: &Path, round: u64) -> Result<()> { + let temporary = root.join("ack.new"); + let mut file = std::fs::File::create(&temporary)?; + file.write_all(&round.to_le_bytes())?; + file.sync_all()?; + std::fs::rename(temporary, root.join("ack"))?; + std::fs::File::open(root)?.sync_all()?; + Ok(()) +} + +#[test] +#[ignore = "subprocess fixture invoked by the real SIGKILL test"] +fn rewrite_child() -> Result<()> { + let root = std::path::PathBuf::from(std::env::var("CONTEXT_CLIENT_JOURNAL_REWRITE_ROOT")?); + let after = std::env::var("CONTEXT_CLIENT_JOURNAL_REWRITE_AFTER")? == "1"; + let runtime = tokio::runtime::Runtime::new()?; + runtime.block_on(async { + let filesystem = Arc::new(FaultFs::default()); + let control = filesystem.control.clone(); + let engine = Engine::open_with(config().engine(&root.join("journal")), filesystem, vec![])?; + let (journal, _) = Journal::with_engine(engine, b"rewrite-process", 2, config())?; + journal + .put( + 0, + Bytes::from_static(b"cold"), + Bytes::from(vec![197; 32768]), + ) + .await?; + journal.put(1, Bytes::new(), hot(0)).await?; + record_ack(&root, 0)?; + *control.rewrite_pause.lock().unwrap() = Some((root.join("paused"), after)); + for round in 1..1024 { + journal.put(1, Bytes::new(), hot(round)).await?; + record_ack(&root, round)?; + } + anyhow::bail!("rewrite sync was not exercised") + }) +} +struct Stop(Child); +impl Drop for Stop { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +#[test] +fn sigkill_before_and_after_rewrite_sync_preserves_every_acknowledged_value() -> Result<()> { + for after in [false, true] { + let root = if let Some(base) = std::env::var_os("CONTEXT_CLIENT_JOURNAL_TEST_ARTIFACTS") { + std::fs::create_dir_all(&base)?; + let mut dir = tempfile::Builder::new() + .prefix("journal-rewrite-") + .tempdir_in(base)?; + dir.disable_cleanup(true); + dir + } else { + tempfile::tempdir()? + }; + let mut child = Stop( + Command::new(std::env::current_exe()?) + .args([ + "--ignored", + "--exact", + "client_journal::tests::rewrite::rewrite_child", + "--nocapture", + ]) + .env("CONTEXT_CLIENT_JOURNAL_REWRITE_ROOT", root.path()) + .env( + "CONTEXT_CLIENT_JOURNAL_REWRITE_AFTER", + if after { "1" } else { "0" }, + ) + .stdin(Stdio::null()) + .stdout(std::fs::File::create(root.path().join("stdout"))?) + .stderr(std::fs::File::create(root.path().join("stderr"))?) + .spawn()?, + ); + let until = Instant::now() + Duration::from_secs(20); + while !root.path().join("paused").exists() { + ensure!( + child.0.try_wait()?.is_none(), + "rewrite child exited before its sync boundary: {}", + root.path().display() + ); + ensure!( + Instant::now() < until, + "rewrite child did not reach sync boundary" + ); + std::thread::sleep(Duration::from_millis(10)); + } + child.0.kill()?; + let status = child.0.wait()?; + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + assert_eq!(status.signal(), Some(9)); + } + let bytes = std::fs::read(root.path().join("ack"))?; + let round = u64::from_le_bytes(bytes.try_into().unwrap()); + let (_journal, values) = Journal::open( + &root.path().join("journal"), + b"rewrite-process", + 2, + config(), + )?; + let cold = values[0].as_ref().context("lost acknowledged cold value")?; + assert_eq!(&cold[..4], b"cold"); + assert_eq!(&cold[4..], vec![197; 32768]); + assert_eq!(values[1].as_ref(), Some(&hot(round))); + std::fs::write( + root.path().join("verified.json"), + serde_json::to_vec( + &serde_json::json!({"after_sync":after,"acknowledged_round":round,"signal":9,"full_values_verified":true}), + )?, + )?; + } + Ok(()) +} diff --git a/crates/lance-context-wal/src/lib.rs b/crates/lance-context-wal/src/lib.rs index fe9b8a0d..e5c7b15a 100644 --- a/crates/lance-context-wal/src/lib.rs +++ b/crates/lance-context-wal/src/lib.rs @@ -28,3 +28,7 @@ pub mod ingest_contract; #[cfg(feature = "vectored-io")] pub mod transport_io; + +/// Local recoverable client state with bounded shared durable barriers. +#[cfg(feature = "client-journal")] +pub mod client_journal; diff --git a/docs/design/client-journal.md b/docs/design/client-journal.md new file mode 100644 index 00000000..83e9114b --- /dev/null +++ b/docs/design/client-journal.md @@ -0,0 +1,89 @@ +# Durable client state with group commit + +Tracked by [#378](https://github.com/lance-format/lance-context/issues/378). +The optional `client-journal` feature provides a local, bounded latest-value +journal for a recoverable ingestion client. Independent writers can share one +physical synchronization barrier while retaining exact pending request bytes and +ACK checkpoints. It uses the existing pinned raft-engine dependency. + +This component does not replicate client state, allocate producers, implement an +HTTP SDK or enable the distributed ingestion path. Callers supply the protocol +state, single-owner scope and recovery validation. The remote WAL still needs its +own durable quorum; a local journal completion cannot substitute for that ACK. + +## API contract + +`Journal::open(path, scope, keys, settings)` opens the engine and returns a handle +plus the latest optional value for each fixed key. It binds version, scope, key +count and settings durably; reopening with a different binding fails. The caller +must provide an existing parent directory; a bare relative name uses the current +directory. Open synchronizes the engine directory and that parent, and rejects +missing parents without creating an unsynchronized ancestor chain. The caller +must distinguish first initialization from a missing/lost journal and validate +every recovered state before sending any network operation. The engine locks its +directory, while the caller owns any surrounding state and ownership protocol. + +`put(key, prefix, body).await` persists the concatenated bytes. The two `Bytes` +parts remain shared until the disk worker assembles an engine value. This avoids +caller-side payload concatenation; engine serialization, checksums and recovery +still copy. The latest value replaces the old value for that key. Concurrent +updates of the same key require the caller's ordering discipline. + +A recoverable client normally writes the exact intent before sending, then writes +the ACK checkpoint before advancing the producer sequence. Multiple such writes +may share a disk barrier. This does not make independent network operations +atomic. A cancelled or failed wait can have an unknown durable outcome: recover +and retry the original protocol operation, never invent a new identity because +the wait ended. + +Open performs blocking disk I/O. Dropping the last handle joins the disk worker +and can wait for pending disk work; place these operations appropriately in an +async application. Keep ownership until the last handle has closed. Copying the +directory does not transfer or fence producer ownership in another process. + +## Durability and bounds + +The dedicated worker gathers requests up to a byte/count limit, waits at most the +configured intentional linger (0–25 ms), and writes a nonempty synced engine +batch. Actual completion can take longer due to queueing and storage. The pinned +engine's empty `sync()` is a no-op and is not used for completion. + +Queued request bytes retain semaphore permits through completion. A separate +latest-state limit bounds logical live values. Disk admission measures regular +files and reserves live-state, rewrite and segment headroom. Engine-managed purge +reclaims obsolete segments; the journal never deletes them itself. These are +admission limits, not an OS disk quota or an RSS bound: engine/encoding buffers, +metadata, allocator overhead and filesystem behavior need separate measurement. +Caller-owned buffers waiting to enter admission are also outside this bound; +applications should bound producers before constructing those buffers. + +Mutation errors and unwinding dependency panics permanently fence the journal +before successful completion can escape. All handles must be dropped and recovery +performed on a fresh engine; a fence cannot be reset. The owner retains the engine +until its worker has stopped, avoiding destruction during mutation unwinding. +Dependency destructor failures or panic-abort builds can still terminate the +process. Process restart and storage power-loss guarantees are distinct. + +Statistics report successful groups/requests/bytes, maximum group size, engine +write time, maintenance and admission wait. Write time includes engine encoding, +append and synchronization; admission wait excludes subsequent worker/linger +waits. These counters do not establish end-to-end ingestion throughput. + +## Validation + +Real-file tests cover concurrent durable replacements, repeated reclamation, +complete latest-value recovery, immutable binding and exclusive ownership, +admission limits, cancellation during actual sync, and injected sync failure with +no premature successful waiter. A parent test invokes the ignored Rust child +fixture, kills it immediately before and after Rewrite-queue sync, then reopens +the files and compares every acknowledged value, including cold retained state. + +CI runs the optional feature explicitly. Run it locally with: + +```sh +cargo test -p lance-context-wal --features client-journal --lib client_journal +``` + +These tests qualify local process recovery on retained files, not volume loss, +distributed producer ownership or a production performance result. Protocol-level +ingress integration and performance comparisons remain tracked separately in #378.