// SPDX-License-Identifier: GPL-2.0

//! CUBELinux store reader — the kernel reading CUBE coordinates off a block device.
//!
//! The store is addressed by coordinate, and the pinned image's layout is:
//!
//! ```text
//! [MAGIC "CUBE" 4][version 1][curve tag]
//! records: [SpaceId 32][Key 24][value length u64 LE][value]
//! ```
//!
//! 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.
//!
//! # Why a reader, and only a reader
//!
//! The store's write authority has not moved yet. `PLAN-kernel-cubelinux.md` records the
//! decision (the kernel owns the store) and the hazard that makes the order matter: a
//! kernel that writes while a userspace daemon still holds the same image in memory loses
//! one of the two writers' work, silently. Until that is settled, a reader is the correct
//! amount of authority for the kernel to hold.
//!
//! # Interface
//!
//! `/dev/cubelinux` — reading it performs the read (through the kernel's own file layer)
//! and returns one line. Nothing happens at module init, so there is no ordering to get
//! wrong against the block driver that provides the device.
//!
//! ```text
//! digest curve=0 bytes=1141 records=11 value_bytes=431 fnv1a64=161113085b1573b2 errors=0
//! ```

use core::fmt::{self, Write};

use kernel::{
    alloc::AllocError,
    bindings, c_str,
    device::Device,
    fs::{File, Kiocb},
    iov::{IovIterDest, IovIterSource},
    miscdevice::{MiscDevice, MiscDeviceOptions, MiscDeviceRegistration},
    prelude::*,
    sync::aref::ARef,
};

module! {
    type: CubeStoreModule,
    name: "cubelinux_store",
    authors: ["CUBE OS"],
    description: "CUBELinux store reader (coordinates in the kernel)",
    license: "GPL",
}

/// The layout, restated here because the kernel cannot depend on the userspace crates.
/// `crates/cube-store-raw` is the source of truth; the digest comparison is what keeps
/// this copy honest.
const MAGIC: &[u8; 4] = b"CUBE";
/// The original format: magic, version, curve, then records until zero padding.
const VERSION_V1: u8 = 1;
/// The current format: the same, plus the image's byte extent and record count.
const VERSION: u8 = 2;
const HEADER_LEN_V1: usize = 6;
const HEADER_LEN_V2: usize = 6 + 8 + 8;
const SPACE_ID_LEN: usize = 32;
const RAW_KEY_LEN: usize = 24;
const RECORD_FIXED: usize = SPACE_ID_LEN + RAW_KEY_LEN + 8;

/// The device to read. A module parameter is the obvious next step; for the boot gate one
/// fixed name is honest and has one less way to be wrong.
const STORE_DEVICE: &core::ffi::CStr = c_str!("/dev/vda");

/// Refuse to pull an unbounded device into memory. The gate's images are tiny; this is the
/// 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;

/// The one mutation this device accepts so far. The coordinate interface's opcodes will be
/// the language's verbs; this is the first of them.
const OP_PUT: u8 = 1;

/// FNV-1a offset basis.
const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;

/// A line of output, built in place. No allocation: the read path formats into a fixed
/// buffer, so it cannot fail for want of memory while holding a file open.
struct Line {
    buf: [u8; 256],
    len: usize,
}

impl Line {
    fn new() -> Self {
        Line {
            buf: [0; 256],
            len: 0,
        }
    }

    fn as_bytes(&self) -> &[u8] {
        &self.buf[..self.len]
    }
}

impl Write for Line {
    fn write_str(&mut self, s: &str) -> fmt::Result {
        let bytes = s.as_bytes();
        if self.len + bytes.len() > self.buf.len() {
            return Err(fmt::Error);
        }
        self.buf[self.len..self.len + bytes.len()].copy_from_slice(bytes);
        self.len += bytes.len();
        Ok(())
    }
}

