Skip corrupted WAL records during recovery instead of failing startup

src/wal.rs

@@ -54,23 +54,41 @@ struct WalState {
     flushed: usize,
 }
 
-fn parse_entries(data: &[u8]) -> std::io::Result<Vec<(String, Vec<u8>)>> {
+/// 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<u8>)> {
     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<u8>)>)> {
         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,

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<u8>) -> Result<(), StorageError> {
         Ok(())
     }
 
-    async fn get(&self, _path: &str) -> Result<Vec<u8>, StorageError> {
-        // return invalid WAL data to trigger a parse error
-        Ok(b"bad\t@@\n".to_vec())
+    async fn get(&self, path: &str) -> Result<Vec<u8>, 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<dyn Storage> = 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<dyn Storage> = 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);
 }

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<dyn Storage> = 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<dyn Storage> = 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);
+}