From 7dfb73e25db1aa93540be81d82b58faedbc44177 Mon Sep 17 00:00:00 2001 From: Santiago Date: Tue, 1 Sep 2026 11:46:32 -0300 Subject: [PATCH] refactor(stelae): move the dolos-free residue out of the profile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first of the four in-workspace steps toward extracting the Stelae bounded context into its own repository. Pure mechanics: code that already belongs to the protocol or to a profile-generic driver crosses the crate boundary behind re-export shims, so every existing import path still resolves and no CLI or test file changes. Protocol residue into `crates/stelae`: - `history_for` becomes `stelae::inscription::history_for`, beside the invariant it is the constructive half of, with its five tests. - The fixed-arity canonical-CBOR decode helpers become `stelae::codec`. `Error` gains `MalformedRecord` and `HistoryBreak` to carry what they raise. New crate `crates/stelae-driver`, with its own `Error` and the same never-`dolos-*` boundary rule as `stelae`: `preflight`, `reporting`, `Standing`, `scope_key` and the `digests` codec, each moved with its tests. `dolos-snapshot` re-exports all of it at the old paths. The two `From` implementations flatten the refusals that moved — `MalformedRecord`, `OutOfOrder`, `NotEnoughSpace`, `MalformedInscription`, `HistoryBreak` — back into the variants they had, so what a caller matches and what an operator reads are unchanged; that is why there is no `Driver(..)` wrapper variant. Also drops the `registry` cargo feature from the CLI module's docs: no manifest defines it, and the OCI surface rides `dolos-snapshot/oci`, which the root manifest enables unconditionally. `crates/snapshot/tests/goldens.rs` passes with zero re-pins, and `cargo tree -e normal --all-features -p stelae -p stelae-driver` matches nothing `^dolos(-|$)`. Plan: plans/dolos-stelae-repo-split-residue-moves.md Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01BV9u2xqpYwrrPyfqiEDfoH --- Cargo.lock | 15 +- Cargo.toml | 1 + crates/snapshot/Cargo.toml | 4 +- crates/snapshot/src/export.rs | 248 +----------------- crates/snapshot/src/layers/blocks.rs | 2 +- crates/snapshot/src/layers/digests.rs | 159 +---------- crates/snapshot/src/layers/indexes.rs | 2 +- crates/snapshot/src/layers/logs.rs | 2 +- crates/snapshot/src/layers/mod.rs | 106 +------- crates/snapshot/src/layers/state.rs | 2 +- crates/snapshot/src/lib.rs | 116 +++++--- crates/snapshot/src/registry.rs | 9 +- crates/snapshot/src/restore.rs | 2 +- crates/stelae-driver/Cargo.toml | 28 ++ crates/stelae-driver/src/digests.rs | 163 ++++++++++++ crates/stelae-driver/src/lib.rs | 247 +++++++++++++++++ .../src/preflight.rs | 0 .../src/reporting.rs | 20 +- crates/stelae/src/codec.rs | 101 +++++++ crates/stelae/src/inscription.rs | 173 ++++++++++++ crates/stelae/src/lib.rs | 37 +++ src/bin/dolos/snapshot/mod.rs | 5 +- 22 files changed, 901 insertions(+), 541 deletions(-) create mode 100644 crates/stelae-driver/Cargo.toml create mode 100644 crates/stelae-driver/src/digests.rs create mode 100644 crates/stelae-driver/src/lib.rs rename crates/{snapshot => stelae-driver}/src/preflight.rs (100%) rename crates/{snapshot => stelae-driver}/src/reporting.rs (91%) create mode 100644 crates/stelae/src/codec.rs diff --git a/Cargo.lock b/Cargo.lock index 8a2a55cdf..20eaf2237 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1639,13 +1639,13 @@ dependencies = [ "dolos-core", "dolos-redb3", "dolos-testing", - "fs4", "hex", "minicbor 0.26.4", "rustls", "serde", "serde_json", "stelae", + "stelae-driver", "tempfile", "thiserror 2.0.18", "tracing", @@ -5385,6 +5385,19 @@ dependencies = [ "zstd", ] +[[package]] +name = "stelae-driver" +version = "1.7.0-alpha.1" +dependencies = [ + "fs4", + "minicbor 0.26.4", + "serde_json", + "stelae", + "tempfile", + "thiserror 2.0.18", + "tracing", +] + [[package]] name = "stringprep" version = "0.1.5" diff --git a/Cargo.toml b/Cargo.toml index 0f4fa445b..2cca11176 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -204,6 +204,7 @@ members = [ "crates/flatfiles", "crates/snapshot", "crates/stelae", + "crates/stelae-driver", "crates/testing", "crates/minibf", "crates/redb3", diff --git a/crates/snapshot/Cargo.toml b/crates/snapshot/Cargo.toml index 566caa54d..805508856 100644 --- a/crates/snapshot/Cargo.toml +++ b/crates/snapshot/Cargo.toml @@ -10,17 +10,17 @@ edition.workspace = true # only writes steles to a directory keeps the dependency tree it had, which is # the whole reason the protocol's transport is behind a feature in the first # place. `dolos`'s own `registry` feature is what turns this on. -oci = ["stelae/oci"] +oci = ["stelae/oci", "stelae-driver/oci"] # This crate is the *profile* half of the Stelae boundary: it may depend on # `stelae` and on `dolos-*`, never the other way around. See # adrs/004_stelae_snapshots.md, "Code layout". [dependencies] stelae = { path = "../stelae" } +stelae-driver = { path = "../stelae-driver" } dolos-core = { path = "../core" } dolos-cardano = { path = "../cardano" } -fs4.workspace = true hex.workspace = true minicbor.workspace = true serde.workspace = true diff --git a/crates/snapshot/src/export.rs b/crates/snapshot/src/export.rs index 0b00515db..2153e3221 100644 --- a/crates/snapshot/src/export.rs +++ b/crates/snapshot/src/export.rs @@ -748,68 +748,11 @@ fn retained_dumps( .collect() } -/// The history a stele at `sequence` carries when it follows `previous`. -/// -/// The three legal readings of what came before, and the one refusal: -/// -/// - **nothing there** — an empty history, which the protocol permits at any -/// sequence. The first stele of a repository carries no history, and so does -/// a publisher deliberately starting a new one at epoch 500; -/// - **the stele before this one** — the old history plus an entry naming it. -/// Contiguous by construction, so the protocol's invariant passes rather than -/// being relied upon; -/// - **anything else** — refused, naming both sequences and, for a gap, the -/// distance between them. A gap means a publisher skipped epochs, an equal -/// sequence means it is republishing one, and a higher one means the -/// repository is ahead of this node. All three are operational faults with -/// different fixes, so the message says which. -/// -/// Whether a deliberate gap ever gets a policy is not this function's to -/// invent; there is no flag here that overrides the refusal. -/// -/// It lives here rather than in [`crate::registry`] because a verifier reaches -/// it without a registry, and because that module is behind a feature: a rule -/// this load-bearing should not be compiled out of a build that still has to -/// reproduce a chained digest. -pub fn history_for( - previous: Option<&Inscription>, - sequence: u64, -) -> Result, Error> { - let Some(previous) = previous else { - return Ok(Vec::new()); - }; - - let latest = previous.sequence; - - let reason = match latest.checked_add(1) { - Some(next) if next == sequence => { - let mut history = previous.history.clone(); - - history.push(HistoryEntry { - sequence: latest, - inscription_digest: previous.digest()?, - }); - - return Ok(history); - } - _ if latest >= sequence => { - "this stele is at or behind the repository's latest; a republish would restart the \ - chain rather than extend it" - .to_owned() - } - _ => format!( - "this node is {} sequences ahead, and a publish must follow the repository's latest \ - stele: this one would leave a gap no later stele could close", - sequence - latest, - ), - }; - - Err(Error::HistoryBreak { - latest, - publishing: sequence, - reason, - }) -} +/// The history rule, which is [`stelae::inscription::history_for`]: it reads a +/// predecessor's sequence and digest and composes nothing this profile owns. +/// Re-exported at its old path, where every caller and the module documentation +/// above already name it. +pub use stelae::inscription::history_for; /// Refuse a predecessor from another chain. /// @@ -829,53 +772,10 @@ pub fn same_network(previous: &Inscription, plan: &Plan) -> Result<(), Error> { Ok(()) } -/// Where a node stands relative to the newest stele already published. -/// -/// The comparison a publisher on a timer needs *before* anything is built, and -/// both halves of it are already in hand: the sequence a repository's latest -/// stele carries, and the sequence [`plan`] derived from the node's cursor. -/// Without it the ordinary case — nothing has closed since last time — arrives -/// as the [`Error::HistoryBreak`] refusal a skipped epoch does, and a job on a -/// timer cannot tell the two apart. -/// -/// A pure comparison over two numbers rather than a method on a transport, so -/// the cases can be checked without one. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum Standing { - /// Nothing has been published; this stele would start the chain. - Empty, - /// The published chain has already reached this node. Not an error: a - /// publisher whose node has not entered a new epoch has nothing to do. - UpToDate { latest: u64 }, - /// The chain ends exactly one sequence back; this stele extends it. - Next { latest: u64 }, - /// The node is further ahead than one sequence, so a publish would leave a - /// gap. `distance` is how far — the number the refusal reports alongside - /// both sequences, because "you skipped some" and "you skipped forty" are - /// different incidents. - Ahead { latest: u64, distance: u64 }, -} - -impl Standing { - /// Read a node at `sequence` against a repository whose latest stele is - /// `latest`. - pub fn read(latest: Option, sequence: u64) -> Self { - let Some(latest) = latest else { - return Self::Empty; - }; - - match sequence.checked_sub(latest) { - None | Some(0) => Self::UpToDate { latest }, - Some(1) => Self::Next { latest }, - Some(distance) => Self::Ahead { latest, distance }, - } - } - - /// Whether a publish should go ahead. - pub fn publishable(&self) -> bool { - matches!(self, Self::Empty | Self::Next { .. }) - } -} +/// Where a node stands against the repository's newest stele, which is +/// [`stelae_driver::Standing`]: a comparison over two sequence numbers. +/// Re-exported at its old path. +pub use stelae_driver::Standing; /// Export a complete stele into `stele`: every layer, then the inscription. /// @@ -2423,89 +2323,6 @@ mod chain_tests { } } - /// The first stele of a repository carries no history, at any sequence. - #[test] - fn an_empty_repository_starts_a_history() { - assert!(history_for(None, 0).unwrap().is_empty()); - assert!(history_for(None, 500).unwrap().is_empty()); - } - - #[test] - fn a_publish_that_follows_latest_extends_the_chain() { - let previous = inscription(3, vec![entry(1), entry(2)]); - - let history = history_for(Some(&previous), 4).unwrap(); - - assert_eq!( - history.iter().map(|e| e.sequence).collect::>(), - vec![1, 2, 3], - "the old history plus an entry naming the stele it came from" - ); - - assert_eq!(history[2].inscription_digest, previous.digest().unwrap()); - - // The invariant holds by construction rather than by inspection: a - // document built on this history validates. - inscription(4, history).validate().unwrap(); - } - - /// All three refusals name both sequences, because which of the three it is - /// decides what the publisher does about it. - #[test] - fn a_publish_that_does_not_follow_latest_is_refused() { - let previous = inscription(497, vec![]); - - for publishing in [500, 497, 496] { - let err = history_for(Some(&previous), publishing).unwrap_err(); - let message = err.to_string(); - - assert!( - matches!(err, Error::HistoryBreak { .. }), - "{publishing}: {err:?}" - ); - - assert!(message.contains("497"), "{publishing}: {message}"); - assert!( - message.contains(&publishing.to_string()), - "{publishing}: {message}" - ); - } - } - - #[test] - fn a_gap_and_a_republish_are_told_apart() { - let previous = inscription(497, vec![]); - - assert!(history_for(Some(&previous), 500) - .unwrap_err() - .to_string() - .contains("gap")); - - assert!(history_for(Some(&previous), 497) - .unwrap_err() - .to_string() - .contains("republish")); - - assert!(history_for(Some(&previous), 496) - .unwrap_err() - .to_string() - .contains("republish")); - } - - /// A gap says how far. "The repository is at 497 and you are at 500" is a - /// different incident from being one epoch out, and the operator reading - /// the message should not have to subtract to find out which they have. - #[test] - fn a_gap_names_the_distance_alongside_both_sequences() { - let previous = inscription(497, vec![]); - - let message = history_for(Some(&previous), 500).unwrap_err().to_string(); - - assert!(message.contains("497"), "{message}"); - assert!(message.contains("500"), "{message}"); - assert!(message.contains("3 sequences ahead"), "{message}"); - } - /// The only thing standing between a publisher and a history chained onto /// another chain's stele. /// @@ -2702,53 +2519,6 @@ mod chain_tests { ); } - /// The four readings of a repository a publisher on a timer meets, and the - /// one that used to arrive as a refusal. - #[test] - fn a_repository_is_read_as_empty_current_next_or_ahead() { - assert_eq!(Standing::read(None, 500), Standing::Empty); - - // The ordinary case for a job that runs more often than epochs close. - assert_eq!( - Standing::read(Some(500), 500), - Standing::UpToDate { latest: 500 } - ); - - // And a node genuinely behind the repository, which is up to date in - // the only sense this comparison is for: there is nothing to publish. - assert_eq!( - Standing::read(Some(501), 500), - Standing::UpToDate { latest: 501 } - ); - - assert_eq!( - Standing::read(Some(499), 500), - Standing::Next { latest: 499 } - ); - - assert_eq!( - Standing::read(Some(497), 500), - Standing::Ahead { - latest: 497, - distance: 3 - } - ); - - for standing in [Standing::Empty, Standing::Next { latest: 1 }] { - assert!(standing.publishable(), "{standing:?}"); - } - - for standing in [ - Standing::UpToDate { latest: 1 }, - Standing::Ahead { - latest: 1, - distance: 2, - }, - ] { - assert!(!standing.publishable(), "{standing:?}"); - } - } - /// A verifier chains with the chain the stele attests, exactly as written. /// /// `Following` is a publisher's rule and refuses what a publisher must diff --git a/crates/snapshot/src/layers/blocks.rs b/crates/snapshot/src/layers/blocks.rs index 11a9f8894..bd3d84ee2 100644 --- a/crates/snapshot/src/layers/blocks.rs +++ b/crates/snapshot/src/layers/blocks.rs @@ -27,8 +27,8 @@ use dolos_core::{BlockHash, BlockSlot}; use stelae::frame::{self, CanonicalCbor}; -use super::{blob, close, fixed, open, uint}; use crate::{Error, BLOCKS}; +use stelae::codec::{blob, close, fixed, open, uint}; /// One block, as the layer carries it. #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/crates/snapshot/src/layers/digests.rs b/crates/snapshot/src/layers/digests.rs index c3e730a10..0ee4f4f87 100644 --- a/crates/snapshot/src/layers/digests.rs +++ b/crates/snapshot/src/layers/digests.rs @@ -1,156 +1,23 @@ -//! The `digests` layer: sha256 of every Cardano immutable-DB file the stele -//! covers. +//! The `digests` layer's codec, which lives in [`stelae_driver::digests`]. //! -//! `[immutable_number, chunk: bytes(32), primary: bytes(32), secondary: -//! bytes(32)]`, ascending `immutable_number`. -//! -//! ## Carries no restorable data -//! -//! Nothing here is written to a store. The layer exists so that block data -//! obtained from *somewhere else* — a Mithril aggregator, a mirror, a relay -//! replay — can be checked against the stele's signed inscription before it is -//! imported. It is also the enabler for a future Mithril-sourced restore mode, -//! which is why the digests are exactly Mithril Cardano DB v2's merkle leaves: -//! sha256 over the raw `.chunk`/`.primary`/`.secondary` file bytes, so a -//! certificate whose beacon covers `lastImmutable` verifies them by merkle -//! proof. -//! -//! The certificate itself is deliberately *not* part of a stele. Certificates -//! are produced on the aggregator's cadence, so two publishers at the same -//! boundary would reference different ones and stop reproducing each other's -//! inscription; the digest values, by contrast, are byte-stable properties of -//! the chain. -//! -//! The layer is optional — a network without a Mithril aggregator simply has no -//! `digests` layer, and the inscription says so. +//! Nothing about the kind is profile-shaped — its records name +//! immutable-database files by content and reach no store — so the codec moved +//! down to the driver crate. What stays here is the shim that keeps its old +//! path resolving, and the two free functions are spelled out rather than +//! re-exported because callers pass them where a `Result<_, crate::Error>` is +//! expected: the module's shape is unchanged, and so is what a failure of +//! either one is. -use stelae::{ - frame::{self, CanonicalCbor}, - Digest, -}; +use stelae::frame::CanonicalCbor; -use super::{close, fixed, open, uint}; -use crate::{Error, DIGESTS}; +pub use stelae_driver::digests::{ImmutableDigests, OrderCheck, DIGESTS}; -/// The three files of one immutable chunk, by content. -/// -/// Uses [`stelae::Digest`] rather than a fourth thirty-two-byte newtype: these -/// are sha256 over bytes, which is what that type is, and it already prints and -/// parses in the `sha256:…` form the rest of a stele uses. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct ImmutableDigests { - pub immutable_number: u64, - pub chunk: Digest, - pub primary: Digest, - pub secondary: Digest, -} +use crate::Error; pub fn encode(record: &ImmutableDigests) -> Result { - Ok(frame::encode(|e| { - e.array(4)? - .u64(record.immutable_number)? - .bytes(record.chunk.as_bytes())? - .bytes(record.primary.as_bytes())? - .bytes(record.secondary.as_bytes())?; - Ok(()) - })?) + Ok(stelae_driver::digests::encode(record)?) } pub fn decode(bytes: &[u8]) -> Result { - let mut decoder = minicbor::Decoder::new(bytes); - - open(DIGESTS, &mut decoder, 4)?; - - let immutable_number = uint(DIGESTS, "immutable_number", &mut decoder)?; - let chunk = fixed::<32>(DIGESTS, "chunk", &mut decoder)?; - let primary = fixed::<32>(DIGESTS, "primary", &mut decoder)?; - let secondary = fixed::<32>(DIGESTS, "secondary", &mut decoder)?; - - close(DIGESTS, &decoder, bytes)?; - - Ok(ImmutableDigests { - immutable_number, - chunk: Digest::from_bytes(chunk), - primary: Digest::from_bytes(primary), - secondary: Digest::from_bytes(secondary), - }) -} - -/// Strictly ascending `immutable_number` — one record per immutable file set. -#[derive(Debug, Default, Clone, Copy)] -pub struct OrderCheck { - last: Option, -} - -impl OrderCheck { - pub fn check(&mut self, record: &ImmutableDigests) -> Result<(), Error> { - if let Some(last) = self.last { - if record.immutable_number <= last { - return Err(Error::out_of_order( - DIGESTS, - format!( - "immutable {} follows immutable {last}", - record.immutable_number - ), - )); - } - } - - self.last = Some(record.immutable_number); - - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn record(number: u64) -> ImmutableDigests { - ImmutableDigests { - immutable_number: number, - chunk: Digest::from_bytes([0x11; 32]), - primary: Digest::from_bytes([0x22; 32]), - secondary: Digest::from_bytes([0x33; 32]), - } - } - - #[test] - fn round_trips() { - for number in [0u64, 6187, u32::MAX as u64 + 1] { - let original = record(number); - let encoded = encode(&original).unwrap(); - - assert_eq!(decode(encoded.as_bytes()).unwrap(), original); - } - } - - #[test] - fn a_wrong_width_digest_is_refused() { - let wire = frame::encode(|e| { - e.array(4)? - .u64(1)? - .bytes(&[0u8; 32])? - .bytes(&[0u8; 20])? - .bytes(&[0u8; 32])?; - Ok(()) - }) - .unwrap(); - - let err = decode(wire.as_bytes()).unwrap_err(); - assert!(matches!(err, Error::MalformedRecord { .. }), "{err:?}"); - } - - #[test] - fn ordering_is_strictly_ascending() { - let mut order = OrderCheck::default(); - - order.check(&record(0)).unwrap(); - order.check(&record(1)).unwrap(); - - for backwards in [record(1), record(0)] { - let err = OrderCheck { last: Some(1) }.check(&backwards).unwrap_err(); - assert!(matches!(err, Error::OutOfOrder { .. }), "{err:?}"); - } - } + Ok(stelae_driver::digests::decode(bytes)?) } diff --git a/crates/snapshot/src/layers/indexes.rs b/crates/snapshot/src/layers/indexes.rs index f56019db0..81271e1e5 100644 --- a/crates/snapshot/src/layers/indexes.rs +++ b/crates/snapshot/src/layers/indexes.rs @@ -39,8 +39,8 @@ use dolos_core::{ }; use stelae::frame::{self, CanonicalCbor}; -use super::{blob, close, fixed, open, text, uint}; use crate::{Error, INDEXES}; +use stelae::codec::{blob, close, fixed, open, text, uint}; /// Leading discriminant of a tag record. pub const TAG_DISCRIMINANT: u64 = 0; diff --git a/crates/snapshot/src/layers/logs.rs b/crates/snapshot/src/layers/logs.rs index 0b3c04e71..0f45054ab 100644 --- a/crates/snapshot/src/layers/logs.rs +++ b/crates/snapshot/src/layers/logs.rs @@ -26,8 +26,8 @@ use dolos_core::{EntityValue, LogKey}; use stelae::frame::{self, CanonicalCbor}; -use super::{blob, close, fixed, open}; use crate::Error; +use stelae::codec::{blob, close, fixed, open}; /// The name this codec refuses under. /// diff --git a/crates/snapshot/src/layers/mod.rs b/crates/snapshot/src/layers/mod.rs index 4eb3e607c..e88314c97 100644 --- a/crates/snapshot/src/layers/mod.rs +++ b/crates/snapshot/src/layers/mod.rs @@ -20,112 +20,12 @@ //! substitute for that promise; it is what catches a driver that merges, //! chunks or parallelizes those iterators and loses it. //! -//! ## Why decoding does not re-validate canonical form -//! -//! Records reach `decode` from [`stelae::dir::Layer::records`] or -//! [`stelae::LayerReader::next_record`], both of which have already validated -//! every byte against the deterministic profile — that is the framing layer's -//! job and it is not repeated here. What `decode` does check is *shape*: the -//! field count, each field's type and width, and that nothing trails the -//! record. +//! The shape checks every `decode` runs — the field count, each field's type +//! and width, and that nothing trails the record — are [`stelae::codec`]'s, and +//! the reason they are not re-validating canonical form is documented there. pub mod blocks; pub mod digests; pub mod indexes; pub mod logs; pub mod state; - -use minicbor::Decoder; - -use crate::Error; - -/// Open a record's outer array, insisting on a definite length of `expected`. -pub(crate) fn open( - kind: &'static str, - decoder: &mut Decoder<'_>, - expected: u64, -) -> Result<(), Error> { - let fields = decoder - .array() - .map_err(|e| Error::malformed(kind, format!("expected an array: {e}")))? - .ok_or_else(|| Error::malformed(kind, "indefinite-length array"))?; - - if fields != expected { - return Err(Error::malformed( - kind, - format!("expected {expected} fields, found {fields}"), - )); - } - - Ok(()) -} - -/// Insist the record ended where the array did. -/// -/// A CBOR sequence has no frame markers, so a record with a tail would be read -/// as one item by the framing layer and as a different, shorter item here — two -/// readers disagreeing about the same bytes, which is how a diffId stops -/// meaning anything. -pub(crate) fn close(kind: &'static str, decoder: &Decoder<'_>, bytes: &[u8]) -> Result<(), Error> { - let read = decoder.position(); - - if read != bytes.len() { - return Err(Error::malformed( - kind, - format!("{} trailing byte(s) after the record", bytes.len() - read), - )); - } - - Ok(()) -} - -pub(crate) fn uint( - kind: &'static str, - field: &str, - decoder: &mut Decoder<'_>, -) -> Result { - decoder - .u64() - .map_err(|e| Error::malformed(kind, format!("{field}: {e}"))) -} - -pub(crate) fn text<'b>( - kind: &'static str, - field: &str, - decoder: &mut Decoder<'b>, -) -> Result<&'b str, Error> { - decoder - .str() - .map_err(|e| Error::malformed(kind, format!("{field}: {e}"))) -} - -pub(crate) fn blob<'b>( - kind: &'static str, - field: &str, - decoder: &mut Decoder<'b>, -) -> Result<&'b [u8], Error> { - decoder - .bytes() - .map_err(|e| Error::malformed(kind, format!("{field}: {e}"))) -} - -/// A byte string of exactly `N` bytes. -/// -/// Width is checked here rather than by a lossy conversion downstream: every -/// fixed-width key type in `dolos-core` (`EntityKey`, `LogKey`, `KeyHash`) -/// converts from a slice by zero-padding or truncating, so a wrong-width field -/// would become a valid-looking key that no lookup can ever reach. -pub(crate) fn fixed( - kind: &'static str, - field: &str, - decoder: &mut Decoder<'_>, -) -> Result<[u8; N], Error> { - let raw = blob(kind, field, decoder)?; - - raw.try_into().map_err(|_| { - Error::malformed( - kind, - format!("{field}: expected {N} bytes, found {}", raw.len()), - ) - }) -} diff --git a/crates/snapshot/src/layers/state.rs b/crates/snapshot/src/layers/state.rs index 72c69ad6b..cf9666f60 100644 --- a/crates/snapshot/src/layers/state.rs +++ b/crates/snapshot/src/layers/state.rs @@ -50,8 +50,8 @@ use dolos_core::{state::KEY_SIZE, EntityKey, EntityValue, Era, EraCbor, Namespace, TxoRef}; use stelae::frame::{self, CanonicalCbor}; -use super::{blob, close, open, uint}; use crate::{namespaces, Error, UTXOS}; +use stelae::codec::{blob, close, open, uint}; /// The name this codec refuses under. /// diff --git a/crates/snapshot/src/lib.rs b/crates/snapshot/src/lib.rs index f67138760..0d4aa9147 100644 --- a/crates/snapshot/src/lib.rs +++ b/crates/snapshot/src/lib.rs @@ -56,12 +56,25 @@ pub mod export; pub mod layers; pub mod namespaces; -pub mod preflight; #[cfg(feature = "oci")] pub mod registry; -mod reporting; pub mod restore; +/// The free-space policy, which is [`stelae_driver`]'s: it is one rule over +/// paths and byte counts and knows nothing about what fills them. Re-exported +/// at its old path. +pub use stelae_driver::preflight; + +/// The layer and record arithmetic both drivers report through, which is +/// [`stelae_driver`]'s for the same reason. Not public here, because it never +/// was. +pub(crate) use stelae_driver::reporting; + +/// The pair that identifies one layer, which is [`stelae_driver::scope_key`]: +/// canonical-JSON equality over a scope this crate composed but the driver only +/// compares. +pub(crate) use stelae_driver::scope_key; + /// The observer seam both drivers report through, re-exported so a binary /// rendering one never has to name the protocol crate — the same property /// [`export::publish`] and [`restore::restore_dir`] hold for the transports. @@ -92,7 +105,10 @@ pub const PROFILE_VERSION: u64 = 1; pub const BLOCKS: &str = "blocks"; pub const INDEXES: &str = "indexes"; -pub const DIGESTS: &str = "digests"; + +/// Spelled by the codec that reads and writes the kind, so the vocabulary has +/// one definition on both sides of the crate boundary. +pub use layers::digests::DIGESTS; /// The namespaces the ledger writes epoch-boundary logs under, byte-sorted. /// @@ -340,30 +356,6 @@ pub fn is_inheritable(kind: &str, scope: &serde_json::Value) -> bool { .is_some() } -/// The pair that identifies one layer: its kind, and the canonical encoding of -/// its profile-owned scope. -/// -/// Canonical rather than [`serde_json::Value`] equality, because two scopes are -/// one layer exactly when they are the same bytes inside the canonical -/// document — the only sense of "the same scope" the protocol has. -/// -/// One function rather than three, and that is the point of it being here -/// instead of beside any one caller. Every table keyed this way is compared -/// against another table keyed this way: the predecessor's inheritable layers -/// against what a publish asks for, an interrupted publish's record against the -/// same, a reproduction's layers against the published ones. Three copies of -/// four lines would agree until one of them was corrected, and the failure that -/// follows is silent — a layer rebuilt instead of inherited, or a divergence -/// reported between two documents that say the same thing. -pub(crate) fn scope_key(kind: &str, scope: &serde_json::Value) -> Result<(String, String), Error> { - let canonical = stelae::inscription::canonical_json(scope)?; - - let canonical = String::from_utf8(canonical) - .map_err(|e| Error::malformed_inscription("layer scope", e.to_string()))?; - - Ok((kind.to_owned(), canonical)) -} - /// The epoch kinds a window always produces a layer for. /// /// The log kinds are the exception, and the only one: a log layer exists if and @@ -497,8 +489,11 @@ pub const COMPRESSION_LEVEL: i32 = 9; /// Errors raised by this profile. #[derive(Debug, thiserror::Error)] pub enum Error { + /// Anything the protocol refused that this enum does not name itself. The + /// driver's refusals arrive here too, through the same flattening — see the + /// two [`From`] implementations below. #[error("stelae error: {0}")] - Stelae(#[from] stelae::Error), + Stelae(stelae::Error), /// Raised where a record's validity is the index store's judgement rather /// than this crate's — exact-key widths, above all. Surfacing the store's @@ -619,8 +614,9 @@ pub enum Error { /// Raised only from a number that was actually measured against free space /// that was actually read — everything else warns and proceeds. One /// variant for both directions because it is one policy; see - /// [`crate::preflight`]. There is deliberately no flag that overrides it: - /// `--scratch-dir` pointed at a bigger volume is the escape hatch. + /// [`crate::preflight`], where it is raised. There is deliberately no flag + /// that overrides it: `--scratch-dir` pointed at a bigger volume is the + /// escape hatch. #[error("not enough space: {0}")] NotEnoughSpace(String), @@ -629,7 +625,8 @@ pub enum Error { /// Both sequences are in the message because the fix depends on which of /// them is wrong: a gap means a publisher skipped epochs, an equal or lower /// sequence means it is republishing one. There is deliberately no flag - /// that overrides this — see [`crate::registry`]. + /// that overrides this — see [`stelae::inscription::history_for`], where it + /// is raised. /// /// `reason` is owned rather than static so a gap can state its *distance*. /// "The repository is at 500 and you are at 540" is a different incident @@ -706,6 +703,63 @@ impl Error { } } +/// Refusals the protocol and the driver raise on this crate's behalf keep the +/// variant they had before those crates existed. +/// +/// `?` still converts, so no caller changed; what does not change either is +/// what a caller *matches* or an operator *reads*. The record-shape checks +/// ([`stelae::codec`]), the free-space policy ([`stelae_driver::preflight`]), +/// the ordering contracts and the history rule +/// ([`stelae::inscription::history_for`]) were all this crate's errors before +/// they moved, and each is matched on somewhere — a test, or the CLI's +/// exit-code mapping. Wrapping them would have renamed every one of those +/// refusals and prefixed every message; a `match` arm apiece is what a move +/// that changes nothing observable costs. +impl From for Error { + fn from(error: stelae::Error) -> Self { + match error { + stelae::Error::MalformedRecord { kind, reason } => { + Self::MalformedRecord { kind, reason } + } + stelae::Error::HistoryBreak { + latest, + publishing, + reason, + } => Self::HistoryBreak { + latest, + publishing, + reason, + }, + other => Self::Stelae(other), + } + } +} + +impl From for Error { + fn from(error: stelae_driver::Error) -> Self { + match error { + stelae_driver::Error::Stelae(error) => error.into(), + stelae_driver::Error::NotEnoughSpace(reason) => Self::NotEnoughSpace(reason), + stelae_driver::Error::MalformedRecord { kind, reason } => { + Self::MalformedRecord { kind, reason } + } + stelae_driver::Error::OutOfOrder { kind, reason } => Self::OutOfOrder { kind, reason }, + stelae_driver::Error::MalformedInscription { field, reason } => { + Self::MalformedInscription { field, reason } + } + stelae_driver::Error::HistoryBreak { + latest, + publishing, + reason, + } => Self::HistoryBreak { + latest, + publishing, + reason, + }, + } + } +} + /// The Dolos profile. /// /// Stateless by design: the [`Profile`] trait answers questions about naming diff --git a/crates/snapshot/src/registry.rs b/crates/snapshot/src/registry.rs index dbc2e85de..ddc796147 100644 --- a/crates/snapshot/src/registry.rs +++ b/crates/snapshot/src/registry.rs @@ -978,7 +978,10 @@ pub fn preflight(registry: &Registry) -> Result<(), Error> { return Ok(()); }; - crate::preflight::check(&[staging_need(staging_peak(registry)?, scratch_dir)]) + Ok(crate::preflight::check(&[staging_need( + staging_peak(registry)?, + scratch_dir, + )])?) } /// What the staging asks of its volume, as [`crate::preflight`] takes it. @@ -1867,7 +1870,9 @@ mod tests { fn a_measured_staging_shortfall_refuses_and_a_first_publish_does_not() { let temp = tempfile::tempdir().unwrap(); - let check = |peak| crate::preflight::check(&[staging_need(peak, temp.path())]); + let check = |peak| -> Result<(), Error> { + Ok(crate::preflight::check(&[staging_need(peak, temp.path())])?) + }; // No predecessor, so nothing sized it, so nothing refuses. check(None).unwrap(); diff --git a/crates/snapshot/src/restore.rs b/crates/snapshot/src/restore.rs index 1c513fd7c..d06591362 100644 --- a/crates/snapshot/src/restore.rs +++ b/crates/snapshot/src/restore.rs @@ -361,7 +361,7 @@ impl Plan { )); } - preflight::check(&needs) + Ok(preflight::check(&needs)?) } } diff --git a/crates/stelae-driver/Cargo.toml b/crates/stelae-driver/Cargo.toml new file mode 100644 index 000000000..e13e6bc87 --- /dev/null +++ b/crates/stelae-driver/Cargo.toml @@ -0,0 +1,28 @@ +[package] +name = "stelae-driver" +description = "Profile-generic driver machinery for the Stelae snapshot protocol" +version.workspace = true +edition.workspace = true + +[features] +# Forwarded to `stelae/oci` so a driver built for a registry-capable binary and +# the protocol crate under it agree on one feature, rather than each consumer +# having to name both. Nothing here is `#[cfg]`-gated on it yet. +oci = ["stelae/oci"] + +# The other half of the Stelae boundary from `stelae` itself: this crate holds +# the lifecycle machinery a publisher or a restorer needs whatever it is +# publishing, and like the protocol crate it must never depend on a `dolos-*` +# package. `cargo tree -p stelae-driver -e normal --all-features` shows the +# boundary holding. See adrs/004_stelae_snapshots.md, "Code layout". +[dependencies] +stelae = { path = "../stelae" } + +fs4.workspace = true +minicbor.workspace = true +serde_json.workspace = true +thiserror.workspace = true +tracing.workspace = true + +[dev-dependencies] +tempfile = "3.20.0" diff --git a/crates/stelae-driver/src/digests.rs b/crates/stelae-driver/src/digests.rs new file mode 100644 index 000000000..55a400b43 --- /dev/null +++ b/crates/stelae-driver/src/digests.rs @@ -0,0 +1,163 @@ +//! The `digests` layer: sha256 of every Cardano immutable-DB file the stele +//! covers. +//! +//! `[immutable_number, chunk: bytes(32), primary: bytes(32), secondary: +//! bytes(32)]`, ascending `immutable_number`. +//! +//! ## Carries no restorable data +//! +//! Nothing here is written to a store. The layer exists so that block data +//! obtained from *somewhere else* — a Mithril aggregator, a mirror, a relay +//! replay — can be checked against the stele's signed inscription before it is +//! imported. It is also the enabler for a future Mithril-sourced restore mode, +//! which is why the digests are exactly Mithril Cardano DB v2's merkle leaves: +//! sha256 over the raw `.chunk`/`.primary`/`.secondary` file bytes, so a +//! certificate whose beacon covers `lastImmutable` verifies them by merkle +//! proof. +//! +//! The certificate itself is deliberately *not* part of a stele. Certificates +//! are produced on the aggregator's cadence, so two publishers at the same +//! boundary would reference different ones and stop reproducing each other's +//! inscription; the digest values, by contrast, are byte-stable properties of +//! the chain. +//! +//! The layer is optional — a network without a Mithril aggregator simply has no +//! `digests` layer, and the inscription says so. + +use stelae::{ + codec::{close, fixed, open, uint}, + frame::{self, CanonicalCbor}, + Digest, +}; + +use crate::Error; + +/// The layer kind these records travel in. +/// +/// Spelled here rather than in the profile that lists it, so the codec and the +/// name its refusals quote cannot drift apart; a profile that carries the layer +/// names this constant in its own kind table. +pub const DIGESTS: &str = "digests"; + +/// The three files of one immutable chunk, by content. +/// +/// Uses [`stelae::Digest`] rather than a fourth thirty-two-byte newtype: these +/// are sha256 over bytes, which is what that type is, and it already prints and +/// parses in the `sha256:…` form the rest of a stele uses. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ImmutableDigests { + pub immutable_number: u64, + pub chunk: Digest, + pub primary: Digest, + pub secondary: Digest, +} + +pub fn encode(record: &ImmutableDigests) -> Result { + Ok(frame::encode(|e| { + e.array(4)? + .u64(record.immutable_number)? + .bytes(record.chunk.as_bytes())? + .bytes(record.primary.as_bytes())? + .bytes(record.secondary.as_bytes())?; + Ok(()) + })?) +} + +pub fn decode(bytes: &[u8]) -> Result { + let mut decoder = minicbor::Decoder::new(bytes); + + open(DIGESTS, &mut decoder, 4)?; + + let immutable_number = uint(DIGESTS, "immutable_number", &mut decoder)?; + let chunk = fixed::<32>(DIGESTS, "chunk", &mut decoder)?; + let primary = fixed::<32>(DIGESTS, "primary", &mut decoder)?; + let secondary = fixed::<32>(DIGESTS, "secondary", &mut decoder)?; + + close(DIGESTS, &decoder, bytes)?; + + Ok(ImmutableDigests { + immutable_number, + chunk: Digest::from_bytes(chunk), + primary: Digest::from_bytes(primary), + secondary: Digest::from_bytes(secondary), + }) +} + +/// Strictly ascending `immutable_number` — one record per immutable file set. +#[derive(Debug, Default, Clone, Copy)] +pub struct OrderCheck { + last: Option, +} + +impl OrderCheck { + pub fn check(&mut self, record: &ImmutableDigests) -> Result<(), Error> { + if let Some(last) = self.last { + if record.immutable_number <= last { + return Err(Error::out_of_order( + DIGESTS, + format!( + "immutable {} follows immutable {last}", + record.immutable_number + ), + )); + } + } + + self.last = Some(record.immutable_number); + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn record(number: u64) -> ImmutableDigests { + ImmutableDigests { + immutable_number: number, + chunk: Digest::from_bytes([0x11; 32]), + primary: Digest::from_bytes([0x22; 32]), + secondary: Digest::from_bytes([0x33; 32]), + } + } + + #[test] + fn round_trips() { + for number in [0u64, 6187, u32::MAX as u64 + 1] { + let original = record(number); + let encoded = encode(&original).unwrap(); + + assert_eq!(decode(encoded.as_bytes()).unwrap(), original); + } + } + + #[test] + fn a_wrong_width_digest_is_refused() { + let wire = frame::encode(|e| { + e.array(4)? + .u64(1)? + .bytes(&[0u8; 32])? + .bytes(&[0u8; 20])? + .bytes(&[0u8; 32])?; + Ok(()) + }) + .unwrap(); + + let err = decode(wire.as_bytes()).unwrap_err(); + assert!(matches!(err, Error::MalformedRecord { .. }), "{err:?}"); + } + + #[test] + fn ordering_is_strictly_ascending() { + let mut order = OrderCheck::default(); + + order.check(&record(0)).unwrap(); + order.check(&record(1)).unwrap(); + + for backwards in [record(1), record(0)] { + let err = OrderCheck { last: Some(1) }.check(&backwards).unwrap_err(); + assert!(matches!(err, Error::OutOfOrder { .. }), "{err:?}"); + } + } +} diff --git a/crates/stelae-driver/src/lib.rs b/crates/stelae-driver/src/lib.rs new file mode 100644 index 000000000..589709499 --- /dev/null +++ b/crates/stelae-driver/src/lib.rs @@ -0,0 +1,247 @@ +//! # Driver machinery for the Stelae protocol +//! +//! [`stelae`] is the protocol: framing, canonicalization, digests and the +//! naming rules. A *profile* — `dolos-snapshot` is this workspace's one — says +//! what a stele contains. Between them sits the work every publisher and every +//! restorer does whatever it is moving: sizing a volume before a run starts, +//! counting layers and records for whoever is watching, reading where a node +//! stands against a repository's newest stele, keying a layer by its kind and +//! scope. +//! +//! None of that is protocol and none of it is profile, so it lives here rather +//! than in either. No type here is a node's or a profile's — the boundary this +//! crate keeps is the same one `stelae` keeps, and the manifest states it. +//! +//! ## Module map +//! +//! - [`preflight`] — the free-space policy a publish and a restore share. +//! - [`reporting`] — the layer and record arithmetic behind +//! [`stelae::progress`]. +//! - [`digests`] — the codec for the `digests` layer kind. Its records are +//! sha256 over immutable-database files, which is a Cardano shape described +//! in Cardano words; the code depends on nothing but this crate and the +//! protocol, which is why it sits here. +//! - [`Standing`] — where a node stands against a repository's latest stele. +//! - [`scope_key`] — the pair that identifies one layer. + +pub mod digests; +pub mod preflight; +pub mod reporting; + +/// Errors raised by the driver. +/// +/// No variant carries a profile's or a node's types: a driver failure is about +/// a volume, a record's shape or a chain of sequences, and a profile wraps this +/// enum in its own rather than the other way round. +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("stelae error: {0}")] + Stelae(stelae::Error), + + /// A volume that cannot hold what the run is about to put on it, refused + /// before the run starts. + /// + /// Raised only from a number that was actually measured against free space + /// that was actually read — everything else warns and proceeds. One + /// variant for both directions because it is one policy; see + /// [`preflight`]. There is deliberately no flag that overrides it: a + /// scratch directory pointed at a bigger volume is the escape hatch. + #[error("not enough space: {0}")] + NotEnoughSpace(String), + + #[error("malformed {kind} record: {reason}")] + MalformedRecord { kind: &'static str, reason: String }, + + #[error("{kind} records are out of order: {reason}")] + OutOfOrder { kind: &'static str, reason: String }, + + /// A field of the inscription a profile owns — `position` or a layer's + /// `scope` — is not a shape that canonicalizes. + #[error("the inscription's {field} is not the shape this profile writes: {reason}")] + MalformedInscription { field: String, reason: String }, + + /// A publish that would not extend the repository's chain. Raised by + /// [`stelae::inscription::history_for`] and carried here unchanged. + #[error( + "this repository's latest stele is sequence {latest} and this publish is sequence \ + {publishing}: {reason}" + )] + HistoryBreak { + latest: u64, + publishing: u64, + reason: String, + }, +} + +/// Protocol refusals this crate also names keep their own variant rather than +/// arriving wrapped. +/// +/// [`stelae::codec`] raises `MalformedRecord` and +/// [`stelae::inscription::history_for`] raises `HistoryBreak`; both were this +/// crate's errors before they moved down into the protocol, and both are +/// matched on by callers. Flattening keeps the variant a caller matches and the +/// message an operator reads exactly what they were, at the cost of a `match` +/// arm per shared refusal — which is the direction a move that is supposed to +/// change nothing observable should pay in. +impl From for Error { + fn from(error: stelae::Error) -> Self { + match error { + stelae::Error::MalformedRecord { kind, reason } => { + Self::MalformedRecord { kind, reason } + } + stelae::Error::HistoryBreak { + latest, + publishing, + reason, + } => Self::HistoryBreak { + latest, + publishing, + reason, + }, + other => Self::Stelae(other), + } + } +} + +impl Error { + pub(crate) fn out_of_order(kind: &'static str, reason: impl Into) -> Self { + Self::OutOfOrder { + kind, + reason: reason.into(), + } + } + + pub(crate) fn malformed_inscription( + field: impl Into, + reason: impl Into, + ) -> Self { + Self::MalformedInscription { + field: field.into(), + reason: reason.into(), + } + } +} + +/// Where a node stands relative to the newest stele already published. +/// +/// The comparison a publisher on a timer needs *before* anything is built, and +/// both halves of it are already in hand: the sequence a repository's latest +/// stele carries, and the sequence the node's cursor derived. Without it the +/// ordinary case — nothing has closed since last time — arrives as the +/// [`Error::HistoryBreak`] refusal a skipped sequence does, and a job on a +/// timer cannot tell the two apart. +/// +/// A pure comparison over two numbers rather than a method on a transport, so +/// the cases can be checked without one. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Standing { + /// Nothing has been published; this stele would start the chain. + Empty, + /// The published chain has already reached this node. Not an error: a + /// publisher whose node has not entered a new epoch has nothing to do. + UpToDate { latest: u64 }, + /// The chain ends exactly one sequence back; this stele extends it. + Next { latest: u64 }, + /// The node is further ahead than one sequence, so a publish would leave a + /// gap. `distance` is how far — the number the refusal reports alongside + /// both sequences, because "you skipped some" and "you skipped forty" are + /// different incidents. + Ahead { latest: u64, distance: u64 }, +} + +impl Standing { + /// Read a node at `sequence` against a repository whose latest stele is + /// `latest`. + pub fn read(latest: Option, sequence: u64) -> Self { + let Some(latest) = latest else { + return Self::Empty; + }; + + match sequence.checked_sub(latest) { + None | Some(0) => Self::UpToDate { latest }, + Some(1) => Self::Next { latest }, + Some(distance) => Self::Ahead { latest, distance }, + } + } + + /// Whether a publish should go ahead. + pub fn publishable(&self) -> bool { + matches!(self, Self::Empty | Self::Next { .. }) + } +} + +/// The pair that identifies one layer: its kind, and the canonical encoding of +/// its profile-owned scope. +/// +/// Canonical rather than [`serde_json::Value`] equality, because two scopes are +/// one layer exactly when they are the same bytes inside the canonical +/// document — the only sense of "the same scope" the protocol has. +/// +/// One function rather than three, and that is the point of it being here +/// instead of beside any one caller. Every table keyed this way is compared +/// against another table keyed this way: the predecessor's inheritable layers +/// against what a publish asks for, an interrupted publish's record against the +/// same, a reproduction's layers against the published ones. Three copies of +/// four lines would agree until one of them was corrected, and the failure that +/// follows is silent — a layer rebuilt instead of inherited, or a divergence +/// reported between two documents that say the same thing. +pub fn scope_key(kind: &str, scope: &serde_json::Value) -> Result<(String, String), Error> { + let canonical = stelae::inscription::canonical_json(scope)?; + + let canonical = String::from_utf8(canonical) + .map_err(|e| Error::malformed_inscription("layer scope", e.to_string()))?; + + Ok((kind.to_owned(), canonical)) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The four readings of a repository a publisher on a timer meets, and the + /// one that used to arrive as a refusal. + #[test] + fn a_repository_is_read_as_empty_current_next_or_ahead() { + assert_eq!(Standing::read(None, 500), Standing::Empty); + + // The ordinary case for a job that runs more often than epochs close. + assert_eq!( + Standing::read(Some(500), 500), + Standing::UpToDate { latest: 500 } + ); + + // And a node genuinely behind the repository, which is up to date in + // the only sense this comparison is for: there is nothing to publish. + assert_eq!( + Standing::read(Some(501), 500), + Standing::UpToDate { latest: 501 } + ); + + assert_eq!( + Standing::read(Some(499), 500), + Standing::Next { latest: 499 } + ); + + assert_eq!( + Standing::read(Some(497), 500), + Standing::Ahead { + latest: 497, + distance: 3 + } + ); + + for standing in [Standing::Empty, Standing::Next { latest: 1 }] { + assert!(standing.publishable(), "{standing:?}"); + } + + for standing in [ + Standing::UpToDate { latest: 1 }, + Standing::Ahead { + latest: 1, + distance: 2, + }, + ] { + assert!(!standing.publishable(), "{standing:?}"); + } + } +} diff --git a/crates/snapshot/src/preflight.rs b/crates/stelae-driver/src/preflight.rs similarity index 100% rename from crates/snapshot/src/preflight.rs rename to crates/stelae-driver/src/preflight.rs diff --git a/crates/snapshot/src/reporting.rs b/crates/stelae-driver/src/reporting.rs similarity index 91% rename from crates/snapshot/src/reporting.rs rename to crates/stelae-driver/src/reporting.rs index 305c3709c..4b73c2346 100644 --- a/crates/snapshot/src/reporting.rs +++ b/crates/stelae-driver/src/reporting.rs @@ -5,7 +5,7 @@ //! fill one in — which layer of how many is in flight, and how often a long //! scan is worth mentioning. //! -//! Shared by [`crate::export`] and [`crate::restore`] because the two count the +//! Shared by a profile's export and restore drivers because the two count the //! same thing and a second copy would be a second answer to "how far along is //! this". Both types are call-scoped: they live on a stack frame for the length //! of one publish or one restore, and neither outlives it. @@ -33,14 +33,14 @@ const RECORD_CADENCE: u64 = 4096; /// one cursor. Positions are display order, nothing else: the inscription /// lists layers by its own rule, so two runs that announce in different /// interleavings still publish the same document. -pub(crate) struct Cursor<'a> { +pub struct Cursor<'a> { observer: &'a Observer, next: std::sync::atomic::AtomicUsize, total: usize, } impl<'a> Cursor<'a> { - pub(crate) fn new(observer: &'a Observer, total: usize) -> Self { + pub fn new(observer: &'a Observer, total: usize) -> Self { Self { observer, next: std::sync::atomic::AtomicUsize::new(0), @@ -49,7 +49,7 @@ impl<'a> Cursor<'a> { } /// Announce a layer and take its position in the run. - pub(crate) fn open(&self, kind: &str, scope: &serde_json::Value) -> usize { + pub fn open(&self, kind: &str, scope: &serde_json::Value) -> usize { let index = self.next.fetch_add(1, std::sync::atomic::Ordering::Relaxed); self.observer.emit(Event::LayerStarted { @@ -63,7 +63,7 @@ impl<'a> Cursor<'a> { } /// Close the layer `index` was handed out for. - pub(crate) fn close(&self, index: usize, kind: &str, outcome: Outcome) { + pub fn close(&self, index: usize, kind: &str, outcome: Outcome) { self.observer.emit(Event::LayerFinished { index, total: self.total, @@ -73,7 +73,7 @@ impl<'a> Cursor<'a> { } /// A record counter reporting to the same place. - pub(crate) fn records(&self) -> Records<'a> { + pub fn records(&self) -> Records<'a> { Records { observer: self.observer, pending: 0, @@ -82,7 +82,7 @@ impl<'a> Cursor<'a> { /// Layers announced so far — what a caller cross-checks its own total /// against once the run is over. - pub(crate) fn opened(&self) -> usize { + pub fn opened(&self) -> usize { self.next.load(std::sync::atomic::Ordering::Relaxed) } } @@ -93,13 +93,13 @@ impl<'a> Cursor<'a> { /// matters is *before* the layer closes: an observer that saw a layer finish /// and then received records for it would have to know which layer they /// belonged to, and the seam deliberately does not carry that. -pub(crate) struct Records<'a> { +pub struct Records<'a> { observer: &'a Observer, pending: u64, } impl Records<'_> { - pub(crate) fn tick(&mut self) { + pub fn tick(&mut self) { self.pending += 1; if self.pending >= RECORD_CADENCE { @@ -107,7 +107,7 @@ impl Records<'_> { } } - pub(crate) fn flush(&mut self) { + pub fn flush(&mut self) { if self.pending > 0 { self.observer.emit(Event::Records(self.pending)); self.pending = 0; diff --git a/crates/stelae/src/codec.rs b/crates/stelae/src/codec.rs new file mode 100644 index 000000000..0459b6e49 --- /dev/null +++ b/crates/stelae/src/codec.rs @@ -0,0 +1,101 @@ +//! Fixed-arity decode helpers for a layer's content records. +//! +//! A profile's records are CBOR arrays of a known width, read out of bytes +//! [`crate::frame`] has already validated against the deterministic profile. +//! What is left to check is *shape* — the field count, each field's type and +//! width, and that nothing trails the record — and every profile checks it the +//! same way, so the helpers live here rather than once per profile. +//! +//! ## Why decoding does not re-validate canonical form +//! +//! Records reach a profile's `decode` from [`crate::dir::Layer::records`] or +//! [`crate::LayerReader::next_record`], both of which have already validated +//! every byte against the deterministic profile — that is the framing layer's +//! job and it is not repeated here. + +use minicbor::Decoder; + +use crate::Error; + +/// Open a record's outer array, insisting on a definite length of `expected`. +pub fn open(kind: &'static str, decoder: &mut Decoder<'_>, expected: u64) -> Result<(), Error> { + let fields = decoder + .array() + .map_err(|e| Error::malformed(kind, format!("expected an array: {e}")))? + .ok_or_else(|| Error::malformed(kind, "indefinite-length array"))?; + + if fields != expected { + return Err(Error::malformed( + kind, + format!("expected {expected} fields, found {fields}"), + )); + } + + Ok(()) +} + +/// Insist the record ended where the array did. +/// +/// A CBOR sequence has no frame markers, so a record with a tail would be read +/// as one item by the framing layer and as a different, shorter item here — two +/// readers disagreeing about the same bytes, which is how a diffId stops +/// meaning anything. +pub fn close(kind: &'static str, decoder: &Decoder<'_>, bytes: &[u8]) -> Result<(), Error> { + let read = decoder.position(); + + if read != bytes.len() { + return Err(Error::malformed( + kind, + format!("{} trailing byte(s) after the record", bytes.len() - read), + )); + } + + Ok(()) +} + +pub fn uint(kind: &'static str, field: &str, decoder: &mut Decoder<'_>) -> Result { + decoder + .u64() + .map_err(|e| Error::malformed(kind, format!("{field}: {e}"))) +} + +pub fn text<'b>( + kind: &'static str, + field: &str, + decoder: &mut Decoder<'b>, +) -> Result<&'b str, Error> { + decoder + .str() + .map_err(|e| Error::malformed(kind, format!("{field}: {e}"))) +} + +pub fn blob<'b>( + kind: &'static str, + field: &str, + decoder: &mut Decoder<'b>, +) -> Result<&'b [u8], Error> { + decoder + .bytes() + .map_err(|e| Error::malformed(kind, format!("{field}: {e}"))) +} + +/// A byte string of exactly `N` bytes. +/// +/// Width is checked here rather than by a lossy conversion downstream: a +/// fixed-width key type typically converts from a slice by zero-padding or +/// truncating, so a wrong-width field would become a valid-looking key that no +/// lookup can ever reach. +pub fn fixed( + kind: &'static str, + field: &str, + decoder: &mut Decoder<'_>, +) -> Result<[u8; N], Error> { + let raw = blob(kind, field, decoder)?; + + raw.try_into().map_err(|_| { + Error::malformed( + kind, + format!("{field}: expected {N} bytes, found {}", raw.len()), + ) + }) +} diff --git a/crates/stelae/src/inscription.rs b/crates/stelae/src/inscription.rs index 3a6f7bf2b..914863e44 100644 --- a/crates/stelae/src/inscription.rs +++ b/crates/stelae/src/inscription.rs @@ -403,6 +403,71 @@ impl Inscription { } } +/// The history a stele at `sequence` carries when it follows `previous`. +/// +/// The constructive half of the invariant [`Inscription::validate`] checks: one +/// builds the chain, the other refuses a document whose chain is broken. +/// +/// The three legal readings of what came before, and the one refusal: +/// +/// - **nothing there** — an empty history, which the protocol permits at any +/// sequence. The first stele of a repository carries no history, and so does +/// a publisher deliberately starting a new one at sequence 500; +/// - **the stele before this one** — the old history plus an entry naming it. +/// Contiguous by construction, so the protocol's invariant passes rather than +/// being relied upon; +/// - **anything else** — refused, naming both sequences and, for a gap, the +/// distance between them. A gap means a publisher skipped sequences, an equal +/// sequence means it is republishing one, and a higher one means the +/// repository is ahead of this node. All three are operational faults with +/// different fixes, so the message says which. +/// +/// Whether a deliberate gap ever gets a policy is not this function's to +/// invent; there is no flag here that overrides the refusal. +/// +/// It lives beside the invariant rather than beside a transport because a +/// verifier reaches it without one, and a rule this load-bearing should not be +/// compiled out of a build that still has to reproduce a chained digest. +pub fn history_for( + previous: Option<&Inscription>, + sequence: u64, +) -> Result, Error> { + let Some(previous) = previous else { + return Ok(Vec::new()); + }; + + let latest = previous.sequence; + + let reason = match latest.checked_add(1) { + Some(next) if next == sequence => { + let mut history = previous.history.clone(); + + history.push(HistoryEntry { + sequence: latest, + inscription_digest: previous.digest()?, + }); + + return Ok(history); + } + _ if latest >= sequence => { + "this stele is at or behind the repository's latest; a republish would restart the \ + chain rather than extend it" + .to_owned() + } + _ => format!( + "this node is {} sequences ahead, and a publish must follow the repository's latest \ + stele: this one would leave a gap no later stele could close", + sequence - latest, + ), + }; + + Err(Error::HistoryBreak { + latest, + publishing: sequence, + reason, + }) +} + /// RFC 8785 canonical JSON encoding of an arbitrary value. /// /// Exposed so the conformance vectors exercise the same code path the @@ -525,6 +590,114 @@ mod tests { inscription } + /// An inscription at `sequence` carrying `history`, for the chain rules + /// below: the fields `history_for` reads and nothing else. + fn at(sequence: u64, history: Vec) -> Inscription { + let mut inscription = Inscription::new( + &Toy, + sequence, + json!({"chapter": sequence}), + json!({"noteWidth": 40}), + Compression { + algo: "zstd".to_owned(), + level: 9, + }, + ); + + inscription.history = history; + inscription + } + + fn entry(sequence: u64) -> HistoryEntry { + HistoryEntry { + sequence, + inscription_digest: Digest::compute(sequence.to_be_bytes()), + } + } + + /// The first stele of a repository carries no history, at any sequence. + #[test] + fn an_empty_repository_starts_a_history() { + assert!(history_for(None, 0).unwrap().is_empty()); + assert!(history_for(None, 500).unwrap().is_empty()); + } + + #[test] + fn a_publish_that_follows_latest_extends_the_chain() { + let previous = at(3, vec![entry(1), entry(2)]); + + let history = history_for(Some(&previous), 4).unwrap(); + + assert_eq!( + history.iter().map(|e| e.sequence).collect::>(), + vec![1, 2, 3], + "the old history plus an entry naming the stele it came from" + ); + + assert_eq!(history[2].inscription_digest, previous.digest().unwrap()); + + // The invariant holds by construction rather than by inspection: a + // document built on this history validates. + at(4, history).validate().unwrap(); + } + + /// All three refusals name both sequences, because which of the three it is + /// decides what the publisher does about it. + #[test] + fn a_publish_that_does_not_follow_latest_is_refused() { + let previous = at(497, vec![]); + + for publishing in [500, 497, 496] { + let err = history_for(Some(&previous), publishing).unwrap_err(); + let message = err.to_string(); + + assert!( + matches!(err, Error::HistoryBreak { .. }), + "{publishing}: {err:?}" + ); + + assert!(message.contains("497"), "{publishing}: {message}"); + assert!( + message.contains(&publishing.to_string()), + "{publishing}: {message}" + ); + } + } + + #[test] + fn a_gap_and_a_republish_are_told_apart() { + let previous = at(497, vec![]); + + assert!(history_for(Some(&previous), 500) + .unwrap_err() + .to_string() + .contains("gap")); + + assert!(history_for(Some(&previous), 497) + .unwrap_err() + .to_string() + .contains("republish")); + + assert!(history_for(Some(&previous), 496) + .unwrap_err() + .to_string() + .contains("republish")); + } + + /// A gap says how far. "The repository is at 497 and you are at 500" is a + /// different incident from being one epoch out, and the operator reading + /// the message should not have to subtract to find out which they have. + #[test] + fn a_gap_names_the_distance_alongside_both_sequences() { + let previous = at(497, vec![]); + + let message = history_for(Some(&previous), 500).unwrap_err().to_string(); + + assert!(message.contains("497"), "{message}"); + assert!(message.contains("500"), "{message}"); + assert!(message.contains("3 sequences ahead"), "{message}"); + } + #[test] fn canonical_form_is_stable_and_parses_back() { let inscription = sample(); diff --git a/crates/stelae/src/lib.rs b/crates/stelae/src/lib.rs index 2d5e3ace9..46a4aa205 100644 --- a/crates/stelae/src/lib.rs +++ b/crates/stelae/src/lib.rs @@ -52,6 +52,7 @@ //! - It has no `dolos-*` dependency, so extracting it later is a directory move //! rather than a refactor. +pub mod codec; pub mod digest; pub mod dir; pub mod frame; @@ -205,6 +206,33 @@ pub enum Error { #[error("history invariant violated: {0}")] HistoryInvariant(String), + /// A publish that would not extend the repository's chain. + /// + /// Both sequences are in the message because the fix depends on which of + /// them is wrong: a gap means a publisher skipped epochs, an equal or lower + /// sequence means it is republishing one. There is deliberately no flag + /// that overrides this — see [`inscription::history_for`]. + /// + /// `reason` is owned rather than static so a gap can state its *distance*. + /// "The repository is at 500 and you are at 540" is a different incident + /// from being one epoch out, and an operator reading the message should not + /// have to subtract. + #[error( + "this repository's latest stele is sequence {latest} and this publish is sequence \ + {publishing}: {reason}" + )] + HistoryBreak { + latest: u64, + publishing: u64, + reason: String, + }, + + /// A profile's content record whose shape is not the one its kind declares + /// — the field count, a field's type or width, or bytes trailing the + /// record. Raised by [`codec`], which is where every profile checks it. + #[error("malformed {kind} record: {reason}")] + MalformedRecord { kind: &'static str, reason: String }, + #[error("invalid digest {value:?}: {reason}")] InvalidDigest { value: String, reason: String }, @@ -339,3 +367,12 @@ pub enum Error { #[error("registry error: {0}")] Registry(#[from] oci_client::errors::OciDistributionError), } + +impl Error { + pub(crate) fn malformed(kind: &'static str, reason: impl Into) -> Self { + Self::MalformedRecord { + kind, + reason: reason.into(), + } + } +} diff --git a/src/bin/dolos/snapshot/mod.rs b/src/bin/dolos/snapshot/mod.rs index 27f876be8..bb3d854d5 100644 --- a/src/bin/dolos/snapshot/mod.rs +++ b/src/bin/dolos/snapshot/mod.rs @@ -9,8 +9,9 @@ //! without pulling a layer. `sign` is the rest of the //! publisher-productization slice, and restore is its own. //! -//! Publishing into a registry is behind the `registry` feature, which is -//! default-off; `--repo` in a build without it is a refusal naming the feature. +//! Publishing into a registry is the OCI transport, behind `dolos-snapshot`'s +//! `oci` feature — which this binary's dependency enables unconditionally (root +//! `Cargo.toml`), so every shipped `dolos` has `--repo`. //! //! ## One epoch selection, however many commands take one //!