/// FNV-1a 64 over a run of bytes, continuing from `h`.
fn fnv1a64(bytes: &[u8], mut h: u64) -> u64 {
    for b in bytes {
        h ^= *b as u64;
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    h
}

/// Read the whole store image through the kernel's file layer.
///
/// `kernel_read` on the block device rather than a raw bio: it is the path this kernel
/// version exposes to Rust, it goes through the page cache the way any other read does, and
/// it needs no C helper. If the store ever has to be read before the VFS is up (a root
/// filesystem, say), that is the moment to reach for the block layer directly.
fn read_image(image: &mut KVVec<u8>) -> Result<()> {
    // O_RDONLY is 0 in Linux; filp_open takes the raw flags word.
    // SAFETY: STORE_DEVICE is a NUL-terminated C string literal, and filp_open either
    // returns a valid `struct file *` or an error pointer, which is checked below.
    let file = unsafe { bindings::filp_open(STORE_DEVICE.as_ptr().cast::<u8>(), 0, 0) };
    let file = kernel::error::from_err_ptr(file)?;
    if file.is_null() {
        return Err(EINVAL);
    }

    let mut pos: bindings::loff_t = 0;
    let mut chunk = KVVec::<u8>::with_capacity(4096, GFP_KERNEL)?;
    chunk.extend_from_slice(&[0u8; 4096][..], GFP_KERNEL)?;
    let mut result = Ok(());

    loop {
        // SAFETY: `file` is a live `struct file *` from filp_open; `chunk` is a 4096-byte
        // kernel buffer we own; `pos` is a valid loff_t. kernel_read reads at most
        // `chunk.len()` bytes into the buffer and does not retain either pointer.
        let n = unsafe {
            bindings::kernel_read(
                file,
                chunk.as_mut_ptr().cast::<core::ffi::c_void>(),
                chunk.len(),
                &mut pos,
            )
        };
        if n < 0 {
            result = Err(Error::from_errno(n as i32));
            break;
        }
        if n == 0 {
            break; // end of device
        }
        let n = n as usize;
        if let Err(e) = image.extend_from_slice(&chunk[..n], GFP_KERNEL) {
            result = Err(e.into());
            break;
        }
        if image.len() >= MAX_BYTES {
            break;
        }
    }

    // SAFETY: `file` came from filp_open and has not been closed; the owner argument is
    // only meaningful for locks that no one holds here.
    unsafe { bindings::filp_close(file, core::ptr::null_mut()) };

    // Less than the shortest header is not an image at all; `digest` reports the rest.
    if result.is_ok() && image.len() < HEADER_LEN_V1 {
        return Err(EINVAL);
    }
    result
}

/// Walk the records of a store image, digesting each one, exactly as
/// `cube-store-raw::iter_records` + `cube-image digest` do.
///
/// The two rules that are easy to get wrong, both from the userspace parser: a record whose
/// value runs past the buffer is a truncated record and an error, and an all-zero frame is
/// *end of records* only when every remaining byte is also zero — otherwise a real record
/// at the origin, followed by padding, would be read as an empty one.
/// What a header says, in both formats.
struct Header {
    version: u8,
    curve: u8,
    image_bytes: Option<u64>,
    record_count: Option<u64>,
}

/// Read the header, or say what is wrong with it.
fn parse_header(image: &[u8]) -> Result<Header, &'static str> {
    if image.len() < 5 || &image[0..4] != MAGIC {
        return Err("not-a-store");
    }
    let version = image[4];
    let curve = image[5];
    match version {
        VERSION_V1 => Ok(Header {
            version,
            curve,
            image_bytes: None,
            record_count: None,
        }),
        VERSION => {
            if image.len() < HEADER_LEN_V2 {
                return Err("short-v2-header");
            }
            let mut word = [0u8; 8];
            word.copy_from_slice(&image[6..14]);
            let image_bytes = u64::from_le_bytes(word);
            word.copy_from_slice(&image[14..22]);
            let record_count = u64::from_le_bytes(word);
            // A header that does not cover itself is corrupt, and guessing at the extent
            // of an image means possibly reading somebody else's bytes.
            if (image_bytes as usize) < HEADER_LEN_V2 {
                return Err("bad-extent");
            }
            Ok(Header {
                version,
                curve,
                image_bytes: Some(image_bytes),
                record_count: Some(record_count),
            })
        }
        _ => Err("unsupported-version"),
    }
}

