1use 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
77pub const DEFAULT_AGE_SECS: u64 = 600;
81pub const DEFAULT_MAX_ASKS: usize = 20;
83const MAX_BACKOFF_DOUBLINGS: u32 = 6;
86const MAX_ASKS_PER_TX: u32 = 16;
89const TIP_RING: usize = 12;
91pub const DEFAULT_SPEND_DEPTH: Height = 1;
94
95#[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 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 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 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 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
177fn 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#[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#[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 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#[derive(Debug, Default)]
282pub struct Courier {
283 carried: HashMap<TxId, std::result::Result<Option<Proof>, String>>,
284}
285
286impl Courier {
287 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#[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#[derive(Debug, Clone, PartialEq, Eq)]
390struct TrackedRow {
391 txid: TxId,
392 built: bool,
393 hints: Vec<StoredHint>,
395 bump: Option<String>,
397 asks: u32,
398 next_ask_at: Option<Timestamp>,
399 reannounce_beef: Option<Vec<u8>>,
401 reannounce_at: Option<Timestamp>,
402 on_accept: Option<String>,
405 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 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 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 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 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
521pub 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 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
674pub 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#[derive(Debug, Clone, PartialEq, Eq)]
701pub enum PostAnswer {
702 Accepted { provider: String },
704 NotFinal { provider: String, said: String },
707 Refused { said: String },
709}
710
711pub 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
724pub 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#[derive(Debug, Clone)]
762pub struct Reannounce {
763 pub beef: Vec<u8>,
765 pub at: Timestamp,
767 pub on_accept: Option<InternalizeActionArgs>,
769}
770
771pub 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 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(¶ms, &headers);
800 if !row.hints.iter().any(|h| h.source == hint.source) {
801 row.hints.push(StoredHint::of(&hint));
802 let _ = state.step(¶ms, &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#[derive(Debug, Clone)]
834pub struct TickOptions {
835 pub age_threshold: Timestamp,
838 pub max_asks: usize,
840 pub spend_depth: Height,
848}
849
850impl TickOptions {
851 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 #[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#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
884pub struct TickReport {
885 pub tracked: usize,
887 pub adopted: usize,
889 pub adopted_mined: Vec<String>,
893 pub adopted_proof_failed: Vec<String>,
896 pub hints: usize,
898 pub asked: usize,
900 pub mined: Vec<String>,
902 pub no_proof: Vec<String>,
904 pub moved: Vec<String>,
906 pub moved_from: Option<Height>,
908 pub reannounced: Vec<String>,
910 pub not_final: Vec<String>,
913 #[serde(skip)]
917 pub on_accept: Vec<(String, String)>,
918 pub faults: Vec<String>,
920 pub stored: Vec<String>,
922}
923
924impl TickReport {
925 pub fn is_quiet(&self) -> bool {
927 self.adopted == 0
928 && self.adopted_mined.is_empty()
929 && self.adopted_proof_failed.is_empty()
930 && self.hints == 0
931 && self.asked == 0
932 && self.reannounced.is_empty()
933 && self.not_final.is_empty()
934 && self.moved.is_empty()
935 && self.faults.is_empty()
936 }
937
938 pub fn summary(&self) -> String {
940 format!(
941 "tracker tick: {} tracked ({} new), {} proven creator(s) adopted ({} mined, {} proof failed), {} hint(s), {} announced again ({} still not final), {} re-ask(s): {} mined, {} without a proof, {} moved, {} fault(s)",
942 self.tracked,
943 self.adopted,
944 self.adopted_mined.len() + self.adopted_proof_failed.len(),
945 self.adopted_mined.len(),
946 self.adopted_proof_failed.len(),
947 self.hints,
948 self.reannounced.len(),
949 self.not_final.len(),
950 self.asked,
951 self.mined.len(),
952 self.no_proof.len(),
953 self.moved.len(),
954 self.faults.len(),
955 )
956 }
957}
958
959async fn read_ring(storage: &StorageSqlx) -> Vec<(Height, String)> {
961 sqlx::query("SELECT value FROM tracker_meta WHERE key = 'tips'")
962 .fetch_optional(storage.pool())
963 .await
964 .ok()
965 .flatten()
966 .and_then(|r| serde_json::from_str(&r.get::<String, _>("value")).ok())
967 .unwrap_or_default()
968}
969
970async fn write_ring(storage: &StorageSqlx, ring: &[(Height, String)]) -> Result<()> {
971 sqlx::query(
972 "INSERT INTO tracker_meta (key, value) VALUES ('tips', ?) \
973 ON CONFLICT(key) DO UPDATE SET value = excluded.value",
974 )
975 .bind(serde_json::to_string(ring)?)
976 .execute(storage.pool())
977 .await?;
978 Ok(())
979}
980
981async fn moved_since<V: WalletServices + ?Sized>(
987 services: &V,
988 snapshot: &mut HeaderSnapshot,
989 mut ring: Vec<(Height, String)>,
990) -> (Option<Height>, Vec<(Height, String)>) {
991 let Some((tip_height, tip_hash)) = snapshot.tip().map(|(h, hash)| (h, hash.to_string())) else {
992 return (None, ring);
993 };
994 let mut moved = None;
995 while let Some((height, hash)) = ring.last().cloned() {
996 let active = if height == tip_height {
997 Some(tip_hash.clone())
998 } else if height > tip_height {
999 None
1000 } else {
1001 snapshot.extend(services, [height]).await;
1002 match snapshot.header_at(height) {
1003 Ok(Some(header)) => Some(header.hash),
1004 _ => return (moved, ring),
1006 }
1007 };
1008 if active.as_deref() == Some(hash.as_str()) {
1009 break;
1010 }
1011 moved = Some(height);
1012 ring.pop();
1013 }
1014 if ring.last().map(|(h, _)| *h) != Some(tip_height) {
1015 ring.push((tip_height, tip_hash));
1016 }
1017 if ring.len() > TIP_RING {
1018 let extra = ring.len() - TIP_RING;
1019 ring.drain(..extra);
1020 }
1021 (moved, ring)
1022}
1023
1024pub async fn tick<V: WalletServices + ?Sized, C: Clock>(
1026 storage: &StorageSqlx,
1027 services: &V,
1028 clock: &C,
1029 opts: &TickOptions,
1030) -> Result<TickReport> {
1031 ensure_tables(storage).await?;
1032 let params = opts.params();
1033 let now = clock.now();
1034 let mut report = TickReport::default();
1035
1036 let adopted = sqlx::query(
1039 "INSERT OR IGNORE INTO tracker_states (txid, word, updated_at) \
1040 SELECT DISTINCT lower(txid), 'built', ? FROM transactions \
1041 WHERE txid IS NOT NULL AND status IN ('unproven', 'sending')",
1042 )
1043 .bind(now as i64)
1044 .execute(storage.pool())
1045 .await?;
1046 report.adopted = adopted.rows_affected() as usize;
1047
1048 let mut snapshot = HeaderSnapshot::take(services, []).await;
1051 let (moved_from, ring) = moved_since(services, &mut snapshot, read_ring(storage).await).await;
1052 report.moved_from = moved_from;
1053
1054 let mut rows: Vec<TrackedRow> = sqlx::query(&format!(
1055 "SELECT {ROW_COLUMNS} FROM tracker_states \
1056 WHERE word IN ('unknown', 'built', 'announced', 'seen', 'stale') \
1057 OR (word = 'mined' AND ? IS NOT NULL AND height >= ?) \
1058 ORDER BY txid"
1059 ))
1060 .bind(moved_from.map(|h| h as i64))
1061 .bind(moved_from.map(|h| h as i64))
1062 .fetch_all(storage.pool())
1063 .await?
1064 .iter()
1065 .map(row_of)
1066 .collect();
1067 report.tracked = rows.len();
1068
1069 snapshot
1071 .extend(services, rows.iter().filter_map(TrackedRow::proof_height))
1072 .await;
1073 let mut hints = MemoryHints::read(storage, &rows).await?;
1074
1075 let mut states: HashMap<TxId, State> = HashMap::new();
1077 for row in rows.iter_mut() {
1078 let (state, fault) = row.replay(¶ms, &snapshot);
1079 if let Some(fault) = fault {
1080 if matches!(fault, CheckError::RootMismatch(_)) {
1083 row.bump = None;
1086 row.chain.clear();
1087 report.moved.push(row.txid.clone());
1088 tracing::warn!(
1089 marker = "tracker_stored_proof_moved",
1090 txid = %row.txid,
1091 "a stored merkle path no longer meets the active header; asking again"
1092 );
1093 } else {
1094 report.faults.push(format!("{}: {fault}", row.txid));
1095 }
1096 row.asks = 0;
1097 row.next_ask_at = None;
1098 }
1099 states.insert(row.txid.clone(), state);
1100 }
1101 while let Ok(Some((txid, hint))) = hints.next_hint() {
1102 let (Some(state), Some(row)) = (
1103 states.get_mut(&txid),
1104 rows.iter_mut().find(|r| r.txid == txid),
1105 ) else {
1106 continue;
1107 };
1108 row.hints.push(StoredHint::of(&hint));
1109 let _ = state.step(¶ms, &snapshot, Input::Hint(hint));
1110 report.hints += 1;
1111 if matches!(state.reask(), Some(Reask::Hint(_))) {
1114 row.asks = 0;
1115 row.next_ask_at = None;
1116 }
1117 }
1118 for row in rows.iter_mut() {
1124 let (Some(beef), Some(at)) = (row.reannounce_beef.clone(), row.reannounce_at) else {
1125 continue;
1126 };
1127 let Some(state) = states.get_mut(&row.txid) else {
1128 continue;
1129 };
1130 if at > now || matches!(state.word(), Word::Mined(_)) {
1131 continue;
1132 }
1133 let pause = Some(now.saturating_add(opts.age_threshold));
1134 let answer = services
1135 .post_beef(&beef, std::slice::from_ref(&row.txid))
1136 .await
1137 .map(|results| read_post(&results));
1138 match answer {
1139 Ok(PostAnswer::Accepted { provider }) => {
1140 let hint = Hint::new(format!("{provider}|accepted"), HintStatus::Accepted, now);
1141 row.hints.push(StoredHint::of(&hint));
1142 let _ = state.step(¶ms, &snapshot, Input::Hint(hint));
1143 row.reannounce_beef = None;
1144 row.reannounce_at = None;
1145 row.asks = 0;
1146 row.next_ask_at = pause;
1147 report.reannounced.push(row.txid.clone());
1148 if let Some(args) = row.on_accept.take() {
1149 report.on_accept.push((row.txid.clone(), args));
1150 }
1151 }
1152 Ok(PostAnswer::NotFinal { .. }) => {
1153 row.asks = row.asks.saturating_add(1);
1154 row.next_ask_at = pause;
1155 if row.asks >= MAX_ASKS_PER_TX {
1156 row.reannounce_at = None;
1157 report.faults.push(format!(
1158 "{}: still not final after {MAX_ASKS_PER_TX} announces; not announced again",
1159 row.txid
1160 ));
1161 } else {
1162 row.reannounce_at = pause;
1163 report.not_final.push(row.txid.clone());
1164 }
1165 }
1166 Ok(PostAnswer::Refused { said }) => {
1167 row.reannounce_at = None;
1168 row.next_ask_at = pause;
1169 report.faults.push(format!(
1170 "{}: refused when announced again: {said}",
1171 row.txid
1172 ));
1173 }
1174 Err(e) => {
1175 row.reannounce_at = pause;
1177 report
1178 .faults
1179 .push(format!("{}: could not announce again: {e}", row.txid));
1180 }
1181 }
1182 }
1183 for state in states.values_mut() {
1184 let _ = state.step(¶ms, &snapshot, Input::Host(HostAction::Tick(now)));
1185 }
1186
1187 let due: Vec<TxId> = rows
1189 .iter()
1190 .filter(|row| {
1191 states[&row.txid].reask().is_some()
1192 && row.asks < MAX_ASKS_PER_TX
1193 && row.next_ask_at.is_none_or(|at| at <= now)
1194 })
1195 .take(opts.max_asks)
1196 .map(|row| row.txid.clone())
1197 .collect();
1198 let mut courier = Courier::carry(services, &due).await;
1199 snapshot.extend(services, courier.heights()).await;
1200
1201 for txid in &due {
1203 let (Some(state), Some(row)) = (
1204 states.get_mut(txid),
1205 rows.iter_mut().find(|r| &r.txid == txid),
1206 ) else {
1207 continue;
1208 };
1209 report.asked += 1;
1210 let reask = state.reask().cloned().unwrap_or(Reask::Age);
1211 let fetched = courier.fetch(txid, &reask);
1212 let proven = match fetched {
1213 Ok(Some(proof)) => {
1214 let hex = proof.path().to_hex();
1215 match state.step(¶ms, &snapshot, Input::Evidence(Evidence::Proof(proof))) {
1216 Ok(()) => {
1217 row.bump = Some(hex);
1218 row.chain.clear();
1219 true
1220 }
1221 Err(fault) => {
1222 report.faults.push(format!("{txid}: {fault}"));
1223 false
1224 }
1225 }
1226 }
1227 Ok(None) => {
1228 report.no_proof.push(txid.clone());
1229 false
1230 }
1231 Err(fault) => {
1232 report
1233 .faults
1234 .push(format!("{txid}: could not ask for a proof: {fault}"));
1235 false
1236 }
1237 };
1238 if proven {
1239 row.asks = 0;
1240 row.next_ask_at = None;
1241 report.mined.push(txid.clone());
1242 } else {
1243 let pause = opts
1244 .age_threshold
1245 .saturating_mul(1 << row.asks.min(MAX_BACKOFF_DOUBLINGS));
1246 row.asks = row.asks.saturating_add(1);
1247 row.next_ask_at = Some(now.saturating_add(pause));
1248 }
1249 }
1250
1251 for row in &rows {
1254 let state = &states[&row.txid];
1255 store_row(storage, row, state, now).await?;
1256 if let (true, Word::Mined(mined)) = (report.mined.contains(&row.txid), state.word()) {
1257 let checked = mined.checked();
1258 let outcome = storage
1259 .ingest_merkle_proof(
1260 &row.txid,
1261 &checked.proof().path().to_binary(),
1262 checked.height(),
1263 &checked.header().hash,
1264 Some(checked.root()),
1265 )
1266 .await;
1267 report.stored.push(match outcome {
1268 Ok(outcome) => format!("{}: {outcome:?}", row.txid),
1269 Err(e) => format!("{}: not stored: {e}", row.txid),
1270 });
1271 }
1272 }
1273 adopt_proven(storage, services, &mut snapshot, ¶ms, now, &mut report).await?;
1274 write_ring(storage, &ring).await?;
1275 Ok(report)
1276}
1277
1278pub const MAX_ADOPT_PER_PASS: usize = 100;
1280
1281async fn adopt_proven<V: WalletServices + ?Sized>(
1299 storage: &StorageSqlx,
1300 services: &V,
1301 snapshot: &mut HeaderSnapshot,
1302 params: &Params,
1303 now: Timestamp,
1304 report: &mut TickReport,
1305) -> Result<()> {
1306 let found = sqlx::query(
1307 "SELECT lower(t.txid) AS txid, MAX(t.is_outgoing) AS made, p.merkle_path \
1308 FROM transactions t \
1309 JOIN proven_txs p ON p.proven_tx_id = t.proven_tx_id \
1310 WHERE t.txid IS NOT NULL AND t.status = 'completed' \
1311 AND NOT EXISTS (SELECT 1 FROM tracker_states s WHERE s.txid = lower(t.txid)) \
1312 AND EXISTS (SELECT 1 FROM outputs o \
1313 WHERE o.transaction_id = t.transaction_id AND o.spent_by IS NULL) \
1314 GROUP BY lower(t.txid) ORDER BY MIN(t.transaction_id) LIMIT ?",
1315 )
1316 .bind(MAX_ADOPT_PER_PASS as i64)
1317 .fetch_all(storage.pool())
1318 .await?;
1319 let mut rows = Vec::with_capacity(found.len());
1320 for r in &found {
1321 let txid: String = r.get("txid");
1322 let path: Option<Vec<u8>> = r.get("merkle_path");
1323 let Some(path) = path
1324 .as_deref()
1325 .and_then(|bytes| MerklePath::from_binary(bytes).ok())
1326 else {
1327 report.faults.push(format!(
1328 "{txid}: the stored proof does not parse; not adopted"
1329 ));
1330 continue;
1331 };
1332 rows.push(TrackedRow {
1333 built: r.get::<i64, _>("made") != 0,
1334 bump: Some(path.to_hex()),
1335 ..TrackedRow::new(&txid)
1336 });
1337 }
1338 snapshot
1339 .extend(services, rows.iter().filter_map(TrackedRow::proof_height))
1340 .await;
1341 for row in rows {
1342 let (state, fault) = row.replay(params, &*snapshot);
1343 match fault {
1344 None if matches!(state.word(), Word::Mined(_)) => {
1345 report.adopted_mined.push(row.txid.clone());
1346 }
1347 Some(CheckError::RootMismatch(_)) => {
1348 tracing::warn!(
1349 marker = "tracker_adopted_proof_failed",
1350 txid = %row.txid,
1351 reask = ?state.reask(),
1352 "a proven creator's stored merkle path does not meet the active header; tracked with its re-ask"
1353 );
1354 report.adopted_proof_failed.push(row.txid.clone());
1355 }
1356 Some(fault) => {
1358 report
1359 .faults
1360 .push(format!("{}: not adopted: {fault}", row.txid));
1361 continue;
1362 }
1363 None => {
1364 report.faults.push(format!(
1365 "{}: not adopted: the stored proof gave the word {}",
1366 row.txid,
1367 word_label(state.word())
1368 ));
1369 continue;
1370 }
1371 }
1372 store_row(storage, &row, &state, now).await?;
1373 }
1374 Ok(())
1375}
1376
1377pub async fn record_accepted<W: WalletInterface>(wallet: &W, report: &mut TickReport) {
1382 for (txid, args) in std::mem::take(&mut report.on_accept) {
1383 let line = match serde_json::from_str::<InternalizeActionArgs>(&args) {
1384 Ok(args) => match wallet.internalize_action(args, "bsv-wallet-cli").await {
1385 Ok(result) if result.accepted => format!("{txid}: recorded in the wallet"),
1386 Ok(_) => format!("{txid}: the wallet did not accept its own transaction"),
1387 Err(e) => format!(
1388 "{txid}: not recorded in the wallet ({e}); `bsv-wallet receive {txid}` records it once it is mined"
1389 ),
1390 },
1391 Err(e) => format!("{txid}: not recorded in the wallet (stored arguments: {e})"),
1392 };
1393 if !line.ends_with("recorded in the wallet") {
1394 tracing::warn!(marker = "own_transaction_not_recorded", "{line}");
1395 }
1396 report.stored.push(line);
1397 }
1398}
1399
1400fn readme_spend_guard<H: Headers>(
1409 state: &mut State,
1410 params: &Params,
1411 headers: &H,
1412) -> std::result::Result<bool, CheckError> {
1413 state.step(params, headers, Input::Host(HostAction::SpendAttempt))?;
1414 state.step(params, headers, Input::Evidence(Evidence::Recheck))?;
1415 Ok(matches!(state.word(), Word::Mined(_)) && !state.suspect())
1416}
1417
1418pub async fn spend_guard<V: WalletServices + ?Sized, C: Clock>(
1425 storage: &StorageSqlx,
1426 services: &V,
1427 clock: &C,
1428 opts: &TickOptions,
1429 txid: &str,
1430) -> Result<std::result::Result<bool, CheckError>> {
1431 ensure_tables(storage).await?;
1432 let Some(row) = load_row(storage, txid).await? else {
1433 return Ok(Ok(false));
1434 };
1435 let params = opts.params();
1436 let snapshot = HeaderSnapshot::take(services, row.proof_height()).await;
1437 let (mut state, fault) = row.replay(¶ms, &snapshot);
1438 match fault {
1439 Some(fault) if !matches!(fault, CheckError::RootMismatch(_)) => return Ok(Err(fault)),
1443 _ => {}
1444 }
1445 let split = under_split(state.word(), open_split_floor(storage).await?);
1446 let verdict = readme_spend_guard(&mut state, ¶ms, &snapshot).and_then(|offered| {
1447 Ok(offered && !split && deep_enough(&state, &snapshot, opts.spend_depth)?)
1448 });
1449 store_row(storage, &row, &state, clock.now()).await?;
1450 Ok(verdict)
1451}
1452
1453#[derive(Debug, Clone, PartialEq, Eq)]
1456pub struct CreatorReading {
1457 pub word: &'static str,
1459 pub depth: u64,
1462 pub suspect: bool,
1465 pub offered: bool,
1468 pub fault: Option<String>,
1471}
1472
1473pub async fn read_creators<V: WalletServices + ?Sized>(
1482 storage: &StorageSqlx,
1483 services: &V,
1484 opts: &TickOptions,
1485 creators: &[(String, Option<Vec<u8>>)],
1486) -> Result<HashMap<String, CreatorReading>> {
1487 ensure_tables(storage).await?;
1488 let params = opts.params();
1489 let mut rows = Vec::with_capacity(creators.len());
1490 for (txid, stored_path) in creators {
1491 let mut row = match load_row(storage, txid).await? {
1492 Some(row) => row,
1493 None => TrackedRow {
1494 built: false,
1495 ..TrackedRow::new(txid)
1496 },
1497 };
1498 if row.bump.is_none() {
1499 row.bump = stored_path
1500 .as_deref()
1501 .and_then(|bytes| MerklePath::from_binary(bytes).ok())
1502 .map(|path| path.to_hex());
1503 }
1504 rows.push(row);
1505 }
1506 let snapshot =
1507 HeaderSnapshot::take(services, rows.iter().filter_map(TrackedRow::proof_height)).await;
1508 let split_floor = open_split_floor(storage).await?;
1509 let mut readings = HashMap::with_capacity(rows.len());
1510 for row in rows {
1511 let (mut state, fault) = row.replay(¶ms, &snapshot);
1512 let fault = match fault {
1513 Some(fault) if !matches!(fault, CheckError::RootMismatch(_)) => Some(fault.to_string()),
1514 _ => None,
1515 };
1516 let verdict = match fault {
1517 Some(_) => Ok(false),
1518 None => readme_spend_guard(&mut state, ¶ms, &snapshot).and_then(|offered| {
1519 Ok(offered && deep_enough(&state, &snapshot, opts.spend_depth)?)
1520 }),
1521 };
1522 let depth = match (state.word().height(), snapshot.tip()) {
1523 (Some(height), Some((tip, _))) => {
1524 (u64::from(tip) + 1).saturating_sub(u64::from(height))
1525 }
1526 _ => 0,
1527 };
1528 let (offered, fault) = match verdict {
1529 Ok(offered) => (offered, fault),
1530 Err(e) => (false, fault.or(Some(e.to_string()))),
1531 };
1532 let split = under_split(state.word(), split_floor);
1533 readings.insert(
1534 row.txid.clone(),
1535 CreatorReading {
1536 word: word_label(state.word()),
1537 depth,
1538 suspect: state.suspect() || split,
1539 offered: offered && !split,
1540 fault,
1541 },
1542 );
1543 }
1544 Ok(readings)
1545}
1546
1547fn deep_enough<H: Headers>(
1552 state: &State,
1553 headers: &H,
1554 spend_depth: Height,
1555) -> std::result::Result<bool, CheckError> {
1556 let Some(height) = state.word().height() else {
1557 return Ok(false);
1558 };
1559 let tip = headers.tip_height().map_err(CheckError::Headers)?;
1560 let depth = (u64::from(tip) + 1).saturating_sub(u64::from(height));
1561 Ok(depth >= u64::from(spend_depth.max(DEFAULT_SPEND_DEPTH)))
1562}
1563
1564async fn read_open_splits<'e, E: sqlx::Executor<'e, Database = sqlx::Sqlite>>(
1572 executor: E,
1573) -> Result<BTreeMap<String, Height>> {
1574 let value: Option<String> =
1575 sqlx::query_scalar("SELECT value FROM tracker_meta WHERE key = 'open_splits'")
1576 .fetch_optional(executor)
1577 .await?;
1578 Ok(match value {
1579 Some(v) => serde_json::from_str(&v)?,
1580 None => BTreeMap::new(),
1581 })
1582}
1583
1584pub async fn open_split_floor(storage: &StorageSqlx) -> Result<Option<Height>> {
1590 ensure_tables(storage).await?;
1591 Ok(read_open_splits(storage.pool()).await?.into_values().min())
1592}
1593
1594fn under_split(word: &Word, floor: Option<Height>) -> bool {
1596 matches!((word, floor), (Word::Mined(m), Some(floor)) if floor <= m.height())
1597}
1598
1599pub async fn feed_cursor(storage: &StorageSqlx) -> Result<u64> {
1602 ensure_tables(storage).await?;
1603 let value: Option<String> =
1604 sqlx::query_scalar("SELECT value FROM tracker_meta WHERE key = 'feed_cursor'")
1605 .fetch_optional(storage.pool())
1606 .await?;
1607 Ok(match value {
1608 Some(v) => v
1609 .parse()
1610 .map_err(|e| anyhow::anyhow!("the stored feed cursor {v:?}: {e}"))?,
1611 None => 0,
1612 })
1613}
1614
1615async fn ensure_frozen(pool: &sqlx::SqlitePool) -> Result<()> {
1619 sqlx::query(
1620 "CREATE TABLE IF NOT EXISTS frozen_outpoints (\
1621 txid TEXT NOT NULL, \
1622 vout INTEGER NOT NULL, \
1623 cursor INTEGER NOT NULL, \
1624 frozen_at INTEGER NOT NULL, \
1625 PRIMARY KEY (txid, vout))",
1626 )
1627 .execute(pool)
1628 .await?;
1629 Ok(())
1630}
1631
1632pub async fn frozen_outpoints(
1634 pool: &sqlx::SqlitePool,
1635) -> Result<std::collections::HashSet<(String, i64)>> {
1636 ensure_frozen(pool).await?;
1637 let rows = sqlx::query("SELECT txid, vout FROM frozen_outpoints")
1638 .fetch_all(pool)
1639 .await?;
1640 Ok(rows
1641 .iter()
1642 .map(|r| (r.get::<String, _>("txid"), r.get::<i64, _>("vout")))
1643 .collect())
1644}
1645
1646pub async fn frozen_count(pool: &sqlx::SqlitePool) -> Result<u64> {
1648 ensure_frozen(pool).await?;
1649 let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM frozen_outpoints")
1650 .fetch_one(pool)
1651 .await?;
1652 Ok(n.max(0) as u64)
1653}
1654
1655#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1657pub struct ChainMove {
1658 pub txid: String,
1660 pub word: String,
1662 pub reask: Option<String>,
1664 pub suspect: bool,
1666 pub fault: Option<String>,
1668}
1669
1670#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1672pub struct ChainDelivery {
1673 pub cursor: u64,
1675 pub kind: &'static str,
1677 pub moved: Vec<ChainMove>,
1679}
1680
1681pub fn event_kind(event: &ChainEvent) -> &'static str {
1683 match event {
1684 ChainEvent::Tip { .. } => "tip",
1685 ChainEvent::Fork { .. } => "fork",
1686 ChainEvent::Reorg { .. } => "reorg",
1687 ChainEvent::Invalidated { .. } => "invalidated",
1688 ChainEvent::Frozen { .. } => "frozen",
1689 ChainEvent::TipAge { .. } => "tipAge",
1690 ChainEvent::Split { .. } => "split",
1691 ChainEvent::SplitResolved { .. } => "splitResolved",
1692 }
1693}
1694
1695struct Replaced<'a> {
1702 snapshot: &'a HeaderSnapshot,
1703 replaced: BTreeMap<Height, Header>,
1704 prior_tip: Option<Height>,
1705}
1706
1707impl Headers for Replaced<'_> {
1708 fn header_at(&self, height: Height) -> std::result::Result<Option<Header>, HeaderError> {
1709 match self.replaced.get(&height) {
1710 Some(header) => Ok(Some(header.clone())),
1711 None => self.snapshot.header_at(height),
1712 }
1713 }
1714
1715 fn tip_height(&self) -> std::result::Result<Height, HeaderError> {
1716 match self.prior_tip {
1717 Some(tip) => Ok(tip),
1718 None => self.snapshot.tip_height(),
1719 }
1720 }
1721}
1722
1723pub async fn deliver_chain_event<V: WalletServices + ?Sized, C: Clock>(
1737 storage: &StorageSqlx,
1738 services: &V,
1739 clock: &C,
1740 opts: &TickOptions,
1741 cursor: u64,
1742 event: ChainEvent,
1743) -> Result<ChainDelivery> {
1744 let kind = event_kind(&event);
1745 if cursor <= feed_cursor(storage).await? {
1746 return Ok(ChainDelivery {
1747 cursor,
1748 kind,
1749 moved: vec![],
1750 });
1751 }
1752 let named = format!("SELECT {ROW_COLUMNS} FROM tracker_states");
1753 let query = match &event {
1754 ChainEvent::Fork { height: from, .. }
1755 | ChainEvent::Reorg {
1756 fork_height: from, ..
1757 }
1758 | ChainEvent::Split {
1759 fork_height: from, ..
1760 }
1761 | ChainEvent::SplitResolved {
1762 fork_height: from, ..
1763 } => Some(
1764 sqlx::query(&format!(
1765 "{named} WHERE word = 'mined' AND height >= ? ORDER BY txid"
1766 ))
1767 .bind(i64::from(*from))
1768 .fetch_all(storage.pool())
1769 .await?,
1770 ),
1771 ChainEvent::Tip { .. } => Some(
1772 sqlx::query(&format!(
1773 "{named} WHERE word = 'mined' AND chain <> '[]' ORDER BY txid"
1774 ))
1775 .fetch_all(storage.pool())
1776 .await?,
1777 ),
1778 ChainEvent::Invalidated { block_hash } => Some(
1779 sqlx::query(&format!("{named} WHERE header_hash = ? ORDER BY txid"))
1780 .bind(block_hash.to_ascii_lowercase())
1781 .fetch_all(storage.pool())
1782 .await?,
1783 ),
1784 ChainEvent::Frozen { .. } | ChainEvent::TipAge { .. } => None,
1785 };
1786 let mut rows: Vec<TrackedRow> = query.iter().flatten().map(row_of).collect();
1787
1788 let now = clock.now();
1789 let mut moved = Vec::new();
1790 if matches!(event, ChainEvent::Frozen { .. }) {
1791 ensure_frozen(storage.pool()).await?;
1792 }
1793 let mut tx = storage.pool().begin().await?;
1794 if let ChainEvent::Frozen { outpoint } = &event {
1795 sqlx::query(
1796 "INSERT INTO frozen_outpoints (txid, vout, cursor, frozen_at) VALUES (?, ?, ?, ?) \
1797 ON CONFLICT(txid, vout) DO NOTHING",
1798 )
1799 .bind(outpoint.txid.to_ascii_lowercase())
1800 .bind(i64::from(outpoint.vout))
1801 .bind(cursor as i64)
1802 .bind(now as i64)
1803 .execute(&mut *tx)
1804 .await?;
1805 }
1806 if !rows.is_empty() {
1807 let snapshot =
1808 HeaderSnapshot::take(services, rows.iter().filter_map(TrackedRow::proof_height)).await;
1809 let (replaced, prior_tip) = match &event {
1810 ChainEvent::Reorg {
1811 deactivated_headers,
1812 ..
1813 } => (
1814 deactivated_headers
1815 .iter()
1816 .map(|h| {
1817 (
1818 h.height,
1819 Header {
1820 hash: h.hash.to_ascii_lowercase(),
1821 merkle_root: h.merkle_root.to_ascii_lowercase(),
1822 },
1823 )
1824 })
1825 .collect(),
1826 deactivated_headers.first().map(|h| h.height),
1827 ),
1828 _ => (BTreeMap::new(), None),
1829 };
1830 let replay_view = Replaced {
1831 snapshot: &snapshot,
1832 replaced,
1833 prior_tip,
1834 };
1835 let mut tracker = Tracker::new(opts.params());
1836 for row in &rows {
1837 row.replay_into(&mut tracker, &replay_view);
1838 }
1839 for update in tracker.on_chain(&snapshot, event.clone()) {
1840 let (Some(state), Some(row)) = (
1841 tracker.get(&update.txid),
1842 rows.iter_mut().find(|r| r.txid == update.txid),
1843 ) else {
1844 continue;
1845 };
1846 row.keep_chain(state, &event)?;
1847 if matches!(update.reask, Some(Reask::Reorg { .. })) {
1848 row.asks = 0;
1850 row.next_ask_at = None;
1851 }
1852 store_row_in(&mut *tx, row, state, now).await?;
1853 let fault = update.result.err().map(|e| e.to_string());
1854 if let Some(fault) = &fault {
1855 tracing::warn!(txid = %update.txid, kind, cursor, %fault, "a chain event's recheck could not look");
1856 }
1857 moved.push(ChainMove {
1858 txid: update.txid,
1859 word: word_label(state.word()).to_string(),
1860 reask: state.reask().map(reask_label),
1861 suspect: state.suspect(),
1862 fault,
1863 });
1864 }
1865 }
1866 if let ChainEvent::Split {
1867 key, fork_height, ..
1868 }
1869 | ChainEvent::SplitResolved {
1870 key, fork_height, ..
1871 } = &event
1872 {
1873 let mut open = read_open_splits(&mut *tx).await?;
1874 if matches!(event, ChainEvent::Split { .. }) {
1875 open.insert(key.clone(), *fork_height);
1876 } else {
1877 open.remove(key);
1878 }
1879 sqlx::query(
1880 "INSERT INTO tracker_meta (key, value) VALUES ('open_splits', ?) \
1881 ON CONFLICT(key) DO UPDATE SET value = excluded.value",
1882 )
1883 .bind(serde_json::to_string(&open)?)
1884 .execute(&mut *tx)
1885 .await?;
1886 }
1887 sqlx::query(
1888 "INSERT INTO tracker_meta (key, value) VALUES ('feed_cursor', ?) \
1889 ON CONFLICT(key) DO UPDATE SET value = excluded.value",
1890 )
1891 .bind(cursor.to_string())
1892 .execute(&mut *tx)
1893 .await?;
1894 tx.commit().await?;
1895 Ok(ChainDelivery {
1896 cursor,
1897 kind,
1898 moved,
1899 })
1900}
1901
1902#[cfg(test)]
1903pub(crate) mod tests {
1904 use super::*;
1905 use bsv_wallet_toolbox::services::mock::{MockResponse, MockWalletServices};
1906 use bsv_wallet_toolbox::services::{BlockHeader, GetMerklePathResult};
1907 use bsv_wallet_toolbox::{
1908 WalletStorageWriter, BROADCAST_STATUS_ACCEPTED, BROADCAST_STATUS_MINED,
1909 };
1910 use chrono::Utc;
1911
1912 pub(crate) struct FixedClock(pub Timestamp);
1914 impl Clock for FixedClock {
1915 fn now(&self) -> Timestamp {
1916 self.0
1917 }
1918 }
1919
1920 pub(crate) fn opts() -> TickOptions {
1921 TickOptions {
1922 age_threshold: 600,
1923 max_asks: 20,
1924 spend_depth: DEFAULT_SPEND_DEPTH,
1925 }
1926 }
1927
1928 pub(crate) async fn wallet_storage() -> (StorageSqlx, i64) {
1930 let storage = StorageSqlx::in_memory().await.unwrap();
1931 storage
1932 .migrate("tracker-tests", &("02".to_string() + &"ab".repeat(32)))
1933 .await
1934 .unwrap();
1935 storage.make_available().await.unwrap();
1936 let (user, _) = storage
1937 .find_or_insert_user(&("02".to_string() + &"cd".repeat(32)))
1938 .await
1939 .unwrap();
1940 (storage, user.user_id)
1941 }
1942
1943 pub(crate) async fn insert_tx(storage: &StorageSqlx, user_id: i64, txid: &str, status: &str) {
1944 sqlx::query(
1945 "INSERT INTO transactions (user_id, status, reference, is_outgoing, satoshis, version, lock_time, description, txid, raw_tx, created_at, updated_at) \
1946 VALUES (?, ?, ?, 1, 0, 1, 0, 'd', ?, X'01000000', ?, ?)",
1947 )
1948 .bind(user_id)
1949 .bind(status)
1950 .bind(&txid[..8])
1951 .bind(txid)
1952 .bind(Utc::now())
1953 .bind(Utc::now())
1954 .execute(storage.pool())
1955 .await
1956 .unwrap();
1957 }
1958
1959 pub(crate) async fn proven_coin(
1965 storage: &StorageSqlx,
1966 user_id: i64,
1967 txid: &str,
1968 height: u32,
1969 satoshis: i64,
1970 ) {
1971 let basket = storage
1972 .find_or_create_default_basket(user_id)
1973 .await
1974 .unwrap()
1975 .basket_id;
1976 let proven = sqlx::query(
1977 "INSERT INTO proven_txs (txid, height, idx, block_hash, merkle_root, merkle_path, raw_tx, created_at, updated_at) \
1978 VALUES (?, ?, 0, ?, ?, ?, X'01000000', ?, ?)",
1979 )
1980 .bind(txid)
1981 .bind(height as i64)
1982 .bind("bb".repeat(32))
1983 .bind(txid)
1984 .bind(MerklePath::from_coinbase_txid(txid, height).to_binary())
1985 .bind(Utc::now())
1986 .bind(Utc::now())
1987 .execute(storage.pool())
1988 .await
1989 .unwrap()
1990 .last_insert_rowid();
1991 let tx = sqlx::query(
1992 "INSERT INTO transactions (user_id, status, reference, is_outgoing, satoshis, version, lock_time, description, txid, raw_tx, proven_tx_id, created_at, updated_at) \
1993 VALUES (?, 'completed', ?, 0, ?, 1, 0, 'Internalize external funding', ?, X'01000000', ?, ?, ?)",
1994 )
1995 .bind(user_id)
1996 .bind(&txid[..8])
1997 .bind(satoshis)
1998 .bind(txid)
1999 .bind(proven)
2000 .bind(Utc::now())
2001 .bind(Utc::now())
2002 .execute(storage.pool())
2003 .await
2004 .unwrap()
2005 .last_insert_rowid();
2006 sqlx::query(
2007 "INSERT INTO outputs (user_id, transaction_id, basket_id, vout, satoshis, locking_script, txid, type, spendable, change, provided_by, purpose, output_description, created_at, updated_at) \
2008 VALUES (?, ?, ?, 0, ?, X'76a914dbc0a7c84983c5bf199b7b2d41b3acf0408ee5aa88ac', ?, 'P2PKH', 1, 1, 'storage', 'change', 'c', ?, ?)",
2009 )
2010 .bind(user_id)
2011 .bind(tx)
2012 .bind(basket)
2013 .bind(satoshis)
2014 .bind(txid)
2015 .bind(Utc::now())
2016 .bind(Utc::now())
2017 .execute(storage.pool())
2018 .await
2019 .unwrap();
2020 }
2021
2022 pub(crate) fn header(height: u32, root: &str, nonce: u32) -> BlockHeader {
2024 BlockHeader {
2025 version: 1,
2026 previous_hash: "00".repeat(32),
2027 merkle_root: root.to_string(),
2028 time: 1_700_000_000,
2029 bits: 0x1d00_ffff,
2030 nonce,
2031 hash: String::new(),
2032 height,
2033 }
2034 }
2035
2036 pub(crate) fn path_of(txid: &str, height: u32) -> String {
2039 MerklePath::from_coinbase_txid(txid, height).to_hex()
2040 }
2041
2042 pub(crate) fn proof_answer(path: Option<String>) -> MockResponse<GetMerklePathResult> {
2043 MockResponse::Success(GetMerklePathResult {
2044 name: Some("Services".to_string()),
2045 merkle_path: path,
2046 header: None,
2047 error: None,
2048 notes: vec![],
2049 })
2050 }
2051
2052 #[tokio::test]
2059 async fn a_pass_tracks_announces_and_proves_one_named_transaction() {
2060 let (storage, user) = wallet_storage().await;
2061 let txid = "a1".repeat(32);
2062 insert_tx(&storage, user, &txid, "unproven").await;
2063 storage
2064 .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_ACCEPTED)
2065 .await
2066 .unwrap();
2067 let services = MockWalletServices::builder()
2068 .get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
2069 .build();
2070 services.set_header_for_height(header(900, &txid, 0));
2071 let now = Utc::now().timestamp() as u64;
2072
2073 let report = tick(&storage, &services, &FixedClock(now), &opts())
2075 .await
2076 .unwrap();
2077 assert_eq!((report.adopted, report.hints, report.asked), (1, 1, 0));
2078 assert_eq!(
2079 stored_word(&storage, &txid).await.unwrap(),
2080 Some(("announced".to_string(), None, None))
2081 );
2082 assert_eq!(services.call_count("get_merkle_path"), 0);
2083
2084 let report = tick(&storage, &services, &FixedClock(now + 601), &opts())
2086 .await
2087 .unwrap();
2088 assert_eq!(report.asked, 1);
2089 assert_eq!(report.mined, vec![txid.clone()]);
2090 assert_eq!(
2091 stored_word(&storage, &txid).await.unwrap(),
2092 Some(("mined".to_string(), Some(900), None))
2093 );
2094 assert_eq!(services.call_count("get_merkle_path"), 1);
2095
2096 let report = tick(&storage, &services, &FixedClock(now + 1300), &opts())
2098 .await
2099 .unwrap();
2100 assert_eq!((report.tracked, report.asked), (0, 0));
2101 assert_eq!(services.call_count("get_merkle_path"), 1);
2102 }
2103
2104 #[tokio::test]
2108 async fn a_broadcasters_mined_is_a_hint_that_asks_for_the_proof() {
2109 let (storage, user) = wallet_storage().await;
2110 let txid = "b2".repeat(32);
2111 insert_tx(&storage, user, &txid, "unproven").await;
2112 storage
2113 .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
2114 .await
2115 .unwrap();
2116 let services = MockWalletServices::builder()
2117 .get_merkle_path_response(proof_answer(None))
2118 .build();
2119 let now = Utc::now().timestamp() as u64;
2120
2121 let report = tick(&storage, &services, &FixedClock(now), &opts())
2122 .await
2123 .unwrap();
2124 assert_eq!(report.asked, 1);
2125 assert_eq!(report.no_proof, vec![txid.clone()]);
2126 assert!(report.mined.is_empty());
2127 let (word, height, reask) = stored_word(&storage, &txid).await.unwrap().unwrap();
2128 assert_eq!((word.as_str(), height), ("built", None));
2129 assert_eq!(reask.as_deref(), Some("hint:arcade|mined"));
2130
2131 let report = tick(&storage, &services, &FixedClock(now + 60), &opts())
2133 .await
2134 .unwrap();
2135 assert_eq!(report.asked, 0);
2136 let report = tick(&storage, &services, &FixedClock(now + 600), &opts())
2138 .await
2139 .unwrap();
2140 assert_eq!(report.asked, 1);
2141 let report = tick(&storage, &services, &FixedClock(now + 1200), &opts())
2142 .await
2143 .unwrap();
2144 assert_eq!(report.asked, 0);
2145 assert_eq!(services.call_count("get_merkle_path"), 2);
2146 }
2147
2148 #[tokio::test]
2151 async fn a_moved_header_reasks_the_rows_at_or_above_it_and_no_other() {
2152 let (storage, user) = wallet_storage().await;
2153 let low = "c3".repeat(32);
2154 let high = "d4".repeat(32);
2155 let services = MockWalletServices::builder()
2156 .get_merkle_path_response(MockResponse::Sequence(vec![
2157 proof_answer(Some(path_of(&low, 100))),
2158 proof_answer(Some(path_of(&high, 200))),
2159 proof_answer(Some(path_of(&high, 201))),
2160 ]))
2161 .build();
2162 services.set_header_for_height(header(100, &low, 0));
2163 services.set_header_for_height(header(200, &high, 0));
2164 services.set_tip_header(header_with_hash(200, &high, 0));
2165 let now = Utc::now().timestamp() as u64;
2166 for txid in [&low, &high] {
2167 insert_tx(&storage, user, txid, "unproven").await;
2168 storage
2169 .record_broadcast_status(txid, "arcade", BROADCAST_STATUS_MINED)
2170 .await
2171 .unwrap();
2172 }
2173 let report = tick(&storage, &services, &FixedClock(now), &opts())
2174 .await
2175 .unwrap();
2176 assert_eq!(report.mined.len(), 2, "{report:?}");
2177
2178 services.set_header_for_height(header(200, &"ee".repeat(32), 7));
2181 services.set_header_for_height(header(201, &high, 0));
2182 services.set_tip_header(header_with_hash(201, &high, 0));
2183 let report = tick(&storage, &services, &FixedClock(now + 60), &opts())
2184 .await
2185 .unwrap();
2186 assert_eq!(report.moved_from, Some(200));
2187 assert_eq!(
2188 report.tracked, 1,
2189 "only the row at or above the moved header"
2190 );
2191 assert_eq!(report.moved, vec![high.clone()]);
2192 assert_eq!(report.mined, vec![high.clone()]);
2193 assert_eq!(
2194 stored_word(&storage, &high).await.unwrap().unwrap().1,
2195 Some(201)
2196 );
2197 assert_eq!(
2198 stored_word(&storage, &low).await.unwrap().unwrap().1,
2199 Some(100)
2200 );
2201 }
2202
2203 #[tokio::test]
2209 async fn a_not_final_transaction_is_announced_again_at_its_lock_time() {
2210 use bsv_wallet_toolbox::services::mock::{
2211 error_post_beef_result, success_post_beef_result,
2212 };
2213 let (storage, _user) = wallet_storage().await;
2214 let txid = "c7".repeat(32);
2215 let lock = 2_000_000_000u64;
2216 let services = MockWalletServices::builder()
2217 .post_beef_response(MockResponse::Sequence(vec![
2218 MockResponse::Success(vec![error_post_beef_result("Arcade", "476 non-final")]),
2219 MockResponse::Success(vec![success_post_beef_result("Arcade", &[&txid])]),
2220 ]))
2221 .get_merkle_path_response(proof_answer(None))
2222 .build();
2223
2224 let hint = Hint::new(
2225 "Arcade|rejected",
2226 HintStatus::Rejected {
2227 reason: "476 non-final".to_string(),
2228 },
2229 lock - 3600,
2230 );
2231 let again = Reannounce {
2232 beef: vec![1, 2, 3],
2233 at: lock,
2234 on_accept: None,
2235 };
2236 let word = heard(
2237 &storage,
2238 &FixedClock(lock - 3600),
2239 &opts(),
2240 &txid,
2241 hint,
2242 Some(lock),
2243 Some(again),
2244 )
2245 .await
2246 .unwrap();
2247 assert_eq!(
2248 word,
2249 (
2250 "built".to_string(),
2251 Some("hint:Arcade|rejected".to_string())
2252 )
2253 );
2254
2255 let report = tick(&storage, &services, &FixedClock(lock - 1), &opts())
2257 .await
2258 .unwrap();
2259 assert_eq!((report.tracked, report.asked), (1, 0), "{report:?}");
2260 assert_eq!(services.call_count("post_beef"), 0);
2261
2262 let report = tick(&storage, &services, &FixedClock(lock), &opts())
2264 .await
2265 .unwrap();
2266 assert_eq!(report.not_final, vec![txid.clone()]);
2267 assert_eq!(services.call_count("post_beef"), 1);
2268 let report = tick(&storage, &services, &FixedClock(lock + 60), &opts())
2269 .await
2270 .unwrap();
2271 assert!(report.not_final.is_empty() && report.reannounced.is_empty());
2272 assert_eq!(services.call_count("post_beef"), 1, "one pause between two");
2273
2274 let report = tick(&storage, &services, &FixedClock(lock + 600), &opts())
2276 .await
2277 .unwrap();
2278 assert_eq!(report.reannounced, vec![txid.clone()]);
2279 assert_eq!(services.call_count("post_beef"), 2);
2280 assert_eq!(
2281 stored_word(&storage, &txid).await.unwrap().unwrap().0,
2282 "announced"
2283 );
2284 assert_eq!(services.call_count("get_merkle_path"), 0);
2285
2286 let report = tick(&storage, &services, &FixedClock(lock + 1300), &opts())
2288 .await
2289 .unwrap();
2290 assert_eq!(report.asked, 1);
2291 assert_eq!(services.call_count("post_beef"), 2);
2292 assert_eq!(services.call_count("get_merkle_path"), 1);
2293 }
2294
2295 #[test]
2297 fn not_final_is_read_from_the_refusal() {
2298 for said in [
2299 "476 non-final",
2300 "bad-txns-nonfinal",
2301 "tx is not final",
2302 "code 476",
2303 ] {
2304 assert!(says_not_final(said), "{said}");
2305 }
2306 for said in [
2307 "fee too low",
2308 "DOUBLE_SPEND_ATTEMPTED",
2309 "txid 4476aa",
2310 "HTTP 500",
2311 ] {
2312 assert!(!says_not_final(said), "{said}");
2313 }
2314 }
2315
2316 pub(crate) fn header_with_hash(height: u32, root: &str, nonce: u32) -> BlockHeader {
2318 let mut h = header(height, root, nonce);
2319 h.hash = header_of(&h.to_binary()).unwrap().hash;
2320 h
2321 }
2322
2323 #[tokio::test]
2328 async fn the_spend_depth_holds_a_coin_until_its_block_is_deep_enough() {
2329 let (storage, user) = wallet_storage().await;
2330 let txid = "e7".repeat(32);
2331 insert_tx(&storage, user, &txid, "unproven").await;
2332 storage
2333 .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
2334 .await
2335 .unwrap();
2336 let services = MockWalletServices::builder()
2337 .get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
2338 .build();
2339 services.set_header_for_height(header(900, &txid, 0));
2340 services.set_tip_header(header_with_hash(900, &txid, 0));
2341 let clock = FixedClock(Utc::now().timestamp() as u64);
2342 tick(&storage, &services, &clock, &opts()).await.unwrap();
2343 let at = |spend_depth| TickOptions {
2344 spend_depth,
2345 ..opts()
2346 };
2347
2348 let one = spend_guard(&storage, &services, &clock, &at(1), &txid)
2349 .await
2350 .unwrap();
2351 let three = spend_guard(&storage, &services, &clock, &at(3), &txid)
2352 .await
2353 .unwrap();
2354 let zero = spend_guard(&storage, &services, &clock, &at(0), &txid)
2355 .await
2356 .unwrap();
2357 println!("one block deep: depth 1 {one:?}, depth 3 {three:?}, depth 0 {zero:?}");
2358 assert_eq!(one, Ok(true));
2359 assert_eq!(three, Ok(false));
2360 assert_eq!(zero, Ok(true), "0 reads as the floor, 1");
2361
2362 services.set_tip_header(header_with_hash(902, &"0c".repeat(32), 0));
2363 let three = spend_guard(&storage, &services, &clock, &at(3), &txid)
2364 .await
2365 .unwrap();
2366 println!("three blocks deep: depth 3 {three:?}");
2367 assert_eq!(three, Ok(true));
2368 assert_eq!(
2369 TickOptions {
2370 spend_depth: 0,
2371 ..opts()
2372 }
2373 .floored()
2374 .spend_depth,
2375 1
2376 );
2377 }
2378
2379 #[tokio::test]
2384 async fn the_readme_spend_guard_runs_as_the_host() {
2385 let (storage, user) = wallet_storage().await;
2386 let txid = "e5".repeat(32);
2387 insert_tx(&storage, user, &txid, "unproven").await;
2388 storage
2389 .record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
2390 .await
2391 .unwrap();
2392 let services = MockWalletServices::builder()
2393 .get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
2394 .build();
2395 services.set_header_for_height(header(900, &txid, 0));
2396 let clock = FixedClock(Utc::now().timestamp() as u64);
2397 tick(&storage, &services, &clock, &opts()).await.unwrap();
2398
2399 let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
2401 .await
2402 .unwrap();
2403 println!(
2404 "spend guard, header in place: {guard:?}, stored {:?}",
2405 stored_word(&storage, &txid).await.unwrap().unwrap()
2406 );
2407 assert_eq!(guard, Ok(true));
2408
2409 services.set_tip_unavailable();
2411 let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
2412 .await
2413 .unwrap();
2414 println!(
2415 "spend guard, no header service: {guard:?}, stored {:?}",
2416 stored_word(&storage, &txid).await.unwrap().unwrap()
2417 );
2418 assert!(matches!(guard, Err(CheckError::Headers(_))), "{guard:?}");
2419 assert_eq!(
2420 stored_word(&storage, &txid).await.unwrap().unwrap().0,
2421 "mined",
2422 "could not look changes no word"
2423 );
2424
2425 services.set_tip_header(header_with_hash(901, &"00".repeat(32), 1));
2427 services.set_header_for_height(header(900, &"ee".repeat(32), 7));
2428 let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
2429 .await
2430 .unwrap();
2431 println!(
2432 "spend guard, header moved: {guard:?}, stored {:?}",
2433 stored_word(&storage, &txid).await.unwrap().unwrap()
2434 );
2435 assert_eq!(guard, Ok(false));
2436
2437 let guard = spend_guard(&storage, &services, &clock, &opts(), &"f6".repeat(32))
2439 .await
2440 .unwrap();
2441 assert_eq!(guard, Ok(false));
2442
2443 let params = opts().params();
2446 services.set_tip_header(header_with_hash(900, &txid, 0));
2447 services.set_header_for_height(header(900, &txid, 0));
2448 let held = HeaderSnapshot::take(&services, [900]).await;
2449 let row = load_row(&storage, &txid).await.unwrap().unwrap();
2450 let (mut state, fault) = row.replay(¶ms, &held);
2451 assert_eq!(fault, None);
2452 let lost = HeaderSnapshot::take(&services, []).await;
2453 let guard = readme_spend_guard(&mut state, ¶ms, &lost);
2454 println!("spend guard, height unanswered: {guard:?}");
2455 assert_eq!(guard, Err(CheckError::Unavailable(900)));
2456 }
2457
2458 #[tokio::test]
2468 async fn the_served_pass_adopts_an_unproven_creator_and_a_proven_completed_one() {
2469 let (storage, user) = wallet_storage().await;
2470 let unproven = "d1".repeat(32);
2471 let earlier = "d2".repeat(32);
2472 insert_tx(&storage, user, &unproven, "unproven").await;
2473 insert_tx(&storage, user, &earlier, "unproven").await;
2474 let services = MockWalletServices::builder().build();
2475 services.set_tip_header(header_with_hash(960, &"0a".repeat(32), 0));
2476 let now = Utc::now().timestamp() as u64;
2477 let pass = crate::broadcast_reconcile::run_served_pass(
2478 &storage,
2479 &services,
2480 &FixedClock(now),
2481 &opts(),
2482 None,
2483 )
2484 .await
2485 .unwrap();
2486 println!("{}", pass.summary());
2487 assert_eq!(pass.tracker.adopted, 2);
2488 assert_eq!(
2489 stored_word(&storage, &earlier).await.unwrap(),
2490 Some(("built".to_string(), None, None))
2491 );
2492
2493 sqlx::query("UPDATE transactions SET status = 'completed' WHERE txid = ?")
2497 .bind(&earlier)
2498 .execute(storage.pool())
2499 .await
2500 .unwrap();
2501 let checks = "d3".repeat(32);
2502 let moved = "d4".repeat(32);
2503 proven_coin(&storage, user, &checks, 900, 3_000).await;
2504 proven_coin(&storage, user, &moved, 950, 2_000).await;
2505 services.set_header_for_height(header(900, &checks, 0));
2506 services.set_header_for_height(header(950, &"ee".repeat(32), 0));
2507 let toolbox_before: Vec<(String, String, Option<i64>)> =
2508 sqlx::query_as("SELECT txid, status, proven_tx_id FROM transactions ORDER BY txid")
2509 .fetch_all(storage.pool())
2510 .await
2511 .unwrap();
2512
2513 let pass = crate::broadcast_reconcile::run_served_pass(
2514 &storage,
2515 &services,
2516 &FixedClock(now + 60),
2517 &opts(),
2518 None,
2519 )
2520 .await
2521 .unwrap();
2522 println!("{}", pass.summary());
2523 assert_eq!(pass.tracker.adopted_mined, vec![checks.clone()]);
2524 assert_eq!(pass.tracker.adopted_proof_failed, vec![moved.clone()]);
2525 assert_eq!(
2526 stored_word(&storage, &checks).await.unwrap(),
2527 Some(("mined".to_string(), Some(900), None))
2528 );
2529 let row = load_row(&storage, &checks).await.unwrap().unwrap();
2530 assert_eq!(
2531 row.bump,
2532 Some(path_of(&checks, 900)),
2533 "the evidence is the stored path"
2534 );
2535 assert!(!row.built, "a received coin's creator was not built here");
2536 assert_eq!(
2537 stored_word(&storage, &moved).await.unwrap(),
2538 Some((
2539 "unknown".to_string(),
2540 None,
2541 Some("proof_failed@950".to_string())
2542 ))
2543 );
2544 assert_eq!(
2545 stored_word(&storage, &earlier).await.unwrap(),
2546 Some(("built".to_string(), None, None)),
2547 "a creator tracked while unproven keeps its row"
2548 );
2549 let toolbox_after: Vec<(String, String, Option<i64>)> =
2550 sqlx::query_as("SELECT txid, status, proven_tx_id FROM transactions ORDER BY txid")
2551 .fetch_all(storage.pool())
2552 .await
2553 .unwrap();
2554 assert_eq!(
2555 toolbox_after, toolbox_before,
2556 "no write to the toolbox's rows"
2557 );
2558 assert_eq!(services.call_count("get_merkle_path"), 0);
2559
2560 let pass = crate::broadcast_reconcile::run_served_pass(
2561 &storage,
2562 &services,
2563 &FixedClock(now + 120),
2564 &opts(),
2565 None,
2566 )
2567 .await
2568 .unwrap();
2569 println!("{}", pass.summary());
2570 assert!(pass.tracker.adopted_mined.is_empty());
2571 assert!(pass.tracker.adopted_proof_failed.is_empty());
2572 }
2573}