use std::time::{Duration, Instant};
use bsv_wallet_toolbox::{services::ARCADE_V2_MAINNET, Chain};
use reqwest::Client;
const DEFAULT_ATTEMPTS: u32 = 12;
const DEFAULT_DELAY_MS: u64 = 2500;
const PROBE_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BroadcastVerification {
Confirmed,
Rejected,
Inconclusive,
}
impl BroadcastVerification {
pub fn into_send_result(self, txid: &str) -> anyhow::Result<()> {
match self {
BroadcastVerification::Rejected => Err(anyhow::anyhow!(
"broadcast rejected: transaction {txid} is absent from BOTH the broadcaster \
it was submitted to AND an independent chain index, after the full probe \
window. The broadcaster dropped it — most likely error 465 \"fee too low\", \
because a monitor-less wallet presented a deep unconfirmed BEEF and ARC \
charged the fee for the whole unconfirmed package. The funds were NOT sent. \
Fetch merkle proofs for the confirmed ancestors (run `bsv-wallet tick` with \
CHAINTRACKS_URL set) or fund from a confirmed UTXO, then retry."
)),
BroadcastVerification::Confirmed | BroadcastVerification::Inconclusive => Ok(()),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Presence {
Present,
Absent,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AbsenceAuthority {
None,
Broadcaster,
ChainIndex,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct AbsenceVotes {
broadcaster: bool,
chain_index: bool,
}
impl AbsenceVotes {
fn record(&mut self, authority: AbsenceAuthority) {
match authority {
AbsenceAuthority::Broadcaster => self.broadcaster = true,
AbsenceAuthority::ChainIndex => self.chain_index = true,
AbsenceAuthority::None => {}
}
}
fn is_definitive(self) -> bool {
self.broadcaster && self.chain_index
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum BroadcastPlane {
ArcadeV2 { base: String },
ClassicArc { base: String },
}
impl BroadcastPlane {
fn resolve(chain: Chain, arcade_mode: bool, arc_url: Option<String>) -> Self {
let arc_url = arc_url
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty());
if arcade_mode {
BroadcastPlane::ArcadeV2 {
base: normalize_base(&arc_url.unwrap_or_else(|| ARCADE_V2_MAINNET.to_string())),
}
} else {
BroadcastPlane::ClassicArc {
base: normalize_base(&arc_url.unwrap_or_else(|| taal_arc_url(chain).to_string())),
}
}
}
fn from_env(chain: Chain) -> Self {
Self::resolve(
chain,
crate::services_env::arcade_mode_enabled(),
std::env::var("ARC_URL").ok(),
)
}
fn base(&self) -> &str {
match self {
BroadcastPlane::ArcadeV2 { base } | BroadcastPlane::ClassicArc { base } => base,
}
}
fn name(&self) -> &'static str {
match self {
BroadcastPlane::ArcadeV2 { .. } => "broadcaster(arcade-v2)",
BroadcastPlane::ClassicArc { .. } => "broadcaster(arc)",
}
}
fn status_template(&self) -> String {
match self {
BroadcastPlane::ArcadeV2 { base } => format!("{base}/tx/{{txid}}"),
BroadcastPlane::ClassicArc { base } => format!("{base}/v1/tx/{{txid}}"),
}
}
}
#[derive(Clone, Debug)]
struct StatusSource {
name: &'static str,
url_template: String,
auth: Option<String>,
absence: AbsenceAuthority,
}
fn build_sources(
chain: Chain,
plane: &BroadcastPlane,
taal_key: Option<String>,
) -> Vec<StatusSource> {
let mut sources = vec![StatusSource {
name: plane.name(),
url_template: plane.status_template(),
auth: match plane {
BroadcastPlane::ArcadeV2 { .. } => None,
BroadcastPlane::ClassicArc { .. } => taal_key.clone(),
},
absence: AbsenceAuthority::Broadcaster,
}];
sources.push(StatusSource {
name: "whatsonchain",
url_template: format!("{}/tx/hash/{{txid}}", woc_base(chain)),
auth: None,
absence: AbsenceAuthority::ChainIndex,
});
if let Some(gp) = gorillapool_arc_url(chain) {
if normalize_base(gp) != plane.base() {
sources.push(StatusSource {
name: "arc-gorillapool",
url_template: format!("{gp}/v1/tx/{{txid}}"),
auth: None,
absence: AbsenceAuthority::None,
});
}
}
if let Some(key) = taal_key {
let taal = taal_arc_url(chain);
if normalize_base(taal) != plane.base() {
sources.push(StatusSource {
name: "arc-taal",
url_template: format!("{taal}/v1/tx/{{txid}}"),
auth: Some(key),
absence: AbsenceAuthority::None,
});
}
}
sources
}
#[derive(Clone)]
pub struct BroadcastVerifier {
client: Client,
sources: Vec<StatusSource>,
attempts: u32,
delay: Duration,
enabled: bool,
}
impl BroadcastVerifier {
pub fn from_env(chain: Chain) -> Self {
let enabled = !env_truthy("BSV_WALLET_SKIP_BROADCAST_VERIFY");
let attempts = std::env::var("BSV_WALLET_BROADCAST_VERIFY_ATTEMPTS")
.ok()
.and_then(|v| v.parse::<u32>().ok())
.filter(|n| *n > 0)
.unwrap_or(DEFAULT_ATTEMPTS);
let delay_ms = std::env::var("BSV_WALLET_BROADCAST_VERIFY_DELAY_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.unwrap_or(DEFAULT_DELAY_MS);
let taal_key = std::env::var("TAAL_API_KEY")
.ok()
.filter(|k| !k.is_empty())
.or_else(|| {
std::env::var("MAIN_TAAL_API_KEY")
.ok()
.filter(|k| !k.is_empty())
});
let plane = BroadcastPlane::from_env(chain);
tracing::debug!(plane = ?plane, "broadcast verifier plane");
Self {
client: Client::new(),
sources: build_sources(chain, &plane, taal_key),
attempts,
delay: Duration::from_millis(delay_ms),
enabled,
}
}
fn absence_window(&self) -> Duration {
self.delay * self.attempts.saturating_sub(1) + PROBE_TIMEOUT
}
pub async fn verify(&self, txid: &str) -> BroadcastVerification {
if !self.enabled || self.sources.is_empty() {
return BroadcastVerification::Inconclusive;
}
let deadline = Instant::now() + self.absence_window();
let mut last_votes: Option<AbsenceVotes> = None;
for attempt in 0..self.attempts {
let mut votes = AbsenceVotes::default();
for src in &self.sources {
match probe(&self.client, src, txid).await {
Presence::Present => return BroadcastVerification::Confirmed,
Presence::Absent => votes.record(src.absence),
Presence::Unknown => {}
}
}
last_votes = Some(votes);
if attempt + 1 < self.attempts {
if Instant::now() >= deadline {
break;
}
tokio::time::sleep(self.delay).await;
}
}
match last_votes {
Some(v) if v.is_definitive() => BroadcastVerification::Rejected,
_ => BroadcastVerification::Inconclusive,
}
}
}
async fn probe(client: &Client, src: &StatusSource, txid: &str) -> Presence {
let url = src.url_template.replace("{txid}", txid);
let mut req = client.get(&url).timeout(PROBE_TIMEOUT);
if let Some(auth) = &src.auth {
req = req.header("Authorization", auth);
}
match req.send().await {
Ok(resp) => {
let status = resp.status().as_u16();
match status {
200 => Presence::Present,
404 => {
if src.absence == AbsenceAuthority::Broadcaster && !is_json(&resp) {
tracing::debug!(
source = src.name,
url = %url,
"broadcaster 404 is not a JSON tx-status body — treating as \
route-not-found (check ARC_URL / path shape), not absence"
);
return Presence::Unknown;
}
Presence::Absent
}
other => {
tracing::debug!(
source = src.name,
status = other,
"broadcast probe inconclusive"
);
Presence::Unknown
}
}
}
Err(e) => {
tracing::debug!(source = src.name, error = %e, "broadcast probe request failed");
Presence::Unknown
}
}
}
fn is_json(resp: &reqwest::Response) -> bool {
resp.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.map(|ct| ct.to_ascii_lowercase().contains("json"))
.unwrap_or(false)
}
fn normalize_base(url: &str) -> String {
url.trim().trim_end_matches('/').to_string()
}
fn taal_arc_url(chain: Chain) -> &'static str {
match chain {
Chain::Main => "https://arc.taal.com",
Chain::Test => "https://arc-test.taal.com",
}
}
fn gorillapool_arc_url(chain: Chain) -> Option<&'static str> {
match chain {
Chain::Main => Some("https://arc.gorillapool.io"),
Chain::Test => None,
}
}
fn woc_base(chain: Chain) -> &'static str {
match chain {
Chain::Main => "https://api.whatsonchain.com/v1/bsv/main",
Chain::Test => "https://api.whatsonchain.com/v1/bsv/test",
}
}
fn env_truthy(key: &str) -> bool {
std::env::var(key)
.map(|v| {
let v = v.trim().to_ascii_lowercase();
v == "1" || v == "true" || v == "yes" || v == "on"
})
.unwrap_or(false)
}
#[cfg(test)]
mod tests {
use super::*;
use axum::http::StatusCode;
use axum::routing::get;
use axum::Router;
use std::net::SocketAddr;
const TXID: &str = "0000000000000000000000000000000000000000000000000000000000000001";
const SYNTHETIC_ARCADE: &str = "https://arcade.invalid";
const SYNTHETIC_ARC: &str = "https://arc.invalid";
const SYNTHETIC_KEY: &str = "test-key-not-a-real-credential";
#[test]
fn arcade_plane_uses_bare_tx_path_not_v1() {
let plane = BroadcastPlane::resolve(
Chain::Main,
true,
Some(SYNTHETIC_ARCADE.to_string()),
);
assert_eq!(
plane.status_template(),
format!("{SYNTHETIC_ARCADE}/tx/{{txid}}")
);
assert!(
!plane.status_template().contains("/v1/"),
"Arcade V2 must NOT be probed on the classic ARC /v1 path"
);
}
#[test]
fn classic_arc_plane_uses_v1_tx_path() {
let plane = BroadcastPlane::resolve(
Chain::Main,
false,
Some(SYNTHETIC_ARC.to_string()),
);
assert_eq!(
plane.status_template(),
format!("{SYNTHETIC_ARC}/v1/tx/{{txid}}")
);
}
#[test]
fn arcade_mode_defaults_to_the_arcade_endpoint_when_arc_url_is_unset() {
let plane = BroadcastPlane::resolve(Chain::Main, true, None);
assert_eq!(plane.base(), ARCADE_V2_MAINNET.trim_end_matches('/'));
}
#[test]
fn classic_mode_defaults_to_taal_and_respects_chain() {
assert_eq!(
BroadcastPlane::resolve(Chain::Main, false, None).base(),
"https://arc.taal.com"
);
assert_eq!(
BroadcastPlane::resolve(Chain::Test, false, None).base(),
"https://arc-test.taal.com"
);
}
#[test]
fn empty_arc_url_falls_back_to_the_default_rather_than_an_empty_base() {
let plane = BroadcastPlane::resolve(Chain::Main, true, Some(" ".to_string()));
assert_eq!(plane.base(), ARCADE_V2_MAINNET.trim_end_matches('/'));
}
#[test]
fn trailing_slash_in_arc_url_does_not_produce_a_double_slash() {
let plane = BroadcastPlane::resolve(
Chain::Main,
true,
Some(format!("{SYNTHETIC_ARCADE}/").to_string()),
);
assert_eq!(
plane.status_template(),
format!("{SYNTHETIC_ARCADE}/tx/{{txid}}")
);
}
#[test]
fn the_broadcaster_we_used_is_always_the_first_source_consulted() {
for plane in [
BroadcastPlane::resolve(Chain::Main, true, Some(SYNTHETIC_ARCADE.to_string())),
BroadcastPlane::resolve(Chain::Main, false, Some(SYNTHETIC_ARC.to_string())),
] {
let sources = build_sources(Chain::Main, &plane, None);
assert_eq!(sources[0].absence, AbsenceAuthority::Broadcaster);
assert!(
sources[0].url_template.starts_with(plane.base()),
"source 0 ({}) must be the configured broadcaster {}",
sources[0].url_template,
plane.base()
);
}
}
#[test]
fn arcade_broadcaster_probe_is_keyless_even_when_a_taal_key_exists() {
let plane = BroadcastPlane::resolve(Chain::Main, true, Some(SYNTHETIC_ARCADE.to_string()));
let sources = build_sources(Chain::Main, &plane, Some(SYNTHETIC_KEY.to_string()));
assert!(sources[0].auth.is_none());
}
#[test]
fn classic_broadcaster_probe_carries_the_taal_key_when_present() {
let plane = BroadcastPlane::resolve(Chain::Main, false, None);
let sources = build_sources(Chain::Main, &plane, Some(SYNTHETIC_KEY.to_string()));
assert_eq!(sources[0].auth.as_deref(), Some(SYNTHETIC_KEY));
}
#[test]
fn keyless_taal_is_not_probed_at_all() {
let plane = BroadcastPlane::resolve(Chain::Main, true, Some(SYNTHETIC_ARCADE.to_string()));
let sources = build_sources(Chain::Main, &plane, None);
assert!(!sources.iter().any(|s| s.name == "arc-taal"));
}
#[test]
fn a_store_is_never_listed_twice_when_it_is_also_the_broadcaster() {
let plane = BroadcastPlane::resolve(
Chain::Main,
false,
Some("https://arc.gorillapool.io".to_string()),
);
let sources = build_sources(Chain::Main, &plane, None);
let gp_rows: Vec<_> = sources
.iter()
.filter(|s| s.url_template.contains("arc.gorillapool.io"))
.collect();
assert_eq!(gp_rows.len(), 1);
assert_eq!(gp_rows[0].absence, AbsenceAuthority::Broadcaster);
}
#[test]
fn a_third_party_arc_store_is_never_authoritative_for_absence() {
let plane = BroadcastPlane::resolve(Chain::Main, true, Some(SYNTHETIC_ARCADE.to_string()));
let sources = build_sources(Chain::Main, &plane, Some(SYNTHETIC_KEY.to_string()));
for s in sources.iter().filter(|s| s.name.starts_with("arc-")) {
assert_eq!(
s.absence,
AbsenceAuthority::None,
"{} is not the broadcaster; its absence must carry no weight",
s.name
);
}
}
#[test]
fn whatsonchain_is_the_chain_index_authority() {
let plane = BroadcastPlane::resolve(Chain::Main, true, Some(SYNTHETIC_ARCADE.to_string()));
let sources = build_sources(Chain::Main, &plane, None);
let woc = sources.iter().find(|s| s.name == "whatsonchain").unwrap();
assert_eq!(woc.absence, AbsenceAuthority::ChainIndex);
}
#[test]
fn absence_is_definitive_only_when_broadcaster_and_chain_index_agree() {
let mut none = AbsenceVotes::default();
assert!(!none.is_definitive(), "no votes is not evidence");
none.record(AbsenceAuthority::None);
assert!(!none.is_definitive());
let mut broadcaster_only = AbsenceVotes::default();
broadcaster_only.record(AbsenceAuthority::Broadcaster);
assert!(
!broadcaster_only.is_definitive(),
"the primary may 404 while the tx went out through the failover provider"
);
let mut index_only = AbsenceVotes::default();
index_only.record(AbsenceAuthority::ChainIndex);
assert!(
!index_only.is_definitive(),
"a chain index can simply be lagging its mempool ingestion"
);
let mut both = AbsenceVotes::default();
both.record(AbsenceAuthority::Broadcaster);
both.record(AbsenceAuthority::ChainIndex);
assert!(both.is_definitive());
}
async fn mock_status_server(code: StatusCode) -> String {
mock_status_server_ct(code, Some("application/json")).await
}
async fn mock_status_server_ct(code: StatusCode, content_type: Option<&'static str>) -> String {
let handler = move || async move {
let mut resp = axum::response::Response::new(axum::body::Body::from("{}"));
*resp.status_mut() = code;
if let Some(ct) = content_type {
resp.headers_mut()
.insert(reqwest::header::CONTENT_TYPE.as_str(), ct.parse().unwrap());
} else {
resp.headers_mut()
.remove(reqwest::header::CONTENT_TYPE.as_str());
}
resp
};
let app = Router::new()
.route("/tx/{txid}", get(handler))
.route("/v1/tx/{txid}", get(handler))
.route("/tx/hash/{txid}", get(handler));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr: SocketAddr = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.ok();
});
format!("http://{}", addr)
}
fn source(name: &'static str, base: &str, absence: AbsenceAuthority) -> StatusSource {
StatusSource {
name,
url_template: format!("{base}/tx/{{txid}}"),
auth: None,
absence,
}
}
fn verifier_with(sources: Vec<StatusSource>) -> BroadcastVerifier {
BroadcastVerifier {
client: Client::new(),
sources,
attempts: 2,
delay: Duration::from_millis(0),
enabled: true,
}
}
#[tokio::test]
async fn rejected_when_broadcaster_and_chain_index_both_report_absent() {
let base = mock_status_server(StatusCode::NOT_FOUND).await;
let verifier = verifier_with(vec![
source("broadcaster", &base, AbsenceAuthority::Broadcaster),
source("chain-index", &base, AbsenceAuthority::ChainIndex),
]);
let outcome = verifier.verify(TXID).await;
assert_eq!(outcome, BroadcastVerification::Rejected);
assert!(
outcome.into_send_result(TXID).is_err(),
"a Rejected verification must map to Err so the send fails loudly"
);
}
#[tokio::test]
async fn the_false_negative_that_motivated_this_fix_is_now_inconclusive() {
let absent = mock_status_server(StatusCode::NOT_FOUND).await;
let verifier = verifier_with(vec![
source(
"broadcaster",
"http://127.0.0.1:1",
AbsenceAuthority::Broadcaster,
),
source("chain-index", &absent, AbsenceAuthority::ChainIndex),
source("arc-third-party", &absent, AbsenceAuthority::None),
]);
assert_eq!(
verifier.verify(TXID).await,
BroadcastVerification::Inconclusive
);
}
#[tokio::test]
async fn third_party_absence_alone_never_rejects() {
let base = mock_status_server(StatusCode::NOT_FOUND).await;
let verifier = verifier_with(vec![
source("arc-third-party-a", &base, AbsenceAuthority::None),
source("arc-third-party-b", &base, AbsenceAuthority::None),
]);
assert_eq!(
verifier.verify(TXID).await,
BroadcastVerification::Inconclusive
);
}
#[tokio::test]
async fn broadcaster_absence_alone_never_rejects() {
let absent = mock_status_server(StatusCode::NOT_FOUND).await;
let verifier = verifier_with(vec![
source("broadcaster", &absent, AbsenceAuthority::Broadcaster),
source(
"chain-index",
"http://127.0.0.1:1",
AbsenceAuthority::ChainIndex,
),
]);
assert_eq!(
verifier.verify(TXID).await,
BroadcastVerification::Inconclusive
);
}
#[tokio::test]
async fn chain_index_absence_alone_never_rejects() {
let absent = mock_status_server(StatusCode::NOT_FOUND).await;
let verifier = verifier_with(vec![
source("broadcaster", &absent, AbsenceAuthority::Broadcaster),
source("chain-index", &absent, AbsenceAuthority::ChainIndex),
]);
assert_eq!(verifier.verify(TXID).await, BroadcastVerification::Rejected);
let unauth = mock_status_server(StatusCode::UNAUTHORIZED).await;
let verifier = verifier_with(vec![
source("broadcaster", &unauth, AbsenceAuthority::Broadcaster),
source("chain-index", &absent, AbsenceAuthority::ChainIndex),
]);
assert_eq!(
verifier.verify(TXID).await,
BroadcastVerification::Inconclusive
);
}
#[tokio::test]
async fn presence_from_any_source_confirms_even_when_others_say_absent() {
let present = mock_status_server(StatusCode::OK).await;
let absent = mock_status_server(StatusCode::NOT_FOUND).await;
let verifier = verifier_with(vec![
source("broadcaster", &absent, AbsenceAuthority::Broadcaster),
source("chain-index", &absent, AbsenceAuthority::ChainIndex),
source("arc-third-party", &present, AbsenceAuthority::None),
]);
let outcome = verifier.verify(TXID).await;
assert_eq!(outcome, BroadcastVerification::Confirmed);
assert!(outcome.into_send_result(TXID).is_ok());
}
#[tokio::test]
async fn confirmed_broadcast_succeeds() {
let base = mock_status_server(StatusCode::OK).await;
let verifier = verifier_with(vec![source(
"broadcaster",
&base,
AbsenceAuthority::Broadcaster,
)]);
let outcome = verifier.verify(TXID).await;
assert_eq!(outcome, BroadcastVerification::Confirmed);
assert!(outcome.into_send_result(TXID).is_ok());
}
#[tokio::test]
async fn unreachable_source_is_inconclusive_not_a_failure() {
let base = mock_status_server(StatusCode::SERVICE_UNAVAILABLE).await;
let verifier = verifier_with(vec![
source("broadcaster", &base, AbsenceAuthority::Broadcaster),
source("chain-index", &base, AbsenceAuthority::ChainIndex),
]);
let outcome = verifier.verify(TXID).await;
assert_eq!(outcome, BroadcastVerification::Inconclusive);
assert!(outcome.into_send_result(TXID).is_ok());
}
#[tokio::test]
async fn a_routing_404_from_the_broadcaster_is_not_absence() {
let text_404 = mock_status_server_ct(StatusCode::NOT_FOUND, Some("text/plain")).await;
let json_404 = mock_status_server(StatusCode::NOT_FOUND).await;
let verifier = verifier_with(vec![
source("broadcaster", &text_404, AbsenceAuthority::Broadcaster),
source("chain-index", &json_404, AbsenceAuthority::ChainIndex),
]);
assert_eq!(
verifier.verify(TXID).await,
BroadcastVerification::Inconclusive
);
}
#[tokio::test]
async fn disabled_verifier_is_inconclusive() {
let base = mock_status_server(StatusCode::NOT_FOUND).await;
let mut verifier = verifier_with(vec![
source("broadcaster", &base, AbsenceAuthority::Broadcaster),
source("chain-index", &base, AbsenceAuthority::ChainIndex),
]);
verifier.enabled = false;
assert_eq!(
verifier.verify(TXID).await,
BroadcastVerification::Inconclusive
);
}
#[test]
fn absence_window_is_bounded_and_reflects_the_configured_rounds() {
let v = BroadcastVerifier {
client: Client::new(),
sources: vec![],
attempts: DEFAULT_ATTEMPTS,
delay: Duration::from_millis(DEFAULT_DELAY_MS),
enabled: true,
};
assert_eq!(
v.absence_window(),
Duration::from_millis(27_500) + PROBE_TIMEOUT
);
}
}