fn digest(image: &[u8]) -> Line {
    let mut line = Line::new();

    let header = match parse_header(image) {
        Ok(h) => h,
        Err(what) => {
            let _ = write!(line, "error={what}");
            return line;
        }
    };
    let end = match header.image_bytes {
        Some(n) => core::cmp::min(n as usize, image.len()),
        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;
    let mut errors: u64 = 0;
    let mut off = if header.version == VERSION_V1 {
        HEADER_LEN_V1
    } else {
        HEADER_LEN_V2
    };
    let mut remaining = header.record_count;

    while off + RECORD_FIXED <= end {
        if remaining == Some(0) {
            break;
        }
        let frame = &image[off..off + RECORD_FIXED];
        // Padding is not a record — but only if it is padding all the way down, and only
        // in v1, where nothing else says where the records stop. A v2 image states its
        // count, so a zero frame inside that count is a record like any other.
        if remaining.is_none()
            && frame.iter().all(|b| *b == 0)
            && image[off..].iter().all(|b| *b == 0)
        {
            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 len_bytes = [0u8; 8];
        len_bytes.copy_from_slice(&image[off + SPACE_ID_LEN + RAW_KEY_LEN..off + RECORD_FIXED]);
        let value_len = u64::from_le_bytes(len_bytes) as usize;

        let value_at = off + RECORD_FIXED;
        // A promised record that is not there is a truncated image, and saying so beats
        // returning a shorter list that looks complete.
        if value_at + value_len > end {
            errors += 1;
            break;
        }
        let value = &image[value_at..value_at + value_len];

        h = fnv1a64(space, h);
        h = fnv1a64(key, h);
        h = fnv1a64(&(value_len as u64).to_le_bytes(), h);
        h = fnv1a64(value, h);
        count += 1;
        value_bytes += value_len as u64;
        off = value_at + value_len;
        if let Some(n) = remaining.as_mut() {
            *n -= 1;
        }
    }
    // The count was a promise; not meeting it is an error the caller must see.
    if let Some(n) = remaining {
        if n > 0 {
            errors += 1;
        }
    }

    // Field order matches cube-image's `digest` line so the two diff directly.
    let _ = write!(
        line,
        "digest version={} curve={} bytes={} records={} value_bytes={} fnv1a64={:016x} errors={}",
        header.version,
        header.curve,
        image.len(),
        count,
        value_bytes,
        h,
        errors
    );
    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)
}

/// Morton-encode a point into the store's 24-byte key — the same interleaving
/// `cube-core`'s `Curve for Morton` does, because a key written here has to be the key a
/// userspace reader decodes. Bit `i` of each axis lands at bit `3i` of the 192-bit key,
/// counting from the least significant bit, which lives in the *last* byte.
fn morton_encode(x: u64, y: u64, z: u64) -> [u8; 24] {
    let mut k = [0u8; 24];
    let put = |n: usize, v: bool, k: &mut [u8; 24]| {
        if !v {
            return;
        }
        let byte = 23 - (n / 8);
        k[byte] |= 1 << (n % 8);
    };
    for i in 0..64 {
        put(3 * i, (x >> i) & 1 != 0, &mut k);
        put(3 * i + 1, (y >> i) & 1 != 0, &mut k);
        put(3 * i + 2, (z >> i) & 1 != 0, &mut k);
    }
    k
}

/// Where the log region starts, given the image's declared extent.
fn log_offset(image_bytes: u64) -> usize {
    let n = image_bytes as usize;
    (n + LOG_ALIGN - 1) & !(LOG_ALIGN - 1)
}

/// Length of the log's valid prefix: the header plus every entry that parses and passes its
/// checksum. This is the append position, and the reason a torn tail is overwritten rather
/// than appended to — the same rule the userspace log uses.
fn log_valid_len(log: &[u8]) -> Result<usize, &'static str> {
    if log.len() < WAL_HEADER_LEN {
        return Err("short-log");
    }
    if &log[0..4] != WAL_MAGIC {
        return Err("no-log");
    }
    if log[4] != WAL_VERSION {
        return Err("unsupported-log-version");
    }
    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);
        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;
        }
        if crc32(&log[start + 5..off + len]) != crc {
            break;
        }
        off += len;
    }
    Ok(off)
}

/// A mutation to append: the byte-plane write the store contract calls `put`.
struct Mutation {
    space: [u8; 32],
    key: [u8; 24],
    value: KVVec<u8>,
}

