cubelinux: replay the write-ahead log, and be compared while doing it

A store on a device is [image][log]: the image is a checkpoint, the log is the
mutations since, in the framing cube-store/src/wal.rs writes (op | crc32 | space
| key | len | value, prefix-trusting recovery). The kernel now reads both and
reports the store they *describe* — the image with the log applied, in the order
a checkpoint would write it.

- the log's offset is fixed by geometry (the first 4 KiB boundary at or after the
  image's declared extent), so no superblock is needed to find it;
- recovery is prefix-trusting, exactly as userspace does it: the first entry that
  is short, mis-framed or fails its CRC ends the log;
- merging is by coordinate with the newest write winning, which is what makes an
  overwrite an overwrite and a delete a delete;
- the digest is taken over the surviving records in sorted order, and `bytes`
  reports the size a checkpoint of this store would produce — so the number can
  be compared with the image userspace actually writes, not merely eyeballed.

Gate (kernel/verify-log-replay.sh), on a pair carrying an overwrite, an
insertion and a deletion:

  reference : bytes=377 records=4 value_bytes=99 fnv1a64=a38c1db74b00966f
  kernel    : bytes=377 records=4 value_bytes=99 fnv1a64=a38c1db74b00966f
              log entries=6 applied over the image

It failed the first time, which is the point: the coalescing kept the earliest
write of a coordinate rather than the latest, resurrecting records the log had
deleted — six records where the fold produced four. A differential gate is what
turns that class of error into a red line instead of a corrupted store.

The image-only gates still pass unchanged (v1 curated, v1 snapshot 35,318
records, v2 store): with no log to apply the digest is the image's own order, so
v1 readers and every existing tool keep seeing exactly what they saw before.

drivers/cube/cubelinux_store.rs

@@ -9,9 +9,16 @@
 //! records: [SpaceId 32][Key 24][value length u64 LE][value]
 //! ```
 //!
-//! This module reads that image from a block device and reports what it holds: the record
-//! count, the total value bytes, and an FNV-1a digest over every `(space, key, length,
-//! value)` in order. The digest exists so the kernel and userspace can be *compared*
+//! A store on a device is that image followed by a **write-ahead log** — the mutations since
+//! the last checkpoint, in the same `op | crc32 | space | key | len | value` framing
+//! `cube-store/src/wal.rs` writes. The log is a delta: the image is authoritative and
+//! self-contained, and a reader that ignores the log loses only the mutations recorded after
+//! the last checkpoint. `DESIGN-cubelinux-write-path.md` is why the write path looks like
+//! this at all.
+//!
+//! This module reads both and reports the store they *describe* — the image with the log
+//! applied: the record count, the total value bytes, and an FNV-1a digest over every
+//! `(space, key, length, value)` in the order a checkpoint would write them. The digest exists so the kernel and userspace can be *compared*
 //! rather than assumed to agree — `cube-image digest <image>` prints the same line in the
 //! same field order, and the QEMU gate fails if they differ by a byte.
 //!
