use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::time::Duration;
use solid_pod_rs::bitcoin_tx::{
anchor_state, gitmark_advance, gitmark_genesis, GitmarkUpdate, MempoolBroadcast, TxoVoucher,
DEFAULT_FEE_SATS,
};
use solid_pod_rs::blocktrail::{verify_anchor_chain, TrailReport};
use solid_pod_rs::mrc20::{bt_address, MempoolLookup, TxIn, TxInfo, TxOut, Utxo};
use solid_pod_rs::payments::PaymentError;
use solid_pod_rs::provenance::{BlockAnchorer, BlockTrailAnchor, ProvenanceError};
pub const MEMPOOL_URL_ENV: &str = "JSS_PAY_MEMPOOL_URL";
pub const DEFAULT_MEMPOOL_URL: &str = "https://mempool.space/testnet4";
pub const MEMPOOL_CONNECT_TIMEOUT_ENV: &str = "JSS_PAY_MEMPOOL_CONNECT_TIMEOUT_SECS";
pub const MEMPOOL_TIMEOUT_ENV: &str = "JSS_PAY_MEMPOOL_TIMEOUT_SECS";
const DEFAULT_CONNECT_TIMEOUT_SECS: u64 = 5;
const DEFAULT_TIMEOUT_SECS: u64 = 20;
fn env_secs(var: &str, default: u64) -> Duration {
let secs = std::env::var(var)
.ok()
.and_then(|v| v.trim().parse::<u64>().ok())
.filter(|s| *s > 0)
.unwrap_or(default);
Duration::from_secs(secs)
}
fn http_client(connect: Duration, total: Duration) -> reqwest::Client {
reqwest::Client::builder()
.connect_timeout(connect)
.timeout(total)
.build()
.unwrap_or_default()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MempoolConfigSource {
Explicit,
Default,
}
impl MempoolConfigSource {
#[must_use]
pub fn as_str(&self) -> &'static str {
match self {
Self::Explicit => "explicit",
Self::Default => "default",
}
}
}
impl std::fmt::Display for MempoolConfigSource {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BitcoinNetwork {
Mainnet,
Testnet3,
Testnet4,
Signet,
Regtest,
Unknown,
}
impl BitcoinNetwork {
#[must_use]
pub fn as_str(&self) -> &'static str {
match self {
Self::Mainnet => "mainnet",
Self::Testnet3 => "testnet3",
Self::Testnet4 => "testnet4",
Self::Signet => "signet",
Self::Regtest => "regtest",
Self::Unknown => "unknown",
}
}
}
impl std::fmt::Display for BitcoinNetwork {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
#[must_use]
pub fn infer_network(base_url: &str) -> BitcoinNetwork {
let trimmed = base_url.trim().trim_end_matches('/');
let after_scheme = match trimmed.find("://") {
Some(i) => &trimmed[i + 3..],
None => trimmed,
};
let (authority, path) = match after_scheme.find('/') {
Some(i) => (&after_scheme[..i], &after_scheme[i + 1..]),
None => (after_scheme, ""),
};
let path = path
.split(['?', '#'])
.next()
.unwrap_or("")
.trim_end_matches('/');
let last_segment = path.rsplit('/').find(|seg| !seg.is_empty()).unwrap_or("");
match last_segment.to_ascii_lowercase().as_str() {
"testnet4" => return BitcoinNetwork::Testnet4,
"testnet" | "testnet3" => return BitcoinNetwork::Testnet3,
"signet" => return BitcoinNetwork::Signet,
"regtest" => return BitcoinNetwork::Regtest,
"mainnet" | "bitcoin" => return BitcoinNetwork::Mainnet,
_ => {}
}
let host = authority.rsplit('@').next().unwrap_or(authority);
let host = if let Some(rest) = host.strip_prefix('[') {
rest.split(']').next().unwrap_or(rest)
} else {
host.split(':').next().unwrap_or(host)
};
let host_lower = host.to_ascii_lowercase();
let is_loopback = host_lower == "localhost"
|| host_lower == "::1"
|| host_lower.starts_with("127.")
|| host_lower.ends_with(".localhost");
if is_loopback {
return BitcoinNetwork::Regtest;
}
if path.is_empty() && (host_lower == "mempool.space" || host_lower.ends_with(".mempool.space"))
{
return BitcoinNetwork::Mainnet;
}
BitcoinNetwork::Unknown
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MempoolSelection {
pub base_url: String,
pub network: BitcoinNetwork,
pub source: MempoolConfigSource,
}
impl MempoolSelection {
#[must_use]
pub fn to_manifest_json(&self) -> serde_json::Value {
serde_json::json!({
"base_url": self.base_url,
"network": self.network.as_str(),
"source": self.source.as_str(),
})
}
}
#[must_use]
pub fn select_mempool_endpoint(configured: Option<&str>) -> MempoolSelection {
let (raw, source) = match configured.map(str::trim).filter(|v| !v.is_empty()) {
Some(v) => (v, MempoolConfigSource::Explicit),
None => (DEFAULT_MEMPOOL_URL, MempoolConfigSource::Default),
};
let base_url = raw.trim_end_matches('/').to_string();
let network = infer_network(&base_url);
MempoolSelection {
base_url,
network,
source,
}
}
#[must_use]
pub fn select_mempool_endpoint_from_env() -> MempoolSelection {
let configured = std::env::var(MEMPOOL_URL_ENV).ok();
select_mempool_endpoint(configured.as_deref())
}
pub fn log_mempool_selection(sel: &MempoolSelection) {
tracing::info!(
target: "solid_pod_rs_server::mempool",
base_url = %sel.base_url,
network = sel.network.as_str(),
source = sel.source.as_str(),
"mempool endpoint selected"
);
if sel.network == BitcoinNetwork::Unknown || sel.source == MempoolConfigSource::Default {
tracing::warn!(
target: "solid_pod_rs_server::mempool",
base_url = %sel.base_url,
network = sel.network.as_str(),
source = sel.source.as_str(),
env_var = MEMPOOL_URL_ENV,
"pod is using an unverified or defaulted Bitcoin endpoint; anchors may be \
written to or verified against an unintended chain — set {} to pin the \
explorer, and therefore the network, explicitly",
MEMPOOL_URL_ENV
);
}
}
pub fn log_mempool_selection_once(sel: &MempoolSelection) {
static ONCE: std::sync::Once = std::sync::Once::new();
ONCE.call_once(|| log_mempool_selection(sel));
}
#[derive(Debug, Clone)]
pub struct MempoolHttpClient {
client: reqwest::Client,
base: String,
selection: MempoolSelection,
}
impl MempoolHttpClient {
#[must_use]
pub fn new(base_url: impl Into<String>) -> Self {
let raw = base_url.into();
let base = raw.trim_end_matches('/').to_string();
let selection = MempoolSelection {
network: infer_network(&base),
base_url: base,
source: MempoolConfigSource::Explicit,
};
Self::from_selection(selection)
}
#[must_use]
pub fn from_selection(selection: MempoolSelection) -> Self {
Self {
client: http_client(
env_secs(MEMPOOL_CONNECT_TIMEOUT_ENV, DEFAULT_CONNECT_TIMEOUT_SECS),
env_secs(MEMPOOL_TIMEOUT_ENV, DEFAULT_TIMEOUT_SECS),
),
base: selection.base_url.clone(),
selection,
}
}
#[must_use]
pub fn with_timeouts(mut self, connect: Duration, total: Duration) -> Self {
self.client = http_client(connect, total);
self
}
#[must_use]
pub fn from_env() -> Self {
let selection = select_mempool_endpoint_from_env();
log_mempool_selection_once(&selection);
Self::from_selection(selection)
}
#[must_use]
pub fn base_url(&self) -> &str {
&self.base
}
#[must_use]
pub fn selection(&self) -> &MempoolSelection {
&self.selection
}
pub async fn transaction_exists(&self, txid: &str) -> Result<bool, PaymentError> {
let url = format!("{}/api/tx/{txid}", self.base);
let resp = self
.client
.get(&url)
.send()
.await
.map_err(|e| PaymentError::InvalidState(format!("mempool request failed: {e}")))?;
if resp.status() == reqwest::StatusCode::NOT_FOUND {
return Ok(false);
}
if !resp.status().is_success() {
return Err(PaymentError::InvalidState(format!(
"mempool API error: {} for {url}",
resp.status().as_u16()
)));
}
Ok(true)
}
async fn get_text(&self, url: &str) -> Result<String, PaymentError> {
let resp = self
.client
.get(url)
.send()
.await
.map_err(|e| PaymentError::InvalidState(format!("mempool request failed: {e}")))?;
let status = resp.status();
if !status.is_success() {
return Err(PaymentError::InvalidState(format!(
"mempool API error: {} for {url}",
status.as_u16()
)));
}
resp.text()
.await
.map_err(|e| PaymentError::InvalidState(format!("mempool body read failed: {e}")))
}
async fn post_text(&self, url: &str, body: &str) -> Result<String, PaymentError> {
let resp = self
.client
.post(url)
.header("Content-Type", "text/plain")
.body(body.to_string())
.send()
.await
.map_err(|e| PaymentError::InvalidState(format!("mempool broadcast failed: {e}")))?;
let status = resp.status();
let text = resp
.text()
.await
.map_err(|e| PaymentError::InvalidState(format!("mempool body read failed: {e}")))?;
if !status.is_success() {
return Err(PaymentError::InvalidState(format!(
"broadcast rejected ({}): {text}",
status.as_u16()
)));
}
Ok(text.trim().to_string())
}
}
#[derive(Debug, Deserialize, Default)]
struct StatusWire {
#[serde(default)]
confirmed: bool,
#[serde(default)]
block_height: Option<u64>,
}
#[derive(Debug, Deserialize)]
struct UtxoWire {
txid: String,
vout: u32,
#[serde(default)]
value: u64,
#[serde(default)]
status: StatusWire,
}
impl From<UtxoWire> for Utxo {
fn from(w: UtxoWire) -> Self {
Utxo {
txid: w.txid,
vout: w.vout,
value: w.value,
confirmed: w.status.confirmed,
block_height: w.status.block_height,
}
}
}
#[derive(Debug, Deserialize, Default)]
struct TxOutWire {
#[serde(default)]
value: u64,
#[serde(default)]
scriptpubkey: Option<String>,
#[serde(default)]
scriptpubkey_address: Option<String>,
}
impl From<TxOutWire> for TxOut {
fn from(w: TxOutWire) -> Self {
TxOut {
value: w.value,
scriptpubkey: w.scriptpubkey,
scriptpubkey_address: w.scriptpubkey_address,
}
}
}
#[derive(Debug, Deserialize)]
struct TxInWire {
#[serde(default)]
txid: String,
#[serde(default)]
vout: u32,
}
#[derive(Debug, Deserialize)]
struct TxWire {
txid: String,
#[serde(default)]
vin: Vec<TxInWire>,
#[serde(default)]
vout: Vec<TxOutWire>,
#[serde(default)]
status: StatusWire,
}
impl From<TxWire> for TxInfo {
fn from(w: TxWire) -> Self {
TxInfo {
txid: w.txid,
vin: w
.vin
.into_iter()
.map(|i| TxIn {
txid: i.txid,
vout: i.vout,
})
.collect(),
vout: w.vout.into_iter().map(TxOut::from).collect(),
confirmed: w.status.confirmed,
block_height: w.status.block_height,
}
}
}
#[async_trait(?Send)]
impl MempoolLookup for MempoolHttpClient {
async fn address_utxos(&self, address: &str) -> Result<Vec<Utxo>, PaymentError> {
let url = format!("{}/api/address/{address}/utxo", self.base);
let body = self.get_text(&url).await?;
let wire: Vec<UtxoWire> = serde_json::from_str(&body)
.map_err(|e| PaymentError::InvalidState(format!("malformed utxo JSON: {e}")))?;
Ok(wire.into_iter().map(Utxo::from).collect())
}
async fn tx(&self, txid: &str) -> Result<TxInfo, PaymentError> {
let url = format!("{}/api/tx/{txid}", self.base);
let body = self.get_text(&url).await?;
let wire: TxWire = serde_json::from_str(&body)
.map_err(|e| PaymentError::InvalidState(format!("malformed tx JSON: {e}")))?;
Ok(TxInfo::from(wire))
}
}
#[async_trait(?Send)]
impl MempoolBroadcast for MempoolHttpClient {
async fn broadcast_tx(&self, raw_hex: &str) -> Result<String, PaymentError> {
let url = format!("{}/api/tx", self.base);
self.post_text(&url, raw_hex).await
}
}
#[derive(Clone)]
pub struct MempoolBlockAnchorer<M: MempoolLookup + MempoolBroadcast + Send + Sync> {
lookup: M,
storage: Option<std::sync::Arc<dyn solid_pod_rs::storage::Storage>>,
}
impl<M: MempoolLookup + MempoolBroadcast + Send + Sync> MempoolBlockAnchorer<M> {
pub fn new(lookup: M) -> Self {
Self {
lookup,
storage: None,
}
}
pub fn with_storage(
lookup: M,
storage: std::sync::Arc<dyn solid_pod_rs::storage::Storage>,
) -> Self {
Self {
lookup,
storage: Some(storage),
}
}
pub fn lookup(&self) -> &M {
&self.lookup
}
}
const PENDING_ANCHOR_PREFIX: &str = "anchor pending: ";
fn pending_anchor(detail: String) -> ProvenanceError {
ProvenanceError::Anchor(format!("{PENDING_ANCHOR_PREFIX}{detail}"))
}
#[cfg(any(feature = "git", test))]
pub(crate) fn is_pending_anchor(e: &ProvenanceError) -> bool {
matches!(e, ProvenanceError::Anchor(m) if m.starts_with(PENDING_ANCHOR_PREFIX))
}
#[async_trait(?Send)]
impl<M: MempoolLookup + MempoolBroadcast + Send + Sync> BlockAnchorer for MempoolBlockAnchorer<M> {
async fn anchor(
&self,
ticker: &str,
state_hash: &str,
network: &str,
) -> Result<BlockTrailAnchor, ProvenanceError> {
use crate::trail_store::{load_trail, save_trail};
let storage = self.storage.as_ref().ok_or_else(|| {
ProvenanceError::Anchor(
"anchor() requires storage; construct with MempoolBlockAnchorer::with_storage"
.into(),
)
})?;
let _guard = crate::trail_store::lock_path(&crate::trail_store::trail_path(ticker)).await;
settle_trail_intent(&self.lookup, storage, ticker).await?;
let mut stored = load_trail(storage, ticker)
.await
.map_err(|e| ProvenanceError::Anchor(format!("load trail {ticker}: {e}")))?
.ok_or_else(|| {
ProvenanceError::Anchor(format!("trail {ticker} not minted on this pod"))
})?;
if stored.network != network {
return Err(ProvenanceError::Anchor(format!(
"network mismatch: trail is {}, requested {network}",
stored.network
)));
}
let public = stored.to_public();
let update = anchor_state(
&public,
&stored.privkey,
state_hash,
DEFAULT_FEE_SATS,
&self.lookup,
)
.await
.map_err(|e| ProvenanceError::Anchor(format!("build anchoring tx: {e}")))?;
let expected = solid_pod_rs::bitcoin_tx::decode_tx_info(&update.tx.raw_hex)
.map_err(|e| ProvenanceError::Anchor(format!("decode anchoring tx: {e}")))?
.txid;
let mut appended = update.trail.clone();
appended.current_txid = expected.clone();
stored.merge_public(&appended);
stored.current_txid = expected.clone();
stored.current_vout = 0;
let intent_path = anchor_intent_path(ticker);
let body = serde_json::to_vec(&PendingAnchor {
txid: expected.clone(),
raw_hex: update.tx.raw_hex.clone(),
stored: stored.clone(),
})
.map_err(|e| ProvenanceError::Anchor(format!("serialise pending anchor: {e}")))?;
storage
.put(&intent_path, bytes::Bytes::from(body), "application/json")
.await
.map_err(|e| ProvenanceError::Anchor(format!("record pending anchor: {e}")))?;
let txid = self
.lookup
.broadcast_tx(&update.tx.raw_hex)
.await
.map_err(|e| pending_anchor(format!("broadcast anchoring tx: {e}")))?;
if txid != expected {
return Err(pending_anchor(format!(
"mempool returned txid {txid} for an anchor that is {expected}"
)));
}
match save_trail(storage, &stored).await {
Ok(()) => {
if let Err(e) = storage.delete(&intent_path).await {
tracing::warn!(%ticker, "anchor intent not cleared: {e}");
}
}
Err(e) => tracing::error!(
%ticker, %txid,
"anchor broadcast accepted but trail save failed; the next advance settles it: {e}"
),
}
Ok(BlockTrailAnchor {
ticker: ticker.to_string(),
state_hash: state_hash.to_string(),
txid,
vout: 0,
address: update.address,
network: network.to_string(),
blockheight: None,
state_strings: appended.state_strings,
pubkey: Some(stored.pubkey_base),
})
}
async fn verify(&self, anchor: &BlockTrailAnchor) -> Result<bool, ProvenanceError> {
Ok(verify_anchor_report(&self.lookup, anchor)
.await?
.is_some_and(|r| r.is_intact()))
}
}
impl<M: MempoolLookup + MempoolBroadcast + Send + Sync> MempoolBlockAnchorer<M> {
pub async fn verify_report(
&self,
anchor: &BlockTrailAnchor,
) -> Result<Option<TrailReport>, ProvenanceError> {
verify_anchor_report(&self.lookup, anchor).await
}
}
pub async fn verify_anchor_report(
lookup: &dyn MempoolLookup,
anchor: &BlockTrailAnchor,
) -> Result<Option<TrailReport>, ProvenanceError> {
let Some(pubkey) = anchor.pubkey.as_deref() else {
return Ok(None);
};
if anchor.state_strings.is_empty() {
return Ok(None);
}
let derived = bt_address(pubkey, &anchor.state_strings, &anchor.network)
.map_err(|e| ProvenanceError::Anchor(format!("address re-derivation failed: {e}")))?;
if derived != anchor.address {
return Ok(None);
}
verify_anchor_chain(
pubkey,
&anchor.state_strings,
&anchor.txid,
anchor.vout,
lookup,
)
.await
.map(Some)
.map_err(|e| ProvenanceError::Anchor(format!("anchor chain walk failed: {e}")))
}
fn already_known(e: &PaymentError) -> bool {
let msg = e.to_string().to_lowercase().replace('-', " ");
msg.contains("already in mempool")
|| msg.contains("already in block chain")
|| msg.contains("already known")
|| msg.contains("already in utxo set")
|| msg.contains("outputs already in")
}
async fn settle_broadcast<M: MempoolLookup + MempoolBroadcast + ?Sized>(
lookup: &M,
txid: &str,
raw_hex: &str,
head: Option<(&str, u32)>,
) -> Result<bool, ProvenanceError> {
let unresolved = |why: &dyn std::fmt::Display| {
ProvenanceError::Anchor(format!(
"a previous mark ({txid}) is unresolved, refusing to advance: {why}"
))
};
let refusal = match lookup.broadcast_tx(raw_hex).await {
Ok(t) if t == txid => return Ok(true),
Ok(t) => {
return Err(ProvenanceError::Anchor(format!(
"mempool returned txid {t} for a mark that is {txid}"
)))
}
Err(e) if already_known(&e) => return Ok(true),
Err(e) if e.to_string().contains("broadcast rejected (4") => e,
Err(e) => return Err(unresolved(&e)),
};
let spent = solid_pod_rs::bitcoin_tx::decode_tx_info(raw_hex)
.map_err(|e| unresolved(&e))?
.vin
.into_iter()
.next()
.ok_or_else(|| unresolved(&"it spends nothing"))?;
if head.is_some_and(|(t, v)| (t, v) != (spent.txid.as_str(), spent.vout) && t != txid) {
return Ok(false);
}
let prev = lookup.tx(&spent.txid).await.map_err(|e| unresolved(&e))?;
let address = prev
.vout
.get(spent.vout as usize)
.and_then(|o| o.scriptpubkey_address.clone())
.ok_or_else(|| unresolved(&refusal))?;
let utxos = lookup
.address_utxos(&address)
.await
.map_err(|e| unresolved(&e))?;
if utxos
.iter()
.any(|u| u.txid == spent.txid && u.vout == spent.vout)
{
Ok(false)
} else {
Err(unresolved(&refusal))
}
}
fn verify_intent_tx(
txid: &str,
raw_hex: &str,
head: Option<(&str, u32)>,
behind: bool,
) -> Result<(), ProvenanceError> {
let bad =
|why: String| ProvenanceError::Anchor(format!("pending intent {txid} refused: {why}"));
let info = solid_pod_rs::bitcoin_tx::decode_tx_info(raw_hex)
.map_err(|e| bad(format!("undecodable transaction: {e}")))?;
if info.txid != txid {
return Err(bad(format!("its transaction is {}", info.txid)));
}
if let (true, Some((head_txid, head_vout))) = (behind, head) {
let spends_head = info
.vin
.first()
.is_some_and(|i| i.txid == head_txid && i.vout == head_vout);
if !spends_head {
return Err(bad("it does not spend the stored head".into()));
}
}
Ok(())
}
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PendingMark {
#[serde(flatten)]
intent: crate::trail_store::GitmarkIntent,
#[serde(default)]
raw_hex: Option<String>,
}
async fn load_pending_mark(
storage: &std::sync::Arc<dyn solid_pod_rs::storage::Storage>,
name: &str,
) -> Result<Option<PendingMark>, ProvenanceError> {
use solid_pod_rs::PodError;
match storage
.get(&crate::trail_store::gitmark_intent_path(name))
.await
{
Ok((bytes, _)) => serde_json::from_slice(&bytes).map(Some).map_err(|e| {
ProvenanceError::Anchor(format!("read pending mark: malformed git-mark intent: {e}"))
}),
Err(PodError::NotFound(_)) => Ok(None),
Err(e) => Err(ProvenanceError::Anchor(format!("read pending mark: {e}"))),
}
}
async fn save_pending_mark(
storage: &std::sync::Arc<dyn solid_pod_rs::storage::Storage>,
pending: &PendingMark,
) -> Result<(), String> {
let body = serde_json::to_vec(pending).map_err(|e| e.to_string())?;
storage
.put(
&crate::trail_store::gitmark_intent_path(&pending.intent.stored.name),
bytes::Bytes::from(body),
"application/json",
)
.await
.map(|_| ())
.map_err(|e| e.to_string())
}
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PendingAnchor {
txid: String,
raw_hex: String,
stored: crate::trail_store::StoredTrail,
}
fn anchor_intent_path(ticker: &str) -> String {
format!("/.well-known/token/{}/intent.json", ticker.to_lowercase())
}
pub(crate) async fn settle_trail_intent<M: MempoolLookup + MempoolBroadcast + ?Sized>(
lookup: &M,
storage: &std::sync::Arc<dyn solid_pod_rs::storage::Storage>,
ticker: &str,
) -> Result<(), ProvenanceError> {
use crate::trail_store::{load_trail, save_trail};
use solid_pod_rs::PodError;
let err =
|what: &str, e: &dyn std::fmt::Display| ProvenanceError::Anchor(format!("{what}: {e}"));
let path = anchor_intent_path(ticker);
let pending: PendingAnchor = match storage.get(&path).await {
Ok((bytes, _)) => {
serde_json::from_slice(&bytes).map_err(|e| err("malformed pending anchor", &e))?
}
Err(PodError::NotFound(_)) => return Ok(()),
Err(e) => return Err(err("read pending anchor", &e)),
};
let stored = load_trail(storage, ticker)
.await
.map_err(|e| err("load trail", &e))?;
verify_intent_tx(
&pending.txid,
&pending.raw_hex,
stored
.as_ref()
.map(|t| (t.current_txid.as_str(), t.current_vout)),
stored.as_ref().map_or(0, |t| t.states.len()) < pending.stored.states.len(),
)?;
let landed = lookup.tx(&pending.txid).await.is_ok()
|| settle_broadcast(
lookup,
&pending.txid,
&pending.raw_hex,
stored
.as_ref()
.map(|t| (t.current_txid.as_str(), t.current_vout)),
)
.await?;
if landed && stored.map_or(0, |t| t.states.len()) < pending.stored.states.len() {
save_trail(storage, &pending.stored)
.await
.map_err(|e| err("save recovered trail", &e))?;
}
match storage.delete(&path).await {
Ok(()) | Err(PodError::NotFound(_)) => Ok(()),
Err(e) => Err(err("clear pending anchor", &e)),
}
}
#[async_trait(?Send)]
pub trait MempoolTxPresence {
async fn transaction_exists(&self, txid: &str) -> Result<bool, PaymentError>;
}
#[async_trait(?Send)]
impl MempoolTxPresence for MempoolHttpClient {
async fn transaction_exists(&self, txid: &str) -> Result<bool, PaymentError> {
MempoolHttpClient::transaction_exists(self, txid).await
}
}
#[derive(Clone)]
pub struct GitmarkAnchorer<M: MempoolLookup + MempoolBroadcast + MempoolTxPresence + Send + Sync> {
lookup: M,
storage: Option<std::sync::Arc<dyn solid_pod_rs::storage::Storage>>,
}
impl<M: MempoolLookup + MempoolBroadcast + MempoolTxPresence + Send + Sync> GitmarkAnchorer<M> {
pub fn new(lookup: M) -> Self {
Self {
lookup,
storage: None,
}
}
pub fn with_storage(
lookup: M,
storage: std::sync::Arc<dyn solid_pod_rs::storage::Storage>,
) -> Self {
Self {
lookup,
storage: Some(storage),
}
}
pub fn lookup(&self) -> &M {
&self.lookup
}
fn storage(
&self,
) -> Result<&std::sync::Arc<dyn solid_pod_rs::storage::Storage>, ProvenanceError> {
self.storage.as_ref().ok_or_else(|| {
ProvenanceError::Anchor(
"a git-mark trail needs storage; construct with GitmarkAnchorer::with_storage"
.into(),
)
})
}
async fn settle_intent(&self, name: &str) -> Result<(), ProvenanceError> {
use crate::trail_store::{clear_gitmark_intent, load_gitmark_trail, save_gitmark_trail};
let storage = self.storage()?;
let err = |what: &str, e: PaymentError| ProvenanceError::Anchor(format!("{what}: {e}"));
let Some(pending) = load_pending_mark(storage, name).await? else {
return Ok(());
};
let intent = &pending.intent;
let exists = self
.lookup
.transaction_exists(&intent.txid)
.await
.map_err(|e| {
ProvenanceError::Anchor(format!(
"a previous mark ({}) is unresolved, refusing to advance: {e}",
intent.txid
))
})?;
let stored = load_gitmark_trail(storage, name)
.await
.map_err(|e| err("load trail", e))?;
if let Some(raw_hex) = &pending.raw_hex {
verify_intent_tx(
&intent.txid,
raw_hex,
stored
.as_ref()
.and_then(|t| t.trail.txo.last())
.map(|m| (m.txid.as_str(), m.vout)),
stored.as_ref().map_or(0, |t| t.trail.txo.len()) < intent.stored.trail.txo.len(),
)?;
}
let landed = if exists {
true
} else if let Some(raw_hex) = &pending.raw_hex {
let head = stored
.as_ref()
.and_then(|t| t.trail.txo.last())
.map(|m| (m.txid.as_str(), m.vout));
settle_broadcast(&self.lookup, &intent.txid, raw_hex, head).await?
} else {
tracing::warn!(trail = %name, txid = %intent.txid, "pending mark has no raw transaction; dropped as not on-chain");
false
};
if landed && stored.map_or(0, |t| t.trail.txo.len()) < intent.stored.trail.txo.len() {
save_gitmark_trail(storage, &intent.stored)
.await
.map_err(|e| err("save recovered trail", e))?;
}
clear_gitmark_intent(storage, name)
.await
.map_err(|e| err("clear pending mark", e))
}
async fn broadcast_and_save(
&self,
name: &str,
privkey: String,
update: GitmarkUpdate,
network: &str,
date_created: String,
) -> Result<BlockTrailAnchor, ProvenanceError> {
use crate::trail_store::{
clear_gitmark_intent, save_gitmark_trail, GitmarkIntent, StoredGitmarkTrail,
};
let storage = self.storage()?;
let expected = solid_pod_rs::bitcoin_tx::decode_tx_info(&update.tx.raw_hex)
.map_err(|e| ProvenanceError::Anchor(format!("decode mark: {e}")))?
.txid;
let mut trail = update.trail;
if let Some(last) = trail.txo.last_mut() {
last.txid = expected.clone();
}
let state_strings = trail
.state_strings()
.map_err(|e| ProvenanceError::Anchor(format!("trail states: {e}")))?;
let pubkey = trail.pubkey_base.clone();
let stored = StoredGitmarkTrail {
name: name.to_string(),
privkey,
trail,
date_created,
};
save_pending_mark(
storage,
&PendingMark {
intent: GitmarkIntent {
txid: expected.clone(),
stored: stored.clone(),
},
raw_hex: Some(update.tx.raw_hex.clone()),
},
)
.await
.map_err(|e| ProvenanceError::Anchor(format!("record pending mark: {e}")))?;
let txid = self
.lookup
.broadcast_tx(&update.tx.raw_hex)
.await
.map_err(|e| pending_anchor(format!("broadcast mark: {e}")))?;
if txid != expected {
return Err(pending_anchor(format!(
"mempool returned txid {txid} for a mark that is {expected}"
)));
}
match save_gitmark_trail(storage, &stored).await {
Ok(()) => {
if let Err(e) = clear_gitmark_intent(storage, name).await {
tracing::warn!(trail = %name, "git-mark intent not cleared: {e}");
}
}
Err(e) => tracing::error!(
trail = %name, txid = %txid,
"git-mark broadcast accepted but trail save failed; the next advance settles it: {e}"
),
}
Ok(BlockTrailAnchor {
ticker: name.to_string(),
state_hash: update.txo.commit.clone().unwrap_or_default(),
txid,
vout: update.txo.vout,
address: update.address,
network: network.to_string(),
blockheight: None,
state_strings,
pubkey,
})
}
pub async fn genesis(
&self,
name: &str,
voucher: &TxoVoucher,
commit: &str,
network: &str,
date_created: &str,
) -> Result<BlockTrailAnchor, ProvenanceError> {
use crate::trail_store::{gitmark_trail_path, load_gitmark_trail};
let storage = self.storage()?;
let _guard = crate::trail_store::lock_path(&gitmark_trail_path(name)).await;
self.settle_intent(name).await?;
if load_gitmark_trail(storage, name)
.await
.map_err(|e| ProvenanceError::Anchor(format!("load trail {name}: {e}")))?
.is_some()
{
return Err(ProvenanceError::Anchor(format!(
"git-mark trail {name} already exists"
)));
}
let update = gitmark_genesis(voucher, commit, network, DEFAULT_FEE_SATS, &self.lookup)
.await
.map_err(|e| ProvenanceError::Anchor(format!("build genesis mark: {e}")))?;
self.broadcast_and_save(
name,
voucher.privkey.clone(),
update,
network,
date_created.to_string(),
)
.await
}
pub async fn verify_report(
&self,
anchor: &BlockTrailAnchor,
) -> Result<Option<TrailReport>, ProvenanceError> {
verify_anchor_report(&self.lookup, anchor).await
}
}
#[async_trait(?Send)]
impl<M: MempoolLookup + MempoolBroadcast + MempoolTxPresence + Send + Sync> BlockAnchorer
for GitmarkAnchorer<M>
{
async fn anchor(
&self,
ticker: &str,
state_hash: &str,
network: &str,
) -> Result<BlockTrailAnchor, ProvenanceError> {
use crate::trail_store::{gitmark_trail_path, load_gitmark_trail};
let _guard = crate::trail_store::lock_path(&gitmark_trail_path(ticker)).await;
self.settle_intent(ticker).await?;
let stored = load_gitmark_trail(self.storage()?, ticker)
.await
.map_err(|e| ProvenanceError::Anchor(format!("load trail {ticker}: {e}")))?
.ok_or_else(|| {
ProvenanceError::Anchor(format!(
"git-mark trail {ticker} has no genesis on this pod"
))
})?;
let trail_network = stored.network();
if trail_network != network {
return Err(ProvenanceError::Anchor(format!(
"network mismatch: trail is {trail_network}, requested {network}"
)));
}
let update = gitmark_advance(
&stored.trail,
&stored.privkey,
state_hash,
DEFAULT_FEE_SATS,
&self.lookup,
)
.await
.map_err(|e| ProvenanceError::Anchor(format!("build mark: {e}")))?;
self.broadcast_and_save(ticker, stored.privkey, update, network, stored.date_created)
.await
}
async fn verify(&self, anchor: &BlockTrailAnchor) -> Result<bool, ProvenanceError> {
Ok(verify_anchor_report(&self.lookup, anchor)
.await?
.is_some_and(|r| r.is_intact()))
}
}
#[cfg(test)]
mod tests {
use super::*;
const UTXO_JSON: &str = include_str!("../tests/fixtures/mempool/address_utxos.json");
const TX_JSON: &str = include_str!("../tests/fixtures/mempool/tx.json");
const EMPTY_UTXO_JSON: &str = "[]";
#[test]
fn utxo_wire_flattens_status() {
let wire: Vec<UtxoWire> = serde_json::from_str(UTXO_JSON).unwrap();
let utxos: Vec<Utxo> = wire.into_iter().map(Utxo::from).collect();
assert_eq!(utxos.len(), 1);
assert_eq!(utxos[0].vout, 0);
assert_eq!(utxos[0].value, 9700);
assert!(
utxos[0].confirmed,
"status.confirmed must flatten onto Utxo"
);
assert_eq!(utxos[0].block_height, Some(42_000));
}
#[test]
fn empty_utxo_set_parses_to_empty_vec() {
let wire: Vec<UtxoWire> = serde_json::from_str(EMPTY_UTXO_JSON).unwrap();
assert!(wire.is_empty());
}
#[test]
fn tx_wire_flattens_outputs_and_status() {
let wire: TxWire = serde_json::from_str(TX_JSON).unwrap();
let tx = TxInfo::from(wire);
assert_eq!(tx.vout.len(), 2);
assert_eq!(tx.vout[0].value, 9700);
assert_eq!(
tx.vout[0].scriptpubkey.as_deref(),
Some("5120aabbccddeeff00112233445566778899aabbccddeeff00112233445566778899")
);
assert_eq!(
tx.vout[0].scriptpubkey_address.as_deref(),
Some("tb1pexampleaddress")
);
assert!(tx.confirmed);
assert_eq!(tx.block_height, Some(42_000));
}
#[test]
fn from_env_defaults_to_testnet4() {
let prev = std::env::var(MEMPOOL_URL_ENV).ok();
std::env::remove_var(MEMPOOL_URL_ENV);
let c = MempoolHttpClient::from_env();
assert_eq!(c.base_url(), DEFAULT_MEMPOOL_URL);
if let Some(v) = prev {
std::env::set_var(MEMPOOL_URL_ENV, v);
}
}
#[test]
fn new_trims_trailing_slash() {
let c = MempoolHttpClient::new("https://mempool.space/testnet4/");
assert_eq!(c.base_url(), "https://mempool.space/testnet4");
}
#[ignore = "hits live mempool.space; opt-in only"]
#[tokio::test]
async fn live_address_utxos_smoke() {
let c = MempoolHttpClient::from_env();
let _ = c
.address_utxos("tb1pqqqqp399et2xygdj5xreqhjjvcmzhxw4aywxecjdzew6hylgvsesf3hn0c")
.await;
}
use std::collections::HashMap;
#[derive(Clone, Default)]
struct FixtureMempool {
utxos: std::sync::Arc<std::sync::Mutex<HashMap<String, Vec<Utxo>>>>,
txs: std::sync::Arc<std::sync::Mutex<HashMap<String, TxInfo>>>,
broadcasts: std::sync::Arc<std::sync::Mutex<Vec<String>>>,
}
impl FixtureMempool {
fn empty() -> Self {
Self::default()
}
fn add_output(&self, txid: &str, vout: u32, spk_hex: &str) {
let mut txs = self.txs.lock().unwrap();
let tx = txs.entry(txid.to_string()).or_insert_with(|| TxInfo {
txid: txid.to_string(),
vin: vec![],
vout: vec![],
confirmed: true,
block_height: Some(41_000),
});
while tx.vout.len() <= vout as usize {
tx.vout.push(TxOut {
value: 0,
scriptpubkey: None,
scriptpubkey_address: None,
});
}
tx.vout[vout as usize].scriptpubkey = Some(spk_hex.to_string());
}
fn add_chain(&self, pubkey: &str, state_strings: &[String]) -> String {
let outputs = solid_pod_rs::mrc20::bt_trail_outputs(pubkey, state_strings).unwrap();
let mut prev = "ff".repeat(32);
for (i, x) in outputs.iter().enumerate() {
let txid = solid_pod_rs::mrc20::sha256_hex(&format!("{pubkey} mark {i}"));
self.txs.lock().unwrap().insert(
txid.clone(),
TxInfo {
txid: txid.clone(),
vin: vec![TxIn {
txid: prev,
vout: 0,
}],
vout: vec![TxOut {
value: 9_700,
scriptpubkey: Some(format!("5120{}", hex::encode(x))),
scriptpubkey_address: None,
}],
confirmed: true,
block_height: Some(42_000 + i as u64),
},
);
prev = txid;
}
prev
}
}
#[async_trait(?Send)]
impl MempoolLookup for FixtureMempool {
async fn address_utxos(&self, address: &str) -> Result<Vec<Utxo>, PaymentError> {
Ok(self
.utxos
.lock()
.unwrap()
.get(address)
.cloned()
.unwrap_or_default())
}
async fn tx(&self, txid: &str) -> Result<TxInfo, PaymentError> {
self.txs
.lock()
.unwrap()
.get(txid)
.cloned()
.ok_or_else(|| PaymentError::InvalidState(format!("tx {txid} not found")))
}
}
#[async_trait(?Send)]
impl MempoolBroadcast for FixtureMempool {
async fn broadcast_tx(&self, raw_hex: &str) -> Result<String, PaymentError> {
let mut info = solid_pod_rs::bitcoin_tx::decode_tx_info(raw_hex)?;
info.confirmed = true;
info.block_height = Some(43_000);
let txid = info.txid.clone();
self.txs.lock().unwrap().insert(txid.clone(), info);
self.broadcasts.lock().unwrap().push(raw_hex.to_string());
Ok(txid)
}
}
#[async_trait(?Send)]
impl MempoolTxPresence for FixtureMempool {
async fn transaction_exists(&self, txid: &str) -> Result<bool, PaymentError> {
Ok(self.txs.lock().unwrap().contains_key(txid))
}
}
const ISSUER_PRIVKEY: &str = "0000000000000000000000000000000000000000000000000000000000000001";
fn issuer_pubkey() -> String {
let sk = k256::SecretKey::from_slice(&hex::decode(ISSUER_PRIVKEY).unwrap()).unwrap();
hex::encode(sk.public_key().to_sec1_bytes())
}
fn consistent_anchor(txid: &str) -> BlockTrailAnchor {
let pubkey = issuer_pubkey();
let state_strings = vec!["{\"seq\":0}".to_string(), "{\"seq\":1}".to_string()];
let address = bt_address(&pubkey, &state_strings, "testnet4").unwrap();
BlockTrailAnchor {
ticker: "PROV".into(),
state_hash: "ff".repeat(32),
txid: txid.to_string(),
vout: 0,
address,
network: "testnet4".into(),
blockheight: Some(42_000),
state_strings,
pubkey: Some(pubkey),
}
}
fn anchored_chain() -> (FixtureMempool, BlockTrailAnchor) {
let mempool = FixtureMempool::empty();
let probe = consistent_anchor("");
let head = mempool.add_chain(&issuer_pubkey(), &probe.state_strings);
(mempool, consistent_anchor(&head))
}
#[test]
fn tx_wire_reads_inputs() {
let wire: TxWire = serde_json::from_str(
r#"{"txid":"aa","vin":[{"txid":"bb","vout":3,"prevout":{}},{"is_coinbase":true}],"vout":[],"status":{"confirmed":false}}"#,
)
.unwrap();
let tx = TxInfo::from(wire);
assert_eq!(tx.vin.len(), 2);
assert_eq!((tx.vin[0].txid.as_str(), tx.vin[0].vout), ("bb", 3));
assert!(!tx.confirmed);
}
#[tokio::test]
async fn block_anchorer_verify_true_when_every_link_is_on_chain() {
let (mempool, anchor) = anchored_chain();
let anchorer = MempoolBlockAnchorer::new(mempool);
assert!(anchorer.verify(&anchor).await.unwrap());
let report = anchorer.verify_report(&anchor).await.unwrap().unwrap();
assert!(report.is_verified());
assert_eq!(report.marks.len(), 2);
}
#[tokio::test]
async fn block_anchorer_verify_false_when_the_anchor_tx_is_absent() {
let anchor = consistent_anchor(&"ab".repeat(32));
let anchorer = MempoolBlockAnchorer::new(FixtureMempool::empty());
assert!(
!anchorer.verify(&anchor).await.unwrap(),
"no transaction behind the anchor ⇒ verify false"
);
}
#[tokio::test]
async fn block_anchorer_verify_false_when_an_earlier_mark_does_not_commit() {
let (mempool, anchor) = anchored_chain();
let genesis = solid_pod_rs::mrc20::sha256_hex(&format!("{} mark 0", issuer_pubkey()));
mempool.txs.lock().unwrap().get_mut(&genesis).unwrap().vout[0].scriptpubkey =
Some(format!("5120{}", "22".repeat(32)));
let anchorer = MempoolBlockAnchorer::new(mempool);
assert!(!anchorer.verify(&anchor).await.unwrap());
let report = anchorer.verify_report(&anchor).await.unwrap().unwrap();
assert_eq!(
report.marks[0].status,
solid_pod_rs::blocktrail::MarkStatus::WrongKey
);
}
#[tokio::test]
async fn block_anchorer_verify_false_when_address_forged() {
let (mempool, mut anchor) = anchored_chain();
anchor.address = "tb1pforged000000000000000000000000000000".into();
let anchorer = MempoolBlockAnchorer::new(mempool);
assert!(
!anchorer.verify(&anchor).await.unwrap(),
"forged address must not verify even with a real chain behind it"
);
assert!(anchorer.verify_report(&anchor).await.unwrap().is_none());
}
#[tokio::test]
async fn block_anchorer_verify_false_without_pubkey() {
let (mempool, mut anchor) = anchored_chain();
anchor.pubkey = None;
let anchorer = MempoolBlockAnchorer::new(mempool);
assert!(!anchorer.verify(&anchor).await.unwrap());
}
#[tokio::test]
async fn block_anchorer_anchor_requires_storage() {
let anchorer = MempoolBlockAnchorer::new(FixtureMempool::empty());
let err = anchorer
.anchor("PROV", "deadbeef", "testnet4")
.await
.unwrap_err();
match err {
ProvenanceError::Anchor(m) => assert!(m.contains("with_storage")),
other => panic!("expected Anchor(requires storage), got {other:?}"),
}
}
use crate::trail_store::{load_gitmark_trail, load_trail, save_trail, StoredTrail};
use solid_pod_rs::bitcoin_tx::mint_token;
use solid_pod_rs::storage::memory::MemoryBackend;
use solid_pod_rs::storage::Storage;
fn issuer_voucher(mempool: &FixtureMempool, txid: &str) -> TxoVoucher {
let sk = k256::SecretKey::from_slice(&hex::decode(ISSUER_PRIVKEY).unwrap()).unwrap();
let compressed = sk.public_key().to_sec1_bytes();
mempool.add_output(txid, 0, &format!("5120{}", hex::encode(&compressed[1..])));
TxoVoucher {
txid: txid.to_string(),
vout: 0,
amount: 100_000,
privkey: ISSUER_PRIVKEY.to_string(),
}
}
async fn mint_and_store(ticker: &str) -> (std::sync::Arc<dyn Storage>, FixtureMempool, String) {
let mempool = FixtureMempool::empty();
let storage: std::sync::Arc<dyn Storage> = std::sync::Arc::new(MemoryBackend::new());
let voucher = issuer_voucher(&mempool, &"11".repeat(32));
let mint = mint_token(ticker, None, 1_000, &voucher, "testnet4", 300, &mempool)
.await
.unwrap();
let mint_txid = mempool.broadcast_tx(&mint.tx.raw_hex).await.unwrap();
let stored = StoredTrail {
ticker: mint.trail.ticker.clone(),
name: mint.trail.name.clone(),
supply: mint.trail.supply,
privkey: ISSUER_PRIVKEY.to_string(),
pubkey_base: mint.trail.pubkey_base.clone(),
states: mint.trail.states.clone(),
state_strings: mint.trail.state_strings.clone(),
current_txid: mint_txid,
current_vout: 0,
current_amount: mint.trail.current_amount,
network: mint.trail.network.clone(),
date_created: "2026-06-13T00:00:00Z".into(),
};
save_trail(&storage, &stored).await.unwrap();
(storage, mempool, ticker.to_string())
}
#[tokio::test]
async fn block_anchorer_anchor_round_trip_and_self_verifies() {
let (storage, mempool, ticker) = mint_and_store("ANCH").await;
let anchorer = MempoolBlockAnchorer::with_storage(mempool.clone(), storage.clone());
let commit_sha = "a1b2c3d4e5f60718293a4b5c6d7e8f9001122334";
let anchor = anchorer
.anchor(&ticker, commit_sha, "testnet4")
.await
.expect("anchor() must build + broadcast + persist");
assert_eq!(anchor.ticker, "ANCH");
assert_eq!(anchor.state_hash, commit_sha);
assert_eq!(anchor.vout, 0);
assert!(anchor.blockheight.is_none());
assert_eq!(anchor.network, "testnet4");
assert!(anchor.pubkey.is_some());
assert_eq!(anchor.state_strings.len(), 2);
let derived = bt_address(
anchor.pubkey.as_deref().unwrap(),
&anchor.state_strings,
"testnet4",
)
.unwrap();
assert_eq!(anchor.address, derived);
let reloaded = load_trail(&storage, "ANCH").await.unwrap().unwrap();
assert_eq!(reloaded.states.len(), 2);
assert_eq!(reloaded.current_txid, anchor.txid);
assert_eq!(reloaded.states[1].anchor.as_deref(), Some(commit_sha));
assert_eq!(reloaded.states[1].ops[0].op, "urn:mono:op:anchor");
assert!(
anchorer.verify(&anchor).await.unwrap(),
"the anchor we just produced must verify, every link"
);
let second = anchorer
.anchor(&ticker, &"b".repeat(40), "testnet4")
.await
.unwrap();
assert_eq!(second.state_strings.len(), 3);
assert!(anchorer.verify(&second).await.unwrap());
assert!(anchorer.verify(&anchor).await.unwrap());
}
#[tokio::test]
async fn block_anchorer_anchor_rejects_unminted_ticker() {
let storage: std::sync::Arc<dyn Storage> = std::sync::Arc::new(MemoryBackend::new());
let anchorer = MempoolBlockAnchorer::with_storage(FixtureMempool::empty(), storage);
let err = anchorer
.anchor("GHOST", "deadbeef", "testnet4")
.await
.unwrap_err();
match err {
ProvenanceError::Anchor(m) => assert!(m.contains("not minted")),
other => panic!("expected not-minted error, got {other:?}"),
}
}
const COMMITS: [&str; 3] = [
"0123456789abcdef0123456789abcdef01234567",
"cf97baba489e88c1ffbe6758c0fe8c18ff83d17d",
"a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2",
];
#[tokio::test]
async fn gitmark_anchorer_genesis_anchor_verify() {
let mempool = FixtureMempool::empty();
let storage: std::sync::Arc<dyn Storage> = std::sync::Arc::new(MemoryBackend::new());
let anchorer = GitmarkAnchorer::with_storage(mempool.clone(), storage.clone());
let voucher = issuer_voucher(&mempool, &"33".repeat(32));
let genesis = anchorer
.genesis(
"prov",
&voucher,
COMMITS[0],
"testnet4",
"2026-10-02T00:00:00Z",
)
.await
.unwrap();
assert_eq!(genesis.state_strings, vec![COMMITS[0].to_string()]);
assert_eq!(genesis.pubkey.as_deref(), Some(issuer_pubkey().as_str()));
assert!(anchorer.verify(&genesis).await.unwrap());
let again = anchorer
.genesis("prov", &voucher, COMMITS[0], "testnet4", "")
.await
.unwrap_err();
assert!(again.to_string().contains("already exists"));
let mut last = genesis.clone();
for commit in &COMMITS[1..] {
last = anchorer.anchor("prov", commit, "testnet4").await.unwrap();
assert_eq!(last.state_hash, *commit);
}
assert_eq!(
last.state_strings,
COMMITS.iter().map(|c| c.to_string()).collect::<Vec<_>>()
);
let report = anchorer.verify_report(&last).await.unwrap().unwrap();
assert!(report.is_verified(), "{report:?}");
assert_eq!(report.marks.len(), 3);
assert!(anchorer.verify(&genesis).await.unwrap());
let stored = load_gitmark_trail(&storage, "prov").await.unwrap().unwrap();
assert_eq!(stored.trail.txo.len(), 3);
assert_eq!(stored.trail.txo[2].txid, last.txid);
let report = solid_pod_rs::blocktrail::verify_blocktrail(&stored.trail, &mempool).await;
assert!(report.is_verified(), "{report:?}");
let (published, _) = storage
.get(&crate::trail_store::gitmark_blocktrails_path("prov"))
.await
.unwrap();
let published: solid_pod_rs::blocktrail::Blocktrail =
serde_json::from_slice(&published).unwrap();
assert_eq!(published, stored.trail);
}
#[tokio::test]
async fn gitmark_anchorer_refusals() {
let mempool = FixtureMempool::empty();
let storage: std::sync::Arc<dyn Storage> = std::sync::Arc::new(MemoryBackend::new());
let anchorer = GitmarkAnchorer::with_storage(mempool.clone(), storage);
let err = anchorer
.anchor("prov", COMMITS[0], "testnet4")
.await
.unwrap_err();
assert!(err.to_string().contains("no genesis"), "{err}");
let voucher = issuer_voucher(&mempool, &"34".repeat(32));
anchorer
.genesis("prov", &voucher, COMMITS[0], "testnet4", "")
.await
.unwrap();
let err = anchorer
.anchor("prov", COMMITS[1], "mainnet")
.await
.unwrap_err();
assert!(err.to_string().contains("network mismatch"), "{err}");
let err = anchorer
.anchor("prov", "not-a-commit", "testnet4")
.await
.unwrap_err();
assert!(err.to_string().contains("commit hash"), "{err}");
let verify_only = GitmarkAnchorer::new(mempool);
let err = verify_only
.anchor("prov", COMMITS[1], "testnet4")
.await
.unwrap_err();
assert!(err.to_string().contains("with_storage"), "{err}");
}
#[derive(Clone)]
struct RacyMempool {
inner: FixtureMempool,
spent: std::sync::Arc<std::sync::Mutex<std::collections::HashSet<(String, u32)>>>,
drop_next_broadcast: std::sync::Arc<std::sync::atomic::AtomicBool>,
index_lag: std::sync::Arc<std::sync::atomic::AtomicBool>,
reject_next: std::sync::Arc<std::sync::Mutex<Option<String>>>,
}
impl RacyMempool {
fn new(inner: FixtureMempool) -> Self {
Self {
inner,
spent: Default::default(),
drop_next_broadcast: Default::default(),
index_lag: Default::default(),
reject_next: Default::default(),
}
}
fn reject_next(&self, msg: &str) {
*self.reject_next.lock().unwrap() = Some(msg.into());
}
fn show_unspent(&self, txid: &str, vout: u32) {
let address = format!("tb1q{txid}{vout}");
self.inner.txs.lock().unwrap().get_mut(txid).unwrap().vout[vout as usize]
.scriptpubkey_address = Some(address.clone());
self.inner.utxos.lock().unwrap().insert(
address,
vec![Utxo {
txid: txid.into(),
vout,
value: 9_700,
confirmed: true,
block_height: Some(1),
}],
);
}
}
#[async_trait(?Send)]
impl MempoolLookup for RacyMempool {
async fn address_utxos(&self, address: &str) -> Result<Vec<Utxo>, PaymentError> {
tokio::task::yield_now().await;
self.inner.address_utxos(address).await
}
async fn tx(&self, txid: &str) -> Result<TxInfo, PaymentError> {
tokio::task::yield_now().await;
self.inner.tx(txid).await
}
}
#[async_trait(?Send)]
impl MempoolBroadcast for RacyMempool {
async fn broadcast_tx(&self, raw_hex: &str) -> Result<String, PaymentError> {
tokio::task::yield_now().await;
if self
.drop_next_broadcast
.swap(false, std::sync::atomic::Ordering::SeqCst)
{
return Err(PaymentError::InvalidState("connection reset".into()));
}
let info = solid_pod_rs::bitcoin_tx::decode_tx_info(raw_hex)?;
if let Some(msg) = self.reject_next.lock().unwrap().take() {
return Err(PaymentError::InvalidState(msg));
}
if self.inner.txs.lock().unwrap().contains_key(&info.txid) {
return Err(PaymentError::InvalidState(
"broadcast rejected (400): txn-already-in-mempool".into(),
));
}
{
let mut spent = self.spent.lock().unwrap();
if info
.vin
.iter()
.any(|i| spent.contains(&(i.txid.clone(), i.vout)))
{
return Err(PaymentError::InvalidState("txn-mempool-conflict".into()));
}
for i in &info.vin {
spent.insert((i.txid.clone(), i.vout));
}
}
self.inner.broadcast_tx(raw_hex).await
}
}
#[async_trait(?Send)]
impl MempoolTxPresence for RacyMempool {
async fn transaction_exists(&self, txid: &str) -> Result<bool, PaymentError> {
tokio::task::yield_now().await;
if self.index_lag.load(std::sync::atomic::Ordering::SeqCst) {
return Ok(false);
}
self.inner.transaction_exists(txid).await
}
}
async fn racy_trail() -> (
RacyMempool,
std::sync::Arc<crate::trail_store::fault_storage::FaultStorage>,
GitmarkAnchorer<RacyMempool>,
GitmarkAnchorer<RacyMempool>,
) {
let fixture = FixtureMempool::empty();
let voucher = issuer_voucher(&fixture, &"35".repeat(32));
let mempool = RacyMempool::new(fixture);
let inner: std::sync::Arc<dyn Storage> = std::sync::Arc::new(MemoryBackend::new());
let storage = crate::trail_store::fault_storage::FaultStorage::new(inner, true);
let a = GitmarkAnchorer::with_storage(mempool.clone(), storage.clone());
let b = GitmarkAnchorer::with_storage(mempool.clone(), storage.clone());
a.genesis("prov", &voucher, COMMITS[0], "testnet4", "")
.await
.unwrap();
(mempool, storage, a, b)
}
#[tokio::test]
async fn concurrent_anchors_build_on_each_others_head() {
let (mempool, storage, a, b) = racy_trail().await;
let (ra, rb) = tokio::join!(
a.anchor("prov", COMMITS[1], "testnet4"),
b.anchor("prov", COMMITS[2], "testnet4"),
);
let (ra, rb) = (ra.unwrap(), rb.unwrap());
let dyn_storage: std::sync::Arc<dyn Storage> = storage.clone();
let stored = load_gitmark_trail(&dyn_storage, "prov")
.await
.unwrap()
.unwrap();
assert_eq!(stored.trail.txo.len(), 3, "both anchors must land");
let heads = [ra.txid.clone(), rb.txid.clone()];
assert_eq!(
heads
.iter()
.filter(|t| **t == stored.trail.txo[2].txid)
.count(),
1
);
assert!(heads.contains(&stored.trail.txo[1].txid));
let report = solid_pod_rs::blocktrail::verify_blocktrail(&stored.trail, &mempool).await;
assert!(report.is_verified(), "{report:?}");
}
#[tokio::test]
async fn broadcast_then_failed_save_is_ok_and_recovered_by_the_next_anchor() {
let (mempool, storage, a, _b) = racy_trail().await;
storage.fail_put(&crate::trail_store::gitmark_trail_path("prov"));
let landed = a
.anchor("prov", COMMITS[1], "testnet4")
.await
.expect("the tx is on-chain: a failed save must not read as a failed anchor");
storage.heal();
let next = a.anchor("prov", COMMITS[2], "testnet4").await.unwrap();
let dyn_storage: std::sync::Arc<dyn Storage> = storage.clone();
let stored = load_gitmark_trail(&dyn_storage, "prov")
.await
.unwrap()
.unwrap();
assert_eq!(stored.trail.txo.len(), 3);
assert_eq!(stored.trail.txo[1].txid, landed.txid);
assert_eq!(stored.trail.txo[2].txid, next.txid);
let report = solid_pod_rs::blocktrail::verify_blocktrail(&stored.trail, &mempool).await;
assert!(report.is_verified(), "{report:?}");
}
#[test]
fn already_known_reads_only_node_statements_about_the_same_tx() {
let e = |m: &str| PaymentError::InvalidState(m.into());
assert!(already_known(&e(
"broadcast rejected (400): txn-already-in-mempool"
)));
assert!(already_known(&e(
"broadcast rejected (400): Transaction already in block chain"
)));
assert!(already_known(&e(
"broadcast rejected (400): txn-already-known"
)));
assert!(!already_known(&e(
"broadcast rejected (400): txn-mempool-conflict"
)));
assert!(!already_known(&e(
"mempool broadcast failed: connection reset"
)));
}
#[tokio::test]
async fn dropped_broadcast_is_resent_by_the_next_anchor() {
let (mempool, storage, a, _b) = racy_trail().await;
mempool
.drop_next_broadcast
.store(true, std::sync::atomic::Ordering::SeqCst);
let lost = a.anchor("prov", COMMITS[1], "testnet4").await;
assert!(
is_pending_anchor(&lost.unwrap_err()),
"a recorded but unconfirmed broadcast is pending, not failed"
);
let dyn_storage: std::sync::Arc<dyn Storage> = storage.clone();
assert!(
crate::trail_store::load_gitmark_intent(&dyn_storage, "prov")
.await
.unwrap()
.is_some()
);
let next = a.anchor("prov", COMMITS[2], "testnet4").await.unwrap();
let stored = load_gitmark_trail(&dyn_storage, "prov")
.await
.unwrap()
.unwrap();
assert_eq!(stored.trail.txo.len(), 3, "the re-sent mark landed");
assert_eq!(stored.trail.txo[2].txid, next.txid);
assert!(
crate::trail_store::load_gitmark_intent(&dyn_storage, "prov")
.await
.unwrap()
.is_none()
);
let report = solid_pod_rs::blocktrail::verify_blocktrail(&stored.trail, &mempool).await;
assert!(report.is_verified(), "{report:?}");
}
#[tokio::test]
async fn unresolved_resend_keeps_the_intent_and_refuses_to_advance() {
let (mempool, storage, a, _b) = racy_trail().await;
let drop_next = || {
mempool
.drop_next_broadcast
.store(true, std::sync::atomic::Ordering::SeqCst)
};
drop_next();
a.anchor("prov", COMMITS[1], "testnet4").await.unwrap_err();
drop_next();
a.anchor("prov", COMMITS[2], "testnet4").await.unwrap_err();
let dyn_storage: std::sync::Arc<dyn Storage> = storage.clone();
assert!(
crate::trail_store::load_gitmark_intent(&dyn_storage, "prov")
.await
.unwrap()
.is_some(),
"an unresolved mark must keep its intent"
);
let stored = load_gitmark_trail(&dyn_storage, "prov")
.await
.unwrap()
.unwrap();
assert_eq!(stored.trail.txo.len(), 1);
}
#[tokio::test]
async fn rejected_resend_with_an_unspent_head_drops_the_intent() {
let (mempool, storage, a, _b) = racy_trail().await;
let dyn_storage: std::sync::Arc<dyn Storage> = storage.clone();
let head = load_gitmark_trail(&dyn_storage, "prov")
.await
.unwrap()
.unwrap()
.trail
.txo[0]
.txid
.clone();
mempool
.drop_next_broadcast
.store(true, std::sync::atomic::Ordering::SeqCst);
a.anchor("prov", COMMITS[1], "testnet4").await.unwrap_err();
mempool.reject_next("broadcast rejected (400): min relay fee not met");
mempool.show_unspent(&head, 0);
a.anchor("prov", COMMITS[2], "testnet4")
.await
.expect("a refused mark must not wedge the trail");
let stored = load_gitmark_trail(&dyn_storage, "prov")
.await
.unwrap()
.unwrap();
assert_eq!(stored.trail.txo.len(), 2, "the refused mark never landed");
assert!(
crate::trail_store::load_gitmark_intent(&dyn_storage, "prov")
.await
.unwrap()
.is_none()
);
}
#[tokio::test]
async fn rejected_resend_without_proof_of_an_unspent_head_keeps_refusing() {
let (mempool, storage, a, _b) = racy_trail().await;
mempool
.drop_next_broadcast
.store(true, std::sync::atomic::Ordering::SeqCst);
a.anchor("prov", COMMITS[1], "testnet4").await.unwrap_err();
mempool.reject_next("broadcast rejected (400): bad-txns-inputs-missingorspent");
a.anchor("prov", COMMITS[2], "testnet4").await.unwrap_err();
let dyn_storage: std::sync::Arc<dyn Storage> = storage.clone();
assert!(
crate::trail_store::load_gitmark_intent(&dyn_storage, "prov")
.await
.unwrap()
.is_some()
);
}
#[tokio::test]
async fn block_anchorer_rejected_resend_with_an_unspent_head_drops_the_intent() {
let (inner, fixture, ticker) = mint_and_store("DRP").await;
let mempool = RacyMempool::new(fixture);
let anchorer = MempoolBlockAnchorer::with_storage(mempool.clone(), inner.clone());
let head = load_trail(&inner, &ticker)
.await
.unwrap()
.unwrap()
.current_txid;
mempool
.drop_next_broadcast
.store(true, std::sync::atomic::Ordering::SeqCst);
anchorer
.anchor(&ticker, &"a".repeat(40), "testnet4")
.await
.unwrap_err();
mempool.reject_next("broadcast rejected (400): min relay fee not met");
mempool.show_unspent(&head, 0);
anchorer
.anchor(&ticker, &"b".repeat(40), "testnet4")
.await
.expect("a refused anchor must not wedge the trail");
let stored = load_trail(&inner, &ticker).await.unwrap().unwrap();
assert_eq!(stored.states.len(), 2);
assert!(!inner.exists(&anchor_intent_path(&ticker)).await.unwrap());
}
#[tokio::test]
async fn block_anchorer_refuses_an_intent_whose_txid_is_not_its_transaction() {
let (inner, fixture, ticker) = mint_and_store("TMP").await;
let mempool = RacyMempool::new(fixture);
let anchorer = MempoolBlockAnchorer::with_storage(mempool.clone(), inner.clone());
mempool
.drop_next_broadcast
.store(true, std::sync::atomic::Ordering::SeqCst);
anchorer
.anchor(&ticker, &"a".repeat(40), "testnet4")
.await
.unwrap_err();
let path = anchor_intent_path(&ticker);
let head = load_trail(&inner, &ticker)
.await
.unwrap()
.unwrap()
.current_txid;
let (bytes, _) = inner.get(&path).await.unwrap();
let mut v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
v["txid"] = serde_json::json!(head);
inner
.put(
&path,
bytes::Bytes::from(serde_json::to_vec(&v).unwrap()),
"application/json",
)
.await
.unwrap();
let before = load_trail(&inner, &ticker).await.unwrap().unwrap();
let err = anchorer
.anchor(&ticker, &"b".repeat(40), "testnet4")
.await
.unwrap_err();
assert!(err.to_string().contains("refused"), "{err}");
let after = load_trail(&inner, &ticker).await.unwrap().unwrap();
assert_eq!(after.states.len(), before.states.len());
assert!(inner.exists(&path).await.unwrap(), "the intent is kept");
}
#[tokio::test]
async fn accepted_but_unindexed_mark_is_not_orphaned_by_the_next_anchor() {
let (mempool, storage, a, _b) = racy_trail().await;
storage.fail_put(&crate::trail_store::gitmark_trail_path("prov"));
let landed = a.anchor("prov", COMMITS[1], "testnet4").await.unwrap();
storage.heal();
mempool
.index_lag
.store(true, std::sync::atomic::Ordering::SeqCst);
let next = a
.anchor("prov", COMMITS[2], "testnet4")
.await
.expect("the next advance must build on the accepted mark, not on the old head");
let dyn_storage: std::sync::Arc<dyn Storage> = storage.clone();
let stored = load_gitmark_trail(&dyn_storage, "prov")
.await
.unwrap()
.unwrap();
assert_eq!(stored.trail.txo.len(), 3);
assert_eq!(stored.trail.txo[1].txid, landed.txid);
assert_eq!(stored.trail.txo[2].txid, next.txid);
let report = solid_pod_rs::blocktrail::verify_blocktrail(&stored.trail, &mempool).await;
assert!(report.is_verified(), "{report:?}");
}
#[tokio::test]
async fn block_anchorer_accepted_broadcast_with_failed_save_is_not_a_failed_anchor() {
use crate::trail_store::fault_storage::FaultStorage;
let (inner, mempool, ticker) = mint_and_store("FLT").await;
let storage = FaultStorage::new(inner, false);
let anchorer = MempoolBlockAnchorer::with_storage(mempool.clone(), storage.clone());
storage.fail_put(&crate::trail_store::trail_path(&ticker));
let first = anchorer
.anchor(&ticker, &"a".repeat(40), "testnet4")
.await
.expect("the tx is on-chain: a failed save must not read as a failed anchor");
storage.heal();
let second = anchorer
.anchor(&ticker, &"b".repeat(40), "testnet4")
.await
.unwrap();
let dyn_storage: std::sync::Arc<dyn Storage> = storage.clone();
let stored = load_trail(&dyn_storage, &ticker).await.unwrap().unwrap();
assert_eq!(stored.states.len(), 3);
assert_eq!(stored.current_txid, second.txid);
let second_tx = mempool
.txs
.lock()
.unwrap()
.get(&second.txid)
.cloned()
.unwrap();
assert_eq!(
second_tx.vin[0].txid, first.txid,
"the second anchor must spend the first, not the old head"
);
}
#[tokio::test]
async fn http_client_presence_check_reaches_the_inherent_method() {
let client = MempoolHttpClient::new("http://127.0.0.1:1");
let out =
<MempoolHttpClient as MempoolTxPresence>::transaction_exists(&client, &"ab".repeat(32))
.await;
assert!(out.is_err(), "{out:?}");
}
}