clay-borg/crates/cb-events/src/store.rs

//! K11 durable log framing, and the `LogStore` port (GameKernel §2.4, AM-11).
//!
//! K11: *"Log format is append-only, length-prefixed, versioned; a
//! truncated tail is detected, not silently accepted."* Until CB-WP-0006
//! T05 the only `fs::` call in the workspace was `read_to_string` for
//! scenario YAML — the rule was satisfied by a `Vec`.
//!
//! **Reimplemented, not assimilated** (ADR-0005 §2). "Own the semantics;
//! assimilate the implementation" exists so we do not reimplement *hard*
//! things — SHA-256 and ChaCha cost 12 crates between them and are kept.
//! A length prefix is not in that category, and an embedded store would
//! spend AM-4a headroom on a format that fits in a paragraph.
//!
//! **The operative clause is *detected*.** A reader that accepts a
//! truncated tail is worse than no format at all, because it silently
//! returns a short history that looks complete. Every corruption below has
//! a negative control in the conformance suite.

use std::fs::OpenOptions;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};

/// File header: magic + format version. Present exactly once, at offset 0.
pub const MAGIC: &[u8; 5] = b"CBLOG";
pub const FORMAT_VERSION: u8 = 1;
pub const HEADER_LEN: usize = MAGIC.len() + 1;

/// Sanity cap. A corrupt length prefix is far more likely than a genuine
/// 64 MiB record, and without a cap it would drive an enormous allocation
/// before the truncation check could fire.
pub const MAX_RECORD: u32 = 64 << 20;

#[derive(Debug, PartialEq, Eq)]
pub enum StoreError {
    /// Not one of our logs at all.
    BadMagic,
    /// A format we do not know how to read. Never guess.
    UnsupportedVersion(u8),
    /// K11's operative clause: the tail is short, and we say so.
    TruncatedTail {
        offset: usize,
        need: usize,
        have: usize,
    },
    /// A length prefix beyond the sanity cap — corruption, not a big record.
    RecordTooLarge(u32),
    Io(String),
}

impl std::fmt::Display for StoreError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::BadMagic => write!(f, "not a CBLOG file (bad magic)"),
            Self::UnsupportedVersion(v) => write!(f, "unsupported format version {v}"),
            Self::TruncatedTail { offset, need, have } => write!(
                f,
                "truncated tail at byte {offset}: need {need} bytes, have {have}"
            ),
            Self::RecordTooLarge(n) => {
                write!(f, "record length {n} exceeds the {MAX_RECORD}-byte cap")
            }
            Self::Io(e) => write!(f, "io: {e}"),
        }
    }
}

/// Serialize a header. Written once when a log is created.
#[must_use]
pub fn header() -> Vec<u8> {
    let mut out = MAGIC.to_vec();
    out.push(FORMAT_VERSION);
    out
}

/// Frame one record: 4-byte little-endian length, then the payload.
#[must_use]
pub fn frame(record: &[u8]) -> Vec<u8> {
    let mut out = Vec::with_capacity(4 + record.len());
    out.extend_from_slice(
        &u32::try_from(record.len())
            .unwrap_or(u32::MAX)
            .to_le_bytes(),
    );
    out.extend_from_slice(record);
    out
}

/// Parse a whole log. Shared by every `LogStore` impl, so the truncation
/// contract cannot differ between them.
pub fn parse(bytes: &[u8]) -> Result<Vec<Vec<u8>>, StoreError> {
    if bytes.len() < HEADER_LEN {
        return Err(StoreError::TruncatedTail {
            offset: 0,
            need: HEADER_LEN,
            have: bytes.len(),
        });
    }
    if &bytes[..MAGIC.len()] != MAGIC {
        return Err(StoreError::BadMagic);
    }
    let version = bytes[MAGIC.len()];
    if version != FORMAT_VERSION {
        return Err(StoreError::UnsupportedVersion(version));
    }

    let mut out = Vec::new();
    let mut at = HEADER_LEN;
    while at < bytes.len() {
        let remaining = bytes.len() - at;
        if remaining < 4 {
            return Err(StoreError::TruncatedTail {
                offset: at,
                need: 4,
                have: remaining,
            });
        }
        let len = u32::from_le_bytes([bytes[at], bytes[at + 1], bytes[at + 2], bytes[at + 3]]);
        if len > MAX_RECORD {
            return Err(StoreError::RecordTooLarge(len));
        }
        let len = len as usize;
        at += 4;
        if bytes.len() - at < len {
            return Err(StoreError::TruncatedTail {
                offset: at,
                need: len,
                have: bytes.len() - at,
            });
        }
        out.push(bytes[at..at + len].to_vec());
        at += len;
    }
    Ok(out)
}

