Skip to content

feat: improve export status info #7383

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
LesnyRumcajs wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

feat: improve export status info #7383

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 8 additions & 9 deletions scripts/tests/calibnet_export_check.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
Expand All @@ -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
Expand All @@ -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"
Expand All @@ -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"
Expand Down
13 changes: 6 additions & 7 deletions scripts/tests/calibnet_export_diff_check.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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"
Expand All @@ -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"
Expand Down
67 changes: 59 additions & 8 deletions src/chain/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -172,6 +173,63 @@ 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::<Sha256, _>(
&db,
&head,
0,
StallingWriter {
write_budget: 0,
stall_on_flush: true,
},
ExportOptions::<CidHashSet>::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());

export.abort();
Ok(())
}

fn six_block_chain() -> anyhow::Result<(Arc<MemoryDB>, 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,
Expand Down Expand Up @@ -210,14 +268,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::<Sha256, _>(
&db,
Expand Down
2 changes: 1 addition & 1 deletion src/cli/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
101 changes: 48 additions & 53 deletions src/cli/subcommands/snapshot_cmd.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -193,66 +194,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")
);
}
}

Expand Down
Loading
Loading