diff --git a/Cargo.lock b/Cargo.lock index ced241dde..e241079a3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1448,7 +1448,6 @@ dependencies = [ "serde_json", "serial_test", "stats_alloc", - "stelae", "tar", "tempfile", "thiserror 2.0.18", diff --git a/Cargo.toml b/Cargo.toml index 217150385..55a4a3b14 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -114,7 +114,6 @@ tracing-subscriber = { version = "0.3.17", features = ["env-filter"] } stats_alloc = "0.1" serial_test = "3" dolos-redb3 = { path = "crates/redb3" } -stelae = { path = "crates/stelae" } [target.'cfg(unix)'.dev-dependencies] nix = { version = "0.30.1", features = ["signal", "resource"] } diff --git a/adrs/004_stelae_snapshots.md b/adrs/004_stelae_snapshots.md index a8bba41a0..d95a45129 100644 --- a/adrs/004_stelae_snapshots.md +++ b/adrs/004_stelae_snapshots.md @@ -309,40 +309,60 @@ 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**, so extracting the protocol later is a directory move rather than a refactor. +Four crates, all workspace members until the extraction. The Stelae half is two of them — the protocol a third party implements from and the profile-generic lifecycle machinery — and the boundary is checkable: **`cargo tree -e normal --all-features` for `stelae` and `stelae-driver` must contain no `dolos-*` package**, so extracting the pair is a directory move rather than a refactor. -``` -crates/stelae/ # package `stelae` — protocol, zero dolos deps +```text +crates/stelae/ # package `stelae` — the wire protocol, zero dolos deps lib.rs # errors, protocol constants, envelope media types - frame.rs # deterministic CBOR-seq record read/write - inscription.rs # schema, JCS encode/verify, digest, history invariant + frame.rs # deterministic CBOR-seq record read/write, Limits + codec.rs # fixed-arity decode helpers for layer content records + inscription.rs # schema, JCS encode/verify, digest, history invariant (history_for) profile.rs # Profile trait, layer-kind registry, media-type & tag naming rules digest.rs # streaming sha256 + zstd (diffId + blob digest in one pass) - sign.rs # Ed25519 detached signatures, trusted keys, k-of-n - plan.rs # restore planning: layer selection, progress file, resume, preflight + layer.rs # reading a layer without holding it + plan.rs # progress file, resume, remaining-bytes accounting + progress.rs # Observer: what a transfer says about itself while running + transport.rs # the SteleReader/SteleWriter seam and the blob index + dir.rs # a stele on a local filesystem oci.rs # feature `oci`: push with blob-skip, pull missing-only, tags, referrers tests/toy_profile.rs # a second, trivial profile — proves the core carries no Dolos assumption +crates/stelae-driver/ # package `stelae-driver` — profile-generic lifecycle, zero dolos deps + lib.rs # the driver's Error + profile.rs # DriverProfile: the dataset policy stelae::Profile deliberately refuses + predecessor.rs # Predecessor/First: what a publish follows and may carry forward + publish.rs # the chained-publish lifecycle: open, Tuning, Publishing, Chained, standing + restore.rs # Budget/Checkpoint/Outlook: restore bounds and the resume checkpoint + preflight.rs # one free-space policy, in both directions + reporting.rs # counting layers and records for the two drivers to report + retry.rs # bounded patience for an external that fails in bursts + digests.rs # the digests-layer codec (Cardano immutable-DB file hashes) + crates/snapshot/ # package `dolos-snapshot` — the io.txpipe.dolos.cardano profile - lib.rs # DolosProfile: name/version, media types, tag rendering, position/parameters + lib.rs # DolosProfile, driver re-exports, profile constants, error mapping + namespaces.rs # the closed set of state namespaces a Dolos stele carries layers/{blocks,indexes,logs,state,digests}.rs export.rs # stores -> layers, generic over dolos_core::Domain - restore.rs # layers -> store writes, per-epoch checkpointing + restore.rs # layer selection and restore into store writes + planning.rs # one epoch selection, one reading of the plan it produces + registry.rs # store-typed publish/preview/restore over OCI; Point, verify, inspect + publisher.rs # publishing as a sequence of steps a command drives + node.rs # registry auth and scratch-dir policy from node configuration + backfill.rs # feature `backfill`: the epoch-at-a-time publisher daemon + +crates/mithril/ # package `dolos-mithril` — the aggregator fetch; no stelae dependency ``` -New deps: `zstd`, `serde_jcs`, `ed25519-dalek`, `oci-client`. The Dolos-side crate keeps the name `snapshot` because that is this project's word for the artifact (`dolos snapshot`, `[snapshot]`, `tests/e2e/snapshot.rs`); `stele` is the protocol's word for the same thing. - -Everything is built against the engine-agnostic core traits. Existing APIs used: `ArchiveStore::get_range` / `iter_logs`, `StateStore::iter_entities` / `read_cursor`, `ArchiveWriter::apply` / `write_log`, `StateWriter::write_entity` / `apply_utxoset` / `set_cursor`, `IndexStore::initialize_schema`, `index_delta_from_utxo_delta`, `seed_wal_from_state`, `CardanoConfig.stop_epoch`. Missing APIs to add (thin wrappers over existing backend internals in both redb and fjall): +The Dolos-side crate keeps the name `snapshot` because that is this project's word for the artifact (`dolos snapshot`, `[snapshot]`, `tests/e2e/snapshot.rs`); `stele` is the protocol's word for the same thing. The planned `sign.rs` (Ed25519 detached signatures, trusted keys, k-of-n) has not been built: signing stays specified above and unimplemented, and nothing else in this section is aspirational. -1. `StateStore::iter_utxos()` — full UTxO-set iteration (export + live-UTxO index rebuild). -2. `IndexStore` iteration of archive tag/exact records by epoch range (export). -3. `IndexWriter::append_prehashed(records)` — direct insertion of pre-hashed records (restore). +Everything is built against the engine-agnostic core traits (`ArchiveStore`, `StateStore`, `IndexStore` and their writers, `seed_wal_from_state`, `CardanoConfig.stop_epoch`). The store APIs the initial design listed as missing — `StateStore::iter_utxos()`, epoch-ranged iteration of archive tag/exact records, `IndexWriter::append_prehashed` — have all since landed in `dolos-core` and its backends. ### CLI and configuration - `dolos snapshot publish [--repo oci://…] [--output-dir DIR] [--epochs N..M] [--dry-run]` — export layers; `--output-dir` writes blobs + inscription to disk, `--repo` pushes with blob-skip and moves tags. - `dolos snapshot digest` — compute and print the canonical inscription and its sha256 from local stores (what independent verifiers run and sign). -- `dolos snapshot verify | sign --key FILE | inspect`. +- `dolos snapshot verify | inspect`; `sign --key FILE` belongs to the signature + phase, which is specified and not yet built ("Code layout" above). - `dolos bootstrap snapshot` gains source-scheme dispatch: `oci://` → new path; https template / `--file` → legacy tarball, unchanged. `--point epoch-E|latest`; existing `--continue` drives resume; `sync.max_history` bounds how much history is fetched (subsumes the old `full`/`ledger` variants). ```toml diff --git a/crates/snapshot/src/lib.rs b/crates/snapshot/src/lib.rs index f89b6e4bd..4e57364b2 100644 --- a/crates/snapshot/src/lib.rs +++ b/crates/snapshot/src/lib.rs @@ -102,6 +102,13 @@ pub(crate) use stelae_driver::scope_key; /// [`export::publish`] and [`restore::restore_dir`] hold for the transports. pub use stelae::progress; +/// The protocol surface a host reaches through this crate rather than by +/// naming `stelae` itself: only this crate depends on the stelae crates, so +/// there is one pin point and one place to absorb upstream churn. These cover +/// what a test or a binary opens a stele with; anything else a consumer needs +/// joins the list rather than justifying a second dependency edge. +pub use stelae::{dir, inscription, transport, SteleReader}; + use dolos_cardano::model::{ AccountEpochLog, AccountState, AssetState, DRepState, DatumState, EpochState, EraSummary, FixedNamespace, GovState, PendingMirState, PendingRewardState, PoolState, ProposalState, diff --git a/crates/snapshot/src/restore.rs b/crates/snapshot/src/restore.rs index 9c0c03225..f7829f12a 100644 --- a/crates/snapshot/src/restore.rs +++ b/crates/snapshot/src/restore.rs @@ -127,10 +127,10 @@ use dolos_core::{ use stelae::{ frame::Limits, inscription::{Inscription, LayerDescriptor}, - plan::{Remaining, RestoreProgress, Resume}, + plan::{Remaining, Resume}, progress::{Event, Observer, Outcome}, transport::{BlobIndex, SteleReader}, - Digest, LayerHeader, + LayerHeader, }; use tracing::info; @@ -142,36 +142,24 @@ use crate::{ RETIRED_SCHEMA_REV, SCOPE_REQUIRED, STATE_KINDS, UTXOS, }; -/// What a restore holds at once. -/// -/// A store writer batches until `commit` and a layer arrives as a stream, so -/// nothing bounds a restore's memory except these numbers. Both commit ceilings -/// are needed and neither subsumes the other: an index record is tens of bytes -/// and only a count bounds it, while one epoch of mainnet blocks is gigabytes -/// and only a byte budget bounds that. -#[derive(Debug, Clone, Copy)] -pub struct Budget { - /// Per-record and window bounds on the layer read itself. - pub limits: Limits, - /// Records accumulated before a write batch is committed. - pub commit_records: usize, - /// Bytes accumulated before a write batch is committed. - pub commit_bytes: usize, -} +pub use stelae_driver::restore::{Budget, Checkpoint, Outlook}; -impl Default for Budget { - fn default() -> Self { - Self { - // The profile's ceiling, not the protocol's default: a restore that - // read under a tighter limit than the publisher wrote under would - // refuse this profile's own steles. - limits: Limits { - max_record: crate::MAX_RECORD, - ..Limits::default() - }, - commit_records: 50_000, - commit_bytes: 64 * 1024 * 1024, - } +/// This profile's restore [`Budget`]. +/// +/// The driver deliberately gives `Budget` no default — the read limits are +/// the publishing profile's ceilings, not the protocol's — so this is where +/// the numbers live. +pub fn default_budget() -> Budget { + Budget { + // The profile's ceiling, not the protocol's default: a restore that + // read under a tighter limit than the publisher wrote under would + // refuse this profile's own steles. + limits: Limits { + max_record: crate::MAX_RECORD, + ..Limits::default() + }, + commit_records: 50_000, + commit_bytes: 64 * 1024 * 1024, } } @@ -811,145 +799,29 @@ pub struct Summary { pub layers_skipped: usize, } -/// Where a restore records what it has finished, and what it inherits. -/// -/// One value rather than three arguments, because the three are one idea: the -/// file, the set of layers it says are done, and the identity of the stele -/// being restored into it. -pub struct Checkpoint { - path: PathBuf, - resume: Resume, - progress: RestoreProgress, -} - -impl Checkpoint { - /// Where a node with storage at `storage_path` keeps its progress file. - pub fn path_in(storage_path: &Path) -> PathBuf { - storage_path.join(PROGRESS_FILE) - } - - /// Open the checkpoint for restoring the stele `identity` into - /// `storage_path`. +impl Summary { + /// Tally [`Checkpoint::fetch`]'s decision. /// - /// `resume` is the operator's `--continue`, and it gates whether anything - /// on disk is *honoured* — not merely whether it is read. A restore - /// that is not resuming is starting over: it takes an empty [`Resume`] - /// and its first checkpoint overwrites whatever was there. - /// - /// That asymmetry is deliberate and is the reason `--force` is safe. A - /// progress file that outlived its stores would name layers whose data is - /// gone, and honouring one nobody asked to honour would skip them onto - /// empty stores — a node missing a slice of chain that nothing would - /// report. Clearing storage removes this file with the rest of it, and - /// the rule here means even a file that somehow survived cannot do that - /// damage. - pub fn open(storage_path: &Path, identity: Digest, resume: bool) -> Result { - let path = Self::path_in(storage_path); - - let existing = match resume { - true => RestoreProgress::load(&path)?, - false => None, - }; - - let resume = Resume::from_progress(existing.as_ref()); - - // The new identity, the old completions. The completions are what the - // resume rule is about — content, not the document that described it — - // and the digest is what tells a later reader which stele a - // half-finished restore was aimed at. - let progress = RestoreProgress { - inscription_digest: identity, - completed: existing.map(|p| p.completed).unwrap_or_default(), - }; - - Ok(Self { - path, - resume, - progress, - }) - } - - /// A restore that checkpoints nowhere. - /// - /// For a caller driving [`restore`] without a node behind it — the test - /// suites, above all, which compare store sets rather than resumes. - pub fn none() -> Self { - Self { - path: PathBuf::new(), - resume: Resume::none(), - progress: RestoreProgress::new(Digest::from_bytes([0; 32])), + /// The driver returns the outcome precisely so the count cannot drift + /// from the checkpoint's own choice. + fn count(&mut self, outcome: Outcome) { + match outcome { + Outcome::Skipped => self.layers_skipped += 1, + _ => self.layers_fetched += 1, } } +} - /// What this checkpoint inherits, for the remaining-bytes accounting. - pub fn resume(&self) -> &Resume { - &self.resume - } - - /// Read `descriptor`'s layer unless an earlier attempt already committed - /// it. - /// - /// The one place a layer is decided about, so that the skip, the count and - /// the checkpoint cannot drift apart. `fetch` runs to completion — every - /// per-kind driver below commits before it returns — and only then is the - /// layer recorded, which is what makes the record mean "committed" rather - /// than "attempted". - /// Returns the outcome alongside the count, rather than leaving a caller to - /// ask the resume the same question a second time: this method is the one - /// place a layer is decided about, and what an observer reports has to be - /// that decision and not a re-derivation of it. - fn fetch( - &mut self, - descriptor: &LayerDescriptor, - summary: &mut Summary, - fetch: impl FnOnce() -> Result, - ) -> Result<(T, Outcome), Error> { - if self.resume.is_done(&descriptor.diff_id) { - info!( - kind = descriptor.kind, - scope = %descriptor.scope, - "skipping a layer an earlier attempt completed" - ); - - summary.layers_skipped += 1; - - return Ok((T::default(), Outcome::Skipped)); - } - - let out = fetch()?; - - summary.layers_fetched += 1; - self.record(descriptor.diff_id)?; - - Ok((out, Outcome::Transferred)) - } - - fn record(&mut self, diff_id: Digest) -> Result<(), Error> { - if self.path.as_os_str().is_empty() { - return Ok(()); - } - - self.progress.record(diff_id); - self.progress.save(&self.path)?; - - Ok(()) - } - - /// Delete the progress file. - /// - /// Called after the live-UTxO rebuild and never after `set_cursor`, which - /// are two different moments and only the later one means the restore is - /// finished. Clearing it at the cursor would take away the resume that - /// repairs exactly the window between them. - fn clear(&self) -> Result<(), Error> { - if self.path.as_os_str().is_empty() { - return Ok(()); - } - - RestoreProgress::remove(&self.path)?; - - Ok(()) - } +/// Where a node with storage at `storage_path` keeps its progress file. +/// +/// The [`Checkpoint`] takes the path it is given — the driver never derives +/// dolos layout — and this is where dolos derives it. The `resume` flag a +/// checkpoint is opened with is the operator's `--continue`, and the driver's +/// ignore-unless-resuming rule is the reason `--force` is safe: clearing +/// storage removes this file with the rest of it, and even a progress file +/// that somehow survived its stores cannot skip layers onto empty ones. +pub fn progress_path_in(storage_path: &Path) -> PathBuf { + storage_path.join(PROGRESS_FILE) } /// The three stores a restore writes into. @@ -1053,11 +925,11 @@ where if let Some(descriptor) = &epoch.blocks { let at = cursor.open(BLOCKS, &descriptor.scope); - let (count, outcome) = checkpoint.fetch(descriptor, &mut summary, || { - restore_blocks(&reader, descriptor, archive) - })?; + let (count, outcome) = + checkpoint.fetch(descriptor, || restore_blocks(&reader, descriptor, archive))?; cursor.close(at, BLOCKS, outcome); + summary.count(outcome); summary.blocks += count; } @@ -1065,22 +937,23 @@ where let kind = descriptor.kind.as_str(); let at = cursor.open(kind, &descriptor.scope); - let (count, outcome) = checkpoint.fetch(descriptor, &mut summary, || { + let (count, outcome) = checkpoint.fetch(descriptor, || { restore_logs(&reader, descriptor, archive, ns) })?; cursor.close(at, kind, outcome); + summary.count(outcome); summary.logs += count; } if let Some(descriptor) = &epoch.indexes { let at = cursor.open(INDEXES, &descriptor.scope); - let (count, outcome) = checkpoint.fetch(descriptor, &mut summary, || { - restore_indexes(&reader, descriptor, indexes) - })?; + let (count, outcome) = + checkpoint.fetch(descriptor, || restore_indexes(&reader, descriptor, indexes))?; cursor.close(at, INDEXES, outcome); + summary.count(outcome); summary.index_records += count; } } @@ -1123,19 +996,6 @@ where Ok(summary) } -/// What a restore is about to do, once the stele has been read. -/// -/// Returned alongside the [`Plan`] so a caller can report the *remaining* -/// download rather than the original one — the whole point of the accounting on -/// a resumed run. -#[derive(Debug, Clone, Copy)] -pub struct Outlook { - /// Layers still to fetch, and what they weigh compressed. - pub remaining: Remaining, - /// Layers an earlier attempt had already committed. - pub inherited: usize, -} - /// Open, verify and read a stele into the stores, in one call. /// /// The front door for a caller holding a source and a configuration — the @@ -1167,7 +1027,8 @@ where let plan = plan(stele, node.network_magic, node.max_history)?; let identity = stele.read_inscription()?.digest()?; - let mut checkpoint = Checkpoint::open(node.storage_path, identity, node.resume)?; + let mut checkpoint = + Checkpoint::open(progress_path_in(node.storage_path), identity, node.resume)?; let index = stele.blob_index()?; @@ -1196,7 +1057,7 @@ where &index, &plan, target, - Budget::default(), + default_budget(), &mut checkpoint, observer, )?; diff --git a/crates/snapshot/tests/restore.rs b/crates/snapshot/tests/restore.rs index 73d66986e..795294501 100644 --- a/crates/snapshot/tests/restore.rs +++ b/crates/snapshot/tests/restore.rs @@ -47,7 +47,7 @@ use dolos_core::{ }; use dolos_snapshot::{ is_state_kind, - restore::{self, Budget, Checkpoint}, + restore::{self, default_budget, progress_path_in, Budget, Checkpoint}, state_layer_count, state_ns_for, DolosProfile, Error, RetainedEpochs, COMPRESSION_LEVEL, KINDS, NAMESPACES, STATE_KINDS, UTXOS, }; @@ -161,7 +161,7 @@ fn a_stele_for_another_network_is_refused_before_anything_is_written() { temp.path(), dolos_snapshot::MAINNET_MAGIC, &blank, - Budget::default(), + default_budget(), ) .unwrap_err(); @@ -312,7 +312,7 @@ fn an_unknown_layer_kind_is_skipped_and_reported() { assert_eq!(stele.blob_index().unwrap().len(), inscription.layers.len()); let blank = Blank::::open(); - let summary = restore_into(temp.path(), magic, &blank, Budget::default()).unwrap(); + let summary = restore_into(temp.path(), magic, &blank, default_budget()).unwrap(); assert_eq!(summary.layers_fetched, plan.layers().count()); @@ -336,7 +336,7 @@ fn a_required_unknown_layer_kind_is_refused_before_anything_is_written() { export_one_kind_ahead(temp.path(), &domain, json!({"epoch": 0, "required": true})); let blank = Blank::::open(); - let err = restore_into(temp.path(), magic, &blank, Budget::default()).unwrap_err(); + let err = restore_into(temp.path(), magic, &blank, default_budget()).unwrap_err(); let Error::RequiredUnknownLayer { kind, scope } = &err else { panic!("{err:?}"); @@ -379,7 +379,7 @@ fn a_restore_that_fails_partway_leaves_no_cursor() { std::fs::remove_file(stele.blob_path(&blob)).unwrap(); let blank = Blank::::open(); - let err = restore_into(temp.path(), magic_of(&domain), &blank, Budget::default()).unwrap_err(); + let err = restore_into(temp.path(), magic_of(&domain), &blank, default_budget()).unwrap_err(); assert!( matches!(err, Error::Stelae(stelae::Error::LayerNotFound { .. })), @@ -518,7 +518,7 @@ fn a_restored_node_is_the_node_it_came_from_on_fjall() { /// restore against the export it came from and so cannot tell a faithful /// restore from one that reproduced the export's own mistake; this can. fn cross_check() { - let (_, blank, _) = round_trip::(Budget::default()); + let (_, blank, _) = round_trip::(default_budget()); let replayed: ToyDomain = harness(); assert_stores_match(&blank, &replayed); @@ -805,7 +805,7 @@ fn restore_watched( let plan = restore::plan(&stele, magic, None)?; let index = stele.blob_index()?; - let mut checkpoint = Checkpoint::open(storage, identity, resume)?; + let mut checkpoint = Checkpoint::open(progress_path_in(storage), identity, resume)?; match stop_at { Some(stop_at) => restore::restore( @@ -816,7 +816,7 @@ fn restore_watched( &index, &plan, target(blank), - Budget::default(), + default_budget(), &mut checkpoint, observer, ), @@ -825,7 +825,7 @@ fn restore_watched( &index, &plan, target(blank), - Budget::default(), + default_budget(), &mut checkpoint, observer, ), @@ -877,7 +877,7 @@ fn kill_and_resume() { // What the killed run left behind: a progress file naming exactly the // layers that committed, and no cursor. - let progress = RestoreProgress::load(&Checkpoint::path_in(storage.path())) + let progress = RestoreProgress::load(&progress_path_in(storage.path())) .unwrap() .expect("a killed restore left no progress file"); @@ -909,7 +909,7 @@ fn kill_and_resume() { // The progress file is gone: the restore finished. assert_eq!( - RestoreProgress::load(&Checkpoint::path_in(storage.path())).unwrap(), + RestoreProgress::load(&progress_path_in(storage.path())).unwrap(), None, "a finished restore left its progress file behind" ); @@ -984,9 +984,9 @@ fn a_stele_with_retained_dumps_restores_the_tip_and_reports_the_dumps() { let dumped_blank = Blank::::open(); let from_plain = - restore_into(plain_root.path(), magic, &plain_blank, Budget::default()).unwrap(); + restore_into(plain_root.path(), magic, &plain_blank, default_budget()).unwrap(); let from_dumped = - restore_into(dumped_root.path(), magic, &dumped_blank, Budget::default()).unwrap(); + restore_into(dumped_root.path(), magic, &dumped_blank, default_budget()).unwrap(); assert_eq!(from_plain, from_dumped); assert_state_matches(dumped_blank.state(), plain_blank.state()); @@ -1036,7 +1036,7 @@ fn a_resumed_restore_of_a_dumped_stele_records_only_the_epoch_layers() { "{err:?}" ); - let progress = RestoreProgress::load(&Checkpoint::path_in(storage.path())) + let progress = RestoreProgress::load(&progress_path_in(storage.path())) .unwrap() .expect("a killed restore left no progress file"); @@ -1068,7 +1068,7 @@ fn a_resumed_restore_of_a_dumped_stele_records_only_the_epoch_layers() { ); assert_eq!( - RestoreProgress::load(&Checkpoint::path_in(storage.path())).unwrap(), + RestoreProgress::load(&progress_path_in(storage.path())).unwrap(), None, "a finished restore left its progress file behind" ); @@ -1171,7 +1171,7 @@ fn a_newer_inscription_keeps_the_epoch_layers_and_redoes_the_tip() { "{err:?}" ); - let seeded = RestoreProgress::load(&Checkpoint::path_in(storage.path())) + let seeded = RestoreProgress::load(&progress_path_in(storage.path())) .unwrap() .expect("the interrupted restore left no progress file") .completed @@ -1289,7 +1289,7 @@ fn a_restore_that_is_not_resuming_honours_no_progress_file() { let inscription = export_to(stele.path(), &domain); let storage = tempfile::tempdir().unwrap(); - let path = Checkpoint::path_in(storage.path()); + let path = progress_path_in(storage.path()); // A progress file claiming every epoch layer is done, over empty stores. let mut progress = RestoreProgress::new(inscription.digest().unwrap()); diff --git a/crates/snapshot/tests/restore_registry.rs b/crates/snapshot/tests/restore_registry.rs index 8265d55d3..e530a1cc9 100644 --- a/crates/snapshot/tests/restore_registry.rs +++ b/crates/snapshot/tests/restore_registry.rs @@ -55,7 +55,7 @@ use dolos_core::{ use dolos_snapshot::{ export::Plan, registry::{self, Point}, - restore::{self, Budget, Checkpoint}, + restore::{self, default_budget, progress_path_in, Checkpoint}, state_layer_count, Error, Network, NAMESPACES, UTXOS, }; use dolos_testing::toy_domain::{MemoryStores, ToyDomain, ToyStores}; @@ -352,7 +352,8 @@ fn a_killed_registry_restore_resumes_where_it_stopped() { assert!(epoch_layers.len() >= 2); // The interruption, at the second epoch layer the driver reaches. - let mut checkpoint = Checkpoint::open(storage.path(), identity, false).unwrap(); + let mut checkpoint = + Checkpoint::open(progress_path_in(storage.path()), identity, false).unwrap(); let err = restore::restore( &Interrupted { @@ -362,7 +363,7 @@ fn a_killed_registry_restore_resumes_where_it_stopped() { &index, &plan, target(&blank), - Budget::default(), + default_budget(), &mut checkpoint, &Observer::silent(), ) @@ -373,7 +374,7 @@ fn a_killed_registry_restore_resumes_where_it_stopped() { "{err:?}" ); - let progress = RestoreProgress::load(&Checkpoint::path_in(storage.path())) + let progress = RestoreProgress::load(&progress_path_in(storage.path())) .unwrap() .expect("a killed restore left no progress file"); @@ -400,7 +401,7 @@ fn a_killed_registry_restore_resumes_where_it_stopped() { assert_eq!(resumed.layers_fetched, PER_PUBLISH - 1); assert_eq!( - RestoreProgress::load(&Checkpoint::path_in(storage.path())).unwrap(), + RestoreProgress::load(&progress_path_in(storage.path())).unwrap(), None, "a finished restore left its progress file behind" ); @@ -463,7 +464,8 @@ fn a_pre_seeded_node_fetches_only_what_it_lacks() { let shards: Vec = plan.tip_layers().map(|l| l.diff_id).collect(); let epoch_layers = plan.immutable_layers().count(); - let mut checkpoint = Checkpoint::open(storage.path(), identity, false).unwrap(); + let mut checkpoint = + Checkpoint::open(progress_path_in(storage.path()), identity, false).unwrap(); restore::restore( &Interrupted { @@ -473,13 +475,13 @@ fn a_pre_seeded_node_fetches_only_what_it_lacks() { &index, &plan, target(&blank), - Budget::default(), + default_budget(), &mut checkpoint, &Observer::silent(), ) .unwrap_err(); - let seeded = RestoreProgress::load(&Checkpoint::path_in(storage.path())) + let seeded = RestoreProgress::load(&progress_path_in(storage.path())) .unwrap() .unwrap() .completed @@ -565,7 +567,7 @@ fn a_point_that_names_no_stele_is_refused() { ); assert_eq!( - RestoreProgress::load(&Checkpoint::path_in(storage.path())).unwrap(), + RestoreProgress::load(&progress_path_in(storage.path())).unwrap(), None, "a restore that never started left a progress file" ); @@ -944,7 +946,8 @@ fn a_resumed_registry_restore_reports_what_it_skipped() { let epoch_layers: Vec = plan.immutable_layers().map(|l| l.diff_id).collect(); assert!(epoch_layers.len() >= 2); - let mut checkpoint = Checkpoint::open(storage.path(), identity, false).unwrap(); + let mut checkpoint = + Checkpoint::open(progress_path_in(storage.path()), identity, false).unwrap(); restore::restore( &Interrupted { @@ -954,7 +957,7 @@ fn a_resumed_registry_restore_reports_what_it_skipped() { &index, &plan, target(&blank), - Budget::default(), + default_budget(), &mut checkpoint, &Observer::silent(), ) diff --git a/crates/stelae-driver/src/lib.rs b/crates/stelae-driver/src/lib.rs index 936f9a7a4..69ab074e3 100644 --- a/crates/stelae-driver/src/lib.rs +++ b/crates/stelae-driver/src/lib.rs @@ -39,6 +39,7 @@ pub mod profile; #[cfg(feature = "oci")] pub mod publish; pub mod reporting; +pub mod restore; pub mod retry; pub use predecessor::{First, Predecessor}; diff --git a/crates/stelae-driver/src/restore.rs b/crates/stelae-driver/src/restore.rs new file mode 100644 index 000000000..a9c9ea66c --- /dev/null +++ b/crates/stelae-driver/src/restore.rs @@ -0,0 +1,251 @@ +//! Restore planning shapes and the resume checkpoint. +//! +//! The profile owns a restore's selection — a layer's `scope` is opaque to the +//! protocol, so only the profile can read an epoch out of one — and the stores +//! it writes into. What lives here is the part that is the same whatever is +//! being restored: how much a restore holds at once ([`Budget`]), what it still +//! has to do ([`Outlook`]), and where it records what it has finished +//! ([`Checkpoint`]). +//! +//! The checkpoint's rule is [`stelae::plan::Resume`]'s — a layer is done when +//! its `diffId` is recorded, which is a fact about bytes and not about the +//! stele they were published in. *Which* layers may be skipped at all is the +//! profile's half of the split, and stays with it: this type is only ever +//! asked about the layers a profile chose to ask about. + +use std::path::PathBuf; + +use stelae::{ + frame::Limits, + inscription::LayerDescriptor, + plan::{Remaining, RestoreProgress, Resume}, + progress::Outcome, + Digest, +}; +use tracing::info; + +use crate::Error; + +/// What a restore holds at once. +/// +/// A store writer batches until `commit` and a layer arrives as a stream, so +/// nothing bounds a restore's memory except these numbers. Both commit ceilings +/// are needed and neither subsumes the other: an index record is tens of bytes +/// and only a count bounds it, while one epoch of blocks can run to gigabytes +/// and only a byte budget bounds that. +/// +/// There is deliberately no `Default`: the read limits are the publishing +/// profile's ceilings, not the protocol's defaults — a restore that read under +/// a tighter limit than the publisher wrote under would refuse that profile's +/// own steles — so the profile supplies its budget. +#[derive(Debug, Clone, Copy)] +pub struct Budget { + /// Per-record and window bounds on the layer read itself. + pub limits: Limits, + /// Records accumulated before a write batch is committed. + pub commit_records: usize, + /// Bytes accumulated before a write batch is committed. + pub commit_bytes: usize, +} + +/// What a restore is about to do, once the stele has been read. +/// +/// Returned alongside the profile's plan so a caller can report the +/// *remaining* download rather than the original one — the whole point of the +/// accounting on a resumed run. +#[derive(Debug, Clone, Copy)] +pub struct Outlook { + /// Layers still to fetch, and what they weigh compressed. + pub remaining: Remaining, + /// Layers an earlier attempt had already committed. + pub inherited: usize, +} + +/// Where a restore records what it has finished, and what it inherits. +/// +/// One value rather than three arguments, because the three are one idea: the +/// file, the set of layers it says are done, and the identity of the stele +/// being restored into it. +pub struct Checkpoint { + path: PathBuf, + resume: Resume, + progress: RestoreProgress, +} + +impl Checkpoint { + /// Open the checkpoint at `path` for restoring the stele `identity`. + /// + /// The path is the caller's: the driver never derives where a node keeps + /// its progress file. `resume` gates whether anything on disk is + /// *honoured* — not merely whether it is read. A restore that is not + /// resuming is starting over: it takes an empty [`Resume`] and its first + /// checkpoint overwrites whatever was there. + /// + /// That asymmetry is deliberate. A progress file that outlived the stores + /// beside it would name layers whose data is gone, and honouring one + /// nobody asked to honour would skip them onto empty stores — a node + /// missing a slice of data that nothing would report. The rule here means + /// even a file that somehow survived its stores cannot do that damage. + pub fn open(path: PathBuf, identity: Digest, resume: bool) -> Result { + let existing = match resume { + true => RestoreProgress::load(&path)?, + false => None, + }; + + let resume = Resume::from_progress(existing.as_ref()); + + // The new identity, the old completions. The completions are what the + // resume rule is about — content, not the document that described it — + // and the digest is what tells a later reader which stele a + // half-finished restore was aimed at. + let progress = RestoreProgress { + inscription_digest: identity, + completed: existing.map(|p| p.completed).unwrap_or_default(), + }; + + Ok(Self { + path, + resume, + progress, + }) + } + + /// A restore that checkpoints nowhere. + /// + /// For a caller driving a restore without a node behind it — the test + /// suites, above all, which compare store sets rather than resumes. + pub fn none() -> Self { + Self { + path: PathBuf::new(), + resume: Resume::none(), + progress: RestoreProgress::new(Digest::from_bytes([0; 32])), + } + } + + /// What this checkpoint inherits, for the remaining-bytes accounting. + pub fn resume(&self) -> &Resume { + &self.resume + } + + /// Read `descriptor`'s layer unless an earlier attempt already committed + /// it. + /// + /// The one place a layer is decided about, so that the skip and the + /// checkpoint cannot drift apart. `fetch` runs to completion — the caller + /// commits before it returns — and only then is the layer recorded, which + /// is what makes the record mean "committed" rather than "attempted". + /// Returns the outcome alongside the value, rather than leaving a caller + /// to ask the resume the same question a second time: what an observer or + /// a summary reports has to be this decision and not a re-derivation of + /// it. + pub fn fetch>( + &mut self, + descriptor: &LayerDescriptor, + fetch: impl FnOnce() -> Result, + ) -> Result<(T, Outcome), E> { + if self.resume.is_done(&descriptor.diff_id) { + info!( + kind = descriptor.kind, + scope = %descriptor.scope, + "skipping a layer an earlier attempt completed" + ); + + return Ok((T::default(), Outcome::Skipped)); + } + + let out = fetch()?; + + self.record(descriptor.diff_id)?; + + Ok((out, Outcome::Transferred)) + } + + fn record(&mut self, diff_id: Digest) -> Result<(), Error> { + if self.path.as_os_str().is_empty() { + return Ok(()); + } + + self.progress.record(diff_id); + self.progress.save(&self.path)?; + + Ok(()) + } + + /// Delete the progress file. + /// + /// For the moment the restore is *finished* — which is the caller's call, + /// not the last `fetch`'s: whatever work follows the final layer is + /// exactly the window a kept progress file lets an operator repair by + /// resuming. + pub fn clear(&self) -> Result<(), Error> { + if self.path.as_os_str().is_empty() { + return Ok(()); + } + + RestoreProgress::remove(&self.path)?; + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn descriptor(byte: u8) -> LayerDescriptor { + LayerDescriptor { + kind: "blocks".into(), + media_type: "application/cbor-seq".into(), + diff_id: Digest::from_bytes([byte; 32]), + records: 1, + uncompressed_size: 1, + scope: serde_json::json!({"epoch": 7}), + } + } + + #[test] + fn a_resumed_checkpoint_skips_what_it_recorded() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("progress.json"); + let identity = Digest::from_bytes([1; 32]); + + let mut first = Checkpoint::open(path.clone(), identity, false).unwrap(); + let (_, outcome) = first.fetch::<(), Error>(&descriptor(2), || Ok(())).unwrap(); + assert!(matches!(outcome, Outcome::Transferred)); + + let mut second = Checkpoint::open(path, identity, true).unwrap(); + let (_, outcome) = second + .fetch::<(), Error>(&descriptor(2), || panic!("must not refetch")) + .unwrap(); + assert!(matches!(outcome, Outcome::Skipped)); + } + + #[test] + fn a_fresh_checkpoint_ignores_what_is_on_disk() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("progress.json"); + let identity = Digest::from_bytes([1; 32]); + + let mut first = Checkpoint::open(path.clone(), identity, false).unwrap(); + first.fetch::<(), Error>(&descriptor(2), || Ok(())).unwrap(); + + let mut fresh = Checkpoint::open(path, identity, false).unwrap(); + let mut fetched = false; + fresh + .fetch::<(), Error>(&descriptor(2), || { + fetched = true; + Ok(()) + }) + .unwrap(); + assert!(fetched); + } + + #[test] + fn a_checkpoint_that_goes_nowhere_records_nothing() { + let mut none = Checkpoint::none(); + none.fetch::<(), Error>(&descriptor(2), || Ok(())).unwrap(); + assert!(none.resume().is_empty()); + + none.clear().unwrap(); + } +} diff --git a/src/adapters/storage.rs b/src/adapters/storage.rs index a5988f072..d50326529 100644 --- a/src/adapters/storage.rs +++ b/src/adapters/storage.rs @@ -1617,7 +1617,7 @@ mod tests { #[cfg(test)] mod lifecycle_tests { - use dolos_snapshot::restore::Checkpoint; + use dolos_snapshot::restore::progress_path_in; /// A `--force` wipe takes the progress file with the data it describes. /// @@ -1638,7 +1638,7 @@ mod lifecycle_tests { std::fs::create_dir_all(&storage).unwrap(); - let progress = Checkpoint::path_in(&storage); + let progress = progress_path_in(&storage); std::fs::write(&progress, b"{}").unwrap(); // A stand-in for the stores the progress file describes, so the @@ -1675,6 +1675,6 @@ mod lifecycle_tests { fn the_progress_file_lives_inside_the_storage_path() { let storage = std::path::Path::new("/var/lib/dolos/data"); - assert!(Checkpoint::path_in(storage).starts_with(storage)); + assert!(progress_path_in(storage).starts_with(storage)); } } diff --git a/src/bin/dolos/feedback.rs b/src/bin/dolos/feedback.rs index 0b0f8e046..130636679 100644 --- a/src/bin/dolos/feedback.rs +++ b/src/bin/dolos/feedback.rs @@ -354,6 +354,21 @@ impl MithrilFeedback { } } +/// A round's bars leave the terminal with the round. +/// +/// Both callers build a receiver per download round against one shared +/// [`MultiProgress`], and a finished bar stays drawn until it is cleared — so +/// without this, a backfill daemon accumulates two dead bars per window for +/// as long as it runs. Tied to drop rather than to a method because the +/// receiver crosses the daemon seam as an opaque `Arc` and never comes back. +#[cfg(feature = "mithril")] +impl Drop for MithrilFeedback { + fn drop(&mut self) { + self.aggregate_pb.finish_and_clear(); + self.validate_pb.finish_and_clear(); + } +} + #[cfg(feature = "mithril")] #[async_trait::async_trait] impl mithril_client::feedback::FeedbackReceiver for MithrilFeedback { diff --git a/tests/index_roundtrip.rs b/tests/index_roundtrip.rs index 93cce77f1..5e5edac60 100644 --- a/tests/index_roundtrip.rs +++ b/tests/index_roundtrip.rs @@ -1132,13 +1132,13 @@ fn mib(bytes: u64) -> f64 { #[test] #[ignore = "measurement, not an assertion"] fn measure_layer_sink_residency() { + use dolos_snapshot::transport::{RecordSink as _, SteleWriter as _}; use dolos_snapshot::{DolosProfile, EpochScope, Scope as _, COMPRESSION_LEVEL, INDEXES}; - use stelae::transport::{RecordSink as _, SteleWriter as _}; const SINKS: u64 = 32; let temp = tempfile::tempdir().expect("tempdir"); - let stele = stelae::dir::SteleDir::create(temp.path()).expect("create stele"); + let stele = dolos_snapshot::dir::SteleDir::create(temp.path()).expect("create stele"); // A record for every sink to chew on, built once: the point is the // compressor's resident state, not the bytes handed to it. @@ -1280,7 +1280,7 @@ fn publish_at_band( store: &S, epochs: u64, band: usize, -) -> (stelae::inscription::Inscription, tempfile::TempDir) { +) -> (dolos_snapshot::inscription::Inscription, tempfile::TempDir) { use dolos_core::{StateStore as _, StateWriter as _}; use dolos_snapshot::{ export::{IndexBand, Plan}, @@ -1324,7 +1324,7 @@ fn publish_at_band( &state, store, None, - &stelae::progress::Observer::silent(), + &dolos_snapshot::progress::Observer::silent(), ) .expect("publish"); diff --git a/tests/snapshot_publish.rs b/tests/snapshot_publish.rs index e00e82063..75b28ac9c 100644 --- a/tests/snapshot_publish.rs +++ b/tests/snapshot_publish.rs @@ -9,8 +9,8 @@ mod node; +use dolos_snapshot::{dir::SteleDir, SteleReader}; use node::{assert_ok, toml_string, Node}; -use stelae::{dir::SteleDir, SteleReader}; /// Done criterion 5. #[test]