blockworx/store/

replay.rs

//! Replaying `log.jsonl` into a [`Repo`], verifying D12 as it goes.
//!
//! Two passes, and the order is the point. The chain is verified first,
//! over parsed records alone: a rewritten record is detected by its
//! *successor's* `parent` link, so checking links before folding is what
//! lets the second pass say something the first cannot — a state stamp
//! that disagrees under an intact chain is not a tampered file, it is the
//! app's fold having changed since the record was written.
//!
//! The stamps are [`Verify::Sampled`] at load and [`Verify::Full`] under
//! `blockworx verify`, per D12's reserved escalation: recomputing every
//! stamp re-serializes the whole document once per record, which cost
//! 16.6 s of a cold open on a 301-record, 2500-block container
//! (`TUNING.md`, Finding 7).
//!
//! The second pass also rebuilds the undo journal (F6). Each record's
//! `kind` crosses into the headless spelling here and is handed to
//! `Repo::replay_one`, which runs the live session's journalling policy —
//! so the depth a document reopens with is the depth it closed with, and
//! one fold per record still serves both the stamp check and the journal.
//!
//! The second pass is also where a record's artwork is read back out of
//! the container ([`super::assets`]) — after the chain, which covers the
//! record as the file holds it, and before the stamp, which then witnesses
//! the payload independently. See [`replay`].
//!
//! Target-independent: the browser replays the same text, and reaches its
//! payloads through the same [`AssetSource`].

use blockworx_doc::{document::FoldError, repo::Repo, rev::Rev};

use super::assets::{AssetFault, AssetSource};
use super::notes::Notes;
use super::record::{Digest, Entry, LogRecord};
use super::tags::Tags;

/// How many records one verified state stamp stands for under
/// [`Verify::Sampled`]. Every record is still folded and every chain link
/// still checked; what this spaces out is `Document::content_hash`, whose
/// cost is the whole document rather than the record.
pub const STAMP_SAMPLE: usize = 32;

/// How much of D12 a replay checks. The chain is always checked in full —
/// it is a hash over one line. The state stamps are what this chooses
/// between, because each one costs a re-serialization of the entire
/// document.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Verify {
    /// The head's stamp and one in every [`STAMP_SAMPLE`]. The head is
    /// never skipped: it is the document the session opens on, and fold
    /// drift that reaches the present reaches it.
    Sampled,
    /// Every stamp — the fsck, and what the editor's opens no longer do.
    Full,
}

impl Verify {
    fn checks_stamp(self, at: usize, head: usize) -> bool {
        match self {
            Verify::Full => true,
            Verify::Sampled => at == head || at.is_multiple_of(STAMP_SAMPLE),
        }
    }
}

/// Where in the file something went wrong — enough to render a span-style
/// report without re-reading it.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Located {
    /// 1-based.
    pub line: usize,
    /// 1-based, in bytes from the start of the line.
    pub column: usize,
    /// Byte offset of the line's first byte from the start of the file.
    pub offset: usize,
    /// The line's length in bytes, its newline excluded.
    pub len: usize,
}

impl std::fmt::Display for Located {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "line {}, column {}", self.line, self.column)
    }
}

/// What a record failed on.
#[derive(Debug)]
pub enum Fault {
    Malformed(serde_json::Error),
    /// D12: the record does not link to the one before it, so the file has
    /// been rewritten, reordered, or spliced at this point.
    Chained {
        expected: Digest,
        found: Digest,
    },
    /// The record claims a position the fold did not assign it.
    Misnumbered {
        expected: Rev,
        found: Rev,
    },
    /// The fold refuses a record it once accepted.
    Refused(FoldError),
    /// The record names artwork in `assets/` that is not there, or is not
    /// what its hash says. The chain still checks out — the log is the
    /// log it always was; what is missing is beside it.
    Artwork(AssetFault),
    /// The chain is intact and the record folds, but to a different
    /// document than the one that wrote it: this is a fold regression in
    /// the *app*, not damage to the file.
    Drift {
        expected: Digest,
        found: Digest,
        blame: Blame,
    },
    /// D18: a tag record stamps the head it was appended under, and this
    /// one does not.
    Mistagged {
        expected: Digest,
        found: Digest,
    },
}

