diff --git a/CHANGELOG.md b/CHANGELOG.md index efaa4247c610..d671c06af10a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,8 @@ - [#7374](https://github.com/ChainSafe/forest/pull/7374): Bounded `nEpochs` in `Forest.StateCompute` RPC method to 2000 which can be overriden by `FOREST_STATE_COMPUTE_MAX_RANGE`. +- [#7383](https://github.com/ChainSafe/forest/pull/7383): `Forest.ChainExportStatus` now reports a `state` (`Idle`/`Running`/`Succeeded`/`Cancelled`/`Failed`) and the export `kind` (`Snapshot`/`DiffSnapshot`/`SnapshotGc`), along with the `error` of a failed export. The `exporting`, `cancelled` and `succeeded` booleans were removed. `forest-cli snapshot export-status --wait` now exits with an error when the watched export fails. + ### Added - [#7376](https://github.com/ChainSafe/forest/pull/7376): Added `--augmented-snapshot` and `--tipset-lookup` to `forest-cli snapshot export`. diff --git a/scripts/tests/calibnet_export_check.sh b/scripts/tests/calibnet_export_check.sh index d486542ae408..c7c2d5747a58 100755 --- a/scripts/tests/calibnet_export_check.sh +++ b/scripts/tests/calibnet_export_check.sh @@ -18,9 +18,9 @@ echo "Cleaning up the initial snapshot" rm --force --verbose ./*.{car,car.zst,sha256sum} output=$($FOREST_CLI_PATH snapshot export-status --format json) -is_exporting=$(echo "$output" | jq -r '.exporting') +state=$(echo "$output" | jq -r '.state') echo "Testing that no export is in progress" -if [ "$is_exporting" == "true" ]; then +if [ "$state" == "Running" ]; then exit 1 fi @@ -29,8 +29,8 @@ $FOREST_CLI_PATH snapshot export --format "$format" > snapshot_export.log 2>&1 & echo "Testing that export is in progress" for ((i=1; i<=retries; i++)); do output=$($FOREST_CLI_PATH snapshot export-status --format json) - is_exporting=$(echo "$output" | jq -r '.exporting') - if [ "$is_exporting" == "true" ]; then + state=$(echo "$output" | jq -r '.state') + if [ "$state" == "Running" ]; then break fi if [ $i -eq $retries ]; then @@ -45,9 +45,8 @@ $FOREST_CLI_PATH snapshot export-cancel echo "Testing that export has been cancelled" for ((i=1; i<=retries; i++)); do output=$($FOREST_CLI_PATH snapshot export-status --format json) - is_exporting=$(echo "$output" | jq -r '.exporting') - is_cancelled=$(echo "$output" | jq -r '.cancelled') - if [ "$is_exporting" == "false" ] && [ "$is_cancelled" == "true" ]; then + state=$(echo "$output" | jq -r '.state') + if [ "$state" == "Cancelled" ]; then break fi if [ $i -eq $retries ]; then @@ -66,7 +65,7 @@ EXPORT_CMD_PID=$! sleep 5 # another export job should be disallowed export_error=$($FOREST_CLI_PATH snapshot export 2>&1 || true) -if echo "$export_error" | grep -q "active chain export job has started"; then +if echo "$export_error" | grep -q "export has been running since"; then echo "verified another export job is disallowed" else echo "another export job should be disallowed" @@ -75,7 +74,7 @@ else fi # another export-diff job should be disallowed export_diff_error=$($FOREST_CLI_PATH snapshot export-diff --from 11000 --to 10100 -d 900 2>&1 || true) -if echo "$export_diff_error" | grep -q "active chain export job has started"; then +if echo "$export_diff_error" | grep -q "export has been running since"; then echo "verified another export-diff job is disallowed" else echo "another export-diff job should be disallowed" diff --git a/scripts/tests/calibnet_export_diff_check.sh b/scripts/tests/calibnet_export_diff_check.sh index b1f1b41f9346..63f97adaf018 100755 --- a/scripts/tests/calibnet_export_diff_check.sh +++ b/scripts/tests/calibnet_export_diff_check.sh @@ -22,8 +22,8 @@ $FOREST_CLI_PATH snapshot export-diff --from "$snapshot_epoch" --to "$((snapshot echo "Testing that export is in progress" for ((i=1; i<=retries; i++)); do output=$($FOREST_CLI_PATH snapshot export-status --format json) - is_exporting=$(echo "$output" | jq -r '.exporting') - if [ "$is_exporting" == "true" ]; then + state=$(echo "$output" | jq -r '.state') + if [ "$state" == "Running" ]; then break fi if [ $i -eq $retries ]; then @@ -38,9 +38,8 @@ $FOREST_CLI_PATH snapshot export-cancel echo "Testing that export has been cancelled" for ((i=1; i<=retries; i++)); do output=$($FOREST_CLI_PATH snapshot export-status --format json) - is_exporting=$(echo "$output" | jq -r '.exporting') - is_cancelled=$(echo "$output" | jq -r '.cancelled') - if [ "$is_exporting" == "false" ] && [ "$is_cancelled" == "true" ]; then + state=$(echo "$output" | jq -r '.state') + if [ "$state" == "Cancelled" ]; then break fi if [ $i -eq $retries ]; then @@ -56,7 +55,7 @@ EXPORT_CMD_PID=$! sleep 5 # another export job should be disallowed export_error=$($FOREST_CLI_PATH snapshot export 2>&1 || true) -if echo "$export_error" | grep -q "active chain export job has started"; then +if echo "$export_error" | grep -q "export has been running since"; then echo "verified another export job is disallowed" else echo "another export job should be disallowed" @@ -65,7 +64,7 @@ else fi # another export-diff job should be disallowed export_diff_error=$($FOREST_CLI_PATH snapshot export-diff --from 11000 --to 10100 -d 900 2>&1 || true) -if echo "$export_diff_error" | grep -q "active chain export job has started"; then +if echo "$export_diff_error" | grep -q "export has been running since"; then echo "verified another export-diff job is disallowed" else echo "another export-diff job should be disallowed" diff --git a/src/chain/tests.rs b/src/chain/tests.rs index 5add8a236e35..59dac1e85fdc 100644 --- a/src/chain/tests.rs +++ b/src/chain/tests.rs @@ -163,6 +163,7 @@ async fn test_export_inner( /// pipeline steps must not be able to wait forever on a stalled writer. mod export_stuckness { use super::*; + use crate::ipld::{CHAIN_EXPORT_STATUS, ChainExportGuard, ChainExportKind, ChainExportState}; use crate::shim::crypto::IPLD_RAW; use crate::utils::db::car_stream::CarBlock; use crate::utils::rand::forest_rng; @@ -172,6 +173,72 @@ mod export_stuckness { use std::task::{Context, Poll}; use tokio::io::AsyncWrite; + /// `start_epoch == 0` on a running export renders as a stuck "0.0%" in + /// `forest-cli snapshot export-status` even while bytes are being written. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[serial_test::serial(chain_export)] + async fn export_status_reports_walk_progress_while_running() -> anyhow::Result<()> { + let (db, head) = six_block_chain()?; + + let _guard = ChainExportGuard::try_start_export(ChainExportKind::Snapshot)?; + // The walk runs on the `par_buffer` producer task, so it progresses despite the + // stalled writer. + let export = tokio::spawn({ + let db = db.clone(); + let head = head.clone(); + async move { + let _ = export::( + &db, + &head, + 0, + StallingWriter { + write_budget: 0, + stall_on_flush: true, + }, + ExportOptions::::default(), + ) + .await; + } + }); + + tokio::time::timeout(Duration::from_secs(5), async { + while CHAIN_EXPORT_STATUS.snapshot().initial_epoch == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("a running export must report walk progress, not epoch 0/0"); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().state, + ChainExportState::Running + ); + assert_eq!(CHAIN_EXPORT_STATUS.snapshot().initial_epoch, head.epoch()); + + // The stalled writer keeps the export from ever finishing, so the status + // above was sampled from a live export, and the abort must land. + assert!(!export.is_finished()); + export.abort(); + assert!( + export + .await + .expect_err("aborted export must not run to completion") + .is_cancelled() + ); + Ok(()) + } + + fn six_block_chain() -> anyhow::Result<(Arc, Tipset)> { + let db = Arc::new(MemoryDB::default()); + let c4u = Chain4U::with_blockstore(db.clone()); + chain4u! { + in c4u; + [_genesis_header] + -> [_b1] -> [_b2] -> [_b3] -> [_b4] -> [b5] + }; + let head = Tipset::load_required(&db, &TipsetKey::from(nunny::vec![b5.cid()]))?; + Ok((db, head)) + } + /// Accepts and discards up to `write_budget` bytes, then stalls forever. struct StallingWriter { write_budget: usize, @@ -210,14 +277,7 @@ mod export_stuckness { /// at 100%. #[tokio::test(start_paused = true)] async fn export_stalled_final_flush_errors_instead_of_hanging() -> anyhow::Result<()> { - let db = Arc::new(MemoryDB::default()); - let c4u = Chain4U::with_blockstore(db.clone()); - chain4u! { - in c4u; - [_genesis_header] - -> [_b1] -> [_b2] -> [_b3] -> [_b4] -> [b5] - }; - let head = Tipset::load_required(&db, &TipsetKey::from(nunny::vec![b5.cid()]))?; + let (db, head) = six_block_chain()?; let export = export::( &db, diff --git a/src/cli/main.rs b/src/cli/main.rs index 62c8fc00b0cd..10bcb1c9d77e 100644 --- a/src/cli/main.rs +++ b/src/cli/main.rs @@ -30,5 +30,5 @@ where let _bg_tasks = logger::setup_logger(&crate::cli_shared::cli::CliOpts::default()); - cmd.run(client).await + cmd.run(client).await.map_err(rpc::humanize_rpc_error) } diff --git a/src/cli/subcommands/snapshot_cmd.rs b/src/cli/subcommands/snapshot_cmd.rs index 52bae165982c..497a754b21a0 100644 --- a/src/cli/subcommands/snapshot_cmd.rs +++ b/src/cli/subcommands/snapshot_cmd.rs @@ -5,6 +5,7 @@ use crate::chain::FilecoinSnapshotVersion; use crate::chain_sync::chain_muxer::DEFAULT_RECENT_STATE_ROOTS; use crate::cli_shared::snapshot::{self, TrustedVendor}; use crate::db::car::forest::tmp_exporting_forest_car_path; +use crate::ipld::ChainExportState; use crate::networks::calibnet; use crate::prelude::*; use crate::rpc::chain::ForestChainExportDiffParams; @@ -202,66 +203,60 @@ impl SnapshotCommands { ForestChainExportStatus::request(())?.with_timeout(Duration::from_secs(30)), ) .await?; - if !result.exporting - && let Format::Text = format - { - if result.cancelled { - println!("No export in progress (last export was cancelled)"); - } else { - println!("No export in progress"); + // A terminal state from a past export is reported as-is, never as the + // watched export's outcome. + if !wait || result.state != ChainExportState::Running { + match format { + Format::Text => println!("{result}"), + Format::Json => println!("{}", serde_json::to_string_pretty(&result)?), } return Ok(()); } - if wait { - let elapsed = chrono::Utc::now() - .signed_duration_since(result.start_time.unwrap_or_default()) - .to_std() - .unwrap_or(Duration::ZERO); - let pb = ProgressBar::new(10000) - .with_elapsed(elapsed) - .with_message("Exporting"); - pb.set_style( - ProgressStyle::with_template( - "[{elapsed_precise}] [{wide_bar}] {percent}% {msg} ", + let watched_start_time = result.start_time; + let elapsed = chrono::Utc::now() + .signed_duration_since(watched_start_time.unwrap_or_default()) + .to_std() + .unwrap_or(Duration::ZERO); + let pb = ProgressBar::new(10000) + .with_elapsed(elapsed) + .with_message("Exporting"); + pb.set_style( + ProgressStyle::with_template( + "[{elapsed_precise}] [{wide_bar}] {percent}% {msg} ", + ) + .expect("indicatif template must be valid") + .progress_chars("#>-"), + ); + let last = loop { + let result = client + .call( + ForestChainExportStatus::request(())? + .with_timeout(Duration::from_secs(30)), ) - .expect("indicatif template must be valid") - .progress_chars("#>-"), - ); - loop { - let result = client - .call( - ForestChainExportStatus::request(())? - .with_timeout(Duration::from_secs(30)), - ) - .await?; - if result.cancelled { - pb.set_message("Export cancelled"); - pb.abandon(); - return Ok(()); - } - let position = (result.progress.clamp(0.0, 1.0) * 10000.0).trunc() as u64; - pb.set_position(position); - - if !result.exporting { - break; - } - tokio::time::sleep(Duration::from_millis(500)).await; + .await?; + if result.start_time != watched_start_time { + // The export slot was reused: the watched export is over, but its + // outcome has been overwritten by the newer export's. + pb.abandon_with_message("Export ended; another export has taken its place"); + return Ok(()); } + let position = (result.progress.clamp(0.0, 1.0) * 10000.0).trunc() as u64; + pb.set_position(position); - pb.finish_with_message(if result.succeeded { - "Export completed" - } else { - "Export failed" - }); - - return Ok(()); - } - match format { - Format::Text => { - println!("Exporting: {:.1}%", result.progress.clamp(0.0, 1.0) * 100.0); + if result.state != ChainExportState::Running { + break result; } - Format::Json => { - println!("{}", serde_json::to_string_pretty(&result)?); + tokio::time::sleep(Duration::from_millis(500)).await; + }; + match last.state { + ChainExportState::Succeeded => pb.finish_with_message("Export completed"), + ChainExportState::Cancelled => pb.abandon_with_message("Export cancelled"), + _ => { + pb.abandon_with_message("Export failed"); + anyhow::bail!( + "export failed: {}", + last.error.as_deref().unwrap_or("unknown error") + ); } } diff --git a/src/db/gc/snapshot.rs b/src/db/gc/snapshot.rs index 8319f814b4a0..1654dcbba94c 100644 --- a/src/db/gc/snapshot.rs +++ b/src/db/gc/snapshot.rs @@ -49,7 +49,7 @@ use crate::db::{ parity_db::GarbageCollectableDb, }; use crate::interpreter::VMTrace; -use crate::ipld::ChainExportGuard; +use crate::ipld::{ChainExportGuard, ChainExportKind}; use crate::prelude::*; use crate::shim::clock::EPOCHS_IN_DAY; use crate::utils::io::EitherMmapOrRandomAccessFile; @@ -76,11 +76,13 @@ pub struct SnapshotGarbageCollector { memory_db: RwLock>>, memory_db_head_key: RwLock>, exported_head_key: RwLock>, - trigger_tx: flume::Sender<()>, - trigger_rx: flume::Receiver<()>, - progress_tx: RwLock>>, + trigger_tx: flume::Sender>, + trigger_rx: flume::Receiver>, } +/// Delivers the outcome of a GC run to a blocking `Forest.SnapshotGC` caller. +type GcOutcomeSender = flume::Sender>; + impl SnapshotGarbageCollector { pub fn new(chain_follower: ChainFollower, config: &crate::Config) -> anyhow::Result { let chain_data_path = chain_path(config); @@ -124,13 +126,12 @@ impl SnapshotGarbageCollector { exported_head_key: RwLock::new(None), trigger_tx, trigger_rx, - progress_tx: RwLock::new(None), }) } pub async fn event_loop(&self) { - while self.trigger_rx.recv_async().await.is_ok() { - self.gc_once().await; + while let Ok(outcome_tx) = self.trigger_rx.recv_async().await { + self.gc_once(outcome_tx).await; } } @@ -170,7 +171,7 @@ impl SnapshotGarbageCollector { && sync_status.is_synced() // chain is in sync && sync_status.active_forks.is_empty() // no active fork && head_epoch - car_db_head_epoch >= snap_gc_interval_epochs // the gap between chain head and car_db head is above threshold - && self.trigger_tx.try_send(()).is_ok() + && self.trigger_tx.try_send(None).is_ok() { tracing::info!(%car_db_head_epoch, %head_epoch, %network_head_epoch, %snap_gc_interval_epochs, "Snap GC scheduled"); } else { @@ -181,44 +182,52 @@ impl SnapshotGarbageCollector { } } - pub fn trigger(&self) -> anyhow::Result> { + pub fn trigger(&self) -> anyhow::Result>> { if self.running.load(Ordering::Relaxed) { anyhow::bail!("snap gc has already been running"); } - if self.trigger_tx.try_send(()).is_err() { + // Travels with the trigger, so another run cannot consume it. + let (outcome_tx, outcome_rx) = flume::bounded(1); + if self.trigger_tx.try_send(Some(outcome_tx)).is_err() { anyhow::bail!("snap gc has already been triggered"); } - - let (progress_tx, progress_rx) = flume::unbounded(); - *self.progress_tx.write() = Some(progress_tx); - Ok(progress_rx) + Ok(outcome_rx) } - async fn gc_once(&self) { + async fn gc_once(&self, outcome_tx: Option) { if self.running.swap(true, Ordering::Relaxed) { tracing::warn!("snap gc has already been running"); return; } - match self.export_snapshot().await { - Ok(_) => { - if let Err(e) = self.cleanup_after_snapshot_export().await { - tracing::warn!("{e:#}"); - } - } + let result = match self.export_snapshot().await { + Ok(()) => self.cleanup_after_snapshot_export().await, Err(e) => { - tracing::error!("{e:#}"); // Unsubscribe on failure path self.db().unsubscribe_write_ops(); + Err(e) } + }; + if let Err(e) = &result { + tracing::error!("{e:#}"); + } + // Dropping the sender also unblocks the caller. + if let Some(outcome_tx) = outcome_tx { + let _ = outcome_tx.send(result); } - // To indicate the completion of GC - drop(self.progress_tx.write().take()); self.running.store(false, Ordering::Relaxed); } async fn export_snapshot(&self) -> anyhow::Result<()> { - let chain_export_guard = ChainExportGuard::try_start_export()?; + let chain_export_guard = ChainExportGuard::try_start_export(ChainExportKind::SnapshotGc)?; + let result = self.export_snapshot_inner(&chain_export_guard).await; + chain_export_guard.finish(result) + } + + async fn export_snapshot_inner( + &self, + chain_export_guard: &ChainExportGuard, + ) -> anyhow::Result<()> { let db = self.db(); tracing::info!( "exporting lite snapshot with {} recent state roots", @@ -314,7 +323,6 @@ impl SnapshotGarbageCollector { _ => {} } joinset.shutdown().await; - chain_export_guard.mark_as_succeeded(); Ok(()) } @@ -440,3 +448,85 @@ impl SnapshotGarbageCollector { &self.chain_follower.sync_status } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::blocks::{CachingBlockHeader, RawBlockHeader}; + use crate::chain_sync::network_context::SyncNetworkContext; + use crate::db::MemoryDB; + use crate::ipld::{ChainExportGuard, ChainExportKind}; + use crate::libp2p::PeerManager; + use crate::message_pool::MessagePool; + use crate::networks::ChainConfig; + use crate::shim::address::Address; + use crate::state_manager::StateManager; + use tokio::task::JoinSet; + + fn test_gc( + data_dir: &std::path::Path, + ) -> (SnapshotGarbageCollector, JoinSet>) { + let (network_send, _network_rx) = flume::bounded(5); + let (_net_event_tx, net_event_rx) = flume::bounded(5); + let mut services = JoinSet::new(); + let db = std::sync::Arc::new(MemoryDB::default()); + let chain_config = std::sync::Arc::new(ChainConfig::default()); + let genesis_header = CachingBlockHeader::new(RawBlockHeader { + miner_address: Address::new_id(0), + timestamp: 7777, + ..Default::default() + }); + let cs = ChainStore::new(db, chain_config, genesis_header.clone()).unwrap(); + let state_manager = StateManager::new(cs.shallow_clone()).unwrap(); + let mpool = MessagePool::new( + cs, + network_send.clone(), + Default::default(), + state_manager.chain_config().clone(), + &mut services, + ) + .unwrap(); + let genesis_ts = Tipset::from(genesis_header); + let peer_manager = std::sync::Arc::new(PeerManager::default()); + let network = SyncNetworkContext::new(network_send, peer_manager, state_manager.db_owned()); + let chain_follower = ChainFollower::new( + state_manager, + network, + genesis_ts, + net_event_rx, + false, + mpool, + ); + let mut config = crate::Config::default(); + config.client.data_dir = data_dir.to_path_buf(); + let gc = SnapshotGarbageCollector::new(chain_follower, &config).unwrap(); + (gc, services) + } + + /// `forest-cli chain prune snap` must not exit 0 with the GC error only in the + /// daemon logs. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[serial_test::serial(chain_export)] + async fn manual_gc_trigger_propagates_failure() { + let tmp = tempfile::TempDir::new().unwrap(); + let (gc, _services) = test_gc(tmp.path()); + let gc = std::sync::Arc::new(gc); + tokio::spawn({ + let gc = gc.clone(); + async move { gc.event_loop().await } + }); + + // Hold the export slot so the GC export cannot start. + let _guard = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap(); + + let progress_rx = gc.trigger().unwrap(); + let outcome = tokio::time::timeout(Duration::from_secs(10), progress_rx.recv_async()) + .await + .expect("GC must complete") + .expect("a blocking GC caller must receive the outcome"); + assert!( + outcome.is_err(), + "GC that could not start must report failure to the caller" + ); + } +} diff --git a/src/ipld/export_status.rs b/src/ipld/export_status.rs new file mode 100644 index 000000000000..ecc580280471 --- /dev/null +++ b/src/ipld/export_status.rs @@ -0,0 +1,384 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +//! Status and life cycle of the chain-export slot shared by user-requested snapshot +//! exports and the automatic snapshot GC. + +use crate::shim::clock::ChainEpoch; +use chrono::{DateTime, Utc}; +use std::sync::Arc; +use std::sync::LazyLock; +use std::sync::atomic::{self, AtomicI64}; +use tokio_util::sync::CancellationToken; + +/// What kind of export is (or was last) holding the chain-export slot. +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + serde::Serialize, + serde::Deserialize, + schemars::JsonSchema, + strum::Display, +)] +pub enum ChainExportKind { + /// A snapshot export requested via `Forest.ChainExport`. + Snapshot, + /// A diff snapshot export requested via `Forest.ChainExportDiff`. + DiffSnapshot, + /// A lite snapshot export performed by the automatic snapshot GC. + SnapshotGc, +} + +/// Transitions only through [`ChainExportGuard`]: `Running` while a guard is held, then +/// exactly one terminal state once it drops. `Idle`: no export has run since node start. +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + serde::Serialize, + serde::Deserialize, + schemars::JsonSchema, + strum::Display, +)] +pub enum ChainExportState { + Idle, + Running, + Succeeded, + Cancelled, + Failed, +} + +/// Cold state behind one mutex for consistent reads; only the per-block epoch counters +/// are hot and lock-free. +#[derive(Default)] +struct StatusInner { + /// `None` while running and before the first export. + outcome: Option, + kind: Option, + start_time: Option>, + error: Option, + cancellation_token: Option, + /// See [`ProgressReporter`]. + counters: Arc, +} + +impl StatusInner { + /// A live cancellation token exists exactly while a [`ChainExportGuard`] is held. + fn is_running(&self) -> bool { + self.cancellation_token.is_some() + } +} + +#[derive(Default)] +struct ProgressCounters { + epoch: AtomicI64, + initial_epoch: AtomicI64, +} + +#[derive(Default)] +pub struct ExportStatus { + inner: parking_lot::Mutex, +} + +/// Read under one lock, so fields are mutually consistent. +pub struct StatusSnapshot { + pub state: ChainExportState, + pub kind: Option, + pub error: Option, + pub start_time: Option>, + pub epoch: ChainEpoch, + pub initial_epoch: ChainEpoch, +} + +pub fn kind_label(kind: Option) -> String { + kind.map(|k| k.to_string()) + .unwrap_or_else(|| "unknown".into()) +} + +pub fn format_start_time(start_time: Option>) -> String { + start_time + .map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)) + .unwrap_or_else(|| "unknown".into()) +} + +impl ExportStatus { + pub fn snapshot(&self) -> StatusSnapshot { + let inner = self.inner.lock(); + StatusSnapshot { + state: if inner.is_running() { + ChainExportState::Running + } else { + inner.outcome.unwrap_or(ChainExportState::Idle) + }, + kind: inner.kind, + error: inner.error.clone(), + start_time: inner.start_time, + epoch: inner.counters.epoch.load(atomic::Ordering::Relaxed), + initial_epoch: inner.counters.initial_epoch.load(atomic::Ordering::Relaxed), + } + } + + /// Check-and-cancel under one lock, so the cancel cannot land on a different export + /// than the one observed. + pub fn cancel_running(&self) -> bool { + if let Some(token) = &self.inner.lock().cancellation_token { + token.cancel(); + true + } else { + false + } + } + + /// Holding the mutex makes check-and-start atomic: the lock is the export slot. + fn try_begin( + &self, + kind: ChainExportKind, + cancellation_token: CancellationToken, + ) -> anyhow::Result<()> { + let mut inner = self.inner.lock(); + anyhow::ensure!( + !inner.is_running(), + "an active {} export has been running since {}; check `forest-cli snapshot export-status`", + kind_label(inner.kind), + format_start_time(inner.start_time), + ); + *inner = StatusInner { + outcome: None, + kind: Some(kind), + start_time: Some(Utc::now()), + error: None, + cancellation_token: Some(cancellation_token), + counters: Arc::new(ProgressCounters::default()), + }; + Ok(()) + } + + /// The first terminal outcome recorded for an export wins; later ones are ignored. + fn record_outcome(&self, outcome: ChainExportState, error: Option) { + let mut inner = self.inner.lock(); + if inner.outcome.is_none() { + inner.outcome = Some(outcome); + inner.error = error; + } + } + + fn end(&self) { + let mut inner = self.inner.lock(); + // Ended without a recorded outcome (panic, or a skipped `finish`): failed. + if inner.outcome.is_none() { + inner.outcome = Some(ChainExportState::Failed); + if !std::thread::panicking() { + tracing::warn!("chain export guard dropped without a recorded outcome"); + } + } + inner.cancellation_token = None; + } + + pub(super) fn progress_reporter(&self) -> ProgressReporter { + ProgressReporter(self.inner.lock().counters.clone()) + } +} + +/// Bound at creation to its export's freshly allocated counters: a producer task that +/// outlives its export (a Tokio abort lands only at the next await) keeps writing into +/// its own orphaned counters, never the next export's. +#[derive(Clone)] +pub struct ProgressReporter(Arc); + +impl ProgressReporter { + pub fn update_epoch(&self, epoch: ChainEpoch) { + self.0.epoch.store(epoch, atomic::Ordering::Relaxed); + _ = self.0.initial_epoch.compare_exchange( + 0, + epoch, + atomic::Ordering::Relaxed, + atomic::Ordering::Relaxed, + ); + } +} + +pub static CHAIN_EXPORT_STATUS: LazyLock = LazyLock::new(ExportStatus::default); + +#[derive(Debug)] +pub struct ChainExportGuard { + cancellation_token: CancellationToken, +} + +impl ChainExportGuard { + pub fn try_start_export(kind: ChainExportKind) -> anyhow::Result { + let cancellation_token = CancellationToken::new(); + CHAIN_EXPORT_STATUS.try_begin(kind, cancellation_token.clone())?; + Ok(Self { cancellation_token }) + } + + /// Every export path that holds a [`ChainExportGuard`] must await its work through + /// this method — an export that does not race against the cancellation token cannot + /// be cancelled and appears stuck until process restart. + pub async fn run_cancellable(&self, fut: F) -> Option { + let output = self.cancellation_token.run_until_cancelled(fut).await; + if output.is_none() { + CHAIN_EXPORT_STATUS.record_outcome(ChainExportState::Cancelled, None); + } + output + } + + /// A cancellation observed by [`Self::run_cancellable`] wins over `result`, so + /// callers need no cancellation special-casing. + pub fn finish(self, result: anyhow::Result) -> anyhow::Result { + match &result { + Ok(_) => CHAIN_EXPORT_STATUS.record_outcome(ChainExportState::Succeeded, None), + Err(e) => { + CHAIN_EXPORT_STATUS.record_outcome(ChainExportState::Failed, Some(format!("{e:#}"))) + } + } + result + } +} + +impl Drop for ChainExportGuard { + fn drop(&mut self) { + // In case some tasks are waiting on this token + self.cancellation_token.cancel(); + CHAIN_EXPORT_STATUS.end(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Pins the invariant documented on [`ChainExportGuard::run_cancellable`]. + #[tokio::test] + #[serial_test::serial(chain_export)] + async fn chain_export_cancel_stops_guarded_export() { + let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap(); + + let fut = g.run_cancellable(std::future::pending::<()>()); + // Cancel exactly as the `Forest.ChainExportCancel` handler does. + assert!(CHAIN_EXPORT_STATUS.cancel_running()); + assert!( + fut.await.is_none(), + "cancellation must interrupt the export" + ); + + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().state, + ChainExportState::Running + ); + drop(g); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().state, + ChainExportState::Cancelled + ); + } + + #[test] + #[serial_test::serial(chain_export)] + fn chain_export_status_reports_kind() { + let g = ChainExportGuard::try_start_export(ChainExportKind::SnapshotGc).unwrap(); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().kind, + Some(ChainExportKind::SnapshotGc) + ); + + // Rejecting a concurrent export must say what kind of export is in the way. + let err = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap_err(); + assert!( + err.to_string().contains("SnapshotGc"), + "unexpected error: {err}" + ); + + // The kind outlives the export; the next export replaces it. + drop(g); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().kind, + Some(ChainExportKind::SnapshotGc) + ); + let _g = ChainExportGuard::try_start_export(ChainExportKind::DiffSnapshot).unwrap(); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().kind, + Some(ChainExportKind::DiffSnapshot) + ); + } + + #[test] + fn chain_export_state_starts_idle() { + assert_eq!( + ExportStatus::default().snapshot().state, + ChainExportState::Idle + ); + } + + /// Pins the transitions documented on [`ChainExportState`]. + #[tokio::test] + #[serial_test::serial(chain_export)] + async fn chain_export_state_machine() { + let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap(); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().state, + ChainExportState::Running + ); + g.finish(anyhow::Ok(())).unwrap(); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().state, + ChainExportState::Succeeded + ); + + // Failure. + let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap(); + g.finish(anyhow::Result::<()>::Err(anyhow::anyhow!( + "missing state root" + ))) + .unwrap_err(); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().state, + ChainExportState::Failed + ); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().error.as_deref(), + Some("missing state root") + ); + + // A guard dropped without `finish` still lands in `Failed`; the previous + // failure's error does not leak into the new export. + let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap(); + drop(g); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().state, + ChainExportState::Failed + ); + assert_eq!(CHAIN_EXPORT_STATUS.snapshot().error, None); + + // A cancelled export whose body then bails: no error recorded. + let g = ChainExportGuard::try_start_export(ChainExportKind::SnapshotGc).unwrap(); + assert!(CHAIN_EXPORT_STATUS.cancel_running()); + assert!( + g.run_cancellable(std::future::pending::<()>()) + .await + .is_none() + ); + g.finish(anyhow::Result::<()>::Err(anyhow::anyhow!( + "snapshot GC export was cancelled" + ))) + .unwrap_err(); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().state, + ChainExportState::Cancelled + ); + assert_eq!(CHAIN_EXPORT_STATUS.snapshot().error, None); + + // A cancel after completion must not flip the terminal state. + let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap(); + g.finish(anyhow::Ok(())).unwrap(); + assert!(!CHAIN_EXPORT_STATUS.cancel_running()); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().state, + ChainExportState::Succeeded + ); + } +} diff --git a/src/ipld/mod.rs b/src/ipld/mod.rs index 9db6886866e5..b82632c9ba56 100644 --- a/src/ipld/mod.rs +++ b/src/ipld/mod.rs @@ -1,9 +1,11 @@ // Copyright 2019-2026 ChainSafe Systems // SPDX-License-Identifier: Apache-2.0, MIT +pub mod export_status; pub mod selector; pub mod util; +pub use export_status::*; pub use ipld_core::ipld::Ipld; pub use util::*; diff --git a/src/ipld/util.rs b/src/ipld/util.rs index 920ad9ab7a3a..614478c3b2e4 100644 --- a/src/ipld/util.rs +++ b/src/ipld/util.rs @@ -4,161 +4,20 @@ use crate::blocks::Tipset; use crate::cid_collections::{CidHashSet, CidHashSetLike}; use crate::ipld::Ipld; +use crate::ipld::export_status::{CHAIN_EXPORT_STATUS, ProgressReporter}; use crate::prelude::*; use crate::shim::clock::ChainEpoch; use crate::shim::executor::Receipt; use crate::utils::db::car_stream::CarBlock; use crate::utils::encoding::extract_cids; use crate::utils::multihash::prelude::*; -use arc_swap::ArcSwapOption; use bytes::Bytes; -use chrono::{DateTime, Utc}; use futures::Stream; use pin_project_lite::pin_project; use std::borrow::Borrow; use std::collections::VecDeque; use std::pin::Pin; -use std::sync::atomic::{AtomicBool, AtomicI64}; -use std::sync::{LazyLock, atomic}; use std::task::{Context, Poll}; -use tokio_util::sync::CancellationToken; - -#[derive(Default)] -pub struct ExportStatus { - pub epoch: AtomicI64, - pub initial_epoch: AtomicI64, - pub exporting: AtomicBool, - pub cancelled: AtomicBool, - pub succeeded: AtomicBool, - pub start_time: ArcSwapOption>, - pub cancellation_token: ArcSwapOption, -} - -impl ExportStatus { - pub fn epoch(&self) -> i64 { - self.epoch.load(atomic::Ordering::Relaxed) - } - - pub fn initial_epoch(&self) -> i64 { - self.initial_epoch.load(atomic::Ordering::Relaxed) - } - - pub fn exporting(&self) -> bool { - self.exporting.load(atomic::Ordering::Relaxed) - } - - pub fn cancelled(&self) -> bool { - self.cancelled.load(atomic::Ordering::Relaxed) - } - - pub fn succeeded(&self) -> bool { - self.succeeded.load(atomic::Ordering::Relaxed) - } - - pub fn start_time(&self) -> Option> { - self.start_time.load().clone().map(Arc::unwrap_or_clone) - } - - pub fn cancellation_token(&self) -> Option { - self.cancellation_token - .load() - .clone() - .map(Arc::unwrap_or_clone) - } -} - -pub static CHAIN_EXPORT_STATUS: LazyLock = LazyLock::new(ExportStatus::default); - -#[derive(Debug)] -pub struct ChainExportGuard { - cancellation_token: CancellationToken, -} - -impl ChainExportGuard { - pub fn try_start_export() -> anyhow::Result { - let cancellation_token = CancellationToken::new(); - start_export(cancellation_token.clone())?; - Ok(Self { cancellation_token }) - } - - pub fn cancel_export(&self) { - cancel_export() - } - - pub fn mark_as_succeeded(&self) { - export_succeeded() - } - - /// Every export path that holds a [`ChainExportGuard`] must await its work through - /// this method — an export that does not race against the cancellation token cannot - /// be cancelled and appears stuck until process restart. On cancellation, marks - /// the export status as cancelled and returns `None`. - pub async fn run_cancellable(&self, fut: F) -> Option { - let output = self.cancellation_token.run_until_cancelled(fut).await; - if output.is_none() { - self.cancel_export(); - } - output - } -} - -impl Drop for ChainExportGuard { - fn drop(&mut self) { - // In case some tasks are waiting on this token - self.cancellation_token.cancel(); - end_export() - } -} - -fn update_epoch(new_value: i64) { - let status = &*CHAIN_EXPORT_STATUS; - status.epoch.store(new_value, atomic::Ordering::Relaxed); - _ = status.initial_epoch.compare_exchange( - 0, - new_value, - atomic::Ordering::Relaxed, - atomic::Ordering::Relaxed, - ); -} - -fn start_export(cancellation_token: CancellationToken) -> anyhow::Result<()> { - let status = &*CHAIN_EXPORT_STATUS; - let export_in_progress = status.exporting.swap(true, atomic::Ordering::Relaxed); - anyhow::ensure!( - !export_in_progress, - "An active chain export job has started at {}, start epoch: {}, current epoch: {}", - status.start_time().unwrap_or_default(), - status.initial_epoch(), - status.epoch(), - ); - status.epoch.store(0, atomic::Ordering::Relaxed); - status.initial_epoch.store(0, atomic::Ordering::Relaxed); - status.cancelled.store(false, atomic::Ordering::Relaxed); - status.succeeded.store(false, atomic::Ordering::Relaxed); - status.start_time.store(Some(Utc::now().into())); - status - .cancellation_token - .store(Some(cancellation_token.into())); - Ok(()) -} - -fn export_succeeded() { - CHAIN_EXPORT_STATUS - .succeeded - .store(true, atomic::Ordering::Relaxed); -} - -fn end_export() { - CHAIN_EXPORT_STATUS - .exporting - .store(false, atomic::Ordering::Relaxed); - CHAIN_EXPORT_STATUS.cancellation_token.store(None); -} - -fn cancel_export() { - let status = &*CHAIN_EXPORT_STATUS; - status.cancelled.store(true, atomic::Ordering::Relaxed); -} fn should_save_block_to_snapshot(cid: Cid) -> bool { // Don't include identity CIDs. @@ -254,7 +113,7 @@ pin_project! { message_receipts: bool, events: bool, tipset_keys:bool, - track_progress: bool, + progress: Option, n_polled: usize, } } @@ -266,7 +125,7 @@ impl ChainStream { } pub fn track_progress(mut self, track_progress: bool) -> Self { - self.track_progress = track_progress; + self.progress = track_progress.then(|| CHAIN_EXPORT_STATUS.progress_reporter()); self } @@ -326,7 +185,7 @@ pub fn stream_chain< message_receipts: false, events: false, tipset_keys: false, - track_progress: false, + progress: None, n_polled: 0, } } @@ -389,8 +248,8 @@ impl, ITER: Iterator + Unpin, S: Cid } } Iterate(epoch, block_cid, _type, cid_vec) => { - if *this.track_progress { - update_epoch(*epoch); + if let Some(progress) = this.progress { + progress.update_epoch(*epoch); } while let Some(cid) = cid_vec.pop() { // The link traversal implementation assumes there are three types of encoding: @@ -456,8 +315,8 @@ impl, ITER: Iterator + Unpin, S: Cid for block in tipset.borrow().block_headers() { let (cid, data) = block.car_block()?; if this.seen.insert(cid)? { - if *this.track_progress { - update_epoch(block.epoch); + if let Some(progress) = this.progress { + progress.update_epoch(block.epoch); } // Make sure we always yield a block otherwise. this.dfs.push_back(Emit(cid, Some(data.into()))); @@ -657,85 +516,4 @@ mod tests { Ok(()) } - - /// Pins the invariant documented on [`ChainExportGuard::run_cancellable`]. - #[tokio::test] - #[serial_test::serial(chain_export)] - async fn chain_export_cancel_stops_guarded_export() { - let g = ChainExportGuard::try_start_export().unwrap(); - - // Cancel via the status-global token, as the `Forest.ChainExportCancel` handler does. - let token = CHAIN_EXPORT_STATUS.cancellation_token().unwrap(); - - let fut = g.run_cancellable(std::future::pending::<()>()); - token.cancel(); - assert!( - fut.await.is_none(), - "cancellation must interrupt the export" - ); - - assert!(CHAIN_EXPORT_STATUS.cancelled()); - assert!(CHAIN_EXPORT_STATUS.exporting()); - - drop(g); - assert!(!CHAIN_EXPORT_STATUS.exporting()); - assert!(CHAIN_EXPORT_STATUS.cancelled()); - } - - #[test] - #[serial_test::serial(chain_export)] - fn test_chain_export_guard() { - // First export (Cancel) - let g = ChainExportGuard::try_start_export().unwrap(); - assert!(CHAIN_EXPORT_STATUS.exporting()); - assert!(!CHAIN_EXPORT_STATUS.succeeded()); - assert!(!CHAIN_EXPORT_STATUS.cancelled()); - - // Another attempt should fail - ChainExportGuard::try_start_export().unwrap_err(); - - // Cancel - g.cancel_export(); - assert!(CHAIN_EXPORT_STATUS.cancelled()); - - // Drop - drop(g); - assert!(!CHAIN_EXPORT_STATUS.exporting()); - assert!(!CHAIN_EXPORT_STATUS.succeeded()); - assert!(CHAIN_EXPORT_STATUS.cancelled()); - - // Second export (Success) - let g = ChainExportGuard::try_start_export().unwrap(); - assert!(CHAIN_EXPORT_STATUS.exporting()); - assert!(!CHAIN_EXPORT_STATUS.succeeded()); - assert!(!CHAIN_EXPORT_STATUS.cancelled()); - - // Another attempt should fail - ChainExportGuard::try_start_export().unwrap_err(); - - // On success - g.mark_as_succeeded(); - assert!(CHAIN_EXPORT_STATUS.succeeded()); - - // Drop - drop(g); - assert!(!CHAIN_EXPORT_STATUS.exporting()); - assert!(CHAIN_EXPORT_STATUS.succeeded()); - assert!(!CHAIN_EXPORT_STATUS.cancelled()); - - // Third export (failure) - let g = ChainExportGuard::try_start_export().unwrap(); - assert!(CHAIN_EXPORT_STATUS.exporting()); - assert!(!CHAIN_EXPORT_STATUS.succeeded()); - assert!(!CHAIN_EXPORT_STATUS.cancelled()); - - // Another attempt should fail - ChainExportGuard::try_start_export().unwrap_err(); - - // Drop - drop(g); - assert!(!CHAIN_EXPORT_STATUS.exporting()); - assert!(!CHAIN_EXPORT_STATUS.succeeded()); - assert!(!CHAIN_EXPORT_STATUS.cancelled()); - } } diff --git a/src/rpc/client.rs b/src/rpc/client.rs index 4e4488d3b0aa..15aec0e1ba92 100644 --- a/src/rpc/client.rs +++ b/src/rpc/client.rs @@ -331,10 +331,56 @@ impl jsonrpsee::core::client::SubscriptionClientT for UrlClient { } } +/// Rewrites a JSON-RPC call error anywhere in the chain to its server-sent message, +/// preserving any context layered on top. +pub fn humanize_rpc_error(e: anyhow::Error) -> anyhow::Error { + match e.downcast_ref::() { + Some(ClientError::Call(obj)) => { + let contexts: Vec = e + .chain() + .take_while(|cause| cause.downcast_ref::().is_none()) + .map(ToString::to_string) + .collect(); + let mut out = match obj.data() { + Some(data) => anyhow::anyhow!("{} (data: {data})", obj.message()), + None => anyhow::anyhow!("{}", obj.message()), + }; + for context in contexts.into_iter().rev() { + out = out.context(context); + } + out + } + _ => e, + } +} + #[cfg(test)] mod tests { use super::*; use crate::cli_shared::FOREST_DATA_DIR_ENV; + use jsonrpsee::types::ErrorObject; + + fn call_error() -> ClientError { + ClientError::Call(ErrorObject::owned(1, "export already running", None::<()>)) + } + + #[test] + fn rpc_call_errors_render_as_their_message() { + assert_eq!( + format!("{:#}", humanize_rpc_error(call_error().into())), + "export already running" + ); + + let wrapped = anyhow::Error::from(call_error()).context("failed to check F3 sync status"); + assert_eq!( + format!("{:#}", humanize_rpc_error(wrapped)), + "failed to check F3 sync status: export already running" + ); + + // Non-RPC errors pass through untouched, chain included. + let other = anyhow::anyhow!("inner").context("outer"); + assert_eq!(format!("{:#}", humanize_rpc_error(other)), "outer: inner"); + } // The RPC client should pick up the admin token from the data directory // pointed to by `FOREST_PATH`, mirroring where a daemon started with the diff --git a/src/rpc/methods/chain.rs b/src/rpc/methods/chain.rs index 8bd5f9633767..1c24b57360ba 100644 --- a/src/rpc/methods/chain.rs +++ b/src/rpc/methods/chain.rs @@ -16,7 +16,7 @@ use crate::db::car::forest::{ tmp_exporting_forest_car_path, }; use crate::ipld::DfsIter; -use crate::ipld::{CHAIN_EXPORT_STATUS, ChainExportGuard}; +use crate::ipld::{CHAIN_EXPORT_STATUS, ChainExportGuard, ChainExportKind}; use crate::lotus_json::{HasLotusJson, LotusJson, lotus_json_with_self}; #[cfg(test)] use crate::lotus_json::{assert_all_snapshots, assert_unchanged_via_json}; @@ -266,7 +266,11 @@ impl RpcMethod<1> for ChainPruneSnapshot { ) -> Result { if let Some(gc) = crate::daemon::GLOBAL_SNAPSHOT_GC.get() { let progress_rx = gc.trigger()?; - while blocking && progress_rx.recv_async().await.is_ok() {} + if blocking { + progress_rx.recv_async().await.map_err(|_| { + anyhow::anyhow!("snapshot GC ended without reporting an outcome") + })??; + } Ok(()) } else { Err(anyhow::anyhow!("snapshot gc is not enabled").into()) @@ -290,191 +294,152 @@ impl RpcMethod<1> for ForestChainExport { (params,): Self::Params, _: &http::Extensions, ) -> Result { - fn save_checksum( - checksum: digest::Output, - snapshot_output_path: &Path, - ) -> anyhow::Result<()> { - let path = forest_car_sha256sum_path(snapshot_output_path); - std::fs::write( - path, - format!( - "{} {}\n", - checksum.encode_hex::(), - snapshot_output_path - .file_name() - .and_then(std::ffi::OsStr::to_str) - .context("Failed to retrieve file name while saving checksum")? - ), - )?; - Ok(()) - } - // Spawn a task so it's not cancelled when CLI client is disconnected. // So do not wrap this with `AbortOnDropHandle` let handle = tokio::spawn(async move { - let ForestChainExportParams { - version, - epoch, - recent_roots, - output_path, - tipset_keys: ApiTipsetKey(tsk), - include_receipts, - include_events, - include_tipset_keys, - augmented_snapshot, - tipset_lookup, - skip_checksum, - dry_run, - } = params; - - let chain_export_guard = ChainExportGuard::try_start_export()?; + let chain_export_guard = ChainExportGuard::try_start_export(ChainExportKind::Snapshot)?; + let result = export_chain_inner(&ctx, params, &chain_export_guard).await; + chain_export_guard.finish(result) + }); + Ok(handle.await??) + } +} - let head = ctx.chain_store().load_required_tipset_or_heaviest(&tsk)?; - let start_ts = ctx - .chain_index() - .load_required_tipset_by_height(epoch, head, ResolveNullTipset::TakeOlder) - .await?; +fn save_checksum( + checksum: digest::Output, + snapshot_output_path: &Path, +) -> anyhow::Result<()> { + let path = forest_car_sha256sum_path(snapshot_output_path); + std::fs::write( + path, + format!( + "{} {}\n", + checksum.encode_hex::(), + snapshot_output_path + .file_name() + .and_then(std::ffi::OsStr::to_str) + .context("Failed to retrieve file name while saving checksum")? + ), + )?; + Ok(()) +} - let options = ExportOptions { - skip_checksum, - include_receipts, - include_events, - include_tipset_keys, - include_tipset_lookup: tipset_lookup, - seen: FileBackedCidHashSet::new(ctx.temp_dir.as_path())?, - }; - let tmp_path = - tempfile::TempPath::try_from_path(tmp_exporting_forest_car_path(&output_path))?; - let writer = if dry_run { - tokio_util::either::Either::Left(VoidAsyncWriter) - } else { - tokio_util::either::Either::Right(tokio::fs::File::create(&tmp_path).await?) +async fn export_chain_inner( + ctx: &Ctx, + params: ForestChainExportParams, + chain_export_guard: &ChainExportGuard, +) -> anyhow::Result { + let ForestChainExportParams { + version, + epoch, + recent_roots, + output_path, + tipset_keys: ApiTipsetKey(tsk), + include_receipts, + include_events, + include_tipset_keys, + augmented_snapshot, + tipset_lookup, + skip_checksum, + dry_run, + } = params; + + let head = ctx.chain_store().load_required_tipset_or_heaviest(&tsk)?; + let start_ts = ctx + .chain_index() + .load_required_tipset_by_height(epoch, head, ResolveNullTipset::TakeOlder) + .await?; + + let options = ExportOptions { + skip_checksum, + include_receipts, + include_events, + include_tipset_keys, + include_tipset_lookup: tipset_lookup, + seen: FileBackedCidHashSet::new(ctx.temp_dir.as_path())?, + }; + let tmp_path = tempfile::TempPath::try_from_path(tmp_exporting_forest_car_path(&output_path))?; + let writer = if dry_run { + tokio_util::either::Either::Left(VoidAsyncWriter) + } else { + tokio_util::either::Either::Right(tokio::fs::File::create(&tmp_path).await?) + }; + let chain_export = match version { + FilecoinSnapshotVersion::V1 => { + crate::chain::export::(ctx.db(), &start_ts, recent_roots, writer, options) + .boxed() + } + FilecoinSnapshotVersion::V2 => { + let f3_snap_tmp_path = { + let mut f3_snap_dir = output_path.clone(); + let mut builder = tempfile::Builder::new(); + let with_suffix = builder.suffix(".f3snap.bin"); + if f3_snap_dir.pop() { + with_suffix.tempfile_in(&f3_snap_dir) + } else { + with_suffix.tempfile_in(".") + }? + .into_temp_path() }; - let chain_export = match version { - FilecoinSnapshotVersion::V1 => crate::chain::export::( - ctx.db(), - &start_ts, - recent_roots, - writer, - options, - ) - .boxed(), - FilecoinSnapshotVersion::V2 => { - let f3_snap_tmp_path = { - let mut f3_snap_dir = output_path.clone(); - let mut builder = tempfile::Builder::new(); - let with_suffix = builder.suffix(".f3snap.bin"); - if f3_snap_dir.pop() { - with_suffix.tempfile_in(&f3_snap_dir) - } else { - with_suffix.tempfile_in(".") - }? - .into_temp_path() - }; - let f3_snap = { - match F3ExportLatestSnapshot::run(f3_snap_tmp_path.display().to_string()) - .await - { - Ok(cid) => Some((cid, File::open(&f3_snap_tmp_path)?)), - Err(e) => { - tracing::error!("Failed to export F3 snapshot: {e:#}"); - None - } - } - }; - crate::chain::export_v2::( - ctx.db(), - f3_snap, - &start_ts, - recent_roots, - writer, - options, - ) - .boxed() + let f3_snap = { + match F3ExportLatestSnapshot::run(f3_snap_tmp_path.display().to_string()).await { + Ok(cid) => Some((cid, File::open(&f3_snap_tmp_path)?)), + Err(e) => { + tracing::error!("Failed to export F3 snapshot: {e:#}"); + None + } } }; - match chain_export_guard.run_cancellable(chain_export).await { - Some(result) => { - let ExportResult { - checksum, - tipset_lookup: hamt, - } = result?; - if !dry_run { - let output_path = output_path.clone(); - spawn_blocking_with_timeout(ASYNC_OPS_TIMEOUT, move || { - tmp_path.persist(&output_path)?; - // The snapshot at `output_path` is usable from this point on; - // a checksum-file failure is not worth failing the export over. - if let Some(checksum) = checksum - && let Err(e) = save_checksum(checksum, &output_path) - { - tracing::warn!( - "failed to save the checksum file for {}: {e:#}", - output_path.display() - ); - } - Ok(()) - }) - .await - .context("failed to persist the exported snapshot")?; - } - match (tipset_lookup, hamt) { - (true, Some(hamt)) => { - let mut hamt = - hamt.context("failed to generate tipset lookup snapshot")?; - let roots = nunny::vec![hamt.flush()?]; - let hamt_output_path = - forest_car_with_filename_suffix(&output_path, "_tipset_lookup")?; - let (mut writer, hamt_output_tmp_path) = if dry_run { - (tokio_util::either::Either::Left(VoidAsyncWriter), None) - } else { - let tmp_path = tempfile::TempPath::try_from_path( - tmp_exporting_forest_car_path(&hamt_output_path), - )?; - ( - tokio_util::either::Either::Right( - tokio::fs::File::create(&tmp_path).await?, - ), - Some(tmp_path), - ) - }; - hamt.into_store() - .export_forest_car(roots, &mut writer) - .await - .context("failed to write tipset lookup snapshot")?; - if let Some(hamt_output_tmp_path) = hamt_output_tmp_path { - hamt_output_tmp_path.persist(&hamt_output_path)?; - if !skip_checksum { - // No need to generate checksum on the fly for small snapshots - save_checksum( - Sha256::digest(std::fs::read(&hamt_output_path)?), - &hamt_output_path, - )?; - } - } - } - (true, None) => { - anyhow::bail!("requested tipset lookup snapshot is missing") - } - (false, Some(_)) => { - anyhow::bail!( - "tipset lookup snapshot should not be generated when it's not requested" - ) - } - _ => {} + crate::chain::export_v2::( + ctx.db(), + f3_snap, + &start_ts, + recent_roots, + writer, + options, + ) + .boxed() + } + }; + match chain_export_guard.run_cancellable(chain_export).await { + Some(result) => { + let ExportResult { + checksum, + tipset_lookup: hamt, + } = result?; + if !dry_run { + let output_path = output_path.clone(); + spawn_blocking_with_timeout(ASYNC_OPS_TIMEOUT, move || { + tmp_path.persist(&output_path)?; + // The snapshot at `output_path` is usable from this point on; + // a checksum-file failure is not worth failing the export over. + if let Some(checksum) = checksum + && let Err(e) = save_checksum(checksum, &output_path) + { + tracing::warn!( + "failed to save the checksum file for {}: {e:#}", + output_path.display() + ); } - if augmented_snapshot { - // It takes <10s on mainnet so export it in sequence for simplicity. - // Some stats: - // calibnet 3895470+2000: 5s 5.4MiB - // mainnet 6193120+2000: 5s 12MiB - let augmented_snapshot_output_path = - forest_car_with_filename_suffix(&output_path, "_receipts_events")?; - let (writer, augmented_snapshot_output_tmp_path) = if dry_run { + Ok(()) + }) + .await + .context("failed to persist the exported snapshot")?; + } + // The auxiliary snapshots must stay cancellable: the guard is still held, + // so an unguarded await here would accept a cancel yet ignore it. + let auxiliary_exports = async { + match (tipset_lookup, hamt) { + (true, Some(hamt)) => { + let mut hamt = hamt.context("failed to generate tipset lookup snapshot")?; + let roots = nunny::vec![hamt.flush()?]; + let hamt_output_path = + forest_car_with_filename_suffix(&output_path, "_tipset_lookup")?; + let (mut writer, hamt_output_tmp_path) = if dry_run { (tokio_util::either::Either::Left(VoidAsyncWriter), None) } else { let tmp_path = tempfile::TempPath::try_from_path( - tmp_exporting_forest_car_path(&augmented_snapshot_output_path), + tmp_exporting_forest_car_path(&hamt_output_path), )?; ( tokio_util::either::Either::Right( @@ -483,38 +448,90 @@ impl RpcMethod<1> for ForestChainExport { Some(tmp_path), ) }; - crate::chain::export_receipts_events_to_forest_car( - ctx.db(), - &start_ts, - recent_roots, - writer, - ) - .await - .context("failed to export message receipts and events snapshot")?; - if let Some(augmented_snapshot_output_tmp_path) = - augmented_snapshot_output_tmp_path - { - augmented_snapshot_output_tmp_path - .persist(&augmented_snapshot_output_path)?; + hamt.into_store() + .export_forest_car(roots, &mut writer) + .await + .context("failed to write tipset lookup snapshot")?; + if let Some(hamt_output_tmp_path) = hamt_output_tmp_path { + hamt_output_tmp_path.persist(&hamt_output_path)?; if !skip_checksum { // No need to generate checksum on the fly for small snapshots save_checksum( - Sha256::digest(std::fs::read(&augmented_snapshot_output_path)?), - &augmented_snapshot_output_path, + Sha256::digest(std::fs::read(&hamt_output_path)?), + &hamt_output_path, )?; } } } - chain_export_guard.mark_as_succeeded(); - anyhow::Ok(ApiExportResult::Done) + (true, None) => { + anyhow::bail!("requested tipset lookup snapshot is missing") + } + (false, Some(_)) => { + anyhow::bail!( + "tipset lookup snapshot should not be generated when it's not requested" + ) + } + _ => {} + } + if augmented_snapshot { + // It takes <10s on mainnet so export it in sequence for simplicity. + // Some stats: + // calibnet 3895470+2000: 5s 5.4MiB + // mainnet 6193120+2000: 5s 12MiB + let augmented_snapshot_output_path = + forest_car_with_filename_suffix(&output_path, "_receipts_events")?; + let (writer, augmented_snapshot_output_tmp_path) = if dry_run { + (tokio_util::either::Either::Left(VoidAsyncWriter), None) + } else { + let tmp_path = tempfile::TempPath::try_from_path( + tmp_exporting_forest_car_path(&augmented_snapshot_output_path), + )?; + ( + tokio_util::either::Either::Right( + tokio::fs::File::create(&tmp_path).await?, + ), + Some(tmp_path), + ) + }; + crate::chain::export_receipts_events_to_forest_car( + ctx.db(), + &start_ts, + recent_roots, + writer, + ) + .await + .context("failed to export message receipts and events snapshot")?; + if let Some(augmented_snapshot_output_tmp_path) = + augmented_snapshot_output_tmp_path + { + augmented_snapshot_output_tmp_path + .persist(&augmented_snapshot_output_path)?; + if !skip_checksum { + // No need to generate checksum on the fly for small snapshots + save_checksum( + Sha256::digest(std::fs::read(&augmented_snapshot_output_path)?), + &augmented_snapshot_output_path, + )?; + } + } + } + anyhow::Ok(()) + }; + match chain_export_guard.run_cancellable(auxiliary_exports).await { + Some(result) => { + result?; + Ok(ApiExportResult::Done) } None => { - tracing::warn!("Snapshot export was cancelled"); - anyhow::Ok(ApiExportResult::Cancelled) + tracing::warn!("Auxiliary snapshot exports were cancelled"); + Ok(ApiExportResult::Cancelled) } } - }); - Ok(handle.await??) + } + None => { + tracing::warn!("Snapshot export was cancelled"); + Ok(ApiExportResult::Cancelled) + } } } @@ -535,9 +552,9 @@ impl RpcMethod<0> for ForestChainExportStatus { (): Self::Params, _: &http::Extensions, ) -> Result { - let status = &*CHAIN_EXPORT_STATUS; - let initial_epoch = status.initial_epoch(); - let epoch = status.epoch(); + let snapshot = CHAIN_EXPORT_STATUS.snapshot(); + let initial_epoch = snapshot.initial_epoch; + let epoch = snapshot.epoch; let progress = if initial_epoch == 0 { 0.0 } else { @@ -551,17 +568,15 @@ impl RpcMethod<0> for ForestChainExportStatus { // only two decimal places let progress = (progress * 100.0).round() / 100.0; - let status = ApiExportStatus { + Ok(ApiExportStatus { + state: snapshot.state, + kind: snapshot.kind, + error: snapshot.error, progress, - exporting: status.exporting(), - cancelled: status.cancelled(), - succeeded: status.succeeded(), - start_time: status.start_time(), + start_time: snapshot.start_time, current_epoch: epoch, start_epoch: initial_epoch, - }; - - Ok(status) + }) } } @@ -582,14 +597,7 @@ impl RpcMethod<0> for ForestChainExportCancel { (): Self::Params, _: &http::Extensions, ) -> Result { - if CHAIN_EXPORT_STATUS.exporting() - && let Some(token) = CHAIN_EXPORT_STATUS.cancellation_token() - { - token.cancel(); - Ok(true) - } else { - Ok(false) - } + Ok(CHAIN_EXPORT_STATUS.cancel_running()) } } @@ -613,58 +621,65 @@ impl RpcMethod<1> for ForestChainExportDiff { // Spawn a task so it's not cancelled when CLI client is disconnected // So do not wrap this with `AbortOnDropHandle` let handle = tokio::spawn(async move { - let chain_export_guard = ChainExportGuard::try_start_export()?; - - let ForestChainExportDiffParams { - from, - to, - depth, - output_path, - } = params; + let chain_export_guard = + ChainExportGuard::try_start_export(ChainExportKind::DiffSnapshot)?; + let result = export_diff_inner(&ctx, params, &chain_export_guard).await; + chain_export_guard.finish(result) + }); + Ok(handle.await??) + } +} - let chain_finality = ctx.chain_config().policy.chain_finality; - anyhow::ensure!( - depth >= chain_finality, - "depth {depth} must be greater than or equal to chain_finality {chain_finality}" - ); +async fn export_diff_inner( + ctx: &Ctx, + params: ForestChainExportDiffParams, + chain_export_guard: &ChainExportGuard, +) -> anyhow::Result { + let ForestChainExportDiffParams { + from, + to, + depth, + output_path, + } = params; + + let chain_finality = ctx.chain_config().policy.chain_finality; + anyhow::ensure!( + depth >= chain_finality, + "depth {depth} must be greater than or equal to chain_finality {chain_finality}" + ); - let head = ctx.chain_store().heaviest_tipset(); - let start_ts = ctx - .chain_index() - .load_required_tipset_by_height(from, head, ResolveNullTipset::TakeOlder) - .await?; - let tmp_path = - tempfile::TempPath::try_from_path(tmp_exporting_forest_car_path(&output_path))?; - let chain_export = crate::tool::subcommands::archive_cmd::do_export( - ctx.chain_index().db(), - start_ts, - Some(ctx.chain_store().genesis_tipset()), - tmp_path.to_path_buf(), - None, - depth, - Some(to), - Some(chain_finality), - true, - ); + let head = ctx.chain_store().heaviest_tipset(); + let start_ts = ctx + .chain_index() + .load_required_tipset_by_height(from, head, ResolveNullTipset::TakeOlder) + .await?; + let tmp_path = tempfile::TempPath::try_from_path(tmp_exporting_forest_car_path(&output_path))?; + let chain_export = crate::tool::subcommands::archive_cmd::do_export( + ctx.chain_index().db(), + start_ts, + Some(ctx.chain_store().genesis_tipset()), + tmp_path.to_path_buf(), + None, + depth, + Some(to), + Some(chain_finality), + true, + ); - match chain_export_guard.run_cancellable(chain_export).await { - Some(result) => { - result?; - spawn_blocking_with_timeout(ASYNC_OPS_TIMEOUT, move || { - Ok(tmp_path.persist(&output_path)?) - }) - .await - .context("failed to persist the exported snapshot")?; - chain_export_guard.mark_as_succeeded(); - anyhow::Ok(ApiExportResult::Done) - } - None => { - tracing::warn!("Diff snapshot export was cancelled"); - anyhow::Ok(ApiExportResult::Cancelled) - } - } - }); - Ok(handle.await??) + match chain_export_guard.run_cancellable(chain_export).await { + Some(result) => { + result?; + spawn_blocking_with_timeout(ASYNC_OPS_TIMEOUT, move || { + Ok(tmp_path.persist(&output_path)?) + }) + .await + .context("failed to persist the exported snapshot")?; + Ok(ApiExportResult::Done) + } + None => { + tracing::warn!("Diff snapshot export was cancelled"); + Ok(ApiExportResult::Cancelled) + } } } diff --git a/src/rpc/mod.rs b/src/rpc/mod.rs index 152e6c00ba67..6394f4fa5fdf 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -27,7 +27,7 @@ use crate::rpc::methods::eth::pubsub_trait::EthPubSubApiServer; use crate::shim::clock::ChainEpoch; use ahash::HashMap; use clap::ValueEnum as _; -pub use client::Client; +pub use client::{Client, humanize_rpc_error}; pub use error::ServerError; use eth::filter::EthEventHandler; use filter_layer::FilterLayer; diff --git a/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap b/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap index 762e9c519f1a..6da2d29ecd88 100644 --- a/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap +++ b/src/rpc/snapshots/forest__rpc__tests__rpc__v0.snap @@ -4853,13 +4853,19 @@ components: ApiExportStatus: type: object properties: - cancelled: - type: boolean current_epoch: type: integer format: int64 - exporting: - type: boolean + error: + description: "Error of the last failed export, if recorded." + type: + - string + - "null" + kind: + description: "`None` if and only if no export has run since the node started." + anyOf: + - $ref: "#/components/schemas/ChainExportKind" + - type: "null" progress: type: number format: double @@ -4871,13 +4877,11 @@ components: - string - "null" format: date-time - succeeded: - type: boolean + state: + $ref: "#/components/schemas/ChainExportState" required: + - state - progress - - exporting - - cancelled - - succeeded - start_epoch - current_epoch ApiF3ParticipationLease: @@ -5320,6 +5324,18 @@ components: - MaxWantedChainsPerInstance - RebroadcastInterval - MaxTimestampAge + ChainExportKind: + description: What kind of export is (or was last) holding the chain-export slot. + oneOf: + - description: "A snapshot export requested via `Forest.ChainExport`." + type: string + const: Snapshot + - description: "A diff snapshot export requested via `Forest.ChainExportDiff`." + type: string + const: DiffSnapshot + - description: A lite snapshot export performed by the automatic snapshot GC. + type: string + const: SnapshotGc ChainExportParams: type: object properties: @@ -5344,6 +5360,15 @@ components: - tipset_keys - skip_checksum - dry_run + ChainExportState: + description: "Transitions only through [`ChainExportGuard`]: `Running` while a guard is held, then\nexactly one terminal state once it drops. `Idle`: no export has run since node start." + type: string + enum: + - Idle + - Running + - Succeeded + - Cancelled + - Failed ChangedType: description: Represents a changed value with before and after states. type: object diff --git a/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap b/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap index 95ce13c39b7f..df4b82c89d73 100644 --- a/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap +++ b/src/rpc/snapshots/forest__rpc__tests__rpc__v1.snap @@ -4922,13 +4922,19 @@ components: ApiExportStatus: type: object properties: - cancelled: - type: boolean current_epoch: type: integer format: int64 - exporting: - type: boolean + error: + description: "Error of the last failed export, if recorded." + type: + - string + - "null" + kind: + description: "`None` if and only if no export has run since the node started." + anyOf: + - $ref: "#/components/schemas/ChainExportKind" + - type: "null" progress: type: number format: double @@ -4940,13 +4946,11 @@ components: - string - "null" format: date-time - succeeded: - type: boolean + state: + $ref: "#/components/schemas/ChainExportState" required: + - state - progress - - exporting - - cancelled - - succeeded - start_epoch - current_epoch ApiF3ParticipationLease: @@ -5389,6 +5393,18 @@ components: - MaxWantedChainsPerInstance - RebroadcastInterval - MaxTimestampAge + ChainExportKind: + description: What kind of export is (or was last) holding the chain-export slot. + oneOf: + - description: "A snapshot export requested via `Forest.ChainExport`." + type: string + const: Snapshot + - description: "A diff snapshot export requested via `Forest.ChainExportDiff`." + type: string + const: DiffSnapshot + - description: A lite snapshot export performed by the automatic snapshot GC. + type: string + const: SnapshotGc ChainExportParams: type: object properties: @@ -5413,6 +5429,15 @@ components: - tipset_keys - skip_checksum - dry_run + ChainExportState: + description: "Transitions only through [`ChainExportGuard`]: `Running` while a guard is held, then\nexactly one terminal state once it drops. `Idle`: no export has run since node start." + type: string + enum: + - Idle + - Running + - Succeeded + - Cancelled + - Failed ChangedType: description: Represents a changed value with before and after states. type: object diff --git a/src/rpc/snapshots/forest__rpc__tests__rpc__v2.snap b/src/rpc/snapshots/forest__rpc__tests__rpc__v2.snap index ef8d3f1bee2b..0af89fa0e625 100644 --- a/src/rpc/snapshots/forest__rpc__tests__rpc__v2.snap +++ b/src/rpc/snapshots/forest__rpc__tests__rpc__v2.snap @@ -1733,13 +1733,19 @@ components: ApiExportStatus: type: object properties: - cancelled: - type: boolean current_epoch: type: integer format: int64 - exporting: - type: boolean + error: + description: "Error of the last failed export, if recorded." + type: + - string + - "null" + kind: + description: "`None` if and only if no export has run since the node started." + anyOf: + - $ref: "#/components/schemas/ChainExportKind" + - type: "null" progress: type: number format: double @@ -1751,13 +1757,11 @@ components: - string - "null" format: date-time - succeeded: - type: boolean + state: + $ref: "#/components/schemas/ChainExportState" required: + - state - progress - - exporting - - cancelled - - succeeded - start_epoch - current_epoch Base64String: @@ -1930,6 +1934,27 @@ components: - $ref: "#/components/schemas/EthInt64" Bloom: type: string + ChainExportKind: + description: What kind of export is (or was last) holding the chain-export slot. + oneOf: + - description: "A snapshot export requested via `Forest.ChainExport`." + type: string + const: Snapshot + - description: "A diff snapshot export requested via `Forest.ChainExportDiff`." + type: string + const: DiffSnapshot + - description: A lite snapshot export performed by the automatic snapshot GC. + type: string + const: SnapshotGc + ChainExportState: + description: "Transitions only through [`ChainExportGuard`]: `Running` while a guard is held, then\nexactly one terminal state once it drops. `Idle`: no export has run since node start." + type: string + enum: + - Idle + - Running + - Succeeded + - Cancelled + - Failed ChainFinalityStatus: description: "Describes how the node is currently determining finality,\ncombining probabilistic EC finality (based on observed chain health) with\nF3 fast finality when available." type: object diff --git a/src/rpc/types/mod.rs b/src/rpc/types/mod.rs index 79d32b4c7713..b617d3f7ef09 100644 --- a/src/rpc/types/mod.rs +++ b/src/rpc/types/mod.rs @@ -567,10 +567,12 @@ impl From for Event { #[derive(Debug, Serialize, Deserialize, JsonSchema, Clone, PartialEq)] pub struct ApiExportStatus { + pub state: crate::ipld::ChainExportState, + /// `None` if and only if no export has run since the node started. + pub kind: Option, + /// Error of the last failed export, if recorded. + pub error: Option, pub progress: f64, - pub exporting: bool, - pub cancelled: bool, - pub succeeded: bool, pub start_epoch: ChainEpoch, pub current_epoch: ChainEpoch, pub start_time: Option>, @@ -578,6 +580,45 @@ pub struct ApiExportStatus { lotus_json_with_self!(ApiExportStatus); +/// The human-readable rendering used by `forest-cli snapshot export-status`. +impl std::fmt::Display for ApiExportStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + use crate::ipld::ChainExportState::*; + use crate::ipld::{format_start_time, kind_label}; + let kind = kind_label(self.kind); + match self.state { + Running => { + let percent = self.progress.clamp(0.0, 1.0) * 100.0; + let started = format_start_time(self.start_time); + if self.start_epoch == 0 { + // Pre-walk and export-at-genesis are indistinguishable here; claim neither. + write!( + f, + "Exporting ({kind}): {percent:.1}% (started at {started})" + ) + } else { + write!( + f, + "Exporting ({kind}): {percent:.1}% (walk at epoch {}, counting down from {}; started at {started})", + self.current_epoch, self.start_epoch, + ) + } + } + Idle => write!(f, "No export in progress"), + Succeeded => write!(f, "No export in progress (last {kind} export succeeded)"), + Cancelled => write!( + f, + "No export in progress (last {kind} export was cancelled)" + ), + Failed => write!( + f, + "No export in progress (last {kind} export failed: {})", + self.error.as_deref().unwrap_or("unknown error"), + ), + } + } +} + #[derive(Debug, Serialize, Deserialize, JsonSchema, Clone, PartialEq, Eq, Hash)] pub enum ApiExportResult { Done, diff --git a/src/rpc/types/tests.rs b/src/rpc/types/tests.rs index d432d6ee7271..bc6420de9c3d 100644 --- a/src/rpc/types/tests.rs +++ b/src/rpc/types/tests.rs @@ -38,3 +38,30 @@ fn test_api_tipset_key_inner(cids: Vec) { .unwrap_or_default(); assert_eq!(cids_from_api_ts, cids); } + +/// Pins the `export-status` text format: a running export always names its kind and, +/// once the walk reports progress, the raw epoch counters. +#[test] +fn api_export_status_display() { + use crate::ipld::{ChainExportKind, ChainExportState}; + let mut status = ApiExportStatus { + state: ChainExportState::Running, + kind: Some(ChainExportKind::Snapshot), + error: None, + progress: 0.0, + start_epoch: 3898735, + current_epoch: 3898000, + start_time: None, + }; + assert_eq!( + status.to_string(), + "Exporting (Snapshot): 0.0% (walk at epoch 3898000, counting down from 3898735; started at unknown)" + ); + + status.state = ChainExportState::Failed; + status.error = Some("missing state root".into()); + assert_eq!( + status.to_string(), + "No export in progress (last Snapshot export failed: missing state root)" + ); +} diff --git a/src/tool/main.rs b/src/tool/main.rs index d27e6c134bf2..a27dbc8053a9 100644 --- a/src/tool/main.rs +++ b/src/tool/main.rs @@ -19,7 +19,7 @@ where let client = crate::rpc::Client::default_or_from_env(None)?; // Run command - match cmd { + let result = match cmd { Subcommand::Backup(cmd) => cmd.run(), Subcommand::Benchmark(cmd) => cmd.run().await, Subcommand::StateMigration(cmd) => cmd.run().await, @@ -33,5 +33,6 @@ where Subcommand::Net(cmd) => cmd.run().await, Subcommand::Shed(cmd) => cmd.run(client).await, Subcommand::Completion(cmd) => cmd.run(&mut std::io::stdout()), - } + }; + result.map_err(crate::rpc::humanize_rpc_error) } diff --git a/src/tool/subcommands/archive_cmd.rs b/src/tool/subcommands/archive_cmd.rs index 4a9f1dc056c8..e006099da456 100644 --- a/src/tool/subcommands/archive_cmd.rs +++ b/src/tool/subcommands/archive_cmd.rs @@ -602,6 +602,11 @@ where let diff_ts: &Tipset = &diff_ts; let diff_limit = diff_depth.map(|depth| diff_ts.epoch() - depth).unwrap_or(0); let store = store.shallow_clone(); + info!( + "building the diff base CID set, from epoch {} down to genesis (state trees for epochs above {diff_limit})...", + diff_ts.epoch() + ); + let start = std::time::Instant::now(); let mut stream = stream_chain( store.shallow_clone(), diff_ts.clone().chain_owned(store.shallow_clone()), @@ -609,6 +614,10 @@ where FileBackedCidHashSet::new_in_temp_dir()?, ); while stream.try_next().await?.is_some() {} + info!( + "built the diff base CID set, took {}", + humantime::format_duration(start.elapsed()) + ); stream.into_seen() } else { FileBackedCidHashSet::new_in_temp_dir()? @@ -1222,6 +1231,85 @@ mod tests { use tempfile::TempDir; use tokio::io::BufReader; + /// `poisoned_state_root` lands on the epoch-3 block, inside the pre-walk's state-root + /// range. Default `chain4u` links are identity CIDs, which the walk skips. + async fn run_diff_export( + poisoned_state_root: Option, + output_path: PathBuf, + ) -> anyhow::Result<()> { + use crate::blocks::{Chain4U, HeaderBuilder, chain4u}; + use crate::db::MemoryDB; + use std::sync::Arc; + + let db = Arc::new(MemoryDB::default()); + let c4u = Chain4U::with_blockstore(db.clone()); + let mut b3 = HeaderBuilder::new(); + if let Some(root) = poisoned_state_root { + b3.with_state_root(root); + } + chain4u! { + in c4u; + genesis_ts @ [_genesis_header] + -> [_b1] -> [_b2] + -> [_b3 = b3] + -> [_b4] -> [_b5] + -> head_ts @ [_b6] + }; + + do_export( + &db, + head_ts.clone(), + Some(genesis_ts.clone()), + output_path, + None, + 6, // depth + Some(4), // diff + Some(2), // diff_depth + true, + ) + .await + } + + /// The diff pre-walk explains the "stuck at `Exporting: 0.0%` with no temporary + /// file" reports; pins that the file and progress appear only after it completes. + #[tokio::test] + #[serial_test::serial(chain_export)] + async fn export_diff_prewalk_reports_no_progress_and_creates_no_file() -> anyhow::Result<()> { + use crate::ipld::{CHAIN_EXPORT_STATUS, ChainExportGuard, ChainExportKind}; + use crate::utils::multihash::prelude::*; + + let tmp = TempDir::new()?; + + // Control: pins the failure below to the pre-walk, not argument validation. + let control_path = tmp.path().join("control.forest.car.zst"); + run_diff_export(None, control_path.clone()).await?; + assert!(control_path.exists()); + + let missing_state_root = Cid::new_v1( + fvm_ipld_encoding::DAG_CBOR, + MultihashCode::Blake2b256.digest(b"missing-state-root"), + ); + let output_path = tmp.path().join("diff.forest.car.zst"); + + // Hold the export guard so the status global behaves as in the RPC handler. + let _guard = ChainExportGuard::try_start_export(ChainExportKind::DiffSnapshot)?; + + run_diff_export(Some(missing_state_root), output_path.clone()) + .await + .expect_err("the diff pre-walk must fail on the missing state root"); + + assert!( + !output_path.exists(), + "output file must not exist: it is only created after the entire diff pre-walk" + ); + assert_eq!( + CHAIN_EXPORT_STATUS.snapshot().initial_epoch, + 0, + "no progress is ever reported during the diff pre-walk" + ); + Ok(()) + } + fn genesis_timestamp(genesis_car: &'static [u8]) -> u64 { let db = crate::db::car::PlainCar::try_from(genesis_car).unwrap(); let ts = db.heaviest_tipset().unwrap(); diff --git a/src/wallet/main.rs b/src/wallet/main.rs index 8327c7945f0d..939f88bdf1da 100644 --- a/src/wallet/main.rs +++ b/src/wallet/main.rs @@ -33,5 +33,7 @@ where let client = rpc::Client::default_or_from_env(opts.token.as_deref())?; // Run command - cmd.run(client, remote_wallet, encrypt).await + cmd.run(client, remote_wallet, encrypt) + .await + .map_err(rpc::humanize_rpc_error) }