diff --git a/crates/didbot-pds/src/journal.rs b/crates/didbot-pds/src/journal.rs new file mode 100644 index 00000000..9521a457 --- /dev/null +++ b/crates/didbot-pds/src/journal.rs @@ -0,0 +1,192 @@ +//! The contracts a durable store implements, and the marks that position it. +//! +//! Everything [`Durable::open`](crate::Durable) holds is a memory store with +//! one log underneath it. What differs between the kinds is what an entry +//! means and how big the state is, never how durability works — so durability +//! is stated once, here, and each store says its own two halves in its own +//! module. +//! +//! # The two halves +//! +//! [`Journaled::apply`] is replay: one fact out of the log, put back. It is +//! total, because a fact was accepted before it was written and there is +//! nothing left to refuse. [`Journaled::checkpoint`] is the other direction: +//! the store's state as the shortest sequence of facts that reproduces it, +//! which is what a compaction writes as the head of the rewritten log. +//! +//! The obligations run both ways. A store must reconstruct from its own +//! `checkpoint()` the state it has now, and where the checkpoint holds less +//! than the history did it must say which observable answers change — the +//! agent token store is the standing example and +//! [`MemoryAgentTokenStore`](crate::credential::MemoryAgentTokenStore)'s +//! `checkpoint` argues it. The caller must apply facts in journal order and +//! must never apply one twice. That second one is not a convenience: most of +//! replay is idempotent and [`Entry::LedgerAppended`] is not, so the strict +//! rule is taken everywhere rather than per store. **A journal position is +//! honoured exactly and never crossed.** +//! +//! # Which contract a store takes +//! +//! The shapes differ along two axes, and neither is the one they are usually +//! named by: +//! +//! - Is a value large enough that the journal should carry a locator rather +//! than the value? Then [`Bodied`]. +//! - Is the state a function of another store's state? Then [`Derived`]. +//! - Everything else is [`Journaled`] and nothing more, whatever its +//! cardinality. +//! +//! "Append-only" is not one of the axes. The ledger and the firehose +//! reservation look alike and are opposites — the reservation's replay is a +//! maximum and its checkpoint is one entry, the ledger's replay appends and +//! its checkpoint is the whole of it — and what separates them is +//! idempotence, which the strict position rule above means neither of them +//! needs the contract to know. + +use std::path::Path; + +use crate::records::MemoryRecordStore; +use crate::wal::Entry; + +/// A position in the journal. +/// +/// One offset into the single total order `pds.wal` writes, counted in bytes +/// from the start of the log: the position immediately after the last entry a +/// mark covers. +/// +/// [`Mark::ORIGIN`] is the position before the first entry, and it is what a +/// store whose only durability is the journal reports from +/// [`Journaled::durable_through`] — its state comes back because the journal +/// holds it, so there is no prefix of the journal that can be dropped on that +/// store's account. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct Mark(pub u64); + +impl Mark { + /// The position before the first entry. + pub const ORIGIN: Self = Self(0); +} + +/// The stores a [`Derived`] one reads itself out of. +/// +/// Handed to [`Derived::observe`] as the stores stand *before* the entry +/// being observed is applied, which is what lets a derivation be taken over +/// the value an entry is about to replace. +#[derive(Debug, Clone, Copy)] +pub struct Stores<'a> { + /// Records, under the keys they were written with. + pub records: &'a MemoryRecordStore, +} + +/// A store whose state is the log replayed into it. +/// +/// Implemented by the memory store rather than the file-backed wrapper +/// wherever the arms touch nothing but memory, so that a store's replay and +/// its checkpoint can be exercised without a data directory. +pub trait Journaled: Send + Sync { + /// This store's name in a checkpoint and in a log line. + const STORE: &'static str; + + /// Whether `entry` is this store's to apply. + /// + /// The answer lives beside the arm that acts on it, so that a variant + /// added to a store is a change to that store's module and to nothing + /// else. `durable::apply` holds the order the stores are asked in and + /// no knowledge of which entries are whose. + fn owns(entry: &Entry) -> bool + where + Self: Sized; + + /// Applies one replayed fact. + /// + /// Total: a fact was accepted before it was written, so there is nothing + /// here to refuse. Called only for entries this store + /// [`owns`](Journaled::owns), in journal order, once each. + fn apply(&self, entry: Entry); + + /// The state, as the shortest sequence of facts that reproduces it. + fn checkpoint(&self) -> Vec; + + /// The journal mark at or before which this store's state is durable + /// without the journal. + /// + /// The trim point for the log as a whole is the minimum over every store, + /// which is what lets one kind move to durability of its own without a + /// flag day: a store that keeps its own state file advances its own mark + /// and the minimum follows it. + fn durable_through(&self) -> Mark; +} + +/// A store whose values are large enough that the journal carries a locator. +/// +/// [`FileBlobStore`](crate::FileBlobStore) is the one such store here and it +/// is spelled by hand: [`Entry::BlobUploaded`] carries a +/// [`BlobRef`](crate::blobs::BlobRef) and the bytes live under +/// [`bodies`](Bodied::bodies). A log that carried the bytes would be a log +/// read into memory in full at every startup to find out what an account is +/// called. +/// +/// The ordering the contract requires is one-directional: **bytes reach the +/// medium before the fact that names them does.** A crash between them leaves +/// bytes nothing references, which is dead space; the reverse leaves a fact +/// naming bytes nothing has, which is a read that cannot be served. +pub trait Bodied: Journaled { + /// Where this store's values live under the data directory. + /// + /// The path a locator is resolved against: the root of a tree for a store + /// that files one value per name, and one file for a store that appends + /// values into it. + fn bodies(&self) -> &Path; +} + +/// A store whose state is a function of another store's. +/// +/// The blob index's reference counts are the case: they are the number of +/// live records pointing at each blob, and +/// [`BlobIndex::snapshot`](crate::blobs) carries no count because the +/// checkpoint restores blobs before the records that name them and replaying +/// those records rebuilds the counts. See `durable::snapshot` for that +/// ordering, which is load-bearing. +pub trait Derived: Journaled { + /// Reads one fact belonging to another store, before that store applies + /// it. + /// + /// Called for every replayed entry in journal order, with `from` as the + /// stores stand before `entry` is applied. That is what a count taken + /// over a replaced value needs: the record a `RecordPut` overwrites is + /// still readable when this runs, and it is the references it held that + /// stop being live. + fn observe(&self, entry: &Entry, from: Stores<'_>); +} + +/// Replays a store's own checkpoint into a fresh one, for a store's +/// round-trip test to compare. +/// +/// Two obligations in one call. Every entry a checkpoint writes is one the +/// store [`owns`](Journaled::owns), because an entry no store claims is an +/// entry a replay drops on the floor. And what comes back out of the fresh +/// store is what went in, which is the contract's own sentence: a store must +/// reconstruct from `checkpoint()` the state it was taken from. +/// +/// Rendered through `serde_json` rather than compared as values, because the +/// serialized form is what actually reaches the log. +#[cfg(test)] +pub(crate) fn round_trip( + store: &S, + fresh: &S, +) -> (serde_json::Value, serde_json::Value) { + let checkpoint = store.checkpoint(); + for entry in &checkpoint { + assert!( + S::owns(entry), + "{}'s checkpoint holds an entry it does not own", + S::STORE + ); + } + let written = serde_json::to_value(&checkpoint).expect("entries serialize"); + for entry in checkpoint { + fresh.apply(entry); + } + let replayed = serde_json::to_value(fresh.checkpoint()).expect("entries serialize"); + (written, replayed) +} diff --git a/crates/didbot-pds/src/lib.rs b/crates/didbot-pds/src/lib.rs index 2a29a736..81d48f7d 100644 --- a/crates/didbot-pds/src/lib.rs +++ b/crates/didbot-pds/src/lib.rs @@ -75,6 +75,7 @@ pub mod evaluation_log; pub mod export; pub mod format; pub mod history; +pub mod journal; pub mod kind; pub mod layout; pub mod ledger;