@@ -36,6 +43,7 @@
 use core::fmt::{self, Write};
 
 use kernel::{
+    alloc::AllocError,
     bindings, c_str,
     device::Device,
     fs::{File, Kiocb},
@@ -75,6 +83,18 @@ const STORE_DEVICE: &core::ffi::CStr = c_str!("/dev/vda");
 /// guard against a wrong device name turning a read into an allocation storm.
 const MAX_BYTES: usize = 64 * 1024 * 1024;
 
+/// The log's own magic, distinct from the image's so a reader that opens the wrong one
+/// cannot mistake it for the other.
+const WAL_MAGIC: [u8; 4] = *b"CUBW";
+const WAL_VERSION: u8 = 1;
+/// Log header: magic(4) + version(1) + curve tag(1).
+const WAL_HEADER_LEN: usize = 6;
+/// Log entry, before the value: op(1) + crc32(4) + space(32) + key(24) + len(4).
+const ENTRY_FIXED: usize = 1 + 4 + 32 + 24 + 4;
+/// The log begins at the first 4 KiB boundary at or after the image. Fixed by geometry so
+/// no superblock is needed to find it, and stated by the v2 header's extent.
+const LOG_ALIGN: usize = 4096;
+
 /// FNV-1a offset basis.
 const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
 
@@ -249,6 +269,24 @@ fn digest(image: &[u8]) -> Line {
         None => image.len(),
     };
 
+    // With a v2 image, the store is the image *plus* whatever log follows it. The order the
+    // digest is taken in is then the order a checkpoint would write: sorted by coordinate,
+    // later writes winning. Without a log there is nothing to merge, and the image's own
+    // order is the answer — which is also what keeps v1 images readable and unchanged.
+    if header.image_bytes.is_some() {
+        match merged_digest(image, end, &header) {
+            Ok(line) => return line,
+            // A log that exists but cannot be read is worth saying out loud: the digest
+            // alone would look like a shorter store.
+            Err(what) if what != "no-log" => {
+                let mut line = Line::new();
+                let _ = write!(line, "error={what}");
+                return line;
+            }
+            Err(_) => {}
+        }
+    }
+
     let mut h = FNV_OFFSET;
     let mut count: u64 = 0;
     let mut value_bytes: u64 = 0;
@@ -323,6 +361,233 @@ fn digest(image: &[u8]) -> Line {
     line
 }
 
+/// One record on its way into the merged store. Fixed size, so a `KVVec` of these sorts
+/// in place; values live in a pool beside them and are referenced by offset.
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+struct Entry {
+    space: [u8; 32],
+    key: [u8; 24],
+    /// Order in which this entry arrived. Breaks ties between the same coordinate, so the
+    /// later write wins after sorting.
+    seq: u32,
+    value_off: u32,
+    value_len: u32,
+    deleted: bool,
+}
+
+/// The store as a sorted set of records, which is what a checkpoint writes and what the
+/// digest is taken over.
+struct Merged {
+    pool: KVVec<u8>,
+    entries: KVVec<Entry>,
+    seq: u32,
+    /// Log entries that were seen but not applied, because they were torn or malformed.
+    dropped: u64,
+}
+
+impl Merged {
+    fn new() -> Result<Self, AllocError> {
+        Ok(Merged {
+            pool: KVVec::new(),
+            entries: KVVec::new(),
+            seq: 0,
+            dropped: 0,
+        })
+    }
+
+    /// Add a record. `deleted` marks a removal, which is kept in the list so it can
+    /// override an older value for the same coordinate.
+    fn add(&mut self, space: &[u8], key: &[u8], value: &[u8], deleted: bool) -> Result<(), AllocError> {
+        let mut sp = [0u8; 32];
+        sp.copy_from_slice(&space[..32]);
+        let mut k = [0u8; 24];
+        k.copy_from_slice(&key[..24]);
+        let value_off = self.pool.len() as u32;
+        self.pool.extend_from_slice(value, GFP_KERNEL)?;
+        let seq = self.seq;
+        self.seq += 1;
+        self.entries.push(
+            Entry {
+                space: sp,
+                key: k,
+                seq,
+                value_off,
+                value_len: value.len() as u32,
+                deleted,
+            },
+            GFP_KERNEL,
+        )?;
+        Ok(())
+    }
+
+    /// Put the entries in the order a checkpoint would write them: by space, then by key,
+    /// with the later write of a coordinate last — so a group's final entry decides it.
+    fn sort_entries(&mut self) {
+        self.entries.as_mut_slice().sort_unstable();
+    }
+
+    fn value(&self, e: &Entry) -> &[u8] {
+        let off = e.value_off as usize;
+        &self.pool.as_slice()[off..off + e.value_len as usize]
+    }
+}
+
+/// Read the log that follows the image, if there is one, and apply its entries.
+///
+/// The log is a delta on the image, so its entries override image records for the same
+/// coordinate. Recovery is prefix-trusting, exactly as userspace does it: entries are
+/// replayed from the start and the first one that is short, mis-framed or fails its
+/// checksum ends the log. A fault in the middle cannot be told from a torn tail without a
+/// second copy, so the log stops there and reports what it dropped instead of guessing.
+fn apply_log(image: &[u8], image_bytes: usize, merged: &mut Merged) -> Result<Option<u64>, &'static str> {
+    let log_off = (image_bytes + LOG_ALIGN - 1) & !(LOG_ALIGN - 1);
+    if log_off + WAL_HEADER_LEN > image.len() {
+        return Ok(None);
+    }
+    let log = &image[log_off..];
+    if &log[0..4] != WAL_MAGIC {
+        return Ok(None);
+    }
+    if log[4] != WAL_VERSION {
+        return Err("unsupported-log-version");
+    }
+
+    let mut applied: u64 = 0;
+    let mut off = WAL_HEADER_LEN;
+    while off + ENTRY_FIXED <= log.len() {
+        let start = off;
+        let op = log[off];
+        if op != 1 && op != 2 {
+            break;
+        }
+        let mut word = [0u8; 4];
+        word.copy_from_slice(&log[off + 1..off + 5]);
+        let crc = u32::from_le_bytes(word);
+        let space = &log[off + 5..off + 37];
+        let key = &log[off + 37..off + 61];
+        word.copy_from_slice(&log[off + 61..off + 65]);
+        let len = u32::from_le_bytes(word) as usize;
+        off += ENTRY_FIXED;
+        if off + len > log.len() {
+            break;
+        }
+        // The checksum covers space, key, length and value, so a corrupted entry is
+        // stopped at rather than applied.
+        if crc32(&log[start + 5..off + len]) != crc {
+            break;
+        }
+        let value = &log[off..off + len];
+        off += len;
+        merged
+            .add(space, key, value, op == 2)
+            .map_err(|_| "out-of-memory")?;
+        applied += 1;
+    }
+
+    merged.dropped = (log.len() - off) as u64;
+    Ok(Some(applied))
+}
+
+/// CRC-32 (IEEE 802.3), bitwise — the same polynomial and the same coverage as
+/// `cube-store/src/wal.rs`, because a log written on either side must validate on both.
+fn crc32(bytes: &[u8]) -> u32 {
+    let mut crc = 0xFFFF_FFFFu32;
+    for byte in bytes {
+        crc ^= *byte as u32;
+        for _ in 0..8 {
+            let mask = (crc & 1).wrapping_neg();
+            crc = (crc >> 1) ^ (0xEDB8_8320 & mask);
+        }
+    }
+    !crc
+}
+
+/// Digest the store the image and its log together describe.
+///
+/// Returns `Err("no-log")` when the device holds no log, so the caller can fall back to
+/// reading the image alone.
+fn merged_digest(image: &[u8], image_bytes: usize, header: &Header) -> Result<Line, &'static str> {
+    let mut merged = Merged::new().map_err(|_| "out-of-memory")?;
+
+    // The image's records first; the log overrides them where they collide.
+    let mut off = HEADER_LEN_V2;
+    let mut image_records: u64 = 0;
+    while off + RECORD_FIXED <= image_bytes {
+        if let Some(c) = header.record_count {
+            if image_records >= c {
+                break;
+            }
+        }
+        let space = &image[off..off + SPACE_ID_LEN];
+        let key = &image[off + SPACE_ID_LEN..off + SPACE_ID_LEN + RAW_KEY_LEN];
+        let mut word = [0u8; 8];
+        word.copy_from_slice(&image[off + SPACE_ID_LEN + RAW_KEY_LEN..off + RECORD_FIXED]);
+        let value_len = u64::from_le_bytes(word) as usize;
+        let value_at = off + RECORD_FIXED;
+        if value_at + value_len > image_bytes {
+            return Err("truncated-image");
+        }
+        merged
+            .add(space, key, &image[value_at..value_at + value_len], false)
+            .map_err(|_| "out-of-memory")?;
+        image_records += 1;
+        off = value_at + value_len;
+    }
+
+    let applied = match apply_log(image, image_bytes, &mut merged)? {
+        Some(n) => n,
+        None => return Err("no-log"),
+    };
+
+    let mut line = Line::new();
+    merged.sort_entries();
+    let survivors = merged.entries.as_slice();
+    let mut h = FNV_OFFSET;
+    let mut count: u64 = 0;
+    let mut value_bytes: u64 = 0;
+    // Sorting put every write of a coordinate together with the later one last, so a group
+    // is decided by its *final* entry — the newest write, or a removal. Keeping the first
+    // instead would resurrect records the log deleted, which is exactly what the gate caught
+    // the first time it ran.
+    let mut i = 0;
+    while i < survivors.len() {
+        let head = &survivors[i];
+        let mut last = i;
+        while last + 1 < survivors.len()
+            && survivors[last + 1].space == head.space
+            && survivors[last + 1].key == head.key
+        {
+            last += 1;
+        }
+        let winner = &survivors[last];
+        i = last + 1;
+        if winner.deleted {
+            continue;
+        }
+        let value = merged.value(winner);
+        h = fnv1a64(&winner.space, h);
+        h = fnv1a64(&winner.key, h);
+        h = fnv1a64(&(value.len() as u64).to_le_bytes(), h);
+        h = fnv1a64(value, h);
+        count += 1;
+        value_bytes += value.len() as u64;
+    }
+
+    // `bytes` is the size a checkpoint of this store would produce — the same number
+    // userspace reports for the image it writes from the same records.
+    let folded = HEADER_LEN_V2 as u64 + count * RECORD_FIXED as u64 + value_bytes;
+    // The digest line is field-for-field what `cube-image digest` prints, so the two are
+    // compared by diff. What the log contributed goes on its own line: a padded log region
+    // makes a byte count meaningless, an entry count does not.
+    let _ = write!(
+        line,
+        "digest version={} curve={} bytes={} records={} value_bytes={} fnv1a64={:016x} errors=0\n",
+        header.version, header.curve, folded, count, value_bytes, h
+    );
+    let _ = write!(line, "log entries={} applied over the image\n", applied);
+    Ok(line)
+}
+
 /// The module's registration; holds the misc device for as long as the module lives.
 #[pin_data]
 struct CubeStoreModule {