blockworx_store/

migrate.rs

//! `blockworx migrate`: a container written when revs were zstd, read by a
//! build that writes them as gzip.
//!
//! One compressor everywhere is what lets a container move between a
//! browser and a desktop untranslated, and zstd's C does not build for
//! `wasm32-unknown-unknown` — so the format changed, and the containers
//! already on the developer's disk are carried over rather than left
//! behind. This module is the last thing in the tree that names zstd and
//! goes with the release after the one that lands it.
//!
//! **The manifest is rewritten, because it has to be.** A row stamps the
//! digest of the bytes the rev file holds, so re-compressing a rev changes
//! what its row must say, and every row after it links to that row. Each
//! row keeps every column it carried — the wall time, the author, the
//! camera, the kind, the label — and gets a new `hash` and a new `parent`,
//! which is exactly the chain the same rows would have had if this build
//! had written them. The file goes down whole or not at all.

use blockworx_doc::rev::Rev;

use crate::container::MANIFEST;
use crate::manifest::{self, End, Row};
use crate::record::Digest;
use crate::revs::{REVS, entry, pack};
use crate::storage::{Entry, Storage, ready_now};

/// What a rev file was called before the encoding changed.
fn zstd_entry(at: Rev) -> Entry {
    Entry::under(REVS, &format!("{:06}.json.zst", at.get()))
}

/// What the migration did.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Migrated {
    /// Every rev was already gzip and every row already stamped it, so
    /// nothing was written. Running the migration twice is running it
    /// once.
    AlreadyDone,
    Rewrote {
        revs: usize,
    },
}

impl std::fmt::Display for Migrated {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Migrated::AlreadyDone => write!(f, "already gzip: nothing to do"),
            Migrated::Rewrote { revs } => {
                write!(
                    f,
                    "{revs} revs rewritten as gzip, and the manifest re-stamped"
                )
            }
        }
    }
}

#[derive(Debug, thiserror::Error)]
pub enum MigrationFailure {
    #[error("the manifest could not be read: {0}")]
    Read(std::io::Error),
    #[error("the manifest does not verify at {0} — migrating it would set the break in stone")]
    Broken(manifest::BreakReport),
    #[error("rev {} is neither gzip nor zstd: there is nothing under either name", .0.get())]
    NoSuchRev(Rev),
    #[error("rev {}'s bytes are not zstd: {why}", .at.get())]
    NotZstd { at: Rev, why: std::io::Error },
    #[error("rev {} could not be written: {why}", .at.get())]
    NotWritten { at: Rev, why: std::io::Error },
    #[error("the re-stamped manifest did not land: {0}")]
    NotStamped(std::io::Error),
    #[error("the migrated container did not reopen: {0}")]
    Reopen(String),
}

/// Rewrite `storage`'s revs from zstd to gzip, re-stamp the rows that name
/// them, and refresh the projection so its stamp names the head as it now
/// stands.
///
/// Each rev lands under its new name before the old one is removed, and
/// the manifest is rewritten only once every rev has: a run that stops
/// halfway leaves a container whose revs are partly converted and whose
/// rows still stamp the zstd bytes, which the next run finishes.
///
/// # Errors
/// [`MigrationFailure`] — the manifest could not be read or does not
/// verify, a rev is missing or is not zstd, or a write did not land.
pub fn to_gzip<S: Storage>(storage: &S) -> Result<Migrated, MigrationFailure> {
    let text = read_manifest(storage)?;
    let scanned = manifest::scan(&text);
    if let End::Broken(report) = scanned.end {
        return Err(MigrationFailure::Broken(report));
    }
    let mut rows: Vec<Row> = scanned
        .rows
        .into_iter()
        .map(|verified| verified.row)
        .collect();
    let mut rewritten = 0;
    let mut stamped: Vec<(Rev, Digest)> = Vec::new();
    for at in rows
        .iter()
        .filter(|row| row.kind.takes_a_rev())
        .map(|row| row.rev)
    {
        let (digest, moved) = carry_over(storage, at)?;
        rewritten += usize::from(moved);
        stamped.push((at, digest));
    }
    restamp(&mut rows, &stamped);

    let text_now = lines(&rows);
    if rewritten == 0 && text_now == text {
        return Ok(Migrated::AlreadyDone);
    }
    ready_now(storage.write(&MANIFEST, text_now.as_bytes()))
        .map_err(MigrationFailure::NotStamped)?;
    Ok(Migrated::Rewrote { revs: rewritten })
}