/// Append a mutation to the log and make it durable.
///
/// The order is the durability contract: the entry is written and flushed to the device
/// *before* the caller is told it happened. A mutation this function has returned `Ok` for
/// survives a power cut; one it has not may or may not, which is exactly the boundary an
/// acknowledgement is supposed to mark.
fn append(image_bytes: u64, m: &Mutation) -> Result<(), Error> {
    let log_off = log_offset(image_bytes);

    // SAFETY: STORE_DEVICE is a NUL-terminated literal; filp_open returns a valid file or an
    // error pointer, which is checked. O_RDWR is 2.
    let file = unsafe { bindings::filp_open(STORE_DEVICE.as_ptr().cast::<u8>(), 2, 0) };
    let file = kernel::error::from_err_ptr(file)?;
    if file.is_null() {
        return Err(EINVAL);
    }

    let mut result = Ok(());
    // The region's header and existing entries, so the append lands after the valid prefix.
    let mut head = [0u8; 4096];
    let mut pos: bindings::loff_t = log_off as bindings::loff_t;
    // SAFETY: `file` is live; `head` is a 4 KiB buffer we own; `pos` is a valid loff_t.
    let n = unsafe {
        bindings::kernel_read(
            file,
            head.as_mut_ptr().cast::<core::ffi::c_void>(),
            head.len(),
            &mut pos,
        )
    };
    if n < 0 {
        result = Err(Error::from_errno(n as i32));
    }

    if result.is_ok() {
        // A log region that has never been written is zeros, not a log. Give it a header
        // first — the same thing the userspace log does when its file does not exist.
        let region = &head[..n as usize];
        let valid = if region.len() < WAL_HEADER_LEN || &region[0..4] != WAL_MAGIC {
            let mut hdr = [0u8; WAL_HEADER_LEN];
            hdr[0..4].copy_from_slice(&WAL_MAGIC);
            hdr[4] = WAL_VERSION;
            hdr[5] = 0; // morton
            let mut at: bindings::loff_t = log_off as bindings::loff_t;
            // SAFETY: `file` is live and writable; `hdr` outlives the call; `at` is valid.
            let wrote = unsafe {
                bindings::kernel_write(
                    file,
                    hdr.as_ptr().cast::<core::ffi::c_void>(),
                    hdr.len(),
                    &mut at,
                )
            };
            if wrote < 0 {
                result = Err(Error::from_errno(wrote as i32));
            }
            WAL_HEADER_LEN
        } else {
            match log_valid_len(region) {
                Ok(v) => v,
                Err(what) => {
                    pr_err!("cubelinux: log region unreadable: {}\n", what);
                    result = Err(EINVAL);
                    WAL_HEADER_LEN
                }
            }
        };
        if result.is_ok() {
        match Ok::<usize, &'static str>(valid) {
            Ok(valid) => {
                // Build the entry: op | crc | space | key | len | value.
                let total = ENTRY_FIXED + m.value.len();
                let mut entry = KVVec::<u8>::with_capacity(total, GFP_KERNEL)?;
                entry.extend_from_slice(&[1u8], GFP_KERNEL)?;
                entry.extend_from_slice(&[0u8; 4][..], GFP_KERNEL)?;
                entry.extend_from_slice(&m.space, GFP_KERNEL)?;
                entry.extend_from_slice(&m.key, GFP_KERNEL)?;
                entry.extend_from_slice(&(m.value.len() as u32).to_le_bytes(), GFP_KERNEL)?;
                entry.extend_from_slice(m.value.as_slice(), GFP_KERNEL)?;
                // The checksum covers everything after the crc field.
                let crc = crc32(&entry.as_slice()[5..]);
                entry.as_mut_slice()[1..5].copy_from_slice(&crc.to_le_bytes());

                let mut at: bindings::loff_t = (log_off + valid) as bindings::loff_t;
                // SAFETY: `file` is live and opened for writing; `entry` is a kernel buffer
                // that outlives the call; `at` is a valid loff_t.
                let written = unsafe {
                    bindings::kernel_write(
                        file,
                        entry.as_slice().as_ptr().cast::<core::ffi::c_void>(),
                        entry.len(),
                        &mut at,
                    )
                };
                if written < 0 {
                    result = Err(Error::from_errno(written as i32));
                } else if written as usize != entry.len() {
                    result = Err(EIO);
                } else {
                    // Durability before acknowledgement: a short write that was never
                    // flushed is a write that did not happen.
                    // SAFETY: `file` is live and writable; datasync is 0 (full sync).
                    let rc = unsafe { bindings::vfs_fsync(file, 0) };
                    if rc < 0 {
                        result = Err(Error::from_errno(rc));
                    }
                }
            }
            Err(what) => {
                pr_err!("cubelinux: log region unreadable: {}\n", what);
                result = Err(EINVAL);
            }
        }
        }
    }

    // SAFETY: `file` came from filp_open and has not been closed.
    unsafe { bindings::filp_close(file, core::ptr::null_mut()) };
    result
}

/// The module's registration; holds the misc device for as long as the module lives.
#[pin_data]
struct CubeStoreModule {
    #[pin]
    _miscdev: MiscDeviceRegistration<CubeStore>,
}

