alkblobs 0.1.0

//! The `mem` kv-tier engine (ADR-011 §2): a `BTreeMap`-shaped ephemeral
//! implementation of the kv contract — the crate's first real engine and
//! the **contract-reference engine**: every other engine's conformance
//! column mirrors its suite rows (store-api.md invariant 7). Feature `mem`,
//! default-on (the 2026-10-15 amendment, ADR-013 §4); never fleet-valid
//! (ADR-011 §4), so no
//! [`gc`][crate::gc] engine-state implementation rides it — the SQL
//! engines' companion seam does not apply.
//!
//! The bodies are the kv body families of ADR-010 §1 in their degenerate
//! form: `get` hands out a read cursor that materializes
//! the whole value (in-memory slices; `len()` from the stored value), and
//! `put_staged` hands out the degenerate in-memory
//! stage whose `commit(key)` publishes through the same whole-value CAS
//! row write as `put` — `entry().or_insert()`, first writer wins, a re-put
//! is dedup and never an overwrite. The key resolves at commit (the
//! staged-put seam's wave-2 amendment: the canonical derivation completes
//! at stream end, so the stage's identity before that point is the
//! engine-internal buffer alone). `commit` cannot fail in normal operation
//! (no media can fail here); the test-only hook noted on the
//! `MemStaging` type
//! exercises the failure form's self-discard hygiene, standing in for the
//! media-failure injection later engines need.

use std::collections::BTreeMap;
use std::sync::Arc;

use parking_lot::Mutex;

use crate::backend::{clamped, Backend, ReadCursor, StagingCursor};
use crate::Error;

/// The `mem` engine — an ephemeral CAS table of opaque byte keys/values
/// under a short-held [`parking_lot::Mutex`] (AGENTS convention 3;
/// parking_lot does not poison, so there is no
/// `unwrap_or_else(|e| e.into_inner())` recovery posture to apply).
///
/// Clone shares the map, so a cloned [`Backend`] handle and every
/// [`MemStaging`] issued from either see the same table — the shape
/// `put_staged`'s out-of-band commit requires. The facade's `mem`
/// constructors construct it with `Default`.
#[derive(Clone, Default)]
pub(crate) struct MemBackend {
    rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
    /// Test-only commit-failure injection, standing in for the media
    /// failure the mem tier cannot produce; mirrors the wave-1 test
    /// double's knob so stage-hygiene's failure form is exercised
    /// against the real engine.
    #[cfg(test)]
    fail_commit: bool,
}

impl Backend for MemBackend {
    type Cursor = MemCursor;
    type Staging = MemStaging;

    fn name(&self) -> &'static str {
        "mem"
    }

    async fn has(&self, key: &[u8]) -> Result<bool, Error> {
        Ok(self.rows.lock().contains_key(key))
    }

    async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error> {
        Ok(self
            .rows
            .lock()
            .get(key)
            .cloned()
            .map(|value| MemCursor { value }))
    }

    async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
        self.rows
            .lock()
            .entry(key.to_vec())
            .or_insert_with(|| value.to_vec());
        Ok(())
    }

    async fn put_staged(&self) -> Result<Self::Staging, Error> {
        Ok(MemStaging {
            staged: Vec::new(),
            rows: Arc::clone(&self.rows),
            #[cfg(test)]
            fail_commit: self.fail_commit,
        })
    }

    async fn delete(&self, key: &[u8]) -> Result<(), Error> {
        match self.rows.lock().remove(key) {
            Some(_) => Ok(()),
            None => Err(Error::Missing),
        }
    }

    async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
        Ok(self.rows.lock().keys().cloned().collect())
    }

    async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
        Ok(self.rows.lock().get(key).map(|value| value.len() as u64))
    }
}

/// The materializing read cursor (ADR-010 §1's kv body family): the whole
/// stored value in memory; range reads are in-memory slices with the
/// cursor contract's EOF clamping, and `len()` is the stored value's
/// length — metadata by construction, there is no content fetch to
/// perform.
pub(crate) struct MemCursor {
    value: Vec<u8>,
}

