From 3994391722de2fe7409864d113eb868b5ecff3ee Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 9 Jun 2026 18:59:01 +0000 Subject: [PATCH] Skip corrupted WAL records during recovery instead of failing startup A single damaged byte anywhere in the commit log (torn write, bad base64, invalid UTF-8 key) made parse_entries return an error, which Database::new propagated, leaving the node unable to start at all. Recovery now drops malformed records with a warning and replays the rest, matching the usual LSM commit-log recovery contract: a torn tail must not take down the node or discard the valid prefix of the log. https://claude.ai/code/session_01CJXWjB9ERGgV6B9mRqrdM9 --- src/wal.rs | 42 +++++++++++++++++++++++++++----------- tests/wal_error_test.rs | 35 +++++++++++++++++++++++-------- tests/wal_recovery_test.rs | 21 +++++++++++++++++++ 3 files changed, 77 insertions(+), 21 deletions(-) diff --git a/src/wal.rs b/src/wal.rs index 51576e1..07f1bf7 100644 --- a/src/wal.rs +++ b/src/wal.rs @@ -54,23 +54,41 @@ struct WalState { flushed: usize, } -fn parse_entries(data: &[u8]) -> std::io::Result)>> { +/// Parse WAL records, skipping any that are corrupted. +/// +/// A torn write or damaged byte must not prevent the node from starting, so +/// malformed records (bad UTF-8 keys, invalid base64 values, or missing +/// separators) are logged and dropped while the remaining entries are +/// replayed. +fn parse_entries(data: &[u8]) -> Vec<(String, Vec)> { let mut res = Vec::new(); + let mut skipped = 0usize; for line in data.split(|b| *b == b'\n') { if line.is_empty() { continue; } - if let Some(pos) = line.iter().position(|b| *b == b'\t') { - let key = std::str::from_utf8(&line[..pos]) - .map_err(std::io::Error::other)? - .to_string(); - let val = STANDARD - .decode(&line[pos + 1..]) - .map_err(std::io::Error::other)?; - res.push((key, val)); - } + let Some(pos) = line.iter().position(|b| *b == b'\t') else { + skipped += 1; + continue; + }; + let Ok(key) = std::str::from_utf8(&line[..pos]) else { + skipped += 1; + continue; + }; + let Ok(val) = STANDARD.decode(&line[pos + 1..]) else { + skipped += 1; + continue; + }; + res.push((key.to_string(), val)); + } + if skipped > 0 { + tracing::warn!( + skipped, + recovered = res.len(), + "skipped corrupted WAL records during recovery" + ); } - Ok(res) + res } fn map_err(e: StorageError) -> std::io::Error { @@ -177,7 +195,7 @@ impl Wal { ) -> std::io::Result<(Self, Vec<(String, Vec)>)> { let path = path.into(); let buf = storage.get(&path).await.unwrap_or_default(); - let entries = parse_entries(&buf)?; + let entries = parse_entries(&buf); let initial_len = buf.len(); let inner = Arc::new(WalInner { path, diff --git a/tests/wal_error_test.rs b/tests/wal_error_test.rs index c82fede..3f98713 100644 --- a/tests/wal_error_test.rs +++ b/tests/wal_error_test.rs @@ -6,17 +6,28 @@ use cass::{ storage::{Storage, StorageError}, }; -struct BadStorage; +/// Storage backend whose WAL contains a mix of corrupted and valid records. +struct CorruptWalStorage; #[async_trait] -impl Storage for BadStorage { +impl Storage for CorruptWalStorage { async fn put(&self, _path: &str, _data: Vec) -> Result<(), StorageError> { Ok(()) } - async fn get(&self, _path: &str) -> Result, StorageError> { - // return invalid WAL data to trigger a parse error - Ok(b"bad\t@@\n".to_vec()) + async fn get(&self, path: &str) -> Result, StorageError> { + if path == "wal.log" { + // invalid base64, missing separator, invalid UTF-8 key, then a + // valid record ("good" -> base64("v1")) + let mut buf = Vec::new(); + buf.extend_from_slice(b"bad\t@@\n"); + buf.extend_from_slice(b"no-separator\n"); + buf.extend_from_slice(b"\xff\xfe\tdjE=\n"); + buf.extend_from_slice(b"good\tdjE=\n"); + Ok(buf) + } else { + Err(StorageError::Io(std::io::Error::other("missing"))) + } } async fn append(&self, _path: &str, _data: &[u8]) -> Result<(), StorageError> { @@ -29,8 +40,14 @@ impl Storage for BadStorage { } #[tokio::test] -async fn database_new_propagates_wal_error() { - let storage: Arc = Arc::new(BadStorage); - let res = Database::new(storage, "wal.log").await; - assert!(res.is_err()); +async fn database_recovers_past_corrupted_wal_records() { + let storage: Arc = Arc::new(CorruptWalStorage); + let db = Database::new(storage, "wal.log") + .await + .expect("corrupted WAL records must not prevent startup"); + // The valid record after the corrupted ones is still replayed. + assert_eq!(db.get("good").await, Some(b"v1".to_vec())); + // The corrupted records are dropped rather than partially applied. + assert_eq!(db.get("bad").await, None); + assert_eq!(db.get("no-separator").await, None); } diff --git a/tests/wal_recovery_test.rs b/tests/wal_recovery_test.rs index e4f36a9..64d691c 100644 --- a/tests/wal_recovery_test.rs +++ b/tests/wal_recovery_test.rs @@ -20,3 +20,24 @@ async fn wal_recovery_after_restart() { let v1 = db.get("k1").await.map(|b| b[8..].to_vec()); assert_eq!(v1, Some(b"v1".to_vec())); } + +#[tokio::test] +async fn wal_recovery_survives_torn_tail() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().to_path_buf(); + let wal = "wal.log"; + { + let storage: Arc = Arc::new(LocalStorage::new(&path)); + let db = Database::new(storage.clone(), wal).await.unwrap(); + db.insert("k1".to_string(), b"v1".to_vec()).await.unwrap(); + db.sync_wal().await.unwrap(); + // Simulate a crash mid-append: a partial record with truncated + // base64 and no trailing newline. + storage.append(wal, b"k2\tdjE").await.unwrap(); + } + let storage: Arc = Arc::new(LocalStorage::new(&path)); + let db = Database::new(storage, wal).await.unwrap(); + let v1 = db.get("k1").await.map(|b| b[8..].to_vec()); + assert_eq!(v1, Some(b"v1".to_vec())); + assert_eq!(db.get("k2").await, None); +}