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

forest/ipld/
export_status.rs