/// Which record a stamp mismatch may be blamed on. Under
/// [`Verify::Sampled`] the stamps between two samples were never
/// recomputed, so the sample that fails is the first *evidence* of the
/// drift rather than its origin, and the report says so.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Blame {
    /// Every stamp before this one verified: the drift is this record.
    ThisRecord,
    /// The drift is at or before this record, and after `verified` — the
    /// last rev whose stamp was recomputed.
    AtOrBefore { verified: Rev },
}

impl std::fmt::Display for Fault {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Fault::Malformed(error) => write!(f, "the record does not parse: {error}"),
            Fault::Chained { expected, found } => write!(
                f,
                "the record links to {found}, but the record before it hashes to {expected} — \
                 the log has been rewritten here",
            ),
            Fault::Misnumbered { expected, found } => {
                write!(
                    f,
                    "the record claims rev {found}, but folds at {expected}",
                    found = found.get(),
                    expected = expected.get()
                )
            }
            Fault::Refused(refusal) => write!(f, "the fold refuses the record: {refusal}"),
            Fault::Artwork(fault) => write!(f, "{fault}"),
            Fault::Drift {
                expected,
                found,
                blame,
            } => write!(
                f,
                "the record folds to {found} but was written at {expected}; the chain is intact, \
                 so this build's fold no longer reproduces what wrote the log{blame}",
            ),
            Fault::Mistagged { expected, found } => write!(
                f,
                "the tag stamps {expected}, but the log it was appended to stands at {found} — \
                 it was written against a history this is not",
            ),
        }
    }
}

impl std::fmt::Display for Blame {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Blame::ThisRecord => Ok(()),
            Blame::AtOrBefore { verified } => write!(
                f,
                " — the drift is at or before this record, and after rev {verified}, whose stamp \
                 was the last one recomputed; `blockworx verify` checks every stamp and names it \
                 exactly",
                verified = verified.get(),
            ),
        }
    }
}

/// A fault and where it is. Kept apart from [`Broken`] so a container can
/// hold on to the report as its read-only reason without also holding the
/// repo it came with.
#[derive(Debug)]
pub struct BreakReport {
    pub at: Located,
    pub fault: Fault,
}

impl std::fmt::Display for BreakReport {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.at, self.fault)
    }
}

/// A final record the writer never finished. Appends are one `write_all`
/// of the line and its newline followed by an fsync, so a log that does
/// not end in a newline ends in a write that was never made durable: the
/// record is dropped and the file truncated back to the last whole line.
#[derive(Clone, Copy, Debug)]
pub struct DroppedTail {
    pub at: Located,
    /// Where the whole log ends.
    pub good_bytes: u64,
}

/// Whether the log ended where it should have. Not a bool: a dropped tail
/// is something the user is told about.
#[derive(Clone, Copy, Debug)]
pub enum Tail {
    Whole,
    Dropped(DroppedTail),
}

/// A log that verified end to end.
pub struct Replayed {
    pub repo: Repo,
    /// The digest the next appended record must name as its parent.
    pub head: Digest,
    pub tail: Tail,
    /// One per commit in `repo`'s log, in the same order — the audit
    /// columns the fold itself has no use for.
    pub entries: Vec<Entry>,
    /// What the log's tag records name its revs (D18). Never in `repo`.
    pub tags: Tags,
    /// What the log's note records say, keyed by the rev each occupies
    /// (D17). Unlike the tags, these *are* in `repo` — as the empty
    /// commits they were written as.
    pub notes: Notes,
    /// How many records were read, tags included — which `entries` is not,
    /// since a tag consumes no rev.
    pub verified: usize,
}

