blockworx/store/

revs.rs

//! `revs/`: the whole document at every rev — the document itself, since
//! P5 (`docs/log-vs-snapshot.md` S3/S4).
//!
//! One file per rev the manifest holds — `revs/000012.json.zst`, zstd −1
//! over the compact `serde_json` bytes of the document at that rev
//! (§14.2). Rev 0 is the empty document and is written nothing, so a
//! container with nothing in it holds an empty directory.
//!
//! **A rev carries no payloads.** `assets/` is the one home for bytes, so
//! a 1 MB PNG is written once under its own hash rather than into every
//! rev that stands after it; a rev file goes out stripped
//! ([`Document::without_assets`]) and comes back with everything it
//! *references* re-attached. `document.json`, the self-contained export
//! form, keeps embedding its payloads.
//!
//! The name and the encoding live here alone, over a [`Backing`] that is
//! a directory for a container and a map for a session with no files — so
//! a scratch session steps exactly as a container does, and the store, the
//! fsck, Save-as and the share bundle cannot spell a rev differently from
//! each other.

use blockworx_doc::{document::Document, rev::Rev};

use super::assets::{AssetFault, AssetSource, hydrate};
use super::record::Digest;

/// The directory a container keeps its rev files in.
pub const REVS: &str = "revs";

/// Where a session keeps the document at every rev: a directory, or a map
/// that dies with the process. The same bytes either way.
pub trait Backing {
    /// # Errors
    /// The write that did not land.
    fn put(&mut self, at: Rev, bytes: &[u8]) -> std::io::Result<()>;

    /// The bytes rev `at` was written as, or `None` where nothing was.
    ///
    /// # Errors
    /// The read that did not work — which is not the same as nothing
    /// being there.
    fn get(&self, at: Rev) -> std::io::Result<Option<Vec<u8>>>;

    /// What a report calls rev `at`'s home.
    fn names(&self, at: Rev) -> String;
}

/// A session with no files: the same compressed bytes, in a map.
///
/// Bounded by what the bytes cost rather than by what the documents do —
/// 121 KB per rev at 2,500 blocks (`TUNING.md`, Finding 8) — which is
/// what makes stepping a scratch session a decode rather than a fold of
/// its whole log prefix.
#[derive(Default, Debug)]
pub struct Memory(std::collections::BTreeMap<Rev, Vec<u8>>);

impl Backing for Memory {
    fn put(&mut self, at: Rev, bytes: &[u8]) -> std::io::Result<()> {
        self.0.insert(at, bytes.to_vec());
        Ok(())
    }

    fn get(&self, at: Rev) -> std::io::Result<Option<Vec<u8>>> {
        Ok(self.0.get(&at).cloned())
    }

    fn names(&self, at: Rev) -> String {
        format!("rev {} of this session", at.get())
    }
}

