diff --git a/adrs/004_stelae_snapshots.md b/adrs/004_stelae_snapshots.md index b9be6c2a..8f0342c8 100644 --- a/adrs/004_stelae_snapshots.md +++ b/adrs/004_stelae_snapshots.md @@ -106,7 +106,7 @@ None of those four goals is Cardano-specific, and neither are the mechanisms tha 10. **Specify the mechanism as a Dolos feature, without a protocol name or profile boundary** (the shape of this ADR before the Stelae amendment) - Pros: one crate, one vocabulary, no extension machinery to design or test. - - Cons: a third-party publisher has no collision-free namespace and would have to fork the spec; the Dolos context absorbs decisions (framing, attestation, transport) unrelated to a data node; and extraction later means renaming every media type, tag and identifier already published. The boundary costs one crate and one CI check today and is irreversible-cheap only before implementation starts. + - Cons: a third-party publisher has no collision-free namespace and would have to fork the spec; the Dolos context absorbs decisions (framing, attestation, transport) unrelated to a data node; and extraction later means renaming every media type, tag and identifier already published. The boundary costs one crate today and is irreversible-cheap only before implementation starts. ## Implementation Details @@ -246,7 +246,7 @@ The arithmetic is counted in layers, because layers are what the ceiling counts: ### Code layout -Two crates, both workspace members. The split is the protocol/profile boundary made mechanical: **`cargo tree -p stelae` must contain no `dolos-*` package**, checked in CI, so extracting the protocol later is a directory move rather than a refactor. +Two crates, both workspace members. The split is the protocol/profile boundary made mechanical: **`cargo tree -p stelae` must contain no `dolos-*` package**, so extracting the protocol later is a directory move rather than a refactor. ``` crates/stelae/ # package `stelae` — protocol, zero dolos deps @@ -335,7 +335,7 @@ Steps 1–2 and the fetch/verify half of steps 4–5 are protocol code; the stor ### Development phases -**1a. Stelae core** — `crates/stelae`: framing, inscription (schema, JCS, digest, history invariant), the `Profile` trait and naming rules, streaming digest/compression, signatures. Verified by CBOR-seq roundtrip and write→read→write byte-identity property tests, a JCS inscription golden test, history-invariant tests (gap/duplicate/out-of-order → reject), fail-closed tests (unknown generic key, unknown profile, higher profile major), a toy non-Dolos profile exercising the full path, and the `cargo tree -p stelae` boundary check. +**1a. Stelae core** — `crates/stelae`: framing, inscription (schema, JCS, digest, history invariant), the `Profile` trait and naming rules, streaming digest/compression, signatures. Verified by CBOR-seq roundtrip and write→read→write byte-identity property tests, a JCS inscription golden test, history-invariant tests (gap/duplicate/out-of-order → reject), fail-closed tests (unknown generic key, unknown profile, higher profile major), and a toy non-Dolos profile exercising the full path. **1b. Dolos profile core** — `crates/snapshot`: `DolosProfile`, layer readers/writers, and the three trait additions with backend impls and adapter enums. Verified by per-layer roundtrip unit tests and golden-digest tests (fixed input → asserted sha256, catching encoding drift). diff --git a/crates/snapshot/src/export.rs b/crates/snapshot/src/export.rs index d5389cc2..e9c9b654 100644 --- a/crates/snapshot/src/export.rs +++ b/crates/snapshot/src/export.rs @@ -263,6 +263,233 @@ impl Predecessor for First { } } +/// The stele this one follows, held as a document rather than as a repository. +/// +/// What a publisher into a registry gets from [`crate::registry`] and what a +/// *verifier* has instead: the predecessor's canonical inscription, off disk or +/// out of a pull, and nothing else. It answers the first of a +/// [`Predecessor`]'s two questions and declines the second — inheriting a layer +/// means arranging for a transport to carry its blob, and a reproduction has no +/// transport and wants to build every layer anyway. +/// +/// It exists so that `dolos snapshot digest --chain-from` extends a history by +/// the same rule a publish does. Two implementations of contiguity is two +/// things to keep honest, and the one that drifts is the one nobody publishes +/// with. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Following { + history: Vec, +} + +impl Following { + /// Read a predecessor's canonical inscription and chain `plan` onto it. + /// + /// The front door for a caller holding bytes rather than a document — the + /// CLI's `--chain-from`, above all — so the `dolos` binary keeps never + /// naming the protocol crate, the same property [`publish`] and + /// [`crate::registry::open`] hold. + /// + /// The bytes must *be* the canonical encoding, exactly as + /// `stelae::dir::SteleDir` requires of the one it holds: a stele is chained + /// to by the digest of its own bytes, so a re-encoded copy names a stele + /// nobody published, and the entry this history would carry would attest a + /// document that does not exist. + pub fn read(raw: &[u8], plan: &Plan) -> Result { + let previous = Inscription::parse(raw)?; + + if previous.canonicalize()? != raw { + return Err(Error::malformed_inscription( + "the document", + "it is not in canonical form; a predecessor is chained to by the digest of its \ + own bytes, so a re-encoded copy names a stele nobody published", + )); + } + + previous.check_profile(&DolosProfile)?; + + Self::new(&previous, plan) + } + + /// The history a stele at `plan.sequence` carries when it follows + /// `previous`. + /// + /// Refuses a predecessor from another network for the same reason a publish + /// does, and refuses one this plan does not follow — a reproduction chained + /// onto the wrong stele would report a digest that is *correct* for a stele + /// nobody published, which is worse than an error. + pub fn new(previous: &Inscription, plan: &Plan) -> Result { + same_network(previous, plan)?; + + Ok(Self { + history: history_for(Some(previous), plan.sequence)?, + }) + } +} + +impl Predecessor for Following { + fn history(&self) -> &[HistoryEntry] { + &self.history + } +} + +/// The chain a published stele attests, taken verbatim. +/// +/// What `verify --reproduce` chains with. [`Following`] extends a +/// predecessor's history by the contiguity rule, because a *publisher* is +/// adding a link; a verifier is not — it is recomputing a document somebody +/// already published, and the `history` inside that document is an input it +/// cannot know any other way. Taking it verbatim is the honest statement of +/// what is being checked: the layers and the document around them, not the +/// chain's provenance. Closing that gap is what signatures are for. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Attested { + history: Vec, +} + +impl Attested { + /// The history `published` carries, as a reproduction's input. + pub fn of(published: &Inscription) -> Self { + Self { + history: published.history.clone(), + } + } +} + +impl Predecessor for Attested { + fn history(&self) -> &[HistoryEntry] { + &self.history + } +} + +/// 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, + }) +} + +/// Refuse a predecessor from another chain. +/// +/// A repository holding two networks' steles is an operator fault, and the +/// check costs nothing: the previous stele's `position` already names its +/// network, and reading it is the same function a restore uses. +pub fn same_network(previous: &Inscription, plan: &Plan) -> Result<(), Error> { + let found = crate::read_position(&previous.position)?.network; + + if found.magic() != plan.network.magic() { + return Err(Error::NetworkMismatch { + expected: plan.network.magic(), + found: found.magic(), + }); + } + + 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 { .. }) + } +} + /// Export a complete stele into `stele`: every layer, then the inscription. /// /// Layers are listed in [`crate::KINDS`] order, and within a kind in ascending @@ -363,6 +590,238 @@ where ) } +/// Reproduce a stele from `plan` and store nothing. +/// +/// The counterpart of [`publish`] for a caller that wants the identity and not +/// the artifact: `dolos snapshot digest`, and the reproduction half of +/// `snapshot verify`. Same walk, same framing, same zstd — see +/// [`stelae::Discarding`] for what is and is not dropped — into a writer with +/// no destination. +/// +/// Here rather than at the call site for the reason [`publish`] is: the profile +/// stays the only thing in the `dolos` binary that names the protocol crate. +/// +/// `previous` is what the reproduction chains onto. It is an input and not +/// something this can work out — `history` is inside the canonical document, so +/// the same stores chained differently are different digests. Pass +/// [`First`] for a stele that starts a chain and [`Following`] for one that +/// extends a predecessor's. +pub fn reproduce( + plan: &Plan, + archive: &A, + state: &S, + indexes: &I, + digest_records: Option<&[digests::ImmutableDigests]>, + previous: &dyn Predecessor, +) -> Result +where + A: ArchiveStore, + S: StateStore, + I: IndexStore, +{ + export( + &stelae::Discarding, + plan, + archive, + state, + indexes, + digest_records, + previous, + ) +} + +/// Reproduce `published` from local stores and hold the two documents against +/// each other. +/// +/// The deferred half of the incremental publish's trade: a layer inherited +/// rather than rebuilt is *attested without being reproduced*, and this is the +/// call that reproduces it. Every layer is rebuilt through the same discarding +/// pipeline [`reproduce`] uses — the published document never short-circuits +/// the walk — and the comparison runs descriptor by descriptor before it runs +/// digest against digest, so a divergence is reported as the layer that +/// diverged rather than as two hashes that differ. +/// +/// The published `history` is taken verbatim ([`Attested`]): a verifier cannot +/// know a chain it did not publish, so a match proves the layers and the +/// document around them, never the chain's provenance. +/// +/// The sequence is checked before a single record is walked. A plan standing +/// at a different epoch than the published stele cannot match it, and finding +/// that out should not cost the hours of compression a full walk does. The +/// network is checked on the same terms, and it is not implied by the sequence: +/// epoch numbers collide across chains, so a preprod stele and a mainnet store +/// can stand at the same sequence and still have nothing to say to each other. +/// [`Attested`] takes the published history verbatim and so — unlike +/// [`Following::new`] — checks nothing about it; this is where that check +/// lands. +pub fn verify_reproduction( + published: &Inscription, + plan: &Plan, + archive: &A, + state: &S, + indexes: &I, + digest_records: Option<&[digests::ImmutableDigests]>, +) -> Result +where + A: ArchiveStore, + S: StateStore, + I: IndexStore, +{ + if plan.sequence != published.sequence { + return Err(Error::ReproductionMismatch { + subject: "sequence".to_owned(), + reason: format!( + "the published stele is sequence {} and these stores stand at sequence {}; a \ + reproduction runs over stores at the epoch the stele was published from", + published.sequence, plan.sequence, + ), + }); + } + + // Same reasoning as the sequence check, and not covered by it: a store on + // another chain cannot reproduce this stele, and epoch numbers collide + // across networks, so same-sequence-different-chain is reachable. + same_network(published, plan)?; + + let reproduced = export( + &stelae::Discarding, + plan, + archive, + state, + indexes, + digest_records, + &Attested::of(published), + )?; + + compare(published, &reproduced)?; + + Ok(reproduced) +} + +/// Hold a reproduced inscription against the published one, layers first. +/// +/// Kind by kind and scope by scope rather than digest first: two hashes that +/// differ say nothing an operator can act on, while "the blocks layer at epoch +/// 412 has a different diffId" is ADR-004's residual risk with a name on it. +/// The digest comparison still runs, last — it covers the fields no layer +/// owns — but by then the layers are known to agree, so a failure there names +/// the generic document instead of hiding a layer divergence behind it. +fn compare(published: &Inscription, reproduced: &Inscription) -> Result<(), Error> { + let published_layers = layers_by_scope(published)?; + let mut reproduced_layers = layers_by_scope(reproduced)?; + + for ((kind, scope), theirs) in &published_layers { + let subject = format!("the {kind} layer at {scope}"); + + let ours = reproduced_layers + .remove(&(kind.clone(), scope.clone())) + .unwrap_or_default(); + + if theirs.len() != ours.len() { + return Err(Error::ReproductionMismatch { + subject, + reason: match ours.len() { + 0 => "the published stele describes it and the reproduction built no such \ + layer; the stores may cover different epochs than the stele, or \ + `--epochs` may select a different window than the publish did" + .to_owned(), + built => format!( + "the published stele describes it {} time(s) and the reproduction built \ + it {built}", + theirs.len(), + ), + }, + }); + } + + for (their, our) in theirs.iter().zip(&ours) { + let fields = [ + ("diffId", their.diff_id.to_string(), our.diff_id.to_string()), + ( + "records", + their.records.to_string(), + our.records.to_string(), + ), + ( + "uncompressedSize", + their.uncompressed_size.to_string(), + our.uncompressed_size.to_string(), + ), + ( + "mediaType", + their.media_type.clone(), + our.media_type.clone(), + ), + ]; + + for (field, published_value, reproduced_value) in fields { + if published_value != reproduced_value { + return Err(Error::ReproductionMismatch { + subject, + reason: format!( + "published {field} is {published_value} and the reproduction \ + computed {reproduced_value}", + ), + }); + } + } + } + } + + if let Some(((kind, scope), _)) = reproduced_layers.into_iter().next() { + return Err(Error::ReproductionMismatch { + subject: format!("the {kind} layer at {scope}"), + reason: "the reproduction built it and the published stele does not describe it" + .to_owned(), + }); + } + + let published_digest = published.digest()?; + let reproduced_digest = reproduced.digest()?; + + if published_digest != reproduced_digest { + return Err(Error::ReproductionMismatch { + subject: "the inscription".to_owned(), + reason: format!( + "every layer agrees and the documents still differ — published \ + {published_digest}, reproduced {reproduced_digest}; the divergence is in a \ + generic field (position, parameters, compression, history) or in layer order", + ), + }); + } + + Ok(()) +} + +/// A stele's layers keyed by the pair that identifies one to an operator: the +/// kind, and the canonical encoding of its profile-owned scope. +/// +/// Canonical rather than [`serde_json::Value`] equality, so two scopes are one +/// key exactly when they are the same bytes inside the canonical document — +/// the only sense of "the same scope" the protocol has. A `Vec` per key rather +/// than a refusal of duplicates: the comparison's job is to report what the +/// documents say, not to relitigate their validity. +fn layers_by_scope( + inscription: &Inscription, +) -> Result>, Error> { + let mut layers: std::collections::BTreeMap<(String, String), Vec<&LayerDescriptor>> = + std::collections::BTreeMap::new(); + + for layer in &inscription.layers { + let scope = stelae::inscription::canonical_json(&layer.scope)?; + + let scope = String::from_utf8(scope) + .map_err(|e| Error::malformed_inscription("layer scope", e.to_string()))?; + + layers + .entry((layer.kind.clone(), scope)) + .or_default() + .push(layer); + } + + Ok(layers) +} + fn sink(stele: &W, spec: &LayerSpec) -> Result { Ok(stele.layer_sink(&DolosProfile, spec, COMPRESSION_LEVEL)?) } @@ -777,3 +1236,419 @@ mod tests { assert_eq!(plan.epochs[0].epoch, 1); } } + +#[cfg(test)] +mod chain_tests { + use dolos_core::{BlockHash, ChainPoint}; + use serde_json::json; + use stelae::Digest; + + use super::*; + use crate::{DolosProfile, Network}; + + fn inscription(sequence: u64, history: Vec) -> Inscription { + let mut inscription = Inscription::new( + &DolosProfile, + sequence, + json!({"epoch": sequence.saturating_sub(1)}), + crate::parameters(), + crate::compression(), + ); + + inscription.history = history; + inscription + } + + fn entry(sequence: u64) -> HistoryEntry { + HistoryEntry { + sequence, + inscription_digest: Digest::compute(sequence.to_be_bytes()), + } + } + + fn plan_at(network: Network) -> Plan { + Plan { + network, + cursor: ChainPoint::Specific(250, BlockHash::from([0xab; 32])), + sequence: 3, + epochs: vec![], + } + } + + /// 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. + /// + /// A publish reads its own magic from genesis and the predecessor's from + /// the predecessor. If they were allowed to differ, the new inscription + /// would attest a chain of steles from a network it has never seen — and + /// nothing downstream re-checks it, because `history` entries carry a + /// sequence and a digest and no position at all. + #[test] + fn a_predecessor_from_another_network_is_refused() { + let preview = Network::for_magic(crate::PREVIEW_MAGIC); + let preprod = Network::for_magic(crate::PREPROD_MAGIC); + + let plan = plan_at(preview.clone()); + + // Built by `crate::position`, not by the `inscription` helper's shape: + // `same_network` reads it back through `crate::read_position`, which + // the helper's bare `{"epoch": n}` would not survive. + let stele = |network: &Network| { + let mut previous = inscription(3, vec![]); + + previous.position = crate::position( + network, + &ChainPoint::Specific(250, BlockHash::from([0xab; 32])), + 2, + ) + .unwrap(); + + previous + }; + + same_network(&stele(&preview), &plan).unwrap(); + + let err = same_network(&stele(&preprod), &plan).unwrap_err(); + + assert!( + matches!( + err, + Error::NetworkMismatch { expected, found } + if expected == preview.magic() && found == preprod.magic() + ), + "{err:?}" + ); + } + + /// `--chain-from` extends a history by the rule a publish extends it by, + /// because it is the same call. + #[test] + fn following_a_predecessor_is_the_publishers_own_rule() { + let network = Network::for_magic(crate::PREVIEW_MAGIC); + + let mut previous = inscription(2, vec![entry(1)]); + previous.position = crate::position( + &network, + &ChainPoint::Specific(150, BlockHash::from([0xcd; 32])), + 1, + ) + .unwrap(); + + let plan = plan_at(network); + + let following = Following::new(&previous, &plan).unwrap(); + + assert_eq!( + following.history(), + history_for(Some(&previous), 3).unwrap() + ); + assert_eq!( + following + .history() + .iter() + .map(|e| e.sequence) + .collect::>(), + vec![1, 2], + ); + + // And a predecessor this plan does not follow is refused here too: a + // reproduction chained onto the wrong stele would report a digest that + // is correct for a stele nobody published. + let mut skipped = previous.clone(); + skipped.sequence = 7; + + assert!(matches!( + Following::new(&skipped, &plan).unwrap_err(), + Error::HistoryBreak { .. } + )); + } + + /// The bytes handed to `--chain-from` have to *be* the document, not merely + /// encode it. + /// + /// Every caller inside this tree hands [`Following::read`] canonical bytes, + /// so the only thing that ever exercises this refusal is an operator's + /// pipeline: a predecessor that went through a pretty-printer, an editor, + /// or a shell that put a newline on the end parses cleanly and + /// describes the very same stele. Chaining onto it anyway would produce + /// a history entry naming the digest of bytes nobody published — a + /// reproduction that reports a digest correct for a stele that does not + /// exist, which is the failure [`Following::read`] exists to prevent. + #[test] + fn a_predecessor_that_is_not_in_canonical_form_is_refused() { + let network = Network::for_magic(crate::PREVIEW_MAGIC); + + let mut previous = inscription(2, vec![entry(1)]); + previous.position = crate::position( + &network, + &ChainPoint::Specific(150, BlockHash::from([0xcd; 32])), + 1, + ) + .unwrap(); + + let plan = plan_at(network); + + let canonical = previous.canonicalize().unwrap(); + + assert_eq!( + Following::read(&canonical, &plan).unwrap(), + Following::new(&previous, &plan).unwrap(), + "the canonical bytes chain exactly as the document does" + ); + + // Both parse, both describe this same stele, and neither is it: one + // re-serialized in the struct's declaration order rather than JCS's + // sorted one, one the canonical bytes with a newline appended. + let reordered = serde_json::to_vec(&previous).unwrap(); + + let mut newline_terminated = canonical.clone(); + newline_terminated.push(b'\n'); + + for (label, raw) in [ + ("reordered keys", reordered), + ("a trailing newline", newline_terminated), + ] { + assert_ne!( + raw, canonical, + "{label}: the fixture is canonical after all" + ); + + assert!( + Inscription::parse(&raw).is_ok(), + "{label}: it must be refused for its encoding, not for being unparseable" + ); + + let err = Following::read(&raw, &plan).unwrap_err(); + + assert!( + matches!(err, Error::MalformedInscription { .. }), + "{label}: {err:?}" + ); + } + } + + /// 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 + /// refuse; `Attested` is a verifier's input and refuses nothing, because + /// the document it comes from was already validated on the way in — and a + /// reproduction that "fixed" the history would compute a digest for a + /// stele nobody published. + #[test] + fn an_attested_history_is_taken_verbatim() { + let published = inscription(3, vec![entry(1), entry(2)]); + + assert_eq!(Attested::of(&published).history(), &published.history[..]); + assert!(Attested::of(&inscription(3, vec![])).history().is_empty()); + } + + fn described(kind: &str, scope: serde_json::Value, identity: u8) -> LayerDescriptor { + LayerDescriptor { + kind: kind.to_owned(), + media_type: format!("application/vnd.dolos.stele.{kind}.v1+zstd"), + diff_id: Digest::compute([identity]), + records: 2, + uncompressed_size: 64, + scope, + } + } + + fn with_layers(layers: Vec) -> Inscription { + let mut document = inscription(3, vec![entry(1), entry(2)]); + document.layers = layers; + document + } + + /// The comparison names the layer, not two hashes. + /// + /// This is the report `verify --reproduce` exists to produce: a diverging + /// layer with its kind and scope is ADR-004's residual risk located, while + /// two inscription digests that differ locate nothing. + #[test] + fn a_diverging_layer_is_named_by_kind_and_scope() { + let scope = json!({"endSlot": 299, "epoch": 2, "startSlot": 200}); + + let published = with_layers(vec![ + described("blocks", scope.clone(), 1), + described("state", json!({"shard": 0}), 2), + ]); + + // Identical documents agree, digest included. + compare(&published, &published.clone()).unwrap(); + + // One layer's bytes came out differently. + let mut diverged = published.clone(); + diverged.layers[0].diff_id = Digest::compute([9]); + + let err = compare(&published, &diverged).unwrap_err(); + let message = err.to_string(); + + assert!(matches!(err, Error::ReproductionMismatch { .. }), "{err:?}"); + assert!(message.contains("blocks"), "{message}"); + assert!(message.contains(r#""epoch":2"#), "{message}"); + assert!(message.contains("diffId"), "{message}"); + + // A layer the reproduction never built. + let missing = with_layers(vec![described("state", json!({"shard": 0}), 2)]); + let message = compare(&published, &missing).unwrap_err().to_string(); + + assert!(message.contains("blocks"), "{message}"); + assert!(message.contains("built no such layer"), "{message}"); + + // And one it built that the published stele does not describe. + let message = compare(&missing, &published).unwrap_err().to_string(); + + assert!(message.contains("blocks"), "{message}"); + assert!(message.contains("does not describe it"), "{message}"); + } + + /// When every layer agrees the digest still has the last word: `history` + /// is inside the canonical document, and two documents over identical + /// layers chained differently are different steles, deliberately. + #[test] + fn identical_layers_chained_differently_still_diverge() { + let layers = vec![described("state", json!({"shard": 0}), 2)]; + + let published = with_layers(layers.clone()); + + let mut unchained = published.clone(); + unchained.history = vec![entry(2)]; + unchained.history[0].sequence = 2; + + let err = compare(&published, &unchained).unwrap_err(); + let message = err.to_string(); + + assert!(message.contains("every layer agrees"), "{message}"); + assert!(message.contains("history"), "{message}"); + } +} diff --git a/crates/snapshot/src/lib.rs b/crates/snapshot/src/lib.rs index 3ae0784e..39b449d3 100644 --- a/crates/snapshot/src/lib.rs +++ b/crates/snapshot/src/lib.rs @@ -210,6 +210,11 @@ pub enum Error { /// 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`]. + /// + /// `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}" @@ -217,7 +222,32 @@ pub enum Error { HistoryBreak { latest: u64, publishing: u64, - reason: &'static str, + reason: String, + }, + + /// A reproduction that arrived at a different answer than the published + /// stele it recomputed. + /// + /// A finding, not a routine failure: same stores, same epochs and the same + /// history reproduce the same document, so a divergence means either the + /// inputs are not what the operator believes — a store at another epoch, a + /// different `--epochs` window than the publish used — or the format's + /// determinism claim has a hole, which is ADR-004's residual risk with a + /// name on it. `subject` names the layer (or the generic field) so the two + /// can be told apart. + #[error("the reproduction does not match the published stele — {subject}: {reason}")] + ReproductionMismatch { subject: String, reason: String }, + + /// A layer that failed its transport verification, named. + /// + /// The wrapper exists because the protocol's refusals name a layer by kind + /// at best, and `snapshot verify`'s deliverable is an exit code plus the + /// offending layer — so the scope rides along with the kind. + #[error("the {kind} layer at {scope} failed verification: {source}")] + LayerVerification { + kind: String, + scope: String, + source: Box, }, } diff --git a/crates/snapshot/src/registry.rs b/crates/snapshot/src/registry.rs index 58491146..f685638c 100644 --- a/crates/snapshot/src/registry.rs +++ b/crates/snapshot/src/registry.rs @@ -17,6 +17,20 @@ //! fault, and silently starting a second chain in one repository is precisely //! what the field exists to prevent. There is no flag here that overrides that. //! +//! The rule itself is [`crate::export::history_for`], one function shared with +//! the verifier that reproduces a chained digest without a registry. +//! +//! ## "Nothing has closed since last time" is not a fault +//! +//! One of those refusals was carrying two meanings. A publisher on a timer +//! whose node has not entered a new epoch is in the *ordinary* case, and it +//! reached [`publish`] as the same `HistoryBreak` a skipped epoch raises. +//! [`standing`] separates them before anything is built, out of the two numbers +//! a publish already has: the sequence the moving tag carries and the one the +//! cursor implies. A node that is not past the repository has nothing to do, +//! which a caller reports and exits zero on; a node further ahead than one +//! sequence still meets the refusal, now with the distance in it. +//! //! ## A reused layer is attested without being reproduced //! //! The layers of an epoch that has closed cannot change, so a publish that @@ -89,8 +103,11 @@ use std::{cell::Cell, collections::BTreeMap}; use dolos_core::{ArchiveStore, IndexStore, StateStore}; use stelae::{ + digest::LayerDigests, + frame::Limits, inscription::{HistoryEntry, Inscription, LayerDescriptor}, - oci::{Options, Registry, Stele, Transfer}, + oci::{Options, Stele, Transfer}, + transport::BlobIndex, Digest, SteleReader as _, }; @@ -105,10 +122,10 @@ use stelae::{ /// [`Auth`] rides along for the same reason: a host resolving its own /// credentials should not have to name the protocol crate to say what it /// resolved them to. -pub use stelae::oci::{Auth, Repository, SCHEME}; +pub use stelae::oci::{Auth, Registry, Repository, SCHEME}; use crate::{ - export::{self, Plan, Predecessor}, + export::{self, history_for, same_network, Plan, Predecessor, Standing}, layers::digests, restore::{Outlook, Restoring, Summary, Target}, DolosProfile, Error, Scope as _, EPOCH_KINDS, STATE_SHARDS, @@ -275,6 +292,202 @@ pub struct Preview { pub layers_built: usize, } +/// Where this node stands relative to what `registry` already holds. +/// +/// One read of the moving tag, and no store is touched. It is what turns "the +/// node has not entered a new epoch" from the same refusal a skipped epoch +/// raises into an answer a job on a timer can act on — see [`Standing`]. +/// +/// Cheap enough to ask before every publish: a manifest pull against the +/// moving tag, which [`publish`] and [`preview`] are each about to make anyway. +/// Asking twice is one HTTP round trip against the alternative, which is +/// threading the answer out of a call that has already started building. +/// +/// **Never call this from inside an async context.** See [`open`]. +pub fn standing(registry: &Registry, plan: &Plan) -> Result { + let latest = registry + .latest(&DolosProfile)? + .map(|stele| stele.read_inscription()) + .transpose()?; + + if let Some(previous) = &latest { + // The same refusal a publish makes, made before the report rather than + // after it: a repository holding another network's chain is not "up to + // date" with this node in any sense worth reporting. + same_network(previous, plan)?; + } + + Ok(Standing::read( + latest.map(|previous| previous.sequence), + plan.sequence, + )) +} + +/// What `snapshot verify` established about a published stele, transport side. +/// +/// By the time this exists, every check that needs no store has run: +/// +/// - **the two documents agree.** The pull itself proved it: the inscription is +/// canonical, its digest is the config digest the manifest names, and every +/// manifest layer matches the inscription's at the same position and `diffId` +/// — `stelae::oci::read_manifest`, the contract the manifest golden freezes. +/// - **the history chain is contiguous back to its first entry** and ends at +/// `sequence - 1`. The protocol refuses to parse an inscription whose chain +/// skips, so a gapped history never reaches the layer checks at all. +/// - **every blob was streamed end to end.** Its bytes hash to the blob digest +/// the manifest addresses it by and stay within the compressed size the +/// manifest claims; what they decompress to stays within the size the +/// descriptor claims and hashes to the layer's `diffId`, with the record +/// count the inscription states. +/// +/// What none of this establishes is *who published any of it*. The history's +/// digests are attested by the newest inscription and nothing here verifies a +/// signature over that inscription — signatures are Phase 5, and a verifier +/// claiming provenance from digests alone would be worse than this sentence. +#[derive(Debug, Clone)] +pub struct Verified { + /// The inscription every layer was checked against. + pub inscription: Inscription, + /// The stele's identity — what a Phase 5 signature will be over. + pub identity: Digest, + /// Compressed bytes streamed and checked, counted off the wire. + pub compressed_bytes: u64, +} + +/// Check a published stele's digests: manifest against inscription, then every +/// blob against both of its digests. +/// +/// The whole report is [`Verified`]; the whole verdict is the `Result`. A +/// failing layer comes back as [`Error::LayerVerification`], naming the layer's +/// kind and scope, because the exit code and the offending layer are the two +/// things a caller is here for. +/// +/// **Never call this from inside an async context.** See [`open`]. +pub fn verify(registry: &Registry, point: Point) -> Result { + let stele = point.pull(registry)?; + let inscription = stele.read_inscription()?; + let blobs = stele.blob_index()?; + + // The profile's record ceiling, not the protocol's default, for the reason + // a restore uses it: a verifier reading under a tighter limit than the + // publisher wrote under would refuse this profile's own steles. + let limits = Limits { + max_record: crate::MAX_RECORD, + ..Limits::default() + }; + + let mut compressed_bytes = 0u64; + + for descriptor in &inscription.layers { + let digests = check_layer(&stele, &blobs, descriptor, limits).map_err(|source| { + Error::LayerVerification { + kind: descriptor.kind.clone(), + scope: descriptor.scope.to_string(), + source: Box::new(source), + } + })?; + + compressed_bytes += digests.compressed_size; + } + + Ok(Verified { + identity: inscription.digest()?, + inscription, + compressed_bytes, + }) +} + +/// Stream one layer end to end and hold every claim against the bytes. +/// +/// `LayerReader::finish` carries most of it — the header record, the +/// decompression ceiling, the `diffId`, the uncompressed size and the record +/// count. What it cannot know is what the *manifest* said, so the two +/// transport facts are checked here: the bytes hash to the blob digest the +/// manifest addresses the layer by, and their count is the compressed size it +/// claims. `pull_blob` already refuses a digest mismatch in flight; comparing +/// the digest again out of [`LayerDigests`] costs nothing and keeps the check +/// in code this crate can point at. +fn check_layer( + stele: &Stele, + blobs: &BlobIndex, + descriptor: &LayerDescriptor, + limits: Limits, +) -> Result { + let reader = stele.stream_layer(blobs, &DolosProfile, descriptor, limits)?; + let digests = reader.finish()?; + + let mismatch = |what: &str, manifest: String, blob: String| { + Error::Stelae(stelae::Error::ManifestMismatch(format!( + "the manifest says this layer's {what} is {manifest} and the blob's is {blob}", + ))) + }; + + if let Some(named) = blobs.blob_for(&descriptor.diff_id) { + if digests.blob_digest != named { + return Err(mismatch( + "blob digest", + named.to_string(), + digests.blob_digest.to_string(), + )); + } + } + + if let Some(claimed) = stele.compressed_size(blobs, descriptor)? { + if digests.compressed_size != claimed { + return Err(mismatch( + "compressed size", + claimed.to_string(), + digests.compressed_size.to_string(), + )); + } + } + + Ok(digests) +} + +/// What a repository holds at a point, read without pulling a single layer. +#[derive(Debug, Clone)] +pub struct Inspected { + pub inscription: Inscription, + /// The stele's identity: sha256 of the canonical inscription. + pub identity: Digest, + /// Each layer's compressed size as the manifest carries it, in inscription + /// order. `None` where the manifest claims something impossible — a + /// negative size — which a report prints as unknown rather than as a + /// number. + pub compressed: Vec>, + /// Compressed bytes across every layer, as the manifest reports them. + pub total_compressed: u64, +} + +/// Read a stele's two documents and nothing else. +/// +/// The first thing an operator does when a restore behaves oddly, priced +/// accordingly: two small GETs — the manifest and the config blob — and no +/// layer is fetched. Everything the pull verifies ([`Registry::pull`]) is +/// verified here too, so what is reported is a stele, not merely a manifest +/// that looked like one. +/// +/// **Never call this from inside an async context.** See [`open`]. +pub fn inspect(registry: &Registry, point: Point) -> Result { + let stele = point.pull(registry)?; + let inscription = stele.read_inscription()?; + let blobs = stele.blob_index()?; + + let compressed = inscription + .layers + .iter() + .map(|descriptor| stele.compressed_size(&blobs, descriptor)) + .collect::, _>>()?; + + Ok(Inspected { + identity: inscription.digest()?, + total_compressed: stele.total_compressed_size(), + inscription, + compressed, + }) +} + /// Publish `plan` into `registry`, chained to whatever is already there. /// /// Reads the repository's moving tag first: an absent one starts a history, a @@ -546,84 +759,12 @@ fn key(kind: &str, scope: &serde_json::Value) -> Result<(String, String), Error> Ok((kind.to_owned(), canonical)) } -/// The history a stele at `sequence` carries when it follows `previous`. -/// -/// The three legal readings of a repository, 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. 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. -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" - } - _ => { - "a publish must follow the repository's latest stele, and this one would leave a gap \ - no later stele could close" - } - }; - - Err(Error::HistoryBreak { - latest, - publishing: sequence, - reason, - }) -} - -/// Refuse a predecessor from another chain. -/// -/// A repository holding two networks' steles is an operator fault, and the -/// check costs nothing: the previous stele's `position` already names its -/// network, and reading it is the same function a restore uses. -fn same_network(previous: &Inscription, plan: &Plan) -> Result<(), Error> { - let found = crate::read_position(&previous.position)?.network; - - if found.magic() != plan.network.magic() { - return Err(Error::NetworkMismatch { - expected: plan.network.magic(), - found: found.magic(), - }); - } - - Ok(()) -} - #[cfg(test)] mod tests { - use dolos_core::{BlockHash, ChainPoint}; use serde_json::json; use stelae::Profile as _; use super::*; - use crate::Network; fn inscription(sequence: u64, history: Vec) -> Inscription { let mut inscription = Inscription::new( @@ -638,82 +779,6 @@ mod tests { 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 = 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")); - } - /// Only the epoch kinds are inheritable, and a state shard is excluded by /// this table rather than by the caller remembering to. #[test] @@ -811,60 +876,6 @@ mod tests { assert!(message.contains("2 records"), "{message}"); } - /// The only thing standing between a publisher and a history chained onto - /// another chain's stele. - /// - /// A publish reads its own magic from genesis and the predecessor's from - /// the predecessor. If they were allowed to differ, the new inscription - /// would attest a chain of steles from a network it has never seen — and - /// nothing downstream re-checks it, because `history` entries carry a - /// sequence and a digest and no position at all. - #[test] - fn a_predecessor_from_another_network_is_refused() { - let preview = Network::for_magic(crate::PREVIEW_MAGIC); - let preprod = Network::for_magic(crate::PREPROD_MAGIC); - - let plan = plan_at(preview.clone()); - - // Built by `crate::position`, not by the `inscription` helper's shape: - // `same_network` reads it back through `crate::read_position`, which - // the helper's bare `{"epoch": n}` would not survive. - let stele = |network: &Network| { - let mut previous = inscription(3, vec![]); - - previous.position = crate::position( - network, - &ChainPoint::Specific(250, BlockHash::from([0xab; 32])), - 2, - ) - .unwrap(); - - previous - }; - - same_network(&stele(&preview), &plan).unwrap(); - - let err = same_network(&stele(&preprod), &plan).unwrap_err(); - - assert!( - matches!( - err, - Error::NetworkMismatch { expected, found } - if expected == preview.magic() && found == preprod.magic() - ), - "{err:?}" - ); - } - - fn plan_at(network: Network) -> Plan { - Plan { - network, - cursor: ChainPoint::Specific(250, BlockHash::from([0xab; 32])), - sequence: 3, - epochs: vec![], - } - } - /// A point round-trips through the profile's own tag rendering, in both /// directions. /// diff --git a/crates/snapshot/tests/export.rs b/crates/snapshot/tests/export.rs index ead08151..54d66800 100644 --- a/crates/snapshot/tests/export.rs +++ b/crates/snapshot/tests/export.rs @@ -19,6 +19,11 @@ //! stores and to the builtin memory stores publish the same inscription //! digest. This is the claim ADR-004 rests on and the first place it is //! measured rather than asserted. +//! 5. **A stele can be reproduced without being stored.** The discarding writer +//! walks the same stores and produces the same canonical document, byte for +//! byte, as the publish that wrote one to disk — which is what `dolos +//! snapshot digest` is, and what makes an independent verifier possible at +//! all. mod common; mod node; @@ -40,7 +45,7 @@ use dolos_snapshot::{ }; use dolos_testing::toy_domain::{FjallStores, MemoryStores, ToyDomain, ToyStores}; use node::{export_to, harness, plan_for}; -use stelae::{dir::SteleDir, SteleReader}; +use stelae::{dir::SteleDir, Discarding, SteleReader}; /// The identity of an export over an empty store set at [`SKELETON_POINT`]. const GOLDEN_SKELETON: &str = @@ -570,3 +575,140 @@ const CANONICAL_SKELETON: &str = concat!( r#"{"diffId":"sha256:0eb6fed603f60e527eb9622e04d72a74c21dcd7ee88e98b95ae6f8895736d5fd","kind":"state","mediaType":"application/vnd.dolos.stele.state.v1+zstd","records":1,"scope":{"shard":15},"uncompressedSize":40}"#, r#"],"parameters":{"indexKeyHash":"xxh3-64","stateShards":16},"position":{"epoch":2,"network":{"magic":764824073,"name":"mainnet"},"point":{"hash":"0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b","slot":250}},"profile":{"name":"io.txpipe.dolos.cardano","version":1},"schema":1,"sequence":3}"#, ); + +// -------------------------------------------------------------------------- +// 5. A stele reproduced without being stored +// -------------------------------------------------------------------------- + +/// The check that the discarding writer is *faithful* rather than merely fast, +/// and no other check stands in for it. +/// +/// Two exports over one store set: one into a directory, which is +/// `snapshot publish --output-dir`, and one into nothing, which is +/// `snapshot digest`. The comparison is on the canonical bytes rather than on +/// the digest — the digest is a function of them, so an equal digest is +/// implied, while an unequal document says *where* in a two-kilobyte JSON the +/// two disagreed. +/// +/// Run over the harness ledger rather than the skeleton on purpose: an empty +/// store set exercises the layer *skeleton*, and what a discarding writer could +/// plausibly get wrong is a layer with records in it — the compressor's state, +/// the record count, the uncompressed size. +#[test] +fn a_discarding_export_reproduces_what_a_publish_stores() { + let domain: ToyDomain = harness(); + let plan = plan_for(&domain); + + let temp = tempfile::tempdir().unwrap(); + let stored = export_to(temp.path(), &domain); + + let reproduced = export::export( + &Discarding, + &plan, + domain.archive(), + domain.state(), + domain.indexes(), + None, + &export::First, + ) + .unwrap(); + + // Non-trivial: a stele of empty layers would compare equal for the wrong + // reason. + assert!( + stored.uncompressed_size() > 1024, + "the fixture has to carry records for this to prove anything" + ); + + assert_eq!( + String::from_utf8(reproduced.canonicalize().unwrap()).unwrap(), + String::from_utf8(stored.canonicalize().unwrap()).unwrap(), + ); + + assert_eq!(reproduced.digest().unwrap(), stored.digest().unwrap()); + + // Nothing was written on the reproduction's behalf: the only stele on disk + // is the one the directory export made, and it holds exactly its own + // blobs. + let blobs = std::fs::read_dir(temp.path().join("blobs").join("sha256")) + .unwrap() + .count(); + + assert_eq!(blobs, distinct_layers(&stored)); +} + +/// A verifier checking a *different* stele than the one published is the +/// failure `--epochs` exists to prevent, so the two commands share one range +/// type and one restriction. +/// +/// The restriction goes through `Plan::restrict_epochs` on both sides here for +/// the same reason: a reproduction over a narrower selection is a different +/// document, and it has to be the *same* different document. +#[test] +fn a_restricted_reproduction_matches_the_same_restricted_publish() { + let domain: ToyDomain = harness(); + let plan = plan_for(&domain).restrict_epochs(Some(1), None); + + assert!( + plan.epochs.is_empty(), + "the harness ledger lives in epoch zero; selecting above it selects nothing" + ); + + let temp = tempfile::tempdir().unwrap(); + + let stored = export::publish( + temp.path(), + &plan, + domain.archive(), + domain.state(), + domain.indexes(), + None, + ) + .unwrap(); + + let reproduced = export::export( + &Discarding, + &plan, + domain.archive(), + domain.state(), + domain.indexes(), + None, + &export::First, + ) + .unwrap(); + + // The state tip alone, which is a legitimate publish and the narrowest one + // there is. + assert_eq!(stored.layers.len(), STATE_SHARDS as usize); + + assert_eq!( + stored.canonicalize().unwrap(), + reproduced.canonicalize().unwrap() + ); + + // And it is genuinely a different stele from the unrestricted one, so the + // equality above is not the equality of two full exports. + let whole = export::export( + &Discarding, + &plan_for(&domain), + domain.archive(), + domain.state(), + domain.indexes(), + None, + &export::First, + ) + .unwrap(); + + assert_ne!(whole.digest().unwrap(), reproduced.digest().unwrap()); +} + +/// How many blobs a stele's layers occupy: distinct `diffId`s, because two +/// layers with identical content are one content-addressed file. +fn distinct_layers(inscription: &stelae::Inscription) -> usize { + inscription + .layers + .iter() + .map(|layer| layer.diff_id) + .collect::>() + .len() +} diff --git a/crates/snapshot/tests/node/mod.rs b/crates/snapshot/tests/node/mod.rs index 1057ac56..31b0d110 100644 --- a/crates/snapshot/tests/node/mod.rs +++ b/crates/snapshot/tests/node/mod.rs @@ -102,6 +102,107 @@ impl Blank { } } +// The re-export mirrors the module's own rule stated above: every test binary +// compiles this file in full, so the suites that never open a registry see an +// import they do not use. +#[cfg(feature = "oci")] +#[allow(unused_imports)] +pub use registry_node::Node; + +/// The registry suites' node: the harness ledger and the two chain points it +/// publishes from. +/// +/// Shared by `tests/publish.rs` and `tests/snapshot_verify.rs`, because a +/// stele the one suite publishes is what the other verifies, and a second +/// copy of the fixture would be a second answer to "where do the two plans +/// stand". +#[cfg(feature = "oci")] +mod registry_node { + use dolos_core::{BlockHash, ChainPoint, Domain as _}; + use dolos_snapshot::{ + export::Plan, + registry::{self, Published, Registry}, + Error, Network, + }; + use dolos_testing::toy_domain::{MemoryStores, ToyDomain}; + + use super::harness; + + /// The harness ledger and the two plans it publishes: sequence 1 standing + /// on the epoch-0 boundary, and sequence 2 one slot past it. + /// + /// That the first cursor sits on the boundary is not a convenience: a + /// stele cut mid-epoch clamps its last window to the cursor, so the same + /// epoch published later in full has a different scope and is correctly + /// rebuilt rather than inherited. See `tests/publish.rs` for the longer + /// argument. + pub struct Node { + pub domain: ToyDomain, + pub first: Plan, + pub second: Plan, + } + + impl Node { + pub fn build() -> Self { + let domain = harness::(); + + let summary = + dolos_cardano::eras::load_chain_summary_from_state(domain.state()).unwrap(); + + let magic = u64::from(domain.genesis().network_magic()); + let network = Network::for_magic(magic); + let boundary = summary.epoch_start(1); + + // Any hash will do: `position` needs one to exist, and nothing in + // an export reads it back out of the store. + let point = |slot| ChainPoint::Specific(slot, BlockHash::new([0xab; 32])); + + let first = Plan::new(&summary, network.clone(), point(boundary - 1)).unwrap(); + let second = Plan::new(&summary, network, point(boundary)).unwrap(); + + assert_eq!(first.sequence, 1); + assert_eq!(second.sequence, 2); + assert_eq!( + first.epochs, + second.epochs[..1], + "epoch 0's window has to be the same in both, or there is nothing to inherit" + ); + + Self { + domain, + first, + second, + } + } + + pub fn publish(&self, repository: &Registry, plan: &Plan, rebuild: bool) -> Published { + registry::publish( + repository, + plan, + self.domain.archive(), + self.domain.state(), + self.domain.indexes(), + None, + rebuild, + ) + .unwrap() + } + + pub fn refuse(&self, repository: &Registry, plan: &Plan) -> Error { + registry::publish( + repository, + plan, + self.domain.archive(), + self.domain.state(), + self.domain.indexes(), + None, + false, + ) + .unwrap_err() + } + } +} + pub fn plan_for(domain: &ToyDomain) -> Plan { export::plan(domain.state(), domain.genesis().network_magic() as u64).unwrap() } diff --git a/crates/snapshot/tests/publish.rs b/crates/snapshot/tests/publish.rs index 3c22da7f..d533e1ec 100644 --- a/crates/snapshot/tests/publish.rs +++ b/crates/snapshot/tests/publish.rs @@ -28,6 +28,14 @@ //! 4. **A different predecessor is a different digest**, deliberately — //! `history` is inside the canonical document, so a repository's identity is //! path-dependent. +//! 5. **A stele published with reuse on can be reproduced from stores alone.** +//! The trust gap the incremental publish opened: layers inherited rather +//! than rebuilt are attested without being reproduced, and until something +//! reproduces them nothing but the publisher has ever checked the +//! attestation. `export::reproduce` is that something. +//! 6. **A repository already at this node's sequence is not a fault.** The +//! detection a publisher on a timer needs, read off the same moving tag a +//! publish reads. //! //! ## Why the two cursors sit where they do //! @@ -48,92 +56,19 @@ mod node; mod registry_fixture; -use dolos_core::{BlockHash, ChainPoint, Domain as _}; +use dolos_core::Domain as _; use dolos_snapshot::{ - export::Plan, - registry::{self, Published}, - DolosProfile, Error, Network, BLOCKS, INDEXES, LOGS, STATE_SHARDS, + export::{self, Following, Predecessor as _, Standing}, + registry, DolosProfile, Error, BLOCKS, INDEXES, LOGS, STATE_SHARDS, }; -use dolos_testing::toy_domain::{MemoryStores, ToyDomain}; -use stelae::{oci::Registry, SteleReader as _}; +use stelae::SteleReader as _; -use node::harness; +use node::Node; use registry_fixture::Fixture; /// Layers a publish of one epoch writes: three epoch kinds plus the state tip. const PER_PUBLISH: usize = 3 + STATE_SHARDS as usize; -// --------------------------------------------------------------------------- -// The node, and the two chain points it publishes from -// --------------------------------------------------------------------------- - -/// The harness ledger and the two plans it publishes: sequence 1 standing on -/// the epoch-0 boundary, and sequence 2 one slot past it. -struct Node { - domain: ToyDomain, - first: Plan, - second: Plan, -} - -impl Node { - fn build() -> Self { - let domain = harness::(); - - let summary = dolos_cardano::eras::load_chain_summary_from_state(domain.state()).unwrap(); - - let magic = u64::from(domain.genesis().network_magic()); - let network = Network::for_magic(magic); - let boundary = summary.epoch_start(1); - - // Any hash will do: `position` needs one to exist, and nothing in an - // export reads it back out of the store. - let point = |slot| ChainPoint::Specific(slot, BlockHash::new([0xab; 32])); - - let first = Plan::new(&summary, network.clone(), point(boundary - 1)).unwrap(); - let second = Plan::new(&summary, network, point(boundary)).unwrap(); - - assert_eq!(first.sequence, 1); - assert_eq!(second.sequence, 2); - assert_eq!( - first.epochs, - second.epochs[..1], - "epoch 0's window has to be the same in both, or there is nothing to inherit" - ); - - Self { - domain, - first, - second, - } - } - - fn publish(&self, repository: &Registry, plan: &Plan, rebuild: bool) -> Published { - registry::publish( - repository, - plan, - self.domain.archive(), - self.domain.state(), - self.domain.indexes(), - None, - rebuild, - ) - .unwrap() - } - - fn refuse(&self, repository: &Registry, plan: &Plan) -> Error { - registry::publish( - repository, - plan, - self.domain.archive(), - self.domain.state(), - self.domain.indexes(), - None, - false, - ) - .unwrap_err() - } -} - // --------------------------------------------------------------------------- // Done criteria 1 and 2 // --------------------------------------------------------------------------- @@ -409,3 +344,148 @@ fn layer<'a>( .find(|layer| layer.kind == kind && layer.scope["epoch"] == epoch) .unwrap_or_else(|| panic!("no {kind} layer for epoch {epoch}")) } + +// --------------------------------------------------------------------------- +// The trust gap, closed +// --------------------------------------------------------------------------- + +/// A stele published **with reuse on** is reproduced from the stores alone. +/// +/// This is the check the incremental publish deferred. `a_second_publish` shows +/// that the second stele inherits epoch 0's three layers and never opens the +/// store for them; `reuse_and_a_forced_rebuild_agree_on_the_digest` shows the +/// same publisher rebuilding them and agreeing. Neither is an *independent* +/// reproduction: both go through the registry publisher, and one of them is +/// the very code whose inheritance is in question. +/// +/// Here the reproduction touches no registry at all. The discarding writer +/// walks the stores, builds every layer including the ones the publish +/// inherited, chains onto the predecessor's inscription through the same +/// `history_for` the publish used — and has to arrive at the digest that is in +/// the repository. +/// +/// The predecessor is an input rather than something the reproduction works +/// out, and it has to be: `history` is inside the canonical document, so a +/// verifier that guessed the chain would compute a digest that is correct for a +/// stele nobody published. That is the residual independence gap, and it is +/// what a signature closes rather than this. +#[test] +#[ignore] +fn a_stele_published_with_reuse_is_reproduced_from_the_stores() { + let fixture = Fixture::spawn(); + let node = Node::build(); + let repository = fixture.repository("dolos/reproduced"); + + let first = node.publish(&repository, &node.first, false); + let second = node.publish(&repository, &node.second, false); + + assert_eq!( + second.layers_reused, 3, + "there is nothing to reproduce unless the publish inherited something" + ); + + // What a verifier is handed: the predecessor's canonical bytes, exactly as + // they sit in the repository's config blob. + let canonical = first.inscription.canonicalize().unwrap(); + let following = Following::read(&canonical, &node.second).unwrap(); + + assert_eq!(following.history().len(), 1); + + let reproduced = export::reproduce( + &node.second, + node.domain.archive(), + node.domain.state(), + node.domain.indexes(), + None, + &following, + ) + .unwrap(); + + assert_eq!( + reproduced.canonicalize().unwrap(), + second.inscription.canonicalize().unwrap(), + "the reproduction and the published stele are not the same document" + ); + + assert_eq!(reproduced.digest().unwrap(), second.identity); + + eprintln!( + "published (3 layers inherited) {} == reproduced from stores {}", + second.identity, + reproduced.digest().unwrap(), + ); + + // And the reproduction is not trivially right: chained onto nothing it is a + // different document, which is the path-dependence the history field is + // for. + let unchained = export::reproduce( + &node.second, + node.domain.archive(), + node.domain.state(), + node.domain.indexes(), + None, + &export::First, + ) + .unwrap(); + + assert_ne!(unchained.digest().unwrap(), second.identity); +} + +// --------------------------------------------------------------------------- +// Incremental detection +// --------------------------------------------------------------------------- + +/// The four readings of a repository, against a real one. +/// +/// The arithmetic is unit-tested in `export`; what this adds is that +/// `registry::standing` reads the *same* moving tag a publish reads and gets +/// the same sequence out of it — the half of the comparison a unit test cannot +/// supply. +#[test] +#[ignore] +fn a_publisher_can_ask_where_it_stands() { + let fixture = Fixture::spawn(); + let node = Node::build(); + let repository = fixture.repository("dolos/standing"); + + assert_eq!( + registry::standing(&repository, &node.first).unwrap(), + Standing::Empty, + "nothing published yet" + ); + + node.publish(&repository, &node.first, false); + + // The ordinary case for a job on a timer: the node has not entered a new + // epoch since the last run. It used to arrive as the same refusal a skipped + // epoch raises. + assert_eq!( + registry::standing(&repository, &node.first).unwrap(), + Standing::UpToDate { latest: 1 }, + ); + + assert_eq!( + registry::standing(&repository, &node.second).unwrap(), + Standing::Next { latest: 1 }, + ); + + // A node three epochs ahead of the repository. + let mut skipped = node.second.clone(); + skipped.sequence = 4; + + assert_eq!( + registry::standing(&repository, &skipped).unwrap(), + Standing::Ahead { + latest: 1, + distance: 3 + }, + ); + + // And the refusal that still stands behind it names both sequences and the + // distance. + let message = node.refuse(&repository, &skipped).to_string(); + + assert!(message.contains('1'), "{message}"); + assert!(message.contains('4'), "{message}"); + assert!(message.contains("3 sequences ahead"), "{message}"); +} diff --git a/crates/snapshot/tests/registry_fixture/mod.rs b/crates/snapshot/tests/registry_fixture/mod.rs index b276c7f1..420d70b7 100644 --- a/crates/snapshot/tests/registry_fixture/mod.rs +++ b/crates/snapshot/tests/registry_fixture/mod.rs @@ -178,6 +178,17 @@ impl Fixture { panic!("the registry never answered GET /v2/ on {address}"); } + /// Where the registry listens, for a suite that has to speak the + /// distribution API directly. + /// + /// The verify suite plants tampered manifests, and planting one is not + /// something the transport offers — deliberately, so nothing in the + /// production tree learns how to publish a manifest that disagrees with + /// its inscription. + pub fn address(&self) -> String { + format!("127.0.0.1:{}", self.port) + } + /// Open a transport onto one repository in this registry. /// /// A fresh transport per call, even for a name already opened: the pending diff --git a/crates/snapshot/tests/snapshot_verify.rs b/crates/snapshot/tests/snapshot_verify.rs new file mode 100644 index 00000000..a93970de --- /dev/null +++ b/crates/snapshot/tests/snapshot_verify.rs @@ -0,0 +1,567 @@ +//! Verifying and inspecting a published stele, end to end. +//! +//! Everything here needs a **registry**, and spawns one: `docker run` of an +//! OCI Distribution server, torn down on the way out. So everything here is +//! `#[ignore]`d, and an `#[ignore]`d test that was never executed proves +//! nothing — run it with: +//! +//! ```text +//! cargo test -p dolos-snapshot --features oci --test snapshot_verify -- --ignored --nocapture +//! ``` +//! +//! ## What this suite proves +//! +//! 1. **A freshly published stele verifies clean**, inherited layers included: +//! the manifest and the inscription agree, and every blob streams back to +//! both of its digests. +//! 2. **Each of the three tampers is refused with the offence named**: a blob +//! that is not the layer it is addressed as, a manifest `diffId` annotation +//! disagreeing with the inscription, and a history that skips a sequence. +//! The tampered artifacts are planted through the raw distribution API, +//! because the transport refuses to write any of them — which is the point. +//! 3. **A reproduction from the publisher's own stores matches**, and a store +//! standing at a different epoch is refused before a single layer is +//! rebuilt. +//! 4. **An inspection reports what the manifest carries**, and its canonical +//! JSON is exactly what `digest --chain-from` takes. +//! +//! ## The registry is content-addressed, so tampering means manifests +//! +//! A stored blob cannot be altered without changing its name; what *can* lie +//! is the mutable tag, and everything it points at. Each tamper here is +//! therefore a manifest (or config blob) rewritten under `latest` — the shape +//! of attack a verifier actually faces. + +#![cfg(feature = "oci")] + +mod node; +mod registry_fixture; + +use dolos_core::Domain as _; +use dolos_snapshot::{ + export::{self, Following, Predecessor as _}, + registry::{self, Point}, + Error, +}; +use serde_json::json; +use stelae::Digest; + +use node::Node; +use registry_fixture::Fixture; + +// --------------------------------------------------------------------------- +// Done criterion 3: verify --repo +// --------------------------------------------------------------------------- + +/// The pass, before the refusals mean anything. +/// +/// The stele verified at `latest` is the interesting one: it inherited epoch +/// 0's three layers from its predecessor, so a clean verification here is the +/// transport half of the trust-gap closure — every attested blob streamed and +/// checked, whether or not this publish built it. +#[test] +#[ignore = "spawns a registry"] +fn a_freshly_published_stele_verifies_clean() { + let fixture = Fixture::spawn(); + let node = Node::build(); + let repository = fixture.repository("dolos/verify-clean"); + + let first = node.publish(&repository, &node.first, false); + let second = node.publish(&repository, &node.second, false); + + assert_eq!( + second.layers_reused, 3, + "the interesting stele is one with inherited layers" + ); + + let verified = registry::verify(&repository, Point::Latest).unwrap(); + + assert_eq!(verified.identity, second.identity); + assert_eq!( + verified.inscription.layers.len(), + second.inscription.layers.len() + ); + assert!(verified.compressed_bytes > 0); + + // The immutable tag reads the predecessor back just as clean. + let predecessor = registry::verify(&repository, Point::Epoch(1)).unwrap(); + + assert_eq!(predecessor.identity, first.identity); + + eprintln!( + "verified latest = {} ({} layers, {} compressed bytes) and epoch-1 = {}", + verified.identity, + verified.inscription.layers.len(), + verified.compressed_bytes, + predecessor.identity, + ); +} + +/// A blob that is not the layer it is addressed as, refused with the layer +/// named. +/// +/// The closest a content-addressed registry comes to a corrupted blob: the +/// bytes are correctly named — so the in-flight blob-digest check passes and +/// the two documents still agree — and they are simply not the layer. Only +/// the streaming check can catch it, and the refusal has to say which layer. +#[test] +#[ignore = "spawns a registry"] +fn a_blob_that_is_not_the_layer_is_refused_with_the_layer_named() { + let fixture = Fixture::spawn(); + let distribution = Distribution::new(&fixture); + let node = Node::build(); + + let name = "dolos/verify-corrupt"; + let repository = fixture.repository(name); + + node.publish(&repository, &node.first, false); + + // A well-formed zstd frame over bytes nobody attested, through the same + // pipeline a real layer takes so nothing about its shape gives it away + // before the content does. + let blob = { + use std::io::Write as _; + + let mut writer = stelae::LayerWriter::new(Vec::new(), 3).unwrap(); + writer.write_all(b"not the layer anybody attested").unwrap(); + writer.finish().unwrap().0 + }; + + let planted = distribution.put_blob(name, &blob); + + // Point the manifest's first layer at it, leaving the annotations and the + // inscription untouched: the pull's document cross-check still passes, + // which is exactly why verify streams. + let mut manifest = distribution.manifest(name, "latest"); + manifest["layers"][0]["digest"] = json!(planted); + manifest["layers"][0]["size"] = json!(blob.len()); + distribution.put_manifest(name, "latest", &manifest); + + let err = registry::verify(&repository, Point::Latest).unwrap_err(); + let message = err.to_string(); + + assert!(matches!(err, Error::LayerVerification { .. }), "{err:?}"); + + // Layer 0 is epoch 0's blocks layer, and the message names it with its + // scope — the two things an operator needs to know what to re-publish. + assert!(message.contains("blocks"), "{message}"); + assert!(message.contains("epoch"), "{message}"); + + eprintln!("corrupted blob: {message}"); +} + +/// A manifest `diffId` annotation that disagrees with the inscription, +/// refused at the pull with the layer position and both identities named. +#[test] +#[ignore = "spawns a registry"] +fn a_diff_id_annotation_that_disagrees_is_refused() { + let fixture = Fixture::spawn(); + let distribution = Distribution::new(&fixture); + let node = Node::build(); + + let name = "dolos/verify-annotation"; + let repository = fixture.repository(name); + + node.publish(&repository, &node.first, false); + + let wrong = Digest::compute([0xee]).to_string(); + + let mut manifest = distribution.manifest(name, "latest"); + manifest["layers"][0]["annotations"]["store.stelae.layer.diffId"] = json!(wrong); + distribution.put_manifest(name, "latest", &manifest); + + let err = registry::verify(&repository, Point::Latest).unwrap_err(); + let message = err.to_string(); + + assert!(message.contains("layer 0"), "{message}"); + assert!(message.contains(&wrong), "{message}"); + + eprintln!("diffId disagreement: {message}"); +} + +/// A history that skips a sequence, refused before a single blob is fetched. +/// +/// The gapped chain is planted as a rewritten config blob under `latest` — +/// the protocol refuses to *produce* one, so the only way it exists is a +/// registry serving a document this code never wrote. The refusal happens at +/// parse, naming the gap. +#[test] +#[ignore = "spawns a registry"] +fn a_history_that_skips_a_sequence_is_refused() { + let fixture = Fixture::spawn(); + let distribution = Distribution::new(&fixture); + let node = Node::build(); + + let name = "dolos/verify-gap"; + let repository = fixture.repository(name); + + let first = node.publish(&repository, &node.first, false); + let second = node.publish(&repository, &node.second, false); + + // Sequence 3 claiming a history of 0 and 2: the entries themselves ascend, + // and the chain still skips sequence 1. + let mut gapped = serde_json::to_value(&second.inscription).unwrap(); + gapped["sequence"] = json!(3); + gapped["history"] = json!([ + {"sequence": 0, "inscriptionDigest": first.identity.to_string()}, + {"sequence": 2, "inscriptionDigest": second.identity.to_string()}, + ]); + + let config = stelae::inscription::canonical_json(&gapped).unwrap(); + let planted = distribution.put_blob(name, &config); + + let mut manifest = distribution.manifest(name, "latest"); + manifest["config"]["digest"] = json!(planted); + manifest["config"]["size"] = json!(config.len()); + distribution.put_manifest(name, "latest", &manifest); + + let err = registry::verify(&repository, Point::Latest).unwrap_err(); + let message = err.to_string(); + + assert!(message.contains("gap"), "{message}"); + assert!(message.contains('0') && message.contains('2'), "{message}"); + + eprintln!("skipped sequence: {message}"); +} + +// --------------------------------------------------------------------------- +// Done criterion 4: verify --repo --reproduce +// --------------------------------------------------------------------------- + +/// The reproduction passes against the store the stele was published from, +/// and a store standing at a different epoch is refused before the walk. +/// +/// The published stele inherited three layers it never rebuilt, so the pass +/// here is the whole trust-gap closure in one assertion: every attested layer +/// came back out of the stores byte-identical, history and all. The refusal +/// is the same ledger one epoch earlier — which is what "a store at a +/// different epoch" is — and it costs a comparison of two sequences, not +/// hours of compression. +#[test] +#[ignore = "spawns a registry"] +fn a_reproduction_passes_at_the_published_epoch_and_fails_at_another() { + let fixture = Fixture::spawn(); + let node = Node::build(); + let repository = fixture.repository("dolos/verify-reproduce"); + + node.publish(&repository, &node.first, false); + let second = node.publish(&repository, &node.second, false); + + assert_eq!( + second.layers_reused, 3, + "there is no trust gap to close unless the publish inherited something" + ); + + let verified = registry::verify(&repository, Point::Latest).unwrap(); + + let reproduced = export::verify_reproduction( + &verified.inscription, + &node.second, + node.domain.archive(), + node.domain.state(), + node.domain.indexes(), + None, + ) + .unwrap(); + + assert_eq!(reproduced.digest().unwrap(), verified.identity); + + let err = export::verify_reproduction( + &verified.inscription, + &node.first, + node.domain.archive(), + node.domain.state(), + node.domain.indexes(), + None, + ) + .unwrap_err(); + + let message = err.to_string(); + + assert!(matches!(err, Error::ReproductionMismatch { .. }), "{err:?}"); + assert!( + message.contains("sequence 2") && message.contains("sequence 1"), + "{message}" + ); + + eprintln!( + "reproduced {} == published {}; a store at another epoch: {message}", + reproduced.digest().unwrap(), + verified.identity, + ); +} + +// --------------------------------------------------------------------------- +// Done criterion 5: inspect --repo +// --------------------------------------------------------------------------- + +/// An inspection lists every layer with the compressed size the manifest +/// carries, and its canonical JSON chains a digest. +/// +/// The round trip is the criterion's own: inspect the predecessor, hand the +/// bytes to the same `Following::read` that `digest --chain-from` calls, and +/// the reproduction arrives at the digest of the stele published against it. +#[test] +#[ignore = "spawns a registry"] +fn an_inspection_reports_the_manifest_and_its_json_chains_a_digest() { + let fixture = Fixture::spawn(); + let node = Node::build(); + let repository = fixture.repository("dolos/inspect"); + + let first = node.publish(&repository, &node.first, false); + let second = node.publish(&repository, &node.second, false); + + let inspected = registry::inspect(&repository, Point::Latest).unwrap(); + + assert_eq!(inspected.identity, second.identity); + assert_eq!( + inspected.compressed.len(), + inspected.inscription.layers.len() + ); + + // Every layer carries a real compressed size — the inherited ones + // included, whose bytes this publish never moved — and they sum to the + // manifest's own total. + let mut total = 0; + + for (index, size) in inspected.compressed.iter().enumerate() { + let size = size.unwrap_or_else(|| panic!("layer {index} carries no compressed size")); + + assert!(size > 0, "layer {index}"); + total += size; + } + + assert_eq!(total, inspected.total_compressed); + + // The `--json` output is the canonical document, verbatim. + let predecessor = registry::inspect(&repository, Point::Epoch(1)).unwrap(); + + assert_eq!(predecessor.identity, first.identity); + + let canonical = predecessor.inscription.canonicalize().unwrap(); + let following = Following::read(&canonical, &node.second).unwrap(); + + assert_eq!(following.history().len(), 1); + + let reproduced = export::reproduce( + &node.second, + node.domain.archive(), + node.domain.state(), + node.domain.indexes(), + None, + &following, + ) + .unwrap(); + + assert_eq!( + reproduced.digest().unwrap(), + second.identity, + "inspect's own output chained the digest of the stele published against it" + ); + + eprintln!( + "inspected {} layers, {} compressed bytes; epoch-1's document chained to {}", + inspected.inscription.layers.len(), + inspected.total_compressed, + reproduced.digest().unwrap(), + ); +} + +// --------------------------------------------------------------------------- +// The raw distribution client the tampers need +// --------------------------------------------------------------------------- + +/// The distribution API, spoken directly. +/// +/// Exists because the tampered artifacts these tests plant are exactly the +/// documents the transport refuses to write. HTTP/1.0 with `Connection: +/// close` over a loopback socket, which keeps the parsing to a status line, +/// a header block and a read-to-EOF body — and assumes the `distribution` +/// server the fixture runs by default. +struct Distribution { + address: String, +} + +impl Distribution { + fn new(fixture: &Fixture) -> Self { + Self { + address: fixture.address(), + } + } + + /// Every socket operation is bounded, for the reason the fixture's + /// `wait_until_ready` bounds its own: a registry that accepts and then says + /// nothing turns an `#[ignore]`d end-to-end test into a CI job that hangs + /// rather than one that fails with a message. A loopback exchange of a few + /// test-sized blobs has no legitimate use for more than this. + const PATIENCE: std::time::Duration = std::time::Duration::from_secs(30); + + fn request( + &self, + method: &str, + path: &str, + headers: &[(&str, String)], + body: &[u8], + ) -> (u16, Vec<(String, String)>, Vec) { + use std::io::{Read as _, Write as _}; + + // `Fixture::address` is `127.0.0.1:{port}`, so this parses without + // resolution — which is what lets the connect itself be bounded. + let socket_address: std::net::SocketAddr = + self.address.parse().expect("a loopback address and a port"); + + let mut socket = + std::net::TcpStream::connect_timeout(&socket_address, Self::PATIENCE).unwrap(); + + socket.set_write_timeout(Some(Self::PATIENCE)).unwrap(); + socket.set_read_timeout(Some(Self::PATIENCE)).unwrap(); + + let credentials = + base64(format!("{}:{}", registry_fixture::USER, registry_fixture::PASSWORD).as_bytes()); + + let mut request = format!( + "{method} {path} HTTP/1.0\r\nHost: {}\r\nAuthorization: Basic {credentials}\r\n\ + Connection: close\r\nContent-Length: {}\r\n", + self.address, + body.len(), + ); + + for (name, value) in headers { + request.push_str(&format!("{name}: {value}\r\n")); + } + + request.push_str("\r\n"); + + socket.write_all(request.as_bytes()).unwrap(); + socket.write_all(body).unwrap(); + + let mut raw = Vec::new(); + socket.read_to_end(&mut raw).unwrap(); + + let boundary = raw + .windows(4) + .position(|window| window == b"\r\n\r\n") + .expect("an HTTP response has a header block"); + + let head = String::from_utf8_lossy(&raw[..boundary]).into_owned(); + let body = raw[boundary + 4..].to_vec(); + + let mut lines = head.lines(); + + let status: u16 = lines + .next() + .and_then(|line| line.split_whitespace().nth(1)) + .and_then(|code| code.parse().ok()) + .unwrap_or_else(|| panic!("no status line in {head:?}")); + + let headers = lines + .filter_map(|line| line.split_once(':')) + .map(|(name, value)| (name.trim().to_ascii_lowercase(), value.trim().to_owned())) + .collect(); + + (status, headers, body) + } + + fn manifest(&self, repository: &str, tag: &str) -> serde_json::Value { + let (status, _, body) = self.request( + "GET", + &format!("/v2/{repository}/manifests/{tag}"), + &[( + "Accept", + "application/vnd.oci.image.manifest.v1+json".to_owned(), + )], + &[], + ); + + assert_eq!(status, 200, "{}", String::from_utf8_lossy(&body)); + + serde_json::from_slice(&body).unwrap() + } + + fn put_manifest(&self, repository: &str, tag: &str, manifest: &serde_json::Value) { + let body = serde_json::to_vec(manifest).unwrap(); + + let (status, _, response) = self.request( + "PUT", + &format!("/v2/{repository}/manifests/{tag}"), + &[( + "Content-Type", + "application/vnd.oci.image.manifest.v1+json".to_owned(), + )], + &body, + ); + + assert_eq!(status, 201, "{}", String::from_utf8_lossy(&response)); + } + + /// Upload a blob and return the digest it is addressed by. + /// + /// The registry checks the digest against the bytes, so this can plant + /// wrong *content* but never a wrong name — which is the property the + /// corrupted-blob test leans on. + fn put_blob(&self, repository: &str, bytes: &[u8]) -> String { + let digest = Digest::compute(bytes).to_string(); + + let (status, headers, response) = self.request( + "POST", + &format!("/v2/{repository}/blobs/uploads/"), + &[], + &[], + ); + + assert_eq!(status, 202, "{}", String::from_utf8_lossy(&response)); + + let location = headers + .iter() + .find(|(name, _)| name == "location") + .map(|(_, value)| value.clone()) + .expect("an upload start answers with a Location"); + + // Absolute or path-relative, per the server's taste. + let location = location + .strip_prefix(&format!("http://{}", self.address)) + .unwrap_or(&location) + .to_owned(); + + let separator = if location.contains('?') { '&' } else { '?' }; + + let (status, _, response) = self.request( + "PUT", + &format!("{location}{separator}digest={digest}"), + &[("Content-Type", "application/octet-stream".to_owned())], + bytes, + ); + + assert_eq!(status, 201, "{}", String::from_utf8_lossy(&response)); + + digest + } +} + +/// Standard base64, for the Basic credential pair. +/// +/// Hand-rolled rather than imported: it is eleven lines, it runs in a test, +/// and a dev-dependency for one header is a dependency review nobody needs. +fn base64(bytes: &[u8]) -> String { + const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + + let mut encoded = String::new(); + + for chunk in bytes.chunks(3) { + let mut word = 0u32; + + for (index, byte) in chunk.iter().enumerate() { + word |= u32::from(*byte) << (16 - 8 * index); + } + + for index in 0..4 { + if index <= chunk.len() { + encoded.push(ALPHABET[((word >> (18 - 6 * index)) & 0x3f) as usize] as char); + } else { + encoded.push('='); + } + } + } + + encoded +} diff --git a/crates/stelae/src/lib.rs b/crates/stelae/src/lib.rs index 17717ee1..4bbacd72 100644 --- a/crates/stelae/src/lib.rs +++ b/crates/stelae/src/lib.rs @@ -30,8 +30,9 @@ //! - [`profile`] — the [`Profile`] trait plus the protocol's media-type, //! profile name and tag naming rules. //! - [`transport`] — the seam a stele is written through and read back from: -//! the two halves a profile uses, and the `diffId`→blob map both transports -//! answer differently. +//! the two halves a profile uses, the `diffId`→blob map both transports +//! answer differently, and the discarding writer that computes a stele's +//! identity without storing it. //! - [`plan`] — what a restore has already done and what it has left to fetch: //! the progress file, the resume rule, and remaining-bytes accounting. //! - [`dir`] — a minimal on-disk stele: the first implementation of that seam, @@ -67,7 +68,10 @@ pub use inscription::{ pub use layer::LayerReader; pub use plan::{Remaining, RestoreProgress, Resume}; pub use profile::{MediaType, Profile}; -pub use transport::{BlobIndex, LayerSpec, RecordSink, SteleReader, SteleWriter, WrittenLayer}; +pub use transport::{ + BlobIndex, Discarding, DiscardingSink, LayerSpec, RecordSink, SteleReader, SteleWriter, + WrittenLayer, +}; /// Artifact type of a stele manifest. Generic tooling discovers stelae of every /// profile by filtering on this; the inscription's `profile` field diff --git a/crates/stelae/src/transport.rs b/crates/stelae/src/transport.rs index 4b4247d2..1c59a879 100644 --- a/crates/stelae/src/transport.rs +++ b/crates/stelae/src/transport.rs @@ -34,6 +34,15 @@ //! rather than parameterizing the reader on "how to find a blob" is what makes //! that difference a cost and not an interface. //! +//! ## A third implementation that stores nothing +//! +//! [`Discarding`] is the write half with the storing taken out: every layer is +//! framed, hashed and compressed exactly as a publish would, and the bytes go +//! to [`std::io::sink`] instead of to a file or a registry. What comes back is +//! the identity — a [`WrittenLayer`] per layer and the inscription's digest +//! from [`SteleWriter::seal`] — which is the whole of what a reproduction needs +//! and none of what it would have to store. +//! //! ## What the seam deliberately does not carry //! //! No notion of *listing* what a repository holds, no tags beyond the two the @@ -42,11 +51,11 @@ //! question and should hold the transport-specific type. use crate::{ - digest::LayerDigests, - frame::{CanonicalCbor, Limits}, + digest::{LayerDigests, LayerWriter}, + frame::{CanonicalCbor, LayerHeader, Limits, SeqWriter}, inscription::{Inscription, LayerDescriptor}, layer::LayerReader, - profile::Profile, + profile::{checked_layer_media_type, Profile}, Digest, Error, }; @@ -228,6 +237,143 @@ pub trait SteleWriter { } } +/// A stele that computes its identity and stores nothing. +/// +/// The write half of the seam with the storing taken out. Every field of a +/// [`WrittenLayer`] is a function of the record stream — [`RecordSink::finish`] +/// reads the descriptor off the digests the pipeline computed, and only *then* +/// does a directory rename onto the content-addressed name — so a writer whose +/// sinks discard their bytes hands back the same descriptors as one that keeps +/// them. Sealing is the same identity a directory returns after writing +/// `inscription.json`: the sha256 of the canonical document. +/// +/// That is the whole of `dolos snapshot digest`. A verifier reproduces a +/// published stele's layers from its own stores and compares descriptors, +/// without provisioning the disk the stele would occupy — hundreds of gigabytes +/// on mainnet — and without touching a registry. +/// +/// ## It compresses +/// +/// The one shortcut this type must not take. `diffId`, `records` and +/// `uncompressedSize` are all fixed before zstd sees a byte, so a writer that +/// skipped compression would reproduce every field the inscription carries and +/// still not be doing what a publish does. The bug class this exists to catch +/// is the one that only appears when the same bytes go through the same +/// pipeline twice, and the pipeline is [`LayerWriter`] — hash in, compress, +/// hash out. What is dropped is the last step, the write to a file, and nothing +/// upstream of it. +/// +/// The cost of that honesty is real: a reproduction pays the compressor in full +/// and saves only the I/O. It buys the blob digest and the compressed size, +/// which a comparison against a registry manifest needs and a document cannot +/// supply. +/// +/// ## What it cannot answer +/// +/// Nothing that needs bytes back. There is no [`SteleReader`] half here and +/// there cannot be one: a stele that stored nothing has nothing to stream, and +/// a caller wanting both halves wants a real transport. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct Discarding; + +/// A layer being written into nothing, one record at a time. +/// +/// The mirror of [`crate::dir::LayerSink`] with the file removed. Nothing is +/// buffered here either — records are framed, hashed and compressed on the way +/// past — so reproducing a mainnet state shard costs the compressor's window +/// and one record, the same as publishing one. +pub struct DiscardingSink { + sequence: SeqWriter>, + kind: String, + media_type: String, + scope: serde_json::Value, +} + +impl RecordSink for DiscardingSink { + fn write_record(&mut self, record: &CanonicalCbor) -> Result<(), Error> { + self.sequence.write_record(record) + } + + fn records(&self) -> u64 { + self.sequence.count() + } + + /// Close the layer and hand back the descriptor it would have been + /// published under. + /// + /// Finishing the compressed frame is not skippable: zstd's epilogue is part + /// of the blob, so the blob digest and the compressed size are only correct + /// once the encoder has been closed. Everything else falls out of the same + /// [`LayerDigests`] a directory reads. + fn finish(self) -> Result { + let Self { + sequence, + kind, + media_type, + scope, + } = self; + + let count = sequence.count(); + let (_, digests) = sequence.into_inner().finish()?; + + Ok(WrittenLayer { + descriptor: LayerDescriptor { + kind, + media_type, + diff_id: digests.diff_id, + records: count, + uncompressed_size: digests.uncompressed_size, + scope, + }, + digests, + }) + } +} + +impl SteleWriter for Discarding { + type Sink = DiscardingSink; + + /// Open a layer that goes nowhere. + /// + /// The media type is still asked of the profile and still validated against + /// the naming rules, and the header record is still the first thing in the + /// sequence. Both are inside the layer's identity, so a reproduction that + /// skipped either would compute a `diffId` no publish ever produces. + fn layer_sink( + &self, + profile: &dyn Profile, + spec: &LayerSpec, + level: i32, + ) -> Result { + let media_type = checked_layer_media_type(profile, &spec.kind)?; + let header = LayerHeader::new(profile.name(), &spec.kind, spec.header_scope.clone()); + + let mut sink = DiscardingSink { + sequence: SeqWriter::with_max_record( + LayerWriter::new(std::io::sink(), level)?, + profile.max_record(), + ), + kind: spec.kind.clone(), + media_type, + scope: spec.scope.clone(), + }; + + sink.write_record(&header.encode()?)?; + + Ok(sink) + } + + /// Return the stele's identity without writing it down. + /// + /// [`Inscription::digest`] canonicalizes, which validates — so a document + /// that could not be sealed into a directory is refused here too, and a + /// reproduction never reports a digest over an inscription no publish could + /// have written. + fn seal(&self, _profile: &dyn Profile, inscription: &Inscription) -> Result { + inscription.digest() + } +} + /// The read half of a stele. /// /// Three operations, in this order: read the inscription, obtain the diff --git a/crates/stelae/tests/toy_profile.rs b/crates/stelae/tests/toy_profile.rs index e42e6101..2591d7b4 100644 --- a/crates/stelae/tests/toy_profile.rs +++ b/crates/stelae/tests/toy_profile.rs @@ -27,7 +27,7 @@ use stelae::{ digest::read_blob, dir::{BlobIndex, LayerSpec, SteleDir, WrittenLayer}, frame::{encode, CanonicalCbor, Limits}, - Compression, Error, Inscription, LayerDescriptor, LayerWriter, Profile, RecordSink, + Compression, Discarding, Error, Inscription, LayerDescriptor, LayerWriter, Profile, RecordSink, SteleReader, SteleWriter, }; @@ -1136,3 +1136,126 @@ fn golden_digests_pin_the_encoding() { ) ); } + +/// The discarding writer is faithful, not merely fast. +/// +/// The same records through both write halves: one into a directory, one into +/// nothing. Every field of the descriptor and every one of the four digests and +/// sizes has to agree — including the *blob* digest and the compressed size, +/// which only exist if zstd actually ran. That is the assertion this test is +/// for: a discarding writer that skipped compression would still reproduce +/// `diffId`, `records` and `uncompressedSize`, and would be exactly as wrong as +/// one that never ran at all. +/// +/// The seal is compared too, since a reproduction reports an identity: a +/// directory's comes from the bytes it wrote to `inscription.json`, and this +/// one from the document in hand. +#[test] +fn a_discarding_writer_reproduces_what_a_directory_stores() { + let temp = tempfile::tempdir().unwrap(); + let (stored, stored_digest) = write_stele(temp.path()); + + let (notes_header_scope, notes_scope) = notes_scopes(); + let (index_header_scope, index_scope) = index_scopes(); + + let notes: Vec = NOTES.iter().map(note_record).collect(); + + let reproduced_notes = Discarding + .write_layer( + &ToyProfile, + &LayerSpec::new("notes", notes_header_scope, notes_scope), + COMPRESSION_LEVEL, + ¬es, + ) + .unwrap(); + + let mut sorted: Vec<&Note> = NOTES.iter().collect(); + sorted.sort_by_key(|n| n.title); + + let mut index_sink = Discarding + .layer_sink( + &ToyProfile, + &LayerSpec::new("index", index_header_scope, index_scope), + COMPRESSION_LEVEL, + ) + .unwrap(); + + for note in sorted { + index_sink.write_record(&index_record(note)).unwrap(); + } + + let reproduced_index = index_sink.finish().unwrap(); + + // The stored stele's own blob digests, recovered the way a directory has + // to: by hashing the files it holds. Nothing in an inscription carries + // them, which is the point — they are transport, and a reproduction that + // agreed on identity while disagreeing on the compressed bytes would still + // publish a different blob. + let stored_blobs: BTreeSet = + std::fs::read_dir(temp.path().join("blobs").join(stelae::Digest::ALGORITHM)) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + + for (stored, reproduced) in stored + .layers + .iter() + .zip([&reproduced_notes, &reproduced_index]) + { + assert_eq!( + *stored, reproduced.descriptor, + "{}: the descriptor a publish would have written", + stored.kind, + ); + + assert!( + stored_blobs.contains(&reproduced.digests.blob_digest.to_hex()), + "{}: the reproduction named a blob the directory does not hold ({})", + stored.kind, + reproduced.digests.blob_digest, + ); + + let on_disk = std::fs::metadata( + temp.path() + .join("blobs") + .join(stelae::Digest::ALGORITHM) + .join(reproduced.digests.blob_digest.to_hex()), + ) + .unwrap() + .len(); + + assert_eq!( + reproduced.digests.compressed_size, on_disk, + "{}: the compressed size only exists if zstd ran", + stored.kind, + ); + } + + // And the identity, over a document assembled exactly as `write_stele` + // assembles it. + let mut inscription = Inscription::new( + &ToyProfile, + 3, + json!({"chapter": 3, "shelf": "east", "curator": {"name": "example", "since": 1998}}), + json!({"noteWidth": 40, "titleOrder": "byte"}), + Compression { + algo: "zstd".to_owned(), + level: COMPRESSION_LEVEL as i64, + }, + ); + + inscription.history = stored.history.clone(); + inscription.layers = vec![ + reproduced_notes.descriptor.clone(), + reproduced_index.descriptor.clone(), + ]; + + assert_eq!( + Discarding.seal(&ToyProfile, &inscription).unwrap(), + stored_digest, + ); + + // Nothing was written anywhere on the way: the only stele on disk is the + // one the directory wrote, and it has exactly its own two blobs. + assert_eq!(stored_blobs.len(), 2); +} diff --git a/src/bin/dolos/snapshot/digest.rs b/src/bin/dolos/snapshot/digest.rs new file mode 100644 index 00000000..7e7b2b0c --- /dev/null +++ b/src/bin/dolos/snapshot/digest.rs @@ -0,0 +1,187 @@ +//! What a stele *would* be, computed from local stores and no registry. +//! +//! `publish` writes a stele and reports its identity. This command computes the +//! identity and writes nothing — the same walk over the same stores through the +//! same framing, hashing and zstd pipeline, into a writer that keeps nothing +//! instead of into a directory or a repository. +//! +//! ## Why it exists +//! +//! A publish into a repository *inherits* the layers of epochs that have +//! already closed rather than rebuilding them, which means the newest +//! inscription describes layers the publishing node never read out of its own +//! stores. Every stele published that way carries an attestation nothing but +//! the publisher has ever reproduced. This is the command that reproduces it: +//! run somewhere else, over another node's stores, it either arrives at the +//! same digest or it does not, and the arithmetic in between is a determinism +//! job rather than a person. +//! +//! ## History is an input, not an inference +//! +//! `history` lives inside the canonical document, so a stele's identity depends +//! on which predecessor it chained from — the same stores chained differently +//! are two different digests, deliberately. A reproduction therefore has to be +//! *told* the chain: +//! +//! - by default there is none, `history: []`, which is the first stele of a +//! repository and what a publisher checks before it has ever published; +//! - `--chain-from FILE` reads a predecessor's canonical inscription off disk +//! and extends it by the rule a publish extends it by, which is +//! [`dolos_snapshot::export::Following`] and is one function shared with the +//! registry publisher rather than a second implementation of contiguity. +//! +//! ## stdout is the document +//! +//! The canonical inscription goes to `stdout` and everything else — the plan, +//! the layer count, the identity — to `stderr`, so `dolos snapshot digest > +//! stele.json` is a file a verifier can hash and `diff`. `--output FILE` writes +//! the same bytes to a file instead, which is what a determinism job comparing +//! two runs wants. +//! +//! ## What it costs +//! +//! What a publish costs, minus the I/O and the upload. Every layer is +//! compressed, because a reproduction that skipped compression would not be +//! doing the work a publish does. On mainnet that is hours. + +use std::path::PathBuf; + +use clap::Parser; +use dolos_core::config::RootConfig; +use dolos_snapshot::export::{self, Following, Plan, Predecessor}; +use miette::{Context as _, IntoDiagnostic as _}; + +use super::EpochRange; + +#[derive(Debug, Parser)] +pub struct Args { + /// epochs to build layers for, e.g. `500..520`, `500..=520`, `500..`, + /// `..520` or `500`; defaults to every epoch below the cursor, and must + /// match what the stele being reproduced was published with + #[arg(long, value_name = "RANGE")] + epochs: Option, + + /// canonical inscription of the stele this one follows, as + /// `inspect --json` or a published `inscription.json` holds it; without it + /// the reproduction carries no history, which is a different digest + #[arg(long, value_name = "FILE")] + chain_from: Option, + + /// write the canonical inscription to FILE instead of to stdout + #[arg(long, value_name = "FILE")] + output: Option, +} + +pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { + let stores = crate::common::open_data_stores(config) + .into_diagnostic() + .context("opening the data stores")?; + + let genesis = crate::common::open_genesis_files(&config.genesis)?; + + let plan = export::plan(&stores.state, u64::from(genesis.network_magic())) + .into_diagnostic() + .context("planning the reproduction")?; + + let plan = super::restrict(plan, args.epochs); + + super::report_plan(&plan)?; + + // Read and chained before the stores are walked: a predecessor this plan + // does not follow is an hour of compression the operator does not have to + // spend to find out. + let previous = args + .chain_from + .as_deref() + .map(|path| chain_from(path, &plan)) + .transpose()?; + + let predecessor: &dyn Predecessor = match &previous { + Some(following) => following, + None => &export::First, + }; + + match previous.as_ref().map(Predecessor::history) { + Some(history) => eprintln!("history: {} entries", history.len()), + None => eprintln!("history: none; this reproduction starts a chain"), + } + + let inscription = export::reproduce( + &plan, + &stores.archive, + &stores.state, + &stores.indexes, + None, + predecessor, + ) + .into_diagnostic() + .context("reproducing the stele")?; + + // The bytes, not a re-encoding of the content: they are what a verifier + // hashes, so anything that pretty-printed them would carry a digest nobody + // else computes. The identity is their sha256, which is what + // `Inscription::digest` is. + let canonical = inscription.canonicalize().into_diagnostic()?; + let identity = inscription.digest().into_diagnostic()?; + + eprintln!("layers: {}", inscription.layers.len()); + eprintln!( + "size: {} uncompressed bytes", + inscription.uncompressed_size() + ); + eprintln!("identity: {identity}"); + + match args.output.as_deref() { + Some(path) => { + std::fs::write(path, &canonical) + .into_diagnostic() + .with_context(|| format!("writing {}", path.display()))?; + + eprintln!("wrote {}", path.display()); + } + None => { + use std::io::Write as _; + + // Written rather than printed, with no trailing newline: the + // document is bytes, `digest > stele.json` has to hash to the + // identity reported above, and `--chain-from` refuses anything + // that is not the canonical encoding itself. + // + // Flushed explicitly, because no trailing newline means a + // line-buffered stdout still holds the tail of the document when + // this returns, and the flush at process exit discards its error: + // `digest > stele.json` onto a full disk would write a truncated + // inscription and exit zero. The `--output` arm gets this from + // `fs::write`. + let mut stdout = std::io::stdout(); + + stdout + .write_all(&canonical) + .into_diagnostic() + .context("writing the inscription to stdout")?; + + stdout + .flush() + .into_diagnostic() + .context("flushing the inscription to stdout")?; + } + } + + Ok(()) +} + +/// Read a predecessor's canonical inscription off disk and chain this plan onto +/// it. +/// +/// Everything the bytes have to satisfy is [`Following::read`]'s: the profile +/// crate owns what makes a predecessor usable, and this adds only the path to +/// the message. +fn chain_from(path: &std::path::Path, plan: &Plan) -> miette::Result { + let raw = std::fs::read(path) + .into_diagnostic() + .with_context(|| format!("reading {}", path.display()))?; + + Following::read(&raw, plan) + .into_diagnostic() + .with_context(|| format!("chaining onto {}", path.display())) +} diff --git a/src/bin/dolos/snapshot/inspect.rs b/src/bin/dolos/snapshot/inspect.rs new file mode 100644 index 00000000..22782ca3 --- /dev/null +++ b/src/bin/dolos/snapshot/inspect.rs @@ -0,0 +1,165 @@ +//! What a published stele contains, without pulling it. +//! +//! The first thing an operator does when a restore behaves oddly, priced +//! accordingly: two small GETs — the manifest and the config blob — and no +//! layer is fetched. Everything a pull verifies is verified on the way in, so +//! what is printed is a stele, not merely a manifest that looked like one. +//! +//! ## `--json` is the document, not a report +//! +//! It emits the canonical inscription verbatim on stdout — the exact bytes the +//! identity is the sha256 of. That makes it both the document a verifier +//! hashes and the thing that pipes into `digest --chain-from`, and it is why +//! nothing is appended to it, a newline included. +//! +//! ## No signers column +//! +//! ADR-004 lists signers among what an inspection reports. There are no +//! signatures to list until Phase 5 delivers `sign`, so the field is simply +//! absent — stated here and in the help, rather than printed as an empty +//! column that reads as "none". + +use clap::Parser; +use dolos_core::config::RootConfig; +use dolos_snapshot::registry::{self, Point, Repository}; +use miette::{Context as _, IntoDiagnostic as _}; + +#[derive(Debug, Parser)] +pub struct Args { + /// OCI repository holding the stele, e.g. + /// `oci://ghcr.io/txpipe/dolos-mainnet` + #[arg(long, value_name = "OCI_URL")] + repo: Repository, + + /// which stele to inspect: `latest`, or `epoch-N` for the stele published + /// at the end of epoch N + #[arg(long, value_name = "POINT", default_value = "latest")] + point: Point, + + /// talk to the repository over plaintext HTTP rather than HTTPS; for a + /// registry on a loopback address or a mirror inside a cluster, and for + /// nothing reachable from outside one + #[arg(long, action)] + insecure: bool, + + /// emit the canonical inscription verbatim instead of the report: the + /// exact bytes the identity is the sha256 of, which is what + /// `digest --chain-from` takes. Signers are not listed anywhere yet — + /// signatures arrive with phase 5 + #[arg(long, action)] + json: bool, +} + +pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { + let auth = crate::common::stele_registry_auth(&config.stelae)?; + + let registry = registry::open(&args.repo, args.insecure, auth) + .into_diagnostic() + .context("opening the repository")?; + + let inspected = registry::inspect(®istry, args.point) + .into_diagnostic() + .context("reading the stele")?; + + if args.json { + use std::io::Write as _; + + // The bytes, not a re-encoding: anything pretty-printed or + // newline-terminated would hash to an identity nobody published and + // be refused by `digest --chain-from`. + let canonical = inspected.inscription.canonicalize().into_diagnostic()?; + + // Flushed explicitly: with no trailing newline a line-buffered stdout + // still holds the tail of the document here, and the flush at process + // exit discards its error — `inspect --json > stele.json` onto a full + // disk or a closed pipe would otherwise write a truncated inscription + // and exit zero. + let mut stdout = std::io::stdout(); + + stdout + .write_all(&canonical) + .into_diagnostic() + .context("writing the inscription to stdout")?; + + stdout + .flush() + .into_diagnostic() + .context("flushing the inscription to stdout")?; + + return Ok(()); + } + + let inscription = &inspected.inscription; + + let position = dolos_snapshot::read_position(&inscription.position) + .into_diagnostic() + .context("reading the stele's position")?; + + println!("{} at {}", args.repo, args.point); + println!("identity: {}", inspected.identity); + println!( + "profile: {} v{}", + inscription.profile.name, inscription.profile.version + ); + println!("sequence: {}", inscription.sequence); + println!( + "position: {} (magic {}), epoch {}, {}", + position.network.name(), + position.network.magic(), + position.epoch, + position.point, + ); + println!( + "compression: {} level {}", + inscription.compression.algo, inscription.compression.level + ); + + match (inscription.history.first(), inscription.history.last()) { + (Some(first), Some(last)) if first.sequence != last.sequence => println!( + "history: {} entries, sequence {} ({}) through sequence {} ({})", + inscription.history.len(), + first.sequence, + first.inscription_digest, + last.sequence, + last.inscription_digest, + ), + (Some(only), _) => println!( + "history: 1 entry, sequence {} ({})", + only.sequence, only.inscription_digest, + ), + _ => println!("history: none; this stele starts a chain"), + } + + println!("layers: {}", inscription.layers.len()); + + let mut records = 0u64; + + for (descriptor, compressed) in inscription.layers.iter().zip(&inspected.compressed) { + records += descriptor.records; + + // The compressed size is the manifest's claim; a manifest claiming + // something impossible prints as unknown rather than as a number. + let compressed = match compressed { + Some(bytes) => bytes.to_string(), + None => "?".to_owned(), + }; + + println!( + " {:<8} {:>10} records {:>14} bytes {:>14} compressed {}", + descriptor.kind, + descriptor.records, + descriptor.uncompressed_size, + compressed, + descriptor.scope, + ); + } + + println!( + "totals: {} records, {} uncompressed bytes, {} compressed bytes", + records, + inscription.uncompressed_size(), + inspected.total_compressed, + ); + + Ok(()) +} diff --git a/src/bin/dolos/snapshot/mod.rs b/src/bin/dolos/snapshot/mod.rs index 2f859bb9..a9c7c178 100644 --- a/src/bin/dolos/snapshot/mod.rs +++ b/src/bin/dolos/snapshot/mod.rs @@ -4,21 +4,47 @@ //! happens here and nowhere else — see `solution/stelae` and //! `adrs/004_stelae_snapshots.md`. //! -//! Only `publish` exists so far. `digest`, `verify`, `inspect` and `sign` are -//! the publisher-productization slice, and restore is its own. +//! `publish` writes one; `digest` says what one *would* be; `verify` checks a +//! published one, digests only; `inspect` reads one's table of contents +//! 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. +//! +//! ## One epoch selection, however many commands take one +//! +//! [`EpochRange`] lives here rather than in either command, because a publisher +//! that names "epochs 500 through 519" to one and gets a different window from +//! the other is a publisher verifying a different stele than the one they +//! published — and being told it matches. One parser, one restriction, applied +//! by [`restrict`]. use clap::{Parser, Subcommand}; use dolos_core::config::RootConfig; +use dolos_snapshot::export::Plan; +use miette::IntoDiagnostic as _; +mod digest; +mod inspect; mod publish; +mod verify; #[derive(Debug, Subcommand)] pub enum Command { /// writes a stele to a local directory or an OCI repository Publish(publish::Args), + + /// computes a stele's inscription and identity without writing one + Digest(digest::Args), + + /// checks a published stele's digests — manifest against inscription, + /// every blob against both of its digests, the history chain's shape — + /// and, with --reproduce, rebuilds every layer from this node's stores + Verify(verify::Args), + + /// prints what a published stele contains, without pulling a layer + Inspect(inspect::Args), } #[derive(Debug, Parser)] @@ -30,5 +56,152 @@ pub struct Args { pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { match &args.command { Command::Publish(x) => publish::run(config, x), + Command::Digest(x) => digest::run(config, x), + Command::Verify(x) => verify::run(config, x), + Command::Inspect(x) => inspect::run(config, x), + } +} + +/// An epoch selection, in Rust's own range spellings. +/// +/// Spelled the way a reader already knows how to read: `..` excludes its end, +/// `..=` includes it. Both are accepted because a publisher naming "epochs 500 +/// through 519" and one naming "up to and including 519" are both natural, and +/// silently picking one of the two meanings is how an operator publishes an +/// epoch short. +#[derive(Debug, Clone, Copy)] +pub struct EpochRange { + first: Option, + last: Option, +} + +impl std::str::FromStr for EpochRange { + type Err = String; + + fn from_str(raw: &str) -> Result { + let bad = |why: &str| format!("{raw:?} is not an epoch range: {why}"); + + let parse = |part: &str| -> Result, String> { + match part.trim() { + "" => Ok(None), + value => value + .parse::() + .map(Some) + .map_err(|e| bad(&format!("{value:?}: {e}"))), + } + }; + + // `..=` first: `..` is a prefix of it, so testing in the other order + // would read `500..=520` as a range ending at `=520`. + let (raw, inclusive) = match raw.split_once("..=") { + Some(_) => (raw.replacen("..=", "..", 1), true), + None => (raw.to_owned(), false), + }; + + let Some((start, end)) = raw.split_once("..") else { + // A bare number: exactly that epoch. + let only = parse(&raw)?.ok_or_else(|| bad("it is empty"))?; + + return Ok(Self { + first: Some(only), + last: Some(only), + }); + }; + + let first = parse(start)?; + let end = parse(end)?; + + let last = match (end, inclusive) { + (Some(end), false) => Some( + end.checked_sub(1) + .ok_or_else(|| bad("an exclusive end of 0 selects nothing"))?, + ), + (None, true) => return Err(bad("`..=` needs an end")), + (end, _) => end, + }; + + if let (Some(first), Some(last)) = (first, last) { + if first > last { + return Err(bad("it starts after it ends")); + } + } + + Ok(Self { first, last }) + } +} + +/// Apply an operator's selection to a plan, or leave it whole. +/// +/// The one place either command narrows a plan. `restrict_epochs` takes two +/// options and a caller could always inline the call; the point is that neither +/// command gets to spell the mapping from a range to those two options its own +/// way. +pub fn restrict(plan: Plan, range: Option) -> Plan { + match range { + Some(range) => plan.restrict_epochs(range.first, range.last), + None => plan, + } +} + +/// The report every command opens with: where the node stands and what the +/// selection covers. +/// +/// Shared so that a publisher comparing a `digest` run against the `publish` +/// that produced a stele is comparing the same four lines. Written to `stderr`, +/// because `digest` puts a document on `stdout` and a report interleaved with +/// it would not be one. +pub fn report_plan(plan: &Plan) -> miette::Result<()> { + let tag = plan.tag().into_diagnostic()?; + + eprintln!( + "network: {} ({})", + plan.network.name(), + plan.network.magic() + ); + eprintln!("cursor: {}", plan.cursor); + eprintln!("sequence: {} (tag {tag})", plan.sequence); + + match (plan.epochs.first(), plan.epochs.last()) { + (Some(first), Some(last)) => eprintln!( + "epochs: {}..={} ({} of them, slots {}..={})", + first.epoch, + last.epoch, + plan.epochs.len(), + first.start_slot, + last.end_slot, + ), + // The state tip alone is a legitimate publish; say so rather than + // printing an empty range and looking like a mistake. + _ => eprintln!("epochs: none selected; the state tip only"), + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn parse(raw: &str) -> (Option, Option) { + let range: EpochRange = raw.parse().unwrap(); + (range.first, range.last) + } + + #[test] + fn epoch_ranges_read_the_way_rust_ranges_do() { + assert_eq!(parse("500..520"), (Some(500), Some(519))); + assert_eq!(parse("500..=520"), (Some(500), Some(520))); + assert_eq!(parse("500.."), (Some(500), None)); + assert_eq!(parse("..520"), (None, Some(519))); + assert_eq!(parse("..=520"), (None, Some(520))); + assert_eq!(parse(".."), (None, None)); + assert_eq!(parse("500"), (Some(500), Some(500))); + } + + #[test] + fn a_nonsensical_range_is_refused() { + for raw in ["520..500", "..0", "abc", "", "500..abc", "500..=", "-1"] { + assert!(raw.parse::().is_err(), "{raw:?}"); + } } } diff --git a/src/bin/dolos/snapshot/publish.rs b/src/bin/dolos/snapshot/publish.rs index b0d3af83..e77c1caf 100644 --- a/src/bin/dolos/snapshot/publish.rs +++ b/src/bin/dolos/snapshot/publish.rs @@ -5,7 +5,12 @@ use dolos_core::config::RootConfig; use dolos_snapshot::export; use miette::{Context as _, IntoDiagnostic as _}; -use dolos_snapshot::registry::{self, Repository}; +use dolos_snapshot::{ + export::Standing, + registry::{self, Registry, Repository}, +}; + +use super::EpochRange; /// Where a stele goes, and it goes to exactly one place. /// @@ -48,74 +53,14 @@ pub struct Args { /// report what would be written and exit #[arg(long, action)] dry_run: bool, -} - -/// An epoch selection, in Rust's own range spellings. -/// -/// Spelled the way a reader already knows how to read: `..` excludes its end, -/// `..=` includes it. Both are accepted because a publisher naming "epochs 500 -/// through 519" and one naming "up to and including 519" are both natural, and -/// silently picking one of the two meanings is how an operator publishes an -/// epoch short. -#[derive(Debug, Clone, Copy)] -struct EpochRange { - first: Option, - last: Option, -} - -impl std::str::FromStr for EpochRange { - type Err = String; - fn from_str(raw: &str) -> Result { - let bad = |why: &str| format!("{raw:?} is not an epoch range: {why}"); - - let parse = |part: &str| -> Result, String> { - match part.trim() { - "" => Ok(None), - value => value - .parse::() - .map(Some) - .map_err(|e| bad(&format!("{value:?}: {e}"))), - } - }; - - // `..=` first: `..` is a prefix of it, so testing in the other order - // would read `500..=520` as a range ending at `=520`. - let (raw, inclusive) = match raw.split_once("..=") { - Some(_) => (raw.replacen("..=", "..", 1), true), - None => (raw.to_owned(), false), - }; - - let Some((start, end)) = raw.split_once("..") else { - // A bare number: exactly that epoch. - let only = parse(&raw)?.ok_or_else(|| bad("it is empty"))?; - - return Ok(Self { - first: Some(only), - last: Some(only), - }); - }; - - let first = parse(start)?; - let end = parse(end)?; - - let last = match (end, inclusive) { - (Some(end), false) => Some( - end.checked_sub(1) - .ok_or_else(|| bad("an exclusive end of 0 selects nothing"))?, - ), - (None, true) => return Err(bad("`..=` needs an end")), - (end, _) => end, - }; - - if let (Some(first), Some(last)) = (first, last) { - if first > last { - return Err(bad("it starts after it ends")); - } - } - - Ok(Self { first, last }) - } + /// fail when the repository is already at this node's sequence, instead of + /// reporting that there is nothing to publish and exiting zero; for an + /// operator who ran this expecting a new stele, and against a job on a + /// timer, for which "nothing has closed since last time" is the ordinary + /// case + #[arg(long, action, conflicts_with = "output_dir")] + require_new: bool, } pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { @@ -129,34 +74,9 @@ pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { .into_diagnostic() .context("planning the publish")?; - let plan = match args.epochs { - Some(range) => plan.restrict_epochs(range.first, range.last), - None => plan, - }; + let plan = super::restrict(plan, args.epochs); - let tag = plan.tag().into_diagnostic()?; - - println!( - "network: {} ({})", - plan.network.name(), - plan.network.magic() - ); - println!("cursor: {}", plan.cursor); - println!("sequence: {} (tag {tag})", plan.sequence); - - match (plan.epochs.first(), plan.epochs.last()) { - (Some(first), Some(last)) => println!( - "epochs: {}..={} ({} of them, slots {}..={})", - first.epoch, - last.epoch, - plan.epochs.len(), - first.start_slot, - last.end_slot, - ), - // The state tip alone is a legitimate publish; say so rather than - // printing an empty range and looking like a mistake. - _ => println!("epochs: none selected; the state tip only"), - } + super::report_plan(&plan)?; match (&args.repo, &args.output_dir) { (Some(repo), _) => to_repository(config, args, repo, &plan, &stores), @@ -224,6 +144,12 @@ fn to_repository( .into_diagnostic() .context("opening the repository")?; + // Before anything is built, and before the dry run too: a publisher asking + // what a publish would do wants the same answer the publish gives. + if !standing(®istry, plan, args)? { + return Ok(()); + } + if args.dry_run { // `None` here and `None` at the `publish` below are one decision: a dry // run describes the publish that follows it, so the two calls are @@ -284,30 +210,55 @@ fn to_repository( Ok(()) } -#[cfg(test)] -mod tests { - use super::*; - - fn parse(raw: &str) -> (Option, Option) { - let range: EpochRange = raw.parse().unwrap(); - (range.first, range.last) - } +/// Read where this node stands against the repository, and report it. +/// +/// Returns whether the publish should go on. Three of the four readings are +/// terminal here, and it is the *middle* one that this exists for: +/// +/// - **nothing published, or exactly one sequence behind** — carry on. +/// - **the repository has already reached this node** — there is nothing to +/// publish. A job on a timer that runs more often than epochs close arrives +/// here every time it runs, and that is not a failure, so it is reported and +/// the process exits zero. `--require-new` makes the same case an error, for +/// an operator who ran it expecting a stele. +/// - **the node is further ahead than one sequence** — refused, and the refusal +/// stands: whether a deliberate gap ever gets a policy is not this command's +/// to invent. What is new is that the message names the distance alongside +/// both sequences, so "the publisher has been down for a day" and "the +/// publisher has been down for a month" do not read the same. +fn standing(registry: &Registry, plan: &export::Plan, args: &Args) -> miette::Result { + let standing = registry::standing(registry, plan) + .into_diagnostic() + .context("reading the repository's latest stele")?; - #[test] - fn epoch_ranges_read_the_way_rust_ranges_do() { - assert_eq!(parse("500..520"), (Some(500), Some(519))); - assert_eq!(parse("500..=520"), (Some(500), Some(520))); - assert_eq!(parse("500.."), (Some(500), None)); - assert_eq!(parse("..520"), (None, Some(519))); - assert_eq!(parse("..=520"), (None, Some(520))); - assert_eq!(parse(".."), (None, None)); - assert_eq!(parse("500"), (Some(500), Some(500))); - } + match standing { + Standing::Empty => { + println!("follows: nothing; this repository holds no stele"); + Ok(true) + } + Standing::Next { latest } => { + println!("follows: sequence {latest}"); + Ok(true) + } + Standing::UpToDate { latest } => { + let message = format!( + "nothing to publish: this repository is at sequence {latest} and this node is at \ + sequence {}", + plan.sequence, + ); + + if args.require_new { + return Err(miette::miette!("{message}")); + } - #[test] - fn a_nonsensical_range_is_refused() { - for raw in ["520..500", "..0", "abc", "", "500..abc", "500..=", "-1"] { - assert!(raw.parse::().is_err(), "{raw:?}"); + println!("{message}"); + Ok(false) } + Standing::Ahead { latest, distance } => Err(miette::miette!( + "this repository's latest stele is sequence {latest} and this node is at sequence \ + {}, {distance} sequences ahead: a publish must follow the repository's latest \ + stele, and this one would leave a gap no later stele could close", + plan.sequence, + )), } } diff --git a/src/bin/dolos/snapshot/verify.rs b/src/bin/dolos/snapshot/verify.rs new file mode 100644 index 00000000..e7a9a1be --- /dev/null +++ b/src/bin/dolos/snapshot/verify.rs @@ -0,0 +1,148 @@ +//! Check a published stele against what it claims — and, opted into, against +//! this node's own stores. +//! +//! Two checks, separable because a verifier with no stores can still run the +//! first: +//! +//! - **Transport, the default.** Pull the manifest and the inscription and +//! check they agree, then stream every blob: its bytes must hash to the blob +//! digest the manifest addresses it by, decompress within the size the +//! descriptor claims, and hash to the layer's `diffId` with the record count +//! the inscription states. The history chain's shape — contiguous back to its +//! first entry — is checked on the way in; a gapped chain never parses. +//! - **`--reproduce`.** Rebuild every layer from the local stores through the +//! same discarding pipeline `digest` uses and require the result to be +//! byte-identical to the published document — descriptors compared kind by +//! kind and scope by scope, then the whole inscription digest, which is what +//! a Phase 5 signature will be over. This is the check that closes the +//! incremental publish's trust gap: a layer inherited rather than rebuilt was +//! attested without being reproduced, until something runs this. +//! +//! ## What a pass does not prove +//! +//! Digests, not signatures. The published `history` is an input — a verifier +//! cannot know a chain it did not publish — so a pass proves the layers and +//! the document around them, never the chain's provenance. Saying otherwise +//! would be worse than the gap; signatures are Phase 5 and `sign` does not +//! exist yet. +//! +//! ## The exit code is the deliverable +//! +//! Non-zero on any mismatch, with the offending layer named. A determinism job +//! is this command and nothing else. + +use clap::Parser; +use dolos_core::config::RootConfig; +use dolos_snapshot::{ + export, + registry::{self, Point, Repository}, +}; +use miette::{Context as _, IntoDiagnostic as _}; + +use super::EpochRange; + +#[derive(Debug, Parser)] +pub struct Args { + /// OCI repository holding the stele, e.g. + /// `oci://ghcr.io/txpipe/dolos-mainnet` + #[arg(long, value_name = "OCI_URL")] + repo: Repository, + + /// which stele to verify: `latest`, or `epoch-N` for the stele published + /// at the end of epoch N + #[arg(long, value_name = "POINT", default_value = "latest")] + point: Point, + + /// talk to the repository over plaintext HTTP rather than HTTPS; for a + /// registry on a loopback address or a mirror inside a cluster, and for + /// nothing reachable from outside one + #[arg(long, action)] + insecure: bool, + + /// also rebuild every layer from this node's stores and require the result + /// to be byte-identical to the published inscription. Costs what a publish + /// costs — hours on mainnet — which is why it is opt-in and does not + /// belong in a pre-flight + #[arg(long, action)] + reproduce: bool, + + /// epochs the reproduction builds layers for, e.g. `500..520`, `500..=520` + /// or `500`; defaults to every epoch below the cursor, and must match what + /// the stele was published with + #[arg(long, value_name = "RANGE", requires = "reproduce")] + epochs: Option, +} + +pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { + let auth = crate::common::stele_registry_auth(&config.stelae)?; + + let registry = registry::open(&args.repo, args.insecure, auth) + .into_diagnostic() + .context("opening the repository")?; + + let verified = registry::verify(®istry, args.point) + .into_diagnostic() + .context("verifying the stele")?; + + println!("verified {} at {}", args.repo, args.point); + println!("sequence: {}", verified.inscription.sequence); + + match verified.inscription.history.as_slice() { + [] => println!("history: none; this stele starts a chain"), + [first, ..] => println!( + "history: {} entries, contiguous back to sequence {}", + verified.inscription.history.len(), + first.sequence, + ), + } + + println!( + "layers: {} streamed and checked ({} compressed bytes)", + verified.inscription.layers.len(), + verified.compressed_bytes, + ); + println!("identity: {}", verified.identity); + + // Where a reader meets the limit of what just passed: digests were + // checked, and the history was an input rather than something this command + // could vouch for. + println!("checked: digests only; signatures and chain provenance are phase 5"); + + if !args.reproduce { + return Ok(()); + } + + let stores = crate::common::open_data_stores(config) + .into_diagnostic() + .context("opening the data stores")?; + + let genesis = crate::common::open_genesis_files(&config.genesis)?; + + let plan = export::plan(&stores.state, u64::from(genesis.network_magic())) + .into_diagnostic() + .context("planning the reproduction")?; + + let plan = super::restrict(plan, args.epochs); + + super::report_plan(&plan)?; + + let reproduced = export::verify_reproduction( + &verified.inscription, + &plan, + &stores.archive, + &stores.state, + &stores.indexes, + None, + ) + .into_diagnostic() + .context("reproducing the published stele from the local stores")?; + + println!( + "reproduced: {} layers rebuilt from the local stores; the documents are byte-identical \ + ({})", + reproduced.layers.len(), + verified.identity, + ); + + Ok(()) +} diff --git a/tests/snapshot_publish.rs b/tests/snapshot_publish.rs index a45cd05a..e20ba56b 100644 --- a/tests/snapshot_publish.rs +++ b/tests/snapshot_publish.rs @@ -19,11 +19,18 @@ fn publish_writes_a_directory_that_opens_and_verifies() { node.sync(); let out = node.root.path().join("stele"); - let stdout = assert_ok(&node.publish(&out, &[])); + let output = node.publish(&out, &[]); + let stdout = assert_ok(&output); + + // The plan report — network, cursor, sequence, epochs — goes to stderr on + // every snapshot command: `digest` puts a document on stdout, the commands + // share one report, and a report split across streams by command would be + // two reports. The result lines stay on stdout. + let stderr = String::from_utf8_lossy(&output.stderr); // The name is derived from the magic, never from the configuration file, // which is what keeps two publishers on one chain from disagreeing. - assert!(stdout.contains("preview (2)"), "{stdout}"); + assert!(stderr.contains("preview (2)"), "{stderr}"); assert!(stdout.contains("identity: sha256:"), "{stdout}"); let stele = SteleDir::open(&out).unwrap();