protoblock 0.1.6

Asynchronous Bitcoin block ingestion pipeline with built-in reorg handling, backpressure, and observability
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

use crate::runtime::protocol::ProtocolError;
use anyhow::Error as AnyError;
use std::fmt;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use tokio::sync::Notify;
use tokio_util::sync::CancellationToken;

#[derive(Clone)]
pub struct FatalErrorHandler {
    inner: Arc<FatalInner>,
}

struct FatalInner {
    triggered: AtomicBool,
    root_shutdown: CancellationToken,
    run_shutdown: CancellationToken,
    captured_error: Mutex<Option<CapturedFatalError>>,
    notify: Notify,
}

#[derive(Clone)]
struct CapturedFatalError {
    inner: Arc<AnyError>,
}

impl CapturedFatalError {
    fn new(inner: AnyError) -> Self {
        Self {
            inner: Arc::new(inner),
        }
    }
}

impl fmt::Debug for CapturedFatalError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_tuple("CapturedFatalError")
            .field(&self.inner)
            .finish()
    }
}

impl fmt::Display for CapturedFatalError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Display::fmt(self.inner.as_ref(), f)
    }
}

impl std::error::Error for CapturedFatalError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(self.inner.as_ref().as_ref())
    }
}

impl FatalErrorHandler {
    pub fn new(root_shutdown: CancellationToken, run_shutdown: CancellationToken) -> Self {
        Self {
            inner: Arc::new(FatalInner {
                triggered: AtomicBool::new(false),
                root_shutdown,
                run_shutdown,
                captured_error: Mutex::new(None),
                notify: Notify::new(),
            }),
        }
    }

    pub fn trigger(&self, error: ProtocolError) -> AnyError {
        let stage = error.stage();

        if self.inner.triggered.swap(true, Ordering::SeqCst) {
            return error.into();
        }

        tracing::error!(
            stage = ?stage,
            error = %error,
            "fatal protocol error; initiating shutdown"
        );

        self.capture_error(CapturedFatalError::new(error.into()))
    }

    pub fn trigger_external(&self, context: &str, error: AnyError) -> AnyError {
        if self.inner.triggered.swap(true, Ordering::SeqCst) {
            return error;
        }

        tracing::error!(
            context,
            error = %error,
            "fatal pipeline error; initiating shutdown"
        );

        self.capture_error(CapturedFatalError::new(error))
    }

    fn capture_error(&self, error: CapturedFatalError) -> AnyError {
        {
            let mut slot = self.inner.captured_error.lock().unwrap();
            if slot.is_none() {
                *slot = Some(error.clone());
            }
        }

        self.inner.run_shutdown.cancel();
        self.inner.root_shutdown.cancel();
        self.inner.notify.notify_waiters();

        error.into()
    }

    pub fn error(&self) -> Option<AnyError> {
        self.inner
            .captured_error
            .lock()
            .unwrap()
            .as_ref()
            .map(|error| error.clone().into())
    }
}