/// What went wrong between a session's revs and the history that names
/// them.
#[derive(Debug, thiserror::Error)]
pub enum RevFault {
    #[error("rev {} was never written to {where_it_would_be}", .at.get())]
    Missing { at: Rev, where_it_would_be: String },
    #[error("rev {}'s copy could not be read: {why}", .at.get())]
    Unreadable { at: Rev, why: std::io::Error },
    #[error("rev {}'s copy is not a document this build reads: {why}", .at.get())]
    NotADocument { at: Rev, why: serde_json::Error },
    #[error("rev {}'s copy holds bytes that hash to {found}, not {named}", .at.get())]
    Tampered {
        at: Rev,
        named: Digest,
        found: Digest,
    },
    #[error(transparent)]
    Artwork(#[from] AssetFault),
    #[error("rev {} could not be written: {why}", .at.get())]
    NotWritten { at: Rev, why: std::io::Error },
}

/// Write `document` as rev `at` and hand back the digest of the bytes that
/// landed — which is what the manifest row stamps, and the only state
/// digest this format has.
///
/// The payloads go to `assets/` first: a rev may not name bytes the store
/// does not already hold, and every payload the document *holds* is put
/// there, not only the ones it currently references, so a payload outlives
/// the reference that brought it in.
///
/// # Errors
/// The write that did not land.
pub fn write(
    into: &mut dyn Backing,
    at: Rev,
    document: &Document,
    payloads: &mut dyn super::assets::AssetSink,
) -> std::io::Result<Digest> {
    for (hash, asset) in document.assets() {
        payloads.put(hash, asset)?;
    }
    let bytes = encode(&document.without_assets())?;
    into.put(at, &bytes)?;
    Ok(Digest::of(&bytes))
}

/// The document rev `at` holds, as the file holds it: no payloads.
/// [`attached`] is what puts them back.
///
/// Rev 0 is the empty document and was never written.
///
/// # Errors
/// [`RevFault`]: nothing there, bytes that will not decompress, or bytes
/// that are not a document this build reads.
pub fn read(from: &dyn Backing, at: Rev) -> Result<Document, RevFault> {
    if at == Rev::ZERO {
        return Ok(Document::default());
    }
    let bytes = bytes(from, at)?;
    let json = unpack(&bytes).map_err(|why| RevFault::Unreadable { at, why })?;
    let mut document: Document =
        serde_json::from_slice(&json).map_err(|why| RevFault::NotADocument { at, why })?;
    document.positioned_at(at);
    Ok(document)
}

/// Put back every payload `document` references, out of the one place
/// bytes live.
///
/// # Errors
/// [`AssetFault`] for a payload the store cannot hand back, or hands back
/// under a hash that is not its own.
pub fn attached(mut document: Document, assets: &dyn AssetSource) -> Result<Document, AssetFault> {
    document.attach_assets(|hash| hydrate(assets, hash))?;
    Ok(document)
}

/// The digest of the bytes rev `at` holds — what a manifest row's `hash`
/// is checked against.
///
/// # Errors
/// [`RevFault`] for a rev whose bytes are not there to be hashed.
pub fn stamp(from: &dyn Backing, at: Rev) -> Result<Digest, RevFault> {
    // The empty document is written nothing, so rev 0 stamps the nothing
    // that was written.
    if at == Rev::ZERO {
        return Ok(Digest::of(&[]));
    }
    Ok(Digest::of(&bytes(from, at)?))
}

/// The bytes rev `at` holds, or the fault that says why they are not
/// there.
///
/// # Errors
/// [`RevFault`] naming the rev.
pub fn bytes(from: &dyn Backing, at: Rev) -> Result<Vec<u8>, RevFault> {
    from.get(at)
        .map_err(|why| RevFault::Unreadable { at, why })?
        .ok_or_else(|| RevFault::Missing {
            at,
            where_it_would_be: from.names(at),
        })
}

/// Whether rev `at` holds the bytes `named` says it does — the fsck's
/// question, and the one a writable open asks of its head.
///
/// # Errors
/// [`RevFault`] for a rev that is missing, unreadable, or not what it is
/// named for.
pub fn witnessed(from: &dyn Backing, at: Rev, named: Digest) -> Result<(), RevFault> {
    let found = stamp(from, at)?;
    if found == named {
        Ok(())
    } else {
        Err(RevFault::Tampered { at, named, found })
    }
}

/// Every rev a history through `at` holds, oldest first. Rev 0 is the
/// empty document and was never written, so counting starts at one.
pub fn through(at: Rev) -> impl Iterator<Item = Rev> {
    std::iter::successors(Some(Rev::ZERO), |rev| Some(rev.next()))
        .skip(1)
        .take_while(move |rev| *rev <= at)
}

/// # Errors
/// The serialization or compression that did not work.
fn encode(document: &Document) -> std::io::Result<Vec<u8>> {
    let json = serde_json::to_vec(document).map_err(std::io::Error::other)?;
    pack(&json)
}

/// §14.2: measured at 30 ms for 13.3 MB, where −9 saves a fifth of the
/// bytes for five times the time.
#[cfg(not(target_arch = "wasm32"))]
const LEVEL: i32 = 1;

#[cfg(not(target_arch = "wasm32"))]
fn pack(json: &[u8]) -> std::io::Result<Vec<u8>> {
    zstd::encode_all(json, LEVEL)
}

#[cfg(not(target_arch = "wasm32"))]
fn unpack(bytes: &[u8]) -> std::io::Result<Vec<u8>> {
    zstd::decode_all(bytes)
}

/// The browser has no zstd (the container it would compress for is
/// native-only), so there a rev is its own JSON. Nothing reads these
/// bytes but the session that wrote them, so the two spellings never meet.
#[cfg(target_arch = "wasm32")]
#[expect(clippy::unnecessary_wraps, reason = "the native pair can fail")]
fn pack(json: &[u8]) -> std::io::Result<Vec<u8>> {
    Ok(json.to_vec())
}

#[cfg(target_arch = "wasm32")]
#[expect(clippy::unnecessary_wraps, reason = "the native pair can fail")]
fn unpack(bytes: &[u8]) -> std::io::Result<Vec<u8>> {
    Ok(bytes.to_vec())
}

#[cfg(not(target_arch = "wasm32"))]
pub use native::{Dir, path};

#[cfg(not(target_arch = "wasm32"))]
mod native {
    use super::{Backing, REVS, Rev};
    use std::path::{Path, PathBuf};

