diff --git a/Cargo.lock b/Cargo.lock index 8297cfc..01c2e7d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2193,6 +2193,7 @@ dependencies = [ "data-encoding", "futures-util", "hubble-commit", + "hubble-sync", "jacquard-api", "jacquard-common", "serde", @@ -2257,7 +2258,6 @@ dependencies = [ "jacquard-api", "jacquard-common", "jacquard-identity", - "jacquard-repo", "k256", "metrics", "multibase", @@ -2731,7 +2731,6 @@ dependencies = [ "ciborium", "ciborium-io", "cid", - "ed25519-dalek", "fluent-uri 0.4.1", "futures", "getrandom 0.2.17", @@ -2869,34 +2868,6 @@ dependencies = [ "trait-variant", ] -[[package]] -name = "jacquard-repo" -version = "0.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "98986367bb78dadaa0f2f07196bab357786c0e3670d8311b350585b91f84d6eb" -dependencies = [ - "bytes", - "cid", - "ed25519-dalek", - "iroh-car", - "jacquard-api", - "jacquard-common", - "jacquard-derive", - "k256", - "miette", - "multihash", - "n0-future", - "p256", - "serde", - "serde_bytes", - "serde_ipld_dagcbor", - "sha2 0.10.9", - "smol_str", - "thiserror 2.0.20", - "tokio", - "trait-variant", -] - [[package]] name = "jobserver" version = "0.1.34" diff --git a/Cargo.toml b/Cargo.toml index 4b7dd5a..b998f87 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,6 @@ jacquard-common = "0.12.1" jacquard-derive = "0.12.1" jacquard-identity = "0.12.1" jacquard-lexicon = "0.12.1" -jacquard-repo = "0.12.1" k256 = { version = "0.13.4", features = ["ecdsa"] } metrics = "0.24" miette = "7.6" # for lexicons codegenned by jacquard-lexicon diff --git a/hubble-commit/corpus/Cargo.toml b/hubble-commit/corpus/Cargo.toml index 575485c..31340d5 100644 --- a/hubble-commit/corpus/Cargo.toml +++ b/hubble-commit/corpus/Cargo.toml @@ -12,6 +12,7 @@ dasl = { workspace = true } data-encoding = "2.11.0" futures-util = { workspace = true } hubble-commit = { workspace = true } +hubble-sync = { workspace = true } jacquard-api = { workspace = true, features = ["streaming"] } jacquard-common = { workspace = true, features = ["streaming"] } serde = { workspace = true } diff --git a/hubble-commit/corpus/src/bench.rs b/hubble-commit/corpus/src/bench.rs index 4dea33b..8334d48 100644 --- a/hubble-commit/corpus/src/bench.rs +++ b/hubble-commit/corpus/src/bench.rs @@ -73,7 +73,10 @@ pub fn run(path: &Path, rounds: usize) -> Result<(), String> { .collect(); println!("{} commits from {}", cases.len(), path.display()); - histogram("ops/commit", &cases.iter().map(|c| c.ops.len()).collect::>()); + histogram( + "ops/commit", + &cases.iter().map(|c| c.ops.len()).collect::>(), + ); histogram( "mst blocks", &cases.iter().map(|c| c.blocks.len()).collect::>(), @@ -96,6 +99,44 @@ pub fn run(path: &Path, rounds: usize) -> Result<(), String> { } } + // how does it scale? bucket by operation count and by tree size, so the + // slope is visible rather than inferred from one aggregate number + let mut by_ops: std::collections::BTreeMap<&str, Vec> = Default::default(); + let mut by_blocks: std::collections::BTreeMap<&str, Vec> = Default::default(); + for _ in 0..rounds { + for p in &prepared { + let start = Instant::now(); + let _ = invert_ops(&p.blocks, p.case.root, &p.case.ops, OpOrder::AsGiven); + let took = start.elapsed(); + let ops = p.case.ops.len(); + by_ops + .entry(match ops { + 1 => "ops 1", + 2..=3 => "ops 2-3", + 4..=7 => "ops 4-7", + _ => "ops 8+", + }) + .or_default() + .push(took); + let blocks = p.case.blocks.len(); + by_blocks + .entry(match blocks { + 0..=5 => "blocks 1-5", + 6..=9 => "blocks 6-9", + 10..=14 => "blocks 10-14", + _ => "blocks 15+", + }) + .or_default() + .push(took); + } + } + println!("\n scaling:"); + for (label, samples) in by_ops.into_iter().chain(by_blocks) { + let n = samples.len() / rounds.max(1); + describe(&format!("{label} (n={n})"), samples); + } + + println!(); for order in [OpOrder::AsGiven, OpOrder::HighestKeyFirst] { let mut samples = Vec::with_capacity(prepared.len() * rounds); for _ in 0..rounds { diff --git a/hubble-commit/corpus/src/capture.rs b/hubble-commit/corpus/src/capture.rs index 061099a..01710c8 100644 --- a/hubble-commit/corpus/src/capture.rs +++ b/hubble-commit/corpus/src/capture.rs @@ -199,7 +199,13 @@ fn case_of(commit: &Commit, counts: &mut Counts) -> Option { } } -pub async fn run(host: &str, want: usize, out: &Path) -> Result { +/// Subscribe to a relay's firehose from the live head. +pub async fn connect( + host: &str, +) -> Result< + impl futures_util::Stream>, + String, +> { let uri = FluentUri::parse(format!("wss://{host}")) .map_err(|e| format!("bad host {host:?}: {e:?}"))?; let stream = TungsteniteClient::new() @@ -207,10 +213,15 @@ pub async fn run(host: &str, want: usize, out: &Path) -> Result .subscribe(&SubscribeRepos { cursor: None }) .await .map_err(|e| format!("subscribe: {e:?}"))?; + let (_sink, messages) = stream.into_stream(); + Ok(messages) +} +pub async fn run(host: &str, want: usize, out: &Path) -> Result { let mut writer = Writer::create(out).map_err(|e| format!("opening {}: {e}", out.display()))?; let mut counts = Counts::default(); - let (_sink, mut messages) = stream.into_stream(); + let messages = connect(host).await?; + futures_util::pin_mut!(messages); while counts.kept < want { let next = tokio::select! { diff --git a/hubble-commit/corpus/src/check.rs b/hubble-commit/corpus/src/check.rs new file mode 100644 index 0000000..cc9a692 --- /dev/null +++ b/hubble-commit/corpus/src/check.rs @@ -0,0 +1,155 @@ +//! Run live firehose events through hubble-sync's prevalidation. +//! +//! The unit tests and the captured corpus exercise inversion directly; this +//! exercises the whole of `FirehoseCommit::prevalidate` — CAR parsing, block +//! CID verification, the commit object, the bonus checks and the inversion — +//! against whatever the network is actually sending right now. It is how we +//! tell that wiring hubble-commit in did not change any verdicts. + +use std::collections::BTreeMap; +use std::time::{Duration, Instant}; + +use futures_util::StreamExt; +use hubble_commit::{CarReader, Cid, verify_block}; +use hubble_sync::{FirehoseCommit, FirehoseSync, FirehoseValidationError}; +use jacquard_api::com_atproto::sync::subscribe_repos::SubscribeReposMessage; +use tracing::{info, warn}; + +use crate::capture; + +/// Which error a rejection was, without the payload-specific detail, so the +/// counts stay legible. +fn reason(e: &FirehoseValidationError) -> &'static str { + use FirehoseValidationError as E; + match e { + E::CarTooBig { .. } => "car_too_big", + E::RecordTooBig { .. } => "record_too_big", + E::MissingOpField { .. } => "missing_op_field", + E::BadOpPath { .. } => "bad_op_path", + E::MissingRecordBlock { .. } => "missing_record_block", + E::PrevDataMismatch { .. } => "prev_data_mismatch", + E::DidMismatch { .. } => "did_mismatch", + E::TidMismatch { .. } => "tid_mismatch", + E::RootNotCommit { .. } => "root_not_commit", + E::Car(_) => "car", + E::Block(_) => "block", + E::Invert(_) => "invert", + E::CommitConvert(_) => "commit_convert", + E::MissingCommitBlock(_) => "missing_commit_block", + E::BadCommitBlock(_) => "bad_commit_block", + E::MissingPrevData => "missing_prev_data", + E::BadCid(_) => "bad_cid", + E::BadTid(_) => "bad_tid", + } +} + +fn percentile(sorted: &[Duration], q: f64) -> f64 { + let idx = ((sorted.len() as f64 * q) as usize).min(sorted.len().saturating_sub(1)); + sorted[idx].as_secs_f64() * 1e6 +} + +pub async fn run(host: &str, want: usize) -> Result<(), String> { + let messages = capture::connect(host).await?; + futures_util::pin_mut!(messages); + + let mut accepted = 0usize; + let mut rejected: BTreeMap<&'static str, usize> = BTreeMap::new(); + let mut samples: Vec = Vec::with_capacity(want); + // the same CAR work prevalidation does, timed on its own, so the split + // between reading the commit and inverting it is visible + let mut car_samples: Vec = Vec::with_capacity(want); + let mut car_bytes: Vec = Vec::with_capacity(want); + let mut syncs = 0usize; + + while samples.len() < want { + let next = tokio::select! { + biased; + _ = tokio::signal::ctrl_c() => break, + next = messages.next() => next, + }; + let Some(message) = next else { + warn!("stream ended early"); + break; + }; + let message = match message { + Ok(m) => m, + Err(e) => { + warn!(error = %e, "stream error, skipping"); + continue; + } + }; + match message { + SubscribeReposMessage::Commit(commit) => { + car_bytes.push(commit.blocks.len()); + let start = Instant::now(); + if let Ok(reader) = CarReader::new(&commit.blocks) { + let mut blocks: std::collections::HashMap = + std::collections::HashMap::new(); + for block in reader { + let Ok((cid, bytes)) = block else { break }; + if verify_block(&cid, bytes).is_err() { + break; + } + blocks.insert(cid, bytes); + } + std::hint::black_box(&blocks); + } + car_samples.push(start.elapsed()); + + let start = Instant::now(); + let verdict = FirehoseCommit::prevalidate(*commit).await; + samples.push(start.elapsed()); + match verdict { + Ok(_) => accepted += 1, + Err(e) => { + *rejected.entry(reason(&e)).or_default() += 1; + warn!(error = %e, "rejected"); + } + } + } + SubscribeReposMessage::Sync(sync) => { + syncs += 1; + if let Err(e) = FirehoseSync::prevalidate(*sync).await { + *rejected.entry(reason(&e)).or_default() += 1; + warn!(error = %e, "rejected #sync"); + } + } + _ => {} + } + } + + samples.sort_unstable(); + info!( + commits = samples.len(), + accepted, + syncs, + ?rejected, + "prevalidation" + ); + if samples.is_empty() { + return Err("no commits seen".into()); + } + car_samples.sort_unstable(); + car_bytes.sort_unstable(); + println!( + "car size: p50 {} bytes p90 {} max {}", + car_bytes[car_bytes.len() / 2], + car_bytes[car_bytes.len() * 9 / 10], + car_bytes[car_bytes.len() - 1], + ); + println!( + "car+verify: p50 {:.1}µs p90 {:.1}µs p99 {:.1}µs max {:.1}µs", + percentile(&car_samples, 0.50), + percentile(&car_samples, 0.90), + percentile(&car_samples, 0.99), + percentile(&car_samples, 1.0), + ); + println!( + "prevalidate: p50 {:.1}µs p90 {:.1}µs p99 {:.1}µs max {:.1}µs", + percentile(&samples, 0.50), + percentile(&samples, 0.90), + percentile(&samples, 0.99), + percentile(&samples, 1.0), + ); + Ok(()) +} diff --git a/hubble-commit/corpus/src/limits.rs b/hubble-commit/corpus/src/limits.rs new file mode 100644 index 0000000..2f6febd --- /dev/null +++ b/hubble-commit/corpus/src/limits.rs @@ -0,0 +1,63 @@ +//! What the spec's size limits cost us. +//! +//! A commit CAR may be 2 MB and a record block 1 MB. Real traffic is nowhere +//! near that, but a hostile PDS picks its own inputs, so the question is what +//! one commit can make the intake task spend before it yields. + +use std::time::Instant; + +use hubble_commit::cid::Codec; +use hubble_commit::{Cid, verify_block}; + +const CAR_LIMIT: usize = 2 * 1024 * 1024; +const RECORD_LIMIT: usize = 1024 * 1024; + +fn time_verify(size: usize, iters: usize) -> f64 { + let block = vec![0x5au8; size]; + let cid = Cid::digest_sha2(Codec::Drisl, &block); + // warm up so the first allocation and cache fill are not in the sample + for _ in 0..3 { + std::hint::black_box(verify_block(&cid, &block).is_ok()); + } + let start = Instant::now(); + for _ in 0..iters { + std::hint::black_box(verify_block(&cid, std::hint::black_box(&block)).is_ok()); + } + start.elapsed().as_secs_f64() * 1e6 / iters as f64 +} + +pub fn run() { + println!("block cid verification"); + let mut per_mb = 0.0; + for (size, iters) in [ + (512, 20_000), + (4096, 10_000), + (65_536, 2_000), + (RECORD_LIMIT, 200), + ] { + let us = time_verify(size, iters); + let mb_s = size as f64 / (us * 1e-6) / 1e6; + if size == RECORD_LIMIT { + per_mb = us; + } + println!(" {size:>9} bytes {us:>9.2}µs {mb_s:>7.0} MB/s"); + } + + println!("\nwhat the spec's limits imply"); + println!( + " one {}-byte record block, hashed: {:>9.0}µs", + RECORD_LIMIT, per_mb + ); + println!( + " a full {}-byte CAR, hashed: {:>9.0}µs", + CAR_LIMIT, + per_mb * (CAR_LIMIT as f64 / RECORD_LIMIT as f64), + ); + // real commits run about 500 bytes per MST node + let nodes = CAR_LIMIT / 506; + println!(" a {}-byte CAR of MST nodes ({nodes} of them),", CAR_LIMIT); + println!( + " inverted at the measured ~2.9µs/node: {:>9.0}µs", + nodes as f64 * 2.9 + ); +} diff --git a/hubble-commit/corpus/src/main.rs b/hubble-commit/corpus/src/main.rs index 1badcb4..1a42d44 100644 --- a/hubble-commit/corpus/src/main.rs +++ b/hubble-commit/corpus/src/main.rs @@ -5,7 +5,9 @@ mod bench; mod capture; +mod check; mod corpus; +mod limits; use std::path::PathBuf; @@ -31,6 +33,16 @@ enum Command { #[arg(long, default_value = "bsky.network")] host: String, }, + /// Run live firehose events through hubble-sync's full prevalidation. + Check { + /// Stop after this many commits. + #[arg(long, default_value_t = 2000)] + commits: usize, + #[arg(long, default_value = "bsky.network")] + host: String, + }, + /// Price the worst case the spec's size limits allow. + Limits, /// Cut a small, diverse subset out of a captured corpus. /// /// Real commits are overwhelmingly one operation over eight or so blocks, @@ -55,22 +67,22 @@ enum Command { async fn main() -> Result<(), String> { tracing_subscriber::fmt() .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| "info".into()), + tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()), ) .init(); match Args::parse().command { - Command::Capture { - out, - commits, - host, - } => { + Command::Capture { out, commits, host } => { let counts = capture::run(&host, commits, &out).await?; counts.report(); println!("wrote {} commits to {}", counts.kept, out.display()); Ok(()) } + Command::Limits => { + limits::run(); + Ok(()) + } + Command::Check { commits, host } => check::run(&host, commits).await, Command::Sample { corpus, out, diff --git a/hubble-commit/src/block.rs b/hubble-commit/src/block.rs new file mode 100644 index 0000000..71ad486 --- /dev/null +++ b/hubble-commit/src/block.rs @@ -0,0 +1,28 @@ +//! Checking a block against the CID that claims to address it. + +use crate::cid::{Cid, Multihash}; + +#[derive(Debug, thiserror::Error)] +pub enum BlockError { + #[error("block hashes to {computed}, not the {claimed} it is filed under")] + CidMismatch { claimed: Cid, computed: Cid }, + #[error("block is addressed by {found:?}, but atproto only uses sha2-256")] + UnsupportedMultihash { found: Multihash }, +} + +/// Check that `block`'s contents really are for its cid. +pub fn verify_block(claimed: &Cid, block: &[u8]) -> Result<(), BlockError> { + let found = claimed.multihash_type(); + if !matches!(found, Multihash::Sha2256) { + return Err(BlockError::UnsupportedMultihash { found }); + } + let computed = Cid::digest_sha2(claimed.codec(), block); + if computed == *claimed { + Ok(()) + } else { + Err(BlockError::CidMismatch { + claimed: *claimed, + computed, + }) + } +} diff --git a/hubble-commit/src/cid.rs b/hubble-commit/src/cid.rs index 5cda90b..a38d9e6 100644 --- a/hubble-commit/src/cid.rs +++ b/hubble-commit/src/cid.rs @@ -26,7 +26,7 @@ const DATA_LEN: usize = PREFIX_LEN + HASH_LEN as usize; const HASH_CODE_SHA2_256: u8 = 0x12; const HASH_CODE_BLAKE3: u8 = 0x1e; -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)] +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Hash)] pub struct Cid { // - 1 byte CID version // - 1 byte Codec @@ -108,6 +108,14 @@ impl From for CidParseError { } } +// DIVERGES FROM UPSTREAM dasl 0.2.0, which derives Debug and so prints a CID +// as 36 raw bytes. this is nicer. +impl std::fmt::Debug for Cid { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{self}") + } +} + impl FromStr for Cid { type Err = CidParseError; @@ -414,7 +422,8 @@ mod tests { #[test] fn short_bodies_are_rejected_not_panics() { for len in 0..PREFIX_LEN { - let bytes = vec![CID_VERSION, Codec::Drisl as u8, HASH_CODE_SHA2_256, 0][..len].to_vec(); + let bytes = + vec![CID_VERSION, Codec::Drisl as u8, HASH_CODE_SHA2_256, 0][..len].to_vec(); assert!( matches!(Cid::from_bytes_raw(&bytes), Err(CidParseError::TooShort)), "{len}-byte body" diff --git a/hubble-commit/src/error.rs b/hubble-commit/src/error.rs index b250d78..256c251 100644 --- a/hubble-commit/src/error.rs +++ b/hubble-commit/src/error.rs @@ -31,6 +31,12 @@ pub enum NodeFault { KeyTooLong { len: usize }, #[error("node has no entries")] Empty, + #[error("node has subtree pointers but no leaves, so it has no layer")] + NoLeaves, + #[error("subtree at layer {child} hangs off a node at layer {parent}")] + SubtreeNotBelow { parent: u8, child: u8 }, + #[error("node at the leaf layer has a subtree pointer")] + SubtreeAtLeafLayer, } /// Errors from [`crate::invert_ops`] and [`crate::Tree`]. @@ -62,10 +68,7 @@ pub enum InvertError { /// A node decoded, but is not a well-formed MST node. `cid` is absent for /// a node we built mid-inversion and have not encoded yet. #[error("malformed node {cid:?}: {detail}")] - MalformedNode { - cid: Option, - detail: NodeFault, - }, + MalformedNode { cid: Option, detail: NodeFault }, /// A node did not decode as DAG-CBOR in the MST node schema. #[error("could not decode node {cid}: {detail}")] @@ -75,11 +78,6 @@ pub enum InvertError { #[error("could not encode node: {detail}")] Encode { detail: String }, - /// The tree is deeper than sha2-256 can place a key, so it is not a tree - /// any honest producer built. Bounds recursion on hostile input. - #[error("tree nests deeper than the {limit} layers sha2-256 allows")] - TooDeep { limit: u8 }, - /// An operation names a key that is not a legal MST key. #[error("invalid key: {0}")] InvalidKey(#[from] KeyError), diff --git a/hubble-commit/src/lib.rs b/hubble-commit/src/lib.rs index e2237c5..2039fae 100644 --- a/hubble-commit/src/lib.rs +++ b/hubble-commit/src/lib.rs @@ -20,6 +20,7 @@ //! //! (in the future maybe we'll add commit coalescing too) +mod block; mod blocks; pub mod car; mod error; @@ -32,6 +33,7 @@ mod tree; // blocks by CID. Re-exported by hubble-sync as `DaslCid`. pub mod cid; +pub use block::{BlockError, verify_block}; pub use blocks::Blocks; pub use car::{CarError, CarReader}; pub use cid::{Cid, CidParseError}; @@ -76,12 +78,12 @@ pub enum OpOrder { /// Invert `ops` against the tree at `root`, returning the root it had before. pub fn invert_ops( - blocks: &impl Blocks, + blocks: &(impl Blocks + Sync), root: Cid, ops: &[Op], order: OpOrder, ) -> Result { - let mut tree = Tree::load(blocks, root)?; + let mut tree = Tree::new(blocks, root); let mut ordered: Vec<&Op> = ops.iter().collect(); if order == OpOrder::HighestKeyFirst { ordered.sort_unstable_by(|a, b| b.key().cmp(a.key())); diff --git a/hubble-commit/src/tree.rs b/hubble-commit/src/tree.rs index 1770c8b..8ec1ed4 100644 --- a/hubble-commit/src/tree.rs +++ b/hubble-commit/src/tree.rs @@ -3,6 +3,10 @@ //! synchronous and single-threaded, since only 200 ops are allowed, and the //! CAR (including records) can only be up to 2MB, so it should be fast enough //! to call from an async context. +//! +//! Nodes are decoded when traversal first reaches them. A proof usually +//! contains exactly the nodes the operations touch, so in the normal case it's +//! the same thing. use std::cmp::Ordering; use std::mem; @@ -10,26 +14,29 @@ use std::mem; use crate::blocks::Blocks; use crate::cid::Cid; use crate::error::{InvertError, NodeFault}; -use crate::key::{MAX_KEY_LEN, MAX_LAYER, layer_of, shared_prefix_len}; +use crate::key::Key; +use crate::key::{MAX_KEY_LEN, layer_of, shared_prefix_len}; use crate::node::{EntryData, NodeData, cid_for}; use crate::op::Op; -use crate::key::Key; -/// A node decoded from a block with no leaf to derive its layer from — only -/// reachable for an empty tree, or a node no honest producer would emit. -/// Resolved from the parent's layer during load. -const UNKNOWN_LAYER: i16 = -1; - -/// Passed to `insert`/`remove` to mean "work the layer out from the key". -/// Distinguished from a real layer so the root can tell it is the root. -const FROM_KEY: i16 = -1; +/// Where blocks come from. +/// +/// `Sync` so a [`Tree`] held across an await point keeps the future `Send`; +/// every real block source is a plain lookup table anyway. +type Source<'a> = &'a (dyn Blocks + Sync); /// One slot in a node: either a record, or a pointer to the subtree covering /// the keys between this slot's neighbours. #[derive(Debug)] enum Entry { - Leaf { key: Box<[u8]>, value: Cid }, - Child { cid: Option, node: Option }, + Leaf { + key: Box<[u8]>, + value: Cid, + }, + Child { + cid: Option, + node: Option, + }, } impl Entry { @@ -37,99 +44,96 @@ impl Entry { matches!(self, Entry::Child { .. }) } - fn child(&self) -> Result<&Node, InvertError> { - match self { - Entry::Child { - node: Some(node), .. - } => Ok(node), - Entry::Child { node: None, cid } => Err(missing(*cid)), - Entry::Leaf { .. } => Err(InvertError::Encode { - detail: "expected a subtree pointer, found a leaf".into(), - }), - } - } - - fn child_mut(&mut self) -> Result<&mut Node, InvertError> { - match self { - Entry::Child { - node: Some(node), .. - } => Ok(node), - Entry::Child { node: None, cid } => Err(missing(*cid)), - Entry::Leaf { .. } => Err(InvertError::Encode { + /// The subtree here, decoding its block if this is the first time anyone + /// has looked, and checking it really does sit below `parent`. + /// + /// Recursion is bounded because an adversarial MST can't be arbitrarily + /// tall, because mining higher-layer keys gets exponentially more expensive + /// (height=80 is ~infeasible) + fn child_mut(&mut self, blocks: Source<'_>, parent: u8) -> Result<&mut Node, InvertError> { + let Entry::Child { cid, node } = self else { + return Err(InvertError::Encode { detail: "expected a subtree pointer, found a leaf".into(), - }), + }); + }; + if node.is_none() { + let expected = parent.checked_sub(1).ok_or(InvertError::MalformedNode { + cid: *cid, + detail: NodeFault::SubtreeAtLeafLayer, + })?; + *node = Some(Node::decode_at(blocks, own_cid(*cid)?, expected)?); } + let child = node.as_mut().ok_or_else(|| InvertError::Encode { + detail: "subtree vanished after decoding".into(), + })?; + child.check_below(parent)?; + Ok(child) } - fn into_child(self) -> Result { - match self { + fn into_child(self, blocks: Source<'_>, parent: u8) -> Result { + let child = match self { Entry::Child { node: Some(node), .. - } => Ok(node), - Entry::Child { node: None, cid } => Err(missing(cid)), - Entry::Leaf { .. } => Err(InvertError::Encode { - detail: "expected a subtree pointer, found a leaf".into(), - }), - } + } => node, + Entry::Child { cid, node: None } => { + let expected = parent.checked_sub(1).ok_or(InvertError::MalformedNode { + cid, + detail: NodeFault::SubtreeAtLeafLayer, + })?; + Node::decode_at(blocks, own_cid(cid)?, expected)? + } + Entry::Leaf { .. } => { + return Err(InvertError::Encode { + detail: "expected a subtree pointer, found a leaf".into(), + }); + } + }; + child.check_below(parent)?; + Ok(child) } } -/// A block we needed and did not have. Upgraded to -/// [`InvertError::ProofIncomplete`] with the offending key at the API edge, -/// where the key is known. -fn missing(cid: Option) -> InvertError { - match cid { - Some(missing) => InvertError::MissingBlock { missing }, - // a pointer with neither a CID nor a node is one we built ourselves - // and have not re-encoded yet, which `root()` always does first - None => InvertError::Encode { - detail: "subtree pointer with neither a CID nor a node".into(), - }, - } +/// A pointer with neither a CID nor a node is one we built ourselves and have +/// not re-encoded yet, which `root()` always does first. +fn own_cid(cid: Option) -> Result { + cid.ok_or_else(|| InvertError::Encode { + detail: "subtree pointer with neither a CID nor a node".into(), + }) } #[derive(Debug, Default)] struct Node { /// Slots in key order, alternating between subtrees and leaves. entries: Vec, - /// Layer this node sits at, counting up from the leaves. - layer: i16, + /// Layer this node sits at, counting up from the leaves. Meaningless while + /// `stub` is set, since the block that would say has not been read. + layer: u8, /// Entries changed since this node was decoded, so `cid` is stale. dirty: bool, cid: Option, - /// We know this node's CID but never had its block. Fine to carry, an - /// error to look inside. + /// We know this node's CID but have not decoded its block. Fine to carry, + /// and fine to answer `root()` with — it is only an error if an operation + /// needs to look inside and the block turns out to be absent. stub: bool, } /// The tree a commit's proof describes, ready to have its operations inverted. -#[derive(Debug)] -pub struct Tree { +pub struct Tree<'a> { root: Node, + blocks: Source<'a>, } -impl Tree { - /// Decode the tree reachable from `root`. - /// - /// Blocks that are absent become stubs rather than errors: a commit only - /// carries a covering proof, so most of a repository is legitimately - /// missing. Touching a stub is what fails, and only if an operation needs - /// it. - pub fn load(blocks: &impl Blocks, root: Cid) -> Result { - // even the root may be absent: a commit that changed nothing needs to - // send nothing, and its previous root is the one it already claims - let Some(mut node) = Node::load(blocks, root, 0)? else { - return Ok(Self { - root: Node { - layer: UNKNOWN_LAYER, - cid: Some(root), - stub: true, - ..Default::default() - }, - }); - }; - node.resolve_layers(); - Ok(Self { root: node }) +impl<'a> Tree<'a> { + /// The tree from `root`, to be decoded as operations reach into it. + pub fn new(blocks: &'a (impl Blocks + Sync), root: Cid) -> Self { + Self { + root: Node { + cid: Some(root), + stub: true, + ..Default::default() + }, + blocks, + } } /// Invert one operation. @@ -137,16 +141,12 @@ impl Tree { /// Fails with [`InvertError::OpMismatch`] if the tree does not hold what /// the operation says it should, which means the commit is not internally /// consistent. - /// - /// Inverting several operations is order-sensitive in what blocks it needs, - /// though not in what root it produces — see [`crate::invert_ops`], which - /// orders them for you. pub fn invert(&mut self, op: &Op) -> Result<(), InvertError> { let key = op.key(); let found = match op { - Op::Create { key, .. } => self.root.remove(key.as_bytes(), FROM_KEY), + Op::Create { key, .. } => self.root.remove_root(self.blocks, key.as_bytes()), Op::Update { key, prev, .. } | Op::Delete { key, prev } => { - self.root.insert(key.as_bytes(), *prev, FROM_KEY) + self.root.insert_root(self.blocks, key.as_bytes(), *prev) } } .map_err(|err| with_key(err, key))?; @@ -187,6 +187,12 @@ impl Tree { } } +impl std::fmt::Debug for Tree<'_> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Tree").field("root", &self.root).finish() + } +} + fn with_key(err: InvertError, key: &Key) -> InvertError { match err { InvertError::MissingBlock { missing } => InvertError::ProofIncomplete { @@ -198,59 +204,86 @@ fn with_key(err: InvertError, key: &Key) -> InvertError { } impl Node { - fn load(blocks: &impl Blocks, cid: Cid, depth: u8) -> Result, InvertError> { - if depth > MAX_LAYER { - return Err(InvertError::TooDeep { limit: MAX_LAYER }); - } - let Some(block) = blocks.get(&cid) else { - return Ok(None); - }; - let data: NodeData = - dasl::drisl::from_slice(block).map_err(|e| InvertError::Decode { - cid, - detail: e.to_string(), - })?; - let mut node = decode(data, cid)?; + /// Decode a subtree, taking `expected` as its layer if it has no keys of + /// its own to say. A node covering a range whose keys all sit further down + /// is leafless and legitimate, and only its parent knows where it belongs. + fn decode_at(blocks: Source<'_>, cid: Cid, expected: u8) -> Result { + let (entries, layer) = read(blocks, cid)?; + Ok(Node { + entries, + layer: layer.unwrap_or(expected), + dirty: false, + cid: Some(cid), + stub: false, + }) + } - for entry in &mut node.entries { - let Entry::Child { - cid: Some(cid), - node: slot, - } = entry - else { - continue; - }; - if let Some(child) = Node::load(blocks, *cid, depth + 1)? { - if node.layer == UNKNOWN_LAYER && child.layer != UNKNOWN_LAYER { - node.layer = child.layer + 1; - } - *slot = Some(child); + /// Decode a root. + /// + /// A root has no parent to say what layer it is, and it does not need one: + /// the root's layer *is* the highest layer any key in the tree reaches, so + /// a non-empty tree's root always holds at least one of those keys. A + /// leafless root would be claiming a layer with nothing at it, which is + /// what `trim` exists to prevent. + fn decode_root(blocks: Source<'_>, cid: Cid) -> Result { + let (entries, layer) = read(blocks, cid)?; + let layer = match layer { + Some(layer) => layer, + None if entries.is_empty() => 0, // the empty tree, and only it + None => { + return Err(InvertError::MalformedNode { + cid: Some(cid), + detail: NodeFault::NoLeaves, + }); } - } - Ok(Some(node)) + }; + Ok(Node { + entries, + layer, + dirty: false, + cid: Some(cid), + stub: false, + }) } - /// Give a layer to any node that had no leaf to derive one from. - fn resolve_layers(&mut self) { - if self.layer <= 0 { - return; + /// Decode this node's own block, if it is still just a pointer. + /// + /// Only ever called on the root: every other node is decoded through its + /// parent, which knows what layer to expect. + fn wake(&mut self, blocks: Source<'_>) -> Result<(), InvertError> { + if !self.stub { + return Ok(()); } - for entry in &mut self.entries { - if let Entry::Child { - node: Some(child), .. - } = entry - { - if child.layer == UNKNOWN_LAYER { - child.layer = self.layer - 1; - } - child.resolve_layers(); - } + *self = Node::decode_root(blocks, own_cid(self.cid)?)?; + Ok(()) + } + + fn check_below(&self, parent: u8) -> Result<(), InvertError> { + // an unread pointer has nothing to check yet; whoever wakes it will + if self.stub || self.layer < parent { + return Ok(()); } + Err(InvertError::MalformedNode { + cid: self.cid, + detail: NodeFault::SubtreeNotBelow { + parent, + child: self.layer, + }, + }) + } + + /// The layer this node's subtrees sit at or below. + fn below(&self) -> Result { + self.layer.checked_sub(1).ok_or(InvertError::MalformedNode { + cid: self.cid, + detail: NodeFault::SubtreeAtLeafLayer, + }) } fn rehash(&mut self) -> Result { if self.stub { - return Err(missing(self.cid)); + // untouched, so its stored CID is still the answer + return own_cid(self.cid); } for entry in &mut self.entries { if let Entry::Child { @@ -268,22 +301,31 @@ impl Node { Ok(cid) } + /// Put `value` at `key` in the tree this node roots. + fn insert_root( + &mut self, + blocks: Source<'_>, + key: &[u8], + value: Cid, + ) -> Result, InvertError> { + self.insert(blocks, key, value, layer_of(key)) + } + /// Put `value` at `key`, returning whatever was there before. - fn insert(&mut self, key: &[u8], value: Cid, layer: i16) -> Result, InvertError> { - if self.stub { - return Err(missing(self.cid)); - } - let layer = if layer == FROM_KEY { - i16::from(layer_of(key)) - } else { - layer - }; + fn insert( + &mut self, + blocks: Source<'_>, + key: &[u8], + value: Cid, + layer: u8, + ) -> Result, InvertError> { + self.wake(blocks)?; if layer > self.layer { - return self.insert_above(key, value, layer); + return self.insert_above(blocks, key, value, layer); } if layer < self.layer { - return self.insert_below(key, value, layer); + return self.insert_below(blocks, key, value, layer); } if let Some(idx) = self.find_leaf(key) { @@ -300,7 +342,7 @@ impl Node { return Ok(Some(prev)); } - let (idx, splits) = self.find_slot(key)?; + let (idx, splits) = self.find_slot(blocks, key)?; self.dirty = true; let leaf = Entry::Leaf { key: key.into(), @@ -312,27 +354,22 @@ impl Node { } // the key lands inside the subtree at `idx`, which has to come apart - let Entry::Child { node, .. } = &mut self.entries[idx] else { - return Err(InvertError::Encode { - detail: "find_slot marked a leaf as splitting".into(), - }); - }; - let (left, right) = node.take().ok_or_else(|| missing(None))?.split(key)?; + let parent = self.layer; + let child = self.entries.remove(idx).into_child(blocks, parent)?; + let (left, right) = child.split(blocks, key)?; self.entries - .splice(idx..=idx, [child_of(left), leaf, child_of(right)]); + .splice(idx..idx, [child_of(left), leaf, child_of(right)]); Ok(None) } /// The key belongs at a higher layer than this node: grow upwards. fn insert_above( &mut self, + blocks: Source<'_>, key: &[u8], value: Cid, - layer: i16, + layer: u8, ) -> Result, InvertError> { - if layer > i16::from(MAX_LAYER) { - return Err(InvertError::TooDeep { limit: MAX_LAYER }); - } if self.entries.is_empty() { // nothing to carry up, so this node just *is* the new layer *self = Node { @@ -349,26 +386,28 @@ impl Node { mem::swap(self, &mut below); self.entries.push(child_of(below)); } - self.insert(key, value, layer) + self.insert(blocks, key, value, layer) } /// The key belongs at a lower layer: descend, making a child if needed. fn insert_below( &mut self, + blocks: Source<'_>, key: &[u8], value: Cid, - layer: i16, + layer: u8, ) -> Result, InvertError> { + let parent = self.layer; if let Some(idx) = self.find_covering_child(key) { - let child = self.entries[idx].child_mut()?; - let prev = child.insert(key, value, layer)?; + let child = self.entries[idx].child_mut(blocks, parent)?; + let prev = child.insert(blocks, key, value, layer)?; if prev != Some(value) { self.dirty = true; } return Ok(prev); } - let (idx, splits) = self.find_slot(key)?; + let (idx, splits) = self.find_slot(blocks, key)?; if splits { return Err(InvertError::Encode { detail: "a key below this layer cannot split a sibling subtree".into(), @@ -376,35 +415,40 @@ impl Node { } self.dirty = true; let mut child = Node { - layer: self.layer - 1, + layer: self.below()?, dirty: true, ..Default::default() }; - child.insert(key, value, layer)?; + child.insert(blocks, key, value, layer)?; self.entries.insert(idx, child_of(child)); Ok(None) } - /// Take `key` out, returning what it held. + /// Take `key` out of the tree this node roots, returning what it held. /// - /// `layer == FROM_KEY` marks the call as being at the root, which is the - /// only place allowed to reshape the top of the tree. - fn remove(&mut self, key: &[u8], layer: i16) -> Result, InvertError> { - if self.stub { - return Err(missing(self.cid)); + /// Only the root may reshape the top of the tree, which is why this is + /// separate from the recursive form. + fn remove_root(&mut self, blocks: Source<'_>, key: &[u8]) -> Result, InvertError> { + let removed = self.remove(blocks, key, layer_of(key))?; + if removed.is_some() { + self.trim(blocks)?; } - let at_root = layer == FROM_KEY; - let layer = if at_root { - i16::from(layer_of(key)) - } else { - layer - }; + Ok(removed) + } + + fn remove( + &mut self, + blocks: Source<'_>, + key: &[u8], + layer: u8, + ) -> Result, InvertError> { + self.wake(blocks)?; if layer > self.layer { return Ok(None); // the key would sit above this tree, so it is absent } if layer < self.layer { - return self.remove_below(key, layer); + return self.remove_below(blocks, key, layer); } let Some(idx) = self.find_leaf(key) else { @@ -419,6 +463,7 @@ impl Node { // a leaf sitting between two subtrees was the only thing separating // them, so they become one + let parent = self.layer; if idx > 0 && idx + 1 < self.entries.len() && self.entries[idx - 1].is_child() @@ -430,29 +475,32 @@ impl Node { detail: "drained fewer entries than the range covers".into(), })?; drop(drained); + let right = right.into_child(blocks, parent)?; self.entries[idx - 1] - .child_mut()? - .merge(right.into_child()?)?; + .child_mut(blocks, parent)? + .merge(blocks, right)?; } else { self.entries.remove(idx); } - - if at_root { - self.trim(); - } Ok(Some(prev)) } - fn remove_below(&mut self, key: &[u8], layer: i16) -> Result, InvertError> { + fn remove_below( + &mut self, + blocks: Source<'_>, + key: &[u8], + layer: u8, + ) -> Result, InvertError> { let Some(idx) = self.find_covering_child(key) else { return Ok(None); }; - let child = self.entries[idx].child_mut()?; - let Some(prev) = child.remove(key, layer)? else { + let parent = self.layer; + let child = self.entries[idx].child_mut(blocks, parent)?; + let Some(prev) = child.remove(blocks, key, layer)? else { return Ok(None); }; self.dirty = true; - if child.entries.is_empty() { + if child.entries.is_empty() && !child.stub { self.entries.remove(idx); } Ok(Some(prev)) @@ -465,31 +513,70 @@ impl Node { /// one it had, or the next insert will build beneath a layer that should /// not exist. Indigo and rsky-relay both leave that second case alone, and /// both compute the wrong root for it. - fn trim(&mut self) { + /// + /// The promoted pointer has to be decoded to know whether it needs + /// trimming in turn — a removal can collapse more than one layer. If its + /// block is absent we stop there and keep the pointer: nothing above it + /// survives to be re-encoded, so its stored CID is already the answer. + fn trim(&mut self, blocks: Source<'_>) -> Result<(), InvertError> { + // a node the loop is about to promote away may legitimately be + // leafless, so track the layer separately rather than parking a + // placeholder in the node itself + let mut layer = Some(self.layer); while self.entries.len() == 1 && self.entries[0].is_child() { let Entry::Child { cid, node } = self.entries.remove(0) else { - return; - }; - let layer = self.layer - 1; - *self = match node { - Some(child) => child, - // the proof stops here, but a bare pointer is a complete - // answer: we already know this subtree's CID - None => Node { - layer, - cid, - stub: true, - ..Default::default() - }, + return Err(InvertError::Encode { + detail: "entry claimed to be a subtree pointer and was not".into(), + }); }; + if let Some(child) = node { + layer = Some(child.layer); + *self = child; + continue; + } + let cid = own_cid(cid)?; + match read(blocks, cid) { + Ok((entries, found)) => { + layer = found; + *self = Node { + entries, + layer: found.unwrap_or_default(), + cid: Some(cid), + ..Default::default() + }; + } + Err(InvertError::MissingBlock { .. }) => { + *self = Node { + cid: Some(cid), + stub: true, + ..Default::default() + }; + return Ok(()); + } + Err(e) => return Err(e), + } } - if self.entries.is_empty() { + + match layer { + Some(layer) => self.layer = layer, + // the loop only stops on a node with a leaf, an empty one, or a + // malformed one with two subtrees side by side + None if self.entries.is_empty() => self.layer = 0, + None => { + return Err(InvertError::MalformedNode { + cid: self.cid, + detail: NodeFault::NoLeaves, + }); + } + } + if self.entries.is_empty() && !self.stub { self.layer = 0; } + Ok(()) } /// Absorb `other`, whose keys all sort after this node's. - fn merge(&mut self, other: Node) -> Result<(), InvertError> { + fn merge(&mut self, blocks: Source<'_>, other: Node) -> Result<(), InvertError> { let seam = self.entries.len(); self.dirty = true; self.cid = None; @@ -504,16 +591,17 @@ impl Node { && self.entries[seam - 1].is_child() && self.entries[seam].is_child() { - let right = self.entries.remove(seam); + let parent = self.layer; + let right = self.entries.remove(seam).into_child(blocks, parent)?; self.entries[seam - 1] - .child_mut()? - .merge(right.into_child()?)?; + .child_mut(blocks, parent)? + .merge(blocks, right)?; } Ok(()) } /// Break this node into the parts either side of `key`. - fn split(mut self, key: &[u8]) -> Result<(Node, Node), InvertError> { + fn split(mut self, blocks: Source<'_>, key: &[u8]) -> Result<(Node, Node), InvertError> { if self.entries.is_empty() { return Err(InvertError::MalformedNode { cid: self.cid, @@ -521,7 +609,7 @@ impl Node { }); } let layer = self.layer; - let (idx, splits) = self.find_slot(key)?; + let (idx, splits) = self.find_slot(blocks, key)?; if !splits { if idx == 0 || idx >= self.entries.len() { @@ -535,14 +623,10 @@ impl Node { // the key lands inside a subtree, so that has to split too let mut right = self.entries.split_off(idx); - let Entry::Child { node, .. } = &mut right[0] else { - return Err(InvertError::Encode { - detail: "find_slot marked a leaf as splitting".into(), - }); - }; - let (left_node, right_node) = node.take().ok_or_else(|| missing(None))?.split(key)?; + let child = right.remove(0).into_child(blocks, layer)?; + let (left_node, right_node) = child.split(blocks, key)?; self.entries.push(child_of(left_node)); - right[0] = child_of(right_node); + right.insert(0, child_of(right_node)); Ok((node_of(self.entries, layer), node_of(right, layer))) } @@ -579,12 +663,11 @@ impl Node { /// Where a new entry for `key` goes, and whether it lands inside the /// subtree already sitting there. - fn find_slot(&self, key: &[u8]) -> Result<(usize, bool), InvertError> { - if self.stub { - return Err(missing(self.cid)); - } - for (idx, entry) in self.entries.iter().enumerate() { - match entry { + fn find_slot(&mut self, blocks: Source<'_>, key: &[u8]) -> Result<(usize, bool), InvertError> { + self.wake(blocks)?; + let parent = self.layer; + for idx in 0..self.entries.len() { + match &self.entries[idx] { Entry::Leaf { key: leaf, .. } => { if key < leaf.as_ref() { return Ok((idx, false)); @@ -598,7 +681,10 @@ impl Node { { continue; } - match entry.child()?.covers(key)? { + let order = self.entries[idx] + .child_mut(blocks, parent)? + .covers(blocks, key)?; + match order { Ordering::Less => return Ok((idx, false)), Ordering::Equal => return Ok((idx, true)), Ordering::Greater => {} @@ -610,10 +696,9 @@ impl Node { } /// Whether `key` falls below, inside, or above this subtree's key range. - fn covers(&self, key: &[u8]) -> Result { - if self.stub { - return Err(missing(self.cid)); - } + fn covers(&mut self, blocks: Source<'_>, key: &[u8]) -> Result { + self.wake(blocks)?; + let parent = self.layer; let (Some(first), Some(last)) = (self.entries.first(), self.entries.last()) else { return Err(InvertError::MalformedNode { cid: self.cid, @@ -630,8 +715,8 @@ impl Node { { return Ok(Ordering::Greater); } - for (idx, entry) in self.entries.iter().enumerate() { - match entry { + for idx in 0..self.entries.len() { + match &self.entries[idx] { Entry::Leaf { key: leaf, .. } => { if key < leaf.as_ref() { return Ok(Ordering::Equal); @@ -644,7 +729,10 @@ impl Node { continue; } let last_slot = idx == self.entries.len() - 1; - return Ok(match entry.child()?.covers(key)? { + let order = self.entries[idx] + .child_mut(blocks, parent)? + .covers(blocks, key)?; + return Ok(match order { Ordering::Less if idx == 0 => Ordering::Less, Ordering::Greater if last_slot => Ordering::Greater, _ => Ordering::Equal, @@ -663,7 +751,7 @@ fn child_of(node: Node) -> Entry { } } -fn node_of(entries: Vec, layer: i16) -> Node { +fn node_of(entries: Vec, layer: u8) -> Node { Node { entries, layer, @@ -672,14 +760,29 @@ fn node_of(entries: Vec, layer: i16) -> Node { } } -/// Rebuild a node's entries from its wire form, expanding prefix compression. -fn decode(data: NodeData, cid: Cid) -> Result { +/// Rebuild a node's entries from its wire form, undoing prefix compression. +/// +/// A node's layer is the layer of its own keys, so it is not something a +/// producer gets to assert. A node with subtrees but no keys has no layer to +/// derive and is not something a canonical tree contains. +fn read(blocks: Source<'_>, cid: Cid) -> Result<(Vec, Option), InvertError> { + let block = blocks + .get(&cid) + .ok_or(InvertError::MissingBlock { missing: cid })?; + let data: NodeData = dasl::drisl::from_slice(block).map_err(|e| InvertError::Decode { + cid, + detail: e.to_string(), + })?; + decode(data, cid) +} + +fn decode(data: NodeData, cid: Cid) -> Result<(Vec, Option), InvertError> { let fault = |detail| InvertError::MalformedNode { cid: Some(cid), detail, }; - let mut layer = UNKNOWN_LAYER; + let mut layer = None; let mut entries = Vec::with_capacity(data.entries.len() * 2 + 1); if let Some(left) = data.left { entries.push(Entry::Child { @@ -708,9 +811,7 @@ fn decode(data: NodeData, cid: Cid) -> Result { let mut key = Vec::with_capacity(len); key.extend_from_slice(&prev[..entry.prefix_len]); key.extend_from_slice(&entry.key_suffix); - if layer == UNKNOWN_LAYER { - layer = i16::from(layer_of(&key)); - } + layer.get_or_insert_with(|| layer_of(&key)); prev.clear(); prev.extend_from_slice(&key); @@ -727,13 +828,7 @@ fn decode(data: NodeData, cid: Cid) -> Result { } } - Ok(Node { - entries, - layer, - dirty: false, - cid: Some(cid), - stub: false, - }) + Ok((entries, layer)) } /// Serialise a node back to canonical DAG-CBOR. @@ -774,11 +869,19 @@ fn encode(entries: &[Entry]) -> Result, InvertError> { }) } -/// Building trees forwards, which only the tests need: inversion is handed a -/// tree and takes it apart. Kept private so the crate's surface stays honest -/// about what it is for. +/// A tree built in memory needs no block source; nothing in it is ever a stub. +#[cfg(test)] +struct NoBlocks; + +#[cfg(test)] +impl Blocks for NoBlocks { + fn get(&self, _: &Cid) -> Option<&[u8]> { + None + } +} + #[cfg(test)] -impl Tree { +impl Tree<'static> { fn empty() -> Self { Self { root: Node { @@ -786,15 +889,16 @@ impl Tree { dirty: true, ..Default::default() }, + blocks: &NoBlocks, } } fn insert(&mut self, key: &str, value: Cid) -> Result, InvertError> { - self.root.insert(key.as_bytes(), value, FROM_KEY) + self.root.insert_root(self.blocks, key.as_bytes(), value) } fn remove(&mut self, key: &str) -> Result, InvertError> { - self.root.remove(key.as_bytes(), FROM_KEY) + self.root.remove_root(self.blocks, key.as_bytes()) } /// Every node in the tree, encoded, as inversion would need to read them. @@ -996,11 +1100,78 @@ mod tests { ); } + /// Counts how many blocks a traversal actually asks for. + struct Counting<'a> { + inner: &'a HashMap>, + reads: std::sync::atomic::AtomicUsize, + } + + impl Blocks for Counting<'_> { + fn get(&self, cid: &Cid) -> Option<&[u8]> { + self.reads + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.inner.get(cid).map(Vec::as_slice) + } + } + + /// A producer may send its whole tree instead of the minimal proof, and + /// extra blocks in a diff are legal. Decoding all of them to then walk + /// eight is work we must not be made to do: one operation should read a + /// path's worth of nodes, not a repo's worth. + #[test] + fn one_op_reads_a_paths_worth_of_nodes() { + let leaf = cid("leaf"); + let keys: Vec = (0..1000) + .map(|i| format!("app.bsky.feed.post/3l{i:013}")) + .collect(); + + let mut before = Tree::empty(); + for key in &keys { + before.insert(key, leaf).expect("insert"); + } + let root_before = before.root().expect("root"); + + let added = "app.bsky.feed.post/3lzzzzzzzzzzzzz"; + let mut after = Tree::empty(); + for key in &keys { + after.insert(key, leaf).expect("insert"); + } + after.insert(added, leaf).expect("insert"); + let root_after = after.root().expect("root"); + + // the whole tree, not the covering proof + let blocks = after.blocks().expect("blocks"); + assert!(blocks.len() > 200, "test tree too small: {}", blocks.len()); + + let counting = Counting { + inner: &blocks, + reads: std::sync::atomic::AtomicUsize::new(0), + }; + let ops = vec![Op::Create { + key: Key::new(added).expect("key"), + cid: leaf, + }]; + assert_eq!( + invert_ops(&counting, root_after, &ops, OpOrder::AsGiven).expect("invert"), + root_before + ); + + let reads = counting.reads.load(std::sync::atomic::Ordering::Relaxed); + assert!( + reads < 32, + "read {reads} blocks out of {} to invert one op", + blocks.len() + ); + } + #[test] fn a_commit_that_changed_nothing_needs_no_blocks() { let root = cid("some root"); let blocks: HashMap> = HashMap::new(); - assert_eq!(invert_ops(&blocks, root, &[], OpOrder::AsGiven).expect("invert"), root); + assert_eq!( + invert_ops(&blocks, root, &[], OpOrder::AsGiven).expect("invert"), + root + ); } #[test] @@ -1085,8 +1256,11 @@ mod tests { keys.dedup(); let value = |n: u64| cid(&format!("record {n}")); - let mut state: Vec<(String, Cid)> = - keys.iter().cloned().map(|k| (k, value(rng.next()))).collect(); + let mut state: Vec<(String, Cid)> = keys + .iter() + .cloned() + .map(|k| (k, value(rng.next()))) + .collect(); let mut before = Tree::empty(); for (key, cid) in &state { @@ -1101,7 +1275,8 @@ mod tests { let choice = if state.is_empty() { 0 } else { rng.below(3) }; match choice { 0 => { - let key = format!("app.bsky.feed.post/3l{:013}", rng.next() % 1_000_000_000); + let key = + format!("app.bsky.feed.post/3l{:013}", rng.next() % 1_000_000_000); if touched.contains(&key) || state.iter().any(|(k, _)| *k == key) { continue; } diff --git a/hubble-commit/tests/firehose_corpus.rs b/hubble-commit/tests/firehose_corpus.rs index ce38f07..3541f00 100644 --- a/hubble-commit/tests/firehose_corpus.rs +++ b/hubble-commit/tests/firehose_corpus.rs @@ -55,7 +55,9 @@ impl Record { self.blocks .iter() .map(|b| { - let bytes = BASE64.decode(b.as_bytes()).expect("fixture block is base64"); + let bytes = BASE64 + .decode(b.as_bytes()) + .expect("fixture block is base64"); (Cid::digest_sha2(Codec::Drisl, &bytes), bytes) }) .collect() diff --git a/hubble-commit/tests/mst_diff_suite.rs b/hubble-commit/tests/mst_diff_suite.rs index 6489a6b..b53ba4f 100644 --- a/hubble-commit/tests/mst_diff_suite.rs +++ b/hubble-commit/tests/mst_diff_suite.rs @@ -96,7 +96,12 @@ fn run_case(case_path: &Path, suite_root: &Path) -> Result<(), String> { let (root_a, blocks_a) = read_car(&resolve(suite_root, &case.inputs.mst_a))?; let (root_b, blocks_b) = read_car(&resolve(suite_root, &case.inputs.mst_b))?; - let ops: Result, String> = case.results.record_ops.iter().map(RecordOp::to_op).collect(); + let ops: Result, String> = case + .results + .record_ops + .iter() + .map(RecordOp::to_op) + .collect(); // deliberately left in the suite's own order let ops = ops?; @@ -105,7 +110,11 @@ fn run_case(case_path: &Path, suite_root: &Path) -> Result<(), String> { let mut all = blocks_b.clone(); all.extend(blocks_a); let got = invert_ops(&all, root_b, &ops, OpOrder::AsGiven).map_err(|e| { - format!("{}: inverting {} ops failed: {e}", case.description, ops.len()) + format!( + "{}: inverting {} ops failed: {e}", + case.description, + ops.len() + ) })?; if got != root_a { return Err(format!( diff --git a/hubble-sync/Cargo.toml b/hubble-sync/Cargo.toml index b08bd1b..c615bf9 100644 --- a/hubble-sync/Cargo.toml +++ b/hubble-sync/Cargo.toml @@ -15,7 +15,6 @@ hubble-commit = { workspace = true } jacquard-api = { workspace = true, features = ["streaming"] } jacquard-common = { workspace = true, features = ["streaming"] } jacquard-identity = { workspace = true } -jacquard-repo = { workspace = true } k256 = { workspace = true } metrics = { workspace = true } multibase = { workspace = true } diff --git a/hubble-sync/src/commit/commit_object.rs b/hubble-sync/src/commit/commit_object.rs index 2840444..73b1685 100644 --- a/hubble-sync/src/commit/commit_object.rs +++ b/hubble-sync/src/commit/commit_object.rs @@ -1,7 +1,5 @@ use dasl::drisl; -use jacquard_common::BosStr; use jacquard_common::types::cid::IpldCid; -use jacquard_repo::commit::Commit as JacqCommit; use repo_stream::Commit as RSCommit; use serde::Serialize; @@ -95,29 +93,6 @@ impl TryFrom<&RSCommit> for CommitObject { } } -impl TryFrom> for CommitObject { - type Error = CommitConvertError; - fn try_from(jac: JacqCommit) -> Result { - let did = Did::raw(jac.did.as_str()); - let data = DaslCid::from_bytes_raw(&jac.data.to_bytes()) - .map_err(|_| CommitConvertError::BadCid("data"))?; - let rev = jac.rev.into(); - let prev = jac - .prev - .map(|t| DaslCid::from_bytes_raw(&t.to_bytes())) - .transpose() - .map_err(|_| CommitConvertError::BadCid("prev"))?; - let sig = jac.sig.to_vec(); - Ok(Self { - did, - data, - rev, - prev, - sig, - }) - } -} - /// the commit representation that gets signed /// /// `sig` is absent, gets dag-cbor (drisl)-encoded diff --git a/hubble-sync/src/firehose/event_validation.rs b/hubble-sync/src/firehose/event_validation.rs index 3876121..1466d3c 100644 --- a/hubble-sync/src/firehose/event_validation.rs +++ b/hubble-sync/src/firehose/event_validation.rs @@ -4,17 +4,16 @@ use std::time::Instant; use bytes::Bytes; use dasl::drisl; +use hubble_commit::{ + BlockError, CarError, CarReader, InvertError, Key, Op as InvertOp, Tree, verify_block, +}; use jacquard_api::com_atproto::sync::subscribe_repos::{ - Commit as JacquardCommit, RepoOpAction, Sync as JacquardSync, + Commit as JacquardCommit, RepoOp, RepoOpAction, Sync as JacquardSync, }; -use jacquard_common::types::cid::IpldCid; -use jacquard_repo::RepoError; -use jacquard_repo::car::{parse_car_bytes, reader::ParsedCar}; -use jacquard_repo::commit::firehose::to_invertible_op; -use jacquard_repo::mst::{Mst, tree::VerifiedWriteOp}; -use jacquard_repo::storage::MemoryBlockStore; +use jacquard_common::types::cid::{CidLink, IpldCid}; use metrics::histogram; -use repo_stream::{CidMismatch, Commit as RSCommit, verify_block_cid}; +use repo_stream::Commit as RSCommit; +use tokio::task::coop::consume_budget; use crate::commit::{CommitConvertError, CommitObject, Op, OpKind}; use crate::metrics::COMMIT_PREVALIDATE_SECONDS; @@ -29,12 +28,16 @@ pub enum FirehoseValidationError { CarTooBig { limit: usize, got: usize }, #[error("Record too big for firehose limit of {limit}: {got}")] RecordTooBig { limit: usize, got: usize }, - #[error("Missing record link for op: {action} {path:?}")] - MissingRecordLink { action: RepoOpAction, path: String }, + #[error("op missing its {field} field: {action} {path:?}")] + MissingOpField { + action: RepoOpAction, + path: String, + field: &'static str, + }, + #[error("op has an invalid record path {path:?}: {detail}")] + BadOpPath { path: String, detail: String }, #[error("Missing record block for {action} op: {cid:?}")] MissingRecordBlock { action: RepoOpAction, cid: String }, - #[error("failed to invert an op: {action} {path:?}")] - FailedToInvert { action: RepoOpAction, path: String }, #[error("prevData mismatch after operation inversion")] PrevDataMismatch { claimed_prev_data: String, @@ -44,8 +47,14 @@ pub enum FirehoseValidationError { DidMismatch { claimed: String, in_commit: String }, #[error("Tid mismatch")] TidMismatch { claimed: String, in_commit: String }, - #[error("CID mismatch")] - CidMismatch(#[from] Box), + #[error("CAR root {root} is not the event's commit link {commit}")] + RootNotCommit { root: DaslCid, commit: DaslCid }, + #[error("CAR: {0}")] + Car(#[from] CarError), + #[error("block: {0}")] + Block(#[from] BlockError), + #[error("operation inversion: {0}")] + Invert(#[from] InvertError), #[error("commit convert: {0}")] CommitConvert(#[from] CommitConvertError), #[error("missing commit block at {0}")] @@ -54,8 +63,6 @@ pub enum FirehoseValidationError { BadCommitBlock(String), #[error("missing prevData field from firehose commit")] MissingPrevData, - #[error("jacquard repo error: {0}")] - RepoError(#[from] RepoError), #[error("invalid CID at {0}")] BadCid(&'static str), #[error("invalid Tid: {0}")] @@ -79,6 +86,108 @@ pub struct FirehoseCommit { pub blocks: HashMap, } +/// Parse a firehose CAR, checking every block against the CID it arrived under. +/// +/// Blocks are returned as `Bytes` views into `car` rather than copies, so this +/// is also the map the commit view hands to consumers: built once, shared. +async fn parse_car( + car: &Bytes, +) -> Result<(DaslCid, HashMap), FirehoseValidationError> { + let reader = CarReader::new(car)?; + let root = reader.root(); + let mut blocks = HashMap::new(); + for block in reader { + // hashing a block is O(its size), and the spec allows 1MB of it, so + // this is a chunk boundary. `consume_budget` returns immediately + // unless the task has actually used its allowance, so the small + // commits that make up essentially all traffic never yield here. + consume_budget().await; + let (cid, bytes) = block?; + // max record size over firehose is 1MB + if bytes.len() > RECORD_SIZE_LIMIT { + return Err(FirehoseValidationError::RecordTooBig { + limit: RECORD_SIZE_LIMIT, + got: bytes.len(), + }); + } + verify_block(&cid, bytes)?; + blocks.insert(cid, car.slice_ref(bytes)); + } + Ok((root, blocks)) +} + +fn link_cid(link: &CidLink, source: &'static str) -> Result { + let ipld = link + .to_ipld() + .map_err(|_| FirehoseValidationError::BadCid(source))?; + to_dasl_cid(ipld, source) +} + +/// One firehose op in both the shapes we need: what inversion consumes, and +/// what the commit view hands to consumers. +fn convert_op(op: &RepoOp) -> Result<(InvertOp, Op), FirehoseValidationError> { + let missing = |field| FirehoseValidationError::MissingOpField { + action: op.action.clone(), + path: op.path.to_string(), + field, + }; + let path: &str = op.path.as_ref(); + let key = Key::new(path).map_err(|e| FirehoseValidationError::BadOpPath { + path: path.to_string(), + detail: e.to_string(), + })?; + let view_path: Arc = path.into(); + + let cid = |field| -> Result { + link_cid(op.cid.as_ref().ok_or_else(|| missing(field))?, "ops[].cid") + }; + let prev = |field| -> Result { + link_cid( + op.prev.as_ref().ok_or_else(|| missing(field))?, + "ops[].prev", + ) + }; + + Ok(match op.action { + RepoOpAction::Create => { + let cid = cid("cid")?; + ( + InvertOp::Create { key, cid }, + Op { + path: view_path, + kind: OpKind::Create { cid }, + }, + ) + } + RepoOpAction::Update => { + let (cid, prev) = (cid("cid")?, prev("prev")?); + ( + InvertOp::Update { key, cid, prev }, + Op { + path: view_path, + kind: OpKind::Update { cid, prev }, + }, + ) + } + RepoOpAction::Delete => { + let prev = prev("prev")?; + ( + InvertOp::Delete { key, prev }, + Op { + path: view_path, + kind: OpKind::Delete { prev }, + }, + ) + } + RepoOpAction::Other(ref action) => { + return Err(FirehoseValidationError::BadOpPath { + path: path.to_string(), + detail: format!("unknown action {action}"), + }); + } + }) +} + impl FirehoseSync { /// Apply firehose sync event valiation, which isn't in the spec (yet?) /// @@ -96,9 +205,11 @@ impl FirehoseSync { }); } - let rs_commit = repo_stream::DriverBuilder::new() - .load_commit(&*s.blocks) - .await + let (root, blocks) = parse_car(&s.blocks).await?; + let commit_block = blocks + .get(&root) + .ok_or_else(|| FirehoseValidationError::MissingCommitBlock(root.to_string()))?; + let rs_commit = drisl::from_slice::(commit_block) .map_err(|e| FirehoseValidationError::BadCommitBlock(e.to_string()))?; let commit: CommitObject = (&rs_commit).try_into()?; @@ -158,55 +269,37 @@ impl FirehoseCommit { // (keep an eye on the time this takes bc all commits serialize through) let invert_start = Instant::now(); - // jacquard's parse only verifies the actual raw car structure - let ParsedCar { root, blocks } = parse_car_bytes(&c.blocks).await?; - - // aditional integrity checks: assert size, CIDs against block bytes - for (cid, block) in &blocks { - // max record size over firehose is 1MB - if block.len() > RECORD_SIZE_LIMIT { - return Err(FirehoseValidationError::RecordTooBig { - limit: RECORD_SIZE_LIMIT, - got: block.len(), - }); - } - // jacquard doesn't assert CIDs against actual block bytes (and - // iroh-car underneath does not either) - verify_block_cid(cid, block)?; - } + let (root, blocks) = parse_car(&c.blocks).await?; // parse the Commit object to ensure its structure is valid. // NOTE: we're not currently validating MST node structures. they should // surface in step 3. let commit_block = blocks .get(&root) - .ok_or(FirehoseValidationError::MissingCommitBlock( - root.to_string(), - ))?; + .ok_or_else(|| FirehoseValidationError::MissingCommitBlock(root.to_string()))?; // this is a bit silly -- we should derive Deserialize on our CommitObject let rs_commit = drisl::from_slice::(commit_block) .map_err(|e| FirehoseValidationError::BadCommitBlock(e.to_string()))?; let commit: CommitObject = (&rs_commit).try_into()?; + let mut invertible = Vec::with_capacity(c.ops.len()); + let mut view_ops = Vec::with_capacity(c.ops.len()); + for op in c.ops.iter() { + let (invert, view) = convert_op(op)?; + invertible.push(invert); + view_ops.push(view); + } + // make sure all record blocks for created/updated ops are present - for op in c - .ops - .iter() - .filter(|o| matches!(o.action, RepoOpAction::Create | RepoOpAction::Update)) - { - let link = op - .cid - .as_ref() - .ok_or(FirehoseValidationError::MissingRecordLink { - action: op.action.clone(), - path: op.path.to_string(), - })? - .to_ipld() - .map_err(|_| FirehoseValidationError::BadCid("#commit.ops[].cid"))?; - if !blocks.contains_key(&link) { + for (op, view) in c.ops.iter().zip(&view_ops) { + if matches!(view.kind, OpKind::Delete { .. }) { + continue; + } + let cid = view.cid(); + if !blocks.contains_key(&cid) { return Err(FirehoseValidationError::MissingRecordBlock { action: op.action.clone(), - cid: format!("{:?}", link), + cid: cid.to_string(), }); } } @@ -214,17 +307,12 @@ impl FirehoseCommit { // 2.5 bonus checks: // - avoid confusion between CAR root link and firehose-commit event's // `commit` link: they must be the same. - let commit_link = c - .commit - .to_ipld() - .map_err(|_| FirehoseValidationError::BadCid("#commit.commit"))?; + let commit_link = link_cid(&c.commit, "#commit.commit")?; if root != commit_link { - return Err(FirehoseValidationError::CidMismatch(Box::new( - CidMismatch { - claimed: commit_link, - computed: root, - }, - ))); + return Err(FirehoseValidationError::RootNotCommit { + root, + commit: commit_link, + }); } // - make sure the commit object's DID matches the event's if commit.did != (&c.repo).into() { @@ -243,47 +331,22 @@ impl FirehoseCommit { // 3. apply operation inversion, matching against claimed `prevData` - // claimed prevData setup - let claimed_prev_data = c + let prev_link = c .prev_data .as_ref() - .ok_or(FirehoseValidationError::MissingPrevData)? - .to_ipld() - .map_err(|_| FirehoseValidationError::BadCid("#commit.prevData"))?; - - // stash a hubble-sync style copy of the map for later - let view_blocks: HashMap = blocks - .iter() - .map(|(ipld, b)| Ok((to_dasl_cid(*ipld, "blocks")?, b.clone()))) - .collect::>()?; - - // for now we use jacquard's memory store impl (CAR is size-bounded) - let store = Arc::new(MemoryBlockStore::new_from_blocks(blocks)); - let mut mst = Mst::load(store, commit.data_ipld(), None); - - // stash for hubble-sync ops for the commit view later - let mut view_ops = Vec::with_capacity(c.ops.len()); - - for op in c.ops.iter() { - let invertible = to_invertible_op(op)?; + .ok_or(FirehoseValidationError::MissingPrevData)?; + let claimed_prev_data = link_cid(prev_link, "#commit.prevData")?; - // we convert from VerifiedWriteOps bc we can do that infallibly - let view_op = Op::try_from(&invertible)?; - - if !mst.invert_op(invertible).await? { - return Err(FirehoseValidationError::FailedToInvert { - action: op.action.clone(), - path: op.path.to_string(), - }); - } - - view_ops.push(view_op); + let mut tree = Tree::new(&blocks, commit.data); + for op in &invertible { + tree.invert(op)?; + consume_budget().await; } - let computed_prev_root = mst.get_pointer().await?; - if computed_prev_root != claimed_prev_data { + let computed = tree.root()?; + if computed != claimed_prev_data { return Err(FirehoseValidationError::PrevDataMismatch { - claimed_prev_data: format!("{:?}", claimed_prev_data), - computed: format!("{}", computed_prev_root), + claimed_prev_data: claimed_prev_data.to_string(), + computed: computed.to_string(), }); } @@ -294,16 +357,14 @@ impl FirehoseCommit { let data = commit.data; - let prev = to_dasl_cid(claimed_prev_data, "prevData")?; - Ok(Self { commit, data, - prev, + prev: claimed_prev_data, rev: c.rev.into(), ops: view_ops, since: c.since.map(Into::into), - blocks: view_blocks, + blocks, }) } } @@ -311,30 +372,3 @@ impl FirehoseCommit { fn to_dasl_cid(ipld: IpldCid, source: &'static str) -> Result { DaslCid::from_bytes_raw(&ipld.to_bytes()).map_err(|_| FirehoseValidationError::BadCid(source)) } - -impl TryFrom<&VerifiedWriteOp> for Op { - type Error = FirehoseValidationError; - fn try_from(r: &VerifiedWriteOp) -> Result { - Ok(match r { - VerifiedWriteOp::Create { key, cid } => Op { - path: key.as_str().into(), - kind: OpKind::Create { - cid: to_dasl_cid(*cid, "ops")?, - }, - }, - VerifiedWriteOp::Update { key, cid, prev } => Op { - path: key.as_str().into(), - kind: OpKind::Update { - cid: to_dasl_cid(*cid, "ops")?, - prev: to_dasl_cid(*prev, "ops")?, - }, - }, - VerifiedWriteOp::Delete { key, prev } => Op { - path: key.as_str().into(), - kind: OpKind::Delete { - prev: to_dasl_cid(*prev, "ops")?, - }, - }, - }) - } -} diff --git a/hubble-sync/src/lib.rs b/hubble-sync/src/lib.rs index 1b849b1..315bbb2 100644 --- a/hubble-sync/src/lib.rs +++ b/hubble-sync/src/lib.rs @@ -25,13 +25,14 @@ use host::Hostname; // public re-exports (TODO: needs cleaning up) -pub use hubble_commit::{Cid as DaslCid, CidParseError}; // vendored there until dasl releases pub use commit::{Commit, CommitObject, Op, OpKind}; pub use config::{DeepCrawl, SensitiveToken, SyncConfig, SyncScope, UpstreamConfig, UpstreamKind}; pub use firehose::{ - FirehoseAck, FirehoseConfig, FirehoseError, FirehoseEvent, FirehosePayload, FirehoseSubscriber, + FirehoseAck, FirehoseCommit, FirehoseConfig, FirehoseError, FirehoseEvent, FirehosePayload, + FirehoseSubscriber, FirehoseSync, FirehoseValidationError, }; pub use host::{Host, HostRegistry, HostRegistryConfig, HostRequestError, Sync11State}; +pub use hubble_commit::{Cid as DaslCid, CidParseError}; // vendored there until dasl releases pub use hubble_sync::{HubbleSync, HubbleSyncError}; pub use identity::{Did, DidMethod, HubbleSyncResolver, Resolve, SignatureError, SigningKey, did}; pub use metrics::{