/// A log that did not. The prefix that *did* verify is still here: a
/// damaged log has a real past, and the reader is owed a look at it —
/// which is also why it is returned boxed, since it carries a whole
/// document and the success path should not widen to hold one.
pub struct Broken {
    pub repo: Repo,
    pub head: Digest,
    pub report: BreakReport,
    /// As [`Replayed::entries`], for the verified prefix.
    pub entries: Vec<Entry>,
    /// As [`Replayed::tags`], for the verified prefix.
    pub tags: Tags,
    /// As [`Replayed::notes`], for the verified prefix.
    pub notes: Notes,
    /// As [`Replayed::verified`], for the verified prefix.
    pub verified: usize,
}

/// Fold `log` record by record, checking every link and the stamps
/// `depth` asks for. Artwork the records only reference is read back out
/// of `assets`.
///
/// The three checks nest, and the nesting is what makes each one mean
/// something. The chain covers the record *as the file holds it* — a
/// hollowed asset op included — so it is settled by the first pass
/// before a payload is read. Hydration then rebuilds the ops, verifying each
/// payload against the hash the record addresses it by. The state stamp
/// comes last, over the document that fold produced, and was written
/// against a document folded from the same whole ops — so it is an
/// independent second witness that the bytes beside the log are the bytes
/// that wrote it.
///
/// # Errors
/// [`Broken`], carrying the verified prefix as a repo, when an interior
/// record does not parse, does not link, names artwork that is not there,
/// does not fold, or does not reproduce the document it was written at.
pub fn replay(log: &str, depth: Verify, assets: &dyn AssetSource) -> Result<Replayed, Box<Broken>> {
    let scan = scan(log);
    let last = scan.records.len().saturating_sub(1);
    let mut repo = Repo::default();
    let mut head = Digest::genesis();
    let mut stamped = Rev::ZERO;

    for (taken, verified) in scan.records.iter().enumerate() {
        let record = &verified.record;
        // D18: a tag is about history rather than part of it. It folds
        // nothing — which is what exempts it from the sequence check — and
        // claims only that the log stood where it says when it was
        // appended, checked on the same schedule as its neighbours' stamps.
        let Some(journal) = record.kind.journals_as() else {
            if depth.checks_stamp(taken, last) {
                let state: Digest = repo.document().content_hash().into();
                if state != record.state {
                    let fault = Fault::Mistagged {
                        expected: record.state,
                        found: state,
                    };
                    return Err(broken(
                        &scan,
                        assets,
                        taken,
                        head,
                        report(verified.at, fault),
                    ));
                }
            }
            head = verified.digest;
            continue;
        };
        let commit = match record.commit(assets) {
            Ok(commit) => commit,
            Err(fault) => {
                let report = report(verified.at, Fault::Artwork(fault));
                return Err(broken(&scan, assets, taken, head, report));
            }
        };
        let document = match repo.replay_one(commit, journal) {
            Ok(document) => document,
            Err(refusal) => {
                let report = report(verified.at, Fault::Refused(refusal));
                return Err(broken(&scan, assets, taken, head, report));
            }
        };
        if document.rev() != record.rev {
            let fault = Fault::Misnumbered {
                expected: document.rev(),
                found: record.rev,
            };
            return Err(broken(
                &scan,
                assets,
                taken,
                head,
                report(verified.at, fault),
            ));
        }
        if depth.checks_stamp(taken, last) {
            let state: Digest = document.content_hash().into();
            if state != record.state {
                let fault = Fault::Drift {
                    expected: record.state,
                    found: state,
                    blame: blame(stamped, record.rev),
                };
                return Err(broken(
                    &scan,
                    assets,
                    taken,
                    head,
                    report(verified.at, fault),
                ));
            }
            stamped = record.rev;
        }
        head = verified.digest;
    }

    let entries = entries(&scan.records);
    let (tags, notes) = projections(&scan.records);
    let verified = scan.records.len();
    match scan.end {
        End::Whole => Ok(Replayed {
            repo,
            head,
            tail: Tail::Whole,
            entries,
            tags,
            notes,
            verified,
        }),
        End::Truncated(dropped) => Ok(Replayed {
            repo,
            head,
            tail: Tail::Dropped(dropped),
            entries,
            tags,
            notes,
            verified,
        }),
        End::Broken(report) => Err(Box::new(Broken {
            repo,
            head,
            report,
            entries,
            tags,
            notes,
            verified,
        })),
    }
}