/// The capability port (AM-11 / M-D4-SWAP). Two impls, one suite.
pub trait LogStore {
    /// Append one record. Append-only: there is no update or delete.
    fn append(&mut self, record: &[u8]) -> Result<(), StoreError>;

    /// Every record, in append order.
    fn records(&self) -> Result<Vec<Vec<u8>>, StoreError>;

    /// Raw framed bytes. Exists so the conformance suite can corrupt a
    /// log and require the *same* rejection from every impl — a format
    /// contract that only one impl enforces is not a contract.
    fn raw(&self) -> Result<Vec<u8>, StoreError>;

    /// Replace the raw bytes. Test-facing, and the reason truncation is
    /// part of the shared suite rather than a file-only special case.
    fn set_raw(&mut self, bytes: &[u8]) -> Result<(), StoreError>;
}

/// In-memory impl — the fast, non-durable half of the AM-11 pair.
#[derive(Debug)]
pub struct MemLogStore {
    bytes: Vec<u8>,
}

impl Default for MemLogStore {
    fn default() -> Self {
        Self::new()
    }
}

impl MemLogStore {
    #[must_use]
    pub fn new() -> Self {
        Self { bytes: header() }
    }
}

impl LogStore for MemLogStore {
    fn append(&mut self, record: &[u8]) -> Result<(), StoreError> {
        self.bytes.extend_from_slice(&frame(record));
        Ok(())
    }

    fn records(&self) -> Result<Vec<Vec<u8>>, StoreError> {
        parse(&self.bytes)
    }

    fn raw(&self) -> Result<Vec<u8>, StoreError> {
        Ok(self.bytes.clone())
    }

    fn set_raw(&mut self, bytes: &[u8]) -> Result<(), StoreError> {
        self.bytes = bytes.to_vec();
        Ok(())
    }
}

/// File-backed impl — the durable half. Opens in append mode per write, so
/// a crash between writes leaves a well-formed prefix rather than a
/// half-updated file.
#[derive(Debug)]
pub struct FileLogStore {
    path: PathBuf,
}

impl FileLogStore {
    /// Create or open a log at `path`, writing the header if new.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, StoreError> {
        let path = path.as_ref().to_path_buf();
        if !path.exists() {
            std::fs::write(&path, header()).map_err(|e| StoreError::Io(e.to_string()))?;
        }
        Ok(Self { path })
    }
}

impl LogStore for FileLogStore {
    fn append(&mut self, record: &[u8]) -> Result<(), StoreError> {
        let mut f = OpenOptions::new()
            .append(true)
            .open(&self.path)
            .map_err(|e| StoreError::Io(e.to_string()))?;
        f.write_all(&frame(record))
            .map_err(|e| StoreError::Io(e.to_string()))
    }

    fn records(&self) -> Result<Vec<Vec<u8>>, StoreError> {
        parse(&self.raw()?)
    }

    fn raw(&self) -> Result<Vec<u8>, StoreError> {
        let mut buf = Vec::new();
        std::fs::File::open(&self.path)
            .and_then(|mut f| f.read_to_end(&mut buf))
            .map_err(|e| StoreError::Io(e.to_string()))?;
        Ok(buf)
    }

    fn set_raw(&mut self, bytes: &[u8]) -> Result<(), StoreError> {
        std::fs::write(&self.path, bytes).map_err(|e| StoreError::Io(e.to_string()))
    }
}

