use std::collections::{BTreeMap, HashMap, VecDeque};
use std::convert::Infallible;
use anyhow::Result;
use bsv_sdk::transaction::MerklePath;
use bsv_sdk::wallet::{InternalizeActionArgs, WalletInterface};
use bsv_tracker::{
decode_envelope, ChainEnvelope, ChainEvent, CheckError, Clock, Evidence, Header, HeaderError,
Headers, Height, Hint, HintSource, HintStatus, HostAction, Input, Params, Proof, ProofFetcher,
Reask, State, Timestamp, Tracker, TxId, Word,
};
use bsv_wallet_toolbox::services::PostBeefResult;
use bsv_wallet_toolbox::{BroadcastStatus, StorageSqlx, WalletServices};
use serde::{Deserialize, Serialize};
use sqlx::Row as _;
pub const DEFAULT_AGE_SECS: u64 = 600;
pub const DEFAULT_MAX_ASKS: usize = 20;
const MAX_BACKOFF_DOUBLINGS: u32 = 6;
const MAX_ASKS_PER_TX: u32 = 16;
const TIP_RING: usize = 12;
pub const DEFAULT_SPEND_DEPTH: Height = 1;
#[derive(Debug, Clone)]
pub struct HeaderSnapshot {
tip: std::result::Result<(Height, String), String>,
headers: BTreeMap<Height, std::result::Result<Header, String>>,
}
impl HeaderSnapshot {
pub async fn take<V: WalletServices + ?Sized>(
services: &V,
heights: impl IntoIterator<Item = Height>,
) -> Self {
let tip = services
.get_chain_tip_header()
.await
.map(|h| (h.height, h.hash.to_ascii_lowercase()))
.map_err(|e| e.to_string());
let mut snapshot = Self {
tip,
headers: BTreeMap::new(),
};
snapshot.extend(services, heights).await;
snapshot
}
pub fn unread() -> Self {
Self {
tip: Err("the header service was not read".to_string()),
headers: BTreeMap::new(),
}
}
pub async fn extend<V: WalletServices + ?Sized>(
&mut self,
services: &V,
heights: impl IntoIterator<Item = Height>,
) {
for height in heights {
if self.headers.contains_key(&height) {
continue;
}
let answer = match services.get_header_for_height(height).await {
Ok(bytes) => header_of(&bytes),
Err(e) => Err(e.to_string()),
};
self.headers.insert(height, answer);
}
}
pub fn tip(&self) -> Option<(Height, &str)> {
self.tip.as_ref().ok().map(|(h, hash)| (*h, hash.as_str()))
}
}
impl Headers for HeaderSnapshot {
fn header_at(&self, height: Height) -> std::result::Result<Option<Header>, HeaderError> {
match self.headers.get(&height) {
Some(Ok(header)) => Ok(Some(header.clone())),
Some(Err(fault)) => Err(HeaderError(fault.clone())),
None => Ok(None),
}
}
fn tip_height(&self) -> std::result::Result<Height, HeaderError> {
self.tip
.as_ref()
.map(|(height, _)| *height)
.map_err(|fault| HeaderError(fault.clone()))
}
}
fn header_of(bytes: &[u8]) -> std::result::Result<Header, String> {
if bytes.len() != 80 {
return Err(format!("a header of {} bytes", bytes.len()));
}
let display = |raw: &[u8]| {
let mut reversed = raw.to_vec();
reversed.reverse();
hex::encode(reversed)
};
Ok(Header {
hash: display(&bsv_sdk::primitives::hash::sha256d(bytes)),
merkle_root: display(&bytes[36..68]),
})
}
#[derive(Debug, Clone, Copy, Default)]
pub struct SystemClock;
impl Clock for SystemClock {
fn now(&self) -> Timestamp {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
}
#[derive(Debug, Default)]
pub struct MemoryHints {
queue: VecDeque<(TxId, Hint)>,
}
impl MemoryHints {
async fn read(storage: &StorageSqlx, rows: &[TrackedRow]) -> Result<Self> {
let txids: Vec<String> = rows.iter().map(|r| r.txid.clone()).collect();
if txids.is_empty() {
return Ok(Self::default());
}
let mut records = storage.broadcast_records(None, &txids).await?;
records.sort_by_key(|r| r.seen_at);
let mut queue = VecDeque::new();
for record in records {
let Some(status) = record.ladder_status() else {
continue;
};
let txid = record.txid.to_ascii_lowercase();
let Some(row) = rows.iter().find(|r| r.txid == txid) else {
continue;
};
let (label, status) = match status {
BroadcastStatus::Accepted => ("accepted", HintStatus::Accepted),
BroadcastStatus::Seen => ("seen", HintStatus::Seen),
BroadcastStatus::Mined => ("mined", HintStatus::Unknown),
BroadcastStatus::Rejected => (
"rejected",
HintStatus::Rejected {
reason: record.status.clone(),
},
),
BroadcastStatus::Unknown => ("unknown", HintStatus::Unknown),
};
let source = format!("{}|{}", record.provider, label);
if row.hints.iter().any(|h| h.source == source) {
continue;
}
let observed = record.seen_at.timestamp().max(0) as Timestamp;
queue.push_back((txid, Hint::new(source, status, observed)));
}
Ok(Self { queue })
}
}
impl HintSource for MemoryHints {
type Error = Infallible;
fn next_hint(&mut self) -> std::result::Result<Option<(TxId, Hint)>, Self::Error> {
Ok(self.queue.pop_front())
}
}
#[derive(Debug, Default)]
pub struct Courier {
carried: HashMap<TxId, std::result::Result<Option<Proof>, String>>,
}
impl Courier {
pub async fn carry<V: WalletServices + ?Sized>(services: &V, txids: &[TxId]) -> Self {
let mut carried = HashMap::new();
for txid in txids {
let answer = match services.get_merkle_path(txid, false).await {
Ok(found) => match found.merkle_path {
Some(hex) => MerklePath::from_hex(&hex)
.map_err(|e| format!("not a merkle path: {e}"))
.and_then(|path| Proof::new(txid.clone(), path).map_err(|e| e.to_string()))
.map(Some),
None => Ok(None),
},
Err(e) => Err(e.to_string()),
};
carried.insert(txid.clone(), answer);
}
Self { carried }
}
fn heights(&self) -> Vec<Height> {
self.carried
.values()
.filter_map(|answer| answer.as_ref().ok()?.as_ref().map(Proof::height))
.collect()
}
}
impl ProofFetcher for Courier {
type Error = String;
fn fetch(
&mut self,
txid: &str,
_reask: &Reask,
) -> std::result::Result<Option<Proof>, Self::Error> {
self.carried.remove(txid).unwrap_or(Ok(None))
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct StoredHint {
source: String,
status: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
height: Option<Height>,
#[serde(default, skip_serializing_if = "Option::is_none")]
reason: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
competitors: Vec<TxId>,
observed: Timestamp,
}
impl StoredHint {
fn of(hint: &Hint) -> Self {
let (status, height, reason, competitors) = match &hint.status {
HintStatus::Accepted => ("accepted", None, None, vec![]),
HintStatus::Seen => ("seen", None, None, vec![]),
HintStatus::Mined { height } => ("mined", Some(*height), None, vec![]),
HintStatus::StaleBlock => ("stale_block", None, None, vec![]),
HintStatus::Rejected { reason } => ("rejected", None, Some(reason.clone()), vec![]),
HintStatus::DoubleSpend { competitors } => {
("double_spend", None, None, competitors.clone())
}
HintStatus::OrphanMempool => ("orphan_mempool", None, None, vec![]),
HintStatus::Unknown => ("unknown", None, None, vec![]),
};
Self {
source: hint.source.clone(),
status: status.to_string(),
height,
reason,
competitors,
observed: hint.observed,
}
}
fn hint(&self) -> Hint {
let status = match self.status.as_str() {
"accepted" => HintStatus::Accepted,
"seen" => HintStatus::Seen,
"mined" => self
.height
.map_or(HintStatus::Unknown, |height| HintStatus::Mined { height }),
"stale_block" => HintStatus::StaleBlock,
"rejected" => HintStatus::Rejected {
reason: self.reason.clone().unwrap_or_default(),
},
"double_spend" => HintStatus::DoubleSpend {
competitors: self.competitors.clone(),
},
"orphan_mempool" => HintStatus::OrphanMempool,
_ => HintStatus::Unknown,
};
Hint::new(self.source.clone(), status, self.observed)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct TrackedRow {
txid: TxId,
built: bool,
hints: Vec<StoredHint>,
bump: Option<String>,
asks: u32,
next_ask_at: Option<Timestamp>,
reannounce_beef: Option<Vec<u8>>,
reannounce_at: Option<Timestamp>,
on_accept: Option<String>,
chain: Vec<String>,
}
impl TrackedRow {
fn new(txid: &str) -> Self {
Self {
txid: txid.to_ascii_lowercase(),
built: true,
hints: vec![],
bump: None,
asks: 0,
next_ask_at: None,
reannounce_beef: None,
reannounce_at: None,
on_accept: None,
chain: vec![],
}
}
fn chain_events(&self) -> Vec<ChainEvent> {
self.chain
.iter()
.filter_map(|json| match decode_envelope(json.as_bytes()) {
Ok(event) => Some(event),
Err(fault) => {
tracing::warn!(txid = %self.txid, %fault, "a stored chain event does not decode");
None
}
})
.collect()
}
fn keep_chain(&mut self, state: &State, event: &ChainEvent) -> Result<()> {
let plain = matches!(state.word(), Word::Mined(_)) && !state.suspect();
if plain && state.reask().is_none() {
self.chain.clear();
return Ok(());
}
if matches!(event, ChainEvent::Tip { .. } | ChainEvent::TipAge { .. }) {
self.chain.retain(|json| {
!matches!(
decode_envelope(json.as_bytes()),
Ok(ChainEvent::Tip { .. } | ChainEvent::TipAge { .. })
)
});
}
self.chain
.push(serde_json::to_string(&ChainEnvelope::new(event.clone()))?);
Ok(())
}
fn replay_into<H: Headers>(&self, tracker: &mut Tracker, headers: &H) {
tracker.track(self.txid.clone());
if self.built {
let _ = tracker.apply(&self.txid, headers, Input::Host(HostAction::Build));
}
for hint in &self.hints {
let _ = tracker.apply(&self.txid, headers, Input::Hint(hint.hint()));
}
if let Some(Ok(proof)) = self.proof() {
let _ = tracker.apply(&self.txid, headers, Input::Evidence(Evidence::Proof(proof)));
}
for event in self.chain_events() {
let _ = tracker.apply(&self.txid, headers, Input::Evidence(Evidence::Chain(event)));
}
}
fn proof(&self) -> Option<std::result::Result<Proof, CheckError>> {
let hex = self.bump.as_deref()?;
Some(
MerklePath::from_hex(hex)
.map_err(|e| CheckError::InvalidProof(e.to_string()))
.and_then(|path| Proof::new(self.txid.clone(), path)),
)
}
fn proof_height(&self) -> Option<Height> {
self.proof()?.ok().map(|p| p.height())
}
fn replay<H: Headers>(&self, params: &Params, headers: &H) -> (State, Option<CheckError>) {
let mut state = State::new(self.txid.clone());
if self.built {
let _ = state.step(params, headers, Input::Host(HostAction::Build));
}
for hint in &self.hints {
let _ = state.step(params, headers, Input::Hint(hint.hint()));
}
let fault = match self.proof() {
Some(Ok(proof)) => state
.step(params, headers, Input::Evidence(Evidence::Proof(proof)))
.err(),
Some(Err(fault)) => Some(fault),
None => None,
};
for event in self.chain_events() {
let _ = state.step(params, headers, Input::Evidence(Evidence::Chain(event)));
}
(state, fault)
}
}
pub fn word_label(word: &Word) -> &'static str {
match word {
Word::Unknown => "unknown",
Word::Built => "built",
Word::Announced => "announced",
Word::Seen => "seen",
Word::Mined(_) => "mined",
Word::Stale => "stale",
Word::Rejected { .. } => "rejected",
Word::Conflicted { .. } => "conflicted",
Word::Abandoned => "abandoned",
}
}
fn reask_label(reask: &Reask) -> String {
match reask {
Reask::Reorg { fork_height } => format!("reorg@{fork_height}"),
Reask::Fork { height } => format!("fork@{height}"),
Reask::Hint(hint) => format!("hint:{}", hint.source),
Reask::ProofFailed { height } => format!("proof_failed@{height}"),
Reask::Recheck => "recheck".to_string(),
Reask::Age => "age".to_string(),
Reask::Spend => "spend".to_string(),
}
}
async fn ensure_tables(storage: &StorageSqlx) -> Result<()> {
for sql in [
"CREATE TABLE IF NOT EXISTS tracker_states (\
txid TEXT PRIMARY KEY, \
built INTEGER NOT NULL DEFAULT 1, \
hints TEXT NOT NULL DEFAULT '[]', \
bump TEXT, \
word TEXT NOT NULL DEFAULT 'unknown', \
height INTEGER, \
header_hash TEXT, \
reask TEXT, \
asks INTEGER NOT NULL DEFAULT 0, \
next_ask_at INTEGER, \
reannounce_beef BLOB, \
reannounce_at INTEGER, \
on_accept TEXT, \
updated_at INTEGER NOT NULL DEFAULT 0)",
"CREATE INDEX IF NOT EXISTS tracker_states_word ON tracker_states (word)",
"CREATE INDEX IF NOT EXISTS tracker_states_height ON tracker_states (height)",
"CREATE TABLE IF NOT EXISTS tracker_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)",
] {
sqlx::query(sql).execute(storage.pool()).await?;
}
let has_chain: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM pragma_table_info('tracker_states') WHERE name = 'chain'",
)
.fetch_one(storage.pool())
.await?;
if has_chain == 0 {
sqlx::query("ALTER TABLE tracker_states ADD COLUMN chain TEXT NOT NULL DEFAULT '[]'")
.execute(storage.pool())
.await?;
}
Ok(())
}
const ROW_COLUMNS: &str =
"txid, built, hints, bump, asks, next_ask_at, reannounce_beef, reannounce_at, on_accept, chain";
fn row_of(row: &sqlx::sqlite::SqliteRow) -> TrackedRow {
let hints: String = row.get("hints");
let chain: String = row.get("chain");
TrackedRow {
txid: row.get("txid"),
built: row.get::<i64, _>("built") != 0,
hints: serde_json::from_str(&hints).unwrap_or_default(),
bump: row.get("bump"),
asks: row.get::<i64, _>("asks").max(0) as u32,
next_ask_at: row
.get::<Option<i64>, _>("next_ask_at")
.map(|t| t.max(0) as Timestamp),
reannounce_beef: row.get("reannounce_beef"),
reannounce_at: row
.get::<Option<i64>, _>("reannounce_at")
.map(|t| t.max(0) as Timestamp),
on_accept: row.get("on_accept"),
chain: serde_json::from_str(&chain).unwrap_or_default(),
}
}
async fn load_row(storage: &StorageSqlx, txid: &str) -> Result<Option<TrackedRow>> {
let row = sqlx::query(&format!(
"SELECT {ROW_COLUMNS} FROM tracker_states WHERE txid = ?"
))
.bind(txid.to_ascii_lowercase())
.fetch_optional(storage.pool())
.await?;
Ok(row.as_ref().map(row_of))
}
async fn store_row(
storage: &StorageSqlx,
row: &TrackedRow,
state: &State,
now: Timestamp,
) -> Result<()> {
store_row_in(storage.pool(), row, state, now).await
}
async fn store_row_in<'e, E: sqlx::Executor<'e, Database = sqlx::Sqlite>>(
executor: E,
row: &TrackedRow,
state: &State,
now: Timestamp,
) -> Result<()> {
let (height, header_hash) = match state.word() {
Word::Mined(mined) => (
Some(mined.height() as i64),
Some(mined.checked().header().hash.clone()),
),
_ => (None, None),
};
sqlx::query(
"INSERT INTO tracker_states \
(txid, built, hints, bump, word, height, header_hash, reask, asks, next_ask_at, \
reannounce_beef, reannounce_at, on_accept, chain, updated_at) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
ON CONFLICT(txid) DO UPDATE SET built = excluded.built, hints = excluded.hints, \
bump = excluded.bump, word = excluded.word, height = excluded.height, \
header_hash = excluded.header_hash, reask = excluded.reask, asks = excluded.asks, \
next_ask_at = excluded.next_ask_at, reannounce_beef = excluded.reannounce_beef, \
reannounce_at = excluded.reannounce_at, on_accept = excluded.on_accept, \
chain = excluded.chain, updated_at = excluded.updated_at",
)
.bind(&row.txid)
.bind(row.built as i64)
.bind(serde_json::to_string(&row.hints)?)
.bind(&row.bump)
.bind(word_label(state.word()))
.bind(height)
.bind(header_hash)
.bind(state.reask().map(reask_label))
.bind(row.asks as i64)
.bind(row.next_ask_at.map(|t| t as i64))
.bind(&row.reannounce_beef)
.bind(row.reannounce_at.map(|t| t as i64))
.bind(&row.on_accept)
.bind(serde_json::to_string(&row.chain)?)
.bind(now as i64)
.execute(executor)
.await?;
Ok(())
}
pub async fn stored_word(
storage: &StorageSqlx,
txid: &str,
) -> Result<Option<(String, Option<Height>, Option<String>)>> {
ensure_tables(storage).await?;
let row = sqlx::query("SELECT word, height, reask FROM tracker_states WHERE txid = ?")
.bind(txid.to_ascii_lowercase())
.fetch_optional(storage.pool())
.await?;
Ok(row.map(|r| {
(
r.get("word"),
r.get::<Option<i64>, _>("height").map(|h| h as Height),
r.get("reask"),
)
}))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PostAnswer {
Accepted { provider: String },
NotFinal { provider: String, said: String },
Refused { said: String },
}
pub fn says_not_final(text: &str) -> bool {
let lower = text.to_ascii_lowercase();
["non-final", "nonfinal", "non final", "not final"]
.iter()
.any(|word| lower.contains(word))
|| lower
.split(|c: char| !c.is_ascii_alphanumeric())
.any(|token| token == "476")
}
pub fn read_post(results: &[PostBeefResult]) -> PostAnswer {
if let Some(accepted) = results.iter().find(|r| r.is_success()) {
return PostAnswer::Accepted {
provider: accepted.name.clone(),
};
}
let said: Vec<(String, String)> = results
.iter()
.map(|r| {
let detail = r
.error
.clone()
.or_else(|| r.txid_results.iter().find_map(|t| t.data.clone()))
.unwrap_or_else(|| r.status.clone());
(r.name.clone(), detail)
})
.collect();
if let Some((provider, detail)) = said.iter().find(|(_, detail)| says_not_final(detail)) {
return PostAnswer::NotFinal {
provider: provider.clone(),
said: detail.clone(),
};
}
PostAnswer::Refused {
said: if said.is_empty() {
"none answered".to_string()
} else {
said.iter()
.map(|(name, detail)| format!("{name}: {detail}"))
.collect::<Vec<_>>()
.join("; ")
},
}
}
#[derive(Debug, Clone)]
pub struct Reannounce {
pub beef: Vec<u8>,
pub at: Timestamp,
pub on_accept: Option<InternalizeActionArgs>,
}
pub async fn heard<C: Clock>(
storage: &StorageSqlx,
clock: &C,
opts: &TickOptions,
txid: &str,
hint: Hint,
not_before: Option<Timestamp>,
reannounce: Option<Reannounce>,
) -> Result<(String, Option<String>)> {
ensure_tables(storage).await?;
let mut row = match load_row(storage, txid).await? {
Some(row) if row.bump.is_some() => {
let stored = stored_word(storage, txid).await?.unwrap_or_default();
return Ok((stored.0, stored.2));
}
Some(row) => row,
None => TrackedRow::new(txid),
};
let params = opts.params();
let headers = HeaderSnapshot::unread();
let (mut state, _) = row.replay(¶ms, &headers);
if !row.hints.iter().any(|h| h.source == hint.source) {
row.hints.push(StoredHint::of(&hint));
let _ = state.step(¶ms, &headers, Input::Hint(hint));
}
row.asks = 0;
row.next_ask_at = not_before;
match reannounce {
Some(again) => {
row.reannounce_beef = Some(again.beef);
row.reannounce_at = Some(again.at);
row.on_accept = match again.on_accept {
Some(args) => Some(serde_json::to_string(&args)?),
None => None,
};
}
None => {
row.reannounce_beef = None;
row.reannounce_at = None;
row.on_accept = None;
}
}
store_row(storage, &row, &state, clock.now()).await?;
Ok((
word_label(state.word()).to_string(),
state.reask().map(reask_label),
))
}
#[derive(Debug, Clone)]
pub struct TickOptions {
pub age_threshold: Timestamp,
pub max_asks: usize,
pub spend_depth: Height,
}
impl TickOptions {
pub fn from_env() -> Self {
fn env<T: std::str::FromStr>(key: &str, default: T) -> T {
std::env::var(key)
.ok()
.and_then(|v| v.trim().parse().ok())
.unwrap_or(default)
}
Self {
age_threshold: env("TRACKER_AGE_SECS", DEFAULT_AGE_SECS).max(1),
max_asks: env("TRACKER_MAX_ASKS", DEFAULT_MAX_ASKS),
spend_depth: env("SPEND_DEPTH", DEFAULT_SPEND_DEPTH),
}
.floored()
}
#[must_use]
pub fn floored(mut self) -> Self {
self.spend_depth = self.spend_depth.max(DEFAULT_SPEND_DEPTH);
self
}
fn params(&self) -> Params {
Params {
age_threshold: self.age_threshold,
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct TickReport {
pub tracked: usize,
pub adopted: usize,
pub adopted_mined: Vec<String>,
pub adopted_proof_failed: Vec<String>,
pub hints: usize,
pub asked: usize,
pub mined: Vec<String>,
pub no_proof: Vec<String>,
pub moved: Vec<String>,
pub moved_from: Option<Height>,
pub reannounced: Vec<String>,
pub not_final: Vec<String>,
#[serde(skip)]
pub on_accept: Vec<(String, String)>,
pub faults: Vec<String>,
pub stored: Vec<String>,
}
impl TickReport {
pub fn is_quiet(&self) -> bool {
self.adopted == 0
&& self.adopted_mined.is_empty()
&& self.adopted_proof_failed.is_empty()
&& self.hints == 0
&& self.asked == 0
&& self.reannounced.is_empty()
&& self.not_final.is_empty()
&& self.moved.is_empty()
&& self.faults.is_empty()
}
pub fn summary(&self) -> String {
format!(
"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)",
self.tracked,
self.adopted,
self.adopted_mined.len() + self.adopted_proof_failed.len(),
self.adopted_mined.len(),
self.adopted_proof_failed.len(),
self.hints,
self.reannounced.len(),
self.not_final.len(),
self.asked,
self.mined.len(),
self.no_proof.len(),
self.moved.len(),
self.faults.len(),
)
}
}
async fn read_ring(storage: &StorageSqlx) -> Vec<(Height, String)> {
sqlx::query("SELECT value FROM tracker_meta WHERE key = 'tips'")
.fetch_optional(storage.pool())
.await
.ok()
.flatten()
.and_then(|r| serde_json::from_str(&r.get::<String, _>("value")).ok())
.unwrap_or_default()
}
async fn write_ring(storage: &StorageSqlx, ring: &[(Height, String)]) -> Result<()> {
sqlx::query(
"INSERT INTO tracker_meta (key, value) VALUES ('tips', ?) \
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
)
.bind(serde_json::to_string(ring)?)
.execute(storage.pool())
.await?;
Ok(())
}
async fn moved_since<V: WalletServices + ?Sized>(
services: &V,
snapshot: &mut HeaderSnapshot,
mut ring: Vec<(Height, String)>,
) -> (Option<Height>, Vec<(Height, String)>) {
let Some((tip_height, tip_hash)) = snapshot.tip().map(|(h, hash)| (h, hash.to_string())) else {
return (None, ring);
};
let mut moved = None;
while let Some((height, hash)) = ring.last().cloned() {
let active = if height == tip_height {
Some(tip_hash.clone())
} else if height > tip_height {
None
} else {
snapshot.extend(services, [height]).await;
match snapshot.header_at(height) {
Ok(Some(header)) => Some(header.hash),
_ => return (moved, ring),
}
};
if active.as_deref() == Some(hash.as_str()) {
break;
}
moved = Some(height);
ring.pop();
}
if ring.last().map(|(h, _)| *h) != Some(tip_height) {
ring.push((tip_height, tip_hash));
}
if ring.len() > TIP_RING {
let extra = ring.len() - TIP_RING;
ring.drain(..extra);
}
(moved, ring)
}
pub async fn tick<V: WalletServices + ?Sized, C: Clock>(
storage: &StorageSqlx,
services: &V,
clock: &C,
opts: &TickOptions,
) -> Result<TickReport> {
ensure_tables(storage).await?;
let params = opts.params();
let now = clock.now();
let mut report = TickReport::default();
let adopted = sqlx::query(
"INSERT OR IGNORE INTO tracker_states (txid, word, updated_at) \
SELECT DISTINCT lower(txid), 'built', ? FROM transactions \
WHERE txid IS NOT NULL AND status IN ('unproven', 'sending')",
)
.bind(now as i64)
.execute(storage.pool())
.await?;
report.adopted = adopted.rows_affected() as usize;
let mut snapshot = HeaderSnapshot::take(services, []).await;
let (moved_from, ring) = moved_since(services, &mut snapshot, read_ring(storage).await).await;
report.moved_from = moved_from;
let mut rows: Vec<TrackedRow> = sqlx::query(&format!(
"SELECT {ROW_COLUMNS} FROM tracker_states \
WHERE word IN ('unknown', 'built', 'announced', 'seen', 'stale') \
OR (word = 'mined' AND ? IS NOT NULL AND height >= ?) \
ORDER BY txid"
))
.bind(moved_from.map(|h| h as i64))
.bind(moved_from.map(|h| h as i64))
.fetch_all(storage.pool())
.await?
.iter()
.map(row_of)
.collect();
report.tracked = rows.len();
snapshot
.extend(services, rows.iter().filter_map(TrackedRow::proof_height))
.await;
let mut hints = MemoryHints::read(storage, &rows).await?;
let mut states: HashMap<TxId, State> = HashMap::new();
for row in rows.iter_mut() {
let (state, fault) = row.replay(¶ms, &snapshot);
if let Some(fault) = fault {
if matches!(fault, CheckError::RootMismatch(_)) {
row.bump = None;
row.chain.clear();
report.moved.push(row.txid.clone());
tracing::warn!(
marker = "tracker_stored_proof_moved",
txid = %row.txid,
"a stored merkle path no longer meets the active header; asking again"
);
} else {
report.faults.push(format!("{}: {fault}", row.txid));
}
row.asks = 0;
row.next_ask_at = None;
}
states.insert(row.txid.clone(), state);
}
while let Ok(Some((txid, hint))) = hints.next_hint() {
let (Some(state), Some(row)) = (
states.get_mut(&txid),
rows.iter_mut().find(|r| r.txid == txid),
) else {
continue;
};
row.hints.push(StoredHint::of(&hint));
let _ = state.step(¶ms, &snapshot, Input::Hint(hint));
report.hints += 1;
if matches!(state.reask(), Some(Reask::Hint(_))) {
row.asks = 0;
row.next_ask_at = None;
}
}
for row in rows.iter_mut() {
let (Some(beef), Some(at)) = (row.reannounce_beef.clone(), row.reannounce_at) else {
continue;
};
let Some(state) = states.get_mut(&row.txid) else {
continue;
};
if at > now || matches!(state.word(), Word::Mined(_)) {
continue;
}
let pause = Some(now.saturating_add(opts.age_threshold));
let answer = services
.post_beef(&beef, std::slice::from_ref(&row.txid))
.await
.map(|results| read_post(&results));
match answer {
Ok(PostAnswer::Accepted { provider }) => {
let hint = Hint::new(format!("{provider}|accepted"), HintStatus::Accepted, now);
row.hints.push(StoredHint::of(&hint));
let _ = state.step(¶ms, &snapshot, Input::Hint(hint));
row.reannounce_beef = None;
row.reannounce_at = None;
row.asks = 0;
row.next_ask_at = pause;
report.reannounced.push(row.txid.clone());
if let Some(args) = row.on_accept.take() {
report.on_accept.push((row.txid.clone(), args));
}
}
Ok(PostAnswer::NotFinal { .. }) => {
row.asks = row.asks.saturating_add(1);
row.next_ask_at = pause;
if row.asks >= MAX_ASKS_PER_TX {
row.reannounce_at = None;
report.faults.push(format!(
"{}: still not final after {MAX_ASKS_PER_TX} announces; not announced again",
row.txid
));
} else {
row.reannounce_at = pause;
report.not_final.push(row.txid.clone());
}
}
Ok(PostAnswer::Refused { said }) => {
row.reannounce_at = None;
row.next_ask_at = pause;
report.faults.push(format!(
"{}: refused when announced again: {said}",
row.txid
));
}
Err(e) => {
row.reannounce_at = pause;
report
.faults
.push(format!("{}: could not announce again: {e}", row.txid));
}
}
}
for state in states.values_mut() {
let _ = state.step(¶ms, &snapshot, Input::Host(HostAction::Tick(now)));
}
let due: Vec<TxId> = rows
.iter()
.filter(|row| {
states[&row.txid].reask().is_some()
&& row.asks < MAX_ASKS_PER_TX
&& row.next_ask_at.is_none_or(|at| at <= now)
})
.take(opts.max_asks)
.map(|row| row.txid.clone())
.collect();
let mut courier = Courier::carry(services, &due).await;
snapshot.extend(services, courier.heights()).await;
for txid in &due {
let (Some(state), Some(row)) = (
states.get_mut(txid),
rows.iter_mut().find(|r| &r.txid == txid),
) else {
continue;
};
report.asked += 1;
let reask = state.reask().cloned().unwrap_or(Reask::Age);
let fetched = courier.fetch(txid, &reask);
let proven = match fetched {
Ok(Some(proof)) => {
let hex = proof.path().to_hex();
match state.step(¶ms, &snapshot, Input::Evidence(Evidence::Proof(proof))) {
Ok(()) => {
row.bump = Some(hex);
row.chain.clear();
true
}
Err(fault) => {
report.faults.push(format!("{txid}: {fault}"));
false
}
}
}
Ok(None) => {
report.no_proof.push(txid.clone());
false
}
Err(fault) => {
report
.faults
.push(format!("{txid}: could not ask for a proof: {fault}"));
false
}
};
if proven {
row.asks = 0;
row.next_ask_at = None;
report.mined.push(txid.clone());
} else {
let pause = opts
.age_threshold
.saturating_mul(1 << row.asks.min(MAX_BACKOFF_DOUBLINGS));
row.asks = row.asks.saturating_add(1);
row.next_ask_at = Some(now.saturating_add(pause));
}
}
for row in &rows {
let state = &states[&row.txid];
store_row(storage, row, state, now).await?;
if let (true, Word::Mined(mined)) = (report.mined.contains(&row.txid), state.word()) {
let checked = mined.checked();
let outcome = storage
.ingest_merkle_proof(
&row.txid,
&checked.proof().path().to_binary(),
checked.height(),
&checked.header().hash,
Some(checked.root()),
)
.await;
report.stored.push(match outcome {
Ok(outcome) => format!("{}: {outcome:?}", row.txid),
Err(e) => format!("{}: not stored: {e}", row.txid),
});
}
}
adopt_proven(storage, services, &mut snapshot, ¶ms, now, &mut report).await?;
write_ring(storage, &ring).await?;
Ok(report)
}
pub const MAX_ADOPT_PER_PASS: usize = 100;
async fn adopt_proven<V: WalletServices + ?Sized>(
storage: &StorageSqlx,
services: &V,
snapshot: &mut HeaderSnapshot,
params: &Params,
now: Timestamp,
report: &mut TickReport,
) -> Result<()> {
let found = sqlx::query(
"SELECT lower(t.txid) AS txid, MAX(t.is_outgoing) AS made, p.merkle_path \
FROM transactions t \
JOIN proven_txs p ON p.proven_tx_id = t.proven_tx_id \
WHERE t.txid IS NOT NULL AND t.status = 'completed' \
AND NOT EXISTS (SELECT 1 FROM tracker_states s WHERE s.txid = lower(t.txid)) \
AND EXISTS (SELECT 1 FROM outputs o \
WHERE o.transaction_id = t.transaction_id AND o.spent_by IS NULL) \
GROUP BY lower(t.txid) ORDER BY MIN(t.transaction_id) LIMIT ?",
)
.bind(MAX_ADOPT_PER_PASS as i64)
.fetch_all(storage.pool())
.await?;
let mut rows = Vec::with_capacity(found.len());
for r in &found {
let txid: String = r.get("txid");
let path: Option<Vec<u8>> = r.get("merkle_path");
let Some(path) = path
.as_deref()
.and_then(|bytes| MerklePath::from_binary(bytes).ok())
else {
report.faults.push(format!(
"{txid}: the stored proof does not parse; not adopted"
));
continue;
};
rows.push(TrackedRow {
built: r.get::<i64, _>("made") != 0,
bump: Some(path.to_hex()),
..TrackedRow::new(&txid)
});
}
snapshot
.extend(services, rows.iter().filter_map(TrackedRow::proof_height))
.await;
for row in rows {
let (state, fault) = row.replay(params, &*snapshot);
match fault {
None if matches!(state.word(), Word::Mined(_)) => {
report.adopted_mined.push(row.txid.clone());
}
Some(CheckError::RootMismatch(_)) => {
tracing::warn!(
marker = "tracker_adopted_proof_failed",
txid = %row.txid,
reask = ?state.reask(),
"a proven creator's stored merkle path does not meet the active header; tracked with its re-ask"
);
report.adopted_proof_failed.push(row.txid.clone());
}
Some(fault) => {
report
.faults
.push(format!("{}: not adopted: {fault}", row.txid));
continue;
}
None => {
report.faults.push(format!(
"{}: not adopted: the stored proof gave the word {}",
row.txid,
word_label(state.word())
));
continue;
}
}
store_row(storage, &row, &state, now).await?;
}
Ok(())
}
pub async fn record_accepted<W: WalletInterface>(wallet: &W, report: &mut TickReport) {
for (txid, args) in std::mem::take(&mut report.on_accept) {
let line = match serde_json::from_str::<InternalizeActionArgs>(&args) {
Ok(args) => match wallet.internalize_action(args, "bsv-wallet-cli").await {
Ok(result) if result.accepted => format!("{txid}: recorded in the wallet"),
Ok(_) => format!("{txid}: the wallet did not accept its own transaction"),
Err(e) => format!(
"{txid}: not recorded in the wallet ({e}); `bsv-wallet receive {txid}` records it once it is mined"
),
},
Err(e) => format!("{txid}: not recorded in the wallet (stored arguments: {e})"),
};
if !line.ends_with("recorded in the wallet") {
tracing::warn!(marker = "own_transaction_not_recorded", "{line}");
}
report.stored.push(line);
}
}
fn readme_spend_guard<H: Headers>(
state: &mut State,
params: &Params,
headers: &H,
) -> std::result::Result<bool, CheckError> {
state.step(params, headers, Input::Host(HostAction::SpendAttempt))?;
state.step(params, headers, Input::Evidence(Evidence::Recheck))?;
Ok(matches!(state.word(), Word::Mined(_)) && !state.suspect())
}
pub async fn spend_guard<V: WalletServices + ?Sized, C: Clock>(
storage: &StorageSqlx,
services: &V,
clock: &C,
opts: &TickOptions,
txid: &str,
) -> Result<std::result::Result<bool, CheckError>> {
ensure_tables(storage).await?;
let Some(row) = load_row(storage, txid).await? else {
return Ok(Ok(false));
};
let params = opts.params();
let snapshot = HeaderSnapshot::take(services, row.proof_height()).await;
let (mut state, fault) = row.replay(¶ms, &snapshot);
match fault {
Some(fault) if !matches!(fault, CheckError::RootMismatch(_)) => return Ok(Err(fault)),
_ => {}
}
let split = under_split(state.word(), open_split_floor(storage).await?);
let verdict = readme_spend_guard(&mut state, ¶ms, &snapshot).and_then(|offered| {
Ok(offered && !split && deep_enough(&state, &snapshot, opts.spend_depth)?)
});
store_row(storage, &row, &state, clock.now()).await?;
Ok(verdict)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreatorReading {
pub word: &'static str,
pub depth: u64,
pub suspect: bool,
pub offered: bool,
pub fault: Option<String>,
}
pub async fn read_creators<V: WalletServices + ?Sized>(
storage: &StorageSqlx,
services: &V,
opts: &TickOptions,
creators: &[(String, Option<Vec<u8>>)],
) -> Result<HashMap<String, CreatorReading>> {
ensure_tables(storage).await?;
let params = opts.params();
let mut rows = Vec::with_capacity(creators.len());
for (txid, stored_path) in creators {
let mut row = match load_row(storage, txid).await? {
Some(row) => row,
None => TrackedRow {
built: false,
..TrackedRow::new(txid)
},
};
if row.bump.is_none() {
row.bump = stored_path
.as_deref()
.and_then(|bytes| MerklePath::from_binary(bytes).ok())
.map(|path| path.to_hex());
}
rows.push(row);
}
let snapshot =
HeaderSnapshot::take(services, rows.iter().filter_map(TrackedRow::proof_height)).await;
let split_floor = open_split_floor(storage).await?;
let mut readings = HashMap::with_capacity(rows.len());
for row in rows {
let (mut state, fault) = row.replay(¶ms, &snapshot);
let fault = match fault {
Some(fault) if !matches!(fault, CheckError::RootMismatch(_)) => Some(fault.to_string()),
_ => None,
};
let verdict = match fault {
Some(_) => Ok(false),
None => readme_spend_guard(&mut state, ¶ms, &snapshot).and_then(|offered| {
Ok(offered && deep_enough(&state, &snapshot, opts.spend_depth)?)
}),
};
let depth = match (state.word().height(), snapshot.tip()) {
(Some(height), Some((tip, _))) => {
(u64::from(tip) + 1).saturating_sub(u64::from(height))
}
_ => 0,
};
let (offered, fault) = match verdict {
Ok(offered) => (offered, fault),
Err(e) => (false, fault.or(Some(e.to_string()))),
};
let split = under_split(state.word(), split_floor);
readings.insert(
row.txid.clone(),
CreatorReading {
word: word_label(state.word()),
depth,
suspect: state.suspect() || split,
offered: offered && !split,
fault,
},
);
}
Ok(readings)
}
fn deep_enough<H: Headers>(
state: &State,
headers: &H,
spend_depth: Height,
) -> std::result::Result<bool, CheckError> {
let Some(height) = state.word().height() else {
return Ok(false);
};
let tip = headers.tip_height().map_err(CheckError::Headers)?;
let depth = (u64::from(tip) + 1).saturating_sub(u64::from(height));
Ok(depth >= u64::from(spend_depth.max(DEFAULT_SPEND_DEPTH)))
}
async fn read_open_splits<'e, E: sqlx::Executor<'e, Database = sqlx::Sqlite>>(
executor: E,
) -> Result<BTreeMap<String, Height>> {
let value: Option<String> =
sqlx::query_scalar("SELECT value FROM tracker_meta WHERE key = 'open_splits'")
.fetch_optional(executor)
.await?;
Ok(match value {
Some(v) => serde_json::from_str(&v)?,
None => BTreeMap::new(),
})
}
pub async fn open_split_floor(storage: &StorageSqlx) -> Result<Option<Height>> {
ensure_tables(storage).await?;
Ok(read_open_splits(storage.pool()).await?.into_values().min())
}
fn under_split(word: &Word, floor: Option<Height>) -> bool {
matches!((word, floor), (Word::Mined(m), Some(floor)) if floor <= m.height())
}
pub async fn feed_cursor(storage: &StorageSqlx) -> Result<u64> {
ensure_tables(storage).await?;
let value: Option<String> =
sqlx::query_scalar("SELECT value FROM tracker_meta WHERE key = 'feed_cursor'")
.fetch_optional(storage.pool())
.await?;
Ok(match value {
Some(v) => v
.parse()
.map_err(|e| anyhow::anyhow!("the stored feed cursor {v:?}: {e}"))?,
None => 0,
})
}
async fn ensure_frozen(pool: &sqlx::SqlitePool) -> Result<()> {
sqlx::query(
"CREATE TABLE IF NOT EXISTS frozen_outpoints (\
txid TEXT NOT NULL, \
vout INTEGER NOT NULL, \
cursor INTEGER NOT NULL, \
frozen_at INTEGER NOT NULL, \
PRIMARY KEY (txid, vout))",
)
.execute(pool)
.await?;
Ok(())
}
pub async fn frozen_outpoints(
pool: &sqlx::SqlitePool,
) -> Result<std::collections::HashSet<(String, i64)>> {
ensure_frozen(pool).await?;
let rows = sqlx::query("SELECT txid, vout FROM frozen_outpoints")
.fetch_all(pool)
.await?;
Ok(rows
.iter()
.map(|r| (r.get::<String, _>("txid"), r.get::<i64, _>("vout")))
.collect())
}
pub async fn frozen_count(pool: &sqlx::SqlitePool) -> Result<u64> {
ensure_frozen(pool).await?;
let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM frozen_outpoints")
.fetch_one(pool)
.await?;
Ok(n.max(0) as u64)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ChainMove {
pub txid: String,
pub word: String,
pub reask: Option<String>,
pub suspect: bool,
pub fault: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ChainDelivery {
pub cursor: u64,
pub kind: &'static str,
pub moved: Vec<ChainMove>,
}
pub fn event_kind(event: &ChainEvent) -> &'static str {
match event {
ChainEvent::Tip { .. } => "tip",
ChainEvent::Fork { .. } => "fork",
ChainEvent::Reorg { .. } => "reorg",
ChainEvent::Invalidated { .. } => "invalidated",
ChainEvent::Frozen { .. } => "frozen",
ChainEvent::TipAge { .. } => "tipAge",
ChainEvent::Split { .. } => "split",
ChainEvent::SplitResolved { .. } => "splitResolved",
}
}
struct Replaced<'a> {
snapshot: &'a HeaderSnapshot,
replaced: BTreeMap<Height, Header>,
prior_tip: Option<Height>,
}
impl Headers for Replaced<'_> {
fn header_at(&self, height: Height) -> std::result::Result<Option<Header>, HeaderError> {
match self.replaced.get(&height) {
Some(header) => Ok(Some(header.clone())),
None => self.snapshot.header_at(height),
}
}
fn tip_height(&self) -> std::result::Result<Height, HeaderError> {
match self.prior_tip {
Some(tip) => Ok(tip),
None => self.snapshot.tip_height(),
}
}
}
pub async fn deliver_chain_event<V: WalletServices + ?Sized, C: Clock>(
storage: &StorageSqlx,
services: &V,
clock: &C,
opts: &TickOptions,
cursor: u64,
event: ChainEvent,
) -> Result<ChainDelivery> {
let kind = event_kind(&event);
if cursor <= feed_cursor(storage).await? {
return Ok(ChainDelivery {
cursor,
kind,
moved: vec![],
});
}
let named = format!("SELECT {ROW_COLUMNS} FROM tracker_states");
let query = match &event {
ChainEvent::Fork { height: from, .. }
| ChainEvent::Reorg {
fork_height: from, ..
}
| ChainEvent::Split {
fork_height: from, ..
}
| ChainEvent::SplitResolved {
fork_height: from, ..
} => Some(
sqlx::query(&format!(
"{named} WHERE word = 'mined' AND height >= ? ORDER BY txid"
))
.bind(i64::from(*from))
.fetch_all(storage.pool())
.await?,
),
ChainEvent::Tip { .. } => Some(
sqlx::query(&format!(
"{named} WHERE word = 'mined' AND chain <> '[]' ORDER BY txid"
))
.fetch_all(storage.pool())
.await?,
),
ChainEvent::Invalidated { block_hash } => Some(
sqlx::query(&format!("{named} WHERE header_hash = ? ORDER BY txid"))
.bind(block_hash.to_ascii_lowercase())
.fetch_all(storage.pool())
.await?,
),
ChainEvent::Frozen { .. } | ChainEvent::TipAge { .. } => None,
};
let mut rows: Vec<TrackedRow> = query.iter().flatten().map(row_of).collect();
let now = clock.now();
let mut moved = Vec::new();
if matches!(event, ChainEvent::Frozen { .. }) {
ensure_frozen(storage.pool()).await?;
}
let mut tx = storage.pool().begin().await?;
if let ChainEvent::Frozen { outpoint } = &event {
sqlx::query(
"INSERT INTO frozen_outpoints (txid, vout, cursor, frozen_at) VALUES (?, ?, ?, ?) \
ON CONFLICT(txid, vout) DO NOTHING",
)
.bind(outpoint.txid.to_ascii_lowercase())
.bind(i64::from(outpoint.vout))
.bind(cursor as i64)
.bind(now as i64)
.execute(&mut *tx)
.await?;
}
if !rows.is_empty() {
let snapshot =
HeaderSnapshot::take(services, rows.iter().filter_map(TrackedRow::proof_height)).await;
let (replaced, prior_tip) = match &event {
ChainEvent::Reorg {
deactivated_headers,
..
} => (
deactivated_headers
.iter()
.map(|h| {
(
h.height,
Header {
hash: h.hash.to_ascii_lowercase(),
merkle_root: h.merkle_root.to_ascii_lowercase(),
},
)
})
.collect(),
deactivated_headers.first().map(|h| h.height),
),
_ => (BTreeMap::new(), None),
};
let replay_view = Replaced {
snapshot: &snapshot,
replaced,
prior_tip,
};
let mut tracker = Tracker::new(opts.params());
for row in &rows {
row.replay_into(&mut tracker, &replay_view);
}
for update in tracker.on_chain(&snapshot, event.clone()) {
let (Some(state), Some(row)) = (
tracker.get(&update.txid),
rows.iter_mut().find(|r| r.txid == update.txid),
) else {
continue;
};
row.keep_chain(state, &event)?;
if matches!(update.reask, Some(Reask::Reorg { .. })) {
row.asks = 0;
row.next_ask_at = None;
}
store_row_in(&mut *tx, row, state, now).await?;
let fault = update.result.err().map(|e| e.to_string());
if let Some(fault) = &fault {
tracing::warn!(txid = %update.txid, kind, cursor, %fault, "a chain event's recheck could not look");
}
moved.push(ChainMove {
txid: update.txid,
word: word_label(state.word()).to_string(),
reask: state.reask().map(reask_label),
suspect: state.suspect(),
fault,
});
}
}
if let ChainEvent::Split {
key, fork_height, ..
}
| ChainEvent::SplitResolved {
key, fork_height, ..
} = &event
{
let mut open = read_open_splits(&mut *tx).await?;
if matches!(event, ChainEvent::Split { .. }) {
open.insert(key.clone(), *fork_height);
} else {
open.remove(key);
}
sqlx::query(
"INSERT INTO tracker_meta (key, value) VALUES ('open_splits', ?) \
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
)
.bind(serde_json::to_string(&open)?)
.execute(&mut *tx)
.await?;
}
sqlx::query(
"INSERT INTO tracker_meta (key, value) VALUES ('feed_cursor', ?) \
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
)
.bind(cursor.to_string())
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(ChainDelivery {
cursor,
kind,
moved,
})
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use bsv_wallet_toolbox::services::mock::{MockResponse, MockWalletServices};
use bsv_wallet_toolbox::services::{BlockHeader, GetMerklePathResult};
use bsv_wallet_toolbox::{
WalletStorageWriter, BROADCAST_STATUS_ACCEPTED, BROADCAST_STATUS_MINED,
};
use chrono::Utc;
pub(crate) struct FixedClock(pub Timestamp);
impl Clock for FixedClock {
fn now(&self) -> Timestamp {
self.0
}
}
pub(crate) fn opts() -> TickOptions {
TickOptions {
age_threshold: 600,
max_asks: 20,
spend_depth: DEFAULT_SPEND_DEPTH,
}
}
pub(crate) async fn wallet_storage() -> (StorageSqlx, i64) {
let storage = StorageSqlx::in_memory().await.unwrap();
storage
.migrate("tracker-tests", &("02".to_string() + &"ab".repeat(32)))
.await
.unwrap();
storage.make_available().await.unwrap();
let (user, _) = storage
.find_or_insert_user(&("02".to_string() + &"cd".repeat(32)))
.await
.unwrap();
(storage, user.user_id)
}
pub(crate) async fn insert_tx(storage: &StorageSqlx, user_id: i64, txid: &str, status: &str) {
sqlx::query(
"INSERT INTO transactions (user_id, status, reference, is_outgoing, satoshis, version, lock_time, description, txid, raw_tx, created_at, updated_at) \
VALUES (?, ?, ?, 1, 0, 1, 0, 'd', ?, X'01000000', ?, ?)",
)
.bind(user_id)
.bind(status)
.bind(&txid[..8])
.bind(txid)
.bind(Utc::now())
.bind(Utc::now())
.execute(storage.pool())
.await
.unwrap();
}
pub(crate) async fn proven_coin(
storage: &StorageSqlx,
user_id: i64,
txid: &str,
height: u32,
satoshis: i64,
) {
let basket = storage
.find_or_create_default_basket(user_id)
.await
.unwrap()
.basket_id;
let proven = sqlx::query(
"INSERT INTO proven_txs (txid, height, idx, block_hash, merkle_root, merkle_path, raw_tx, created_at, updated_at) \
VALUES (?, ?, 0, ?, ?, ?, X'01000000', ?, ?)",
)
.bind(txid)
.bind(height as i64)
.bind("bb".repeat(32))
.bind(txid)
.bind(MerklePath::from_coinbase_txid(txid, height).to_binary())
.bind(Utc::now())
.bind(Utc::now())
.execute(storage.pool())
.await
.unwrap()
.last_insert_rowid();
let tx = sqlx::query(
"INSERT INTO transactions (user_id, status, reference, is_outgoing, satoshis, version, lock_time, description, txid, raw_tx, proven_tx_id, created_at, updated_at) \
VALUES (?, 'completed', ?, 0, ?, 1, 0, 'Internalize external funding', ?, X'01000000', ?, ?, ?)",
)
.bind(user_id)
.bind(&txid[..8])
.bind(satoshis)
.bind(txid)
.bind(proven)
.bind(Utc::now())
.bind(Utc::now())
.execute(storage.pool())
.await
.unwrap()
.last_insert_rowid();
sqlx::query(
"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) \
VALUES (?, ?, ?, 0, ?, X'76a914dbc0a7c84983c5bf199b7b2d41b3acf0408ee5aa88ac', ?, 'P2PKH', 1, 1, 'storage', 'change', 'c', ?, ?)",
)
.bind(user_id)
.bind(tx)
.bind(basket)
.bind(satoshis)
.bind(txid)
.bind(Utc::now())
.bind(Utc::now())
.execute(storage.pool())
.await
.unwrap();
}
pub(crate) fn header(height: u32, root: &str, nonce: u32) -> BlockHeader {
BlockHeader {
version: 1,
previous_hash: "00".repeat(32),
merkle_root: root.to_string(),
time: 1_700_000_000,
bits: 0x1d00_ffff,
nonce,
hash: String::new(),
height,
}
}
pub(crate) fn path_of(txid: &str, height: u32) -> String {
MerklePath::from_coinbase_txid(txid, height).to_hex()
}
pub(crate) fn proof_answer(path: Option<String>) -> MockResponse<GetMerklePathResult> {
MockResponse::Success(GetMerklePathResult {
name: Some("Services".to_string()),
merkle_path: path,
header: None,
error: None,
notes: vec![],
})
}
#[tokio::test]
async fn a_pass_tracks_announces_and_proves_one_named_transaction() {
let (storage, user) = wallet_storage().await;
let txid = "a1".repeat(32);
insert_tx(&storage, user, &txid, "unproven").await;
storage
.record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_ACCEPTED)
.await
.unwrap();
let services = MockWalletServices::builder()
.get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
.build();
services.set_header_for_height(header(900, &txid, 0));
let now = Utc::now().timestamp() as u64;
let report = tick(&storage, &services, &FixedClock(now), &opts())
.await
.unwrap();
assert_eq!((report.adopted, report.hints, report.asked), (1, 1, 0));
assert_eq!(
stored_word(&storage, &txid).await.unwrap(),
Some(("announced".to_string(), None, None))
);
assert_eq!(services.call_count("get_merkle_path"), 0);
let report = tick(&storage, &services, &FixedClock(now + 601), &opts())
.await
.unwrap();
assert_eq!(report.asked, 1);
assert_eq!(report.mined, vec![txid.clone()]);
assert_eq!(
stored_word(&storage, &txid).await.unwrap(),
Some(("mined".to_string(), Some(900), None))
);
assert_eq!(services.call_count("get_merkle_path"), 1);
let report = tick(&storage, &services, &FixedClock(now + 1300), &opts())
.await
.unwrap();
assert_eq!((report.tracked, report.asked), (0, 0));
assert_eq!(services.call_count("get_merkle_path"), 1);
}
#[tokio::test]
async fn a_broadcasters_mined_is_a_hint_that_asks_for_the_proof() {
let (storage, user) = wallet_storage().await;
let txid = "b2".repeat(32);
insert_tx(&storage, user, &txid, "unproven").await;
storage
.record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
.await
.unwrap();
let services = MockWalletServices::builder()
.get_merkle_path_response(proof_answer(None))
.build();
let now = Utc::now().timestamp() as u64;
let report = tick(&storage, &services, &FixedClock(now), &opts())
.await
.unwrap();
assert_eq!(report.asked, 1);
assert_eq!(report.no_proof, vec![txid.clone()]);
assert!(report.mined.is_empty());
let (word, height, reask) = stored_word(&storage, &txid).await.unwrap().unwrap();
assert_eq!((word.as_str(), height), ("built", None));
assert_eq!(reask.as_deref(), Some("hint:arcade|mined"));
let report = tick(&storage, &services, &FixedClock(now + 60), &opts())
.await
.unwrap();
assert_eq!(report.asked, 0);
let report = tick(&storage, &services, &FixedClock(now + 600), &opts())
.await
.unwrap();
assert_eq!(report.asked, 1);
let report = tick(&storage, &services, &FixedClock(now + 1200), &opts())
.await
.unwrap();
assert_eq!(report.asked, 0);
assert_eq!(services.call_count("get_merkle_path"), 2);
}
#[tokio::test]
async fn a_moved_header_reasks_the_rows_at_or_above_it_and_no_other() {
let (storage, user) = wallet_storage().await;
let low = "c3".repeat(32);
let high = "d4".repeat(32);
let services = MockWalletServices::builder()
.get_merkle_path_response(MockResponse::Sequence(vec![
proof_answer(Some(path_of(&low, 100))),
proof_answer(Some(path_of(&high, 200))),
proof_answer(Some(path_of(&high, 201))),
]))
.build();
services.set_header_for_height(header(100, &low, 0));
services.set_header_for_height(header(200, &high, 0));
services.set_tip_header(header_with_hash(200, &high, 0));
let now = Utc::now().timestamp() as u64;
for txid in [&low, &high] {
insert_tx(&storage, user, txid, "unproven").await;
storage
.record_broadcast_status(txid, "arcade", BROADCAST_STATUS_MINED)
.await
.unwrap();
}
let report = tick(&storage, &services, &FixedClock(now), &opts())
.await
.unwrap();
assert_eq!(report.mined.len(), 2, "{report:?}");
services.set_header_for_height(header(200, &"ee".repeat(32), 7));
services.set_header_for_height(header(201, &high, 0));
services.set_tip_header(header_with_hash(201, &high, 0));
let report = tick(&storage, &services, &FixedClock(now + 60), &opts())
.await
.unwrap();
assert_eq!(report.moved_from, Some(200));
assert_eq!(
report.tracked, 1,
"only the row at or above the moved header"
);
assert_eq!(report.moved, vec![high.clone()]);
assert_eq!(report.mined, vec![high.clone()]);
assert_eq!(
stored_word(&storage, &high).await.unwrap().unwrap().1,
Some(201)
);
assert_eq!(
stored_word(&storage, &low).await.unwrap().unwrap().1,
Some(100)
);
}
#[tokio::test]
async fn a_not_final_transaction_is_announced_again_at_its_lock_time() {
use bsv_wallet_toolbox::services::mock::{
error_post_beef_result, success_post_beef_result,
};
let (storage, _user) = wallet_storage().await;
let txid = "c7".repeat(32);
let lock = 2_000_000_000u64;
let services = MockWalletServices::builder()
.post_beef_response(MockResponse::Sequence(vec![
MockResponse::Success(vec![error_post_beef_result("Arcade", "476 non-final")]),
MockResponse::Success(vec![success_post_beef_result("Arcade", &[&txid])]),
]))
.get_merkle_path_response(proof_answer(None))
.build();
let hint = Hint::new(
"Arcade|rejected",
HintStatus::Rejected {
reason: "476 non-final".to_string(),
},
lock - 3600,
);
let again = Reannounce {
beef: vec![1, 2, 3],
at: lock,
on_accept: None,
};
let word = heard(
&storage,
&FixedClock(lock - 3600),
&opts(),
&txid,
hint,
Some(lock),
Some(again),
)
.await
.unwrap();
assert_eq!(
word,
(
"built".to_string(),
Some("hint:Arcade|rejected".to_string())
)
);
let report = tick(&storage, &services, &FixedClock(lock - 1), &opts())
.await
.unwrap();
assert_eq!((report.tracked, report.asked), (1, 0), "{report:?}");
assert_eq!(services.call_count("post_beef"), 0);
let report = tick(&storage, &services, &FixedClock(lock), &opts())
.await
.unwrap();
assert_eq!(report.not_final, vec![txid.clone()]);
assert_eq!(services.call_count("post_beef"), 1);
let report = tick(&storage, &services, &FixedClock(lock + 60), &opts())
.await
.unwrap();
assert!(report.not_final.is_empty() && report.reannounced.is_empty());
assert_eq!(services.call_count("post_beef"), 1, "one pause between two");
let report = tick(&storage, &services, &FixedClock(lock + 600), &opts())
.await
.unwrap();
assert_eq!(report.reannounced, vec![txid.clone()]);
assert_eq!(services.call_count("post_beef"), 2);
assert_eq!(
stored_word(&storage, &txid).await.unwrap().unwrap().0,
"announced"
);
assert_eq!(services.call_count("get_merkle_path"), 0);
let report = tick(&storage, &services, &FixedClock(lock + 1300), &opts())
.await
.unwrap();
assert_eq!(report.asked, 1);
assert_eq!(services.call_count("post_beef"), 2);
assert_eq!(services.call_count("get_merkle_path"), 1);
}
#[test]
fn not_final_is_read_from_the_refusal() {
for said in [
"476 non-final",
"bad-txns-nonfinal",
"tx is not final",
"code 476",
] {
assert!(says_not_final(said), "{said}");
}
for said in [
"fee too low",
"DOUBLE_SPEND_ATTEMPTED",
"txid 4476aa",
"HTTP 500",
] {
assert!(!says_not_final(said), "{said}");
}
}
pub(crate) fn header_with_hash(height: u32, root: &str, nonce: u32) -> BlockHeader {
let mut h = header(height, root, nonce);
h.hash = header_of(&h.to_binary()).unwrap().hash;
h
}
#[tokio::test]
async fn the_spend_depth_holds_a_coin_until_its_block_is_deep_enough() {
let (storage, user) = wallet_storage().await;
let txid = "e7".repeat(32);
insert_tx(&storage, user, &txid, "unproven").await;
storage
.record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
.await
.unwrap();
let services = MockWalletServices::builder()
.get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
.build();
services.set_header_for_height(header(900, &txid, 0));
services.set_tip_header(header_with_hash(900, &txid, 0));
let clock = FixedClock(Utc::now().timestamp() as u64);
tick(&storage, &services, &clock, &opts()).await.unwrap();
let at = |spend_depth| TickOptions {
spend_depth,
..opts()
};
let one = spend_guard(&storage, &services, &clock, &at(1), &txid)
.await
.unwrap();
let three = spend_guard(&storage, &services, &clock, &at(3), &txid)
.await
.unwrap();
let zero = spend_guard(&storage, &services, &clock, &at(0), &txid)
.await
.unwrap();
println!("one block deep: depth 1 {one:?}, depth 3 {three:?}, depth 0 {zero:?}");
assert_eq!(one, Ok(true));
assert_eq!(three, Ok(false));
assert_eq!(zero, Ok(true), "0 reads as the floor, 1");
services.set_tip_header(header_with_hash(902, &"0c".repeat(32), 0));
let three = spend_guard(&storage, &services, &clock, &at(3), &txid)
.await
.unwrap();
println!("three blocks deep: depth 3 {three:?}");
assert_eq!(three, Ok(true));
assert_eq!(
TickOptions {
spend_depth: 0,
..opts()
}
.floored()
.spend_depth,
1
);
}
#[tokio::test]
async fn the_readme_spend_guard_runs_as_the_host() {
let (storage, user) = wallet_storage().await;
let txid = "e5".repeat(32);
insert_tx(&storage, user, &txid, "unproven").await;
storage
.record_broadcast_status(&txid, "arcade", BROADCAST_STATUS_MINED)
.await
.unwrap();
let services = MockWalletServices::builder()
.get_merkle_path_response(proof_answer(Some(path_of(&txid, 900))))
.build();
services.set_header_for_height(header(900, &txid, 0));
let clock = FixedClock(Utc::now().timestamp() as u64);
tick(&storage, &services, &clock, &opts()).await.unwrap();
let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
.await
.unwrap();
println!(
"spend guard, header in place: {guard:?}, stored {:?}",
stored_word(&storage, &txid).await.unwrap().unwrap()
);
assert_eq!(guard, Ok(true));
services.set_tip_unavailable();
let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
.await
.unwrap();
println!(
"spend guard, no header service: {guard:?}, stored {:?}",
stored_word(&storage, &txid).await.unwrap().unwrap()
);
assert!(matches!(guard, Err(CheckError::Headers(_))), "{guard:?}");
assert_eq!(
stored_word(&storage, &txid).await.unwrap().unwrap().0,
"mined",
"could not look changes no word"
);
services.set_tip_header(header_with_hash(901, &"00".repeat(32), 1));
services.set_header_for_height(header(900, &"ee".repeat(32), 7));
let guard = spend_guard(&storage, &services, &clock, &opts(), &txid)
.await
.unwrap();
println!(
"spend guard, header moved: {guard:?}, stored {:?}",
stored_word(&storage, &txid).await.unwrap().unwrap()
);
assert_eq!(guard, Ok(false));
let guard = spend_guard(&storage, &services, &clock, &opts(), &"f6".repeat(32))
.await
.unwrap();
assert_eq!(guard, Ok(false));
let params = opts().params();
services.set_tip_header(header_with_hash(900, &txid, 0));
services.set_header_for_height(header(900, &txid, 0));
let held = HeaderSnapshot::take(&services, [900]).await;
let row = load_row(&storage, &txid).await.unwrap().unwrap();
let (mut state, fault) = row.replay(¶ms, &held);
assert_eq!(fault, None);
let lost = HeaderSnapshot::take(&services, []).await;
let guard = readme_spend_guard(&mut state, ¶ms, &lost);
println!("spend guard, height unanswered: {guard:?}");
assert_eq!(guard, Err(CheckError::Unavailable(900)));
}
#[tokio::test]
async fn the_served_pass_adopts_an_unproven_creator_and_a_proven_completed_one() {
let (storage, user) = wallet_storage().await;
let unproven = "d1".repeat(32);
let earlier = "d2".repeat(32);
insert_tx(&storage, user, &unproven, "unproven").await;
insert_tx(&storage, user, &earlier, "unproven").await;
let services = MockWalletServices::builder().build();
services.set_tip_header(header_with_hash(960, &"0a".repeat(32), 0));
let now = Utc::now().timestamp() as u64;
let pass = crate::broadcast_reconcile::run_served_pass(
&storage,
&services,
&FixedClock(now),
&opts(),
None,
)
.await
.unwrap();
println!("{}", pass.summary());
assert_eq!(pass.tracker.adopted, 2);
assert_eq!(
stored_word(&storage, &earlier).await.unwrap(),
Some(("built".to_string(), None, None))
);
sqlx::query("UPDATE transactions SET status = 'completed' WHERE txid = ?")
.bind(&earlier)
.execute(storage.pool())
.await
.unwrap();
let checks = "d3".repeat(32);
let moved = "d4".repeat(32);
proven_coin(&storage, user, &checks, 900, 3_000).await;
proven_coin(&storage, user, &moved, 950, 2_000).await;
services.set_header_for_height(header(900, &checks, 0));
services.set_header_for_height(header(950, &"ee".repeat(32), 0));
let toolbox_before: Vec<(String, String, Option<i64>)> =
sqlx::query_as("SELECT txid, status, proven_tx_id FROM transactions ORDER BY txid")
.fetch_all(storage.pool())
.await
.unwrap();
let pass = crate::broadcast_reconcile::run_served_pass(
&storage,
&services,
&FixedClock(now + 60),
&opts(),
None,
)
.await
.unwrap();
println!("{}", pass.summary());
assert_eq!(pass.tracker.adopted_mined, vec![checks.clone()]);
assert_eq!(pass.tracker.adopted_proof_failed, vec![moved.clone()]);
assert_eq!(
stored_word(&storage, &checks).await.unwrap(),
Some(("mined".to_string(), Some(900), None))
);
let row = load_row(&storage, &checks).await.unwrap().unwrap();
assert_eq!(
row.bump,
Some(path_of(&checks, 900)),
"the evidence is the stored path"
);
assert!(!row.built, "a received coin's creator was not built here");
assert_eq!(
stored_word(&storage, &moved).await.unwrap(),
Some((
"unknown".to_string(),
None,
Some("proof_failed@950".to_string())
))
);
assert_eq!(
stored_word(&storage, &earlier).await.unwrap(),
Some(("built".to_string(), None, None)),
"a creator tracked while unproven keeps its row"
);
let toolbox_after: Vec<(String, String, Option<i64>)> =
sqlx::query_as("SELECT txid, status, proven_tx_id FROM transactions ORDER BY txid")
.fetch_all(storage.pool())
.await
.unwrap();
assert_eq!(
toolbox_after, toolbox_before,
"no write to the toolbox's rows"
);
assert_eq!(services.call_count("get_merkle_path"), 0);
let pass = crate::broadcast_reconcile::run_served_pass(
&storage,
&services,
&FixedClock(now + 120),
&opts(),
None,
)
.await
.unwrap();
println!("{}", pass.summary());
assert!(pass.tracker.adopted_mined.is_empty());
assert!(pass.tracker.adopted_proof_failed.is_empty());
}
}