forest/cli/subcommands/
sync_cmd.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use crate::blocks::TipsetKey;
5use crate::chain_sync::{NodeSyncStatus, SyncStatusReport};
6use crate::rpc::sync::{SnapshotProgressState, SyncStatus};
7use crate::rpc::{self, prelude::*};
8use anyhow::Context;
9use cid::Cid;
10use clap::Subcommand;
11use dialoguer::console::{Term, measure_text_width};
12use std::{
13    io::{Write, stdout},
14    time::Duration,
15};
16use tokio::time;
17use tokio::time::sleep;
18
19#[derive(Debug, Subcommand)]
20pub enum SyncCommands {
21    /// Display continuous sync data until sync is complete
22    Wait {
23        /// Don't exit after node is synced
24        #[arg(short)]
25        watch: bool,
26    },
27    /// Check sync status
28    Status,
29    /// Check if a given block is marked bad, and for what reason
30    CheckBad {
31        #[arg(short)]
32        /// The block CID to check
33        cid: Cid,
34    },
35    /// Mark a given block as bad
36    MarkBad {
37        /// The block CID to mark as a bad block
38        #[arg(short)]
39        cid: Cid,
40    },
41}
42
43impl SyncCommands {
44    pub async fn run(self, client: rpc::Client) -> anyhow::Result<()> {
45        match self {
46            Self::Wait { watch } => {
47                let mut term = Term::buffered_stdout();
48                let mut last_term_frame: Option<(usize, (u16, u16))> = None;
49
50                handle_initial_snapshot_check(&client).await?;
51
52                let mut interval = tokio::time::interval(Duration::from_secs(1));
53                loop {
54                    interval.tick().await;
55                    let report = SyncStatus::call(&client, ())
56                        .await
57                        .context("Failed to get sync status")?;
58
59                    wait_for_node_to_start_syncing(&client).await?;
60
61                    let size = term.size();
62                    match last_term_frame {
63                        None => {}
64                        Some((_, last_size)) if last_size != size => term.clear_screen()?,
65                        Some((rows, _)) => clear_previous_lines(&term, rows)?,
66                    }
67
68                    let rows = print_sync_report_details(&mut term, &report, size.1 as usize)
69                        .context("Failed to print sync status report")?;
70                    last_term_frame = Some((rows, size));
71                    term.flush()?;
72
73                    // Exit if synced and not in watch mode.
74                    if !watch && report.status == NodeSyncStatus::Synced {
75                        writeln!(term, "\nSync complete!")?;
76                        term.flush()?;
77                        break;
78                    }
79                }
80
81                Ok(())
82            }
83
84            Self::Status => {
85                let sync_status = client.call(SyncStatus::request(())?).await?;
86                if sync_status.status == NodeSyncStatus::Initializing {
87                    // If a snapshot is required and not yet complete, return here
88                    if !check_snapshot_progress(&client, false)
89                        .await?
90                        .is_not_required()
91                    {
92                        println!("Please try again later, once the snapshot is downloaded...");
93                        return Ok(());
94                    };
95                }
96
97                // Print the status report once, without row counting for clearing
98                let mut term = Term::buffered_stdout();
99                let width = term.size().1 as usize;
100                _ = print_sync_report_details(&mut term, &sync_status, width)
101                    .context("Failed to print sync status report")?;
102                term.flush()?;
103
104                Ok(())
105            }
106            Self::CheckBad { cid } => {
107                let response = SyncCheckBad::call(&client, (cid,)).await?;
108                if response.is_empty() {
109                    println!("Block \"{cid}\" is not marked as a bad block");
110                } else {
111                    println!("{response}");
112                }
113                Ok(())
114            }
115            Self::MarkBad { cid } => {
116                SyncMarkBad::call(&client, (cid,)).await?;
117                println!("OK");
118                Ok(())
119            }
120        }
121    }
122}
123
124/// Writes the sync status report and returns the number of terminal *rows* it
125/// occupies.
126///
127/// Rows, not lines: a line wider than the terminal wraps onto several rows, and
128/// the caller clears the frame by moving the cursor up by rows. Counting
129/// `writeln!` calls instead leaves the topmost row of each frame behind on narrow
130/// terminals. See <https://github.com/ChainSafe/forest/issues/7366>.
131fn print_sync_report_details(
132    out: &mut impl Write,
133    report: &SyncStatusReport,
134    term_width: usize,
135) -> anyhow::Result<usize> {
136    let head_key_str = report
137        .current_head_key
138        .as_ref()
139        .map(tipset_key_to_string)
140        .unwrap_or_else(|| "[unknown]".to_string());
141
142    let mut lines = vec![
143        format!(
144            "Status: {:?} ({} epochs behind)",
145            report.status, report.epochs_behind
146        ),
147        format!(
148            "Node Head: Epoch {} ({head_key_str})",
149            report.current_head_epoch
150        ),
151        format!("Network Head: Epoch {}", report.network_head_epoch),
152        format!("Last Update: {}", report.last_updated.to_rfc3339()),
153    ];
154
155    // Print active sync tasks (forks)
156    let active_forks = &report.active_forks;
157    if active_forks.is_empty() {
158        lines.push("Active Sync Tasks: None".into());
159    } else {
160        lines.push("Active Sync Tasks:".into());
161        let mut sorted_forks = active_forks.clone();
162        sorted_forks.sort_by_key(|f| std::cmp::Reverse(f.target_epoch));
163        for fork in &sorted_forks {
164            let total_epochs_for_this_fork = fork
165                .target_epoch
166                .saturating_sub(fork.target_sync_epoch_start);
167            lines.push(format!(
168                "  - Fork Target: {} ({}), Stage: {}, Syncing Range: [{}..{}] ({} epochs)",
169                fork.target_epoch,
170                tipset_key_to_string(&fork.target_tipset_key),
171                fork.stage,
172                fork.target_sync_epoch_start,
173                fork.target_epoch,
174                total_epochs_for_this_fork
175            ));
176        }
177    }
178
179    let mut rows = 0;
180    for line in &lines {
181        writeln!(out, "{line}")?;
182        // Measured in terminal columns, so ANSI escapes are ignored and wide
183        // characters count for two.
184        rows += measure_text_width(line).div_ceil(term_width).max(1);
185    }
186
187    Ok(rows)
188}
189
190fn clear_previous_lines(term: &Term, rows: usize) -> anyhow::Result<()> {
191    term.clear_last_lines(rows)?;
192    Ok(())
193}
194
195fn tipset_key_to_string(key: &TipsetKey) -> String {
196    let cids = key.to_cids();
197    match cids.len() {
198        0 => "[]".to_string(),
199        _ => format!("[{}, ...]", cids.first()),
200    }
201}
202
203/// Check if the snapshot download is in progress, if wait is true,
204/// wait till snapshot download is completed else return after checking once
205async fn check_snapshot_progress(
206    client: &rpc::Client,
207    wait: bool,
208) -> anyhow::Result<SnapshotProgressState> {
209    let mut interval = time::interval(Duration::from_secs(5));
210    let mut stdout = stdout();
211    loop {
212        interval.tick().await;
213
214        let progress_state = client.call(SyncSnapshotProgress::request(())?).await?;
215
216        write!(
217            stdout,
218            "\r{}{}Snapshot status: {}\n",
219            anes::MoveCursorUp(1),
220            anes::ClearLine::All,
221            progress_state
222        )?;
223        stdout.flush()?;
224
225        match progress_state {
226            SnapshotProgressState::Completed | SnapshotProgressState::NotRequired => {
227                println!();
228                return Ok(progress_state);
229            }
230            _ if !wait => {
231                return Ok(progress_state);
232            }
233            _ => {} // continue
234        }
235    }
236}
237
238/// Waits for node initialization to complete (start `Syncing`).
239async fn wait_for_node_to_start_syncing(client: &rpc::Client) -> anyhow::Result<()> {
240    let mut is_msg_printed = false;
241    let term = Term::stdout();
242    const POLLING_INTERVAL: Duration = Duration::from_secs(1);
243
244    loop {
245        let report = SyncStatus::call(client, ())
246            .await
247            .context("Failed to get sync status while waiting for initialization to complete")?;
248
249        if report.status == NodeSyncStatus::Initializing {
250            term.write_str("\r🔄 Node syncing is initializing, please wait...")?;
251            term.flush()?;
252            is_msg_printed = true;
253
254            sleep(POLLING_INTERVAL).await;
255        } else {
256            if is_msg_printed {
257                term.clear_line()
258                    .context("Failed to clear initializing message")?;
259            }
260
261            break;
262        }
263    }
264
265    Ok(())
266}
267
268/// Checks if a snapshot download is required or in progress when the node is initializing.
269/// If a snapshot download is in progress, it waits for completion before starting the sync monitor.
270async fn handle_initial_snapshot_check(client: &rpc::Client) -> anyhow::Result<()> {
271    let initial_report = SyncStatus::call(client, ())
272        .await
273        .context("Failed to get sync status")?;
274    if initial_report.status == NodeSyncStatus::Initializing {
275        // if the snapshot download is not required, then return,
276        // else wait till the snapshot download is completed.
277        if !check_snapshot_progress(client, false)
278            .await?
279            .is_not_required()
280        {
281            check_snapshot_progress(client, true).await?;
282        }
283    }
284
285    Ok(())
286}
forest/cli/subcommands/sync_cmd.rs

forest/cli/subcommands/
sync_cmd.rs
