forest/ipld/
export_status.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4//! Status and life cycle of the chain-export slot shared by user-requested snapshot
5//! exports and the automatic snapshot GC.
6
7use crate::shim::clock::ChainEpoch;
8use chrono::{DateTime, Utc};
9use std::sync::Arc;
10use std::sync::LazyLock;
11use std::sync::atomic::{self, AtomicI64};
12use tokio_util::sync::CancellationToken;
13
14/// What kind of export is (or was last) holding the chain-export slot.
15#[derive(
16    Debug,
17    Clone,
18    Copy,
19    PartialEq,
20    Eq,
21    serde::Serialize,
22    serde::Deserialize,
23    schemars::JsonSchema,
24    strum::Display,
25)]
26pub enum ChainExportKind {
27    /// A snapshot export requested via `Forest.ChainExport`.
28    Snapshot,
29    /// A diff snapshot export requested via `Forest.ChainExportDiff`.
30    DiffSnapshot,
31    /// A lite snapshot export performed by the automatic snapshot GC.
32    SnapshotGc,
33}
34
35/// Transitions only through [`ChainExportGuard`]: `Running` while a guard is held, then
36/// exactly one terminal state once it drops. `Idle`: no export has run since node start.
37#[derive(
38    Debug,
39    Clone,
40    Copy,
41    PartialEq,
42    Eq,
43    serde::Serialize,
44    serde::Deserialize,
45    schemars::JsonSchema,
46    strum::Display,
47)]
48pub enum ChainExportState {
49    Idle,
50    Running,
51    Succeeded,
52    Cancelled,
53    Failed,
54}
55
56/// Cold state behind one mutex for consistent reads; only the per-block epoch counters
57/// are hot and lock-free.
58#[derive(Default)]
59struct StatusInner {
60    /// `None` while running and before the first export.
61    outcome: Option<ChainExportState>,
62    kind: Option<ChainExportKind>,
63    start_time: Option<DateTime<Utc>>,
64    error: Option<String>,
65    cancellation_token: Option<CancellationToken>,
66    /// See [`ProgressReporter`].
67    counters: Arc<ProgressCounters>,
68}
69
70impl StatusInner {
71    /// A live cancellation token exists exactly while a [`ChainExportGuard`] is held.
72    fn is_running(&self) -> bool {
73        self.cancellation_token.is_some()
74    }
75}
76
77#[derive(Default)]
78struct ProgressCounters {
79    epoch: AtomicI64,
80    initial_epoch: AtomicI64,
81}
82
83#[derive(Default)]
84pub struct ExportStatus {
85    inner: parking_lot::Mutex<StatusInner>,
86}
87
88/// Read under one lock, so fields are mutually consistent.
89pub struct StatusSnapshot {
90    pub state: ChainExportState,
91    pub kind: Option<ChainExportKind>,
92    pub error: Option<String>,
93    pub start_time: Option<DateTime<Utc>>,
94    pub epoch: ChainEpoch,
95    pub initial_epoch: ChainEpoch,
96}
97
98pub fn kind_label(kind: Option<ChainExportKind>) -> String {
99    kind.map(|k| k.to_string())
100        .unwrap_or_else(|| "unknown".into())
101}
102
103pub fn format_start_time(start_time: Option<DateTime<Utc>>) -> String {
104    start_time
105        .map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
106        .unwrap_or_else(|| "unknown".into())
107}
108
109impl ExportStatus {
110    pub fn snapshot(&self) -> StatusSnapshot {
111        let inner = self.inner.lock();
112        StatusSnapshot {
113            state: if inner.is_running() {
114                ChainExportState::Running
115            } else {
116                inner.outcome.unwrap_or(ChainExportState::Idle)
117            },
118            kind: inner.kind,
119            error: inner.error.clone(),
120            start_time: inner.start_time,
121            epoch: inner.counters.epoch.load(atomic::Ordering::Relaxed),
122            initial_epoch: inner.counters.initial_epoch.load(atomic::Ordering::Relaxed),
123        }
124    }
125
126    /// Check-and-cancel under one lock, so the cancel cannot land on a different export
127    /// than the one observed.
128    pub fn cancel_running(&self) -> bool {
129        if let Some(token) = &self.inner.lock().cancellation_token {
130            token.cancel();
131            true
132        } else {
133            false
134        }
135    }
136
137    /// Holding the mutex makes check-and-start atomic: the lock is the export slot.
138    fn try_begin(
139        &self,
140        kind: ChainExportKind,
141        cancellation_token: CancellationToken,
142    ) -> anyhow::Result<()> {
143        let mut inner = self.inner.lock();
144        anyhow::ensure!(
145            !inner.is_running(),
146            "an active {} export has been running since {}; check `forest-cli snapshot export-status`",
147            kind_label(inner.kind),
148            format_start_time(inner.start_time),
149        );
150        *inner = StatusInner {
151            outcome: None,
152            kind: Some(kind),
153            start_time: Some(Utc::now()),
154            error: None,
155            cancellation_token: Some(cancellation_token),
156            counters: Arc::new(ProgressCounters::default()),
157        };
158        Ok(())
159    }
160
161    /// The first terminal outcome recorded for an export wins; later ones are ignored.
162    fn record_outcome(&self, outcome: ChainExportState, error: Option<String>) {
163        let mut inner = self.inner.lock();
164        if inner.outcome.is_none() {
165            inner.outcome = Some(outcome);
166            inner.error = error;
167        }
168    }
169
170    fn end(&self) {
171        let mut inner = self.inner.lock();
172        // Ended without a recorded outcome (panic, or a skipped `finish`): failed.
173        if inner.outcome.is_none() {
174            inner.outcome = Some(ChainExportState::Failed);
175            if !std::thread::panicking() {
176                tracing::warn!("chain export guard dropped without a recorded outcome");
177            }
178        }
179        inner.cancellation_token = None;
180    }
181
182    pub(super) fn progress_reporter(&self) -> ProgressReporter {
183        ProgressReporter(self.inner.lock().counters.clone())
184    }
185}
186
187/// Bound at creation to its export's freshly allocated counters: a producer task that
188/// outlives its export (a Tokio abort lands only at the next await) keeps writing into
189/// its own orphaned counters, never the next export's.
190#[derive(Clone)]
191pub struct ProgressReporter(Arc<ProgressCounters>);
192
193impl ProgressReporter {
194    pub fn update_epoch(&self, epoch: ChainEpoch) {
195        self.0.epoch.store(epoch, atomic::Ordering::Relaxed);
196        _ = self.0.initial_epoch.compare_exchange(
197            0,
198            epoch,
199            atomic::Ordering::Relaxed,
200            atomic::Ordering::Relaxed,
201        );
202    }
203}
204
205pub static CHAIN_EXPORT_STATUS: LazyLock<ExportStatus> = LazyLock::new(ExportStatus::default);
206
207#[derive(Debug)]
208pub struct ChainExportGuard {
209    cancellation_token: CancellationToken,
210}
211
212impl ChainExportGuard {
213    pub fn try_start_export(kind: ChainExportKind) -> anyhow::Result<Self> {
214        let cancellation_token = CancellationToken::new();
215        CHAIN_EXPORT_STATUS.try_begin(kind, cancellation_token.clone())?;
216        Ok(Self { cancellation_token })
217    }
218
219    /// Every export path that holds a [`ChainExportGuard`] must await its work through
220    /// this method — an export that does not race against the cancellation token cannot
221    /// be cancelled and appears stuck until process restart.
222    pub async fn run_cancellable<F: Future>(&self, fut: F) -> Option<F::Output> {
223        let output = self.cancellation_token.run_until_cancelled(fut).await;
224        if output.is_none() {
225            CHAIN_EXPORT_STATUS.record_outcome(ChainExportState::Cancelled, None);
226        }
227        output
228    }
229
230    /// A cancellation observed by [`Self::run_cancellable`] wins over `result`, so
231    /// callers need no cancellation special-casing.
232    pub fn finish<T>(self, result: anyhow::Result<T>) -> anyhow::Result<T> {
233        match &result {
234            Ok(_) => CHAIN_EXPORT_STATUS.record_outcome(ChainExportState::Succeeded, None),
235            Err(e) => {
236                CHAIN_EXPORT_STATUS.record_outcome(ChainExportState::Failed, Some(format!("{e:#}")))
237            }
238        }
239        result
240    }
241}
242
243impl Drop for ChainExportGuard {
244    fn drop(&mut self) {
245        // In case some tasks are waiting on this token
246        self.cancellation_token.cancel();
247        CHAIN_EXPORT_STATUS.end();
248    }
249}
250
251#[cfg(test)]
252mod tests {
253    use super::*;
254
255    /// Pins the invariant documented on [`ChainExportGuard::run_cancellable`].
256    #[tokio::test]
257    #[serial_test::serial(chain_export)]
258    async fn chain_export_cancel_stops_guarded_export() {
259        let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap();
260
261        let fut = g.run_cancellable(std::future::pending::<()>());
262        // Cancel exactly as the `Forest.ChainExportCancel` handler does.
263        assert!(CHAIN_EXPORT_STATUS.cancel_running());
264        assert!(
265            fut.await.is_none(),
266            "cancellation must interrupt the export"
267        );
268
269        assert_eq!(
270            CHAIN_EXPORT_STATUS.snapshot().state,
271            ChainExportState::Running
272        );
273        drop(g);
274        assert_eq!(
275            CHAIN_EXPORT_STATUS.snapshot().state,
276            ChainExportState::Cancelled
277        );
278    }
279
280    #[test]
281    #[serial_test::serial(chain_export)]
282    fn chain_export_status_reports_kind() {
283        let g = ChainExportGuard::try_start_export(ChainExportKind::SnapshotGc).unwrap();
284        assert_eq!(
285            CHAIN_EXPORT_STATUS.snapshot().kind,
286            Some(ChainExportKind::SnapshotGc)
287        );
288
289        // Rejecting a concurrent export must say what kind of export is in the way.
290        let err = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap_err();
291        assert!(
292            err.to_string().contains("SnapshotGc"),
293            "unexpected error: {err}"
294        );
295
296        // The kind outlives the export; the next export replaces it.
297        drop(g);
298        assert_eq!(
299            CHAIN_EXPORT_STATUS.snapshot().kind,
300            Some(ChainExportKind::SnapshotGc)
301        );
302        let _g = ChainExportGuard::try_start_export(ChainExportKind::DiffSnapshot).unwrap();
303        assert_eq!(
304            CHAIN_EXPORT_STATUS.snapshot().kind,
305            Some(ChainExportKind::DiffSnapshot)
306        );
307    }
308
309    #[test]
310    fn chain_export_state_starts_idle() {
311        assert_eq!(
312            ExportStatus::default().snapshot().state,
313            ChainExportState::Idle
314        );
315    }
316
317    /// Pins the transitions documented on [`ChainExportState`].
318    #[tokio::test]
319    #[serial_test::serial(chain_export)]
320    async fn chain_export_state_machine() {
321        let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap();
322        assert_eq!(
323            CHAIN_EXPORT_STATUS.snapshot().state,
324            ChainExportState::Running
325        );
326        g.finish(anyhow::Ok(())).unwrap();
327        assert_eq!(
328            CHAIN_EXPORT_STATUS.snapshot().state,
329            ChainExportState::Succeeded
330        );
331
332        // Failure.
333        let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap();
334        g.finish(anyhow::Result::<()>::Err(anyhow::anyhow!(
335            "missing state root"
336        )))
337        .unwrap_err();
338        assert_eq!(
339            CHAIN_EXPORT_STATUS.snapshot().state,
340            ChainExportState::Failed
341        );
342        assert_eq!(
343            CHAIN_EXPORT_STATUS.snapshot().error.as_deref(),
344            Some("missing state root")
345        );
346
347        // A guard dropped without `finish` still lands in `Failed`; the previous
348        // failure's error does not leak into the new export.
349        let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap();
350        drop(g);
351        assert_eq!(
352            CHAIN_EXPORT_STATUS.snapshot().state,
353            ChainExportState::Failed
354        );
355        assert_eq!(CHAIN_EXPORT_STATUS.snapshot().error, None);
356
357        // A cancelled export whose body then bails: no error recorded.
358        let g = ChainExportGuard::try_start_export(ChainExportKind::SnapshotGc).unwrap();
359        assert!(CHAIN_EXPORT_STATUS.cancel_running());
360        assert!(
361            g.run_cancellable(std::future::pending::<()>())
362                .await
363                .is_none()
364        );
365        g.finish(anyhow::Result::<()>::Err(anyhow::anyhow!(
366            "snapshot GC export was cancelled"
367        )))
368        .unwrap_err();
369        assert_eq!(
370            CHAIN_EXPORT_STATUS.snapshot().state,
371            ChainExportState::Cancelled
372        );
373        assert_eq!(CHAIN_EXPORT_STATUS.snapshot().error, None);
374
375        // A cancel after completion must not flip the terminal state.
376        let g = ChainExportGuard::try_start_export(ChainExportKind::Snapshot).unwrap();
377        g.finish(anyhow::Ok(())).unwrap();
378        assert!(!CHAIN_EXPORT_STATUS.cancel_running());
379        assert_eq!(
380            CHAIN_EXPORT_STATUS.snapshot().state,
381            ChainExportState::Succeeded
382        );
383    }
384}
forest/ipld/export_status.rs

forest/ipld/
export_status.rs