    /// Where rev `at` is written. Zero-padded so the directory sorts as
    /// the history reads.
    pub fn path(root: &Path, at: Rev) -> PathBuf {
        root.join(REVS).join(format!("{:06}.json.zst", at.get()))
    }

    /// A container's `revs/` directory.
    pub struct Dir(PathBuf);

    impl Dir {
        pub fn at(root: &Path) -> Self {
            Self(root.to_path_buf())
        }
    }

    impl Backing for Dir {
        fn put(&mut self, at: Rev, bytes: &[u8]) -> std::io::Result<()> {
            // The directory is laid down at creation; a container whose
            // `revs/` somebody removed is repaired by the write that
            // needs it.
            std::fs::create_dir_all(self.0.join(REVS))?;
            crate::atomic::write_atomically(&path(&self.0, at), bytes)
        }

        fn get(&self, at: Rev) -> std::io::Result<Option<Vec<u8>>> {
            match std::fs::read(path(&self.0, at)) {
                Ok(bytes) => Ok(Some(bytes)),
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
                Err(error) => Err(error),
            }
        }

        fn names(&self, at: Rev) -> String {
            path(&self.0, at).display().to_string()
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::store::assets::Held;
    use crate::store::tests::fixture;
    use blockworx_doc::fixtures::rev;

    /// The encoding is a round trip, and a rev's own digest is the digest
    /// of the bytes that landed.
    #[test]
    fn a_rev_round_trips_and_stamps_the_bytes_it_was_written_as() {
        let document = fixture::documents(2).pop().expect("two documents");
        let mut backing = Memory::default();
        let mut payloads = Held::default();
        let stamped = write(&mut backing, rev(1), &document, &mut payloads).expect("the rev lands");

        assert_eq!(
            stamp(&backing, rev(1)).expect("the bytes are there"),
            stamped
        );
        assert_eq!(read(&backing, rev(1)).expect("it reads back"), document);
        witnessed(&backing, rev(1), stamped).expect("the rev is what it says");
    }

    /// Rev 0 was never written, and says so rather than being invented.
    #[test]
    fn rev_zero_is_the_empty_document_and_holds_no_bytes() {
        let backing = Memory::default();
        assert_eq!(
            read(&backing, Rev::ZERO).expect("the empty document"),
            Document::default(),
        );
        assert!(matches!(
            read(&backing, rev(4)),
            Err(RevFault::Missing { .. }),
        ));
    }

    /// The payload rule: a rev file holds none, `assets/` holds them all,
    /// and reading a rev puts back exactly the ones it references.
    #[test]
    fn a_rev_file_holds_no_payloads_and_reads_back_the_ones_it_references() {
        let asset = fixture::svg(1);
        let document = fixture::with_icon(&asset);
        assert_eq!(document.assets().count(), 1, "precondition: it holds one");

        let mut backing = Memory::default();
        let mut payloads = Held::default();
        write(&mut backing, rev(1), &document, &mut payloads).expect("the rev lands");

        let stripped = read(&backing, rev(1)).expect("it reads back");
        assert_eq!(stripped.assets().count(), 0, "the rev carries bytes");
        assert_eq!(
            attached(stripped, &payloads).expect("the payload is in the store"),
            document,
        );
    }

    /// Tampering with a rev's bytes is caught by the hash its row stamps,
    /// which is the whole of what replaced D12's folded-state stamp.
    #[test]
    fn bytes_that_are_not_what_the_row_names_are_refused() {
        let document = fixture::documents(1).pop().expect("one document");
        let mut backing = Memory::default();
        let mut payloads = Held::default();
        let stamped = write(&mut backing, rev(1), &document, &mut payloads).expect("the rev lands");

        backing
            .put(rev(1), b"not a document")
            .expect("it is rewritten");
        assert!(matches!(
            witnessed(&backing, rev(1), stamped),
            Err(RevFault::Tampered { .. }),
        ));
        assert!(matches!(
            read(&backing, rev(1)),
            Err(RevFault::Unreadable { .. } | RevFault::NotADocument { .. }),
        ));
    }

    #[test]
    fn the_revs_through_a_head_start_at_one() {
        assert_eq!(
            through(rev(3)).collect::<Vec<_>>(),
            [rev(1), rev(2), rev(3)]
        );
        assert_eq!(through(Rev::ZERO).count(), 0);
    }
}