bsv_wallet_cli/
chain_feed.rs

1//! The header service's chain-event feed, held by the daemon.
2//!
3//! The header service (`CHAINTRACKS_URL`, required, no default: P0-1c)
4//! journals every tip move, fork, reorg, invalidation, freeze, tip age and
5//! rung split as an envelope with a cursor (chaintracks-cloudflare v0.4.1
6//! `docs/CHAIN-EVENTS.md`, "Delivery and cursors"). The daemon holds one
7//! loop over it:
8//!
9//! 1. On start, replay from the stored cursor (`0` at birth) through the
10//!    page endpoint, `GET /events?since=<cursor>`, until `hasMore` is false.
11//! 2. Then hold the stream, `GET /events/stream?since=<cursor>` (SSE: an
12//!    `id:` per envelope, comment heartbeats).
13//! 3. Every envelope goes through `bsv_tracker::decode_envelope` and
14//!    `Tracker::on_chain` with the daemon's headers
15//!    ([`tracker_host::deliver_chain_event`]); the cursor is stored after
16//!    each delivery, in the same transaction as the rows it moved, in the
17//!    database the tracker's rows live in (`tracker_meta`, key
18//!    `feed_cursor`: one row, one column).
19//! 4. A dropped stream (EOF without an error event) reconnects: the page
20//!    from the stored cursor, then the stream with `Last-Event-ID`.
21//! 5. A decode fault, a cursor ahead of the journal (HTTP 409
22//!    `cursorAhead`), an `event: error` (`feedUnavailable`) or a refused
23//!    request (400, 404) is reported through the daemon's status
24//!    ([`FeedStatus::fault`]: its kind, the cursor the loop stopped at, the
25//!    offending envelope's cursor, the fault text) and the loop stops there:
26//!    no skip, no acknowledgment, no silent reconnect past it (the tracker's
27//!    host rule, bsv-tracker 0.4.0 `docs/CHAIN-EVENTS.md`). A restart of the
28//!    daemon starts again from the stored cursor.
29//!
30//! # Load
31//!
32//! One outbound stream per daemon, plus one page request per (re)connect
33//! (more only while the journal answers `hasMore`). The daemon runs nothing
34//! periodic of its own for this: between envelopes it waits on the stream,
35//! and the header service sends the heartbeats. A transport failure (no
36//! answer, a 5xx, a 429) is retried after a pause that doubles from one
37//! second to five minutes. The header service is read beyond the feed only
38//! when an envelope names a tracked transaction: its tip and the heights of
39//! the named transactions' proofs.
40
41use std::time::Duration;
42
43use anyhow::Result;
44use bsv_tracker::{decode_envelope, Clock};
45use bsv_wallet_toolbox::{StorageSqlx, WalletServices};
46use serde::{Deserialize, Serialize};
47
48use crate::tracker_host::{self, TickOptions};
49
50/// The pauses of a reconnect after a transport failure.
51#[derive(Debug, Clone)]
52pub struct FeedOptions {
53    /// The first pause, and the pause after a stream that delivered.
54    pub min_pause: Duration,
55    /// The pause doubles up to this.
56    pub max_pause: Duration,
57}
58
59impl Default for FeedOptions {
60    fn default() -> Self {
61        Self {
62            min_pause: Duration::from_secs(1),
63            max_pause: Duration::from_secs(300),
64        }
65    }
66}
67
68/// Why the loop stopped. The loop never moves past it.
69#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
70#[serde(rename_all = "camelCase")]
71pub struct FeedFault {
72    /// `decode` (an envelope or page the decoder refused), `cursorAhead`
73    /// (HTTP 409), `feedUnavailable` (an SSE `event: error`), or `refused`
74    /// (HTTP 400 or 404: a malformed cursor, or no feed at this source).
75    pub kind: String,
76    /// The stored cursor: the last envelope delivered, where the loop
77    /// stopped.
78    pub cursor: String,
79    /// The offending envelope's cursor, when it carried one.
80    pub event_cursor: Option<String>,
81    /// The fault, as the decoder or the header service gave it.
82    pub detail: String,
83}
84
85/// The daemon's view of its feed, stored in `tracker_meta` (key
86/// `feed_status`) and served at `GET /chain-feed`.
87#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
88#[serde(rename_all = "camelCase")]
89pub struct FeedStatus {
90    /// The header service (`CHAINTRACKS_URL`).
91    pub source: String,
92    /// `replaying` (pages), `following` (the stream), `reconnecting`
93    /// (after a transport failure or a dropped stream) or `stopped`.
94    pub state: String,
95    /// The stored cursor.
96    pub cursor: String,
97    /// Envelopes delivered since the daemon started.
98    pub delivered: u64,
99    /// Stream connections after the first.
100    pub reconnects: u64,
101    /// Set when the loop stopped: what it stopped on.
102    pub fault: Option<FeedFault>,
103    /// The last transport failure, retried.
104    pub last_error: Option<String>,
105    /// Seconds since the epoch, at the last change.
106    pub updated_at: u64,
107}
108
109/// The status the feed stored last, if it ever ran here.
110pub async fn read_status(storage: &StorageSqlx) -> Result<Option<FeedStatus>> {
111    let value: Option<String> =
112        sqlx::query_scalar("SELECT value FROM tracker_meta WHERE key = 'feed_status'")
113            .fetch_optional(storage.pool())
114            .await
115            .unwrap_or(None);
116    Ok(match value {
117        Some(v) => Some(serde_json::from_str(&v)?),
118        None => None,
119    })
120}
121
122async fn write_status(storage: &StorageSqlx, status: &FeedStatus) {
123    let written = sqlx::query(
124        "INSERT INTO tracker_meta (key, value) VALUES ('feed_status', ?) \
125         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
126    )
127    .bind(serde_json::to_string(status).unwrap_or_default())
128    .execute(storage.pool())
129    .await;
130    if let Err(e) = written {
131        tracing::warn!(error = %e, "chain feed: the status was not stored");
132    }
133}
134
135/// How one leg (the page replay, or one stream connection) ended.
136enum Leg {
137    /// The stream ended without an error event.
138    Dropped { delivered: bool },
139    /// Stop here and report.
140    Stop(FeedFault),
141    /// Retry after a pause.
142    Transient(String),
143}
144
145struct Feed<'a, V: ?Sized, C> {
146    base: String,
147    client: reqwest::Client,
148    storage: &'a StorageSqlx,
149    services: &'a V,
150    clock: &'a C,
151    opts: &'a TickOptions,
152    status: FeedStatus,
153}
154
155/// Hold the feed at `base` until it stops on a fault; return the status it
156/// stopped with (also stored). See the module docs.
157pub async fn run<V: WalletServices + ?Sized, C: Clock>(
158    base: &str,
159    storage: &StorageSqlx,
160    services: &V,
161    clock: &C,
162    opts: &TickOptions,
163    feed_opts: &FeedOptions,
164) -> FeedStatus {
165    let base = base.trim_end_matches('/').to_string();
166    let cursor = match tracker_host::feed_cursor(storage).await {
167        Ok(c) => c,
168        Err(e) => {
169            let status = FeedStatus {
170                source: base,
171                state: "stopped".into(),
172                cursor: "?".into(),
173                delivered: 0,
174                reconnects: 0,
175                fault: Some(FeedFault {
176                    kind: "decode".into(),
177                    cursor: "?".into(),
178                    event_cursor: None,
179                    detail: format!("the stored cursor: {e}"),
180                }),
181                last_error: None,
182                updated_at: clock.now(),
183            };
184            write_status(storage, &status).await;
185            return status;
186        }
187    };
188    let client = reqwest::Client::builder()
189        .connect_timeout(Duration::from_secs(10))
190        .build()
191        .unwrap_or_default();
192    let mut feed = Feed {
193        status: FeedStatus {
194            source: base.clone(),
195            state: "replaying".into(),
196            cursor: cursor.to_string(),
197            delivered: 0,
198            reconnects: 0,
199            fault: None,
200            last_error: None,
201            updated_at: clock.now(),
202        },
203        base,
204        client,
205        storage,
206        services,
207        clock,
208        opts,
209    };
210    let mut pause = feed_opts.min_pause;
211    let mut connects: u64 = 0;
212    loop {
213        feed.set_state("replaying").await;
214        let leg = match feed.replay().await {
215            Ok(()) => {
216                feed.set_state("following").await;
217                let leg = feed.stream(connects > 0).await;
218                connects += 1;
219                feed.status.reconnects = connects - 1;
220                leg
221            }
222            Err(leg) => leg,
223        };
224        match leg {
225            Leg::Stop(fault) => return feed.stop(fault).await,
226            Leg::Dropped { delivered } => {
227                if delivered {
228                    pause = feed_opts.min_pause;
229                }
230                feed.status.last_error = Some("the stream ended without an error event".into());
231                tracing::info!(
232                    cursor = %feed.status.cursor,
233                    "chain feed: the stream dropped; reconnecting from the stored cursor"
234                );
235            }
236            Leg::Transient(e) => {
237                tracing::warn!(cursor = %feed.status.cursor, error = %e, "chain feed: retrying");
238                feed.status.last_error = Some(e);
239            }
240        }
241        feed.set_state("reconnecting").await;
242        tokio::time::sleep(pause).await;
243        pause = (pause * 2).min(feed_opts.max_pause);
244    }
245}
246
247impl<V: WalletServices + ?Sized, C: Clock> Feed<'_, V, C> {
248    async fn set_state(&mut self, state: &str) {
249        self.status.state = state.into();
250        self.status.updated_at = self.clock.now();
251        write_status(self.storage, &self.status).await;
252    }
253
254    async fn stop(mut self, fault: FeedFault) -> FeedStatus {
255        tracing::error!(
256            kind = %fault.kind,
257            cursor = %fault.cursor,
258            event_cursor = ?fault.event_cursor,
259            detail = %fault.detail,
260            "chain feed stopped at its cursor; nothing past it is delivered until the daemon restarts"
261        );
262        self.status.fault = Some(fault);
263        self.set_state("stopped").await;
264        self.status
265    }
266
267    fn fault(&self, kind: &str, event_cursor: Option<&str>, detail: impl Into<String>) -> Leg {
268        Leg::Stop(FeedFault {
269            kind: kind.into(),
270            cursor: self.status.cursor.clone(),
271            event_cursor: event_cursor.map(str::to_string),
272            detail: detail.into(),
273        })
274    }
275
276    /// Read an HTTP answer's refusals: 409 and 400/404 stop, other
277    /// failures retry.
278    async fn refusal(&self, resp: reqwest::Response) -> Leg {
279        let code = resp.status();
280        let body = resp.text().await.unwrap_or_default();
281        let detail = format!("HTTP {}: {}", code.as_u16(), body.trim());
282        match code.as_u16() {
283            409 => self.fault("cursorAhead", None, detail),
284            400 | 404 | 405 | 410 => self.fault("refused", None, detail),
285            _ => Leg::Transient(detail),
286        }
287    }
288
289    /// Deliver one envelope at `cursor`. A fault stops the loop at the
290    /// stored cursor; a storage failure is retried from it.
291    async fn deliver(&mut self, cursor: &str, bytes: &[u8]) -> std::result::Result<(), Leg> {
292        let Ok(at) = cursor.trim().parse::<u64>() else {
293            return Err(self.fault("decode", Some(cursor), "a cursor that is not a decimal"));
294        };
295        let event = match decode_envelope(bytes) {
296            Ok(event) => event,
297            Err(fault) => return Err(self.fault("decode", Some(cursor), fault.to_string())),
298        };
299        let delivery = tracker_host::deliver_chain_event(
300            self.storage,
301            self.services,
302            self.clock,
303            self.opts,
304            at,
305            event,
306        )
307        .await
308        .map_err(|e| Leg::Transient(format!("the delivery at {cursor} was not stored: {e}")))?;
309        if at > self.status.cursor.parse().unwrap_or(0) {
310            self.status.cursor = at.to_string();
311            self.status.delivered += 1;
312        }
313        for moved in &delivery.moved {
314            tracing::info!(
315                cursor = at,
316                kind = delivery.kind,
317                txid = %moved.txid,
318                word = %moved.word,
319                reask = ?moved.reask,
320                suspect = moved.suspect,
321                "chain feed: a chain event moved a tracked transaction"
322            );
323        }
324        Ok(())
325    }
326
327    /// The pages from the stored cursor until `hasMore` is false.
328    async fn replay(&mut self) -> std::result::Result<(), Leg> {
329        loop {
330            let since = self.status.cursor.clone();
331            let url = format!("{}/events?since={since}", self.base);
332            let resp = self
333                .client
334                .get(&url)
335                .send()
336                .await
337                .map_err(|e| Leg::Transient(format!("GET /events: {e}")))?;
338            if !resp.status().is_success() {
339                return Err(self.refusal(resp).await);
340            }
341            let page: Page = match resp.json().await {
342                Ok(page) => page,
343                Err(e) => return Err(self.fault("decode", None, format!("a page: {e}"))),
344            };
345            for item in &page.events {
346                let bytes = serde_json::to_vec(&item.event).unwrap_or_default();
347                self.deliver(&item.cursor, &bytes).await?;
348            }
349            if !page.has_more {
350                return Ok(());
351            }
352            if page.events.is_empty() || page.cursor == since {
353                return Err(Leg::Transient(format!(
354                    "a page at {since} said hasMore and moved no cursor"
355                )));
356            }
357        }
358    }
359
360    /// One stream connection from the stored cursor.
361    async fn stream(&mut self, reconnect: bool) -> Leg {
362        let since = self.status.cursor.clone();
363        let url = format!("{}/events/stream?since={since}", self.base);
364        let mut request = self.client.get(&url).header("Accept", "text/event-stream");
365        if reconnect {
366            request = request.header("Last-Event-ID", &since);
367        }
368        let mut resp = match request.send().await {
369            Ok(resp) => resp,
370            Err(e) => return Leg::Transient(format!("GET /events/stream: {e}")),
371        };
372        if !resp.status().is_success() {
373            return self.refusal(resp).await;
374        }
375        let mut buffer: Vec<u8> = Vec::new();
376        let mut delivered = false;
377        loop {
378            let chunk = match resp.chunk().await {
379                Ok(Some(chunk)) => chunk,
380                Ok(None) | Err(_) => return Leg::Dropped { delivered },
381            };
382            buffer.extend(chunk.iter().filter(|b| **b != b'\r'));
383            while let Some(end) = buffer.windows(2).position(|w| w == b"\n\n") {
384                let block: Vec<u8> = buffer.drain(..end + 2).collect();
385                let message = SseMessage::parse(&String::from_utf8_lossy(&block));
386                if message.event.as_deref() == Some("error") {
387                    return self.fault(
388                        "feedUnavailable",
389                        message.id.as_deref(),
390                        message.data.unwrap_or_default(),
391                    );
392                }
393                let Some(data) = message.data else {
394                    continue;
395                };
396                let Some(id) = message.id else {
397                    return self.fault("decode", None, format!("an envelope with no id: {data}"));
398                };
399                if let Err(leg) = self.deliver(&id, data.as_bytes()).await {
400                    return leg;
401                }
402                delivered = true;
403            }
404        }
405    }
406}
407
408/// `GET /events` (the page carrier, version 1).
409#[derive(Debug, Deserialize)]
410#[serde(rename_all = "camelCase")]
411struct Page {
412    cursor: String,
413    events: Vec<PageItem>,
414    has_more: bool,
415}
416
417#[derive(Debug, Deserialize)]
418struct PageItem {
419    cursor: String,
420    event: serde_json::Value,
421}
422
423/// One SSE message: its `id`, `event` and joined `data` lines; comments
424/// (heartbeats) are dropped.
425#[derive(Debug, Default, PartialEq, Eq)]
426struct SseMessage {
427    id: Option<String>,
428    event: Option<String>,
429    data: Option<String>,
430}
431
432impl SseMessage {
433    fn parse(block: &str) -> Self {
434        let mut message = Self::default();
435        for line in block.lines() {
436            if line.is_empty() || line.starts_with(':') {
437                continue;
438            }
439            let (field, value) = line.split_once(':').unwrap_or((line, ""));
440            let value = value.strip_prefix(' ').unwrap_or(value);
441            match field {
442                "id" => message.id = Some(value.to_string()),
443                "event" => message.event = Some(value.to_string()),
444                "data" => {
445                    message.data = Some(match message.data.take() {
446                        Some(previous) => format!("{previous}\n{value}"),
447                        None => value.to_string(),
448                    })
449                }
450                _ => {}
451            }
452        }
453        message
454    }
455}
456
457#[cfg(test)]
458#[path = "chain_feed_tests.rs"]
459mod tests;
bsv_wallet_cli/chain_feed.rs

bsv_wallet_cli/
chain_feed.rs