/// The audit columns, one per *commit* — a tag consumes no rev, so it
/// contributes no entry and the pairing with the repo's log holds.
fn entries(verified: &[Verified]) -> Vec<Entry> {
    verified
        .iter()
        .filter(|v| v.record.kind.journals_as().is_some())
        .map(|v| Entry::from(&v.record))
        .collect()
}

/// The two projections the fold does not keep: what the tag records name
/// (D18) and what the note records say (D17). Built from the records
/// rather than beside the fold, so a whole log and a verified prefix are
/// projected by the same pass.
fn projections(verified: &[Verified]) -> (Tags, Notes) {
    let mut tags = Tags::default();
    let mut notes = Notes::default();
    for record in verified.iter().map(|v| &v.record) {
        if record.kind.journals_as().is_none() {
            tags.set(record.rev, &record.label);
        }
        notes.take(record);
    }
    (tags, notes)
}

fn blame(stamped: Rev, at: Rev) -> Blame {
    if stamped.next() == at {
        Blame::ThisRecord
    } else {
        Blame::AtOrBefore { verified: stamped }
    }
}

/// The break at record `taken`, whose repo is the prefix *before* it —
/// re-replayed, because the growing one has already taken the offending
/// record (it folded legally; what it failed was its own stamp). Only the
/// failure path pays for that, and the records being re-folded have each
/// folded once already — which is also why the refusal arms cannot be
/// reached and simply stop.
fn broken(
    scan: &Scan,
    assets: &dyn AssetSource,
    taken: usize,
    head: Digest,
    report: BreakReport,
) -> Box<Broken> {
    let mut repo = Repo::default();
    for verified in &scan.records[..taken] {
        let Some(journal) = verified.record.kind.journals_as() else {
            continue;
        };
        let Ok(commit) = verified.record.commit(assets) else {
            break;
        };
        if repo.replay_one(commit, journal).is_err() {
            break;
        }
    }
    let (tags, notes) = projections(&scan.records[..taken]);
    Box::new(Broken {
        repo,
        head,
        report,
        entries: entries(&scan.records[..taken]),
        tags,
        notes,
        verified: taken,
    })
}

fn report(at: Located, fault: Fault) -> BreakReport {
    BreakReport { at, fault }
}

struct Verified {
    record: LogRecord,
    at: Located,
    digest: Digest,
}

/// How the record sequence stopped.
enum End {
    Whole,
    Truncated(DroppedTail),
    Broken(BreakReport),
}

struct Scan {
    records: Vec<Verified>,
    end: End,
}

/// Pass one: parse and link. Stops at the first record that does not
/// belong, keeping everything before it.
fn scan(log: &str) -> Scan {
    let mut records: Vec<Verified> = Vec::new();
    let mut parent = Digest::genesis();
    let mut offset = 0;

    for (number, chunk) in log.split_inclusive('\n').enumerate() {
        let line = chunk.strip_suffix('\n');
        let text = line.unwrap_or(chunk);
        let at = Located {
            line: number + 1,
            column: 1,
            offset,
            len: text.len(),
        };
        if line.is_none() {
            return Scan {
                records,
                end: End::Truncated(DroppedTail {
                    at,
                    good_bytes: offset as u64,
                }),
            };
        }
        offset += chunk.len();

        let record: LogRecord = match serde_json::from_str(text) {
            Ok(record) => record,
            Err(error) => {
                let at = Located {
                    column: error.column().max(1),
                    ..at
                };
                return Scan {
                    records,
                    end: End::Broken(BreakReport {
                        at,
                        fault: Fault::Malformed(error),
                    }),
                };
            }
        };
        if record.parent != parent {
            let fault = Fault::Chained {
                expected: parent,
                found: record.parent,
            };
            return Scan {
                records,
                end: End::Broken(BreakReport { at, fault }),
            };
        }
        parent = record.digest_as_written(text);
        records.push(Verified {
            record,
            at,
            digest: parent,
        });
    }

    Scan {
        records,
        end: End::Whole,
    }
}