impl ReadCursor for MemCursor {
    fn len(&self) -> u64 {
        self.value.len() as u64
    }

    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        let (start, end) = clamped(offset, len, self.value.len());
        Ok(self.value[start..end].to_vec())
    }
}

/// The degenerate in-memory staging cursor (ADR-010 §1): appends extend
/// the buffer; `commit(key)` publishes the whole staged bytes through the
/// same CAS clause as [`Backend::put`] — under the store-core-resolved
/// canonical key (the staged-put seam's key-at-commit design point),
/// first writer wins, losing the race is dedup (`Ok`), never a failure —
/// and every non-commit path, including the failure form, converges on
/// zero residue. `commit` takes the buffer via `std::mem::take` before
/// the table lock, so a failed publish cannot strand the staged bytes
/// in the map.
pub(crate) struct MemStaging {
    staged: Vec<u8>,
    rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
    /// Test-only injection, shared with the issuing [`MemBackend`] — see
    /// its field note.
    #[cfg(test)]
    fail_commit: bool,
}

impl ReadCursor for MemStaging {
    fn len(&self) -> u64 {
        self.staged.len() as u64
    }

    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        let (start, end) = clamped(offset, len, self.staged.len());
        Ok(self.staged[start..end].to_vec())
    }
}

impl StagingCursor for MemStaging {
    async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
        self.staged.extend_from_slice(chunk);
        Ok(())
    }

    async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
        #[cfg(test)]
        if self.fail_commit {
            self.staged.clear();
            return Err(Error::Io(
                "engine commit failure (mem, injected)".to_owned(),
            ));
        }
        let value = std::mem::take(&mut self.staged);
        self.rows.lock().entry(key.to_vec()).or_insert(value);
        Ok(())
    }

    async fn discard(mut self) {
        self.staged.clear();
    }
}

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

    const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";

    /// The real stored value, observed through the read seam — the
    /// mirror of the test double's `row()` peek.
    async fn stored_value(backend: &MemBackend, key: &[u8]) -> Option<Vec<u8>> {
        let cursor = backend.get(key).await.expect("get succeeds")?;
        let (bytes, _) = cursor.read_range(0, cursor.len()).await.expect("read");
        Some(bytes)
    }

    /// The Backend method contract: the name, the seven methods'
    /// absence/value/error disciplines, CAS first-writer-wins, and
    /// virgin-store reads as no-ops — mirroring the wave-1 double's
    /// `backend_method_contract`, now against the real engine.
    #[tokio::test]
    async fn backend_method_contract() {
        let backend = MemBackend::default();
        let key = b"\x01opaque-key-bytes".to_vec();
        assert_eq!(backend.name(), "mem", "reporting identity");

        assert!(
            !backend.has(&key).await.unwrap(),
            "virgin store: has is a no-op read"
        );
        assert!(
            backend.get(&key).await.unwrap().is_none(),
            "virgin store: get is a no-op read"
        );
        assert_eq!(
            backend.size(&key).await.unwrap(),
            None,
            "virgin store: size is a no-op read"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "virgin store: list is empty"
        );
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "virgin store: delete reports Missing"
        );

        backend.put(&key, CONTENT).await.unwrap();
        assert!(backend.has(&key).await.unwrap(), "put lands");
        assert_eq!(
            backend.size(&key).await.unwrap(),
            Some(CONTENT.len() as u64),
            "size is metadata-only and exact"
        );
        backend.put(&key, b"overwritten").await.unwrap();
        assert_eq!(
            stored_value(&backend, &key).await.as_deref(),
            Some(CONTENT),
            "CAS first-writer wins: re-put is a no-op"
        );
        assert_eq!(
            backend.list().await.unwrap(),
            vec![key.clone()],
            "list is complete, keys not values"
        );

        backend.delete(&key).await.unwrap();
        assert!(!backend.has(&key).await.unwrap(), "delete lands");
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "delete reports Missing when absent"
        );
    }

    /// The `list` contract is complete AND well-formed: every key in the
    /// table, no values leaked, and (BTreeMap ordering) lexicographic —
    /// asserted order-insensitively against the map's keys plus a
    /// determinism check.
    #[tokio::test]
    async fn list_is_complete_and_well_formed() {
        let backend = MemBackend::default();
        let keys = [
            b"\x02b-key".to_vec(),
            b"\x02a-key".to_vec(),
            b"\x02c-key".to_vec(),
        ];
        for key in &keys {
            backend.put(key, b"value").await.unwrap();
        }
        let mut listed = backend.list().await.unwrap();
        let mut expected = keys.to_vec();
        listed.sort();
        expected.sort();
        assert_eq!(listed, expected, "list is complete and well-formed");
        for key in &expected {
            assert!(
                backend.has(key).await.unwrap(),
                "no spurious or mangled keys"
            );
        }
    }

    /// The materialized cursor body's read contract: `len()` exact,
    /// EOF-clamped reads (over-read, tail, at-EOF empty, zero-length),
    /// and the ADR-006 range seam's digest over exactly the returned
    /// bytes — mirroring the double's `cursor_contract_materialized_body`.
    #[tokio::test]
    async fn cursor_contract_materialized_body() {
        let backend = MemBackend::default();
        let key = b"\x01cursor-key".to_vec();
        backend.put(&key, CONTENT).await.unwrap();

        let cursor = backend
            .get(&key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        let len = CONTENT.len() as u64;
        assert_eq!(cursor.len(), len, "len is the entry length");

        assert_eq!(cursor.read_at(0, len).await.unwrap(), CONTENT, "full read");
        assert_eq!(
            cursor.read_at(0, len + 17).await.unwrap(),
            CONTENT,
            "over-read clamps at EOF, pread-style"
        );
        let mid = (CONTENT.len() / 2) as u64;
        assert_eq!(
            cursor.read_at(mid, len).await.unwrap(),
            &CONTENT[mid as usize..],
            "tail read clamps"
        );
        assert_eq!(
            cursor.read_at(len, 5).await.unwrap(),
            Vec::<u8>::new(),
            "read at EOF is empty, never an error"
        );
        assert_eq!(
            cursor.read_at(0, 0).await.unwrap(),
            Vec::<u8>::new(),
            "zero-length read"
        );

        let (slice, digest) = cursor.read_range(mid, 7).await.unwrap();
        assert_eq!(slice, &CONTENT[mid as usize..(mid as usize + 7)]);
        assert_eq!(digest, crate::backend::slice_digest(&slice));
    }

    /// The degenerate case: a zero-byte entry round-trips (put empty,
    /// size exact, cursor body reads empty), and the staged form handles
    /// an empty append-and-commit identically.
    #[tokio::test]
    async fn empty_entry_round_trips() {
        let backend = MemBackend::default();
        let key = b"\x01empty-key".to_vec();

        backend.put(&key, b"").await.unwrap();
        assert!(backend.has(&key).await.unwrap(), "empty value still lands");
        assert_eq!(backend.size(&key).await.unwrap(), Some(0), "exact zero");
        let cursor = backend.get(&key).await.unwrap().expect("entry present");
        assert_eq!(cursor.len(), 0);
        assert_eq!(cursor.read_at(0, 8).await.unwrap(), Vec::<u8>::new());

        let staged_key = b"\x01empty-staged-key".to_vec();
        let mut staging = backend.put_staged().await.unwrap();
        staging.append(b"").await.unwrap();
        assert_eq!(staging.len(), 0, "empty append stays empty");
        staging
            .commit(&staged_key)
            .await
            .expect("empty commit publishes");
        assert!(
            backend.has(&staged_key).await.unwrap(),
            "empty entry landed"
        );
        assert_eq!(backend.size(&staged_key).await.unwrap(), Some(0));
        let committed = backend
            .get(&staged_key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        assert_eq!(
            committed.read_at(0, 8).await.unwrap(),
            Vec::<u8>::new(),
            "committed empty body reads empty"
        );
    }

    /// The staged put's happy path: handout, appends, the verification
    /// pass reading the staged artifact back as a whole through the read
    /// seam, commit publishing through the same CAS clause, and the
    /// committed entry round-tripping through get's cursor — mirroring
    /// the double's `staged_put_round_trips_through_the_read_seam`.
    #[tokio::test]
    async fn staged_put_round_trips_through_the_read_seam() {
        let backend = MemBackend::default();
        let key = b"\x01staged-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(&CONTENT[..8]).await.unwrap();
        staging.append(&CONTENT[8..16]).await.unwrap();
        staging.append(&CONTENT[16..]).await.unwrap();

        assert_eq!(
            staging.len() as usize,
            CONTENT.len(),
            "staged length is the appended length"
        );
        let (bytes, digest) = staging
            .read_range(0, staging.len())
            .await
            .expect("staged bytes readable");
        assert_eq!(bytes, CONTENT, "staged artifact readable as a whole");
        assert_eq!(digest, crate::backend::slice_digest(CONTENT));

        staging.commit(&key).await.expect("commit publishes");
        assert!(backend.has(&key).await.unwrap(), "entry published");

        let cursor = backend
            .get(&key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        let (bytes, digest) = cursor
            .read_range(0, cursor.len())
            .await
            .expect("committed read");
        assert_eq!(bytes, CONTENT);
        assert_eq!(digest, crate::backend::slice_digest(CONTENT));

        // The staged form lands CAS-first-writer too: a subsequent
        // re-put — and a re-staged commit against the live entry — are
        // both dedup no-ops that never overwrite.
        backend.put(&key, b"overwritten").await.unwrap();
        assert_eq!(
            stored_value(&backend, &key).await.as_deref(),
            Some(CONTENT),
            "whole-value re-put does not overwrite"
        );
        let mut second = backend.put_staged().await.unwrap();
        second.append(b"overwritten-by-stage").await.unwrap();
        second
            .commit(&key)
            .await
            .expect("commit is dedup, not failure");
        assert_eq!(
            stored_value(&backend, &key).await.as_deref(),
            Some(CONTENT),
            "staged re-commit does not overwrite"
        );
    }

    /// Stage hygiene, discard form: every non-commit path converges on
    /// discard with zero residue — nothing was staged into the table.
    #[tokio::test]
    async fn staged_discard_leaves_zero_residue() {
        let backend = MemBackend::default();
        let key = b"\x01discard-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(b"residue").await.unwrap();
        staging.discard().await;

        assert!(
            !backend.has(&key).await.unwrap(),
            "discarded stage published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "no entry landed");
        assert!(
            backend.list().await.unwrap().is_empty(),
            "the table holds exactly zero entries after discard"
        );
    }

    /// Stage hygiene, failure form: a failed commit is already
    /// self-discarded — zero residue, entry unpublished, no post-failure
    /// discard left to drive (the consuming call took the cursor). The
    /// injection stands in for the media failure the mem tier cannot
    /// itself produce; the shape holds regardless of the failure's cause.
    #[tokio::test]
    async fn commit_failure_self_discards() {
        let backend = MemBackend {
            fail_commit: true,
            ..Default::default()
        };
        let key = b"\x01failure-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(CONTENT).await.unwrap();
        let err = staging.commit(&key).await.unwrap_err();
        assert!(matches!(err, Error::Io(_)), "engine media failure");

        assert!(
            !backend.has(&key).await.unwrap(),
            "failed commit published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "zero residue");
        assert!(
            backend.list().await.unwrap().is_empty(),
            "zero residue in the table itself"
        );
    }
}