Omar SobhandClaude Opus 4.7
0cf00e5954
Build with clawstor cache / Cargo build (clawstor-cached) (pull_request) Successful in 17s
Turns the single-file Phase 4c WAL into a segmented log so it
can grow past a single file safely. This unblocks every
downstream Phase 4d/4e integration — reconnect + push loop
can't rely on an unbounded single file.
Layout change:
<root>/segment-<20-digit-first-seq>.bin
20-digit zero-padded first-seq means lex sort == numeric sort,
so `read_dir + sort_by_key` recovers the natural order.
Rotation policy:
* `max_segment_bytes` default 8 MiB, overridable via
`open_with_options`.
* `append` rolls to a fresh segment BEFORE writing when the
current tail is non-empty AND at/above the cap. A single
oversize record always lands in one segment — we never split
a record.
Truncation across segments:
* whole segments with `last_seq <= watermark` are `unlink`'d
* the boundary segment (if any) is rewritten in place via
`tempfile-in-parent + rename` + parent-dir fsync
* full truncation resets head/tail to 0 and the next append
creates a fresh segment
Legacy compat: on open, if a pre-4d `log.bin` is present and
no `segment-*.bin` files exist, it is scanned for its first
seq and renamed to the correct segment name. Refuses to
silently overwrite on filename collision.
Tests (18, all green): rotation-happens-at-cap, reopen-
enumerates-all-segments, truncate-drops-whole-segments,
truncate-partial-rewrites-boundary, oversize-record-still-
fits-one-segment, legacy-log.bin-migration, plus the full
Phase 4c suite (fresh open, append, iter partial ranges,
reopen recovers tail, torn-write truncation, corruption is
hard error, full truncation appendable, below-head no-op,
large payload, empty payload, append-after-reopen).
942-line file, comfortably under the 1300-line ceiling.
Co-Authored-By: Claude Opus 4.7 <[email protected]>
943 lines
34 KiB
Rust
943 lines
34 KiB
Rust
//! Phase 4c/4d (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 WAL root is scanned. Segments are named
|
|
//! `segment-<20digit-first-seq>.bin` so lexical sort equals seq
|
|
//! order. Each segment is scanned linearly. A short read or wrong
|
|
//! length prefix on the *last* segment stops the scan — the file
|
|
//! is size-truncated to the last complete record. Corruption on a
|
|
//! full-length record surfaces as a hard error.
|
|
//!
|
|
//! Phase 4d (2026-07-13): segment rotation. New records land in the
|
|
//! newest segment until it exceeds `max_segment_bytes` (default
|
|
//! 8 MiB), at which point `append` opens a new segment on the next
|
|
//! seq. Old segments become read-only. `truncate_up_to` drops whole
|
|
//! segments below the watermark and rewrites the boundary one.
|
|
//!
|
|
//! 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;
|
|
|
|
/// Default per-segment soft cap. A single record can still push a
|
|
/// segment past this — rotation is checked *before* the write so the
|
|
/// current record always lands cleanly.
|
|
pub const DEFAULT_MAX_SEGMENT_BYTES: u64 = 8 * 1024 * 1024;
|
|
|
|
/// Segment filename width for the zero-padded first-seq field. 20
|
|
/// digits covers `u64::MAX`; lex-sort == numeric-sort.
|
|
const SEQ_FIELD_WIDTH: 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>,
|
|
}
|
|
|
|
/// One immutable segment on disk. The tail segment is mutable —
|
|
/// this struct only tracks *state* (its bounds and byte length).
|
|
#[derive(Debug, Clone)]
|
|
struct Segment {
|
|
path: PathBuf,
|
|
/// First seq value stored in this segment. Encoded in the filename.
|
|
first_seq: u64,
|
|
/// Last seq value stored. Zero for a freshly-created empty segment
|
|
/// — treated as "unknown, please scan" only during open.
|
|
last_seq: u64,
|
|
/// Byte length on disk (as of last scan or write).
|
|
byte_len: u64,
|
|
}
|
|
|
|
/// File-backed segmented WAL. Single-writer; `append` takes
|
|
/// `&mut self` so serialization is enforced at the type level.
|
|
/// Readers (`iter_from`, `head_seq`, `tail_seq`) work against
|
|
/// `&self` and open their own file handles.
|
|
pub struct WriteAheadLog {
|
|
root: PathBuf,
|
|
segments: Vec<Segment>,
|
|
/// Cached append handle for the tail segment. Opened in append
|
|
/// mode so offsets always land at the end.
|
|
writer: tokio::fs::File,
|
|
max_segment_bytes: u64,
|
|
head_seq: u64,
|
|
tail_seq: u64,
|
|
}
|
|
|
|
impl WriteAheadLog {
|
|
/// Open (or create) the WAL rooted at `root`. Scans the segment
|
|
/// directory to recover `head_seq` / `tail_seq`. A partial
|
|
/// trailing record on the last segment is size-truncated in
|
|
/// place. Uses the default per-segment cap.
|
|
pub async fn open(root: impl Into<PathBuf>) -> Result<Self> {
|
|
Self::open_with_options(root, DEFAULT_MAX_SEGMENT_BYTES).await
|
|
}
|
|
|
|
/// Open with a caller-supplied per-segment cap. Values below
|
|
/// `RECORD_HEADER_LEN` are rounded up — you can't produce a
|
|
/// segment that fits zero records.
|
|
pub async fn open_with_options(
|
|
root: impl Into<PathBuf>,
|
|
max_segment_bytes: u64,
|
|
) -> Result<Self> {
|
|
let root = root.into();
|
|
tokio::fs::create_dir_all(&root)
|
|
.await
|
|
.with_context(|| format!("creating WAL root {}", root.display()))?;
|
|
|
|
let max_segment_bytes = max_segment_bytes.max(RECORD_HEADER_LEN as u64);
|
|
|
|
// Migrate the pre-4d single-file layout if we find one.
|
|
migrate_legacy_log(&root).await?;
|
|
|
|
let mut segments = load_segments(&root).await?;
|
|
|
|
// Scan the tail segment (if any) to recover last_seq +
|
|
// fix a torn trailing record. Middle segments are trusted
|
|
// to be intact — a crash can only tear the file currently
|
|
// being appended to.
|
|
if let Some(tail) = segments.last_mut() {
|
|
let (last_seq, valid_len) = scan_segment_tail(&tail.path).await?;
|
|
if valid_len != tail.byte_len {
|
|
truncate_file(&tail.path, valid_len).await?;
|
|
tail.byte_len = valid_len;
|
|
}
|
|
if last_seq >= tail.first_seq {
|
|
tail.last_seq = last_seq;
|
|
} else {
|
|
// Segment is empty on disk. Delete + drop.
|
|
let path = tail.path.clone();
|
|
segments.pop();
|
|
let _ = tokio::fs::remove_file(&path).await;
|
|
}
|
|
}
|
|
|
|
let head_seq = segments.first().map(|s| s.first_seq).unwrap_or(0);
|
|
let tail_seq = segments.last().map(|s| s.last_seq).unwrap_or(0);
|
|
|
|
// Ensure we have a writable tail segment. If the WAL is
|
|
// empty, the first append will create segment-0000...001.
|
|
let writer = match segments.last() {
|
|
Some(tail) => open_append(&tail.path).await?,
|
|
None => {
|
|
// Placeholder handle that will be replaced on first
|
|
// append. Point it at the root dir sentinel we know
|
|
// exists so we don't hold onto stale state.
|
|
//
|
|
// Open a self-closing tempfile inside `root` and drop
|
|
// it — the `writer` field will be reassigned before
|
|
// any write.
|
|
let placeholder = root.join(".wal-writer-placeholder");
|
|
let f = OpenOptions::new()
|
|
.create(true)
|
|
.write(true)
|
|
.truncate(true)
|
|
.open(&placeholder)
|
|
.await?;
|
|
let _ = tokio::fs::remove_file(&placeholder).await;
|
|
f
|
|
}
|
|
};
|
|
|
|
Ok(Self {
|
|
root,
|
|
segments,
|
|
writer,
|
|
max_segment_bytes,
|
|
head_seq,
|
|
tail_seq,
|
|
})
|
|
}
|
|
|
|
/// Root directory this WAL lives in.
|
|
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
|
|
}
|
|
|
|
/// Number of segments currently on disk. Test hook — production
|
|
/// callers should not depend on this.
|
|
pub fn segment_count(&self) -> usize {
|
|
self.segments.len()
|
|
}
|
|
|
|
/// Append a payload. Returns the assigned seq. Persisted with
|
|
/// `fsync` before returning so callers can treat success as
|
|
/// durable. Rolls a new segment when the current one is at or
|
|
/// above `max_segment_bytes` *and* non-empty (so a single
|
|
/// oversize record still lands in one segment).
|
|
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 frame_len = (RECORD_HEADER_LEN + payload.len()) as u64;
|
|
let should_roll = match self.segments.last() {
|
|
Some(tail) => tail.byte_len > 0 && tail.byte_len >= self.max_segment_bytes,
|
|
None => true,
|
|
};
|
|
if should_roll {
|
|
self.roll_to_new_segment(seq).await?;
|
|
}
|
|
|
|
let mut frame = Vec::with_capacity(frame_len as usize);
|
|
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
|
|
.context("appending WAL frame")?;
|
|
self.writer.flush().await.context("flushing WAL append")?;
|
|
self.writer.sync_all().await.context("fsync WAL append")?;
|
|
|
|
let tail = self
|
|
.segments
|
|
.last_mut()
|
|
.expect("tail segment created above");
|
|
tail.byte_len += frame_len;
|
|
tail.last_seq = seq;
|
|
if self.head_seq == 0 {
|
|
self.head_seq = seq;
|
|
}
|
|
self.tail_seq = seq;
|
|
Ok(seq)
|
|
}
|
|
|
|
/// Read every record with `seq >= start_seq` in seq order.
|
|
pub async fn iter_from(&self, start_seq: u64) -> Result<Vec<WalRecord>> {
|
|
let mut out = Vec::new();
|
|
for seg in &self.segments {
|
|
if seg.last_seq < start_seq {
|
|
continue;
|
|
}
|
|
let mut file = tokio::fs::File::open(&seg.path).await.with_context(|| {
|
|
format!("opening WAL segment {}", seg.path.display())
|
|
})?;
|
|
loop {
|
|
match read_next_record(&mut file).await? {
|
|
Some(rec) if rec.seq >= start_seq => out.push(rec),
|
|
Some(_) => {}
|
|
None => break,
|
|
}
|
|
}
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
/// Drop every record with `seq <= watermark`. Whole segments
|
|
/// whose `last_seq <= watermark` are `unlink`'d. The segment
|
|
/// containing the watermark (if any) is rewritten in place via
|
|
/// `tempfile-in-parent + rename`. No-op when the watermark is
|
|
/// below the current head or the log is empty.
|
|
pub async fn truncate_up_to(&mut self, watermark: u64) -> Result<()> {
|
|
if self.tail_seq == 0 || watermark < self.head_seq {
|
|
return Ok(());
|
|
}
|
|
|
|
// Drop whole segments that end at or below the watermark.
|
|
let mut kept = Vec::with_capacity(self.segments.len());
|
|
for seg in self.segments.drain(..) {
|
|
if seg.last_seq <= watermark {
|
|
let _ = tokio::fs::remove_file(&seg.path).await;
|
|
} else {
|
|
kept.push(seg);
|
|
}
|
|
}
|
|
self.segments = kept;
|
|
|
|
// Partial-drop the boundary segment if it straddles the
|
|
// watermark. After the drain loop, if the surviving head
|
|
// starts below-or-at the watermark, it needs a rewrite.
|
|
if let Some(first) = self.segments.first().cloned() {
|
|
if first.first_seq <= watermark {
|
|
self.rewrite_segment_above(watermark).await?;
|
|
}
|
|
}
|
|
|
|
// Recompute bounds. If everything was dropped, reset both
|
|
// and close the append handle onto a placeholder so the
|
|
// next `append` creates a fresh segment cleanly.
|
|
if self.segments.is_empty() {
|
|
self.head_seq = 0;
|
|
self.tail_seq = 0;
|
|
self.writer = open_placeholder(&self.root).await?;
|
|
} else {
|
|
self.head_seq = self.segments.first().unwrap().first_seq;
|
|
self.tail_seq = self.segments.last().unwrap().last_seq;
|
|
self.writer = open_append(&self.segments.last().unwrap().path).await?;
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Roll the tail to a brand-new segment starting at
|
|
/// `first_seq`. Fsyncs the old tail first so nothing straddles.
|
|
async fn roll_to_new_segment(&mut self, first_seq: u64) -> Result<()> {
|
|
// Fsync the current tail (harmless if placeholder).
|
|
let _ = self.writer.sync_all().await;
|
|
|
|
let path = segment_path(&self.root, first_seq);
|
|
let f = OpenOptions::new()
|
|
.create_new(true)
|
|
.append(true)
|
|
.open(&path)
|
|
.await
|
|
.with_context(|| format!("creating segment {}", path.display()))?;
|
|
self.writer = f;
|
|
self.segments.push(Segment {
|
|
path,
|
|
first_seq,
|
|
last_seq: 0,
|
|
byte_len: 0,
|
|
});
|
|
// Fsync the parent directory so the new file is durable.
|
|
fsync_dir(&self.root).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Rewrite the boundary segment: keep records with `seq >
|
|
/// watermark`, drop the rest. Atomic via `tempfile-in-parent +
|
|
/// rename + parent fsync`.
|
|
async fn rewrite_segment_above(&mut self, watermark: u64) -> Result<()> {
|
|
// Boundary segment is always segments[0] after the drain
|
|
// loop above.
|
|
let boundary = self.segments[0].clone();
|
|
let tmp = tempfile::NamedTempFile::new_in(&self.root)
|
|
.with_context(|| format!("tempfile in {}", self.root.display()))?;
|
|
let tmp_path = tmp.path().to_path_buf();
|
|
drop(tmp);
|
|
|
|
let mut new_first = 0u64;
|
|
let mut new_last = 0u64;
|
|
let mut new_bytes = 0u64;
|
|
{
|
|
let mut writer = OpenOptions::new()
|
|
.create(true)
|
|
.write(true)
|
|
.truncate(true)
|
|
.open(&tmp_path)
|
|
.await
|
|
.with_context(|| format!("opening rewrite tempfile {}", tmp_path.display()))?;
|
|
let mut reader = tokio::fs::File::open(&boundary.path).await?;
|
|
loop {
|
|
match read_next_record(&mut reader).await? {
|
|
Some(rec) if rec.seq > watermark => {
|
|
let frame_len = (RECORD_HEADER_LEN + rec.payload.len()) as u64;
|
|
let mut frame = Vec::with_capacity(frame_len as usize);
|
|
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_first == 0 {
|
|
new_first = rec.seq;
|
|
}
|
|
new_last = rec.seq;
|
|
new_bytes += frame_len;
|
|
}
|
|
Some(_) => {}
|
|
None => break,
|
|
}
|
|
}
|
|
writer.flush().await?;
|
|
writer.sync_all().await?;
|
|
}
|
|
|
|
if new_first == 0 {
|
|
// Everything in the boundary was <= watermark. Drop it.
|
|
let _ = tokio::fs::remove_file(&boundary.path).await;
|
|
let _ = tokio::fs::remove_file(&tmp_path).await;
|
|
self.segments.remove(0);
|
|
return Ok(());
|
|
}
|
|
|
|
// Rename tempfile → new segment path. If the new first seq
|
|
// differs from the boundary's, drop the old file first.
|
|
let new_path = segment_path(&self.root, new_first);
|
|
if new_path != boundary.path {
|
|
let _ = tokio::fs::remove_file(&boundary.path).await;
|
|
}
|
|
tokio::fs::rename(&tmp_path, &new_path).await.with_context(|| {
|
|
format!("renaming {} → {}", tmp_path.display(), new_path.display())
|
|
})?;
|
|
fsync_dir(&self.root).await?;
|
|
|
|
self.segments[0] = Segment {
|
|
path: new_path,
|
|
first_seq: new_first,
|
|
last_seq: new_last,
|
|
byte_len: new_bytes,
|
|
};
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
fn segment_path(root: &Path, first_seq: u64) -> PathBuf {
|
|
root.join(format!("segment-{:0width$}.bin", first_seq, width = SEQ_FIELD_WIDTH))
|
|
}
|
|
|
|
fn parse_segment_first_seq(name: &str) -> Option<u64> {
|
|
let core = name.strip_prefix("segment-")?.strip_suffix(".bin")?;
|
|
if core.len() != SEQ_FIELD_WIDTH {
|
|
return None;
|
|
}
|
|
core.parse::<u64>().ok()
|
|
}
|
|
|
|
async fn open_append(path: &Path) -> Result<tokio::fs::File> {
|
|
OpenOptions::new()
|
|
.create(true)
|
|
.append(true)
|
|
.open(path)
|
|
.await
|
|
.with_context(|| format!("open-append {}", path.display()))
|
|
}
|
|
|
|
async fn open_placeholder(root: &Path) -> Result<tokio::fs::File> {
|
|
let placeholder = root.join(".wal-writer-placeholder");
|
|
let f = OpenOptions::new()
|
|
.create(true)
|
|
.write(true)
|
|
.truncate(true)
|
|
.open(&placeholder)
|
|
.await?;
|
|
let _ = tokio::fs::remove_file(&placeholder).await;
|
|
Ok(f)
|
|
}
|
|
|
|
async fn truncate_file(path: &Path, len: u64) -> Result<()> {
|
|
let f = OpenOptions::new()
|
|
.write(true)
|
|
.open(path)
|
|
.await
|
|
.with_context(|| format!("opening {} for truncation", path.display()))?;
|
|
f.set_len(len)
|
|
.await
|
|
.with_context(|| format!("truncating {} to {}", path.display(), len))?;
|
|
f.sync_all().await.context("fsync after truncation")?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn fsync_dir(dir: &Path) -> Result<()> {
|
|
let handle = tokio::fs::File::open(dir)
|
|
.await
|
|
.with_context(|| format!("opening WAL dir {}", dir.display()))?;
|
|
handle
|
|
.sync_all()
|
|
.await
|
|
.with_context(|| format!("fsync WAL dir {}", dir.display()))?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn load_segments(root: &Path) -> Result<Vec<Segment>> {
|
|
let mut out = Vec::new();
|
|
let mut rd = tokio::fs::read_dir(root)
|
|
.await
|
|
.with_context(|| format!("reading WAL root {}", root.display()))?;
|
|
while let Some(entry) = rd.next_entry().await? {
|
|
let name = match entry.file_name().into_string() {
|
|
Ok(n) => n,
|
|
Err(_) => continue,
|
|
};
|
|
let first_seq = match parse_segment_first_seq(&name) {
|
|
Some(v) => v,
|
|
None => continue,
|
|
};
|
|
let meta = entry.metadata().await?;
|
|
if !meta.is_file() {
|
|
continue;
|
|
}
|
|
out.push(Segment {
|
|
path: entry.path(),
|
|
first_seq,
|
|
last_seq: 0, // filled by scan on the tail; middle segments compute lazily below
|
|
byte_len: meta.len(),
|
|
});
|
|
}
|
|
out.sort_by_key(|s| s.first_seq);
|
|
|
|
// For middle segments, cheaply derive last_seq by reading the
|
|
// tail header (last 20 bytes could still be a valid header
|
|
// *record*, but we need the payload len too — easiest: full
|
|
// linear scan of each). Middle segments are trusted to be
|
|
// intact so a short-circuit is fine.
|
|
let last_idx = out.len().saturating_sub(1);
|
|
for (i, seg) in out.iter_mut().enumerate() {
|
|
if i == last_idx {
|
|
continue;
|
|
}
|
|
let (last_seq, _valid_len) = scan_segment_tail(&seg.path).await?;
|
|
seg.last_seq = last_seq.max(seg.first_seq);
|
|
}
|
|
|
|
Ok(out)
|
|
}
|
|
|
|
/// Legacy compat: rename `log.bin` (pre-4d) to a properly-named
|
|
/// segment so it participates in the new enumeration. Zero-op when
|
|
/// no legacy file exists.
|
|
async fn migrate_legacy_log(root: &Path) -> Result<()> {
|
|
let legacy = root.join("log.bin");
|
|
if tokio::fs::metadata(&legacy).await.is_err() {
|
|
return Ok(());
|
|
}
|
|
// Peek at the first record to learn first_seq.
|
|
let mut f = tokio::fs::File::open(&legacy).await?;
|
|
let first_seq = match read_next_record(&mut f).await? {
|
|
Some(r) => r.seq,
|
|
None => {
|
|
// Empty file — just delete.
|
|
let _ = tokio::fs::remove_file(&legacy).await;
|
|
return Ok(());
|
|
}
|
|
};
|
|
drop(f);
|
|
let new_path = segment_path(root, first_seq);
|
|
if tokio::fs::metadata(&new_path).await.is_ok() {
|
|
// Collision — refuse rather than silently overwriting. This
|
|
// is a hand-repair case; document with the error.
|
|
bail!(
|
|
"legacy WAL log.bin found alongside {}; refusing to overwrite",
|
|
new_path.display()
|
|
);
|
|
}
|
|
tokio::fs::rename(&legacy, &new_path).await?;
|
|
fsync_dir(root).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// First 8 bytes of BLAKE3 over `seq || len || payload`. Keyless
|
|
/// checksum: not tamper-resistant; catches 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 —
|
|
/// 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 => {
|
|
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 segment)",
|
|
seq,
|
|
start
|
|
);
|
|
}
|
|
Ok(Some(WalRecord { seq, payload }))
|
|
}
|
|
|
|
/// Scan a segment forward. Returns `(last_seq, valid_byte_len)` —
|
|
/// `last_seq` is 0 if no complete records are present, and
|
|
/// `valid_byte_len` is the offset of the first partial/absent
|
|
/// record (so callers can size-truncate at that boundary).
|
|
async fn scan_segment_tail(path: &Path) -> Result<(u64, u64)> {
|
|
let mut file = tokio::fs::File::open(path)
|
|
.await
|
|
.with_context(|| format!("scanning WAL segment {}", path.display()))?;
|
|
let mut last_seq = 0u64;
|
|
let mut valid = 0u64;
|
|
loop {
|
|
match read_next_record(&mut file).await? {
|
|
Some(rec) => {
|
|
last_seq = rec.seq;
|
|
valid += (RECORD_HEADER_LEN + rec.payload.len()) as u64;
|
|
}
|
|
None => break,
|
|
}
|
|
}
|
|
Ok((last_seq, valid))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
async fn open_wal(root: &tempfile::TempDir) -> WriteAheadLog {
|
|
WriteAheadLog::open(root.path().join("wal")).await.unwrap()
|
|
}
|
|
|
|
async fn open_wal_capped(root: &tempfile::TempDir, cap: u64) -> WriteAheadLog {
|
|
WriteAheadLog::open_with_options(root.path().join("wal"), cap)
|
|
.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.segment_count(), 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);
|
|
assert_eq!(wal.segment_count(), 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);
|
|
let tail = wal.iter_from(3).await.unwrap();
|
|
assert_eq!(tail.len(), 2);
|
|
assert_eq!(tail[0].seq, 3);
|
|
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);
|
|
}
|
|
|
|
#[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();
|
|
}
|
|
// Find the tail segment and append 5 stray bytes.
|
|
let mut names: Vec<_> = std::fs::read_dir(tmp.path().join("wal"))
|
|
.unwrap()
|
|
.filter_map(|e| e.ok().map(|e| e.path()))
|
|
.filter(|p| {
|
|
p.file_name()
|
|
.and_then(|s| s.to_str())
|
|
.map(|n| n.starts_with("segment-"))
|
|
.unwrap_or(false)
|
|
})
|
|
.collect();
|
|
names.sort();
|
|
let tail = names.last().cloned().unwrap();
|
|
let mut f = OpenOptions::new().append(true).open(&tail).await.unwrap();
|
|
f.write_all(&[0xAB, 0xCD, 0xEF, 0x01, 0x02]).await.unwrap();
|
|
f.sync_all().await.unwrap();
|
|
drop(f);
|
|
|
|
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);
|
|
let meta = tokio::fs::metadata(&tail).await.unwrap();
|
|
assert_eq!(meta.len(), (RECORD_HEADER_LEN as u64 + 3) * 2);
|
|
}
|
|
|
|
#[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();
|
|
}
|
|
let names: Vec<_> = std::fs::read_dir(tmp.path().join("wal"))
|
|
.unwrap()
|
|
.filter_map(|e| e.ok().map(|e| e.path()))
|
|
.filter(|p| {
|
|
p.file_name()
|
|
.and_then(|s| s.to_str())
|
|
.map(|n| n.starts_with("segment-"))
|
|
.unwrap_or(false)
|
|
})
|
|
.collect();
|
|
let seg = names[0].clone();
|
|
let mut f = OpenOptions::new()
|
|
.read(true)
|
|
.write(true)
|
|
.open(&seg)
|
|
.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);
|
|
assert!(WriteAheadLog::open(tmp.path().join("wal")).await.is_err());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn rotation_opens_new_segment_when_cap_exceeded() {
|
|
let tmp = tempfile::TempDir::new().unwrap();
|
|
// Cap is 40 bytes: two 20-byte-header + 0-byte-payload records
|
|
// fit (40 bytes exactly triggers rotation next).
|
|
let cap = 40;
|
|
let mut wal = open_wal_capped(&tmp, cap).await;
|
|
wal.append(&[]).await.unwrap(); // seq 1
|
|
wal.append(&[]).await.unwrap(); // seq 2
|
|
assert_eq!(wal.segment_count(), 1);
|
|
wal.append(&[]).await.unwrap(); // seq 3 → rolls
|
|
assert_eq!(wal.segment_count(), 2);
|
|
wal.append(&[]).await.unwrap(); // seq 4
|
|
assert_eq!(wal.segment_count(), 2);
|
|
let recs = wal.iter_from(0).await.unwrap();
|
|
assert_eq!(recs.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![1, 2, 3, 4]);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn reopen_enumerates_all_segments() {
|
|
let tmp = tempfile::TempDir::new().unwrap();
|
|
{
|
|
let mut wal = open_wal_capped(&tmp, 40).await;
|
|
for _ in 0..5 {
|
|
wal.append(&[]).await.unwrap();
|
|
}
|
|
}
|
|
let wal = open_wal_capped(&tmp, 40).await;
|
|
assert_eq!(wal.tail_seq(), 5);
|
|
assert_eq!(wal.head_seq(), 1);
|
|
assert!(wal.segment_count() >= 2);
|
|
assert_eq!(wal.iter_from(0).await.unwrap().len(), 5);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn truncate_up_to_drops_whole_segments() {
|
|
let tmp = tempfile::TempDir::new().unwrap();
|
|
let mut wal = open_wal_capped(&tmp, 40).await;
|
|
for _ in 0..5 {
|
|
wal.append(&[]).await.unwrap();
|
|
}
|
|
let before = wal.segment_count();
|
|
// watermark 4 → seqs 1-4 gone.
|
|
wal.truncate_up_to(4).await.unwrap();
|
|
assert_eq!(wal.head_seq(), 5);
|
|
assert_eq!(wal.tail_seq(), 5);
|
|
let recs = wal.iter_from(0).await.unwrap();
|
|
assert_eq!(recs.len(), 1);
|
|
assert_eq!(recs[0].seq, 5);
|
|
assert!(wal.segment_count() < before);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn truncate_partial_rewrites_boundary_segment() {
|
|
let tmp = tempfile::TempDir::new().unwrap();
|
|
let mut wal = open_wal_capped(&tmp, 100).await; // big enough to hold several
|
|
for _ in 0..5 {
|
|
wal.append(&[0xAA]).await.unwrap();
|
|
}
|
|
// All 5 in one segment. watermark 3 → 4,5 survive.
|
|
wal.truncate_up_to(3).await.unwrap();
|
|
assert_eq!(wal.head_seq(), 4);
|
|
assert_eq!(wal.tail_seq(), 5);
|
|
assert_eq!(wal.segment_count(), 1);
|
|
let recs = wal.iter_from(0).await.unwrap();
|
|
assert_eq!(recs.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![4, 5]);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn truncate_full_leaves_log_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());
|
|
assert_eq!(wal.segment_count(), 0);
|
|
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);
|
|
wal.truncate_up_to(0).await.unwrap();
|
|
assert_eq!(wal.head_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];
|
|
let seq = wal.append(&big).await.unwrap();
|
|
let wal2 = open_wal(&tmp).await;
|
|
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 oversize_record_still_fits_one_segment() {
|
|
let tmp = tempfile::TempDir::new().unwrap();
|
|
// Cap is 32 bytes but payload alone is 1000. Rotation must
|
|
// not trigger for a single-record segment.
|
|
let mut wal = open_wal_capped(&tmp, 32).await;
|
|
wal.append(&vec![0x77; 1000]).await.unwrap();
|
|
assert_eq!(wal.segment_count(), 1);
|
|
let recs = wal.iter_from(0).await.unwrap();
|
|
assert_eq!(recs[0].payload.len(), 1000);
|
|
}
|
|
|
|
#[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);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn legacy_log_bin_is_migrated_on_open() {
|
|
let tmp = tempfile::TempDir::new().unwrap();
|
|
let root = tmp.path().join("wal");
|
|
std::fs::create_dir_all(&root).unwrap();
|
|
// Hand-craft a legacy log.bin with two records (seq 1, 2).
|
|
let legacy = root.join("log.bin");
|
|
let mut bytes = Vec::new();
|
|
for (seq, payload) in [(1u64, &b"hi"[..]), (2u64, &b"bye"[..])] {
|
|
bytes.extend_from_slice(&seq.to_le_bytes());
|
|
bytes.extend_from_slice(&(payload.len() as u32).to_le_bytes());
|
|
bytes.extend_from_slice(&checksum(seq, payload));
|
|
bytes.extend_from_slice(payload);
|
|
}
|
|
std::fs::write(&legacy, &bytes).unwrap();
|
|
|
|
let wal = WriteAheadLog::open(&root).await.unwrap();
|
|
assert_eq!(wal.head_seq(), 1);
|
|
assert_eq!(wal.tail_seq(), 2);
|
|
assert!(!legacy.exists(), "legacy log.bin should be renamed away");
|
|
assert_eq!(wal.segment_count(), 1);
|
|
let recs = wal.iter_from(0).await.unwrap();
|
|
assert_eq!(recs.len(), 2);
|
|
assert_eq!(recs[1].payload, b"bye".to_vec());
|
|
}
|
|
|
|
#[test]
|
|
fn segment_name_round_trips() {
|
|
let path = segment_path(Path::new("/tmp/w"), 42);
|
|
assert!(path.to_string_lossy().ends_with("segment-00000000000000000042.bin"));
|
|
assert_eq!(
|
|
parse_segment_first_seq("segment-00000000000000000042.bin"),
|
|
Some(42)
|
|
);
|
|
assert_eq!(parse_segment_first_seq("segment-42.bin"), None); // wrong width
|
|
assert_eq!(parse_segment_first_seq("junk"), None);
|
|
}
|
|
}
|