Skip to main content

bsv_wallet_cli/
tracker_host.rs

1//! The CLI as a host of the transaction tracker (`bsv-tracker`).
2//!
3//! The tracker holds one word per transaction, the evidence behind it, the
4//! hints it has heard and the next re-ask. Only headers and proofs change a
5//! word; every broadcaster word is a hint that schedules a re-ask. The crate
6//! fetches nothing, reads no clock and stores nothing: the host supplies the
7//! four things below, runs the re-asks, and keeps the state.
8//!
9//! # The four host things
10//!
11//! | the tracker's trait | here | what it is |
12//! |---|---|---|
13//! | `Headers` | [`HeaderSnapshot`] | the header service's tip and the headers at the heights a pass needs, read once per pass through the wallet's services; a height not read is "no answer", a fault is a fault |
14//! | `ProofFetcher` | [`Courier`] | `get_merkle_path` for one named transaction; what it carries is a raw path, checked by the tracker against the snapshot |
15//! | `HintSource` | [`MemoryHints`] | the broadcast memory rows of the wallet's storage (what a broadcaster answered, what Arcade pushed over SSE or the webhook), each word fed once |
16//! | `Clock` | [`SystemClock`] | seconds since the epoch |
17//!
18//! The tracker's traits are synchronous and the wallet's transports are
19//! not, so a pass reads first (headers, memory rows, proofs for the re-asks
20//! that are due) and steps second.
21//!
22//! # The state, in the CLI's storage
23//!
24//! One row per tracked transaction in `tracker_states`, in the wallet's own
25//! database. A row holds inputs, never a trusted word: whether the host
26//! built it, the hints in the order heard, and the raw merkle path. Loading
27//! a row replays those inputs, so a stored path is checked again against
28//! the active header every time it is read (the tracker has no
29//! deserializer for a checked proof, by design). The `word`, `height` and
30//! `reask` columns are what the tracker said at the last step, kept for
31//! reading and for the indexes; nothing is decided from them.
32//!
33//! # What a pass asks, and what it never asks
34//!
35//! A pass ([`tick`]) names transactions; it never walks the chain and
36//! never asks a chain index. It loads the rows that are not yet mined, feeds
37//! the new hints, ticks the clock, and runs the re-asks that are due: one
38//! proof request for one transaction, with a doubling pause between two
39//! requests for the same transaction. A mined row is read again only when
40//! the header at or below its height moved (the last twelve tips are kept
41//! to see that) or when a spend names it ([`spend_guard`]).
42//!
43//! # What the header service pushes
44//!
45//! The daemon's chain-event feed ([`crate::chain_feed`]) hands each
46//! envelope to [`deliver_chain_event`]: the rows the event names by index
47//! (the mined rows at or above a fork, reorg or split height, the rows
48//! marked by an earlier event for a tip, the rows on an invalidated
49//! block's hash) are loaded into a [`Tracker`], and `Tracker::on_chain`
50//! moves their words. An event that left a row anything but a plain
51//! checked `mined` is kept on the row as one more input to replay (the
52//! `chain` column), so a `stale` or a suspect mark survives the next load;
53//! a new proof, or a stored path dropped because it no longer checks,
54//! clears it. The cursor is stored in the same transaction.
55//!
56//! One re-ask is an announce, not a proof request: a transaction a
57//! broadcaster called not final (a forced `gift-claim` before its lock
58//! time) is kept with its BEEF and handed to the broadcasters again at the
59//! lock time ([`heard`], and the re-announce step of [`tick`]).
60
61use std::collections::{BTreeMap, HashMap, VecDeque};
62use std::convert::Infallible;
63
64use anyhow::Result;
65use bsv_sdk::transaction::MerklePath;
66use bsv_sdk::wallet::{InternalizeActionArgs, WalletInterface};
67use bsv_tracker::{
68    decode_envelope, ChainEnvelope, ChainEvent, CheckError, Clock, Evidence, Header, HeaderError,
69    Headers, Height, Hint, HintSource, HintStatus, HostAction, Input, Params, Proof, ProofFetcher,
70    Reask, State, Timestamp, Tracker, TxId, Word,
71};
72use bsv_wallet_toolbox::services::PostBeefResult;
73use bsv_wallet_toolbox::{BroadcastStatus, StorageSqlx, WalletServices};
74use serde::{Deserialize, Serialize};
75use sqlx::Row as _;
76
77/// Default seconds without a new hint before an announced or seen
78/// transaction is asked for its proof (`TRACKER_AGE_SECS`): one block
79/// interval.
80pub const DEFAULT_AGE_SECS: u64 = 600;
81/// Default re-asks one pass runs (`TRACKER_MAX_ASKS`).
82pub const DEFAULT_MAX_ASKS: usize = 20;
83/// The pause between two re-asks of one transaction doubles up to this
84/// many times (600 s, then 1200 s, up to 64 times the age threshold).
85const MAX_BACKOFF_DOUBLINGS: u32 = 6;
86/// A transaction asked this many times in a row with no proof is not asked
87/// again until a new hint names it (about five days at the default age).
88const MAX_ASKS_PER_TX: u32 = 16;
89/// Recent tips kept to see a moved header.
90const TIP_RING: usize = 12;
91/// Default spend depth (`SPEND_DEPTH`), and its floor: a coin is offered
92/// when the block holding its transaction is at or below the tip.
93pub const DEFAULT_SPEND_DEPTH: Height = 1;
94
95// ---------------------------------------------------------------------------
96// Headers
97// ---------------------------------------------------------------------------
98
99/// One pass's view of the active chain: the tip and the headers at the
100/// heights the pass named, read from the header service through the
101/// wallet's services. Fails closed: a height that was not read has no
102/// answer, a read that failed is a fault, and with no tip nothing checks.
103#[derive(Debug, Clone)]
104pub struct HeaderSnapshot {
105    tip: std::result::Result<(Height, String), String>,
106    headers: BTreeMap<Height, std::result::Result<Header, String>>,
107}
108
109impl HeaderSnapshot {
110    /// Read the tip and the headers at `heights`.
111    pub async fn take<V: WalletServices + ?Sized>(
112        services: &V,
113        heights: impl IntoIterator<Item = Height>,
114    ) -> Self {
115        let tip = services
116            .get_chain_tip_header()
117            .await
118            .map(|h| (h.height, h.hash.to_ascii_lowercase()))
119            .map_err(|e| e.to_string());
120        let mut snapshot = Self {
121            tip,
122            headers: BTreeMap::new(),
123        };
124        snapshot.extend(services, heights).await;
125        snapshot
126    }
127
128    /// A snapshot of nothing: no tip, no header. Every check fails closed.
129    pub fn unread() -> Self {
130        Self {
131            tip: Err("the header service was not read".to_string()),
132            headers: BTreeMap::new(),
133        }
134    }
135
136    /// Read the headers at `heights` that this snapshot does not hold yet.
137    pub async fn extend<V: WalletServices + ?Sized>(
138        &mut self,
139        services: &V,
140        heights: impl IntoIterator<Item = Height>,
141    ) {
142        for height in heights {
143            if self.headers.contains_key(&height) {
144                continue;
145            }
146            let answer = match services.get_header_for_height(height).await {
147                Ok(bytes) => header_of(&bytes),
148                Err(e) => Err(e.to_string()),
149            };
150            self.headers.insert(height, answer);
151        }
152    }
153
154    /// The tip's height and hash, when the header service gave one.
155    pub fn tip(&self) -> Option<(Height, &str)> {
156        self.tip.as_ref().ok().map(|(h, hash)| (*h, hash.as_str()))
157    }
158}
159
160impl Headers for HeaderSnapshot {
161    fn header_at(&self, height: Height) -> std::result::Result<Option<Header>, HeaderError> {
162        match self.headers.get(&height) {
163            Some(Ok(header)) => Ok(Some(header.clone())),
164            Some(Err(fault)) => Err(HeaderError(fault.clone())),
165            None => Ok(None),
166        }
167    }
168
169    fn tip_height(&self) -> std::result::Result<Height, HeaderError> {
170        self.tip
171            .as_ref()
172            .map(|(height, _)| *height)
173            .map_err(|fault| HeaderError(fault.clone()))
174    }
175}
176
177/// The tracker's header projection of an 80-byte block header: its hash
178/// and its merkle root, both as displayed.
179fn header_of(bytes: &[u8]) -> std::result::Result<Header, String> {
180    if bytes.len() != 80 {
181        return Err(format!("a header of {} bytes", bytes.len()));
182    }
183    let display = |raw: &[u8]| {
184        let mut reversed = raw.to_vec();
185        reversed.reverse();
186        hex::encode(reversed)
187    };
188    Ok(Header {
189        hash: display(&bsv_sdk::primitives::hash::sha256d(bytes)),
190        merkle_root: display(&bytes[36..68]),
191    })
192}
193
194// ---------------------------------------------------------------------------
195// Clock
196// ---------------------------------------------------------------------------
197
198/// The host's clock: seconds since the epoch.
199#[derive(Debug, Clone, Copy, Default)]
200pub struct SystemClock;
201
202impl Clock for SystemClock {
203    fn now(&self) -> Timestamp {
204        std::time::SystemTime::now()
205            .duration_since(std::time::UNIX_EPOCH)
206            .map(|d| d.as_secs())
207            .unwrap_or(0)
208    }
209}
210
211// ---------------------------------------------------------------------------
212// Hints
213// ---------------------------------------------------------------------------
214
215/// The hints of one pass: the broadcast memory rows of the tracked
216/// transactions whose `(provider, word)` the row has not heard yet. A word
217/// a broadcaster repeats every minute is one hint, so a refreshed `seen`
218/// never resets the age clock.
219#[derive(Debug, Default)]
220pub struct MemoryHints {
221    queue: VecDeque<(TxId, Hint)>,
222}
223
224impl MemoryHints {
225    async fn read(storage: &StorageSqlx, rows: &[TrackedRow]) -> Result<Self> {
226        let txids: Vec<String> = rows.iter().map(|r| r.txid.clone()).collect();
227        if txids.is_empty() {
228            return Ok(Self::default());
229        }
230        let mut records = storage.broadcast_records(None, &txids).await?;
231        records.sort_by_key(|r| r.seen_at);
232        let mut queue = VecDeque::new();
233        for record in records {
234            let Some(status) = record.ladder_status() else {
235                continue;
236            };
237            let txid = record.txid.to_ascii_lowercase();
238            let Some(row) = rows.iter().find(|r| r.txid == txid) else {
239                continue;
240            };
241            // A broadcaster's "mined" carries no height here and is no
242            // proof: it reads as a word that disagrees and asks for one.
243            let (label, status) = match status {
244                BroadcastStatus::Accepted => ("accepted", HintStatus::Accepted),
245                BroadcastStatus::Seen => ("seen", HintStatus::Seen),
246                BroadcastStatus::Mined => ("mined", HintStatus::Unknown),
247                BroadcastStatus::Rejected => (
248                    "rejected",
249                    HintStatus::Rejected {
250                        reason: record.status.clone(),
251                    },
252                ),
253                BroadcastStatus::Unknown => ("unknown", HintStatus::Unknown),
254            };
255            let source = format!("{}|{}", record.provider, label);
256            if row.hints.iter().any(|h| h.source == source) {
257                continue;
258            }
259            let observed = record.seen_at.timestamp().max(0) as Timestamp;
260            queue.push_back((txid, Hint::new(source, status, observed)));
261        }
262        Ok(Self { queue })
263    }
264}
265
266impl HintSource for MemoryHints {
267    type Error = Infallible;
268
269    fn next_hint(&mut self) -> std::result::Result<Option<(TxId, Hint)>, Self::Error> {
270        Ok(self.queue.pop_front())
271    }
272}
273
274// ---------------------------------------------------------------------------
275// Proofs
276// ---------------------------------------------------------------------------
277
278/// The host's proof transport: `get_merkle_path` for the transactions whose
279/// re-ask is due, read before the tracker steps. What it hands over is a
280/// raw path bound to its txid; the tracker checks it.
281#[derive(Debug, Default)]
282pub struct Courier {
283    carried: HashMap<TxId, std::result::Result<Option<Proof>, String>>,
284}
285
286impl Courier {
287    /// Ask for the proof of each of `txids`, once.
288    pub async fn carry<V: WalletServices + ?Sized>(services: &V, txids: &[TxId]) -> Self {
289        let mut carried = HashMap::new();
290        for txid in txids {
291            let answer = match services.get_merkle_path(txid, false).await {
292                Ok(found) => match found.merkle_path {
293                    Some(hex) => MerklePath::from_hex(&hex)
294                        .map_err(|e| format!("not a merkle path: {e}"))
295                        .and_then(|path| Proof::new(txid.clone(), path).map_err(|e| e.to_string()))
296                        .map(Some),
297                    None => Ok(None),
298                },
299                Err(e) => Err(e.to_string()),
300            };
301            carried.insert(txid.clone(), answer);
302        }
303        Self { carried }
304    }
305
306    fn heights(&self) -> Vec<Height> {
307        self.carried
308            .values()
309            .filter_map(|answer| answer.as_ref().ok()?.as_ref().map(Proof::height))
310            .collect()
311    }
312}
313
314impl ProofFetcher for Courier {
315    type Error = String;
316
317    fn fetch(
318        &mut self,
319        txid: &str,
320        _reask: &Reask,
321    ) -> std::result::Result<Option<Proof>, Self::Error> {
322        self.carried.remove(txid).unwrap_or(Ok(None))
323    }
324}
325
326// ---------------------------------------------------------------------------
327// The stored row
328// ---------------------------------------------------------------------------
329
330#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
331struct StoredHint {
332    source: String,
333    status: String,
334    #[serde(default, skip_serializing_if = "Option::is_none")]
335    height: Option<Height>,
336    #[serde(default, skip_serializing_if = "Option::is_none")]
337    reason: Option<String>,
338    #[serde(default, skip_serializing_if = "Vec::is_empty")]
339    competitors: Vec<TxId>,
340    observed: Timestamp,
341}
342
343impl StoredHint {
344    fn of(hint: &Hint) -> Self {
345        let (status, height, reason, competitors) = match &hint.status {
346            HintStatus::Accepted => ("accepted", None, None, vec![]),
347            HintStatus::Seen => ("seen", None, None, vec![]),
348            HintStatus::Mined { height } => ("mined", Some(*height), None, vec![]),
349            HintStatus::StaleBlock => ("stale_block", None, None, vec![]),
350            HintStatus::Rejected { reason } => ("rejected", None, Some(reason.clone()), vec![]),
351            HintStatus::DoubleSpend { competitors } => {
352                ("double_spend", None, None, competitors.clone())
353            }
354            HintStatus::OrphanMempool => ("orphan_mempool", None, None, vec![]),
355            HintStatus::Unknown => ("unknown", None, None, vec![]),
356        };
357        Self {
358            source: hint.source.clone(),
359            status: status.to_string(),
360            height,
361            reason,
362            competitors,
363            observed: hint.observed,
364        }
365    }
366
367    fn hint(&self) -> Hint {
368        let status = match self.status.as_str() {
369            "accepted" => HintStatus::Accepted,
370            "seen" => HintStatus::Seen,
371            "mined" => self
372                .height
373                .map_or(HintStatus::Unknown, |height| HintStatus::Mined { height }),
374            "stale_block" => HintStatus::StaleBlock,
375            "rejected" => HintStatus::Rejected {
376                reason: self.reason.clone().unwrap_or_default(),
377            },
378            "double_spend" => HintStatus::DoubleSpend {
379                competitors: self.competitors.clone(),
380            },
381            "orphan_mempool" => HintStatus::OrphanMempool,
382            _ => HintStatus::Unknown,
383        };
384        Hint::new(self.source.clone(), status, self.observed)
385    }
386}
387
388/// One tracked transaction as the CLI's storage holds it: inputs to replay.
389#[derive(Debug, Clone, PartialEq, Eq)]
390struct TrackedRow {
391    txid: TxId,
392    built: bool,
393    /// Oldest first.
394    hints: Vec<StoredHint>,
395    /// The raw merkle path, hex. Checked again on every load.
396    bump: Option<String>,
397    asks: u32,
398    next_ask_at: Option<Timestamp>,
399    /// The BEEF to hand to the broadcasters again, and when.
400    reannounce_beef: Option<Vec<u8>>,
401    reannounce_at: Option<Timestamp>,
402    /// What the wallet records once a broadcaster takes it: the
403    /// `internalizeAction` arguments, JSON.
404    on_accept: Option<String>,
405    /// Chain events that left the checked word other than a plain
406    /// `mined` (stale, or suspect), oldest first, as their envelopes.
407    /// Replayed after the proof on every load.
408    chain: Vec<String>,
409}
410
411impl TrackedRow {
412    fn new(txid: &str) -> Self {
413        Self {
414            txid: txid.to_ascii_lowercase(),
415            built: true,
416            hints: vec![],
417            bump: None,
418            asks: 0,
419            next_ask_at: None,
420            reannounce_beef: None,
421            reannounce_at: None,
422            on_accept: None,
423            chain: vec![],
424        }
425    }
426
427    /// The stored chain events, decoded. They were decoded once before
428    /// they were stored; one that no longer decodes is reported and not
429    /// replayed.
430    fn chain_events(&self) -> Vec<ChainEvent> {
431        self.chain
432            .iter()
433            .filter_map(|json| match decode_envelope(json.as_bytes()) {
434                Ok(event) => Some(event),
435                Err(fault) => {
436                    tracing::warn!(txid = %self.txid, %fault, "a stored chain event does not decode");
437                    None
438                }
439            })
440            .collect()
441    }
442
443    /// Keep `event` on the row when it left the state anything but a
444    /// plain checked `mined`; forget every kept event when it did. One
445    /// `tip` is kept at most (each rechecks the same suspect mark).
446    fn keep_chain(&mut self, state: &State, event: &ChainEvent) -> Result<()> {
447        let plain = matches!(state.word(), Word::Mined(_)) && !state.suspect();
448        if plain && state.reask().is_none() {
449            self.chain.clear();
450            return Ok(());
451        }
452        if matches!(event, ChainEvent::Tip { .. } | ChainEvent::TipAge { .. }) {
453            self.chain.retain(|json| {
454                !matches!(
455                    decode_envelope(json.as_bytes()),
456                    Ok(ChainEvent::Tip { .. } | ChainEvent::TipAge { .. })
457                )
458            });
459        }
460        self.chain
461            .push(serde_json::to_string(&ChainEnvelope::new(event.clone()))?);
462        Ok(())
463    }
464
465    /// Feed the stored inputs to `tracker`, as [`TrackedRow::replay`] feeds
466    /// them to one state.
467    fn replay_into<H: Headers>(&self, tracker: &mut Tracker, headers: &H) {
468        tracker.track(self.txid.clone());
469        if self.built {
470            let _ = tracker.apply(&self.txid, headers, Input::Host(HostAction::Build));
471        }
472        for hint in &self.hints {
473            let _ = tracker.apply(&self.txid, headers, Input::Hint(hint.hint()));
474        }
475        if let Some(Ok(proof)) = self.proof() {
476            let _ = tracker.apply(&self.txid, headers, Input::Evidence(Evidence::Proof(proof)));
477        }
478        for event in self.chain_events() {
479            let _ = tracker.apply(&self.txid, headers, Input::Evidence(Evidence::Chain(event)));
480        }
481    }
482
483    fn proof(&self) -> Option<std::result::Result<Proof, CheckError>> {
484        let hex = self.bump.as_deref()?;
485        Some(
486            MerklePath::from_hex(hex)
487                .map_err(|e| CheckError::InvalidProof(e.to_string()))
488                .and_then(|path| Proof::new(self.txid.clone(), path)),
489        )
490    }
491
492    fn proof_height(&self) -> Option<Height> {
493        self.proof()?.ok().map(|p| p.height())
494    }
495
496    /// Replay the stored inputs. The stored path is checked against
497    /// `headers` like any other proof; a path that no longer checks leaves
498    /// the hint-tier word and a re-ask, and is returned as the fault.
499    fn replay<H: Headers>(&self, params: &Params, headers: &H) -> (State, Option<CheckError>) {
500        let mut state = State::new(self.txid.clone());
501        if self.built {
502            let _ = state.step(params, headers, Input::Host(HostAction::Build));
503        }
504        for hint in &self.hints {
505            let _ = state.step(params, headers, Input::Hint(hint.hint()));
506        }
507        let fault = match self.proof() {
508            Some(Ok(proof)) => state
509                .step(params, headers, Input::Evidence(Evidence::Proof(proof)))
510                .err(),
511            Some(Err(fault)) => Some(fault),
512            None => None,
513        };
514        for event in self.chain_events() {
515            let _ = state.step(params, headers, Input::Evidence(Evidence::Chain(event)));
516        }
517        (state, fault)
518    }
519}
520
521/// The tracker's word, as one lowercase label.
522pub fn word_label(word: &Word) -> &'static str {
523    match word {
524        Word::Unknown => "unknown",
525        Word::Built => "built",
526        Word::Announced => "announced",
527        Word::Seen => "seen",
528        Word::Mined(_) => "mined",
529        Word::Stale => "stale",
530        Word::Rejected { .. } => "rejected",
531        Word::Conflicted { .. } => "conflicted",
532        Word::Abandoned => "abandoned",
533    }
534}
535
536fn reask_label(reask: &Reask) -> String {
537    match reask {
538        Reask::Reorg { fork_height } => format!("reorg@{fork_height}"),
539        Reask::Fork { height } => format!("fork@{height}"),
540        Reask::Hint(hint) => format!("hint:{}", hint.source),
541        Reask::ProofFailed { height } => format!("proof_failed@{height}"),
542        Reask::Recheck => "recheck".to_string(),
543        Reask::Age => "age".to_string(),
544        Reask::Spend => "spend".to_string(),
545    }
546}
547
548async fn ensure_tables(storage: &StorageSqlx) -> Result<()> {
549    for sql in [
550        "CREATE TABLE IF NOT EXISTS tracker_states (\
551            txid TEXT PRIMARY KEY, \
552            built INTEGER NOT NULL DEFAULT 1, \
553            hints TEXT NOT NULL DEFAULT '[]', \
554            bump TEXT, \
555            word TEXT NOT NULL DEFAULT 'unknown', \
556            height INTEGER, \
557            header_hash TEXT, \
558            reask TEXT, \
559            asks INTEGER NOT NULL DEFAULT 0, \
560            next_ask_at INTEGER, \
561            reannounce_beef BLOB, \
562            reannounce_at INTEGER, \
563            on_accept TEXT, \
564            updated_at INTEGER NOT NULL DEFAULT 0)",
565        "CREATE INDEX IF NOT EXISTS tracker_states_word ON tracker_states (word)",
566        "CREATE INDEX IF NOT EXISTS tracker_states_height ON tracker_states (height)",
567        "CREATE TABLE IF NOT EXISTS tracker_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)",
568    ] {
569        sqlx::query(sql).execute(storage.pool()).await?;
570    }
571    // Added with the chain-event feed (0.9.0); a table made before it
572    // gains the column empty.
573    let has_chain: i64 = sqlx::query_scalar(
574        "SELECT COUNT(*) FROM pragma_table_info('tracker_states') WHERE name = 'chain'",
575    )
576    .fetch_one(storage.pool())
577    .await?;
578    if has_chain == 0 {
579        sqlx::query("ALTER TABLE tracker_states ADD COLUMN chain TEXT NOT NULL DEFAULT '[]'")
580            .execute(storage.pool())
581            .await?;
582    }
583    Ok(())
584}
585
586const ROW_COLUMNS: &str =
587    "txid, built, hints, bump, asks, next_ask_at, reannounce_beef, reannounce_at, on_accept, chain";
588
589fn row_of(row: &sqlx::sqlite::SqliteRow) -> TrackedRow {
590    let hints: String = row.get("hints");
591    let chain: String = row.get("chain");
592    TrackedRow {
593        txid: row.get("txid"),
594        built: row.get::<i64, _>("built") != 0,
595        hints: serde_json::from_str(&hints).unwrap_or_default(),
596        bump: row.get("bump"),
597        asks: row.get::<i64, _>("asks").max(0) as u32,
598        next_ask_at: row
599            .get::<Option<i64>, _>("next_ask_at")
600            .map(|t| t.max(0) as Timestamp),
601        reannounce_beef: row.get("reannounce_beef"),
602        reannounce_at: row
603            .get::<Option<i64>, _>("reannounce_at")
604            .map(|t| t.max(0) as Timestamp),
605        on_accept: row.get("on_accept"),
606        chain: serde_json::from_str(&chain).unwrap_or_default(),
607    }
608}
609
610async fn load_row(storage: &StorageSqlx, txid: &str) -> Result<Option<TrackedRow>> {
611    let row = sqlx::query(&format!(
612        "SELECT {ROW_COLUMNS} FROM tracker_states WHERE txid = ?"
613    ))
614    .bind(txid.to_ascii_lowercase())
615    .fetch_optional(storage.pool())
616    .await?;
617    Ok(row.as_ref().map(row_of))
618}
619
620async fn store_row(
621    storage: &StorageSqlx,
622    row: &TrackedRow,
623    state: &State,
624    now: Timestamp,
625) -> Result<()> {
626    store_row_in(storage.pool(), row, state, now).await
627}
628
629async fn store_row_in<'e, E: sqlx::Executor<'e, Database = sqlx::Sqlite>>(
630    executor: E,
631    row: &TrackedRow,
632    state: &State,
633    now: Timestamp,
634) -> Result<()> {
635    let (height, header_hash) = match state.word() {
636        Word::Mined(mined) => (
637            Some(mined.height() as i64),
638            Some(mined.checked().header().hash.clone()),
639        ),
640        _ => (None, None),
641    };
642    sqlx::query(
643        "INSERT INTO tracker_states \
644            (txid, built, hints, bump, word, height, header_hash, reask, asks, next_ask_at, \
645             reannounce_beef, reannounce_at, on_accept, chain, updated_at) \
646         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
647         ON CONFLICT(txid) DO UPDATE SET built = excluded.built, hints = excluded.hints, \
648            bump = excluded.bump, word = excluded.word, height = excluded.height, \
649            header_hash = excluded.header_hash, reask = excluded.reask, asks = excluded.asks, \
650            next_ask_at = excluded.next_ask_at, reannounce_beef = excluded.reannounce_beef, \
651            reannounce_at = excluded.reannounce_at, on_accept = excluded.on_accept, \
652            chain = excluded.chain, updated_at = excluded.updated_at",
653    )
654    .bind(&row.txid)
655    .bind(row.built as i64)
656    .bind(serde_json::to_string(&row.hints)?)
657    .bind(&row.bump)
658    .bind(word_label(state.word()))
659    .bind(height)
660    .bind(header_hash)
661    .bind(state.reask().map(reask_label))
662    .bind(row.asks as i64)
663    .bind(row.next_ask_at.map(|t| t as i64))
664    .bind(&row.reannounce_beef)
665    .bind(row.reannounce_at.map(|t| t as i64))
666    .bind(&row.on_accept)
667    .bind(serde_json::to_string(&row.chain)?)
668    .bind(now as i64)
669    .execute(executor)
670    .await?;
671    Ok(())
672}
673
674/// What the storage holds for `txid`: the word, the mined height and the
675/// pending re-ask the tracker gave at its last step. A view for a person;
676/// nothing is decided from it.
677pub async fn stored_word(
678    storage: &StorageSqlx,
679    txid: &str,
680) -> Result<Option<(String, Option<Height>, Option<String>)>> {
681    ensure_tables(storage).await?;
682    let row = sqlx::query("SELECT word, height, reask FROM tracker_states WHERE txid = ?")
683        .bind(txid.to_ascii_lowercase())
684        .fetch_optional(storage.pool())
685        .await?;
686    Ok(row.map(|r| {
687        (
688            r.get("word"),
689            r.get::<Option<i64>, _>("height").map(|h| h as Height),
690            r.get("reask"),
691        )
692    }))
693}
694
695// ---------------------------------------------------------------------------
696// What a broadcaster answered to a transaction the host just handed over
697// ---------------------------------------------------------------------------
698
699/// The broadcasters' answer to one post, read as a hint.
700#[derive(Debug, Clone, PartialEq, Eq)]
701pub enum PostAnswer {
702    /// A broadcaster took it.
703    Accepted { provider: String },
704    /// None took it and one called it not final: its lock time has not
705    /// passed for the node behind that broadcaster. Transient.
706    NotFinal { provider: String, said: String },
707    /// None took it, for some other reason (or none answered).
708    Refused { said: String },
709}
710
711/// Does a broadcaster's refusal say "not final"? The node's word is
712/// `non-final`; ARC and Arcade carry it under code 476, which the toolbox
713/// reads as retryable (bsv-wallet-toolbox-rs `ARCADE_CODE_NON_FINAL`).
714pub fn says_not_final(text: &str) -> bool {
715    let lower = text.to_ascii_lowercase();
716    ["non-final", "nonfinal", "non final", "not final"]
717        .iter()
718        .any(|word| lower.contains(word))
719        || lower
720            .split(|c: char| !c.is_ascii_alphanumeric())
721            .any(|token| token == "476")
722}
723
724/// Read the results of one `post_beef`.
725pub fn read_post(results: &[PostBeefResult]) -> PostAnswer {
726    if let Some(accepted) = results.iter().find(|r| r.is_success()) {
727        return PostAnswer::Accepted {
728            provider: accepted.name.clone(),
729        };
730    }
731    let said: Vec<(String, String)> = results
732        .iter()
733        .map(|r| {
734            let detail = r
735                .error
736                .clone()
737                .or_else(|| r.txid_results.iter().find_map(|t| t.data.clone()))
738                .unwrap_or_else(|| r.status.clone());
739            (r.name.clone(), detail)
740        })
741        .collect();
742    if let Some((provider, detail)) = said.iter().find(|(_, detail)| says_not_final(detail)) {
743        return PostAnswer::NotFinal {
744            provider: provider.clone(),
745            said: detail.clone(),
746        };
747    }
748    PostAnswer::Refused {
749        said: if said.is_empty() {
750            "none answered".to_string()
751        } else {
752            said.iter()
753                .map(|(name, detail)| format!("{name}: {detail}"))
754                .collect::<Vec<_>>()
755                .join("; ")
756        },
757    }
758}
759
760/// A BEEF to hand to the broadcasters again, and when.
761#[derive(Debug, Clone)]
762pub struct Reannounce {
763    /// The BEEF, as it was posted.
764    pub beef: Vec<u8>,
765    /// Not before this time (seconds since the epoch).
766    pub at: Timestamp,
767    /// What the wallet records when a broadcaster takes it then.
768    pub on_accept: Option<InternalizeActionArgs>,
769}
770
771/// Track a transaction this wallet built and just handed to a broadcaster,
772/// and feed what the broadcaster said, as a hint. `not_before` holds every
773/// re-ask until that time (a transaction that cannot be mined before its
774/// lock time has no proof to ask for before it); `reannounce` keeps the
775/// BEEF to hand over again then. Returns the tracker's word and its
776/// pending re-ask.
777pub async fn heard<C: Clock>(
778    storage: &StorageSqlx,
779    clock: &C,
780    opts: &TickOptions,
781    txid: &str,
782    hint: Hint,
783    not_before: Option<Timestamp>,
784    reannounce: Option<Reannounce>,
785) -> Result<(String, Option<String>)> {
786    ensure_tables(storage).await?;
787    let mut row = match load_row(storage, txid).await? {
788        // A row with a stored path is past this: its word is read, never
789        // rewritten from a hint.
790        Some(row) if row.bump.is_some() => {
791            let stored = stored_word(storage, txid).await?.unwrap_or_default();
792            return Ok((stored.0, stored.2));
793        }
794        Some(row) => row,
795        None => TrackedRow::new(txid),
796    };
797    let params = opts.params();
798    let headers = HeaderSnapshot::unread();
799    let (mut state, _) = row.replay(&params, &headers);
800    if !row.hints.iter().any(|h| h.source == hint.source) {
801        row.hints.push(StoredHint::of(&hint));
802        let _ = state.step(&params, &headers, Input::Hint(hint));
803    }
804    row.asks = 0;
805    row.next_ask_at = not_before;
806    match reannounce {
807        Some(again) => {
808            row.reannounce_beef = Some(again.beef);
809            row.reannounce_at = Some(again.at);
810            row.on_accept = match again.on_accept {
811                Some(args) => Some(serde_json::to_string(&args)?),
812                None => None,
813            };
814        }
815        None => {
816            row.reannounce_beef = None;
817            row.reannounce_at = None;
818            row.on_accept = None;
819        }
820    }
821    store_row(storage, &row, &state, clock.now()).await?;
822    Ok((
823        word_label(state.word()).to_string(),
824        state.reask().map(reask_label),
825    ))
826}
827
828// ---------------------------------------------------------------------------
829// The pass
830// ---------------------------------------------------------------------------
831
832/// The host's parameters of a pass.
833#[derive(Debug, Clone)]
834pub struct TickOptions {
835    /// Seconds without a new hint before an announced or seen transaction
836    /// is asked for its proof (the tracker's `Params::age_threshold`).
837    pub age_threshold: Timestamp,
838    /// Re-asks one pass runs.
839    pub max_asks: usize,
840    /// How deep the block holding a coin's transaction must be before the
841    /// coin is offered (`SPEND_DEPTH`): the tip is depth 1. The stack's
842    /// wallet charter R3 (b), ruled 2026-10-12 (bsv-stack-lean
843    /// `lean/Wallet.lean`, `guardDepth`): the spend depth is the host's
844    /// parameter, named in its config, with 1 the floor; a number below 1
845    /// reads as 1. Read by [`spend_guard`], where the host decides whether
846    /// a coin may be spent.
847    pub spend_depth: Height,
848}
849
850impl TickOptions {
851    /// `TRACKER_AGE_SECS`, `TRACKER_MAX_ASKS` and `SPEND_DEPTH`, with their
852    /// defaults.
853    pub fn from_env() -> Self {
854        fn env<T: std::str::FromStr>(key: &str, default: T) -> T {
855            std::env::var(key)
856                .ok()
857                .and_then(|v| v.trim().parse().ok())
858                .unwrap_or(default)
859        }
860        Self {
861            age_threshold: env("TRACKER_AGE_SECS", DEFAULT_AGE_SECS).max(1),
862            max_asks: env("TRACKER_MAX_ASKS", DEFAULT_MAX_ASKS),
863            spend_depth: env("SPEND_DEPTH", DEFAULT_SPEND_DEPTH),
864        }
865        .floored()
866    }
867
868    /// The options with the spend depth at its floor of 1 or above (R3).
869    #[must_use]
870    pub fn floored(mut self) -> Self {
871        self.spend_depth = self.spend_depth.max(DEFAULT_SPEND_DEPTH);
872        self
873    }
874
875    fn params(&self) -> Params {
876        Params {
877            age_threshold: self.age_threshold,
878        }
879    }
880}
881
882/// What one pass did.
883#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
884pub struct TickReport {
885    /// Rows read this pass (never the mined ones, unless a header moved).
886    pub tracked: usize,
887    /// The wallet's own unproven transactions tracked for the first time.
888    pub adopted: usize,
889    /// New hints fed.
890    pub hints: usize,
891    /// Re-asks run: one proof request each.
892    pub asked: usize,
893    /// Transactions whose proof checked this pass: the word is `mined`.
894    pub mined: Vec<String>,
895    /// Re-asks that brought no proof; each is asked again after its pause.
896    pub no_proof: Vec<String>,
897    /// Mined rows whose stored path no longer meets the active header.
898    pub moved: Vec<String>,
899    /// The lowest height whose header moved since the last pass, if any.
900    pub moved_from: Option<Height>,
901    /// Transactions handed to the broadcasters again and accepted.
902    pub reannounced: Vec<String>,
903    /// Transactions handed over again and still called not final; each is
904    /// handed over again after a pause.
905    pub not_final: Vec<String>,
906    /// What the wallet is to record for each transaction accepted on a
907    /// re-announce (`txid`, the `internalizeAction` arguments): the caller
908    /// that holds the wallet runs [`record_accepted`].
909    #[serde(skip)]
910    pub on_accept: Vec<(String, String)>,
911    /// Faults the host reports: `txid: what failed`.
912    pub faults: Vec<String>,
913    /// What the wallet's storage did with each checked proof.
914    pub stored: Vec<String>,
915}
916
917impl TickReport {
918    /// Nothing moved and nothing was asked.
919    pub fn is_quiet(&self) -> bool {
920        self.adopted == 0
921            && self.hints == 0
922            && self.asked == 0
923            && self.reannounced.is_empty()
924            && self.not_final.is_empty()
925            && self.moved.is_empty()
926            && self.faults.is_empty()
927    }
928
929    /// One line.
930    pub fn summary(&self) -> String {
931        format!(
932            "tracker tick: {} tracked ({} new), {} hint(s), {} announced again ({} still not final), {} re-ask(s): {} mined, {} without a proof, {} moved, {} fault(s)",
933            self.tracked,
934            self.adopted,
935            self.hints,
936            self.reannounced.len(),
937            self.not_final.len(),
938            self.asked,
939            self.mined.len(),
940            self.no_proof.len(),
941            self.moved.len(),
942            self.faults.len(),
943        )
944    }
945}
946
947/// The recent tips, newest last, read from and written to `tracker_meta`.
948async fn read_ring(storage: &StorageSqlx) -> Vec<(Height, String)> {
949    sqlx::query("SELECT value FROM tracker_meta WHERE key = 'tips'")
950        .fetch_optional(storage.pool())
951        .await
952        .ok()
953        .flatten()
954        .and_then(|r| serde_json::from_str(&r.get::<String, _>("value")).ok())
955        .unwrap_or_default()
956}
957
958async fn write_ring(storage: &StorageSqlx, ring: &[(Height, String)]) -> Result<()> {
959    sqlx::query(
960        "INSERT INTO tracker_meta (key, value) VALUES ('tips', ?) \
961         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
962    )
963    .bind(serde_json::to_string(ring)?)
964    .execute(storage.pool())
965    .await?;
966    Ok(())
967}
968
969/// The lowest height whose header is no longer the one a past pass saw,
970/// and the ring with the moved entries dropped and the tip appended. One
971/// header read when the tip advanced, none when it did not, and one more
972/// per replaced block. Deeper than the ring, the ring's lowest height is
973/// the answer and the host says so.
974async fn moved_since<V: WalletServices + ?Sized>(
975    services: &V,
976    snapshot: &mut HeaderSnapshot,
977    mut ring: Vec<(Height, String)>,
978) -> (Option<Height>, Vec<(Height, String)>) {
979    let Some((tip_height, tip_hash)) = snapshot.tip().map(|(h, hash)| (h, hash.to_string())) else {
980        return (None, ring);
981    };
982    let mut moved = None;
983    while let Some((height, hash)) = ring.last().cloned() {
984        let active = if height == tip_height {
985            Some(tip_hash.clone())
986        } else if height > tip_height {
987            None
988        } else {
989            snapshot.extend(services, [height]).await;
990            match snapshot.header_at(height) {
991                Ok(Some(header)) => Some(header.hash),
992                // Could not look: nothing is concluded this pass.
993                _ => return (moved, ring),
994            }
995        };
996        if active.as_deref() == Some(hash.as_str()) {
997            break;
998        }
999        moved = Some(height);
1000        ring.pop();
1001    }
1002    if ring.last().map(|(h, _)| *h) != Some(tip_height) {
1003        ring.push((tip_height, tip_hash));
1004    }
1005    if ring.len() > TIP_RING {
1006        let extra = ring.len() - TIP_RING;
1007        ring.drain(..extra);
1008    }
1009    (moved, ring)
1010}
1011
1012/// One pass of the host. See the module docs.
1013pub async fn tick<V: WalletServices + ?Sized, C: Clock>(
1014    storage: &StorageSqlx,
1015    services: &V,
1016    clock: &C,
1017    opts: &TickOptions,
1018) -> Result<TickReport> {
1019    ensure_tables(storage).await?;
1020    let params = opts.params();
1021    let now = clock.now();
1022    let mut report = TickReport::default();
1023
1024    // The wallet's own transactions that are out and unproven, tracked
1025    // from the first pass that sees them. Never anyone else's.
1026    let adopted = sqlx::query(
1027        "INSERT OR IGNORE INTO tracker_states (txid, word, updated_at) \
1028         SELECT DISTINCT lower(txid), 'built', ? FROM transactions \
1029         WHERE txid IS NOT NULL AND status IN ('unproven', 'sending')",
1030    )
1031    .bind(now as i64)
1032    .execute(storage.pool())
1033    .await?;
1034    report.adopted = adopted.rows_affected() as usize;
1035
1036    // Did a header we stood on move? Then the mined rows at or above it
1037    // are read again, and no other mined row is.
1038    let mut snapshot = HeaderSnapshot::take(services, []).await;
1039    let (moved_from, ring) = moved_since(services, &mut snapshot, read_ring(storage).await).await;
1040    report.moved_from = moved_from;
1041
1042    let mut rows: Vec<TrackedRow> = sqlx::query(&format!(
1043        "SELECT {ROW_COLUMNS} FROM tracker_states \
1044         WHERE word IN ('unknown', 'built', 'announced', 'seen', 'stale') \
1045            OR (word = 'mined' AND ? IS NOT NULL AND height >= ?) \
1046         ORDER BY txid"
1047    ))
1048    .bind(moved_from.map(|h| h as i64))
1049    .bind(moved_from.map(|h| h as i64))
1050    .fetch_all(storage.pool())
1051    .await?
1052    .iter()
1053    .map(row_of)
1054    .collect();
1055    report.tracked = rows.len();
1056
1057    // Read: the headers the stored paths name, and the new hints.
1058    snapshot
1059        .extend(services, rows.iter().filter_map(TrackedRow::proof_height))
1060        .await;
1061    let mut hints = MemoryHints::read(storage, &rows).await?;
1062
1063    // Step: replay each row, feed its hints, tick its clock.
1064    let mut states: HashMap<TxId, State> = HashMap::new();
1065    for row in rows.iter_mut() {
1066        let (state, fault) = row.replay(&params, &snapshot);
1067        if let Some(fault) = fault {
1068            // The stored path does not check against the active header:
1069            // the row is back at its hint-tier word with a re-ask, due now.
1070            if matches!(fault, CheckError::RootMismatch(_)) {
1071                // The path proves inclusion in a block that is no longer
1072                // active: it is dropped, and a new one is asked for.
1073                row.bump = None;
1074                row.chain.clear();
1075                report.moved.push(row.txid.clone());
1076                tracing::warn!(
1077                    marker = "tracker_stored_proof_moved",
1078                    txid = %row.txid,
1079                    "a stored merkle path no longer meets the active header; asking again"
1080                );
1081            } else {
1082                report.faults.push(format!("{}: {fault}", row.txid));
1083            }
1084            row.asks = 0;
1085            row.next_ask_at = None;
1086        }
1087        states.insert(row.txid.clone(), state);
1088    }
1089    while let Ok(Some((txid, hint))) = hints.next_hint() {
1090        let (Some(state), Some(row)) = (
1091            states.get_mut(&txid),
1092            rows.iter_mut().find(|r| r.txid == txid),
1093        ) else {
1094            continue;
1095        };
1096        row.hints.push(StoredHint::of(&hint));
1097        let _ = state.step(&params, &snapshot, Input::Hint(hint));
1098        report.hints += 1;
1099        // A word that disagrees is asked about now, not after the pause
1100        // an earlier re-ask earned.
1101        if matches!(state.reask(), Some(Reask::Hint(_))) {
1102            row.asks = 0;
1103            row.next_ask_at = None;
1104        }
1105    }
1106    // Announce again what a broadcaster called not final, at its time.
1107    // The lock time passing on our clock is not the node's rule passing
1108    // (the node reads the median time past, about an hour behind; the rule
1109    // is the sibling's, bsv-script-lean `docs/BLOCK-CONSENSUS.md` section
1110    // 2.2), so "not final" again is expected and waits one more pause.
1111    for row in rows.iter_mut() {
1112        let (Some(beef), Some(at)) = (row.reannounce_beef.clone(), row.reannounce_at) else {
1113            continue;
1114        };
1115        let Some(state) = states.get_mut(&row.txid) else {
1116            continue;
1117        };
1118        if at > now || matches!(state.word(), Word::Mined(_)) {
1119            continue;
1120        }
1121        let pause = Some(now.saturating_add(opts.age_threshold));
1122        let answer = services
1123            .post_beef(&beef, std::slice::from_ref(&row.txid))
1124            .await
1125            .map(|results| read_post(&results));
1126        match answer {
1127            Ok(PostAnswer::Accepted { provider }) => {
1128                let hint = Hint::new(format!("{provider}|accepted"), HintStatus::Accepted, now);
1129                row.hints.push(StoredHint::of(&hint));
1130                let _ = state.step(&params, &snapshot, Input::Hint(hint));
1131                row.reannounce_beef = None;
1132                row.reannounce_at = None;
1133                row.asks = 0;
1134                row.next_ask_at = pause;
1135                report.reannounced.push(row.txid.clone());
1136                if let Some(args) = row.on_accept.take() {
1137                    report.on_accept.push((row.txid.clone(), args));
1138                }
1139            }
1140            Ok(PostAnswer::NotFinal { .. }) => {
1141                row.asks = row.asks.saturating_add(1);
1142                row.next_ask_at = pause;
1143                if row.asks >= MAX_ASKS_PER_TX {
1144                    row.reannounce_at = None;
1145                    report.faults.push(format!(
1146                        "{}: still not final after {MAX_ASKS_PER_TX} announces; not announced again",
1147                        row.txid
1148                    ));
1149                } else {
1150                    row.reannounce_at = pause;
1151                    report.not_final.push(row.txid.clone());
1152                }
1153            }
1154            Ok(PostAnswer::Refused { said }) => {
1155                row.reannounce_at = None;
1156                row.next_ask_at = pause;
1157                report.faults.push(format!(
1158                    "{}: refused when announced again: {said}",
1159                    row.txid
1160                ));
1161            }
1162            Err(e) => {
1163                // Could not look: the same time again, one pause on.
1164                row.reannounce_at = pause;
1165                report
1166                    .faults
1167                    .push(format!("{}: could not announce again: {e}", row.txid));
1168            }
1169        }
1170    }
1171    for state in states.values_mut() {
1172        let _ = state.step(&params, &snapshot, Input::Host(HostAction::Tick(now)));
1173    }
1174
1175    // Read again: a proof for each re-ask that is due, and its header.
1176    let due: Vec<TxId> = rows
1177        .iter()
1178        .filter(|row| {
1179            states[&row.txid].reask().is_some()
1180                && row.asks < MAX_ASKS_PER_TX
1181                && row.next_ask_at.is_none_or(|at| at <= now)
1182        })
1183        .take(opts.max_asks)
1184        .map(|row| row.txid.clone())
1185        .collect();
1186    let mut courier = Courier::carry(services, &due).await;
1187    snapshot.extend(services, courier.heights()).await;
1188
1189    // Step again: each proof through the tracker's check.
1190    for txid in &due {
1191        let (Some(state), Some(row)) = (
1192            states.get_mut(txid),
1193            rows.iter_mut().find(|r| &r.txid == txid),
1194        ) else {
1195            continue;
1196        };
1197        report.asked += 1;
1198        let reask = state.reask().cloned().unwrap_or(Reask::Age);
1199        let fetched = courier.fetch(txid, &reask);
1200        let proven = match fetched {
1201            Ok(Some(proof)) => {
1202                let hex = proof.path().to_hex();
1203                match state.step(&params, &snapshot, Input::Evidence(Evidence::Proof(proof))) {
1204                    Ok(()) => {
1205                        row.bump = Some(hex);
1206                        row.chain.clear();
1207                        true
1208                    }
1209                    Err(fault) => {
1210                        report.faults.push(format!("{txid}: {fault}"));
1211                        false
1212                    }
1213                }
1214            }
1215            Ok(None) => {
1216                report.no_proof.push(txid.clone());
1217                false
1218            }
1219            Err(fault) => {
1220                report
1221                    .faults
1222                    .push(format!("{txid}: could not ask for a proof: {fault}"));
1223                false
1224            }
1225        };
1226        if proven {
1227            row.asks = 0;
1228            row.next_ask_at = None;
1229            report.mined.push(txid.clone());
1230        } else {
1231            let pause = opts
1232                .age_threshold
1233                .saturating_mul(1 << row.asks.min(MAX_BACKOFF_DOUBLINGS));
1234            row.asks = row.asks.saturating_add(1);
1235            row.next_ask_at = Some(now.saturating_add(pause));
1236        }
1237    }
1238
1239    // Keep the state, and hand each checked proof to the wallet's own
1240    // storage through its one proof funnel.
1241    for row in &rows {
1242        let state = &states[&row.txid];
1243        store_row(storage, row, state, now).await?;
1244        if let (true, Word::Mined(mined)) = (report.mined.contains(&row.txid), state.word()) {
1245            let checked = mined.checked();
1246            let outcome = storage
1247                .ingest_merkle_proof(
1248                    &row.txid,
1249                    &checked.proof().path().to_binary(),
1250                    checked.height(),
1251                    &checked.header().hash,
1252                    Some(checked.root()),
1253                )
1254                .await;
1255            report.stored.push(match outcome {
1256                Ok(outcome) => format!("{}: {outcome:?}", row.txid),
1257                Err(e) => format!("{}: not stored: {e}", row.txid),
1258            });
1259        }
1260    }
1261    write_ring(storage, &ring).await?;
1262    Ok(report)
1263}
1264
1265/// Record in the wallet what a pass's re-announces got accepted (E4): the
1266/// wallet's storage is the verdict for its own actions, so a claim the
1267/// wallet made is internalized from the BEEF it already holds, and nothing
1268/// scans the chain for it. Each outcome is added to `report.stored`.
1269pub async fn record_accepted<W: WalletInterface>(wallet: &W, report: &mut TickReport) {
1270    for (txid, args) in std::mem::take(&mut report.on_accept) {
1271        let line = match serde_json::from_str::<InternalizeActionArgs>(&args) {
1272            Ok(args) => match wallet.internalize_action(args, "bsv-wallet-cli").await {
1273                Ok(result) if result.accepted => format!("{txid}: recorded in the wallet"),
1274                Ok(_) => format!("{txid}: the wallet did not accept its own transaction"),
1275                Err(e) => format!(
1276                    "{txid}: not recorded in the wallet ({e}); `bsv-wallet receive {txid}` records it once it is mined"
1277                ),
1278            },
1279            Err(e) => format!("{txid}: not recorded in the wallet (stored arguments: {e})"),
1280        };
1281        if !line.ends_with("recorded in the wallet") {
1282            tracing::warn!(marker = "own_transaction_not_recorded", "{line}");
1283        }
1284        report.stored.push(line);
1285    }
1286}
1287
1288// ---------------------------------------------------------------------------
1289// The spend guard
1290// ---------------------------------------------------------------------------
1291
1292/// The wallet spend guard, as the tracker's README gives it (bsv-tracker
1293/// `README.md`, "Wallet: spend guard"), unchanged: the scheduled recheck
1294/// runs before the coin is used, and an unavailable header is an error that
1295/// prevents the spend.
1296fn readme_spend_guard<H: Headers>(
1297    state: &mut State,
1298    params: &Params,
1299    headers: &H,
1300) -> std::result::Result<bool, CheckError> {
1301    state.step(params, headers, Input::Host(HostAction::SpendAttempt))?;
1302    state.step(params, headers, Input::Evidence(Evidence::Recheck))?;
1303    Ok(matches!(state.word(), Word::Mined(_)) && !state.suspect())
1304}
1305
1306/// Run the spend guard for a coin of the tracked transaction `txid`, as
1307/// the host: the row is loaded (its stored path checked against a fresh
1308/// header), the README's guard runs on it, and the row is stored again.
1309/// `Ok(true)` only for a checked, non-suspect `mined`; `Err` when the
1310/// header service could not answer. A transaction the host does not track
1311/// is not mined.
1312pub async fn spend_guard<V: WalletServices + ?Sized, C: Clock>(
1313    storage: &StorageSqlx,
1314    services: &V,
1315    clock: &C,
1316    opts: &TickOptions,
1317    txid: &str,
1318) -> Result<std::result::Result<bool, CheckError>> {
1319    ensure_tables(storage).await?;
1320    let Some(row) = load_row(storage, txid).await? else {
1321        return Ok(Ok(false));
1322    };
1323    let params = opts.params();
1324    let snapshot = HeaderSnapshot::take(services, row.proof_height()).await;
1325    let (mut state, fault) = row.replay(&params, &snapshot);
1326    match fault {
1327        // Could not look (no tip, no header at the height, a path that
1328        // does not parse): an error, no spend, and the row is left as it
1329        // was. "Could not look" changes no word.
1330        Some(fault) if !matches!(fault, CheckError::RootMismatch(_)) => return Ok(Err(fault)),
1331        _ => {}
1332    }
1333    let verdict = readme_spend_guard(&mut state, &params, &snapshot)
1334        .and_then(|offered| Ok(offered && deep_enough(&state, &snapshot, opts.spend_depth)?));
1335    store_row(storage, &row, &state, clock.now()).await?;
1336    Ok(verdict)
1337}
1338
1339/// The guard's reading of one coin's creator, for the coin gate
1340/// ([`crate::spend_gate`]).
1341#[derive(Debug, Clone, PartialEq, Eq)]
1342pub struct CreatorReading {
1343    /// The tracker's word after the guard's recheck, as its label.
1344    pub word: &'static str,
1345    /// How deep the creator's block is under the tip (the tip is 1); 0 for
1346    /// a word with no block.
1347    pub depth: u64,
1348    /// A fork in progress at or below its height marks it suspect.
1349    pub suspect: bool,
1350    /// `mined`, checked against the headers of the moment, not suspect, at
1351    /// the spend depth (floor 1): the guard offers it with no rule.
1352    pub offered: bool,
1353    /// Could not look (no tip, no header at the height): the fault. Nothing
1354    /// is offered on it.
1355    pub fault: Option<String>,
1356}
1357
1358/// Run the spend guard over many creators against one read of the headers:
1359/// the tip once, and one header per height a stored path names. Each
1360/// creator is the tracked row when the tracker holds one, else the
1361/// toolbox's stored merkle path (`proven_txs`, given here as its bytes)
1362/// replayed as a row's input: a stored proof is checked again against the
1363/// headers of the moment, never trusted by age. A creator with neither
1364/// reads `unknown`. Nothing is written: the tracker's rows are not stored
1365/// back and no row is adopted.
1366pub async fn read_creators<V: WalletServices + ?Sized>(
1367    storage: &StorageSqlx,
1368    services: &V,
1369    opts: &TickOptions,
1370    creators: &[(String, Option<Vec<u8>>)],
1371) -> Result<HashMap<String, CreatorReading>> {
1372    ensure_tables(storage).await?;
1373    let params = opts.params();
1374    let mut rows = Vec::with_capacity(creators.len());
1375    for (txid, stored_path) in creators {
1376        let mut row = match load_row(storage, txid).await? {
1377            Some(row) => row,
1378            None => TrackedRow {
1379                built: false,
1380                ..TrackedRow::new(txid)
1381            },
1382        };
1383        if row.bump.is_none() {
1384            row.bump = stored_path
1385                .as_deref()
1386                .and_then(|bytes| MerklePath::from_binary(bytes).ok())
1387                .map(|path| path.to_hex());
1388        }
1389        rows.push(row);
1390    }
1391    let snapshot =
1392        HeaderSnapshot::take(services, rows.iter().filter_map(TrackedRow::proof_height)).await;
1393    let mut readings = HashMap::with_capacity(rows.len());
1394    for row in rows {
1395        let (mut state, fault) = row.replay(&params, &snapshot);
1396        let fault = match fault {
1397            Some(fault) if !matches!(fault, CheckError::RootMismatch(_)) => Some(fault.to_string()),
1398            _ => None,
1399        };
1400        let verdict = match fault {
1401            Some(_) => Ok(false),
1402            None => readme_spend_guard(&mut state, &params, &snapshot).and_then(|offered| {
1403                Ok(offered && deep_enough(&state, &snapshot, opts.spend_depth)?)
1404            }),
1405        };
1406        let depth = match (state.word().height(), snapshot.tip()) {
1407            (Some(height), Some((tip, _))) => {
1408                (u64::from(tip) + 1).saturating_sub(u64::from(height))
1409            }
1410            _ => 0,
1411        };
1412        let (offered, fault) = match verdict {
1413            Ok(offered) => (offered, fault),
1414            Err(e) => (false, fault.or(Some(e.to_string()))),
1415        };
1416        readings.insert(
1417            row.txid.clone(),
1418            CreatorReading {
1419                word: word_label(state.word()),
1420                depth,
1421                suspect: state.suspect(),
1422                offered,
1423                fault,
1424            },
1425        );
1426    }
1427    Ok(readings)
1428}
1429
1430/// Is the block holding a `mined` word's transaction at least
1431/// `spend_depth` deep under `headers`' tip (the tip is depth 1; the stack's
1432/// `Tracker.depth hd h = hd.tip + 1 - h`)? The depth reads as 1 at least
1433/// (R3's floor). An unreadable tip is an error, as for the guard.
1434fn deep_enough<H: Headers>(
1435    state: &State,
1436    headers: &H,
1437    spend_depth: Height,
1438) -> std::result::Result<bool, CheckError> {
1439    let Some(height) = state.word().height() else {
1440        return Ok(false);
1441    };
1442    let tip = headers.tip_height().map_err(CheckError::Headers)?;
1443    let depth = (u64::from(tip) + 1).saturating_sub(u64::from(height));
1444    Ok(depth >= u64::from(spend_depth.max(DEFAULT_SPEND_DEPTH)))
1445}
1446
1447// ---------------------------------------------------------------------------
1448// Chain events from the header service's feed
1449// ---------------------------------------------------------------------------
1450
1451/// The feed's stored cursor: the last envelope delivered, `0` at birth.
1452/// One row of `tracker_meta`, beside the tracker's rows.
1453pub async fn feed_cursor(storage: &StorageSqlx) -> Result<u64> {
1454    ensure_tables(storage).await?;
1455    let value: Option<String> =
1456        sqlx::query_scalar("SELECT value FROM tracker_meta WHERE key = 'feed_cursor'")
1457            .fetch_optional(storage.pool())
1458            .await?;
1459    Ok(match value {
1460        Some(v) => v
1461            .parse()
1462            .map_err(|e| anyhow::anyhow!("the stored feed cursor {v:?}: {e}"))?,
1463        None => 0,
1464    })
1465}
1466
1467/// The frozen record: an outpoint the feed's `frozen` event named, with
1468/// the envelope's cursor and when it was delivered. The spend gate reads it
1469/// and never offers such a coin (R6: nothing here clears a row).
1470async fn ensure_frozen(pool: &sqlx::SqlitePool) -> Result<()> {
1471    sqlx::query(
1472        "CREATE TABLE IF NOT EXISTS frozen_outpoints (\
1473            txid TEXT NOT NULL, \
1474            vout INTEGER NOT NULL, \
1475            cursor INTEGER NOT NULL, \
1476            frozen_at INTEGER NOT NULL, \
1477            PRIMARY KEY (txid, vout))",
1478    )
1479    .execute(pool)
1480    .await?;
1481    Ok(())
1482}
1483
1484/// Every frozen outpoint: (lowercase txid, vout).
1485pub async fn frozen_outpoints(
1486    pool: &sqlx::SqlitePool,
1487) -> Result<std::collections::HashSet<(String, i64)>> {
1488    ensure_frozen(pool).await?;
1489    let rows = sqlx::query("SELECT txid, vout FROM frozen_outpoints")
1490        .fetch_all(pool)
1491        .await?;
1492    Ok(rows
1493        .iter()
1494        .map(|r| (r.get::<String, _>("txid"), r.get::<i64, _>("vout")))
1495        .collect())
1496}
1497
1498/// How many outpoints the frozen record holds.
1499pub async fn frozen_count(pool: &sqlx::SqlitePool) -> Result<u64> {
1500    ensure_frozen(pool).await?;
1501    let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM frozen_outpoints")
1502        .fetch_one(pool)
1503        .await?;
1504    Ok(n.max(0) as u64)
1505}
1506
1507/// What one delivered envelope did to one tracked transaction.
1508#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1509pub struct ChainMove {
1510    /// The transaction the event named, by index.
1511    pub txid: String,
1512    /// The tracker's word after the event.
1513    pub word: String,
1514    /// Its pending re-ask, if any.
1515    pub reask: Option<String>,
1516    /// Marked by a fork or a split, awaiting a recheck.
1517    pub suspect: bool,
1518    /// The tracker's fault for this transaction, reported, if any.
1519    pub fault: Option<String>,
1520}
1521
1522/// One delivered envelope.
1523#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1524pub struct ChainDelivery {
1525    /// The envelope's cursor, now stored.
1526    pub cursor: u64,
1527    /// The event's kind.
1528    pub kind: &'static str,
1529    /// The transactions it named.
1530    pub moved: Vec<ChainMove>,
1531}
1532
1533/// The event's wire kind.
1534pub fn event_kind(event: &ChainEvent) -> &'static str {
1535    match event {
1536        ChainEvent::Tip { .. } => "tip",
1537        ChainEvent::Fork { .. } => "fork",
1538        ChainEvent::Reorg { .. } => "reorg",
1539        ChainEvent::Invalidated { .. } => "invalidated",
1540        ChainEvent::Frozen { .. } => "frozen",
1541        ChainEvent::TipAge { .. } => "tipAge",
1542        ChainEvent::Split { .. } => "split",
1543        ChainEvent::SplitResolved { .. } => "splitResolved",
1544    }
1545}
1546
1547/// The headers a reorg replaced, over the daemon's snapshot: a stored
1548/// proof is replayed against the header it was checked against (the
1549/// event's deactivated header at its height, the prior tip as the tip),
1550/// so the tracker sees the `mined` word the reorg takes away. The event
1551/// writes no `mined` word through this view; `Tracker::on_chain` turns
1552/// the words it names `stale`.
1553struct Replaced<'a> {
1554    snapshot: &'a HeaderSnapshot,
1555    replaced: BTreeMap<Height, Header>,
1556    prior_tip: Option<Height>,
1557}
1558
1559impl Headers for Replaced<'_> {
1560    fn header_at(&self, height: Height) -> std::result::Result<Option<Header>, HeaderError> {
1561        match self.replaced.get(&height) {
1562            Some(header) => Ok(Some(header.clone())),
1563            None => self.snapshot.header_at(height),
1564        }
1565    }
1566
1567    fn tip_height(&self) -> std::result::Result<Height, HeaderError> {
1568        match self.prior_tip {
1569            Some(tip) => Ok(tip),
1570            None => self.snapshot.tip_height(),
1571        }
1572    }
1573}
1574
1575/// Deliver one chain envelope (already decoded, at `cursor`) to the
1576/// tracked transactions it names, through `Tracker::on_chain` with the
1577/// daemon's headers, and store the moved rows and the cursor in one
1578/// transaction. A cursor at or below the stored one was delivered before
1579/// (delivery is at least once) and changes nothing.
1580///
1581/// The rows are named by index, never by a scan of the chain: a fork, a
1582/// reorg or a split names the mined rows at or above its height; a tip,
1583/// the rows an earlier event marked; an invalidated block, the rows on its
1584/// hash; a freeze or a tip age names none here (they are the spend and
1585/// freshness guards' inputs). A freeze writes its outpoint to the frozen
1586/// record ([`frozen_outpoints`]) in the same transaction as the cursor. The header service is read only when a row
1587/// is named: the tip and the heights of the named rows' proofs.
1588pub async fn deliver_chain_event<V: WalletServices + ?Sized, C: Clock>(
1589    storage: &StorageSqlx,
1590    services: &V,
1591    clock: &C,
1592    opts: &TickOptions,
1593    cursor: u64,
1594    event: ChainEvent,
1595) -> Result<ChainDelivery> {
1596    let kind = event_kind(&event);
1597    if cursor <= feed_cursor(storage).await? {
1598        return Ok(ChainDelivery {
1599            cursor,
1600            kind,
1601            moved: vec![],
1602        });
1603    }
1604    let named = format!("SELECT {ROW_COLUMNS} FROM tracker_states");
1605    let query = match &event {
1606        ChainEvent::Fork { height: from, .. }
1607        | ChainEvent::Reorg {
1608            fork_height: from, ..
1609        }
1610        | ChainEvent::Split {
1611            fork_height: from, ..
1612        }
1613        | ChainEvent::SplitResolved {
1614            fork_height: from, ..
1615        } => Some(
1616            sqlx::query(&format!(
1617                "{named} WHERE word = 'mined' AND height >= ? ORDER BY txid"
1618            ))
1619            .bind(i64::from(*from))
1620            .fetch_all(storage.pool())
1621            .await?,
1622        ),
1623        ChainEvent::Tip { .. } => Some(
1624            sqlx::query(&format!(
1625                "{named} WHERE word = 'mined' AND chain <> '[]' ORDER BY txid"
1626            ))
1627            .fetch_all(storage.pool())
1628            .await?,
1629        ),
1630        ChainEvent::Invalidated { block_hash } => Some(
1631            sqlx::query(&format!("{named} WHERE header_hash = ? ORDER BY txid"))
1632                .bind(block_hash.to_ascii_lowercase())
1633                .fetch_all(storage.pool())
1634                .await?,
1635        ),
1636        ChainEvent::Frozen { .. } | ChainEvent::TipAge { .. } => None,
1637    };
1638    let mut rows: Vec<TrackedRow> = query.iter().flatten().map(row_of).collect();
1639
1640    let now = clock.now();
1641    let mut moved = Vec::new();
1642    if matches!(event, ChainEvent::Frozen { .. }) {
1643        ensure_frozen(storage.pool()).await?;
1644    }
1645    let mut tx = storage.pool().begin().await?;
1646    if let ChainEvent::Frozen { outpoint } = &event {
1647        sqlx::query(
1648            "INSERT INTO frozen_outpoints (txid, vout, cursor, frozen_at) VALUES (?, ?, ?, ?) \
1649             ON CONFLICT(txid, vout) DO NOTHING",
1650        )
1651        .bind(outpoint.txid.to_ascii_lowercase())
1652        .bind(i64::from(outpoint.vout))
1653        .bind(cursor as i64)
1654        .bind(now as i64)
1655        .execute(&mut *tx)
1656        .await?;
1657    }
1658    if !rows.is_empty() {
1659        let snapshot =
1660            HeaderSnapshot::take(services, rows.iter().filter_map(TrackedRow::proof_height)).await;
1661        let (replaced, prior_tip) = match &event {
1662            ChainEvent::Reorg {
1663                deactivated_headers,
1664                ..
1665            } => (
1666                deactivated_headers
1667                    .iter()
1668                    .map(|h| {
1669                        (
1670                            h.height,
1671                            Header {
1672                                hash: h.hash.to_ascii_lowercase(),
1673                                merkle_root: h.merkle_root.to_ascii_lowercase(),
1674                            },
1675                        )
1676                    })
1677                    .collect(),
1678                deactivated_headers.first().map(|h| h.height),
1679            ),
1680            _ => (BTreeMap::new(), None),
1681        };
1682        let replay_view = Replaced {
1683            snapshot: &snapshot,
1684            replaced,
1685            prior_tip,
1686        };
1687        let mut tracker = Tracker::new(opts.params());
1688        for row in &rows {
1689            row.replay_into(&mut tracker, &replay_view);
1690        }
1691        for update in tracker.on_chain(&snapshot, event.clone()) {
1692            let (Some(state), Some(row)) = (
1693                tracker.get(&update.txid),
1694                rows.iter_mut().find(|r| r.txid == update.txid),
1695            ) else {
1696                continue;
1697            };
1698            row.keep_chain(state, &event)?;
1699            if matches!(update.reask, Some(Reask::Reorg { .. })) {
1700                // Asked now, not after the pause an earlier re-ask earned.
1701                row.asks = 0;
1702                row.next_ask_at = None;
1703            }
1704            store_row_in(&mut *tx, row, state, now).await?;
1705            let fault = update.result.err().map(|e| e.to_string());
1706            if let Some(fault) = &fault {
1707                tracing::warn!(txid = %update.txid, kind, cursor, %fault, "a chain event's recheck could not look");
1708            }
1709            moved.push(ChainMove {
1710                txid: update.txid,
1711                word: word_label(state.word()).to_string(),
1712                reask: state.reask().map(reask_label),
1713                suspect: state.suspect(),
1714                fault,
1715            });
1716        }
1717    }
1718    sqlx::query(
1719        "INSERT INTO tracker_meta (key, value) VALUES ('feed_cursor', ?) \
1720         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
1721    )
1722    .bind(cursor.to_string())
1723    .execute(&mut *tx)
1724    .await?;
1725    tx.commit().await?;
1726    Ok(ChainDelivery {
1727        cursor,
1728        kind,
1729        moved,
1730    })
1731}
1732
1733#[cfg(test)]
1734pub(crate) mod tests {
1735    use super::*;
1736    use bsv_wallet_toolbox::services::mock::{MockResponse, MockWalletServices};
1737    use bsv_wallet_toolbox::services::{BlockHeader, GetMerklePathResult};
1738    use bsv_wallet_toolbox::{
1739        WalletStorageWriter, BROADCAST_STATUS_ACCEPTED, BROADCAST_STATUS_MINED,
1740    };
1741    use chrono::Utc;
1742
1743    /// A clock the test sets.
1744    pub(crate) struct FixedClock(pub Timestamp);
1745    impl Clock for FixedClock {
1746        fn now(&self) -> Timestamp {
1747            self.0
1748        }
1749    }
1750
1751    pub(crate) fn opts() -> TickOptions {
1752        TickOptions {
1753            age_threshold: 600,
1754            max_asks: 20,
1755            spend_depth: DEFAULT_SPEND_DEPTH,
1756        }
1757    }
1758
1759    /// A migrated in-memory wallet with one user.
1760    pub(crate) async fn wallet_storage() -> (StorageSqlx, i64) {
1761        let storage = StorageSqlx::in_memory().await.unwrap();
1762        storage
1763            .migrate("tracker-tests", &("02".to_string() + &"ab".repeat(32)))
1764            .await
1765            .unwrap();
1766        storage.make_available().await.unwrap();
1767        let (user, _) = storage
1768            .find_or_insert_user(&("02".to_string() + &"cd".repeat(32)))
1769            .await
1770            .unwrap();
1771        (storage, user.user_id)
1772    }
1773
1774    pub(crate) async fn insert_tx(storage: &StorageSqlx, user_id: i64, txid: &str, status: &str) {
1775        sqlx::query(
1776            "INSERT INTO transactions (user_id, status, reference, is_outgoing, satoshis, version, lock_time, description, txid, raw_tx, created_at, updated_at) \
1777             VALUES (?, ?, ?, 1, 0, 1, 0, 'd', ?, X'01000000', ?, ?)",
1778        )
1779        .bind(user_id)
1780        .bind(status)
1781        .bind(&txid[..8])
1782        .bind(txid)
1783        .bind(Utc::now())
1784        .bind(Utc::now())
1785        .execute(storage.pool())
1786        .await
1787        .unwrap();
1788    }
1789
1790    /// A header at `height` whose merkle root is `root`.
1791    pub(crate) fn header(height: u32, root: &str, nonce: u32) -> BlockHeader {
1792        BlockHeader {
1793            version: 1,
1794            previous_hash: "00".repeat(32),
1795            merkle_root: root.to_string(),
1796            time: 1_700_000_000,
1797            bits: 0x1d00_ffff,
1798            nonce,
1799            hash: String::new(),
1800            height,
1801        }
1802    }
1803
1804    /// A path proving `txid` alone in a block at `height`: its root is the
1805    /// txid.
1806    pub(crate) fn path_of(txid: &str, height: u32) -> String {
1807        MerklePath::from_coinbase_txid(txid, height).to_hex()
1808    }
1809
1810    pub(crate) fn proof_answer(path: Option<String>) -> MockResponse<GetMerklePathResult> {
1811        MockResponse::Success(GetMerklePathResult {
1812            name: Some("Services".to_string()),
1813            merkle_path: path,
1814            header: None,
1815            error: None,
1816            notes: vec![],
1817        })
1818    }
1819
1820    /// E3 (the rulings of 2026-10-09). The wallet's own unproven
1821    /// transaction is tracked, a broadcaster's acceptance announces it, and
1822    /// its age asks for a proof: one request, for that transaction. The
1823    /// proof checks against the header service's header and the word is
1824    /// `mined`. The row is in the wallet's database, and the next pass
1825    /// reads no mined row and asks nothing.
1826    #[tokio::test]
1827    async fn a_pass_tracks_announces_and_proves_one_named_transaction() {
1828        let (storage, user) = wallet_storage().await;
1829        let txid = "a1".repeat(32);
1830        insert_tx(&storage, user, &txid, "unproven").await;
1831        storage
1832            .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_ACCEPTED)
1833            .await
1834            .unwrap();
1835        let services = MockWalletServices::builder()
1836            .get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
1837            .build();
1838        services.set_header_for_height(header(900, &txid, 0));
1839        let now = Utc::now().timestamp() as u64;
1840
1841        // Young: announced on the hint, nothing asked.
1842        let report = tick(&storage, &services, &FixedClock(now), &opts())
1843            .await
1844            .unwrap();
1845        assert_eq!((report.adopted, report.hints, report.asked), (1, 1, 0));
1846        assert_eq!(
1847            stored_word(&storage, &txid).await.unwrap(),
1848            Some(("announced".to_string(), None, None))
1849        );
1850        assert_eq!(services.call_count("get_merkle_path"), 0);
1851
1852        // Past the age threshold: the re-ask runs, the proof checks.
1853        let report = tick(&storage, &services, &FixedClock(now + 601), &opts())
1854            .await
1855            .unwrap();
1856        assert_eq!(report.asked, 1);
1857        assert_eq!(report.mined, vec![txid.clone()]);
1858        assert_eq!(
1859            stored_word(&storage, &txid).await.unwrap(),
1860            Some(("mined".to_string(), Some(900), None))
1861        );
1862        assert_eq!(services.call_count("get_merkle_path"), 1);
1863
1864        // A mined row is not read again while no header moved.
1865        let report = tick(&storage, &services, &FixedClock(now + 1300), &opts())
1866            .await
1867            .unwrap();
1868        assert_eq!((report.tracked, report.asked), (0, 0));
1869        assert_eq!(services.call_count("get_merkle_path"), 1);
1870    }
1871
1872    /// The evidence rule in the host: a broadcaster's "mined" is a hint. It
1873    /// asks for the proof at once; with no proof the word does not move,
1874    /// and the same transaction is not asked again until its pause is over.
1875    #[tokio::test]
1876    async fn a_broadcasters_mined_is_a_hint_that_asks_for_the_proof() {
1877        let (storage, user) = wallet_storage().await;
1878        let txid = "b2".repeat(32);
1879        insert_tx(&storage, user, &txid, "unproven").await;
1880        storage
1881            .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
1882            .await
1883            .unwrap();
1884        let services = MockWalletServices::builder()
1885            .get_merkle_path_response(proof_answer(None))
1886            .build();
1887        let now = Utc::now().timestamp() as u64;
1888
1889        let report = tick(&storage, &services, &FixedClock(now), &opts())
1890            .await
1891            .unwrap();
1892        assert_eq!(report.asked, 1);
1893        assert_eq!(report.no_proof, vec![txid.clone()]);
1894        assert!(report.mined.is_empty());
1895        let (word, height, reask) = stored_word(&storage, &txid).await.unwrap().unwrap();
1896        assert_eq!((word.as_str(), height), ("built", None));
1897        assert_eq!(reask.as_deref(), Some("hint:arcade|mined"));
1898
1899        // One minute later: the pause is not over, nothing is asked.
1900        let report = tick(&storage, &services, &FixedClock(now + 60), &opts())
1901            .await
1902            .unwrap();
1903        assert_eq!(report.asked, 0);
1904        // After the pause it is asked again, and the pause doubles.
1905        let report = tick(&storage, &services, &FixedClock(now + 600), &opts())
1906            .await
1907            .unwrap();
1908        assert_eq!(report.asked, 1);
1909        let report = tick(&storage, &services, &FixedClock(now + 1200), &opts())
1910            .await
1911            .unwrap();
1912        assert_eq!(report.asked, 0);
1913        assert_eq!(services.call_count("get_merkle_path"), 2);
1914    }
1915
1916    /// A header we stood on moved: the mined rows at or above it are read
1917    /// again and asked again, and a mined row below it is not touched.
1918    #[tokio::test]
1919    async fn a_moved_header_reasks_the_rows_at_or_above_it_and_no_other() {
1920        let (storage, user) = wallet_storage().await;
1921        let low = "c3".repeat(32);
1922        let high = "d4".repeat(32);
1923        let services = MockWalletServices::builder()
1924            .get_merkle_path_response(MockResponse::Sequence(vec![
1925                proof_answer(Some(path_of(&low, 100))),
1926                proof_answer(Some(path_of(&high, 200))),
1927                proof_answer(Some(path_of(&high, 201))),
1928            ]))
1929            .build();
1930        services.set_header_for_height(header(100, &low, 0));
1931        services.set_header_for_height(header(200, &high, 0));
1932        services.set_tip_header(header_with_hash(200, &high, 0));
1933        let now = Utc::now().timestamp() as u64;
1934        for txid in [&low, &high] {
1935            insert_tx(&storage, user, txid, "unproven").await;
1936            storage
1937                .record_broadcast_status(txid, "arcade", BROADCAST_STATUS_MINED)
1938                .await
1939                .unwrap();
1940        }
1941        let report = tick(&storage, &services, &FixedClock(now), &opts())
1942            .await
1943            .unwrap();
1944        assert_eq!(report.mined.len(), 2, "{report:?}");
1945
1946        // Height 200 is replaced by a block without `high`; it is mined
1947        // again at 201.
1948        services.set_header_for_height(header(200, &"ee".repeat(32), 7));
1949        services.set_header_for_height(header(201, &high, 0));
1950        services.set_tip_header(header_with_hash(201, &high, 0));
1951        let report = tick(&storage, &services, &FixedClock(now + 60), &opts())
1952            .await
1953            .unwrap();
1954        assert_eq!(report.moved_from, Some(200));
1955        assert_eq!(
1956            report.tracked, 1,
1957            "only the row at or above the moved header"
1958        );
1959        assert_eq!(report.moved, vec![high.clone()]);
1960        assert_eq!(report.mined, vec![high.clone()]);
1961        assert_eq!(
1962            stored_word(&storage, &high).await.unwrap().unwrap().1,
1963            Some(201)
1964        );
1965        assert_eq!(
1966            stored_word(&storage, &low).await.unwrap().unwrap().1,
1967            Some(100)
1968        );
1969    }
1970
1971    /// E2 (the rulings of 2026-10-09). A transaction a broadcaster called
1972    /// not final is held with its BEEF: nothing is asked or announced
1973    /// before its lock time; at the lock time it is announced again; "not
1974    /// final" again (the node's clock is behind ours) waits one pause; an
1975    /// acceptance announces it, and from there its age asks for the proof.
1976    #[tokio::test]
1977    async fn a_not_final_transaction_is_announced_again_at_its_lock_time() {
1978        use bsv_wallet_toolbox::services::mock::{
1979            error_post_beef_result, success_post_beef_result,
1980        };
1981        let (storage, _user) = wallet_storage().await;
1982        let txid = "c7".repeat(32);
1983        let lock = 2_000_000_000u64;
1984        let services = MockWalletServices::builder()
1985            .post_beef_response(MockResponse::Sequence(vec![
1986                MockResponse::Success(vec![error_post_beef_result("Arcade", "476 non-final")]),
1987                MockResponse::Success(vec![success_post_beef_result("Arcade", &[&txid])]),
1988            ]))
1989            .get_merkle_path_response(proof_answer(None))
1990            .build();
1991
1992        let hint = Hint::new(
1993            "Arcade|rejected",
1994            HintStatus::Rejected {
1995                reason: "476 non-final".to_string(),
1996            },
1997            lock - 3600,
1998        );
1999        let again = Reannounce {
2000            beef: vec![1, 2, 3],
2001            at: lock,
2002            on_accept: None,
2003        };
2004        let word = heard(
2005            &storage,
2006            &FixedClock(lock - 3600),
2007            &opts(),
2008            &txid,
2009            hint,
2010            Some(lock),
2011            Some(again),
2012        )
2013        .await
2014        .unwrap();
2015        assert_eq!(
2016            word,
2017            (
2018                "built".to_string(),
2019                Some("hint:Arcade|rejected".to_string())
2020            )
2021        );
2022
2023        // Before the lock time: nothing announced, no proof asked.
2024        let report = tick(&storage, &services, &FixedClock(lock - 1), &opts())
2025            .await
2026            .unwrap();
2027        assert_eq!((report.tracked, report.asked), (1, 0), "{report:?}");
2028        assert_eq!(services.call_count("post_beef"), 0);
2029
2030        // At the lock time: announced again; the node still says not final.
2031        let report = tick(&storage, &services, &FixedClock(lock), &opts())
2032            .await
2033            .unwrap();
2034        assert_eq!(report.not_final, vec![txid.clone()]);
2035        assert_eq!(services.call_count("post_beef"), 1);
2036        let report = tick(&storage, &services, &FixedClock(lock + 60), &opts())
2037            .await
2038            .unwrap();
2039        assert!(report.not_final.is_empty() && report.reannounced.is_empty());
2040        assert_eq!(services.call_count("post_beef"), 1, "one pause between two");
2041
2042        // One pause on: accepted. The word is the tracker's `announced`.
2043        let report = tick(&storage, &services, &FixedClock(lock + 600), &opts())
2044            .await
2045            .unwrap();
2046        assert_eq!(report.reannounced, vec![txid.clone()]);
2047        assert_eq!(services.call_count("post_beef"), 2);
2048        assert_eq!(
2049            stored_word(&storage, &txid).await.unwrap().unwrap().0,
2050            "announced"
2051        );
2052        assert_eq!(services.call_count("get_merkle_path"), 0);
2053
2054        // It is not announced a third time; its age asks for the proof.
2055        let report = tick(&storage, &services, &FixedClock(lock + 1300), &opts())
2056            .await
2057            .unwrap();
2058        assert_eq!(report.asked, 1);
2059        assert_eq!(services.call_count("post_beef"), 2);
2060        assert_eq!(services.call_count("get_merkle_path"), 1);
2061    }
2062
2063    /// The words that mean "not final", and one that does not.
2064    #[test]
2065    fn not_final_is_read_from_the_refusal() {
2066        for said in [
2067            "476 non-final",
2068            "bad-txns-nonfinal",
2069            "tx is not final",
2070            "code 476",
2071        ] {
2072            assert!(says_not_final(said), "{said}");
2073        }
2074        for said in [
2075            "fee too low",
2076            "DOUBLE_SPEND_ATTEMPTED",
2077            "txid 4476aa",
2078            "HTTP 500",
2079        ] {
2080            assert!(!says_not_final(said), "{said}");
2081        }
2082    }
2083
2084    /// A tip header carrying the hash its bytes have.
2085    pub(crate) fn header_with_hash(height: u32, root: &str, nonce: u32) -> BlockHeader {
2086        let mut h = header(height, root, nonce);
2087        h.hash = header_of(&h.to_binary()).unwrap().hash;
2088        h
2089    }
2090
2091    /// R3 (b): the spend depth is the host's, named in its config, 1 the
2092    /// floor. A coin whose block is the tip (one block deep) is offered at
2093    /// depth 1 and held at depth 3; two blocks later it is offered at 3; a
2094    /// depth of 0 reads as 1.
2095    #[tokio::test]
2096    async fn the_spend_depth_holds_a_coin_until_its_block_is_deep_enough() {
2097        let (storage, user) = wallet_storage().await;
2098        let txid = "e7".repeat(32);
2099        insert_tx(&storage, user, &txid, "unproven").await;
2100        storage
2101            .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
2102            .await
2103            .unwrap();
2104        let services = MockWalletServices::builder()
2105            .get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
2106            .build();
2107        services.set_header_for_height(header(900, &txid, 0));
2108        services.set_tip_header(header_with_hash(900, &txid, 0));
2109        let clock = FixedClock(Utc::now().timestamp() as u64);
2110        tick(&storage, &services, &clock, &opts()).await.unwrap();
2111        let at = |spend_depth| TickOptions {
2112            spend_depth,
2113            ..opts()
2114        };
2115
2116        let one = spend_guard(&storage, &services, &clock, &at(1), &txid)
2117            .await
2118            .unwrap();
2119        let three = spend_guard(&storage, &services, &clock, &at(3), &txid)
2120            .await
2121            .unwrap();
2122        let zero = spend_guard(&storage, &services, &clock, &at(0), &txid)
2123            .await
2124            .unwrap();
2125        println!("one block deep: depth 1 {one:?}, depth 3 {three:?}, depth 0 {zero:?}");
2126        assert_eq!(one, Ok(true));
2127        assert_eq!(three, Ok(false));
2128        assert_eq!(zero, Ok(true), "0 reads as the floor, 1");
2129
2130        services.set_tip_header(header_with_hash(902, &"0c".repeat(32), 0));
2131        let three = spend_guard(&storage, &services, &clock, &at(3), &txid)
2132            .await
2133            .unwrap();
2134        println!("three blocks deep: depth 3 {three:?}");
2135        assert_eq!(three, Ok(true));
2136        assert_eq!(
2137            TickOptions {
2138                spend_depth: 0,
2139                ..opts()
2140            }
2141            .floored()
2142            .spend_depth,
2143            1
2144        );
2145    }
2146
2147    /// The README's wallet spend guard, run by the host over a stored row
2148    /// (bsv-tracker `README.md`, "Wallet: spend guard"). Three runs: the
2149    /// header is the one the proof was checked against; the header service
2150    /// cannot answer; the header moved.
2151    #[tokio::test]
2152    async fn the_readme_spend_guard_runs_as_the_host() {
2153        let (storage, user) = wallet_storage().await;
2154        let txid = "e5".repeat(32);
2155        insert_tx(&storage, user, &txid, "unproven").await;
2156        storage
2157            .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
2158            .await
2159            .unwrap();
2160        let services = MockWalletServices::builder()
2161            .get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
2162            .build();
2163        services.set_header_for_height(header(900, &txid, 0));
2164        let clock = FixedClock(Utc::now().timestamp() as u64);
2165        tick(&storage, &services, &clock, &opts()).await.unwrap();
2166
2167        // 1. Mined, checked, not suspect: the coin may be spent.
2168        let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
2169            .await
2170            .unwrap();
2171        println!(
2172            "spend guard, header in place:   {guard:?}, stored {:?}",
2173            stored_word(&storage, &txid).await.unwrap().unwrap()
2174        );
2175        assert_eq!(guard, Ok(true));
2176
2177        // 2. No tip from the header service: an error, and no spend.
2178        services.set_tip_unavailable();
2179        let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
2180            .await
2181            .unwrap();
2182        println!(
2183            "spend guard, no header service: {guard:?}, stored {:?}",
2184            stored_word(&storage, &txid).await.unwrap().unwrap()
2185        );
2186        assert!(matches!(guard, Err(CheckError::Headers(_))), "{guard:?}");
2187        assert_eq!(
2188            stored_word(&storage, &txid).await.unwrap().unwrap().0,
2189            "mined",
2190            "could not look changes no word"
2191        );
2192
2193        // 3. The header at the height moved: not mined, a re-ask pending.
2194        services.set_tip_header(header_with_hash(901, &"00".repeat(32), 1));
2195        services.set_header_for_height(header(900, &"ee".repeat(32), 7));
2196        let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
2197            .await
2198            .unwrap();
2199        println!(
2200            "spend guard, header moved:      {guard:?}, stored {:?}",
2201            stored_word(&storage, &txid).await.unwrap().unwrap()
2202        );
2203        assert_eq!(guard, Ok(false));
2204
2205        // A transaction the host does not track is not mined.
2206        let guard = spend_guard(&storage, &services, &clock, &opts(), &"f6".repeat(32))
2207            .await
2208            .unwrap();
2209        assert_eq!(guard, Ok(false));
2210
2211        // The guard's own error path, on a state held in memory: mined,
2212        // then the header service loses the height.
2213        let params = opts().params();
2214        services.set_tip_header(header_with_hash(900, &txid, 0));
2215        services.set_header_for_height(header(900, &txid, 0));
2216        let held = HeaderSnapshot::take(&services, [900]).await;
2217        let row = load_row(&storage, &txid).await.unwrap().unwrap();
2218        let (mut state, fault) = row.replay(&params, &held);
2219        assert_eq!(fault, None);
2220        let lost = HeaderSnapshot::take(&services, []).await;
2221        let guard = readme_spend_guard(&mut state, &params, &lost);
2222        println!("spend guard, height unanswered: {guard:?}");
2223        assert_eq!(guard, Err(CheckError::Unavailable(900)));
2224    }
2225}