/// The AM-11 conformance suite: **one** suite, driven by every impl.
///
/// Before this existed, `M-D4-SWAP` was a bool over "impls passing the
/// same conformance suite" and no suite existed — the RNG pair was
/// exercised by two separate, non-shared tests, so `AM-11 | met, narrow`
/// was never earned. ADR-0005 §4 downgraded it to unmet; this is what
/// re-earns it.
///
/// Panics with a labelled message so a failure names the impl.
pub fn conformance<S: LogStore>(label: &str, mut make: impl FnMut() -> S) {
    // A fresh log holds no records but is still well-formed.
    let s = make();
    assert_eq!(
        s.records().unwrap(),
        Vec::<Vec<u8>>::new(),
        "{label}: a new log must read back as zero records"
    );

    // Append order is read order.
    let mut s = make();
    let records: Vec<Vec<u8>> = vec![b"first".to_vec(), b"second".to_vec(), b"third".to_vec()];
    for r in &records {
        s.append(r).unwrap();
    }
    assert_eq!(
        s.records().unwrap(),
        records,
        "{label}: records must read back in append order"
    );

    // Binary payloads, including empty and high bytes, survive intact.
    let mut s = make();
    let tricky: Vec<Vec<u8>> = vec![
        Vec::new(),
        vec![0u8; 8],
        vec![0xFF; 8],
        (0u8..=255).collect(),
    ];
    for r in &tricky {
        s.append(r).unwrap();
    }
    assert_eq!(
        s.records().unwrap(),
        tricky,
        "{label}: binary payloads must round-trip byte-for-byte"
    );

    // --- K11's operative clause: corruption is DETECTED, not accepted ---

    // A tail short by one byte.
    let mut s = make();
    s.append(b"payload").unwrap();
    let good = s.raw().unwrap();
    s.set_raw(&good[..good.len() - 1]).unwrap();
    assert!(
        matches!(s.records(), Err(StoreError::TruncatedTail { .. })),
        "{label}: a tail short by one byte must be rejected, got {:?}",
        s.records()
    );

    // A length prefix cut in half.
    let mut s = make();
    s.append(b"payload").unwrap();
    let good = s.raw().unwrap();
    s.set_raw(&good[..HEADER_LEN + 2]).unwrap();
    assert!(
        matches!(s.records(), Err(StoreError::TruncatedTail { .. })),
        "{label}: a half-written length prefix must be rejected"
    );

    // A corrupted length prefix claiming more than the file holds.
    let mut s = make();
    s.append(b"payload").unwrap();
    let mut bad = s.raw().unwrap();
    bad[HEADER_LEN] = 0xFF;
    bad[HEADER_LEN + 1] = 0xFF;
    s.set_raw(&bad).unwrap();
    assert!(
        matches!(
            s.records(),
            Err(StoreError::TruncatedTail { .. }) | Err(StoreError::RecordTooLarge(_))
        ),
        "{label}: a corrupt length prefix must be rejected"
    );

    // Wrong magic and unknown version are refused rather than guessed at.
    let mut s = make();
    s.set_raw(b"NOPE!\x01").unwrap();
    assert_eq!(
        s.records(),
        Err(StoreError::BadMagic),
        "{label}: foreign bytes must not be read as a log"
    );

    let mut s = make();
    let mut future = header();
    future[MAGIC.len()] = FORMAT_VERSION + 1;
    s.set_raw(&future).unwrap();
    assert_eq!(
        s.records(),
        Err(StoreError::UnsupportedVersion(FORMAT_VERSION + 1)),
        "{label}: a newer format must be refused, not guessed at"
    );
}

#[cfg(test)]
mod tests {
    use super::*;

    /// AM-11: the SAME suite, both impls. This is the whole point — two
    /// separate tests would reproduce the defect ADR-0005 §4 recorded.
    #[test]
    fn mem_store_passes_the_conformance_suite() {
        conformance("MemLogStore", MemLogStore::new);
    }

    #[test]
    fn file_store_passes_the_conformance_suite() {
        let dir = std::env::temp_dir().join(format!("cb-log-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let mut n = 0;
        conformance("FileLogStore", || {
            n += 1;
            let p = dir.join(format!("{n}.cblog"));
            let _ = std::fs::remove_file(&p);
            FileLogStore::open(&p).unwrap()
        });
        std::fs::remove_dir_all(&dir).ok();
    }

    /// K11 framing is shared, so it is also pinned directly.
    #[test]
    fn framing_round_trips_and_rejects_a_truncated_tail() {
        let mut bytes = header();
        bytes.extend_from_slice(&frame(b"abc"));
        assert_eq!(parse(&bytes).unwrap(), vec![b"abc".to_vec()]);
        assert!(matches!(
            parse(&bytes[..bytes.len() - 1]),
            Err(StoreError::TruncatedTail { .. })
        ));
    }
}