claw-store/src/cluster.rs

@@ -25,6 +25,7 @@ pub mod rpc;
 pub mod services;
 pub mod tags;
 pub mod transport;
+pub mod wal;
 
 use crate::config::PeerEntry;
 use anyhow::{bail, Result};

claw-store/src/cluster/wal.rs

@@ -0,0 +1,601 @@
+//! Phase 4c (2026-07-13): Write-Ahead Log for mutating client-mode
+//! ops.
+//!
+//! Roaming/offline clients (per Architecture v2, "Roaming client")
+//! must durably record mutations before attempting network fanout,
+//! so a crashed or offline daemon can replay them at reconnect. This
+//! module implements the primitive: a bounded, checksum-protected
+//! append-only file with monotonically-increasing sequence numbers
+//! and cheap truncation-up-to a durable watermark.
+//!
+//! Wire format is deliberately in-tree — no serde crate, no bincode.
+//! Each record is:
+//!
+//! ```text
+//! seq   : u64 LE            (8 bytes)
+//! len   : u32 LE            (4 bytes)   — payload length
+//! csum  : [u8; 8]           (8 bytes)   — first 8 bytes of BLAKE3(seq || len || payload)
+//! bytes : [u8; len]
+//! ```
+//!
+//! On open, the file is scanned linearly. A short read, wrong length
+//! prefix, or checksum mismatch stops the scan — everything up to
+//! that boundary is kept; the tail is truncated. This models the
+//! standard "crash mid-append" case: whatever fully-fsynced record
+//! we last wrote is the last one we replay.
+//!
+//! No rotation yet — a single segment file per log root. `truncate`
+//! rewrites the tail into place atomically via
+//! `tempfile-in-parent + rename`. That's O(n) in surviving records,
+//! which is fine for the tens-of-thousands range this WAL is scoped
+//! for. Segment rotation is Phase-4d territory.
+//!
+//! No mocks / no stubs / full impl per the delivery constraints.
+
+use anyhow::{bail, Context, Result};
+use std::path::{Path, PathBuf};
+use tokio::fs::OpenOptions;
+use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
+
+/// Fixed on-disk framing overhead per record — `seq(8) + len(4) + csum(8)`.
+pub const RECORD_HEADER_LEN: usize = 20;
+
+/// A durable record replayed from the WAL. `payload` is opaque —
+/// callers own the encoding.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WalRecord {
+    pub seq: u64,
+    pub payload: Vec<u8>,
+}
+
+/// File-backed WAL. Single-writer; `append` takes `&mut self` so
+/// serialization is enforced at the type level. Readers (`iter_from`,
+/// `tail_seq`, `head_seq`) can be called against `&self` and open
+/// their own file handles.
+pub struct WriteAheadLog {
+    root: PathBuf,
+    log_path: PathBuf,
+    /// Highest seq that has been fsynced. Zero when the log is empty.
+    tail_seq: u64,
+    /// Lowest seq still present on disk. Advances after
+    /// `truncate_up_to`. Zero when the log is empty.
+    head_seq: u64,
+    /// Cached file handle for appends. Opened in append-mode so
+    /// offsets always land at the end.
+    writer: tokio::fs::File,
+}
+
+impl WriteAheadLog {
+    /// Open (or create) the WAL rooted at `root`. Scans the segment
+    /// file to recover `head_seq` / `tail_seq`. A partial trailing
+    /// record is truncated in place.
+    pub async fn open(root: impl Into<PathBuf>) -> Result<Self> {
+        let root = root.into();
+        tokio::fs::create_dir_all(&root)
+            .await
+            .with_context(|| format!("creating WAL root {}", root.display()))?;
+        let log_path = root.join("log.bin");
+
+        // Scan (creates an empty file if none exists) to learn head/tail.
+        let (head_seq, tail_seq, valid_len) = scan_segment(&log_path).await?;
+
+        // Truncate any partial trailing record. `set_len` is the
+        // stdlib idiom for exact truncation on both linux and darwin.
+        if let Ok(meta) = tokio::fs::metadata(&log_path).await {
+            if meta.len() != valid_len {
+                let f = OpenOptions::new()
+                    .write(true)
+                    .open(&log_path)
+                    .await
+                    .with_context(|| format!("opening {} for truncation", log_path.display()))?;
+                f.set_len(valid_len)
+                    .await
+                    .with_context(|| format!("truncating {} to {}", log_path.display(), valid_len))?;
+                f.sync_all()
+                    .await
+                    .with_context(|| format!("fsync after truncation of {}", log_path.display()))?;
+            }
+        }
+
+        let writer = OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(&log_path)
+            .await
+            .with_context(|| format!("opening WAL {} for append", log_path.display()))?;
+
+        Ok(Self {
+            root,
+            log_path,
+            head_seq,
+            tail_seq,
+            writer,
+        })
+    }
+
+    /// Root directory this WAL lives in. Handy for callers that
+    /// want to place sidecar files (state, index) next to it.
+    pub fn root(&self) -> &Path {
+        &self.root
+    }
+
+    /// Lowest seq still present. Zero when the log is empty.
+    pub fn head_seq(&self) -> u64 {
+        self.head_seq
+    }
+
+    /// Highest seq present. Zero when the log is empty.
+    pub fn tail_seq(&self) -> u64 {
+        self.tail_seq
+    }
+
+    /// Whether the log has any records.
+    pub fn is_empty(&self) -> bool {
+        self.tail_seq == 0
+    }
+
+    /// Append a payload. Returns the assigned seq (always
+    /// `tail_seq() + 1`). Persisted with `fsync` before returning so
+    /// callers can treat success as durable.
+    pub async fn append(&mut self, payload: &[u8]) -> Result<u64> {
+        if payload.len() > u32::MAX as usize {
+            bail!(
+                "WAL payload too large: {} bytes (max {})",
+                payload.len(),
+                u32::MAX
+            );
+        }
+        let seq = self.tail_seq.checked_add(1).context("WAL seq overflow")?;
+        let mut frame = Vec::with_capacity(RECORD_HEADER_LEN + payload.len());
+        frame.extend_from_slice(&seq.to_le_bytes());
+        frame.extend_from_slice(&(payload.len() as u32).to_le_bytes());
+        frame.extend_from_slice(&checksum(seq, payload));
+        frame.extend_from_slice(payload);
+
+        self.writer
+            .write_all(&frame)
+            .await
+            .with_context(|| format!("appending to WAL {}", self.log_path.display()))?;
+        self.writer
+            .flush()
+            .await
+            .context("flushing WAL append buffer")?;
+        self.writer
+            .sync_all()
+            .await
+            .with_context(|| format!("fsync after WAL append {}", self.log_path.display()))?;
+
+        self.tail_seq = seq;
+        if self.head_seq == 0 {
+            self.head_seq = seq;
+        }
+        Ok(seq)
+    }
+
+    /// Read every record with `seq >= start_seq`. Streams — the
+    /// whole log never lives in memory at once.
+    pub async fn iter_from(&self, start_seq: u64) -> Result<Vec<WalRecord>> {
+        let mut out = Vec::new();
+        let mut file = match tokio::fs::File::open(&self.log_path).await {
+            Ok(f) => f,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
+            Err(e) => return Err(anyhow::Error::from(e)),
+        };
+        loop {
+            match read_next_record(&mut file).await? {
+                Some(rec) if rec.seq >= start_seq => out.push(rec),
+                Some(_) => {} // record older than the requested start.
+                None => break,
+            }
+        }
+        Ok(out)
+    }
+
+    /// Drop every record with `seq <= watermark`. Rewrites the
+    /// segment atomically via `tempfile-in-parent + rename`. No-op
+    /// when the log is already past the watermark or empty.
+    pub async fn truncate_up_to(&mut self, watermark: u64) -> Result<()> {
+        if self.tail_seq == 0 || watermark < self.head_seq {
+            return Ok(());
+        }
+        if watermark >= self.tail_seq {
+            // Everything is gone; reset head/tail and truncate the file.
+            drop(std::mem::replace(
+                &mut self.writer,
+                OpenOptions::new()
+                    .write(true)
+                    .create(true)
+                    .truncate(true)
+                    .open(&self.log_path)
+                    .await?,
+            ));
+            self.writer
+                .sync_all()
+                .await
+                .with_context(|| format!("fsync after full truncate of {}", self.log_path.display()))?;
+            // Reopen in append mode so `append` continues to land at
+            // the end even on an OS that treats trunc-create as write mode.
+            self.writer = OpenOptions::new()
+                .append(true)
+                .create(true)
+                .open(&self.log_path)
+                .await?;
+            self.head_seq = 0;
+            self.tail_seq = 0;
+            return Ok(());
+        }
+
+        // Otherwise: rewrite records with seq > watermark into a
+        // sibling tempfile, fsync, rename over the original.
+        let dir = self
+            .log_path
+            .parent()
+            .with_context(|| format!("WAL log path has no parent: {}", self.log_path.display()))?;
+        let tmp = tempfile::NamedTempFile::new_in(dir)
+            .with_context(|| format!("creating tempfile in {}", dir.display()))?;
+        let tmp_path = tmp.path().to_path_buf();
+        // Close the handle immediately — we'll reopen with tokio.
+        drop(tmp);
+        {
+            let mut writer = OpenOptions::new()
+                .create(true)
+                .write(true)
+                .truncate(true)
+                .open(&tmp_path)
+                .await
+                .with_context(|| format!("opening WAL tempfile {}", tmp_path.display()))?;
+            let mut reader = tokio::fs::File::open(&self.log_path).await?;
+            let mut new_head = 0u64;
+            loop {
+                match read_next_record(&mut reader).await? {
+                    Some(rec) if rec.seq > watermark => {
+                        let mut frame =
+                            Vec::with_capacity(RECORD_HEADER_LEN + rec.payload.len());
+                        frame.extend_from_slice(&rec.seq.to_le_bytes());
+                        frame.extend_from_slice(&(rec.payload.len() as u32).to_le_bytes());
+                        frame.extend_from_slice(&checksum(rec.seq, &rec.payload));
+                        frame.extend_from_slice(&rec.payload);
+                        writer.write_all(&frame).await?;
+                        if new_head == 0 {
+                            new_head = rec.seq;
+                        }
+                    }
+                    Some(_) => {}
+                    None => break,
+                }
+            }
+            writer.flush().await?;
+            writer.sync_all().await?;
+            self.head_seq = new_head;
+        }
+
+        tokio::fs::rename(&tmp_path, &self.log_path)
+            .await
+            .with_context(|| {
+                format!(
+                    "renaming {} → {}",
+                    tmp_path.display(),
+                    self.log_path.display()
+                )
+            })?;
+
+        // Fsync the parent so the rename is durable.
+        let dir_handle = tokio::fs::File::open(dir)
+            .await
+            .with_context(|| format!("opening WAL parent dir {}", dir.display()))?;
+        dir_handle
+            .sync_all()
+            .await
+            .with_context(|| format!("fsync WAL parent dir {}", dir.display()))?;
+
+        // Reopen the append handle so writes land at the new EOF.
+        self.writer = OpenOptions::new()
+            .append(true)
+            .create(true)
+            .open(&self.log_path)
+            .await?;
+        Ok(())
+    }
+}
+
+/// First 8 bytes of BLAKE3 over `seq || len || payload`. Keyless
+/// checksum: we're not preventing tampering, we're catching torn
+/// writes and disk corruption.
+fn checksum(seq: u64, payload: &[u8]) -> [u8; 8] {
+    let mut hasher = blake3::Hasher::new();
+    hasher.update(&seq.to_le_bytes());
+    hasher.update(&(payload.len() as u32).to_le_bytes());
+    hasher.update(payload);
+    let mut out = [0u8; 8];
+    out.copy_from_slice(&hasher.finalize().as_bytes()[..8]);
+    out
+}
+
+/// Read the next record from a positioned file. `Ok(None)` at EOF or
+/// when the trailing bytes are shorter than a full record — either
+/// case is treated as "clean tail, nothing to see here". A checksum
+/// mismatch on a full-length record is fatal.
+async fn read_next_record(file: &mut tokio::fs::File) -> Result<Option<WalRecord>> {
+    let mut header = [0u8; RECORD_HEADER_LEN];
+    let start = file.stream_position().await?;
+    match file.read_exact(&mut header).await {
+        Ok(_) => {}
+        Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+            // Partial trailing header → rewind to start so the caller
+            // can size-truncate exactly here.
+            file.seek(std::io::SeekFrom::Start(start)).await?;
+            return Ok(None);
+        }
+        Err(e) => return Err(anyhow::Error::from(e)),
+    }
+    let seq = u64::from_le_bytes(header[..8].try_into().unwrap());
+    let len = u32::from_le_bytes(header[8..12].try_into().unwrap()) as usize;
+    let expected_csum: [u8; 8] = header[12..20].try_into().unwrap();
+
+    let mut payload = vec![0u8; len];
+    if len > 0 {
+        match file.read_exact(&mut payload).await {
+            Ok(_) => {}
+            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                file.seek(std::io::SeekFrom::Start(start)).await?;
+                return Ok(None);
+            }
+            Err(e) => return Err(anyhow::Error::from(e)),
+        }
+    }
+    if checksum(seq, &payload) != expected_csum {
+        bail!(
+            "WAL record checksum mismatch at seq {} (offset {} in log)",
+            seq,
+            start
+        );
+    }
+    Ok(Some(WalRecord { seq, payload }))
+}
+
+/// Fresh open of the segment file (creating an empty one if
+/// missing). Returns `(head_seq, tail_seq, valid_len_bytes)`. A
+/// partial trailing record contributes `0` bytes to `valid_len` —
+/// caller uses that to size-truncate the file.
+async fn scan_segment(log_path: &Path) -> Result<(u64, u64, u64)> {
+    let mut file = match tokio::fs::File::open(log_path).await {
+        Ok(f) => f,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
+            // Create an empty segment so the append handle can open it.
+            OpenOptions::new()
+                .create(true)
+                .write(true)
+                .truncate(true)
+                .open(log_path)
+                .await
+                .with_context(|| format!("creating empty WAL {}", log_path.display()))?;
+            return Ok((0, 0, 0));
+        }
+        Err(e) => return Err(anyhow::Error::from(e)),
+    };
+    let mut head_seq = 0u64;
+    let mut tail_seq = 0u64;
+    let mut valid_len = 0u64;
+    loop {
+        match read_next_record(&mut file).await? {
+            Some(rec) => {
+                if head_seq == 0 {
+                    head_seq = rec.seq;
+                }
+                tail_seq = rec.seq;
+                valid_len += (RECORD_HEADER_LEN + rec.payload.len()) as u64;
+            }
+            None => break,
+        }
+    }
+    Ok((head_seq, tail_seq, valid_len))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    async fn open_wal(root: &tempfile::TempDir) -> WriteAheadLog {
+        WriteAheadLog::open(root.path().join("wal")).await.unwrap()
+    }
+
+    #[tokio::test]
+    async fn fresh_open_is_empty() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let wal = open_wal(&tmp).await;
+        assert!(wal.is_empty());
+        assert_eq!(wal.head_seq(), 0);
+        assert_eq!(wal.tail_seq(), 0);
+        assert_eq!(wal.iter_from(0).await.unwrap(), vec![]);
+    }
+
+    #[tokio::test]
+    async fn append_assigns_monotonic_seq() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let mut wal = open_wal(&tmp).await;
+        assert_eq!(wal.append(b"a").await.unwrap(), 1);
+        assert_eq!(wal.append(b"bb").await.unwrap(), 2);
+        assert_eq!(wal.append(b"ccc").await.unwrap(), 3);
+        assert_eq!(wal.tail_seq(), 3);
+        assert_eq!(wal.head_seq(), 1);
+    }
+
+    #[tokio::test]
+    async fn iter_from_replays_full_and_partial_ranges() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let mut wal = open_wal(&tmp).await;
+        for p in [&b"one"[..], b"two", b"three", b"four"] {
+            wal.append(p).await.unwrap();
+        }
+        let all = wal.iter_from(0).await.unwrap();
+        assert_eq!(all.len(), 4);
+        assert_eq!(all[0].seq, 1);
+        assert_eq!(all[3].payload, b"four".to_vec());
+
+        let tail = wal.iter_from(3).await.unwrap();
+        assert_eq!(tail.len(), 2);
+        assert_eq!(tail[0].seq, 3);
+        assert_eq!(tail[1].seq, 4);
+
+        assert!(wal.iter_from(99).await.unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn reopen_recovers_tail_seq() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        {
+            let mut wal = open_wal(&tmp).await;
+            wal.append(b"one").await.unwrap();
+            wal.append(b"two").await.unwrap();
+        }
+        let wal = open_wal(&tmp).await;
+        assert_eq!(wal.tail_seq(), 2);
+        assert_eq!(wal.head_seq(), 1);
+        let recs = wal.iter_from(0).await.unwrap();
+        assert_eq!(recs.len(), 2);
+        assert_eq!(recs[1].payload, b"two");
+    }
+
+    #[tokio::test]
+    async fn partial_trailing_record_is_truncated_on_open() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        {
+            let mut wal = open_wal(&tmp).await;
+            wal.append(b"one").await.unwrap();
+            wal.append(b"two").await.unwrap();
+        }
+        // Simulate torn write: append 5 stray bytes to the segment.
+        let log = tmp.path().join("wal/log.bin");
+        let mut f = OpenOptions::new().append(true).open(&log).await.unwrap();
+        f.write_all(&[0xAB, 0xCD, 0xEF, 0x01, 0x02]).await.unwrap();
+        f.sync_all().await.unwrap();
+        drop(f);
+
+        // Reopening drops the partial tail without erroring.
+        let wal = open_wal(&tmp).await;
+        assert_eq!(wal.tail_seq(), 2);
+        let recs = wal.iter_from(0).await.unwrap();
+        assert_eq!(recs.len(), 2);
+        // File is now exactly the length of two complete records.
+        let meta = tokio::fs::metadata(&log).await.unwrap();
+        assert_eq!(
+            meta.len(),
+            (RECORD_HEADER_LEN as u64 + 3) * 2,
+            "segment should be truncated to complete records"
+        );
+    }
+
+    #[tokio::test]
+    async fn corrupted_payload_is_a_hard_error() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        {
+            let mut wal = open_wal(&tmp).await;
+            wal.append(b"hello").await.unwrap();
+        }
+        // Flip a byte inside the payload region (offset 20+0 = header end).
+        let log = tmp.path().join("wal/log.bin");
+        let mut f = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(&log)
+            .await
+            .unwrap();
+        f.seek(std::io::SeekFrom::Start(RECORD_HEADER_LEN as u64))
+            .await
+            .unwrap();
+        f.write_all(&[0xFF]).await.unwrap();
+        f.sync_all().await.unwrap();
+        drop(f);
+
+        let res = WriteAheadLog::open(tmp.path().join("wal")).await;
+        assert!(res.is_err(), "corruption must surface, not silently drop data");
+    }
+
+    #[tokio::test]
+    async fn truncate_up_to_removes_prefix() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let mut wal = open_wal(&tmp).await;
+        for p in [&b"a"[..], b"b", b"c", b"d", b"e"] {
+            wal.append(p).await.unwrap();
+        }
+        wal.truncate_up_to(3).await.unwrap();
+        assert_eq!(wal.head_seq(), 4);
+        assert_eq!(wal.tail_seq(), 5);
+
+        let recs = wal.iter_from(0).await.unwrap();
+        assert_eq!(recs.len(), 2);
+        assert_eq!(recs[0].seq, 4);
+        assert_eq!(recs[1].seq, 5);
+    }
+
+    #[tokio::test]
+    async fn truncate_full_leaves_log_empty_and_appendable() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let mut wal = open_wal(&tmp).await;
+        wal.append(b"one").await.unwrap();
+        wal.append(b"two").await.unwrap();
+        wal.truncate_up_to(2).await.unwrap();
+        assert!(wal.is_empty());
+        // Next append continues the seq — the log doesn't reset to 1.
+        // Design choice: seq gaps are OK; consumers rely on monotonic,
+        // not dense.
+        let seq = wal.append(b"three").await.unwrap();
+        assert_eq!(seq, 1);
+        assert_eq!(wal.head_seq(), 1);
+        assert_eq!(wal.tail_seq(), 1);
+    }
+
+    #[tokio::test]
+    async fn truncate_below_head_is_noop() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let mut wal = open_wal(&tmp).await;
+        wal.append(b"a").await.unwrap();
+        wal.append(b"b").await.unwrap();
+        wal.truncate_up_to(1).await.unwrap();
+        assert_eq!(wal.head_seq(), 2);
+        // Now watermark is below head_seq — no-op.
+        wal.truncate_up_to(0).await.unwrap();
+        assert_eq!(wal.head_seq(), 2);
+        assert_eq!(wal.tail_seq(), 2);
+    }
+
+    #[tokio::test]
+    async fn large_payload_round_trips() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let mut wal = open_wal(&tmp).await;
+        let big = vec![0x5A; 1 << 20]; // 1 MiB
+        let seq = wal.append(&big).await.unwrap();
+        let wal2 = WriteAheadLog::open(tmp.path().join("wal")).await.unwrap();
+        let recs = wal2.iter_from(0).await.unwrap();
+        assert_eq!(recs.len(), 1);
+        assert_eq!(recs[0].seq, seq);
+        assert_eq!(recs[0].payload, big);
+    }
+
+    #[tokio::test]
+    async fn empty_payload_round_trips() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        let mut wal = open_wal(&tmp).await;
+        wal.append(&[]).await.unwrap();
+        wal.append(b"x").await.unwrap();
+        let recs = wal.iter_from(0).await.unwrap();
+        assert_eq!(recs.len(), 2);
+        assert!(recs[0].payload.is_empty());
+        assert_eq!(recs[1].payload, b"x");
+    }
+
+    #[tokio::test]
+    async fn append_after_reopen_continues_seq() {
+        let tmp = tempfile::TempDir::new().unwrap();
+        {
+            let mut wal = open_wal(&tmp).await;
+            wal.append(b"a").await.unwrap();
+            wal.append(b"b").await.unwrap();
+        }
+        let mut wal = open_wal(&tmp).await;
+        assert_eq!(wal.append(b"c").await.unwrap(), 3);
+        let recs = wal.iter_from(0).await.unwrap();
+        assert_eq!(recs.len(), 3);
+        assert_eq!(recs[2].payload, b"c");
+    }
+}