/// Read rev `at` as zstd and write it back as gzip, answering the digest
/// its row must now stamp and whether anything moved.
fn carry_over<S: Storage>(storage: &S, at: Rev) -> Result<(Digest, bool), MigrationFailure> {
    let old = zstd_entry(at);
    let new = entry(at);
    let Some(bytes) = read(storage, &old, at)? else {
        // Already carried over — by an earlier run, or by this build.
        let held = read(storage, &new, at)?.ok_or(MigrationFailure::NoSuchRev(at))?;
        return Ok((Digest::of(&held), false));
    };
    let json =
        zstd::decode_all(bytes.as_slice()).map_err(|why| MigrationFailure::NotZstd { at, why })?;
    let packed = pack(&json).map_err(|why| MigrationFailure::NotWritten { at, why })?;
    let written =
        ready_now(storage.write(&new, &packed)).and_then(|()| ready_now(storage.remove(&old)));
    written.map_err(|why| MigrationFailure::NotWritten { at, why })?;
    Ok((Digest::of(&packed), true))
}

fn read<S: Storage>(
    storage: &S,
    at: &Entry,
    rev: Rev,
) -> Result<Option<Vec<u8>>, MigrationFailure> {
    ready_now(storage.read(at)).map_err(|why| MigrationFailure::NotWritten { at: rev, why })
}

/// Give every row the hash its rev now has, and chain the file again.
///
/// A row that spends a rev stamps that rev's own bytes; one that does not
/// — a tag — stamps the head it was appended under, which is the last rev
/// before it in the file. The same rule [`Store`](crate::handle::Store) writes
/// rows under, read
/// off a file instead of a session.
fn restamp(rows: &mut [Row], stamped: &[(Rev, Digest)]) {
    let mut head = Digest::of(&[]);
    let mut parent = Digest::genesis();
    for row in rows.iter_mut() {
        if row.kind.takes_a_rev()
            && let Some((_, digest)) = stamped.iter().find(|(rev, _)| *rev == row.rev)
        {
            head = *digest;
        }
        row.hash = head;
        row.parent = parent;
        parent = row.digest();
    }
}

fn lines(rows: &[Row]) -> String {
    let mut text = String::new();
    for row in rows {
        text.push_str(&String::from_utf8_lossy(&row.canonical_bytes()));
        text.push('\n');
    }
    text
}