impl kernel::InPlaceModule for CubeStoreModule {
    fn init(_module: &'static ThisModule) -> impl PinInit<Self, Error> {
        pr_info!("cubelinux: store reader registered at /dev/cubelinux\n");
        try_pin_init!(Self {
            _miscdev <- MiscDeviceRegistration::register(MiscDeviceOptions {
                name: c_str!("cubelinux"),
            }),
        })
    }
}

#[pin_data]
struct CubeStore {
    dev: ARef<Device>,
}

#[vtable]
impl MiscDevice for CubeStore {
    type Ptr = Pin<KBox<Self>>;

    fn open(_file: &File, misc: &MiscDeviceRegistration<Self>) -> Result<Pin<KBox<Self>>> {
        KBox::try_pin_init(
            try_pin_init! {
                CubeStore { dev: ARef::from(misc.device()) }
            },
            GFP_KERNEL,
        )
    }

    /// Write one mutation, framed as the argument block the coordinate interface would
    /// pass: `op(1) | space(32) | x(8) | y(8) | z(8) | len(4) | value[len]`, little-endian.
    ///
    /// This is the byte plane — a `put` of bytes at a coordinate, with no header written
    /// beside them — which is what the store contract defines and what the differential
    /// gate compares against userspace's `cell put`. The header tier sits above this.
    fn write_iter(kiocb: Kiocb<'_, Self::Ptr>, iov: &mut IovIterSource<'_>) -> Result<usize> {
        let mut buf = KVVec::<u8>::new();
        let len = iov.copy_from_iter_vec(&mut buf, GFP_KERNEL)?;
        let bytes = buf.as_slice();
        if bytes.len() < 1 + 32 + 24 + 4 {
            return Err(EINVAL);
        }
        if bytes[0] != OP_PUT {
            return Err(EINVAL);
        }
        let mut space = [0u8; 32];
        space.copy_from_slice(&bytes[1..33]);
        let axis = |at: usize| -> u64 {
            let mut w = [0u8; 8];
            w.copy_from_slice(&bytes[at..at + 8]);
            u64::from_le_bytes(w)
        };
        let (x, y, z) = (axis(33), axis(41), axis(49));
        let mut w = [0u8; 4];
        w.copy_from_slice(&bytes[57..61]);
        let value_len = u32::from_le_bytes(w) as usize;
        if bytes.len() < 61 + value_len {
            return Err(EINVAL);
        }

        let mut value = KVVec::<u8>::new();
        value.extend_from_slice(&bytes[61..61 + value_len], GFP_KERNEL)?;
        let mutation = Mutation {
            space,
            key: morton_encode(x, y, z),
            value,
        };

        // The image's extent says where the log begins. Reading it is not optional: the log
        // has no fixed place of its own, on purpose — no superblock to keep in step.
        let mut head = KVVec::<u8>::new();
        read_image(&mut head)?;
        let header = match parse_header(&head) {
            Ok(h) => h,
            Err(what) => {
                pr_err!("cubelinux: cannot append: {}\n", what);
                return Err(EINVAL);
            }
        };
        let image_bytes = match header.image_bytes {
            Some(n) => n,
            None => {
                pr_err!("cubelinux: cannot append to a v1 store (no extent for the log)\n");
                return Err(EINVAL);
            }
        };

        let me = kiocb.file();
        match append(image_bytes, &mutation) {
            Ok(()) => {
                dev_info!(
                    me.dev,
                    "cubelinux: appended {} bytes at the log ({} value bytes)\n",
                    ENTRY_FIXED + value_len,
                    value_len
                );
                Ok(len)
            }
            Err(e) => {
                dev_err!(me.dev, "cubelinux: append failed: {:?}\n", e);
                Err(e)
            }
        }
    }

    fn read_iter(mut kiocb: Kiocb<'_, Self::Ptr>, iov: &mut IovIterDest<'_>) -> Result<usize> {
        // The work happens here, on demand, rather than at init: by the time anything can
        // open this device, the block driver that provides the store is certainly up.
        let mut image = KVVec::<u8>::new();
        let line = match read_image(&mut image) {
            Ok(()) => digest(&image),
            Err(e) => {
                let mut line = Line::new();
                let _ = write!(line, "error={:?}", e);
                line
            }
        };
        pr_info!("cubelinux: {}\n", core::str::from_utf8(line.as_bytes()).unwrap_or("<bad utf8>"));

        let me = kiocb.file();
        dev_info!(me.dev, "cubelinux: store read complete\n");
        iov.simple_read_from_buffer(kiocb.ki_pos_mut(), line.as_bytes())
    }
}