fn read_manifest<S: Storage>(storage: &S) -> Result<String, MigrationFailure> {
    let bytes = ready_now(storage.read(&MANIFEST))
        .map_err(MigrationFailure::Read)?
        .ok_or_else(|| {
            MigrationFailure::Read(std::io::Error::from(std::io::ErrorKind::NotFound))
        })?;
    String::from_utf8(bytes).map_err(|why| {
        MigrationFailure::Read(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            why.to_string(),
        ))
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fixture;
    use crate::handle::Store;
    use crate::storage::Native;
    use crate::tags::Tagging;

    /// The whole claim in one test: every rev comes over, the manifest
    /// still verifies against the bytes that are now there, and the
    /// document reads back unchanged.
    #[test]
    fn a_zstd_container_comes_over_and_still_verifies() {
        let dir = fixture::dir("migrate-gzip");
        let root = dir.join("doc.bwx");
        let document = {
            let mut store =
                Store::create(Native::at(&root), fixture::clock()).expect("the container");
            for commit in fixture::edits(3) {
                store
                    .submit_edit(commit, &fixture::author())
                    .expect("the edit lands");
            }
            store
                .tag(
                    blockworx_doc::fixtures::rev(2),
                    "Initial Draft",
                    Tagging::Added,
                    &fixture::author(),
                )
                .expect("the tag lands");
            store.document().clone()
        };
        roll_back_to_zstd(&root);

        let migrated = to_gzip(&Native::at(&root)).expect("the container comes over");
        assert_eq!(migrated, Migrated::Rewrote { revs: 3 });

        for at in crate::revs::through(blockworx_doc::fixtures::rev(3)) {
            assert!(
                root.join(entry(at).as_str()).is_file(),
                "rev {} was not written as gzip",
                at.get(),
            );
            assert!(
                !root.join(zstd_entry(at).as_str()).exists(),
                "rev {}'s old file was left behind",
                at.get(),
            );
        }

        let reopened = Store::open(Native::at(&root), fixture::clock()).expect("it reopens");
        assert!(
            reopened.read_only_reason().is_none(),
            "the migrated manifest does not verify: {:?}",
            reopened.read_only_reason(),
        );
        assert_eq!(reopened.document().clone(), document);
        assert_eq!(
            reopened.tags().of(blockworx_doc::fixtures::rev(2)),
            ["Initial Draft"],
            "the tag row did not come over",
        );
        assert!(
            crate::dump::verify(Native::at(&root))
                .expect("the manifest reads")
                .broken
                .is_none(),
            "the fsck does not accept the migrated container",
        );
    }

    /// Running it twice is running it once.
    #[test]
    fn a_container_already_gzip_is_left_alone() {
        let dir = fixture::dir("migrate-idempotent");
        let root = dir.join("doc.bwx");
        {
            let mut store =
                Store::create(Native::at(&root), fixture::clock()).expect("the container");
            for commit in fixture::edits(2) {
                store
                    .submit_edit(commit, &fixture::author())
                    .expect("the edit lands");
            }
        }
        let before = std::fs::read(root.join(MANIFEST.as_str())).expect("the manifest");

        assert_eq!(
            to_gzip(&Native::at(&root)).expect("it answers"),
            Migrated::AlreadyDone,
        );
        assert_eq!(
            std::fs::read(root.join(MANIFEST.as_str())).expect("the manifest"),
            before,
            "a container with nothing to migrate had its manifest rewritten",
        );
    }

    /// A run cut short leaves a half-converted container, and the next one
    /// finishes it.
    #[test]
    fn a_half_converted_container_is_finished_by_the_next_run() {
        let dir = fixture::dir("migrate-half");
        let root = dir.join("doc.bwx");
        {
            let mut store =
                Store::create(Native::at(&root), fixture::clock()).expect("the container");
            for commit in fixture::edits(3) {
                store
                    .submit_edit(commit, &fixture::author())
                    .expect("the edit lands");
            }
        }
        roll_back_to_zstd(&root);
        // One rev carried over by hand, as an interrupted run would leave
        // it: the new file down, the old one gone, the rows untouched.
        let at = blockworx_doc::fixtures::rev(1);
        let json = zstd::decode_all(
            std::fs::read(root.join(zstd_entry(at).as_str()))
                .expect("the old rev")
                .as_slice(),
        )
        .expect("it decodes");
        std::fs::write(
            root.join(entry(at).as_str()),
            pack(&json).expect("it packs"),
        )
        .expect("the new rev");
        std::fs::remove_file(root.join(zstd_entry(at).as_str())).expect("the old rev goes");

        assert_eq!(
            to_gzip(&Native::at(&root)).expect("the rest comes over"),
            Migrated::Rewrote { revs: 2 },
        );
        let reopened = Store::open(Native::at(&root), fixture::clock()).expect("it reopens");
        assert!(reopened.read_only_reason().is_none());
    }

    /// A manifest that does not verify is left where it is: migrating it
    /// would set the break in stone under a fresh chain.
    #[test]
    fn a_broken_manifest_is_refused() {
        let dir = fixture::dir("migrate-broken");
        let root = dir.join("doc.bwx");
        {
            let mut store =
                Store::create(Native::at(&root), fixture::clock()).expect("the container");
            for commit in fixture::edits(3) {
                store
                    .submit_edit(commit, &fixture::author())
                    .expect("the edit lands");
            }
        }
        let mut lines = crate::tests::manifest_lines(&root);
        lines[1] = lines[1].replace("Added a block", "Added a blork");
        crate::tests::write_manifest(&root, &lines);

        assert!(matches!(
            to_gzip(&Native::at(&root)),
            Err(MigrationFailure::Broken(_)),
        ));
    }

    /// Rewrite a gzip container's revs as zstd and re-stamp its rows, so a
    /// test can stand a container as the old build would have left it.
    fn roll_back_to_zstd(root: &std::path::Path) {
        let text = std::fs::read_to_string(root.join(MANIFEST.as_str())).expect("the manifest");
        let mut rows: Vec<Row> = manifest::scan(&text)
            .rows
            .into_iter()
            .map(|verified| verified.row)
            .collect();
        let mut stamped: Vec<(Rev, Digest)> = Vec::new();
        for at in rows
            .iter()
            .filter(|row| row.kind.takes_a_rev())
            .map(|row| row.rev)
        {
            let gz = root.join(entry(at).as_str());
            let json = crate::revs::unpack(&std::fs::read(&gz).expect("the gzip rev"))
                .expect("it decodes");
            let packed = zstd::encode_all(json.as_slice(), 1).expect("it compresses");
            std::fs::write(root.join(zstd_entry(at).as_str()), &packed).expect("the zstd rev");
            std::fs::remove_file(&gz).expect("the gzip rev goes");
            stamped.push((at, Digest::of(&packed)));
        }
        restamp(&mut rows, &stamped);
        std::fs::write(root.join(MANIFEST.as_str()), lines(&rows)).expect("the manifest");
    }
}