commit 0b53546c861f33ed588dc3bba314b2c7fcd9389c
parent 324a965cc1d702d8ca7dc413596ee7758e025547
Author: Antoine A <>
Date: Fri, 3 Jul 2026 16:43:38 +0200
Improve sync state machine, improve db test and rewrite test harness
Diffstat:
36 files changed, 2338 insertions(+), 2964 deletions(-)
diff --git a/.gitignore b/.gitignore
@@ -17,9 +17,10 @@ taler.conf
*.mk
testbench/instrumentation
testbench/env
+depolymerizer-bitcoin/test
tools
debian/depolymerizer-bitcoin
debian/files
debian/*.substvars
debian/*debhelper*
-.history
-\ No newline at end of file
+.*history
+\ No newline at end of file
diff --git a/Cargo.toml b/Cargo.toml
@@ -3,8 +3,7 @@ resolver = "3"
members = [
"depolymerizer-bitcoin",
"depolymerizer-common",
- "uri-pack",
- "testbench",
+ "uri-pack"
]
[workspace.package]
@@ -35,6 +34,7 @@ taler-common = { git = "git://git.taler.net/taler-rust.git/" }
taler-api = { git = "git://git.taler.net/taler-rust.git/" }
taler-build = { git = "git://git.taler.net/taler-rust.git/" }
taler-test-utils = { git = "git://git.taler.net/taler-rust.git/" }
+failure-injection = { git = "git://git.taler.net/taler-rust.git/" }
depolymerizer-common = { path = "depolymerizer-common" }
bitcoin = { version = "0.32.5", features = [
"std",
@@ -50,3 +50,4 @@ rand = { version = "0.10.1" }
tracing-subscriber = "0.3"
jiff = { version = "0.2", default-features = false, features = ["perf-inline", "std"] }
compact_str = { version = "0.9.0", features = ["serde", "sqlx-postgres"] }
+owo-colors = "4.2.3"
diff --git a/depolymerizer-bitcoin/Cargo.toml b/depolymerizer-bitcoin/Cargo.toml
@@ -8,10 +8,6 @@ homepage.workspace = true
repository.workspace = true
license-file.workspace = true
-[features]
-# Enable random failures
-fail = []
-
[dependencies]
bech32 = "0.12.0"
serde_repr = "0.1.16"
@@ -19,13 +15,14 @@ depolymerizer-common.workspace = true
bitcoin.workspace = true
clap.workspace = true
serde.workspace = true
-serde_json.workspace = true
+serde_json = {workspace = true, features = ["raw_value"]}
thiserror.workspace = true
hex.workspace = true
anyhow.workspace = true
taler-api.workspace = true
taler-build.workspace = true
taler-common.workspace = true
+failure-injection.workspace = true
sqlx.workspace = true
tokio.workspace = true
tracing.workspace = true
@@ -35,10 +32,11 @@ rand.workspace = true
url.workspace = true
jiff.workspace = true
compact_str.workspace = true
+owo-colors.workspace = true
+taler-test-utils.workspace = true
[dev-dependencies]
criterion.workspace = true
-taler-test-utils.workspace = true
[[bench]]
name = "metadata"
diff --git a/depolymerizer-bitcoin/db/depolymerizer-bitcoin-0001.sql b/depolymerizer-bitcoin/db/depolymerizer-bitcoin-0001.sql
@@ -1,6 +1,6 @@
--
-- This file is part of TALER
--- Copyright (C) 2025, 2026 Taler Systems SA
+-- Copyright (C) 2025-2026 Taler Systems SA
--
-- TALER is free software; you can redistribute it and/or modify it under the
-- terms of the GNU General Public License as published by the Free Software
@@ -91,6 +91,7 @@ COMMENT ON TABLE pending_recurrent_in IS 'Pending recurrent incoming transaction
CREATE TYPE debit_status AS ENUM(
'requested',
+ 'ignored',
'sent',
'confirmed'
);
@@ -107,22 +108,16 @@ CREATE TABLE transfer (
exchange_url TEXT NOT NULL,
request_uid BYTEA NOT NULL UNIQUE CHECK (LENGTH(request_uid)=64),
metadata TEXT,
- status debit_status NOT NULL
+ status debit_status NOT NULL,
+ status_msg TEXT
);
COMMENT ON TABLE transfer IS 'Wire Gateway transfers';
-CREATE TYPE bounce_status AS ENUM(
- 'requested',
- 'ignored',
- 'sent',
- 'confirmed'
-);
-COMMENT ON TYPE bounce_status IS 'Status of a bounce';
-
CREATE TABLE bounced (
tx_in_id INT8 NOT NULL UNIQUE REFERENCES tx_in(tx_in_id) ON DELETE SET NULL,
txid BYTEA UNIQUE CHECK (LENGTH(txid)=32),
- reason TEXT,
- status bounce_status NOT NULL
+ reason TEXT NOT NULL,
+ status debit_status NOT NULL,
+ status_msg TEXT
);
COMMENT ON TABLE state IS 'Bounced incoming transactions';
\ No newline at end of file
diff --git a/depolymerizer-bitcoin/db/depolymerizer-bitcoin-procedures.sql b/depolymerizer-bitcoin/db/depolymerizer-bitcoin-procedures.sql
@@ -232,7 +232,7 @@ COMMENT ON FUNCTION register_bounce_tx_in IS 'Register an incoming transaction a
CREATE FUNCTION sync_out(
IN in_txid BYTEA,
- IN in_replace_txid BYTEA,
+ IN in_replaces_txid BYTEA,
IN in_amount taler_amount,
IN in_credit_acc TEXT,
IN in_wtid BYTEA,
@@ -240,6 +240,7 @@ CREATE FUNCTION sync_out(
IN in_metadata TEXT,
IN in_bounced_txid BYTEA,
IN in_created_at INT8,
+ IN in_confirmed BOOLEAN,
IN in_now INT8,
-- Success return
OUT out_tx_row_id INT8,
@@ -250,67 +251,92 @@ CREATE FUNCTION sync_out(
LANGUAGE plpgsql AS $$
DECLARE
local_id INT8;
+ local_status debit_status;
+ local_update BOOLEAN;
BEGIN
-UPDATE tx_out SET txid=in_txid WHERE txid=in_replace_txid;
-out_replaced=FOUND;
-out_new=NOT EXISTS(SELECT FROM tx_out WHERE txid = in_txid);
-IF NOT out_new THEN
- RETURN;
+IF in_confirmed THEN
+ local_status='confirmed';
+ELSE
+ local_status='sent';
END IF;
-
--- Insert new outgoing transaction
-INSERT INTO tx_out (
- amount,
- credit_acc,
- txid,
- created_at
-) VALUES (
- in_amount,
- in_credit_acc,
- in_txid,
- in_created_at
-) RETURNING tx_out_id INTO out_tx_row_id;
--- Notify new outgoing transaction registration
-PERFORM pg_notify('tx_out', out_tx_row_id || '');
-
IF in_wtid IS NOT NULL THEN
- -- Insert new outgoing talerable transaction
- INSERT INTO taler_out (
- tx_out_id,
- wtid,
- exchange_base_url,
- metadata
- ) VALUES (
- out_tx_row_id,
- in_wtid,
- in_exchange_base_url,
- in_metadata
- ) ON CONFLICT (wtid) DO NOTHING;
- IF FOUND THEN
- -- Notify new outgoing talerable transaction registration
- PERFORM pg_notify('taler_out', out_tx_row_id || '');
+ -- Sync transfer status
+ SELECT
+ txid=in_replaces_txid,
+ txid IS NULL,
+ status!=local_status OR txid!=in_txid
+ INTO
+ out_replaced,
+ out_recovered,
+ local_update
+ FROM transfer
+ WHERE wtid=in_wtid;
+ IF local_update THEN
+ UPDATE transfer SET status=local_status,txid=in_txid
+ WHERE wtid=in_wtid;
END IF;
- -- Update transfer state
- UPDATE transfer SET status='confirmed',txid=in_txid WHERE wtid=in_wtid;
- out_recovered=FOUND;
ELSIF in_bounced_txid IS NOT NULL THEN
- -- Update bounce state
- IF NOT EXISTS(SELECT FROM bounced JOIN tx_in USING (tx_in_id) WHERE tx_in.txid=in_bounced_txid) THEN
- INSERT INTO bounced (
- tx_in_id,
- txid,
- status
- ) VALUES (
- (SELECT tx_in_id FROM tx_in WHERE txid=in_bounced_txid),
- in_txid,
- 'confirmed'
- );
- out_recovered=TRUE;
- ELSE
- UPDATE bounced SET status='confirmed',txid=in_txid
+ -- Sync bounce status
+ SELECT
+ bounced.txid=in_replaces_txid,
+ bounced.txid IS NULL,
+ status!=local_status OR bounced.txid!=in_txid
+ INTO
+ out_replaced,
+ out_recovered,
+ local_update
+ FROM bounced JOIN tx_in USING (tx_in_id)
+ WHERE tx_in.txid=in_bounced_txid;
+ IF local_update THEN
+ UPDATE bounced SET status=local_status,txid=in_txid
FROM tx_in
WHERE bounced.tx_in_id=tx_in.tx_in_id AND tx_in.txid=in_bounced_txid;
- out_recovered=FALSE;
+ END IF;
+END IF;
+
+IF in_confirmed THEN
+ -- Sync tx_out status
+ UPDATE tx_out SET txid=in_txid WHERE txid=in_replaces_txid;
+ out_replaced=out_replaced OR FOUND;
+ SELECT tx_out_id INTO out_tx_row_id
+ FROM tx_out WHERE txid=in_txid;
+ IF FOUND THEN
+ RETURN;
+ END IF;
+ out_new = TRUE;
+
+ -- Insert new outgoing transaction
+ INSERT INTO tx_out (
+ amount,
+ credit_acc,
+ txid,
+ created_at
+ ) VALUES (
+ in_amount,
+ in_credit_acc,
+ in_txid,
+ in_created_at
+ ) RETURNING tx_out_id INTO out_tx_row_id;
+ -- Notify new outgoing transaction registration
+ PERFORM pg_notify('tx_out', out_tx_row_id || '');
+
+ IF in_wtid IS NOT NULL THEN
+ -- Insert new outgoing talerable transaction
+ INSERT INTO taler_out (
+ tx_out_id,
+ wtid,
+ exchange_base_url,
+ metadata
+ ) VALUES (
+ out_tx_row_id,
+ in_wtid,
+ in_exchange_base_url,
+ in_metadata
+ ) ON CONFLICT (wtid) DO NOTHING;
+ IF FOUND THEN
+ -- Notify new outgoing talerable transaction registration
+ PERFORM pg_notify('taler_out', out_tx_row_id || '');
+ END IF;
END IF;
END IF;
END $$;
diff --git a/depolymerizer-bitcoin/src/api.rs b/depolymerizer-bitcoin/src/api.rs
@@ -1,6 +1,6 @@
/*
This file is part of TALER
- Copyright (C) 2025, 2026 Taler Systems SA
+ Copyright (C) 2025-2026 Taler Systems SA
TALER is free software; you can redistribute it and/or modify it under the
terms of the GNU Affero General Public License as published by the Free Software
@@ -378,7 +378,7 @@ pub mod test {
use crate::{
CONFIG_SOURCE,
api::ServerState,
- db::{TxOutKind, sync_out, test::rand_tx_id},
+ db::{TxOut, TxOutKind, sync_out, test::rand_tx_id},
payto::FullBtcPayto,
};
@@ -442,16 +442,19 @@ pub mod test {
let sub = &OutgoingSubject::rand();
sync_out(
&db,
- &rand_tx_id(),
- None,
- &amount("BTC:10"),
- &EXCHANGE.0,
+ &TxOut {
+ id: rand_tx_id(),
+ replaces_txid: None,
+ amount: amount("BTC:10"),
+ credit_acc: &EXCHANGE.0,
+ block_time: Timestamp::now(),
+ },
&TxOutKind::Talerable {
wtid: &sub.wtid,
url: &sub.exchange_base_url,
metadata: sub.metadata.as_deref(),
},
- &Timestamp::now(),
+ true,
)
.await
.unwrap();
diff --git a/depolymerizer-bitcoin/src/bin/btc-harness.rs b/depolymerizer-bitcoin/src/bin/btc-harness.rs
@@ -0,0 +1,1319 @@
+/*
+ This file is part of TALER
+ Copyright (C) 2026 Taler Systems SA
+
+ TALER is free software; you can redistribute it and/or modify it under the
+ terms of the GNU Affero General Public License as published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
+
+ You should have received a copy of the GNU Affero General Public License along with
+ TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+*/
+
+use std::{
+ assert_matches,
+ collections::BTreeSet,
+ net::{Ipv4Addr, SocketAddr, SocketAddrV4, TcpListener},
+ process::{Child, Stdio},
+ str::FromStr,
+ time::Duration,
+};
+
+use bitcoin::{Address, Amount, Txid, address::NetworkUnchecked, hashes::Hash};
+use clap::Parser as _;
+use depolymerizer_bitcoin::{
+ CONFIG_SOURCE, RpcApiExtended as _,
+ config::{RpcAuth, RpcCfg, WorkerCfg},
+ db::{self, dbinit, incoming_history, pool},
+ loops::{LoopError, LoopResult, analysis::analysis, worker::worker_step},
+ payto::{BtcWallet, FullBtcPayto},
+ rpc::{Category, Error, ErrorCode, Rpc, RpcApi, WalletRpc},
+ setup::setup,
+ taler_utils::{btc_to_taler, taler_to_btc},
+};
+use failure_injection::set_failure_scenario;
+use owo_colors::OwoColorize as _;
+use sqlx::PgPool;
+use taler_api::notification::dummy_listen;
+use taler_build::long_version;
+use taler_common::{
+ CommonArgs,
+ api::{
+ EddsaPublicKey, HashCode, ShortHashCode,
+ params::{History, Page, Pooling},
+ wire::{IncomingBankTransaction, TransferRequest, TransferState},
+ },
+ config::Config,
+ taler_main,
+ types::{
+ amount::{Currency, amount},
+ url,
+ },
+};
+use taler_test_utils::repl::Repl;
+use tracing::info;
+
+#[derive(Debug)]
+struct PanicErr;
+
+type PanicRes<T> = Result<T, PanicErr>;
+
+impl<E: std::fmt::Display> From<E> for PanicErr {
+ #[track_caller]
+ fn from(value: E) -> Self {
+ panic!("{value}")
+ }
+}
+
+/// Bictoin Adapter harness test suite
+#[derive(clap::Parser, Debug)]
+#[command(long_version = long_version(), about, long_about = None)]
+struct Args {
+ #[clap(flatten)]
+ common: CommonArgs,
+
+ #[command(subcommand)]
+ cmd: Command,
+}
+
+#[derive(Debug, Clone, clap::ValueEnum, PartialEq, Eq)]
+enum Network {
+ Local,
+ Test,
+}
+
+#[derive(clap::Subcommand, Debug)]
+enum Command {
+ /// Run logic tests
+ Logic {
+ #[arg(short, long)]
+ reset: bool,
+ },
+ /// Run online tests
+ Online {
+ #[arg(short, long)]
+ reset: bool,
+ network: Network,
+ },
+}
+
+fn step(step: &str) {
+ println!("{}", step.green());
+}
+
+#[derive(Debug)]
+struct SystemCtx {
+ n1_dir: String,
+ n2_dir: String,
+ c_dir: String,
+ peer_addr: SocketAddr,
+ n1_addr: SocketAddr,
+ n2_addr: SocketAddr,
+ n1_cookie: String,
+ n2_cookie: String,
+}
+
+impl SystemCtx {
+ fn new(reset: bool) -> Self {
+ // Prepare dir
+ let dir = "depolymerizer-bitcoin/test/logic";
+ std::fs::create_dir_all(dir).unwrap();
+ if reset {
+ std::fs::remove_dir_all(dir).unwrap();
+ }
+ let n1_dir = format!("{dir}/n1");
+ let n2_dir = format!("{dir}/n2");
+ let c_dir = format!("{dir}/c");
+ for dir in [&n1_dir, &n2_dir] {
+ std::fs::create_dir_all(dir).unwrap();
+ }
+
+ // Choose unused port
+ let btc_port = unused_port();
+ let btc_rpc_port = unused_port();
+ let btc2_port = unused_port();
+ let btc2_rpc_port = unused_port();
+
+ // Bitcoin config
+ for (root, btc, rpc) in [
+ (&n1_dir, btc_port, btc_rpc_port),
+ (&n2_dir, btc2_port, btc2_rpc_port),
+ ] {
+ std::fs::write(
+ format!("{root}/bitcoin.conf"),
+ format!(
+ "
+ regtest=1
+ txindex=1
+ maxtxfee=0.01
+ fallbackfee=0.00000001
+ rpcservertimeout=0
+ dbcache=4
+ maxmempool=5
+ par=2
+ rpcthreads=5
+
+ [regtest]
+ bind=127.0.0.1:{btc}
+ rpcport={rpc}
+ "
+ ),
+ )
+ .unwrap();
+ }
+
+ Self {
+ n1_addr: SocketAddrV4::new(Ipv4Addr::LOCALHOST, btc_rpc_port).into(),
+ n1_cookie: format!("{n1_dir}/regtest/.cookie"),
+ n2_addr: SocketAddrV4::new(Ipv4Addr::LOCALHOST, btc2_rpc_port).into(),
+ n2_cookie: format!("{n2_dir}/regtest/.cookie"),
+ peer_addr: SocketAddrV4::new(Ipv4Addr::LOCALHOST, btc2_port).into(),
+ n1_dir,
+ n2_dir,
+ c_dir,
+ }
+ }
+
+ async fn test_cfg(&self, btc_patch: &str, cfg_patch: &str) {
+ let port = unused_port();
+ let Self { c_dir, .. } = self;
+ std::fs::remove_dir_all(c_dir).ok();
+ std::fs::create_dir_all(c_dir).unwrap();
+ std::fs::write(
+ format!("{c_dir}/bitcoin.conf"),
+ format!(
+ "
+ regtest=1
+ txindex=1
+ maxtxfee=0.01
+ fallbackfee=0.00000001
+ rpcservertimeout=0
+ dbcache=4
+ maxmempool=5
+ par=2
+ rpcthreads=5
+
+ [regtest]
+ rpcport={port}
+
+ {btc_patch}
+ "
+ ),
+ )
+ .unwrap();
+ let cfg = Config::from_mem_with_env(
+ CONFIG_SOURCE,
+ &format!(
+ "
+ [depolymerizer-bitcoin-worker]
+ RPC_BIND=127.0.0.1:{port}
+
+ {cfg_patch}
+ "
+ ),
+ )
+ .unwrap();
+ let mut n = cmd("bitcoind", &[format!("-datadir={}", self.c_dir).as_str()]);
+ let cfg = RpcCfg::parse(&cfg).unwrap();
+ wait_up(&cfg).await;
+ n.0.kill().unwrap();
+ n.0.wait().unwrap();
+ }
+
+ async fn start_n1(&self, args: &[&str]) -> (ChildGuard, Rpc) {
+ let Self {
+ n1_dir,
+ n1_addr,
+ n1_cookie,
+ ..
+ } = self;
+ let n = cmd(
+ "bitcoind",
+ Vec::from_iter(
+ [format!("-datadir={n1_dir}").as_str()]
+ .iter()
+ .chain(args)
+ .copied(),
+ )
+ .as_slice(),
+ );
+ let mut rpc = wait_up(&RpcCfg {
+ addr: *n1_addr,
+ auth: RpcAuth::Cookie(n1_cookie.clone()),
+ })
+ .await;
+ for name in ["wire", "client", "reserve"] {
+ if let Err(e) = rpc.load_wallet(name).await {
+ if let Error::RPC {
+ code: ErrorCode::RpcWalletNotFound,
+ ..
+ } = e
+ {
+ rpc.create_wallet(name, "").await.unwrap();
+ } else {
+ break;
+ }
+ }
+ }
+ (n, rpc)
+ }
+
+ async fn start_n2(&self, rpc1: &mut impl RpcApi) -> (ChildGuard, Rpc) {
+ let n = cmd("bitcoind", &[&format!("-datadir={}", self.n2_dir)]);
+ let rpc = wait_up(&RpcCfg {
+ addr: self.n2_addr,
+ auth: RpcAuth::Cookie(self.n2_cookie.clone()),
+ })
+ .await;
+ rpc1.add_node(&self.peer_addr.to_string()).await.unwrap();
+ (n, rpc)
+ }
+}
+
+struct Ctx<'a> {
+ sc: &'a SystemCtx,
+ n1: &'a mut ChildGuard,
+ pool: &'a PgPool,
+ rpc: Rpc,
+ rpc2: Rpc,
+ wire_addr: &'a Address,
+ client_addr: &'a Address,
+ reserve_addr: &'a Address,
+ s: WorkerCfg,
+ status: bool,
+}
+
+impl<'a> Ctx<'a> {
+ /// Run the worker once
+ async fn run_worker(&mut self) -> LoopResult<()> {
+ let db = &mut self.pool.acquire().await.unwrap().detach();
+ let res = worker_step(
+ db,
+ &mut self.rpc.wallet("wire"),
+ &mut self.s,
+ &mut self.status,
+ )
+ .await;
+ if let Err(e) = &res {
+ tracing::error!(target: "worker", "{e}");
+ }
+ res
+ }
+
+ /// Run the worker once successfuly
+ async fn worker(&mut self) {
+ self.run_worker().await.unwrap();
+ }
+
+ /// Run the worker once and fail on an injected error
+ async fn w_injected_err(&mut self) {
+ assert_matches!(self.run_worker().await, Err(LoopError::Injected(_)));
+ }
+
+ /// Run the worker once and fail on a rpc error
+ async fn w_rpc_err(&mut self) {
+ assert_matches!(
+ self.run_worker().await,
+ Err(LoopError::Rpc(Error::RPC { .. }))
+ );
+ }
+
+ async fn restart_node(&mut self, args: &[&str]) {
+ // Stop node cleanly
+ self.rpc.stop().await.unwrap();
+ self.n1.0.wait().unwrap();
+ // Start a new node
+ let (n, rpc) = self.sc.start_n1(args).await;
+ *self.n1 = n;
+ self.rpc = rpc;
+ self.rpc
+ .add_node(&self.sc.peer_addr.to_string())
+ .await
+ .unwrap();
+ }
+
+ async fn cluster_deco(&mut self) {
+ self.rpc
+ .disconnect_node(&self.sc.peer_addr.to_string())
+ .await
+ .unwrap();
+ }
+
+ async fn cluster_fork(&mut self) -> u32 {
+ let n1h = self.rpc.get_blockchain_info().await.unwrap().blocks;
+ let n2h = self.rpc2.get_blockchain_info().await.unwrap().blocks;
+ let diff = n1h - n2h;
+ self.rpc2.mine(diff + 1, self.reserve_addr).await.unwrap();
+ self.rpc
+ .add_node(&self.sc.peer_addr.to_string())
+ .await
+ .unwrap();
+ self.rpc.wait_for_block_height(n1h + 1).await.unwrap();
+ diff + 2
+ }
+
+ async fn forget_tx(&self, tx: &Txid) {
+ sqlx::query("UPDATE tx_in SET txid=NULL WHERE txid=$1")
+ .bind(tx.as_byte_array())
+ .execute(self.pool)
+ .await
+ .unwrap();
+ }
+
+ /* ----- Rpc ----- */
+
+ fn client(&mut self) -> WalletRpc<'_> {
+ self.rpc.wallet("client")
+ }
+
+ fn wire(&mut self) -> WalletRpc<'_> {
+ self.rpc.wallet("wire")
+ }
+
+ fn reserve(&mut self) -> WalletRpc<'_> {
+ self.rpc.wallet("reserve")
+ }
+
+ /* ----- Balance ---- */
+
+ async fn c_balance(&mut self) -> Amount {
+ self.client().get_balance().await.unwrap()
+ }
+
+ async fn w_balance(&mut self) -> Amount {
+ self.wire().get_balance().await.unwrap()
+ }
+
+ async fn expect_c_balance(&mut self, amount: Amount) {
+ assert_eq!(self.c_balance().await, amount)
+ }
+
+ async fn expect_w_balance(&mut self, amount: Amount) {
+ assert_eq!(self.w_balance().await, amount)
+ }
+
+ /* ----- Mining ----- */
+
+ /// Mine nb block
+ async fn mine(&mut self, nb: u32) {
+ self.rpc.mine(nb, self.reserve_addr).await.unwrap();
+ }
+
+ /// Mine enough block to reach confirmation
+ async fn next_conf(&mut self) {
+ self.mine(self.s.conf).await
+ }
+
+ /// Mine one block
+ async fn next_block(&mut self) {
+ self.mine(1).await
+ }
+
+ /// Mine while txs are pending
+ async fn mine_pending(&mut self) {
+ while self.rpc.get_mempool_info().await.unwrap().size > 0 {
+ self.mine(1).await
+ }
+ }
+
+ /// Mine while txs are pending or unconfirmed
+ async fn mine_conf(&mut self) {
+ self.mine_pending().await;
+ self.next_conf().await;
+ }
+
+ /* ----- Transaction ----- */
+
+ async fn withdrawal(&mut self, amount: Amount) -> (Txid, EddsaPublicKey) {
+ let key = EddsaPublicKey::rand();
+ let id = self.withdrawal_with_key(amount, &key).await;
+ (id, key)
+ }
+
+ async fn withdrawal_with_key(&mut self, amount: Amount, key: &EddsaPublicKey) -> Txid {
+ loop {
+ match self
+ .rpc
+ .wallet("client")
+ .send_segwit_key(self.wire_addr, amount, &key)
+ .await
+ {
+ Ok(id) => return id,
+ Err(e) => match e {
+ Error::RPC {
+ code: ErrorCode::RpcWalletError,
+ ..
+ } => {
+ self.mine(1).await;
+ }
+ _ => panic!("{e:?}"),
+ },
+ }
+ }
+ }
+
+ async fn transfer_to(&mut self, addr: &Address, amount: Amount) -> u64 {
+ let payto = FullBtcPayto::new(BtcWallet(addr.clone()), "Bitcoin Client");
+ match db::transfer(
+ self.pool,
+ &payto,
+ &TransferRequest {
+ request_uid: HashCode::rand(),
+ amount: btc_to_taler(&amount.to_signed().unwrap(), &Currency::KUDOS),
+ exchange_base_url: url("https://test.com"),
+ metadata: None,
+ wtid: ShortHashCode::rand(),
+ credit_account: payto.as_uri(),
+ },
+ )
+ .await
+ .unwrap()
+ {
+ db::TransferResult::Success(it) => it.row_id,
+ it => panic!("{it:?}"),
+ }
+ }
+
+ async fn transfer(&mut self, amount: Amount) -> u64 {
+ self.transfer_to(self.client_addr, amount).await
+ }
+
+ async fn c_send(&mut self, addr: &Address, amount: Amount) -> Txid {
+ self.rpc
+ .wallet("client")
+ .send(addr, amount, None, false)
+ .await
+ .unwrap()
+ }
+
+ async fn malformed_credit(&mut self, amount: Amount) -> Txid {
+ self.c_send(self.wire_addr, amount).await
+ }
+
+ async fn abandon(mut rpc: impl RpcApi) {
+ let list = rpc.list_since_block(None, 1).await.unwrap();
+ for tx in list.transactions {
+ if tx.category == Category::Send && tx.confirmations == 0 {
+ rpc.abandon_tx(&tx.txid).await.unwrap();
+ }
+ }
+ }
+
+ async fn abandon_wire(&mut self) {
+ Self::abandon(self.wire()).await;
+ }
+
+ async fn abandon_client(&mut self) {
+ Self::abandon(self.client()).await;
+ }
+
+ /* ----- Checks ----- */
+
+ async fn expect_incoming(&self, key: EddsaPublicKey) {
+ let transfer = incoming_history(
+ self.pool,
+ &History {
+ page: Page {
+ limit: -1,
+ offset: None,
+ },
+ pooling: Pooling { timeout_ms: None },
+ },
+ &Currency::TEST,
+ dummy_listen,
+ )
+ .await
+ .unwrap();
+ assert_matches!(
+ transfer.first().unwrap(),
+ IncomingBankTransaction::Reserve { reserve_pub, .. } if *reserve_pub == key,
+ );
+ }
+
+ async fn expect_transfer_status(&self, id: u64, status: TransferState) {
+ let mut attempts = 0;
+ loop {
+ let transfer = db::transfer_by_id(self.pool, id, &Currency::KUDOS)
+ .await
+ .unwrap()
+ .unwrap();
+ if (transfer.status, transfer.status_msg.as_deref()) == (status, None) {
+ return;
+ }
+ if attempts > 40 {
+ assert_eq!(
+ (transfer.status, transfer.status_msg.as_deref()),
+ (status, None)
+ );
+ }
+ attempts += 1;
+ tokio::time::sleep(Duration::from_millis(200)).await;
+ }
+ }
+}
+
+fn unused_port() -> u16 {
+ TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0))
+ .unwrap()
+ .local_addr()
+ .unwrap()
+ .port()
+}
+
+async fn wait_up(cfg: &RpcCfg) -> Rpc {
+ for _ in 0..100 {
+ if let Ok(mut rpc) = Rpc::new(cfg).await
+ && rpc.get_blockchain_info().await.is_ok()
+ {
+ return rpc;
+ }
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ }
+ let mut rpc = Rpc::new(cfg).await.unwrap();
+ rpc.get_blockchain_info().await.unwrap();
+ rpc
+}
+
+struct ChildGuard(pub Child);
+
+impl Drop for ChildGuard {
+ fn drop(&mut self) {
+ self.0.kill().ok();
+ }
+}
+
+#[track_caller]
+fn cmd(cmd: &str, args: &[&str]) -> ChildGuard {
+ let child = std::process::Command::new(cmd)
+ .args(args)
+ .stderr(Stdio::null())
+ .stdout(Stdio::null())
+ .stdin(Stdio::null())
+ .spawn()
+ .unwrap();
+ ChildGuard(child)
+}
+
+/// Run logic tests against local regtest bitcoin chain
+async fn logic_harness(reset: bool) -> PanicRes<()> {
+ step("Run Bitcoin logic harness tests");
+
+ step("Start bitcoin network");
+ let sc = SystemCtx::new(reset);
+
+ // Start bitcoin nodes
+ let (mut n1, mut rpc) = sc.start_n1(&[]).await;
+ let (mut n2, rpc2) = sc.start_n2(&mut rpc).await;
+ let reserve_addr = &rpc.wallet("reserve").gen_addr().await?;
+ let client_addr = &rpc.wallet("client").gen_addr().await?;
+ let wire_addr = &rpc.wallet("wire").gen_addr().await?;
+
+ let cfg = Config::from_mem_with_env(
+ CONFIG_SOURCE,
+ &format!(
+ "
+ [depolymerizer-bitcoin]
+ CURRENCY = BTC
+ NAME = Exchange Owner
+ WALLET = {wire_addr}
+
+ [depolymerizer-bitcoin-worker]
+ BOUNCE_FEE = BTC:0.00001
+ CONFIRMATION = 3
+ RPC_BIND = {}
+ WALLET_NAME = wire
+ RPC_COOKIE_FILE = {}
+ ",
+ sc.n1_addr, sc.n1_cookie
+ ),
+ )?;
+
+ step("Setup");
+ dbinit(&cfg, reset).await?;
+ setup(&cfg, reset).await?;
+ let state = WorkerCfg::parse(&cfg)?;
+ let pool = &pool(&cfg).await?;
+
+ let c = &mut Ctx {
+ sc: &sc,
+ n1: &mut n1,
+ pool,
+ rpc,
+ rpc2,
+ client_addr,
+ wire_addr,
+ reserve_addr,
+ s: state,
+ status: true,
+ };
+
+ if c.client().get_balance().await? < Amount::ONE_BTC {
+ while c.reserve().get_balance().await? < Amount::ONE_BTC {
+ c.mine(1).await;
+ }
+ c.reserve()
+ .send(client_addr, Amount::ONE_BTC, None, false)
+ .await
+ .unwrap();
+ }
+
+ let fee = c.s.bounce_fee;
+
+ step("Warmup");
+ c.worker().await;
+ c.mine_conf().await;
+ c.worker().await;
+
+ step("Withdrawal");
+ let amount = Amount::from_sat(330000);
+ let b = c.wire().get_balance().await?;
+ // Send tx
+ let (_, reserve_pub) = c.withdrawal(amount).await;
+ // Confirm tx
+ c.mine_conf().await;
+ // Check received
+ c.expect_w_balance(b + amount).await;
+ // Work register it
+ c.worker().await;
+ // Check registered
+ c.expect_incoming(reserve_pub).await;
+
+ step("Deposit");
+ let amount = Amount::from_sat(34000);
+ let b = c.c_balance().await;
+ // Request tx
+ let id = c.transfer(amount).await;
+ // Check start pending
+ c.expect_transfer_status(id, TransferState::pending).await;
+ // Send it
+ c.worker().await;
+ // Mine it
+ c.mine_pending().await;
+ // Client already received the fund
+ c.expect_c_balance(b + amount).await;
+ // Still pending until confirmed
+ c.worker().await;
+ c.expect_transfer_status(id, TransferState::pending).await;
+ // Confirm it
+ c.mine_conf().await;
+ c.worker().await;
+ // Now success
+ c.expect_transfer_status(id, TransferState::success).await;
+ c.expect_c_balance(b + amount).await;
+
+ step("Bounce");
+ let b = c.w_balance().await;
+ let amount = Amount::from_sat(10000);
+ // Send malformed
+ c.malformed_credit(amount).await;
+ // Ignore until confirmed
+ c.mine_conf().await;
+ c.expect_w_balance(b + amount).await;
+ // Confirm it & bounce
+ c.mine_conf().await;
+ c.worker().await;
+ // Keep fee
+ c.expect_w_balance(b + fee).await;
+ // Confirm the bounce
+ c.mine_conf().await;
+ c.worker().await;
+
+ // TODO lifetime ?
+ // TODO reconnect rpc and database?
+
+ step("Transfer to unknown account");
+ c.transfer_to(
+ &Address::<NetworkUnchecked>::from_str("bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh")
+ .unwrap()
+ .assume_checked(),
+ Amount::from_sat(34100),
+ )
+ .await;
+ c.worker().await;
+
+ step("Debit failure");
+ set_failure_scenario(&["debit"]);
+ let amount = Amount::from_sat(34000);
+ let b = c.c_balance().await;
+ let id = c.transfer(amount).await;
+ // Check idempotent
+ c.w_injected_err().await;
+ c.worker().await;
+ // Confirm it
+ c.mine_conf().await;
+ c.worker().await;
+ c.expect_transfer_status(id, TransferState::success).await;
+ // Check only sent once
+ c.expect_c_balance(b + amount).await;
+
+ step("Bounce failure");
+ set_failure_scenario(&["bounce"]);
+ let b = c.w_balance().await;
+ let amount = Amount::from_sat(20000);
+ // Send malformed
+ c.malformed_credit(amount).await;
+ c.mine_pending().await;
+ // Ignore until confirmed
+ c.mine_conf().await;
+ c.expect_w_balance(b + amount).await;
+ // Check idempotent
+ c.w_injected_err().await;
+ c.worker().await;
+ // Kept fee
+ c.expect_w_balance(b + fee).await;
+ c.mine_conf().await;
+ c.worker().await;
+ c.expect_w_balance(b + fee).await;
+
+ step("Conflict send");
+ let amount = Amount::from_sat(35000);
+ let b = c.c_balance().await;
+ // Perform deposit
+ c.transfer(amount).await;
+ c.worker().await;
+ // Abandon pending transaction
+ c.restart_node(&["-minrelaytxfee=0.0001"]).await;
+ c.abandon_wire().await;
+ c.expect_c_balance(b).await;
+ // Generate conflict
+ let conflict = Amount::from_sat(40000);
+ c.transfer(conflict).await;
+ c.worker().await;
+ c.mine_pending().await;
+ c.expect_c_balance(b + conflict).await;
+ // Handle conflict
+ c.worker().await;
+ c.mine_pending().await;
+ c.expect_c_balance(b + amount + conflict).await;
+ c.mine_conf().await;
+ c.worker().await;
+
+ step("Conflict bounce");
+ let b = c.w_balance().await;
+ let amount = Amount::from_sat(20000);
+ // Perform bounce
+ c.malformed_credit(amount).await;
+ c.mine_pending().await;
+ c.expect_w_balance(b + amount).await;
+ c.mine_conf().await;
+ c.worker().await;
+ c.expect_w_balance(b + fee).await;
+ // Abandon pending transaction
+ c.restart_node(&["-minrelaytxfee=0.0002"]).await;
+ c.abandon_wire().await;
+ c.expect_w_balance(b + amount).await;
+ // Generate conflict
+ c.malformed_credit(amount).await;
+ c.mine_conf().await;
+ c.worker().await;
+ c.expect_w_balance(b + amount + fee).await;
+ // Handle conflict
+ c.mine_pending().await;
+ c.worker().await;
+ c.expect_w_balance(b + fee * 2).await;
+ c.mine_conf().await;
+ c.worker().await;
+
+ step("Reorg withdrawal");
+ c.cluster_deco().await;
+ let before = c.w_balance().await;
+ c.withdrawal(Amount::from_sat(100000)).await;
+ c.mine_conf().await;
+ let after = c.w_balance().await;
+ c.worker().await;
+ assert!(c.status);
+ let fork = c.cluster_fork().await;
+ c.worker().await;
+ assert!(!c.status);
+ c.expect_w_balance(before).await;
+ c.mine(fork).await;
+ c.worker().await;
+ assert!(c.status);
+ c.expect_w_balance(after).await;
+
+ step("Reorg deposit");
+ c.cluster_deco().await;
+ let before = c.c_balance().await;
+ c.transfer(Amount::from_sat(1111)).await;
+ c.worker().await;
+ c.mine_conf().await;
+ c.worker().await;
+ let after = c.c_balance().await;
+ assert!(c.status);
+ let fork = c.cluster_fork().await;
+ c.worker().await;
+ assert!(c.status);
+ c.expect_c_balance(before).await;
+ c.mine(fork).await;
+ c.worker().await;
+ assert!(c.status);
+ c.expect_c_balance(after).await;
+
+ step("Reorg bounce");
+ c.cluster_deco().await;
+ c.malformed_credit(Amount::from_sat(100000)).await;
+ c.mine_conf().await;
+ c.worker().await;
+ c.mine_pending().await;
+ assert!(c.status);
+ let fork = c.cluster_fork().await;
+ c.worker().await;
+ assert!(!c.status);
+ c.mine(fork).await;
+ c.worker().await;
+ assert!(c.status);
+
+ c.s.bump_delay = Some(1);
+
+ step("Bump fee");
+ let amount = Amount::from_sat(40000);
+ let b = c.c_balance().await;
+ c.transfer(amount).await;
+ c.worker().await;
+ let until = tokio::time::Instant::now() + Duration::from_secs(1);
+ c.worker().await;
+ c.restart_node(&["-minrelaytxfee=0.0003"]).await;
+ tokio::time::sleep_until(until).await;
+ c.worker().await;
+ c.expect_c_balance(b).await;
+ c.mine_conf().await;
+ c.expect_c_balance(b + amount).await;
+ c.worker().await;
+
+ step("Bump fee failure");
+ let amount = Amount::from_sat(41000);
+ let b = c.c_balance().await;
+ c.transfer(amount).await;
+ c.worker().await;
+ let until = tokio::time::Instant::now() + Duration::from_secs(1);
+ c.worker().await;
+ c.restart_node(&["-minrelaytxfee=0.0004"]).await;
+ tokio::time::sleep_until(until).await;
+ set_failure_scenario(&["bumpfee"]);
+ c.w_injected_err().await;
+ c.worker().await;
+ c.expect_c_balance(b).await;
+ c.mine_conf().await;
+ c.expect_c_balance(b + amount).await;
+ c.worker().await;
+
+ step("Bump fee reorg");
+ c.cluster_deco().await;
+ let amount = Amount::from_sat(42000);
+ let b = c.c_balance().await;
+ c.transfer(amount).await;
+ c.worker().await;
+ let until = tokio::time::Instant::now() + Duration::from_secs(1);
+ c.worker().await;
+ c.cluster_fork().await;
+ c.restart_node(&["-minrelaytxfee=0.0005"]).await;
+ c.next_block().await;
+ tokio::time::sleep_until(until).await;
+ c.worker().await;
+ c.expect_c_balance(b).await;
+ c.mine_conf().await;
+ c.expect_c_balance(b + amount).await;
+ c.worker().await;
+
+ c.s.bump_delay = None;
+
+ step("Reorg conflict withdrawal");
+ let b = c.w_balance().await;
+ c.cluster_deco().await;
+
+ // Withdrawal
+ let amount = Amount::from_sat(420000);
+ let (id, _) = c.withdrawal(amount).await;
+ c.mine_conf().await;
+ c.worker().await;
+ c.expect_w_balance(b + amount).await;
+
+ // Perform fork
+ assert!(c.status);
+ c.cluster_fork().await;
+ c.worker().await;
+ assert!(!c.status);
+
+ // Generate conflict
+ c.restart_node(&["-minrelaytxfee=0.0006"]).await;
+ c.abandon_client().await;
+ let conflict = Amount::from_sat(54000);
+ c.withdrawal(conflict).await;
+ c.mine_pending().await;
+ c.expect_w_balance(b + conflict).await;
+
+ // Check cannot recover again adversarial attack
+ c.mine_conf().await;
+ c.worker().await;
+ assert!(!c.status);
+
+ // Manual drop from db
+ c.forget_tx(&id).await;
+ c.worker().await;
+ assert!(c.status);
+ c.expect_w_balance(b + conflict).await;
+
+ step("Reorg conflict bounce");
+ let b = c.w_balance().await;
+ c.cluster_deco().await;
+
+ // Withdrawal
+ let amount = Amount::from_sat(420000);
+ let id = c.malformed_credit(amount).await;
+ c.mine_conf().await;
+ c.worker().await;
+ c.expect_w_balance(b + fee).await;
+
+ // Perform fork
+ assert!(c.status);
+ c.cluster_fork().await;
+ c.worker().await;
+ assert!(!c.status);
+
+ // Generate conflict
+ c.restart_node(&["-minrelaytxfee=0.0007"]).await;
+ c.abandon_client().await;
+ let conflict = Amount::from_sat(54000);
+ c.withdrawal(conflict).await;
+ c.mine_pending().await;
+ c.expect_w_balance(b + conflict).await;
+
+ // Check cannot recover again adversarial attack
+ c.mine_conf().await;
+ c.worker().await;
+ assert!(!c.status);
+
+ // Manual drop from db
+ c.forget_tx(&id).await;
+ c.worker().await;
+ assert!(c.status);
+ c.expect_w_balance(b + conflict).await;
+
+ step("Analysis");
+ assert_eq!(c.s.conf, 3);
+ c.s.conf = analysis(&mut c.rpc, c.s.conf, c.s.max_conf).await?;
+ assert_eq!(c.s.conf, 6);
+
+ step("Maxfee");
+ c.restart_node(&["-maxtxfee=0.0000001", "-minrelaytxfee=0.0000001"])
+ .await;
+ let amount = Amount::from_sat(34500);
+ let b = c.c_balance().await;
+ c.transfer(amount).await;
+ c.w_rpc_err().await;
+ c.mine_pending().await;
+ c.w_rpc_err().await;
+ c.expect_c_balance(b).await;
+ c.restart_node(&[]).await;
+ c.worker().await;
+ c.mine_conf().await;
+ c.expect_c_balance(b + amount).await;
+ c.worker().await;
+
+ c.abandon_client().await;
+ c.abandon_wire().await;
+
+ step("Stop bitcoin network");
+ let cb = c.c_balance().await;
+ let wb = c.w_balance().await;
+ c.worker().await;
+ c.mine_conf().await;
+ c.worker().await;
+ c.expect_c_balance(cb).await;
+ c.expect_w_balance(wb).await;
+
+ c.rpc.stop().await?;
+ c.rpc2.stop().await?;
+
+ n1.0.wait()?;
+ n2.0.wait()?;
+
+ step("Config");
+ // Connect with custom cookie files
+ sc.test_cfg(
+ "
+ rpccookiefile=catch_me_if_you_can
+ ",
+ &format!(
+ "
+ [depolymerizer-bitcoin-worker]
+ RPC_COOKIE_FILE={}/regtest/catch_me_if_you_can
+ ",
+ sc.c_dir
+ ),
+ )
+ .await;
+ // Connect with password
+ sc.test_cfg(
+ "
+ rpcuser=bob
+ rpcpassword=password
+ ",
+ "
+ [depolymerizer-bitcoin-worker]
+ RPC_AUTH_METHOD=basic
+ RPC_USERNAME=bob
+ RPC_PASSWORD=password
+ ",
+ )
+ .await;
+ // Connect with token
+ sc.test_cfg(
+ "
+ rpcauth=bob:9641cec731e1fad1ded02e1d31536e44$36b8b8af0a38104997a57f017805ff56bf8963ae4a2ed40252ca0e0e070fc19e
+ ",
+ "
+ [depolymerizer-bitcoin-worker]
+ RPC_AUTH_METHOD=basic
+ RPC_USERNAME=bob
+ RPC_PASSWORD=password
+ ",
+ )
+ .await;
+ Ok(())
+}
+
+#[derive(clap::Parser, Debug)]
+#[command(no_binary_name = true)]
+struct Shell {
+ #[clap(subcommand)]
+ cmd: Option<Cmd>,
+}
+#[derive(Clone, clap::Subcommand, Debug)]
+enum Cmd {
+ Setup,
+ Reset,
+ ResetDb,
+ Sync,
+ Credit,
+ Debit,
+ Mine {
+ amount: Option<u32>,
+ addr: Option<Address<NetworkUnchecked>>,
+ },
+ Exit,
+ Tx {
+ txid: Txid,
+ },
+ Track {
+ txid: Txid,
+ },
+ Untrack {
+ txid: Txid,
+ },
+}
+
+async fn online_harness(network: Network, reset: bool) -> PanicRes<()> {
+ // Prepare dir
+ let network_dir = match network {
+ Network::Local => "btc-local",
+ Network::Test => "btc-test",
+ };
+ let dir = format!("depolymerizer-bitcoin/test/{network_dir}");
+ std::fs::create_dir_all(&dir)?;
+ if reset {
+ std::fs::remove_dir_all(&dir)?;
+ }
+
+ // Bitcoin config
+ let cfg = match network {
+ Network::Local => {
+ "
+ chain=regtest
+ txindex=1
+ rpcservertimeout=0
+ fallbackfee=0.00000001
+ [regtest]
+ rpcport=18345
+ "
+ }
+ Network::Test => {
+ "
+ chain=signet
+ txindex=1
+ rpcservertimeout=0
+ fallbackfee=0.00000001
+ [signet]
+ rpcport=18345
+ "
+ }
+ };
+ std::fs::create_dir_all(&dir)?;
+ std::fs::write(format!("{dir}/bitcoin.conf"), cfg)?;
+
+ let mut n = cmd("bitcoind", &[format!("-datadir={dir}").as_str()]);
+ let addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 18345).into();
+
+ let datadir = match network {
+ Network::Local => format!("{dir}/regtest"),
+ Network::Test => format!("{dir}/signet"),
+ };
+ let cookie = format!("{datadir}/.cookie");
+ let mut rpc = wait_up(&RpcCfg {
+ addr,
+ auth: RpcAuth::Cookie(cookie.clone()),
+ })
+ .await;
+
+ for name in ["wire", "client"] {
+ if let Err(e) = rpc.load_wallet(name).await {
+ if let Error::RPC {
+ code: ErrorCode::RpcWalletNotFound,
+ ..
+ } = e
+ {
+ rpc.create_wallet(name, "").await?;
+ } else {
+ break;
+ }
+ }
+ }
+
+ let wire_addr = &rpc.wallet("wire").gen_addr().await?;
+ let client_addr = &rpc.wallet("client").gen_addr().await?;
+
+ let cfg = Config::from_mem_with_env(
+ CONFIG_SOURCE,
+ &format!(
+ "
+ [depolymerizer-bitcoin]
+ CURRENCY = BTC
+ NAME = Exchange Owner
+ WALLET = {wire_addr}
+
+ [depolymerizer-bitcoin-worker]
+ BOUNCE_FEE = BTC:0.00001
+ CONFIRMATION = 1
+ RPC_BIND = {addr}
+ WALLET_NAME = wire
+ RPC_COOKIE_FILE = {cookie}
+ "
+ ),
+ )
+ .unwrap();
+
+ step("Setup");
+ let mut pool = dbinit(&cfg, reset).await?;
+ setup(&cfg, reset).await?;
+ let mut s = WorkerCfg::parse(&cfg)?;
+ let mut status = true;
+
+ let mut repl: Repl<Shell> = Repl::new(".btc_history");
+ let mut tracked = BTreeSet::new();
+ loop {
+ let info = rpc.get_blockchain_info().await.unwrap();
+ let wire_balance = rpc.wallet("wire").get_balance().await.unwrap();
+ println!("wire {wire_addr} {wire_balance}");
+ let client_balance = rpc.wallet("client").get_balance().await.unwrap();
+ println!("client {client_addr} {client_balance}");
+ for txid in &tracked {
+ match rpc.wallet("client").get_tx(txid).await {
+ Ok(info) => println!(
+ "{} {txid} {} {}",
+ "tx".cyan(),
+ info.amount,
+ info.confirmations
+ ),
+ Err(e) => println!("{} {txid} {}", "tx".cyan(), e.red()),
+ }
+ }
+ if let Some(shell) =
+ repl.read_line(&info.chain, &format!("{:.6}", info.verification_progress))
+ {
+ let Some(cmd) = shell.cmd else { continue };
+ match cmd {
+ Cmd::Setup => setup(&cfg, false).await?,
+ Cmd::Reset => {
+ pool = dbinit(&cfg, true).await?;
+ }
+ Cmd::ResetDb => {
+ pool = dbinit(&cfg, true).await?;
+ setup(&cfg, false).await?;
+ }
+ Cmd::Sync => {
+ let db = &mut pool.acquire().await.unwrap().detach();
+ let res = worker_step(db, &mut rpc.wallet("wire"), &mut s, &mut status).await;
+ if let Err(e) = &res {
+ tracing::error!(target: "worker", "{e}");
+ }
+ }
+ Cmd::Credit => {
+ let reserve_pub = EddsaPublicKey::rand();
+ let amount = amount("DEVBTC:0.00012");
+ match rpc
+ .wallet("client")
+ .send_segwit_key(&wire_addr, taler_to_btc(&amount), &reserve_pub)
+ .await
+ {
+ Ok(txid) => {
+ tracked.insert(txid);
+ info!(target: "testbench", "Credit {reserve_pub} {amount} {txid} to {wire_addr}");
+ }
+ Err(e) => tracing::error!(target: "worker", "{e}"),
+ }
+ }
+ Cmd::Debit => {
+ let payto = FullBtcPayto::new(BtcWallet(client_addr.clone()), "Bitcoin Client");
+ let wtid = ShortHashCode::rand();
+ let amount = amount("DEVBTC:0.00011");
+ match db::transfer(
+ &pool,
+ &payto,
+ &TransferRequest {
+ request_uid: HashCode::rand(),
+ amount,
+ exchange_base_url: url("https://test.com"),
+ metadata: None,
+ wtid,
+ credit_account: payto.as_uri(),
+ },
+ )
+ .await
+ .unwrap()
+ {
+ db::TransferResult::Success(_) => {}
+ it => panic!("{it:?}"),
+ };
+ info!(target: "testbench", "Debit {wtid} {amount} to {wire_addr}");
+ }
+ Cmd::Mine { amount, addr } => {
+ let amount = amount.unwrap_or(1);
+ let addr = addr.map(|a| a.assume_checked());
+ let addr = addr.as_ref().unwrap_or(&client_addr);
+ rpc.wallet("client").mine(amount, addr).await.unwrap();
+ }
+ Cmd::Exit => break,
+ Cmd::Tx { txid } => {
+ let info = rpc.wallet("client").get_tx(&txid).await.unwrap();
+ info!(target: "testbench", "{txid} {} {}", info.amount, info.confirmations);
+ }
+ Cmd::Track { txid } => {
+ tracked.insert(txid);
+ }
+ Cmd::Untrack { txid } => {
+ tracked.retain(|id| *id != txid);
+ }
+ }
+ } else {
+ break;
+ }
+ }
+
+ step("Finish");
+ rpc.stop().await?;
+ n.0.wait()?;
+
+ Ok(())
+}
+
+fn main() {
+ let args = Args::parse();
+ taler_main(CONFIG_SOURCE, args.common, async |_| match args.cmd {
+ Command::Logic { reset } => Ok(logic_harness(reset).await.unwrap()),
+ Command::Online { reset, network } => Ok(online_harness(network, reset).await.unwrap()),
+ });
+}
diff --git a/depolymerizer-bitcoin/src/cli.rs b/depolymerizer-bitcoin/src/cli.rs
@@ -1,6 +1,6 @@
/*
This file is part of TALER
- Copyright (C) 2025, 2026 Taler Systems SA
+ Copyright (C) 2025-2026 Taler Systems SA
TALER is free software; you can redistribute it and/or modify it under the
terms of the GNU Affero General Public License as published by the Free Software
@@ -17,18 +17,13 @@
use axum::{Router, middleware};
use taler_api::api::TalerRouter as _;
use taler_build::long_version;
-use taler_common::{
- CommonArgs,
- cli::ConfigCmd,
- config::Config,
- db::{dbinit, pool},
-};
+use taler_common::{CommonArgs, cli::ConfigCmd, config::Config};
use tracing::info;
use crate::{
- CONFIG_SOURCE, DB_SCHEMA,
api::{ServerState, status_middleware},
- config::{ServeCfg, WorkerCfg, parse_db_cfg},
+ config::{ServeCfg, WorkerCfg},
+ db::{dbinit, pool},
loops::{
watcher::watcher,
worker::{worker_loop, worker_transient},
@@ -81,33 +76,20 @@ pub enum Command {
pub async fn run(cmd: Command, cfg: &Config) -> anyhow::Result<()> {
match cmd {
Command::Dbinit { reset } => {
- let cfg = parse_db_cfg(cfg)?;
- let pool = pool(cfg.cfg, DB_SCHEMA).await?;
- let mut conn = pool.acquire().await?;
- dbinit(
- &mut conn,
- cfg.sql_dir.as_ref(),
- CONFIG_SOURCE.component_name,
- reset,
- )
- .await?;
+ dbinit(cfg, reset).await?;
}
Command::Setup { reset } => {
setup(cfg, reset).await?;
}
Command::Worker { transient } => {
let state = WorkerCfg::parse(cfg)?;
- let db_cfg = parse_db_cfg(cfg)?;
- let pool = pool(db_cfg.cfg, DB_SCHEMA).await?;
-
- #[cfg(feature = "fail")]
- tracing::warn!("Running with random failures");
+ let pool = pool(cfg).await?;
if transient {
worker_transient(state, pool).await?;
} else {
tokio::spawn(watcher(state.rpc_cfg.clone(), pool.clone()));
- tokio::spawn(worker_loop(state, pool)).await?;
+ worker_loop(state, pool.clone()).await;
info!("btc-wire stopped");
}
@@ -119,8 +101,7 @@ pub async fn run(cmd: Command, cfg: &Config) -> anyhow::Result<()> {
std::process::exit(1);
}
} else {
- let db = parse_db_cfg(cfg)?;
- let pool = pool(db.cfg, DB_SCHEMA).await?;
+ let pool = pool(cfg).await?;
let cfg = ServeCfg::parse(cfg)?;
let api = ServerState::start(pool, cfg.payto, cfg.currency).await;
let mut router = Router::new();
diff --git a/depolymerizer-bitcoin/src/db.rs b/depolymerizer-bitcoin/src/db.rs
@@ -1,6 +1,6 @@
/*
This file is part of TALER
- Copyright (C) 2025, 2026 Taler Systems SA
+ Copyright (C) 2025-2026 Taler Systems SA
TALER is free software; you can redistribute it and/or modify it under the
terms of the GNU Affero General Public License as published by the Free Software
@@ -34,6 +34,7 @@ use taler_common::{
TransferResponse, TransferState, TransferStatus,
},
},
+ config::Config,
db::IncomingType,
types::amount::{Amount, Currency},
};
@@ -41,10 +42,33 @@ use tokio::sync::watch::Receiver;
use url::Url;
use crate::{
+ config::parse_db_cfg,
payto::FullBtcPayto,
sql::{sql_addr, sql_btc_amount, sql_generic_payto, sql_payto},
};
+const SCHEMA: &str = "depolymerizer_bitcoin";
+
+pub async fn pool(cfg: &Config) -> anyhow::Result<PgPool> {
+ let db = parse_db_cfg(cfg)?;
+ let pool = taler_common::db::pool(db.cfg, SCHEMA).await?;
+ Ok(pool)
+}
+
+pub async fn dbinit(cfg: &Config, reset: bool) -> anyhow::Result<PgPool> {
+ let db_cfg = parse_db_cfg(cfg)?;
+ let pool = taler_common::db::pool(db_cfg.cfg, SCHEMA).await?;
+ let mut db = pool.acquire().await?;
+ taler_common::db::dbinit(
+ &mut db,
+ db_cfg.sql_dir.as_ref(),
+ "depolymerizer-bitcoin",
+ reset,
+ )
+ .await?;
+ Ok(pool)
+}
+
/// Initialize the worker status
pub async fn init_status(db: &PgPool) -> sqlx::Result<()> {
sqlx::query(
@@ -163,9 +187,8 @@ pub async fn transfer_page(
Some(s) => match s {
TransferState::pending => Some(DebitStatus::requested),
TransferState::success => Some(DebitStatus::sent),
- TransferState::transient_failure
- | TransferState::permanent_failure
- | TransferState::late_failure => {
+ TransferState::permanent_failure => Some(DebitStatus::ignored),
+ TransferState::transient_failure | TransferState::late_failure => {
return Ok(Vec::new());
}
},
@@ -197,10 +220,7 @@ pub async fn transfer_page(
|r: PgRow| {
Ok(TransferListStatus {
row_id: r.try_get_u64(0)?,
- status: match r.try_get(1)? {
- DebitStatus::requested | DebitStatus::sent => TransferState::pending,
- DebitStatus::confirmed => TransferState::success,
- },
+ status: r.try_get::<DebitStatus, _>(1)?.into(),
amount: r.try_get_amount(2, currency)?,
credit_account: sql_payto(&r, 3, 4)?,
timestamp: r.try_get_taler_timestamp(5)?,
@@ -221,6 +241,7 @@ pub async fn transfer_by_id(
"
SELECT
status,
+ status_msg,
amount,
exchange_url,
wtid,
@@ -234,17 +255,14 @@ pub async fn transfer_by_id(
.bind(id as i64)
.try_map(|r: PgRow| {
Ok(TransferStatus {
- status: match r.try_get(0)? {
- DebitStatus::requested | DebitStatus::sent => TransferState::pending,
- DebitStatus::confirmed => TransferState::success,
- },
- status_msg: None,
- amount: r.try_get_amount(1, currency)?,
- origin_exchange_url: r.try_get(2)?,
- wtid: r.try_get(3)?,
- credit_account: sql_payto(&r, 4, 5)?,
- metadata: r.try_get(6)?,
- timestamp: r.try_get_taler_timestamp(7)?,
+ status: r.try_get::<DebitStatus, _>(0)?.into(),
+ status_msg: r.try_get(1)?,
+ amount: r.try_get_amount(2, currency)?,
+ origin_exchange_url: r.try_get(3)?,
+ wtid: r.try_get(4)?,
+ credit_account: sql_payto(&r, 5, 6)?,
+ metadata: r.try_get(7)?,
+ timestamp: r.try_get_taler_timestamp(8)?,
})
})
.fetch_optional(db)
@@ -544,7 +562,7 @@ pub async fn transfer_unregister(
}
/// Update a transaction id after bumping it
-pub async fn bump_tx_id(
+pub async fn transfer_bumpfee(
db: &mut PgConnection,
to: &Txid,
wtid: &ShortHashCode,
@@ -632,8 +650,14 @@ pub async fn reorg<'a>(e: impl PgExecutor<'a>, ids: &[Txid]) -> sqlx::Result<Vec
.await
}
-#[derive(Debug)]
-pub enum SyncOutResult {
+#[derive(Debug, PartialEq, Eq)]
+pub struct SyncOutResult {
+ pub id: Option<u64>,
+ pub state: SyncOutState,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum SyncOutState {
New,
Replaced,
Recovered,
@@ -651,25 +675,30 @@ pub enum TxOutKind<'a> {
},
}
+pub struct TxOut<'a> {
+ pub id: Txid,
+ pub replaces_txid: Option<Txid>,
+ pub amount: Amount,
+ pub credit_acc: &'a Address,
+ pub block_time: Timestamp,
+}
+
pub async fn sync_out<'a>(
e: impl PgExecutor<'a>,
- txid: &Txid,
- replace_txid: Option<&Txid>,
- amount: &Amount,
- credit_acc: &Address,
+ tx: &TxOut<'_>,
kind: &TxOutKind<'_>,
- created: &Timestamp,
+ confirmed: bool,
) -> sqlx::Result<SyncOutResult> {
let query = sqlx::query(
"
- SELECT out_replaced, out_recovered, out_new
- FROM sync_out($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+ SELECT out_tx_row_id, out_replaced, out_recovered, out_new
+ FROM sync_out($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
",
)
- .bind(txid.as_byte_array())
- .bind(replace_txid.map(|it| it.as_byte_array()))
- .bind(amount)
- .bind(credit_acc.to_string());
+ .bind(tx.id.as_byte_array())
+ .bind(tx.replaces_txid.as_ref().map(|it| it.as_byte_array()))
+ .bind(tx.amount)
+ .bind(tx.credit_acc.to_string());
match kind {
TxOutKind::Simple => query
.bind(None::<&[u8]>)
@@ -691,17 +720,21 @@ pub async fn sync_out<'a>(
.bind(metadata)
.bind(None::<&[u8]>),
}
- .bind_timestamp(created)
+ .bind_timestamp(&tx.block_time)
+ .bind(confirmed)
.bind_timestamp(&Timestamp::now())
.try_map(|r: PgRow| {
- Ok(if r.try_get_flag(0)? {
- SyncOutResult::Replaced
- } else if r.try_get_flag(1)? {
- SyncOutResult::Recovered
- } else if r.try_get_flag(2)? {
- SyncOutResult::New
- } else {
- SyncOutResult::None
+ Ok(SyncOutResult {
+ id: r.try_get_opt_u64(0)?,
+ state: if r.try_get_flag(1)? {
+ SyncOutState::Replaced
+ } else if r.try_get_flag(2)? {
+ SyncOutState::Recovered
+ } else if r.try_get_flag(3)? {
+ SyncOutState::New
+ } else {
+ SyncOutState::None
+ },
})
})
.fetch_one(e)
@@ -713,7 +746,7 @@ pub async fn pending_transfer<'a>(
currency: &Currency,
) -> sqlx::Result<
Option<(
- i64,
+ u64,
bitcoin::Amount,
ShortHashCode,
Address,
@@ -736,7 +769,7 @@ pub async fn pending_transfer<'a>(
)
.try_map(|r: PgRow| {
Ok((
- r.try_get(0)?,
+ r.try_get_u64(0)?,
sql_btc_amount(&r, 1, currency)?,
r.try_get(2)?,
sql_addr(&r, 3)?,
@@ -748,10 +781,24 @@ pub async fn pending_transfer<'a>(
.await
}
+/// Update transfer status to 'ignored' and bind it to a txid
+pub async fn transfer_ignored<'a>(
+ e: impl PgExecutor<'a>,
+ id: u64,
+ reason: &str,
+) -> sqlx::Result<()> {
+ sqlx::query("UPDATE transfer SET status='ignored', status_msg=$2 WHERE transfer_id=$1")
+ .bind(id as i64)
+ .bind(reason)
+ .execute(e)
+ .await?;
+ Ok(())
+}
+
/// Update transfer status to 'sent' and bind it to a txid
-pub async fn transfer_sent<'a>(e: impl PgExecutor<'a>, id: i64, txid: &Txid) -> sqlx::Result<()> {
+pub async fn transfer_sent<'a>(e: impl PgExecutor<'a>, id: u64, txid: &Txid) -> sqlx::Result<()> {
sqlx::query("UPDATE transfer SET status='sent', txid=$2 WHERE transfer_id=$1")
- .bind(id)
+ .bind(id as i64)
.bind(txid.as_byte_array())
.execute(e)
.await?;
@@ -772,7 +819,7 @@ pub async fn transfer_conflict<'a>(e: impl PgExecutor<'a>, id: &Txid) -> sqlx::R
pub async fn pending_bounce<'a>(
e: impl PgExecutor<'a>,
-) -> sqlx::Result<Option<(i64, Txid, Option<String>)>> {
+) -> sqlx::Result<Option<(i64, Txid, CompactString)>> {
sqlx::query(
"
SELECT
@@ -781,7 +828,8 @@ pub async fn pending_bounce<'a>(
reason
FROM bounced
JOIN tx_in USING (tx_in_id)
- WHERE status='requested' ORDER BY received_at LIMIT 1
+ WHERE status='requested' AND tx_in.txid IS NOT NULL
+ ORDER BY received_at LIMIT 1
",
)
.try_map(|r: PgRow| {
@@ -848,10 +896,11 @@ pub mod test {
CONFIG_SOURCE,
api::test::CLIENT,
db::{
- AddIncomingResult, ProblematicTx, SyncOutResult, TransferResult, TxOutKind, bounce,
- bounce_sent, bump_tx_id, get_sync_state, incoming_history, init_status,
+ AddIncomingResult, ProblematicTx, SyncOutResult, SyncOutState, TransferResult, TxOut,
+ TxOutKind, bounce, bounce_sent, get_sync_state, incoming_history, init_status,
init_sync_state, pending_bounce, register_tx_in, register_tx_in_admin, reorg,
- revenue_history, swap_sync_state, sync_out, transfer, update_status,
+ revenue_history, swap_sync_state, sync_out, transfer, transfer_bumpfee,
+ transfer_conflict, transfer_ignored, transfer_sent, update_status,
},
};
@@ -977,14 +1026,14 @@ pub mod test {
// Reserve transaction
routine(
- &Some(IncomingSubject::Reserve(first.clone())),
+ &Some(IncomingSubject::Reserve(first)),
&Some(IncomingSubject::Reserve(second)),
)
.await;
// Kyc transaction
routine(
- &Some(IncomingSubject::Kyc(first.clone())),
+ &Some(IncomingSubject::Kyc(first)),
&Some(IncomingSubject::Kyc(first)),
)
.await;
@@ -1070,6 +1119,408 @@ pub mod test {
}
#[tokio::test]
+ async fn sync_out_simple() {
+ let (_, pool) = setup().await;
+ let amount = amount("KUDOS:10");
+ let now = now_sql_stable_ts();
+
+ // Sync
+ let txid = rand_tx_id();
+ let out = TxOut {
+ id: txid,
+ replaces_txid: None,
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ assert_eq!(
+ sync_out(&pool, &out, &TxOutKind::Simple, false)
+ .await
+ .unwrap(),
+ SyncOutResult {
+ id: None,
+ state: SyncOutState::None
+ }
+ );
+ assert_eq!(
+ sync_out(&pool, &out, &TxOutKind::Simple, true)
+ .await
+ .unwrap(),
+ SyncOutResult {
+ id: Some(1),
+ state: SyncOutState::New
+ }
+ );
+ assert_eq!(
+ sync_out(&pool, &out, &TxOutKind::Simple, true)
+ .await
+ .unwrap(),
+ SyncOutResult {
+ id: Some(1),
+ state: SyncOutState::None
+ }
+ );
+
+ // Replaced
+ let out = TxOut {
+ id: rand_tx_id(),
+ replaces_txid: Some(txid),
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ assert_eq!(
+ sync_out(&pool, &out, &TxOutKind::Simple, false)
+ .await
+ .unwrap(),
+ SyncOutResult {
+ id: None,
+ state: SyncOutState::None
+ }
+ );
+ assert_eq!(
+ sync_out(&pool, &out, &TxOutKind::Simple, true)
+ .await
+ .unwrap(),
+ SyncOutResult {
+ id: Some(1),
+ state: SyncOutState::Replaced
+ }
+ );
+ assert_eq!(
+ sync_out(&pool, &out, &TxOutKind::Simple, true)
+ .await
+ .unwrap(),
+ SyncOutResult {
+ id: Some(1),
+ state: SyncOutState::None
+ }
+ );
+ }
+
+ #[tokio::test]
+ async fn sync_out_talerable() {
+ let (mut db, poll) = setup().await;
+ let amount = amount("KUDOS:10");
+ let now = now_sql_stable_ts();
+
+ let prepare_transfer = async || {
+ let t = TransferRequest {
+ amount,
+ exchange_base_url: url("https://exchange.example.com"),
+ request_uid: HashCode::rand(),
+ wtid: ShortHashCode::rand(),
+ metadata: None,
+ credit_account: CLIENT.as_uri(),
+ };
+ let id = match transfer(&poll, &CLIENT, &t).await.unwrap() {
+ TransferResult::Success(res) => res.row_id,
+ _ => unreachable!(),
+ };
+ (id, t.wtid, t.exchange_base_url)
+ };
+
+ // Sync
+ let (id, wtid, url) = &prepare_transfer().await;
+ let out = TxOut {
+ id: rand_tx_id(),
+ replaces_txid: None,
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Talerable {
+ wtid,
+ url,
+ metadata: None,
+ };
+ transfer_sent(&mut *db, *id, &out.id).await.unwrap();
+ assert_eq!(
+ sync_out(&poll, &out, &kind, false).await.unwrap(),
+ SyncOutResult {
+ id: None,
+ state: SyncOutState::None
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::New
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::None
+ }
+ );
+
+ // Conflict
+ let (id, wtid, url) = &prepare_transfer().await;
+ let tx_id = rand_tx_id();
+ let conflict_id = rand_tx_id();
+ let out = TxOut {
+ id: conflict_id,
+ replaces_txid: None,
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Talerable {
+ wtid,
+ url,
+ metadata: None,
+ };
+ transfer_sent(&mut *db, *id, &tx_id).await.unwrap();
+ transfer_conflict(&mut *db, &conflict_id).await.unwrap();
+ assert_eq!(
+ sync_out(&poll, &out, &kind, false).await.unwrap(),
+ SyncOutResult {
+ id: None,
+ state: SyncOutState::None
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::New
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::None
+ }
+ );
+
+ // Bump fee
+ let (id, wtid, url) = &prepare_transfer().await;
+ let tx_id = rand_tx_id();
+ let bump_id = rand_tx_id();
+ let out = TxOut {
+ id: bump_id,
+ replaces_txid: Some(tx_id),
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Talerable {
+ wtid,
+ url,
+ metadata: None,
+ };
+ transfer_sent(&mut *db, *id, &tx_id).await.unwrap();
+ transfer_bumpfee(&mut db, &bump_id, wtid).await.unwrap();
+ assert_eq!(
+ sync_out(&poll, &out, &kind, false).await.unwrap(),
+ SyncOutResult {
+ id: None,
+ state: SyncOutState::None
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::New
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::None
+ }
+ );
+
+ // Recover
+ let (id, wtid, url) = &prepare_transfer().await;
+ let out = TxOut {
+ id: rand_tx_id(),
+ replaces_txid: None,
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Talerable {
+ wtid,
+ url,
+ metadata: None,
+ };
+ assert_eq!(
+ sync_out(&poll, &out, &kind, false).await.unwrap(),
+ SyncOutResult {
+ id: None,
+ state: SyncOutState::Recovered
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::New
+ }
+ );
+
+ // Recover completed
+ let (id, wtid, url) = &prepare_transfer().await;
+ let out = TxOut {
+ id: rand_tx_id(),
+ replaces_txid: None,
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Talerable {
+ wtid,
+ url,
+ metadata: None,
+ };
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::Recovered
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::None
+ }
+ );
+
+ // Recover bump
+ let (id, wtid, url) = &prepare_transfer().await;
+ let out = TxOut {
+ id: rand_tx_id(),
+ replaces_txid: Some(rand_tx_id()),
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Talerable {
+ wtid,
+ url,
+ metadata: None,
+ };
+ assert_eq!(
+ sync_out(&poll, &out, &kind, false).await.unwrap(),
+ SyncOutResult {
+ id: None,
+ state: SyncOutState::Recovered
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::New
+ }
+ );
+
+ // Recover bump completed
+ let (id, wtid, url) = &prepare_transfer().await;
+ let out = TxOut {
+ id: rand_tx_id(),
+ replaces_txid: Some(rand_tx_id()),
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Talerable {
+ wtid,
+ url,
+ metadata: None,
+ };
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::Recovered
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::None
+ }
+ );
+
+ // Recover sent bump
+ let (id, wtid, url) = &prepare_transfer().await;
+ let txid = rand_tx_id();
+ let out = TxOut {
+ id: rand_tx_id(),
+ replaces_txid: Some(txid),
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Talerable {
+ wtid,
+ url,
+ metadata: None,
+ };
+ transfer_sent(&mut *db, *id, &txid).await.unwrap();
+ assert_eq!(
+ sync_out(&poll, &out, &kind, false).await.unwrap(),
+ SyncOutResult {
+ id: None,
+ state: SyncOutState::Replaced
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::New
+ }
+ );
+
+ // Recover sent bump completed
+ let (id, wtid, url) = &prepare_transfer().await;
+ let txid = rand_tx_id();
+ let out = TxOut {
+ id: rand_tx_id(),
+ replaces_txid: Some(txid),
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Talerable {
+ wtid,
+ url,
+ metadata: None,
+ };
+ transfer_sent(&mut *db, *id, &txid).await.unwrap();
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::Replaced
+ }
+ );
+ assert_eq!(
+ sync_out(&poll, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(*id),
+ state: SyncOutState::None
+ }
+ );
+
+ // Failure
+ let (id, _, _) = &prepare_transfer().await;
+ transfer_ignored(&mut *db, *id, "Oh nooooo").await.unwrap();
+ }
+
+ #[tokio::test]
async fn bounces() {
let (_, db) = setup().await;
let amount = amount("KUDOS:10");
@@ -1095,34 +1546,28 @@ pub mod test {
}
_ => unreachable!(),
}
- assert_matches!(
- sync_out(
- &db,
- &bounce_txid,
- None,
- &amount,
- &ADDR,
- &TxOutKind::Bounce(bounced_txid),
- &now,
- )
- .await
- .unwrap(),
- SyncOutResult::New
+ let out = TxOut {
+ id: bounce_txid,
+ replaces_txid: None,
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ };
+ let kind = TxOutKind::Bounce(bounced_txid);
+ assert_eq!(
+ sync_out(&db, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(1),
+ state: SyncOutState::New
+ }
);
assert_eq!(pending_bounce(&db).await.unwrap(), None);
- assert_matches!(
- sync_out(
- &db,
- &bounce_txid,
- None,
- &amount,
- &ADDR,
- &TxOutKind::Bounce(bounced_txid),
- &now,
- )
- .await
- .unwrap(),
- SyncOutResult::None
+ assert_eq!(
+ sync_out(&db, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(1),
+ state: SyncOutState::None
+ }
);
// Recovered
@@ -1138,113 +1583,29 @@ pub mod test {
..
}
);
- assert_matches!(
- sync_out(
- &db,
- &bounce_txid,
- None,
- &amount,
- &ADDR,
- &TxOutKind::Bounce(bounced_txid),
- &now,
- )
- .await
- .unwrap(),
- SyncOutResult::Recovered
- );
- assert_matches!(
- sync_out(
- &db,
- &bounce_txid,
- None,
- &amount,
- &ADDR,
- &TxOutKind::Bounce(bounced_txid),
- &now,
- )
- .await
- .unwrap(),
- SyncOutResult::None
- );
- assert_eq!(pending_bounce(&db).await.unwrap(), None);
- }
-
- #[tokio::test]
- async fn sync_out_talerable_and_replace() {
- let (mut db, poll) = setup().await;
- let amount = amount("KUDOS:10");
- let now = now_sql_stable_ts();
-
- // 1. Simple Sync Out
- let txid = rand_tx_id();
- assert_matches!(
- sync_out(&poll, &txid, None, &amount, &ADDR, &TxOutKind::Simple, &now,)
- .await
- .unwrap(),
- SyncOutResult::New
- );
- assert_matches!(
- sync_out(&poll, &txid, None, &amount, &ADDR, &TxOutKind::Simple, &now,)
- .await
- .unwrap(),
- SyncOutResult::None
- );
-
- // 2. Replace (Fee Bump)
- assert_matches!(
- sync_out(
- &poll,
- &rand_tx_id(),
- Some(&txid),
- &amount,
- &ADDR,
- &TxOutKind::Simple,
- &now,
- )
- .await
- .unwrap(),
- SyncOutResult::Replaced
- );
-
- // 3. Recover Talerable Transfer
- let t = TransferRequest {
+ let out = TxOut {
+ id: bounce_txid,
+ replaces_txid: None,
amount,
- exchange_base_url: url("https://exchange.example.com"),
- request_uid: HashCode::rand(),
- wtid: ShortHashCode::rand(),
- metadata: None,
- credit_account: CLIENT.as_uri(),
+ credit_acc: &ADDR,
+ block_time: now,
};
-
- // Create the pending transfer
- assert_matches!(
- transfer(&poll, &CLIENT, &t).await.unwrap(),
- TransferResult::Success(_)
+ let kind = TxOutKind::Bounce(bounced_txid);
+ assert_eq!(
+ sync_out(&db, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(2),
+ state: SyncOutState::New
+ }
);
-
- let txid = rand_tx_id();
- // Sync it out
- assert_matches!(
- sync_out(
- &poll,
- &txid,
- None,
- &amount,
- &ADDR,
- &TxOutKind::Talerable {
- wtid: &t.wtid,
- url: &t.exchange_base_url,
- metadata: None,
- },
- &now,
- )
- .await
- .unwrap(),
- SyncOutResult::Recovered
+ assert_eq!(
+ sync_out(&db, &out, &kind, true).await.unwrap(),
+ SyncOutResult {
+ id: Some(2),
+ state: SyncOutState::None
+ }
);
-
- // Bump fee
- bump_tx_id(&mut db, &rand_tx_id(), &t.wtid).await.unwrap();
+ assert_eq!(pending_bounce(&db).await.unwrap(), None);
}
#[tokio::test]
@@ -1262,7 +1623,7 @@ pub mod test {
&amount,
&ADDR,
&now,
- &Some(IncomingSubject::Reserve(reserve_pub.clone())),
+ &Some(IncomingSubject::Reserve(reserve_pub)),
)
.await
.unwrap();
@@ -1275,12 +1636,15 @@ pub mod test {
.unwrap();
sync_out(
&pool,
- &txid_bounce,
- None,
- &amount,
- &ADDR,
+ &TxOut {
+ id: txid_bounce,
+ replaces_txid: None,
+ amount,
+ credit_acc: &ADDR,
+ block_time: now,
+ },
&TxOutKind::Bounce(txid_bounced),
- &now,
+ true,
)
.await
.unwrap();
diff --git a/depolymerizer-bitcoin/src/fail_point.rs b/depolymerizer-bitcoin/src/fail_point.rs
@@ -1,31 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2022-2025 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU Affero General Public License as published by the Free Software
- Foundation; either version 3, or (at your option) any later version.
-
- TALER is distributed in the hope that it will be useful, but WITHOUT ANY
- WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
- A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
-
- You should have received a copy of the GNU Affero General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
-*/
-#[derive(Debug, thiserror::Error)]
-#[error("{0}")]
-pub struct Injected(&'static str);
-
-/// Inject random failure when 'fail' feature is used
-#[allow(unused_variables)]
-pub fn fail_point(msg: &'static str, prob: f32) -> Result<(), Injected> {
- #[cfg(feature = "fail")]
- return if rand::random::<f32>() < prob {
- Err(Injected(msg))
- } else {
- Ok(())
- };
-
- Ok(())
-}
diff --git a/depolymerizer-bitcoin/src/lib.rs b/depolymerizer-bitcoin/src/lib.rs
@@ -16,19 +16,18 @@
use std::str::FromStr;
use bitcoin::{Address, Amount, Network, Txid, hashes::hex::FromHex};
-use rpc::{Category, Rpc, Transaction};
-use rpc_utils::{segwit_min_amount, sender_address};
+use rpc::{Category, Transaction};
+use rpc_utils::segwit_min_amount;
use segwit::{decode_segwit_msg, encode_segwit_key};
use taler_common::{api::EddsaPublicKey, config::parser::ConfigSource};
-use crate::segwit::DecodeSegWitErr;
+use crate::{rpc::RpcApi, segwit::DecodeSegWitErr};
pub mod api;
pub mod cli;
pub mod config;
pub mod db;
-mod fail_point;
-mod loops;
+pub mod loops;
pub mod payto;
pub mod rpc;
pub mod rpc_utils;
@@ -38,7 +37,6 @@ pub mod sql;
pub mod taler_utils;
pub const CONFIG_SOURCE: ConfigSource = ConfigSource::simple("depolymerizer-bitcoin");
-pub const DB_SCHEMA: &str = "depolymerizer_bitcoin";
#[derive(Debug, thiserror::Error)]
pub enum GetOpReturnErr {
@@ -49,12 +47,12 @@ pub enum GetOpReturnErr {
}
/// An extended bitcoincore JSON-RPC api client who can send and retrieve metadata with their transaction
-impl Rpc {
+pub trait RpcApiExtended: RpcApi {
/// Send a transaction with a 32B key as metadata encoded using fake segwit addresses
- pub async fn send_segwit_key(
+ async fn send_segwit_key(
&mut self,
to: &Address,
- amount: &Amount,
+ amount: Amount,
metadata: &[u8; 32],
) -> rpc::Result<Txid> {
let network = guess_network(to);
@@ -71,12 +69,12 @@ impl Rpc {
];
let mut recipients = vec![(to, amount)];
let min = segwit_min_amount();
- recipients.extend(addresses.iter().map(|addr| (addr, &min)));
+ recipients.extend(addresses.iter().map(|addr| (addr, min)));
self.send_many(recipients).await
}
/// Get detailed information about an in-wallet transaction and it's 32B metadata key encoded using fake segwit addresses
- pub async fn get_tx_segwit_key(
+ async fn get_tx_segwit_key(
&mut self,
id: &Txid,
) -> Result<(Transaction, Result<EddsaPublicKey, DecodeSegWitErr>), rpc::Error> {
@@ -98,7 +96,7 @@ impl Rpc {
}
/// Get detailed information about an in-wallet transaction and its op_return metadata
- pub async fn get_tx_op_return(
+ async fn get_tx_op_return(
&mut self,
id: &Txid,
) -> Result<(Transaction, Vec<u8>), GetOpReturnErr> {
@@ -118,12 +116,27 @@ impl Rpc {
Ok((full, metadata))
}
+ /// Get the first sender address from a raw transaction
+ async fn sender_address(&mut self, full: &Transaction) -> rpc::Result<Address> {
+ let first = &full.decoded.vin[0];
+ let tx = self.get_input_output(&first.txid.unwrap()).await?;
+ Ok(tx
+ .vout
+ .into_iter()
+ .find(|it| it.n == first.vout.unwrap())
+ .unwrap()
+ .script_pub_key
+ .address
+ .unwrap()
+ .assume_checked())
+ }
+
/// Bounce a transaction bask to its sender
///
/// There is no reliable way to bounce a transaction as you cannot know if the addresses
/// used are shared or come from a third-party service. We only send back to the first input
/// address as a best-effort gesture.
- pub async fn bounce(
+ async fn bounce(
&mut self,
id: &Txid,
bounce_fee: &Amount,
@@ -134,13 +147,15 @@ impl Rpc {
assert!(detail.category == Category::Receive);
let amount = detail.amount.to_unsigned().unwrap();
- let sender = sender_address(self, &full).await?;
+ let sender = self.sender_address(&full).await?;
let bounce_amount = Amount::from_sat(amount.to_sat().saturating_sub(bounce_fee.to_sat()));
// Send refund making recipient pay the transaction fees
- self.send(&sender, &bounce_amount, metadata, true).await
+ self.send(&sender, bounce_amount, metadata, true).await
}
}
+impl<T: RpcApi> RpcApiExtended for T {}
+
pub fn guess_network(address: &Address) -> Network {
let addr = address.as_unchecked();
for network in [
diff --git a/depolymerizer-bitcoin/src/loops.rs b/depolymerizer-bitcoin/src/loops.rs
@@ -14,7 +14,9 @@
TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
*/
-use crate::{fail_point::Injected, rpc};
+use failure_injection::InjectedErr;
+
+use crate::rpc;
pub mod analysis;
pub mod watcher;
@@ -29,7 +31,7 @@ pub enum LoopError {
#[error("Another btc-wire process is running concurrently")]
Concurrency,
#[error(transparent)]
- Injected(#[from] Injected),
+ Injected(#[from] InjectedErr),
}
pub type LoopResult<T> = Result<T, LoopError>;
diff --git a/depolymerizer-bitcoin/src/loops/analysis.rs b/depolymerizer-bitcoin/src/loops/analysis.rs
@@ -17,10 +17,10 @@
use tracing::warn;
use super::LoopResult;
-use crate::rpc::{ChainTipsStatus, Rpc};
+use crate::rpc::{ChainTipsStatus, RpcApi};
/// Analyse blockchain behavior and return the new confirmation delay
-pub async fn analysis(rpc: &mut Rpc, current: u32, max: u32) -> LoopResult<u32> {
+pub async fn analysis(rpc: &mut impl RpcApi, current: u32, max: u32) -> LoopResult<u32> {
// Get biggest known valid fork
let fork = rpc
.get_chain_tips()
diff --git a/depolymerizer-bitcoin/src/loops/watcher.rs b/depolymerizer-bitcoin/src/loops/watcher.rs
@@ -20,14 +20,17 @@ use tokio::time::sleep;
use tracing::error;
use super::LoopResult;
-use crate::{config::RpcCfg, rpc::rpc_common};
+use crate::{
+ config::RpcCfg,
+ rpc::{Rpc, RpcApi as _},
+};
/// Wait for new block and notify arrival with postgreSQL notifications
pub async fn watcher(rpc_cfg: RpcCfg, pool: PgPool) {
let mut jitter = ExpoBackoffDecorr::default();
loop {
let result: LoopResult<()> = async {
- let mut rpc = rpc_common(&rpc_cfg).await?;
+ let mut rpc = Rpc::new(&rpc_cfg).await?;
loop {
sqlx::query("NOTIFY new_block").execute(&pool).await?;
rpc.wait_for_new_block().await?;
diff --git a/depolymerizer-bitcoin/src/loops/worker.rs b/depolymerizer-bitcoin/src/loops/worker.rs
@@ -17,6 +17,7 @@ use std::{fmt::Write, time::SystemTime};
use bitcoin::{Amount as BtcAmount, Txid, hashes::Hash};
use depolymerizer_common::metadata::OutMetadata;
+use failure_injection::fail_point;
use jiff::Timestamp;
use sqlx::{
Acquire, Either, PgConnection, PgPool,
@@ -29,12 +30,14 @@ use tracing::{debug, error, info, trace, warn};
use super::{LoopError, LoopResult, analysis::analysis};
use crate::{
- GetOpReturnErr,
+ GetOpReturnErr, RpcApiExtended as _,
config::WorkerCfg,
- db::{self, AddIncomingResult, SyncOutResult, TxOutKind},
- fail_point::fail_point,
- rpc::{self, Category, ErrorCode, ListSinceBlock, ListTransaction, Rpc, rpc_wallet},
- rpc_utils::sender_address,
+ db::{self, AddIncomingResult, SyncOutState, TxOut, TxOutKind},
+ rpc::{
+ self, Category,
+ ErrorCode::{self, RpcInvalidAddressOrKey},
+ ListSinceBlock, ListTransaction, Rpc, RpcApi,
+ },
taler_utils::btc_to_taler,
};
@@ -47,7 +50,12 @@ pub async fn worker_loop(mut state: WorkerCfg, pool: PgPool) {
loop {
let result: LoopResult<()> = async {
// Connect
- let rpc = &mut rpc_wallet(&state.rpc_cfg, &state.wallet_cfg).await?;
+ let rpc = &mut Rpc::new(&state.rpc_cfg).await?;
+ rpc.load_wallet(&state.wallet_cfg.name).await?;
+ if let Some(password) = &state.wallet_cfg.password {
+ rpc.unlock_wallet(password).await?;
+ }
+ let rpc = &mut rpc.wallet(&state.wallet_cfg.name);
let db = &mut PgListener::connect_with(&pool).await?;
// Listen to all channels
@@ -104,7 +112,7 @@ pub async fn worker_loop(mut state: WorkerCfg, pool: PgPool) {
// Perform analysis
state.conf = analysis(rpc, state.conf, state.max_conf).await?;
- worker_step(rpc, lock.as_mut(), &mut state, &mut status).await?;
+ worker_step(lock.as_mut(), rpc, &mut state, &mut status).await?;
skip_notification = false;
jitter.reset();
@@ -120,7 +128,10 @@ pub async fn worker_loop(mut state: WorkerCfg, pool: PgPool) {
skip_notification = match e {
LoopError::DB(_) | LoopError::Injected(_) | LoopError::Concurrency => true,
LoopError::Rpc(e) => match e {
- rpc::Error::Transport(_) | rpc::Error::Connect(_) => true,
+ rpc::Error::Transport(_)
+ | rpc::Error::Cookie(_, _)
+ | rpc::Error::Tcp(_, _)
+ | rpc::Error::Elapsed(_, _) => true,
rpc::Error::RPC { code, .. } => code == ErrorCode::RpcWalletError,
rpc::Error::Bitcoin(_) | rpc::Error::Json { .. } | rpc::Error::Null => false,
},
@@ -136,7 +147,12 @@ pub async fn worker_transient(mut state: WorkerCfg, pool: PgPool) -> LoopResult<
let mut status = true;
// Connect
- let rpc = &mut rpc_wallet(&state.rpc_cfg, &state.wallet_cfg).await?;
+ let rpc = &mut Rpc::new(&state.rpc_cfg).await?;
+ rpc.load_wallet(&state.wallet_cfg.name).await?;
+ if let Some(password) = &state.wallet_cfg.password {
+ rpc.unlock_wallet(password).await?;
+ }
+ let rpc = &mut rpc.wallet(&state.wallet_cfg.name);
let mut db = pool.acquire().await?;
// It is not possible to atomically update the blockchain and the database.
@@ -153,19 +169,19 @@ pub async fn worker_transient(mut state: WorkerCfg, pool: PgPool) -> LoopResult<
return Err(LoopError::Concurrency);
};
- worker_step(rpc, lock.as_mut(), &mut state, &mut status).await?;
+ worker_step(lock.as_mut(), rpc, &mut state, &mut status).await?;
Ok(())
}
/// Synchronize local db with blockchain and perform transactions
-async fn worker_step(
- rpc: &mut Rpc,
+pub async fn worker_step(
db: &mut PgConnection,
+ rpc: &mut impl RpcApi,
state: &mut WorkerCfg,
status: &mut bool,
) -> LoopResult<()> {
// Sync chain
- if let Some(stuck) = sync_chain(rpc, db, state, status).await? {
+ if let Some(stuck) = sync_chain(db, rpc, state, status).await? {
// As we are now in sync with the blockchain if a transaction has Requested status it have not been sent
// Send requested debits
@@ -174,9 +190,9 @@ async fn worker_step(
// Bump stuck transactions
for (txid, wtid) in stuck {
let bump = rpc.bump_fee(&txid).await?;
- fail_point("(injected) fail bump", 0.3)?;
- db::bump_tx_id(&mut *db, &bump.txid, &wtid).await?;
- info!(target: "worker", ">> (bump) {wtid} {txid} -> {}", bump.txid);
+ failure_injection::fail_point("bumpfee")?;
+ db::transfer_bumpfee(&mut *db, &bump.txid, &wtid).await?;
+ info!(target: "worker", ">> bump {wtid} {txid} -> {}", bump.txid);
}
// Send requested bounce
@@ -187,8 +203,8 @@ async fn worker_step(
/// Parse new transactions, return stuck transactions if the database is up to date with the latest mined block
async fn sync_chain(
- rpc: &mut Rpc,
db: &mut PgConnection,
+ rpc: &mut impl RpcApi,
state: &WorkerCfg,
status: &mut bool,
) -> LoopResult<Option<Vec<(Txid, ShortHashCode)>>> {
@@ -329,13 +345,13 @@ async fn sync_chain_removed(
/// Sync database with an incoming confirmed transaction
async fn sync_chain_incoming_confirmed(
txid: &Txid,
- rpc: &mut Rpc,
+ rpc: &mut impl RpcApi,
db: &mut PgConnection,
state: &WorkerCfg,
) -> Result<(), LoopError> {
let (tx, metadata) = rpc.get_tx_segwit_key(txid).await?;
// Store transactions in database
- let debit_addr = sender_address(rpc, &tx).await?;
+ let debit_addr = rpc.sender_address(&tx).await?;
let amount = btc_to_taler(&tx.amount, &state.currency);
let time = Timestamp::from_second(tx.time as i64).unwrap();
let ty = IncomingType::reserve;
@@ -347,7 +363,7 @@ async fn sync_chain_incoming_confirmed(
&amount,
&debit_addr,
&Timestamp::from_second(tx.time as i64).unwrap(),
- &Some(IncomingSubject::Reserve(reserve_pub.clone())),
+ &Some(IncomingSubject::Reserve(reserve_pub)),
)
.await?
{
@@ -377,10 +393,12 @@ async fn sync_chain_incoming_confirmed(
async fn sync_chain_outgoing(
txid: &Txid,
confirmations: i32,
- rpc: &mut Rpc,
+ rpc: &mut impl RpcApi,
db: &mut PgConnection,
state: &WorkerCfg,
) -> LoopResult<Option<ShortHashCode>> {
+ let confirmed = confirmations >= state.conf as i32;
+ let chain = if confirmed { "onchain" } else { "sent" };
match rpc
.get_tx_op_return(txid)
.await
@@ -397,54 +415,56 @@ async fn sync_chain_outgoing(
metadata,
} => {
if confirmations < 0 {
- // Handle conflicting tx
- if tx.replaced_by_txid.is_none() && db::transfer_conflict(db, txid).await? {
- warn!(target: "worker", ">> (conflict) {wtid} {txid} {credit_addr} {amount}");
+ // Handle conflict
+ if db::transfer_conflict(db, txid).await? {
+ warn!(target: "worker", ">> conflict {wtid} {txid} {credit_addr} {amount}");
}
- } else if confirmations > state.conf as i32 {
+ } else {
+ // Sync db state
match db::sync_out(
db,
- txid,
- tx.replaced_by_txid.as_ref(),
- &amount,
- &credit_addr,
+ &TxOut {
+ id: *txid,
+ replaces_txid: tx.replaces_txid,
+ amount,
+ credit_acc: &credit_addr,
+ block_time: created_at,
+ },
&TxOutKind::Talerable {
wtid: &wtid,
url: &url,
metadata: metadata.as_deref(),
},
- &created_at,
+ confirmed,
)
.await?
+ .state
{
- SyncOutResult::New => {
- info!(target: "worker", ">> (onchain) {wtid} {txid} {credit_addr} {amount}");
+ SyncOutState::New => {
+ info!(target: "worker", ">> {chain} {wtid} {txid} {credit_addr} {amount}");
}
- SyncOutResult::Replaced => {
- info!(
+ SyncOutState::Replaced => {
+ warn!(
target: "worker",
- ">> (recovered) {wtid} {txid} -> {} {credit_addr} {amount}",
- tx.replaced_by_txid.unwrap()
+ ">> recovered {chain} {wtid} {txid} -> {} {credit_addr} {amount}",
+ tx.replaces_txid.unwrap()
)
}
- SyncOutResult::Recovered => {
- warn!(target: "worker", ">> (recovered) {wtid} {txid} {credit_addr} {amount}")
+ SyncOutState::Recovered => {
+ warn!(target: "worker", ">> recovered {chain} {wtid} {txid} {credit_addr} {amount}")
}
- SyncOutResult::None => {}
+ SyncOutState::None => {}
}
- } else {
- // TODO sync transfer sent ?
-
- // Check if stuck
if let Some(delay) = state.bump_delay
&& confirmations == 0
&& tx.replaced_by_txid.is_none()
{
+ // Check stuck
let now = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.unwrap()
.as_secs();
- if now - tx.time > delay as u64 {
+ if now - tx.time >= delay as u64 {
return Ok(Some(wtid));
}
}
@@ -453,39 +473,42 @@ async fn sync_chain_outgoing(
OutMetadata::Bounce { bounced } => {
let bounced = Txid::from_byte_array(bounced);
if confirmations < 0 {
- // Handle conflicting tx
+ // Handle conflict
if db::bounce_conflict(db, txid).await? {
- warn!(target: "worker", "|| (conflict) {bounced} {txid}");
+ warn!(target: "worker", "|| conflict {bounced} {txid}");
}
- } else if confirmations > state.conf as i32 {
+ } else {
+ // Sync db state
match db::sync_out(
db,
- txid,
- tx.replaces_txid.as_ref(),
- &amount,
- &credit_addr,
+ &TxOut {
+ id: *txid,
+ replaces_txid: tx.replaces_txid,
+ amount,
+ credit_acc: &credit_addr,
+ block_time: created_at,
+ },
&TxOutKind::Bounce(bounced),
- &created_at,
+ confirmed,
)
.await?
+ .state
{
- SyncOutResult::New => {
- info!(target: "worker", "|| (onchain) {bounced} {txid}")
+ SyncOutState::New => {
+ info!(target: "worker", "|| {chain} {bounced} {txid}")
}
- SyncOutResult::Replaced => {
- info!(
+ SyncOutState::Replaced => {
+ warn!(
target: "worker",
- "|| (recovered) {bounced} {txid} -> {}",
+ "|| recovered {chain} {bounced} {txid} -> {}",
tx.replaced_by_txid.unwrap()
)
}
- SyncOutResult::Recovered => {
- warn!(target: "worker", "|| (recovered) {bounced} {txid}")
+ SyncOutState::Recovered => {
+ warn!(target: "worker", "|| recovered {chain} {bounced} {txid}")
}
- SyncOutResult::None => {}
+ SyncOutState::None => {}
}
- } else {
- // TODO sync bounce sent ?
}
}
}
@@ -500,24 +523,41 @@ async fn sync_chain_outgoing(
}
/// Send a debit transaction on the blockchain, return false if no more requested transactions are found
-async fn debit(db: &mut PgConnection, rpc: &mut Rpc, state: &WorkerCfg) -> LoopResult<bool> {
+async fn debit(
+ db: &mut PgConnection,
+ rpc: &mut impl RpcApi,
+ state: &WorkerCfg,
+) -> LoopResult<bool> {
// We rely on the advisory lock to ensure we are the only one sending transactions
if let Some((id, amount, wtid, addr, url, metadata)) =
db::pending_transfer(&mut *db, &state.currency).await?
{
let metadata = OutMetadata::Debit {
- wtid: wtid.clone(),
+ wtid,
url,
metadata,
};
- let txid = rpc
- .send(&addr, &amount, Some(&metadata.encode().unwrap()), false)
- .await?;
- fail_point("(injected) fail debit", 0.3)?;
+ let txid = match rpc
+ .send(&addr, amount, Some(&metadata.encode().unwrap()), false)
+ .await
+ {
+ Ok(id) => id,
+
+ Err(e) => match &e {
+ rpc::Error::RPC { code, msg, .. } if *code == RpcInvalidAddressOrKey => {
+ let reason = format!("{code:?} {msg}");
+ db::transfer_ignored(db, id, &reason).await?;
+ warn!(target: "worker", ">> ignored {wtid} {addr} {amount}: {reason}");
+ return Ok(true);
+ }
+ _ => return Err(e.into()),
+ },
+ };
+ fail_point("debit")?;
db::transfer_sent(db, id, &txid).await?;
let amount = btc_to_taler(&amount.to_signed().unwrap(), &state.currency);
- info!(target: "worker", ">> (sent) {wtid} {txid} {addr} {amount}");
+ info!(target: "worker", ">> sent {wtid} {txid} {addr} {amount}");
Ok(true)
} else {
Ok(false)
@@ -525,7 +565,7 @@ async fn debit(db: &mut PgConnection, rpc: &mut Rpc, state: &WorkerCfg) -> LoopR
}
/// Bounce a transaction on the blockchain, return false if no more requested transactions are found
-async fn bounce(db: &mut PgConnection, rpc: &mut Rpc, fee: &BtcAmount) -> LoopResult<bool> {
+async fn bounce(db: &mut PgConnection, rpc: &mut impl RpcApi, fee: &BtcAmount) -> LoopResult<bool> {
// We rely on the advisory lock to ensure we are the only one sending transactions
if let Some((id, bounced, reason)) = db::pending_bounce(&mut *db).await? {
let metadata = OutMetadata::Bounce {
@@ -537,13 +577,9 @@ async fn bounce(db: &mut PgConnection, rpc: &mut Rpc, fee: &BtcAmount) -> LoopRe
.await
{
Ok(txid) => {
- fail_point("(injected) fail bounce", 0.3)?;
+ fail_point("bounce")?;
db::bounce_sent(db, id, &txid).await?;
- if let Some(reason) = reason {
- info!(target: "worker", "|| (sent) {bounced} {txid}: {reason}");
- } else {
- info!(target: "worker", "|| (sent) {bounced} {txid}");
- }
+ info!(target: "worker", "|| sent {bounced} {txid}: {reason}");
}
Err(err) => match err {
rpc::Error::RPC {
diff --git a/depolymerizer-bitcoin/src/payto.rs b/depolymerizer-bitcoin/src/payto.rs
@@ -23,10 +23,7 @@ use taler_common::types::payto::{FullPayto, Payto, PaytoErr, PaytoImpl, PaytoURI
pub struct BtcWallet(pub Address);
const BITCOIN: &str = "bitcoin";
-
-#[derive(Debug, thiserror::Error)]
-#[error("missing wallet addr in path")]
-pub struct MissingAddr;
+const SEGMENT: &str = "wallet addr";
impl PaytoImpl for BtcWallet {
fn as_uri(&self) -> PaytoURI {
@@ -42,13 +39,13 @@ impl PaytoImpl for BtcWallet {
));
}
let Some(mut segments) = url.path_segments() else {
- return Err(PaytoErr::MissingSegment("wallet addr"));
+ return Err(PaytoErr::MissingSegment(SEGMENT));
};
let Some(first) = segments.next() else {
- return Err(PaytoErr::MissingSegment("wallet addr"));
+ return Err(PaytoErr::MissingSegment(SEGMENT));
};
let address =
- Address::from_str(first).map_err(|e| PaytoErr::malformed_segment("wallet addr", e))?;
+ Address::from_str(first).map_err(|e| PaytoErr::malformed_segment(SEGMENT, e))?;
Ok(Self(address.assume_checked()))
}
}
diff --git a/depolymerizer-bitcoin/src/rpc.rs b/depolymerizer-bitcoin/src/rpc.rs
@@ -45,39 +45,23 @@ use tokio::{
};
use tracing::trace;
-use crate::config::{RpcAuth, RpcCfg, WalletCfg};
+use crate::config::{RpcAuth, RpcCfg};
-/// Create a rpc connection with an unlocked wallet
-pub async fn rpc_wallet(config: &RpcCfg, wallet: &WalletCfg) -> Result<Rpc> {
- let mut rpc = Rpc::wallet(config, &wallet.name).await?;
- rpc.load_wallet(&wallet.name).await?;
- if let Some(password) = &wallet.password {
- rpc.unlock_wallet(password).await?;
- }
- Ok(rpc)
-}
-
-/// Create a rpc connection
-pub async fn rpc_common(config: &RpcCfg) -> Result<Rpc> {
- Ok(Rpc::common(config).await?)
-}
+const RPC_VERSION: &str = "2.0";
#[derive(Debug, serde::Serialize)]
struct RpcRequest<'a, T: serde::Serialize> {
+ jsonrpc: &'static str,
method: &'a str,
id: u64,
params: &'a T,
}
#[derive(Debug, serde::Deserialize)]
-#[serde(untagged)]
-enum RpcResponse<T> {
- RpcResponse {
- result: Option<T>,
- error: Option<RpcError>,
- id: u64,
- },
- Error(CompactString),
+struct RpcResponse<T> {
+ result: Option<T>,
+ error: Option<RpcError>,
+ id: u64,
}
#[derive(Debug, serde::Deserialize)]
@@ -103,8 +87,12 @@ pub enum Error {
method: CompactString,
e: serde_json::Error,
},
- #[error("connect: {0}")]
- Connect(#[from] RpcConnectErr),
+ #[error("failed to read cookie file at '{0}': {1}")]
+ Cookie(String, ErrorKind),
+ #[error("failed to connect {0}: {1}")]
+ Tcp(SocketAddr, ErrorKind),
+ #[error("failed to connect {0}: {1}")]
+ Elapsed(SocketAddr, tokio::time::error::Elapsed),
#[error("Null rpc, no result or error")]
Null,
}
@@ -120,15 +108,20 @@ fn expect_null(result: Result<()>) -> Result<()> {
}
}
+#[derive(Debug)]
pub struct JsonSocket {
- path: CompactString,
- cookie: CompactString,
sock: TcpStream,
buf: Vec<u8>,
}
impl JsonSocket {
- async fn call<T>(&mut self, method: &str, body: &impl serde::Serialize) -> Result<T>
+ async fn call<T>(
+ &mut self,
+ path: &str,
+ method: &str,
+ cookie: &str,
+ body: &impl serde::Serialize,
+ ) -> Result<T>
where
T: serde::de::DeserializeOwned,
{
@@ -142,10 +135,10 @@ impl JsonSocket {
let body_len = buf.len();
// Write HTTP request
- writeln!(buf, "POST {} HTTP/1.1\r", self.path)?;
+ writeln!(buf, "POST {path} HTTP/1.1\r")?;
// Write headers
writeln!(buf, "Accept: application/json-rpc\r")?;
- writeln!(buf, "Authorization: {}\r", self.cookie)?;
+ writeln!(buf, "Authorization: {cookie}\r")?;
writeln!(buf, "Content-Type: application/json-rpc\r")?;
writeln!(buf, "Content-Length: {body_len}\r")?;
// Write separator
@@ -184,115 +177,144 @@ impl JsonSocket {
}
}
-#[derive(Debug, thiserror::Error)]
-pub enum RpcConnectErr {
- #[error("failed to read cookie file at '{0}': {1}")]
- Cookie(String, ErrorKind),
- #[error("failed to connect {0}: {1}")]
- Tcp(SocketAddr, ErrorKind),
- #[error("failed to connect {0}: {1}")]
- Elapsed(SocketAddr, tokio::time::error::Elapsed),
-}
-
/// Bitcoin RPC connection
pub struct Rpc {
socket: JsonSocket,
+ cookie: CompactString,
id: u64,
}
impl Rpc {
/// Start a RPC connection
- pub async fn common(cfg: &RpcCfg) -> std::result::Result<Self, RpcConnectErr> {
- Self::new(cfg, None).await
- }
-
- /// Start a wallet RPC connection
- pub async fn wallet(cfg: &RpcCfg, wallet: &str) -> std::result::Result<Self, RpcConnectErr> {
- Self::new(cfg, Some(wallet)).await
- }
-
- async fn new(cfg: &RpcCfg, wallet: Option<&str>) -> std::result::Result<Self, RpcConnectErr> {
- let path = if let Some(wallet) = wallet {
- format_compact!("/wallet/{wallet}")
- } else {
- CompactString::const_new("/")
- };
-
+ pub async fn new(cfg: &RpcCfg) -> std::result::Result<Self, Error> {
let token = match &cfg.auth {
RpcAuth::Basic(s) => s.as_bytes().to_vec(),
RpcAuth::Cookie(path) => match std::fs::read(path) {
Ok(content) => content,
Err(e) if e.kind() == ErrorKind::IsADirectory => {
let path = PathBuf::from_str(path).unwrap().join(".cookie");
- std::fs::read(&path).map_err(|e| {
- RpcConnectErr::Cookie(path.to_string_lossy().to_string(), e.kind())
- })?
+ std::fs::read(&path)
+ .map_err(|e| Error::Cookie(path.to_string_lossy().to_string(), e.kind()))?
}
- Err(e) => return Err(RpcConnectErr::Cookie(path.clone(), e.kind())),
+ Err(e) => return Err(Error::Cookie(path.clone(), e.kind())),
},
};
// Open connection
let sock = timeout(Duration::from_secs(5), TcpStream::connect(&cfg.addr))
.await
- .map_err(|e| RpcConnectErr::Elapsed(cfg.addr, e))?
- .map_err(|e| RpcConnectErr::Tcp(cfg.addr, e.kind()))?;
+ .map_err(|e| Error::Elapsed(cfg.addr, e))?
+ .map_err(|e| Error::Tcp(cfg.addr, e.kind()))?;
sock.set_nodelay(true).ok();
Ok(Self {
id: 0,
+ cookie: format_compact!("Basic {}", BASE64_STANDARD.encode(&token)),
socket: JsonSocket {
- path,
- cookie: format_compact!("Basic {}", BASE64_STANDARD.encode(&token)),
sock,
buf: Vec::with_capacity(16 * 1024),
},
})
}
- async fn call<T>(&mut self, method: &str, params: &(impl serde::Serialize + Debug)) -> Result<T>
+ /// Set current wallet name
+ pub fn wallet(&mut self, wallet: &str) -> WalletRpc<'_> {
+ WalletRpc {
+ rpc: self,
+ path: format_compact!("/wallet/{wallet}"),
+ }
+ }
+
+ async fn call_rpc<T>(
+ &mut self,
+ path: &str,
+ method: &str,
+ params: &(impl serde::Serialize + Debug),
+ ) -> Result<T>
where
T: serde::de::DeserializeOwned + Debug,
{
trace!("RPC > {method} {params:?}");
let request = RpcRequest {
+ jsonrpc: RPC_VERSION,
method,
id: self.id,
params,
};
- let response: RpcResponse<T> = self.socket.call(method, &request).await?;
+ let response: RpcResponse<T> = self
+ .socket
+ .call(path, method, &self.cookie, &request)
+ .await?;
trace!("RPC < {response:?}");
- match response {
- RpcResponse::RpcResponse { result, error, id } => {
- assert_eq!(self.id, id);
- self.id += 1;
- if let Some(ok) = result {
- Ok(ok)
- } else {
- Err(match error {
- Some(err) => Error::RPC {
- method: method.into(),
- code: err.code,
- msg: err.message,
- },
- None => Error::Null,
- })
- }
- }
- RpcResponse::Error(msg) => Err(Error::Bitcoin(msg)),
+ let RpcResponse { result, error, id } = response;
+ assert_eq!(self.id, id);
+ self.id += 1;
+ if let Some(ok) = result {
+ Ok(ok)
+ } else {
+ Err(match error {
+ Some(err) => Error::RPC {
+ method: method.into(),
+ code: err.code,
+ msg: err.message,
+ },
+ None => Error::Null,
+ })
}
}
+}
+
+impl RpcApi for Rpc {
+ async fn call<
+ T: serde::de::DeserializeOwned + Debug + Send,
+ A: serde::Serialize + Debug + Send,
+ >(
+ &mut self,
+ method: &str,
+ params: &A,
+ ) -> Result<T> {
+ self.call_rpc("/", method, params).await
+ }
+}
+
+pub struct WalletRpc<'a> {
+ rpc: &'a mut Rpc,
+ path: CompactString,
+}
+
+impl<'a> RpcApi for WalletRpc<'a> {
+ async fn call<
+ T: serde::de::DeserializeOwned + Debug + Send,
+ A: serde::Serialize + Debug + Send,
+ >(
+ &mut self,
+ method: &str,
+ params: &A,
+ ) -> Result<T> {
+ self.rpc.call_rpc(&self.path, method, params).await
+ }
+}
+
+pub trait RpcApi {
+ async fn call<
+ T: serde::de::DeserializeOwned + Debug + Send,
+ A: serde::Serialize + Debug + Send,
+ >(
+ &mut self,
+ method: &str,
+ params: &A,
+ ) -> Result<T>;
/* ----- Wallet management ----- */
/// Create encrypted native bitcoin wallet
- pub async fn create_wallet(&mut self, name: &str, passwd: &str) -> Result<Wallet> {
+ async fn create_wallet(&mut self, name: &str, passwd: &str) -> Result<Wallet> {
self.call("createwallet", &(name, (), (), passwd, (), true))
.await
}
/// Load existing wallet
- pub async fn load_wallet(&mut self, name: &str) -> Result<Wallet> {
+ async fn load_wallet(&mut self, name: &str) -> Result<Wallet> {
match self.call("loadwallet", &[name]).await {
Err(Error::RPC {
code: ErrorCode::RpcWalletAlreadyLoaded,
@@ -305,82 +327,92 @@ impl Rpc {
}
/// Unlock loaded wallet
- pub async fn unlock_wallet(&mut self, passwd: &str) -> Result<()> {
+ async fn unlock_wallet(&mut self, passwd: &str) -> Result<()> {
expect_null(self.call("walletpassphrase", &(passwd, 100000000)).await)
}
/* ----- Wallet utils ----- */
/// Generate a new address for the current wallet
- pub async fn gen_addr(&mut self) -> Result<Address> {
+ async fn gen_addr(&mut self) -> Result<Address> {
Ok(self
- .call::<Address<NetworkUnchecked>>("getnewaddress", &EMPTY)
+ .call::<Address<NetworkUnchecked>, _>("getnewaddress", &EMPTY)
.await?
.assume_checked())
}
/// Get current balance amount
- pub async fn get_balance(&mut self) -> Result<Amount> {
+ async fn get_balance(&mut self) -> Result<Amount> {
let btc: f64 = self.call("getbalance", &EMPTY).await?;
Ok(Amount::from_btc(btc).unwrap())
}
/// Get current balance amount
- pub async fn addr_info(&mut self, addr: &Address) -> Result<AddressInfo> {
+ async fn addr_info(&mut self, addr: &Address) -> Result<AddressInfo> {
self.call("getaddressinfo", &[addr]).await
}
/* ----- Mining ----- */
/// Mine a certain amount of block to profit a given address
- pub async fn mine(&mut self, nb: u16, address: &Address) -> Result<Vec<BlockHash>> {
+ async fn mine(&mut self, nb: u32, address: &Address) -> Result<Vec<BlockHash>> {
self.call("generatetoaddress", &(nb, address)).await
}
+ /// Wait for a block height
+ async fn wait_for_block_height(&mut self, height: u32) -> Result<Nothing> {
+ self.call("waitforblockheight", &[height]).await
+ }
+
/* ----- Getter ----- */
/// Get blockchain info
- pub async fn get_blockchain_info(&mut self) -> Result<BlockchainInfo> {
+ async fn get_blockchain_info(&mut self) -> Result<BlockchainInfo> {
self.call("getblockchaininfo", &EMPTY).await
}
/// Get mempool info
- pub async fn get_mempool_info(&mut self) -> Result<MemPoolInfo> {
+ async fn get_mempool_info(&mut self) -> Result<MemPoolInfo> {
self.call("getmempoolinfo", &EMPTY).await
}
+ /// Get mempool entry
+ async fn get_mempool_entry(&mut self, id: &Txid) -> Result<MemPoolEntry> {
+ self.call("getmempoolentry", &[id]).await
+ }
+
/// Get blockchain info
- pub async fn get_wallet_info(&mut self) -> Result<WalletInfo> {
+ async fn get_wallet_info(&mut self) -> Result<WalletInfo> {
self.call("getwalletinfo", &EMPTY).await
}
/// Get chain tips
- pub async fn get_chain_tips(&mut self) -> Result<Vec<ChainTips>> {
+ async fn get_chain_tips(&mut self) -> Result<Vec<ChainTips>> {
self.call("getchaintips", &EMPTY).await
}
/// Get wallet transaction info from id
- pub async fn get_tx(&mut self, id: &Txid) -> Result<Transaction> {
+ async fn get_tx(&mut self, id: &Txid) -> Result<Transaction> {
self.call("gettransaction", &(id, (), true)).await
}
/// Get transaction inputs and outputs
- pub async fn get_input_output(&mut self, id: &Txid) -> Result<InputOutput> {
+ async fn get_input_output(&mut self, id: &Txid) -> Result<InputOutput> {
self.call("getrawtransaction", &(id, true)).await
}
/// Get genesis block hash
- pub async fn get_genesis(&mut self) -> Result<BlockHash> {
+ async fn get_genesis(&mut self) -> Result<BlockHash> {
self.call("getblockhash", &[0]).await
}
/* ----- Transactions ----- */
/// Send bitcoin transaction
- pub async fn send(
+ async fn send(
&mut self,
to: &Address,
- amount: &Amount,
+ amount: Amount,
data: Option<&[u8]>,
subtract_fee: bool,
) -> Result<Txid> {
@@ -390,9 +422,9 @@ impl Rpc {
}
/// Send bitcoin transaction with multiple recipients
- pub async fn send_many<'a>(
+ async fn send_many<'a>(
&mut self,
- to: impl IntoIterator<Item = (&'a Address, &'a Amount)>,
+ to: impl IntoIterator<Item = (&'a Address, Amount)>,
) -> Result<Txid> {
self.send_custom([], to, None, false)
.await
@@ -402,7 +434,7 @@ impl Rpc {
async fn send_custom<'a>(
&mut self,
from: impl IntoIterator<Item = &'a Txid>,
- to: impl IntoIterator<Item = (&'a Address, &'a Amount)>,
+ to: impl IntoIterator<Item = (&'a Address, Amount)>,
data: Option<&[u8]>,
subtract_fee: bool,
) -> Result<SendResult> {
@@ -445,24 +477,24 @@ impl Rpc {
}
/// Bump transaction fees of a wallet debit
- pub async fn bump_fee(&mut self, id: &Txid) -> Result<BumpResult> {
+ async fn bump_fee(&mut self, id: &Txid) -> Result<BumpResult> {
self.call("bumpfee", &[id]).await
}
/// Abandon a pending transaction
- pub async fn abandon_tx(&mut self, id: &Txid) -> Result<()> {
+ async fn abandon_tx(&mut self, id: &Txid) -> Result<()> {
expect_null(self.call("abandontransaction", &[&id]).await)
}
/* ----- Watcher ----- */
/// Block until a new block is mined
- pub async fn wait_for_new_block(&mut self) -> Result<Nothing> {
+ async fn wait_for_new_block(&mut self) -> Result<Nothing> {
self.call("waitfornewblock", &[0]).await
}
/// List new and removed transaction since a block
- pub async fn list_since_block(
+ async fn list_since_block(
&mut self,
hash: Option<&BlockHash>,
confirmation: u32,
@@ -477,20 +509,20 @@ impl Rpc {
/* ----- Cluster ----- */
/// Try a connection to a node once
- pub async fn add_node(&mut self, addr: &str) -> Result<()> {
+ async fn add_node(&mut self, addr: &str) -> Result<()> {
expect_null(self.call("addnode", &(addr, "onetry")).await)
}
/// Immediately disconnects from the specified peer node.
- pub async fn disconnect_node(&mut self, addr: &str) -> Result<()> {
+ async fn disconnect_node(&mut self, addr: &str) -> Result<()> {
expect_null(self.call("disconnectnode", &(addr, ())).await)
}
/* ----- Control ------ */
/// Request a graceful shutdown
- pub async fn stop(&mut self) -> Result<String> {
- self.call("stop", &()).await
+ fn stop(&mut self) -> impl Future<Output = Result<String>> {
+ self.call("stop", &())
}
}
@@ -506,7 +538,7 @@ pub struct BlockchainInfo {
pub verification_progress: f32,
#[serde(rename = "initialblockdownload")]
pub initial_block_download: bool,
- pub blocks: u64,
+ pub blocks: u32,
#[serde(rename = "bestblockhash")]
pub best_block_hash: BlockHash,
}
@@ -517,6 +549,11 @@ pub struct MemPoolInfo {
}
#[derive(Clone, Debug, serde::Deserialize)]
+pub struct MemPoolEntry {
+ pub height: u32,
+}
+
+#[derive(Clone, Debug, serde::Deserialize)]
#[serde(untagged)]
pub enum Scanning {
Active { duration: u64, progress: f32 },
@@ -627,6 +664,7 @@ pub struct Transaction {
pub replaced_by_txid: Option<Txid>,
pub details: Vec<TransactionDetail>,
pub decoded: InputOutput,
+ //pub height: u32,
}
#[derive(Clone, PartialEq, Eq, serde::Deserialize, Debug)]
diff --git a/depolymerizer-bitcoin/src/rpc_utils.rs b/depolymerizer-bitcoin/src/rpc_utils.rs
@@ -15,9 +15,7 @@
*/
use std::{path::PathBuf, str::FromStr};
-use bitcoin::{Address, Amount, Network};
-
-use crate::rpc::{self, Rpc, Transaction};
+use bitcoin::{Amount, Network};
/// Default chain dir <https://github.com/bitcoin/bitcoin/blob/master/doc/files.md#data-directory-location>
pub fn chain_dir(network: Network) -> &'static str {
@@ -65,18 +63,3 @@ pub fn segwit_min_amount() -> Amount {
// https://github.com/bitcoin/bitcoin/blob/master/src/policy/policy.cpp
Amount::from_sat(294)
}
-
-/// Get the first sender address from a raw transaction
-pub async fn sender_address(rpc: &mut Rpc, full: &Transaction) -> rpc::Result<Address> {
- let first = &full.decoded.vin[0];
- let tx = rpc.get_input_output(&first.txid.unwrap()).await?;
- Ok(tx
- .vout
- .into_iter()
- .find(|it| it.n == first.vout.unwrap())
- .unwrap()
- .script_pub_key
- .address
- .unwrap()
- .assume_checked())
-}
diff --git a/depolymerizer-bitcoin/src/segwit.rs b/depolymerizer-bitcoin/src/segwit.rs
@@ -63,11 +63,11 @@ pub fn encode_segwit_key(hrp: Hrp, msg: &[u8; 32]) -> [String; 2] {
#[derive(Debug, thiserror::Error)]
pub enum DecodeSegWitErr {
- #[error("There is less than 2 segwit addresses")]
+ #[error("there is less than 2 segwit addresses")]
MissingSegWitAddress,
- #[error("No addresses are sharing a common prefix")]
+ #[error("no addresses shares a common prefix")]
NoPrefixMatch,
- #[error("More than two addresses are sharing a common prefix")]
+ #[error("more than two addresses shares a common prefix")]
PrefixCollision,
#[error(transparent)]
Malformed(EddsaPublicKeyError),
diff --git a/depolymerizer-bitcoin/src/setup.rs b/depolymerizer-bitcoin/src/setup.rs
@@ -15,29 +15,24 @@
*/
use anyhow::bail;
-use taler_common::{config::Config, db::pool};
+use taler_common::config::Config;
use tokio::try_join;
use tracing::{info, warn};
use crate::{
- DB_SCHEMA,
- config::{WorkerCfg, parse_db_cfg},
- db,
+ config::WorkerCfg,
+ db::{self, pool},
payto::BtcWallet,
- rpc::{Rpc, Scanning},
+ rpc::{Rpc, RpcApi as _, Scanning},
};
pub async fn setup(cfg: &Config, reset: bool) -> anyhow::Result<()> {
info!(target: "setup", "Check bitcoind RPC connection");
let worker_cfg = WorkerCfg::parse(cfg)?;
- let mut rpc = Rpc::wallet(&worker_cfg.rpc_cfg, &worker_cfg.wallet_cfg.name).await?;
+ let mut rpc = Rpc::new(&worker_cfg.rpc_cfg).await?;
let info = rpc.get_blockchain_info().await?;
info!(target: "setup", "Running on {} chain", info.chain);
- #[cfg(feature = "fail")]
- if info.chain != "regtest" {
- anyhow::bail!("Running with random failures is unsuitable for production");
- }
let genesis_hash = rpc.get_genesis().await.unwrap();
info!(target: "setup", "Check wallet");
@@ -45,28 +40,28 @@ pub async fn setup(cfg: &Config, reset: bool) -> anyhow::Result<()> {
if let Some(password) = &worker_cfg.wallet_cfg.password {
rpc.unlock_wallet(password).await?;
}
+ let rpc = &mut rpc.wallet(&worker_cfg.wallet_cfg.name);
info!(target: "setup", "Check address");
let wallet: BtcWallet = cfg
.section("depolymerizer-bitcoin")
.parse("bitcoin wallet address", "WALLET")
.require()?;
- let addr_info = rpc.addr_info(&wallet.0).await?;
+ let addr = wallet.0;
+ let addr_info = rpc.addr_info(&addr).await?;
if !addr_info.ismine {
bail!(
- "Address {} does not belong to wallet '{}'",
- wallet.0,
+ "Address {addr} does not belong to wallet '{}'",
worker_cfg.wallet_cfg.name
);
} else if addr_info.iswatchonly {
- bail!("Address {} is watchonly", wallet.0);
+ bail!("Address {addr} is watchonly");
} else if !addr_info.solvable {
- bail!("Address {} is not solvable", wallet.0);
+ bail!("Address {addr} is not solvable");
}
info!(target: "setup", "Setup database state");
- let db_cfg = parse_db_cfg(cfg)?;
- let pool = pool(db_cfg.cfg, DB_SCHEMA).await?;
+ let pool = pool(cfg).await?;
// Init status to true
try_join!(
diff --git a/depolymerizer-bitcoin/src/sql.rs b/depolymerizer-bitcoin/src/sql.rs
@@ -14,7 +14,7 @@
TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
*/
-use bitcoin::{Address, Amount as BtcAmount, Txid, address::NetworkUnchecked, hashes::Hash};
+use bitcoin::{Address, Amount as BtcAmount, address::NetworkUnchecked};
use sqlx::{Row, postgres::PgRow};
use taler_api::db::TypeHelper;
use taler_common::types::{
@@ -37,11 +37,6 @@ pub fn sql_addr(row: &PgRow, idx: usize) -> sqlx::Result<Address> {
.assume_checked())
}
-/// Bitcoin transaction id from sql
-pub fn sql_txid(row: &PgRow, idx: usize) -> sqlx::Result<Txid> {
- row.try_get_map(idx, Txid::from_slice)
-}
-
pub fn sql_payto<I: sqlx::ColumnIndex<PgRow>>(
r: &PgRow,
addr: I,
diff --git a/depolymerizer-common/src/status.rs b/depolymerizer-common/src/status.rs
@@ -15,8 +15,11 @@
*/
//! Transactions status in database
+use taler_common::api::wire::TransferState;
+
/// Debit transaction status
/// -> Requested API request
+/// Requested -> Ignored Invalid address
/// Requested -> Sent Announced to the bitcoin network
/// Sent -> Confirmed Mined in an N old block
/// Confirmed -> Requested Conflicting transaction (reorg)
@@ -26,12 +29,24 @@
pub enum DebitStatus {
/// Debit have been requested (default)
requested,
+ /// Debit will never be sent (e.g: address does not work for this network)
+ ignored,
/// Debit have been announced to the bitcoin network
sent,
/// Debit have been mined in an N old block
confirmed,
}
+impl Into<TransferState> for DebitStatus {
+ fn into(self) -> TransferState {
+ match self {
+ DebitStatus::requested | DebitStatus::sent => TransferState::pending,
+ DebitStatus::confirmed => TransferState::success,
+ DebitStatus::ignored => TransferState::permanent_failure,
+ }
+ }
+}
+
/// Bounce transaction status
/// -> Requested Credit in wrong format
/// Requested -> Ignored Insufficient found
@@ -40,7 +55,7 @@ pub enum DebitStatus {
/// Confirmed -> Requested Conflicting transaction (reorg)
#[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type)]
#[allow(non_camel_case_types)]
-#[sqlx(type_name = "bounce_status")]
+#[sqlx(type_name = "debit_status")]
pub enum BounceStatus {
/// Bounce have been requested (default)
requested,
diff --git a/makefile b/makefile
@@ -15,8 +15,8 @@ all: build
build:
cargo build --release
-.PHONY: install-nobuild-files
-install-nobuild-files:
+.PHONY: install-files
+install-files:
install -m 644 -D -t $(share_dir)/depolymerizer-bitcoin/config.d depolymerizer-bitcoin/depolymerizer-bitcoin.conf
install -m 644 -D -t $(share_dir)/depolymerizer-bitcoin/sql depolymerizer-common/db/versioning.sql
install -m 644 -D -t $(share_dir)/depolymerizer-bitcoin/sql depolymerizer-bitcoin/db/depolymerizer-bitcoin*.sql
@@ -25,16 +25,15 @@ install-nobuild-files:
install -D -t $(bin_dir) contrib/depolymerizer-bitcoin-dbconfig
.PHONY: install
-install: build install-nobuild-files
+install: build install-files
install -D -t $(bin_dir) target/release/depolymerizer-bitcoin
.PHONY: check
-check: install-nobuild-files
- cargo clippy --all-targets
+check: install-files
cargo test
.PHONY: test
-test: install-nobuild-files
+test: install-files
RUST_BACKTRACE=true cargo run --profile dev --bin testbench -- instrumentation
.PHONY: doc
@@ -56,4 +55,7 @@ fmt:
.PHONY: coverage-cyclos
coverage:
- cargo llvm-cov test
-\ No newline at end of file
+ cargo llvm-cov clean --workspace
+ cargo llvm-cov test --no-clean
+ cargo llvm-cov run --bin btc-harness --no-clean -- logic
+ cargo llvm-cov report --html
+\ No newline at end of file
diff --git a/testbench/Cargo.toml b/testbench/Cargo.toml
@@ -1,31 +0,0 @@
-[package]
-name = "testbench"
-version = "0.1.0"
-edition.workspace = true
-authors.workspace = true
-homepage.workspace = true
-repository.workspace = true
-license-file.workspace = true
-
-[dependencies]
-# Cli args parser
-clap.workspace = true
-# Bitcoin
-depolymerizer-bitcoin = { path = "../depolymerizer-bitcoin" }
-bitcoin.workspace = true
-# Wire Gateway
-ureq = { version = "3.0.0", features = ["json"] }
-# terminal color
-owo-colors = "4.0.0"
-# Edit toml files
-rust-ini = "0.21.0"
-# Progress reporting
-indicatif = "0.18.0"
-taler-common.workspace = true
-taler-test-utils.workspace = true
-tokio.workspace = true
-anyhow.workspace = true
-tracing-subscriber.workspace = true
-tracing.workspace = true
-sqlx.workspace = true
-url.workspace = true
diff --git a/testbench/conf/bitcoin.conf b/testbench/conf/bitcoin.conf
@@ -1,13 +0,0 @@
-regtest=1
-txindex=1
-maxtxfee=0.01
-fallbackfee=0.00000001
-rpcservertimeout=0
-dbcache=4
-maxmempool=5
-par=2
-rpcthreads=5
-
-[regtest]
-port=8345
-rpcport=18345
-\ No newline at end of file
diff --git a/testbench/conf/bitcoindev.conf b/testbench/conf/bitcoindev.conf
@@ -1,4 +0,0 @@
-regtest=1
-txindex=1
-rpcservertimeout=0
-rpcport=18345
-\ No newline at end of file
diff --git a/testbench/conf/taler_btc.conf b/testbench/conf/taler_btc.conf
@@ -1,14 +0,0 @@
-[depolymerizer-bitcoin]
-CURRENCY = DEVBTC
-
-[depolymerizer-bitcoin-worker]
-BOUNCE_FEE = DEVBTC:0.00001
-CONFIRMATION = 3
-
-[depolymerizer-bitcoin-httpd-wire-gateway-api]
-ENABLED = YES
-AUTH_METHOD = none
-
-[depolymerizer-bitcoin-httpd-revenue-api]
-ENABLED = YES
-AUTH_METHOD = none
diff --git a/testbench/conf/taler_btc_bump.conf b/testbench/conf/taler_btc_bump.conf
@@ -1,15 +0,0 @@
-[depolymerizer-bitcoin]
-CURRENCY = DEVBTC
-
-[depolymerizer-bitcoin-worker]
-BOUNCE_FEE = DEVBTC:0.00001
-CONFIRMATION = 3
-BUMP_DELAY = 5
-
-[depolymerizer-bitcoin-httpd-wire-gateway-api]
-ENABLED = YES
-AUTH_METHOD = none
-
-[depolymerizer-bitcoin-httpd-revenue-api]
-ENABLED = YES
-AUTH_METHOD = none
-\ No newline at end of file
diff --git a/testbench/conf/taler_btc_dev.conf b/testbench/conf/taler_btc_dev.conf
@@ -1,17 +0,0 @@
-[depolymerizer-bitcoin]
-CURRENCY = DEVBTC
-NAME = Test
-
-[depolymerizer-bitcoin-worker]
-WALLET_NAME = wire
-RPC_COOKIE_FILE = testbenches/btc-regtest/bitcoin/testnet4/.cookie
-RPC_BIND = 127.0.0.1:18345
-CONFIRMATION = 3
-
-[depolymerizer-bitcoin-httpd-wire-gateway-api]
-ENABLED = YES
-AUTH_METHOD = none
-
-[depolymerizer-bitcoin-httpd-revenue-api]
-ENABLED = YES
-AUTH_METHOD = none
diff --git a/testbench/conf/taler_btc_lifetime.conf b/testbench/conf/taler_btc_lifetime.conf
@@ -1,18 +0,0 @@
-[depolymerizer-bitcoin]
-CURRENCY = DEVBTC
-
-[depolymerizer-bitcoin-worker]
-BOUNCE_FEE = DEVBTC:0.00001
-CONFIRMATION = 3
-LIFETIME = 10
-
-[depolymerizer-bitcoin-httpd]
-LIFETIME = 10
-
-[depolymerizer-bitcoin-httpd-wire-gateway-api]
-ENABLED = YES
-AUTH_METHOD = none
-
-[depolymerizer-bitcoin-httpd-revenue-api]
-ENABLED = YES
-AUTH_METHOD = none
diff --git a/testbench/conf/taler_eth.conf b/testbench/conf/taler_eth.conf
@@ -1,14 +0,0 @@
-[depolymerizer-ethereum]
-CURRENCY = DEVETH
-
-[depolymerizer-ethereum-worker]
-BOUNCE_FEE = DEVETH:0.00001
-CONFIRMATION = 3
-
-[depolymerizer-ethereum-httpd-wire-gateway-api]
-ENABLED = YES
-AUTH_METHOD = none
-
-[depolymerizer-ethereum-httpd-revenue-api]
-ENABLED = YES
-AUTH_METHOD = none
diff --git a/testbench/conf/taler_eth_bump.conf b/testbench/conf/taler_eth_bump.conf
@@ -1,15 +0,0 @@
-[depolymerizer-ethereum]
-CURRENCY = DEVETH
-
-[depolymerizer-ethereum-worker]
-BOUNCE_FEE = DEVETH:0.00001
-CONFIRMATION = 3
-BUMP_DELAY = 5
-
-[depolymerizer-ethereum-httpd-wire-gateway-api]
-ENABLED = YES
-AUTH_METHOD = none
-
-[depolymerizer-ethereum-httpd-revenue-api]
-ENABLED = YES
-AUTH_METHOD = none
-\ No newline at end of file
diff --git a/testbench/conf/taler_eth_lifetime.conf b/testbench/conf/taler_eth_lifetime.conf
@@ -1,18 +0,0 @@
-[depolymerizer-ethereum]
-CURRENCY = DEVETH
-
-[depolymerizer-ethereum-worker]
-BOUNCE_FEE = DEVETH:0.00001
-CONFIRMATION = 3
-LIFETIME = 10
-
-[depolymerizer-ethereum-httpd]
-LIFETIME = 10
-
-[depolymerizer-ethereum-httpd-wire-gateway-api]
-ENABLED = YES
-AUTH_METHOD = none
-
-[depolymerizer-ethereum-httpd-revenue-api]
-ENABLED = YES
-AUTH_METHOD = none
-\ No newline at end of file
diff --git a/testbench/src/btc.rs b/testbench/src/btc.rs
@@ -1,1102 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2022-2025, 2026 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU Affero General Public License as published by the Free Software
- Foundation; either version 3, or (at your option) any later version.
-
- TALER is distributed in the hope that it will be useful, but WITHOUT ANY
- WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
- A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
-
- You should have received a copy of the GNU Affero General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
-*/
-
-use std::{
- net::{IpAddr, Ipv4Addr, SocketAddr},
- ops::{Deref, DerefMut},
- path::Path,
- time::Duration,
-};
-
-use bitcoin::{Address, Amount};
-use depolymerizer_bitcoin::{
- CONFIG_SOURCE,
- config::{RpcAuth, RpcCfg, ServeCfg, WorkerCfg},
- payto::BtcWallet,
- rpc::{Category, Error, ErrorCode, Rpc},
- rpc_utils::segwit_min_amount,
- taler_utils::btc_to_taler,
-};
-use ini::Ini;
-use taler_common::{
- api::{EddsaPublicKey, ShortHashCode},
- config::Config,
- types::{
- base32::Base32,
- payto::{Payto, PaytoImpl},
- },
-};
-
-use crate::{
- retry, retry_opt,
- utils::{ChildGuard, TalerCtx, TestCtx, cmd_redirect, patch_config, transfer, unused_port},
-};
-
-pub struct BtcCtx {
- btc_node: ChildGuard,
- _btc_node2: ChildGuard,
- common_rpc: Rpc,
- common_rpc2: Rpc,
- wire_rpc: Rpc,
- client_rpc: Rpc,
- reserve_rpc: Rpc,
- wire_addr: Address,
- pub client_addr: Address,
- reserve_addr: Address,
- worker_cfg: WorkerCfg,
- serve_cfg: ServeCfg,
- conf: u16,
- ctx: TalerCtx,
- node2_addr: String,
-}
-
-impl Deref for BtcCtx {
- type Target = TalerCtx;
-
- fn deref(&self) -> &Self::Target {
- &self.ctx
- }
-}
-
-impl DerefMut for BtcCtx {
- fn deref_mut(&mut self) -> &mut Self::Target {
- &mut self.ctx
- }
-}
-
-impl BtcCtx {
- pub async fn config(
- ctx: &TestCtx,
- btc_patch: impl FnOnce(&mut Ini, &Path),
- cfg_patch: impl FnOnce(&mut Ini, &Path),
- ) {
- // Patch configs
- let port = unused_port();
- let rpc_port = unused_port();
-
- patch_config(
- "testbench/conf/bitcoin.conf",
- ctx.dir.join("bitcoin.conf"),
- |cfg| {
- cfg.with_section(Some("regtest"))
- .set("bind", format!("127.0.0.1:{port}"))
- .set("rpcport", format!("{rpc_port}"));
- btc_patch(cfg, &ctx.dir)
- },
- );
- patch_config(
- "testbench/conf/taler_btc.conf",
- ctx.dir.join("config.conf"),
- |cfg| {
- cfg.with_section(Some("depolymerizer-bitcoin-worker"))
- .set("RPC_BIND", format!("127.0.0.1:{rpc_port}"));
- cfg_patch(cfg, &ctx.dir)
- },
- );
- // Load config
- let cfg = Config::load(CONFIG_SOURCE, Some(ctx.dir.join("config.conf"))).unwrap();
- let rpc_cfg = RpcCfg::parse(&cfg).unwrap();
- // Start bitcoin nodes
- let _node = cmd_redirect(
- "bitcoind",
- &[&format!("-datadir={}", ctx.dir.to_string_lossy())],
- ctx.log("bitcoind"),
- );
- // Connect
- retry_opt!(async {
- let mut client = Rpc::common(&rpc_cfg).await?;
- client.get_blockchain_info().await?;
- Ok::<_, anyhow::Error>(())
- });
- }
-
- fn patch_btc_config(from: impl AsRef<Path>, to: impl AsRef<Path>, port: u16, rpc_port: u16) {
- patch_config(from, to, |cfg| {
- cfg.with_section(Some("regtest"))
- .set("bind", format!("127.0.0.1:{port}"))
- .set("rpcport", format!("{rpc_port}"));
- })
- }
-
- pub async fn setup(ctx: &TestCtx, config: &str, stressed: bool) -> Self {
- let mut ctx = TalerCtx::new(ctx, "depolymerizer-bitcoin", config, stressed);
-
- // Choose unused port
- let btc_port = unused_port();
- let btc_rpc_port = unused_port();
- let btc2_port = unused_port();
- let btc2_rpc_port = unused_port();
-
- // Bitcoin config
- Self::patch_btc_config(
- "testbench/conf/bitcoin.conf",
- ctx.wire_dir.join("bitcoin.conf"),
- btc_port,
- btc_rpc_port,
- );
- Self::patch_btc_config(
- "testbench/conf/bitcoin.conf",
- ctx.wire2_dir.join("bitcoin.conf"),
- btc2_port,
- btc2_rpc_port,
- );
- patch_config(&ctx.conf, &ctx.conf, |cfg| {
- cfg.with_section(Some("depolymerizer-bitcoin-worker"))
- .set("RPC_BIND", format!("127.0.0.1:{btc_rpc_port}"))
- .set("WALLET_NAME", "wire")
- .set(
- "RPC_COOKIE_FILE",
- ctx.wire_dir.join("regtest/.cookie").to_string_lossy(),
- );
- });
-
- // Load config
- let config = Config::load(CONFIG_SOURCE, Some(&ctx.conf)).unwrap();
- let cfg = WorkerCfg::parse(&config).unwrap();
- // Start bitcoin nodes
- let btc_node = cmd_redirect(
- "bitcoind",
- &[&format!("-datadir={}", ctx.wire_dir.to_string_lossy())],
- ctx.log("bitcoind"),
- );
- let _btc_node2 = cmd_redirect(
- "bitcoind",
- &[&format!("-datadir={}", ctx.wire2_dir.to_string_lossy())],
- ctx.log("bitcoind2"),
- );
-
- // Setup wallets
- let mut common_rpc = retry_opt!(Rpc::common(&cfg.rpc_cfg));
- retry_opt!(common_rpc.get_blockchain_info());
- let node2_addr = format!("127.0.0.1:{btc2_port}");
- common_rpc.add_node(&node2_addr).await.unwrap();
-
- for name in ["wire", "client", "reserve"] {
- if let Err(e) = common_rpc.load_wallet(name).await {
- if let Error::RPC {
- code: ErrorCode::RpcWalletNotFound,
- ..
- } = e
- {
- common_rpc.create_wallet(name, "").await.unwrap();
- } else {
- break;
- }
- }
- }
- let common_rpc2 = retry_opt!(Rpc::common(&RpcCfg {
- addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), btc2_rpc_port),
- auth: RpcAuth::Cookie(
- ctx.wire2_dir
- .join("regtest/.cookie")
- .to_string_lossy()
- .to_string(),
- ),
- }));
-
- // Generate money
- let mut reserve_rpc = Rpc::wallet(&cfg.rpc_cfg, "reserve").await.unwrap();
- let mut client_rpc = Rpc::wallet(&cfg.rpc_cfg, "client").await.unwrap();
- let mut wire_rpc = Rpc::wallet(&cfg.rpc_cfg, "wire").await.unwrap();
- let reserve_addr = reserve_rpc.gen_addr().await.unwrap();
- let client_addr = client_rpc.gen_addr().await.unwrap();
- let wire_addr = wire_rpc.gen_addr().await.unwrap();
- common_rpc.mine(101, &reserve_addr).await.unwrap();
- reserve_rpc
- .send(&client_addr, &(Amount::ONE_BTC * 10), None, false)
- .await
- .unwrap();
- common_rpc.mine(1, &reserve_addr).await.unwrap();
-
- patch_config(&ctx.conf, &ctx.conf, |cfg| {
- cfg.with_section(Some("depolymerizer-bitcoin"))
- .set("NAME", "Exchange Owner")
- .set("WALLET", wire_addr.to_string());
- });
-
- let config = Config::load(CONFIG_SOURCE, Some(&ctx.conf)).unwrap();
- let serve_cfg = ServeCfg::parse(&config).unwrap();
-
- // Setup & run
- ctx.dbinit();
- ctx.setup();
- ctx.run().await;
-
- Self {
- ctx,
- btc_node,
- common_rpc,
- wire_rpc,
- client_rpc,
- reserve_rpc,
- wire_addr,
- client_addr,
- reserve_addr,
- conf: cfg.conf as u16,
- worker_cfg: cfg,
- serve_cfg,
- _btc_node2,
- common_rpc2,
- node2_addr,
- }
- }
-
- pub fn reset_db(&mut self) {
- self.ctx.reset_db();
- self.ctx.setup();
- }
-
- pub async fn stop_node(&mut self) {
- // We need to kill bitcoin gracefully to avoid corruption
- self.common_rpc.stop().await.unwrap();
- self.btc_node.0.wait().unwrap();
- }
-
- pub async fn cluster_deco(&mut self) {
- self.common_rpc
- .disconnect_node(&self.node2_addr)
- .await
- .unwrap();
- }
-
- pub async fn cluster_fork(&mut self) -> u16 {
- let node1_height = self.common_rpc.get_blockchain_info().await.unwrap().blocks;
- let node2_height = self.common_rpc2.get_blockchain_info().await.unwrap().blocks;
- let diff = node1_height - node2_height;
- self.common_rpc2
- .mine(diff as u16 + 1, &self.reserve_addr)
- .await
- .unwrap();
- self.common_rpc.add_node(&self.node2_addr).await.unwrap();
- diff as u16 + 2
- }
-
- pub async fn restart_node(&mut self, additional_args: &[&str]) {
- self.stop_node().await;
- self.resume_node(additional_args).await;
- }
-
- pub async fn resume_node(&mut self, additional_args: &[&str]) {
- let datadir = format!("-datadir={}", self.ctx.wire_dir.to_string_lossy());
- let mut args = vec![datadir.as_str()];
- args.extend_from_slice(additional_args);
- self.btc_node = cmd_redirect("bitcoind", &args, self.ctx.log("bitcoind"));
- self.common_rpc = retry_opt!(Rpc::common(&self.worker_cfg.rpc_cfg));
- self.common_rpc.add_node(&self.node2_addr).await.unwrap();
- for name in ["client", "reserve", "wire"] {
- self.common_rpc.load_wallet(name).await.ok();
- }
-
- self.reserve_rpc = Rpc::wallet(&self.worker_cfg.rpc_cfg, "reserve")
- .await
- .unwrap();
- self.client_rpc = Rpc::wallet(&self.worker_cfg.rpc_cfg, "client")
- .await
- .unwrap();
- self.wire_rpc = Rpc::wallet(&self.worker_cfg.rpc_cfg, "wire").await.unwrap();
- tokio::time::sleep(Duration::from_millis(100)).await;
- }
-
- /* ----- Transaction ------ */
-
- pub async fn credit(&mut self, amount: Amount, metadata: &EddsaPublicKey) {
- while let Err(e) = self
- .client_rpc
- .send_segwit_key(&self.wire_addr, &amount, metadata)
- .await
- {
- match e {
- Error::RPC {
- code: ErrorCode::RpcWalletError,
- ..
- } => {
- self.mine(1).await;
- }
- _ => panic!("{e:?}"),
- }
- }
- }
-
- pub async fn debit(&mut self, amount: Amount, metadata: &ShortHashCode) {
- transfer(
- &self.ctx.gateway_url,
- metadata,
- Payto::new(BtcWallet(self.client_addr.clone())).as_full_uri("name"),
- &btc_to_taler(&amount.to_signed().unwrap(), &self.worker_cfg.currency),
- )
- .await
- }
-
- pub async fn malformed_credit(&mut self, amount: &Amount) {
- self.client_rpc
- .send(&self.wire_addr, amount, None, false)
- .await
- .unwrap();
- }
-
- pub async fn reset_wallet(&mut self) {
- let amount = self.wire_balance().await;
- self.wire_rpc
- .send(&self.client_addr, &amount, None, true)
- .await
- .unwrap();
- self.next_block().await;
- }
-
- async fn abandon(rpc: &mut Rpc) {
- let list = rpc.list_since_block(None, 1).await.unwrap();
- for tx in list.transactions {
- if tx.category == Category::Send && tx.confirmations == 0 {
- rpc.abandon_tx(&tx.txid).await.unwrap();
- }
- }
- }
-
- pub async fn abandon_wire(&mut self) {
- Self::abandon(&mut self.wire_rpc).await;
- }
-
- pub async fn abandon_client(&mut self) {
- Self::abandon(&mut self.client_rpc).await;
- }
-
- /* ----- Mining ----- */
-
- async fn mine(&mut self, nb: u16) {
- self.common_rpc.mine(nb, &self.reserve_addr).await.unwrap();
- }
-
- pub async fn next_conf(&mut self) {
- self.mine(self.conf).await
- }
-
- pub async fn next_block(&mut self) {
- self.mine(1).await
- }
-
- pub async fn mine_pending(&mut self) {
- while self.common_rpc.get_mempool_info().await.unwrap().size > 0 {
- self.mine(1).await
- }
- }
-
- pub async fn mine_conf(&mut self) {
- self.mine_pending().await;
- self.next_conf().await;
- }
-
- /* ----- Balances ----- */
-
- pub async fn client_balance(&mut self) -> Amount {
- self.client_rpc.get_balance().await.unwrap()
- }
-
- pub async fn wire_balance(&mut self) -> Amount {
- self.wire_rpc.get_balance().await.unwrap()
- }
-
- pub async fn expect_c_balance(&mut self, balance: Amount, mine: bool) {
- retry!(async {
- let check = balance == self.client_balance().await;
- if !check && mine {
- self.mine_conf().await;
- }
- check
- });
- }
-
- pub async fn expect_w_balance(&mut self, balance: Amount, mine: bool) {
- retry!(async {
- let check = balance == self.wire_balance().await;
- if !check && mine {
- self.mine_conf().await;
- }
- check
- });
- }
-
- pub async fn expect_w_balance_less(&mut self, balance: Amount, mine: bool) {
- retry!(async {
- let check = self.wire_balance().await < balance;
- if !check && mine {
- self.mine_conf().await;
- }
- check
- });
- }
-
- /* ----- Wire Gateway ----- */
-
- pub async fn expect_credits(&mut self, txs: &[(EddsaPublicKey, Amount)], mine: bool) {
- let txs: Vec<_> = txs
- .iter()
- .map(|(metadata, amount)| {
- (
- metadata.clone(),
- btc_to_taler(&amount.to_signed().unwrap(), &self.worker_cfg.currency),
- )
- })
- .collect();
- retry!(async {
- let check = self.ctx.expect_credits(&txs).await;
- if !check && mine {
- self.mine_conf().await;
- }
- check
- });
- }
-
- pub async fn expect_debits(&mut self, txs: &[(ShortHashCode, Amount)], mine: bool) {
- let txs: Vec<_> = txs
- .iter()
- .map(|(metadata, amount)| {
- (
- metadata.clone(),
- btc_to_taler(&amount.to_signed().unwrap(), &self.worker_cfg.currency),
- )
- })
- .collect();
- retry!(async {
- let check = self.ctx.expect_debits(&txs).await;
- if !check && mine {
- self.mine_conf().await;
- }
- check
- });
- }
-}
-
-/// Test btc-wire correctly receive and send transactions on the blockchain
-pub async fn wire(ctx: TestCtx) {
- ctx.step("Setup");
- let mut ctx = BtcCtx::setup(&ctx, "taler_btc.conf", false).await;
-
- ctx.step("Credit");
- {
- // Send transactions
- let mut balance = ctx.wire_balance().await;
- let mut txs = Vec::new();
- for n in 10..100 {
- let metadata = EddsaPublicKey::rand();
- let amount = Amount::from_sat(n * 1000);
- ctx.credit(amount, &metadata).await;
- txs.push((metadata, amount));
- balance += amount;
- }
- ctx.expect_credits(&txs, true).await;
- ctx.expect_w_balance(balance, false).await;
- };
-
- ctx.step("Debit");
- {
- let mut balance = ctx.client_balance().await;
- let mut txs = Vec::new();
- for n in 10..100 {
- let metadata = Base32::rand();
- let amount = Amount::from_sat(n * 100);
- balance += amount;
- ctx.debit(amount, &metadata).await;
- txs.push((metadata, amount));
- }
- ctx.expect_debits(&txs, true).await;
- ctx.expect_c_balance(balance, false).await;
- }
-
- ctx.step("Bounce");
- {
- ctx.reset_wallet().await;
- let mut balance = ctx.wire_balance().await;
- for n in 10..40 {
- ctx.malformed_credit(&Amount::from_sat(n * 1000)).await;
- balance += ctx.worker_cfg.bounce_fee;
- }
- ctx.expect_w_balance(balance, true).await;
- }
-}
-
-/// Check btc-wire and wire-gateway correctly stop when a lifetime limit is configured
-pub async fn lifetime(ctx: TestCtx) {
- ctx.step("Setup");
- let mut ctx = BtcCtx::setup(&ctx, "taler_btc_lifetime.conf", false).await;
- ctx.step("Check lifetime");
- // Start up
- retry!(async { ctx.wire_running() && ctx.gateway_running() });
- // Consume wire lifetime
- for _ in 0..ctx.worker_cfg.lifetime.unwrap() {
- ctx.mine(1).await;
- }
- retry!(async {
- let check = !ctx.wire_running() && ctx.gateway_running();
- if !check {
- ctx.mine(1).await;
- }
- check
- });
- // Consume gateway lifetime
- for _ in 0..=ctx.serve_cfg.lifetime.unwrap() {
- ctx.debit(segwit_min_amount(), &Base32::rand()).await;
- }
- // End down
- retry!(async { !ctx.wire_running() && !ctx.gateway_running() });
-}
-
-/// Check the capacity of wire-gateway and btc-wire to recover from database and node loss
-pub async fn reconnect(ctx: TestCtx) {
- // TODO check recover metadata
- ctx.step("Setup");
- let mut ctx = BtcCtx::setup(&ctx, "taler_btc.conf", false).await;
-
- let mut credits = Vec::new();
- let mut debits = Vec::new();
-
- ctx.step("With DB");
- {
- let metadata = EddsaPublicKey::rand();
- let amount = Amount::from_sat(42000);
- ctx.credit(amount, &metadata).await;
- credits.push((metadata, amount));
- ctx.mine_conf().await;
- ctx.expect_credits(&credits, false).await;
- };
-
- ctx.step("Without DB");
- {
- ctx.stop_db();
- ctx.malformed_credit(&Amount::from_sat(24000)).await;
- let metadata = EddsaPublicKey::rand();
- let amount = Amount::from_sat(40000);
- ctx.credit(amount, &metadata).await;
- credits.push((metadata, amount));
- ctx.stop_node().await;
- ctx.expect_gateway_down().await;
- }
-
- ctx.step("Reconnect DB");
- {
- ctx.resume_db();
- ctx.resume_node(&[]).await;
- let metadata = Base32::rand();
- let amount = Amount::from_sat(2000);
- ctx.debit(amount, &metadata).await;
- debits.push((metadata, amount));
- ctx.expect_debits(&debits, true).await;
- ctx.expect_credits(&credits, false).await;
- }
-
- ctx.step("Recover DB");
- {
- let balance = ctx.wire_balance().await;
- ctx.reset_db();
- ctx.expect_debits(&debits, false).await;
- ctx.expect_credits(&credits, false).await;
- ctx.expect_w_balance(balance, false).await;
- }
-}
-
-/// Test btc-wire ability to recover from errors in correctness critical paths and prevent concurrent sending
-pub async fn stress(ctx: TestCtx) {
- ctx.step("Setup");
- let mut ctx = BtcCtx::setup(&ctx, "taler_btc.conf", true).await;
-
- let mut credits = Vec::new();
- let mut debits = Vec::new();
-
- ctx.step("Credit");
- {
- let mut balance = ctx.wire_balance().await;
- for n in 10..30 {
- let metadata = EddsaPublicKey::rand();
- let amount = Amount::from_sat(n * 10000);
- ctx.credit(amount, &metadata).await;
- credits.push((metadata, amount));
- balance += amount;
- }
- ctx.expect_credits(&credits, true).await;
- ctx.expect_w_balance(balance, false).await;
- };
-
- ctx.step("Debit");
- {
- let mut balance = ctx.client_balance().await;
- for n in 10..30 {
- let metadata = Base32::rand();
- let amount = Amount::from_sat(n * 100);
- balance += amount;
- ctx.debit(amount, &metadata).await;
- debits.push((metadata, amount));
- }
- ctx.expect_debits(&debits, true).await;
- ctx.expect_c_balance(balance, false).await;
- }
-
- ctx.step("Bounce");
- {
- ctx.reset_wallet().await;
- let mut balance = ctx.wire_balance().await;
- for n in 10..30 {
- ctx.malformed_credit(&Amount::from_sat(n * 1000)).await;
- balance += ctx.worker_cfg.bounce_fee;
- }
- ctx.expect_w_balance(balance, true).await;
- }
-
- ctx.step("Recover DB");
- {
- let balance = ctx.wire_balance().await;
- ctx.reset_db();
- ctx.expect_debits(&debits, false).await;
- ctx.expect_credits(&credits, false).await;
- ctx.expect_w_balance(balance, false).await;
- }
-}
-
-/// Test btc-wire ability to handle conflicting outgoing transactions
-pub async fn conflict(tctx: TestCtx) {
- tctx.step("Setup");
- let mut ctx = BtcCtx::setup(&tctx, "taler_btc.conf", false).await;
-
- ctx.step("Conflict send");
- {
- // Perform credit
- let amount = Amount::from_sat(4200000);
- ctx.credit(amount, &EddsaPublicKey::rand()).await;
- ctx.next_conf().await;
- ctx.expect_w_balance(amount, false).await;
- let client = ctx.client_balance().await;
- let wire = ctx.wire_balance().await;
-
- // Perform debit
- ctx.debit(Amount::from_sat(400000), &Base32::rand()).await;
- ctx.expect_w_balance_less(wire, false).await;
-
- // Abandon pending transaction
- ctx.restart_node(&["-minrelaytxfee=0.0001"]).await;
- ctx.abandon_wire().await;
- ctx.expect_c_balance(client, false).await;
- ctx.expect_w_balance(wire, false).await;
-
- // Generate conflict
- ctx.debit(Amount::from_sat(500000), &Base32::rand()).await;
- ctx.expect_w_balance_less(wire, false).await;
-
- // Resend conflicting transaction
- let wire = ctx.wire_balance().await;
- ctx.restart_node(&[]).await;
- ctx.expect_w_balance_less(wire, true).await;
- }
-
- ctx.step("Setup");
- drop(ctx);
- let mut ctx = BtcCtx::setup(&tctx, "taler_btc.conf", false).await;
- ctx.credit(Amount::from_sat(3000000), &EddsaPublicKey::rand())
- .await;
- ctx.next_block().await;
-
- ctx.step("Conflict bounce");
- {
- // Perform bounce
- let wire = ctx.wire_balance().await;
- let bounce_amount = Amount::from_sat(4000000);
- ctx.malformed_credit(&bounce_amount).await;
- ctx.next_conf().await;
- let fee = ctx.worker_cfg.bounce_fee;
- ctx.expect_w_balance(wire + fee, true).await;
-
- // Abandon pending transaction
- ctx.restart_node(&["-minrelaytxfee=0.0001"]).await;
- ctx.abandon_wire().await;
- ctx.expect_w_balance(wire + bounce_amount, false).await;
-
- // Generate conflict
- ctx.debit(Amount::from_sat(50000), &Base32::rand()).await;
- ctx.expect_w_balance_less(wire + bounce_amount, false).await;
-
- // Resend conflicting transaction
- let wire = ctx.wire_balance().await;
- ctx.restart_node(&[]).await;
- ctx.expect_w_balance_less(wire, true).await;
- }
-}
-
-/// Test btc-wire correctness when a blockchain reorganization occurs
-pub async fn reorg(ctx: TestCtx) {
- ctx.step("Setup");
- let mut ctx = BtcCtx::setup(&ctx, "taler_btc.conf", false).await;
-
- ctx.step("Handle reorg incoming transactions");
- {
- // Loose second bitcoin node
- ctx.cluster_deco().await;
-
- // Perform credits
- let before = ctx.wire_balance().await;
- for n in 10..21 {
- ctx.credit(Amount::from_sat(n * 10000), &EddsaPublicKey::rand())
- .await;
- }
- ctx.mine_conf().await;
- let after = ctx.wire_balance().await;
-
- // Perform fork and check btc-wire hard error
- ctx.expect_gateway_up().await;
- let fork = ctx.cluster_fork().await;
- ctx.expect_w_balance(before, false).await;
- ctx.expect_gateway_down().await;
-
- // Recover orphaned transaction
- ctx.mine(fork).await;
- ctx.expect_w_balance(after, false).await;
- ctx.expect_gateway_up().await;
- }
-
- ctx.step("Handle reorg outgoing transactions");
- {
- // Loose second bitcoin node
- ctx.cluster_deco().await;
-
- // Perform debits
- let before = ctx.client_balance().await;
- let mut after = ctx.client_balance().await;
- for n in 10..21 {
- let amount = Amount::from_sat(n * 100);
- ctx.debit(amount, &Base32::rand()).await;
- after += amount;
- }
- ctx.expect_c_balance(after, true).await;
-
- // Perform fork and check btc-wire still up
- ctx.expect_gateway_up().await;
- ctx.cluster_fork().await;
- ctx.expect_c_balance(before, false).await;
- ctx.expect_gateway_up().await;
-
- // Recover orphaned transaction
- ctx.next_conf().await;
- ctx.expect_c_balance(after, false).await;
- }
-
- ctx.step("Handle reorg bounce");
- {
- ctx.reset_wallet().await;
-
- // Loose second bitcoin node
- ctx.cluster_deco().await;
-
- // Perform bounce
- let before = ctx.wire_balance().await;
- let mut after = ctx.wire_balance().await;
- for n in 10..21 {
- ctx.malformed_credit(&Amount::from_sat(n * 1000)).await;
- after += ctx.worker_cfg.bounce_fee;
- }
- ctx.expect_w_balance(after, true).await;
-
- // Perform fork and check btc-wire hard error
- ctx.expect_gateway_up().await;
- let fork = ctx.cluster_fork().await;
- ctx.expect_w_balance(before, false).await;
- ctx.expect_gateway_down().await;
-
- // Recover orphaned transaction
- ctx.mine(fork).await;
- ctx.expect_w_balance(after, false).await;
- ctx.expect_gateway_up().await;
- }
-}
-
-/// Test btc-wire correctness when a blockchain reorganization occurs leading to past incoming transaction conflict
-pub async fn hell(tctx: TestCtx) {
- macro_rules! step {
- ($ctx:ident, $action:expr) => {
- // Loose second bitcoin node
- $ctx.cluster_deco().await;
-
- // Perform action
- $action;
-
- // Perform fork and check btc-wire hard error
- $ctx.expect_gateway_up().await;
- $ctx.cluster_fork().await;
- $ctx.expect_gateway_down().await;
-
- // Generate conflict
- $ctx.restart_node(&["-minrelaytxfee=0.001"]).await;
- $ctx.abandon_client().await;
- let amount = Amount::from_sat(54000);
- $ctx.credit(amount, &EddsaPublicKey::rand()).await;
- $ctx.expect_w_balance(amount, true).await;
-
- // Check btc-wire suspend operation
- let bounce_amount = Amount::from_sat(34000);
- $ctx.malformed_credit(&bounce_amount).await;
- $ctx.next_conf().await;
- $ctx.expect_w_balance(amount + bounce_amount, true).await;
- $ctx.expect_gateway_down().await;
- };
- }
-
- tctx.step("Setup");
- let mut ctx = BtcCtx::setup(&tctx, "taler_btc.conf", false).await;
- ctx.step("Handle reorg conflicting incoming credit");
- step!(ctx, {
- let amount = Amount::from_sat(420000);
- ctx.credit(amount, &EddsaPublicKey::rand()).await;
- ctx.mine_conf().await;
- ctx.expect_w_balance(amount, true).await;
- });
-
- drop(ctx);
- tctx.step("Setup");
- let mut ctx = BtcCtx::setup(&tctx, "taler_btc.conf", false).await;
- ctx.step("Handle reorg conflicting bounce");
- step!(ctx, {
- let amount = Amount::from_sat(420000);
- ctx.malformed_credit(&amount).await;
- ctx.mine_conf().await;
- let fee = ctx.worker_cfg.bounce_fee;
- ctx.expect_w_balance(fee, true).await;
- });
-}
-
-/// Test btc-wire ability to learn and protect itself from blockchain behavior
-pub async fn analysis(ctx: TestCtx) {
- ctx.step("Setup");
- let mut ctx = BtcCtx::setup(&ctx, "taler_btc.conf", false).await;
-
- ctx.step("Learn from reorg");
-
- // Loose second bitcoin node
- ctx.cluster_deco().await;
-
- // Perform credit
- let before = ctx.wire_balance().await;
- ctx.credit(Amount::from_sat(42000), &EddsaPublicKey::rand())
- .await;
- ctx.mine_conf().await;
- let after = ctx.wire_balance().await;
-
- // Perform fork and check btc-wire hard error
- ctx.expect_gateway_up().await;
- ctx.cluster_fork().await;
- ctx.expect_w_balance(before, false).await;
- ctx.expect_gateway_down().await;
-
- // Recover orphaned transaction
- ctx.mine_conf().await;
- ctx.next_block().await; // Conf have changed
- ctx.expect_w_balance(after, false).await;
- ctx.expect_gateway_up().await;
-
- // Loose second bitcoin node
- ctx.cluster_deco().await;
-
- // Perform credit
- let before = ctx.wire_balance().await;
- ctx.credit(Amount::from_sat(42000), &EddsaPublicKey::rand())
- .await;
- ctx.mine_conf().await;
-
- // Perform fork and check btc-wire learned from previous attack
- ctx.expect_gateway_up().await;
- ctx.cluster_fork().await;
- ctx.expect_w_balance(before, false).await;
- ctx.expect_gateway_up().await;
-}
-
-/// Test btc-wire ability to handle stuck transaction correctly
-pub async fn bumpfee(tctx: TestCtx) {
- tctx.step("Setup");
- let mut ctx = BtcCtx::setup(&tctx, "taler_btc_bump.conf", false).await;
-
- // Perform credits to allow wire to perform debits latter
- for n in 10..13 {
- ctx.credit(Amount::from_sat(n * 100000), &EddsaPublicKey::rand())
- .await;
- }
- ctx.mine_conf().await;
-
- ctx.step("Bump fee");
- {
- // Perform debit
- let mut client = ctx.client_balance().await;
- let wire = ctx.wire_balance().await;
- let amount = Amount::from_sat(40000);
- ctx.debit(amount, &Base32::rand()).await;
- ctx.expect_w_balance_less(wire, false).await;
-
- // Bump min relay fee making the previous debit stuck
- ctx.restart_node(&["-minrelaytxfee=0.0001"]).await;
-
- // Check bump happen
- client += amount;
- ctx.expect_c_balance(client, true).await;
- }
-
- ctx.step("Bump fee reorg");
- {
- // Loose second bitcoin node
- ctx.cluster_deco().await;
-
- // Perform debit
- let mut client = ctx.client_balance().await;
- let wire = ctx.wire_balance().await;
- let amount = Amount::from_sat(40000);
- ctx.debit(amount, &Base32::rand()).await;
- ctx.expect_w_balance_less(wire, false).await;
-
- // Bump min relay fee and fork making the previous debit stuck and problematic
- ctx.cluster_fork().await;
- ctx.restart_node(&["-minrelaytxfee=0.0001"]).await;
-
- // Check bump happen
- client += amount;
- ctx.expect_c_balance(client, true).await;
- }
- ctx.step("Setup");
- drop(ctx);
- let mut ctx = BtcCtx::setup(&tctx, "taler_btc_bump.conf", true).await;
-
- // Perform credits to allow wire to perform debits latter
- for n in 10..61 {
- ctx.credit(Amount::from_sat(n * 100000), &EddsaPublicKey::rand())
- .await;
- }
- ctx.mine_conf().await;
-
- ctx.step("Bump fee stress");
- {
- // Loose second bitcoin node
- ctx.cluster_deco().await;
-
- // Perform debits
- let client = ctx.client_balance().await;
- let wire = ctx.wire_balance().await;
- let mut total_amount = Amount::ZERO;
- for n in 10..30 {
- let amount = Amount::from_sat(n * 10000);
- total_amount += amount;
- ctx.debit(amount, &Base32::rand()).await;
- }
- ctx.expect_w_balance_less(wire - total_amount, false).await;
-
- // Bump min relay fee making the previous debits stuck
- ctx.restart_node(&["-minrelaytxfee=0.0001"]).await;
-
- // Check bump happen
- ctx.expect_c_balance(client + total_amount, true).await;
- }
-}
-
-/// Test btc-wire handle transaction fees exceeding limits
-pub async fn maxfee(ctx: TestCtx) {
- ctx.step("Setup");
- let mut ctx = BtcCtx::setup(&ctx, "taler_btc.conf", false).await;
-
- // Perform credits to allow wire to perform debits latter
- for n in 10..31 {
- ctx.credit(Amount::from_sat(n * 100000), &EddsaPublicKey::rand())
- .await;
- }
- ctx.mine_pending().await;
-
- let client = ctx.client_balance().await;
- let wire = ctx.wire_balance().await;
- let mut total_amount = Amount::ZERO;
-
- ctx.step("Too high fee");
- {
- // Change fee config
- ctx.restart_node(&["-maxtxfee=0.0000001", "-minrelaytxfee=0.0000001"])
- .await;
-
- // Perform debits
- for n in 10..31 {
- let amount = Amount::from_sat(n * 10000);
- total_amount += amount;
- ctx.debit(amount, &Base32::rand()).await;
- }
- ctx.mine_pending().await;
-
- // Check no transaction happen
- ctx.expect_w_balance(wire, false).await;
- ctx.expect_c_balance(client, false).await;
- }
-
- ctx.step("Good feed");
- {
- // Restore default config
- ctx.restart_node(&[]).await;
-
- // Check transaction now have been made
- ctx.expect_c_balance(client + total_amount, true).await;
- }
-}
-
-/// Test btc-wire ability to configure itself from bitcoin configuration
-pub async fn config(ctx: TestCtx) {
- // Connect with cookie files
- ctx.step("Cookie");
- BtcCtx::config(
- &ctx,
- |btc, dir| {
- btc.with_section(None::<&str>).set(
- "rpccookiefile",
- dir.join("catch_me_if_you_can").to_string_lossy(),
- );
- },
- |cfg, dir| {
- cfg.with_section(Some("depolymerizer-bitcoin-worker")).set(
- "RPC_COOKIE_FILE",
- dir.join("catch_me_if_you_can").to_string_lossy(),
- );
- },
- )
- .await;
-
- // Connect with password
- ctx.step("Password");
- BtcCtx::config(
- &ctx,
- |btc, _| {
- btc.with_section(None::<&str>)
- .set("rpcuser", "bob")
- .set("rpcpassword", "password");
- },
- |cfg, _| {
- cfg.with_section(Some("depolymerizer-bitcoin-worker"))
- .set("RPC_AUTH_METHOD", "basic")
- .set("RPC_USERNAME", "bob")
- .set("RPC_PASSWORD", "password");
- },
- )
- .await;
-
- // Connect with token
- ctx.step("Token");
- BtcCtx::config(
- &ctx,
- |btc, _| {
- btc.with_section(None::<&str>)
- .set("rpcauth", "bob:9641cec731e1fad1ded02e1d31536e44$36b8b8af0a38104997a57f017805ff56bf8963ae4a2ed40252ca0e0e070fc19e");
- },
- |cfg, _| {
- cfg.with_section(Some("depolymerizer-bitcoin-worker"))
- .set("RPC_AUTH_METHOD", "basic")
- .set("RPC_USERNAME", "bob")
- .set("RPC_PASSWORD", "password");
- },
- ).await;
-}
diff --git a/testbench/src/main.rs b/testbench/src/main.rs
@@ -1,527 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2022-2025, 2026 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU Affero General Public License as published by the Free Software
- Foundation; either version 3, or (at your option) any later version.
-
- TALER is distributed in the hope that it will be useful, but WITHOUT ANY
- WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
- A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
-
- You should have received a copy of the GNU Affero General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
-*/
-
-use std::{
- panic::UnwindSafe,
- path::{Path, PathBuf},
- str::FromStr,
- string::String,
- sync::{Arc, Mutex},
- time::{Duration, Instant},
-};
-
-use bitcoin::{Address, Txid, address::NetworkUnchecked};
-use clap::{Parser, ValueEnum};
-use depolymerizer_bitcoin::{
- CONFIG_SOURCE, DB_SCHEMA,
- cli::{Command, run},
- config::{WalletCfg, WorkerCfg, parse_db_cfg},
- db,
- payto::BtcWallet,
- rpc::{Error, ErrorCode, Rpc, rpc_common, rpc_wallet},
- taler_utils::taler_to_btc,
-};
-use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
-use owo_colors::OwoColorize;
-use sqlx::PgPool;
-use taler_common::{
- api::{EddsaPublicKey, HashCode, ShortHashCode, wire::TransferRequest},
- config::Config,
- db::pool,
- log::taler_logger,
- types::{amount::amount, payto::FullPayto},
-};
-use taler_test_utils::repl::Repl;
-use tokio::task::{JoinError, JoinHandle};
-use tracing::info;
-use tracing_subscriber::util::SubscriberInitExt as _;
-use url::Url;
-
-use crate::utils::{ChildGuard, LocalDb, TestCtx, cmd_redirect, patch_config, try_cmd_redirect};
-
-mod btc;
-mod utils;
-
-/// Depolymerizer instrumentation test
-#[derive(clap::Parser, Debug)]
-enum Testbench {
- /// Start instrumentation tests for offline testing
- Instrumentation {
- /// With tests to run
- #[clap(global = true, default_value = "")]
- filters: Vec<String>,
- },
- /// Start bitcoin testbench for online testing
- Bitcoin { network: Network },
-}
-
-struct Tmp<'a> {
- root: &'a Path,
- filters: &'a [String],
- m: MultiProgress,
- start_style: ProgressStyle,
- ok_style: ProgressStyle,
- err_style: ProgressStyle,
- tasks: Vec<(
- &'static str,
- JoinHandle<(Result<(), JoinError>, Duration, String)>,
- )>,
- db: Arc<LocalDb>,
- start: Instant,
-}
-
-impl<'a> Tmp<'a> {
- async fn check<T, F>(&mut self, name: &'static str, task: T)
- where
- T: FnOnce(TestCtx) -> F + Send + UnwindSafe + 'static,
- F: Future<Output = ()> + Send + 'static,
- {
- if self.filters.is_empty() || self.filters.iter().any(|f| name.starts_with(f)) {
- let pb = self.m.add(ProgressBar::new_spinner());
- pb.set_style(self.start_style.clone());
- pb.set_prefix(name);
- pb.set_message("Init");
- pb.enable_steady_tick(Duration::from_millis(1000));
- let ok_style = self.ok_style.clone();
- let err_style = self.err_style.clone();
- let start = self.start;
- let db = self.db.clone();
- let ctx: TestCtx = TestCtx::new(self.root, name, pb.clone(), db);
- self.tasks.push((
- name,
- tokio::spawn(async move {
- let result = tokio::spawn(task(ctx)).await;
- if result.is_ok() {
- pb.set_style(ok_style.clone());
- pb.finish_with_message("OK");
- } else {
- pb.set_style(err_style.clone());
- pb.finish();
- }
- (result, start.elapsed(), pb.message())
- }),
- ));
- }
- }
-}
-
-#[tokio::main]
-pub async fn main() {
- taler_logger(None, false).init();
- match Testbench::parse() {
- Testbench::Instrumentation { filters } => instrumentation(filters).await,
- Testbench::Bitcoin { network } => bitcoin(network).await,
- }
-}
-
-#[derive(Debug, Clone, ValueEnum, PartialEq, Eq)]
-enum Network {
- Local,
- Test,
-}
-struct BtcEnv {
- network: Network,
- db: LocalDb,
- bitcoind: ChildGuard,
- wire_rpc: Rpc,
- client_rpc: Rpc,
- cfg: Config,
- wire_addr: Address,
- client_addr: Address,
- tracked: Vec<Txid>,
- pool: PgPool,
-}
-
-impl BtcEnv {
- pub async fn init(network: Network) -> Self {
- let network_dir = match network {
- Network::Local => "btc-local",
- Network::Test => "btc-test",
- };
- println!("Setup bitcoin {network:?} env in {network_dir}");
- let root_dir = PathBuf::from_str("./testbench/env")
- .unwrap()
- .join(network_dir);
- std::fs::create_dir_all(root_dir.join("bitcoin")).unwrap();
- let root_dir = root_dir.canonicalize().unwrap();
-
- // Generate bitcoind config
- let cfg = match network {
- Network::Local => {
- "
- chain=regtest
- txindex=1
- rpcservertimeout=0
- fallbackfee=0.00000001
-
- [regtest]
- rpcport=18345
- "
- }
- Network::Test => {
- "
- chain=signet
- txindex=1
- rpcservertimeout=0
- fallbackfee=0.000001
- [signet]
- rpcport=18345
- "
- }
- };
- std::fs::write(root_dir.join("bitcoin/bitcoin.conf"), cfg).unwrap();
- let datadir = match network {
- Network::Local => root_dir.join("bitcoin").join("regtest"),
- Network::Test => root_dir.join("bitcoin").join("signet"),
- };
-
- // Start database
- let db = LocalDb::new(&root_dir);
-
- // Generate Taler config
- let taler_cfg_path = root_dir.join("taler.conf");
- patch_config(
- "testbench/conf/taler_btc_dev.conf",
- &taler_cfg_path,
- |ini| {
- ini.with_section(Some("depolymerizer-bitcoin-worker"))
- .set("RPC_COOKIE_FILE", datadir.to_string_lossy());
- ini.with_section(Some("depolymerizer-bitcoindb-postgres"))
- .set("CONFIG", db.postgres_uri("depolymerizer_bitcoin"));
- },
- );
-
- // Start node
- let bitcoind = cmd_redirect(
- "bitcoind",
- &[&format!(
- "-datadir={}",
- root_dir.join("bitcoin").to_string_lossy()
- )],
- root_dir.join("bitcoind.log"),
- );
- let cfg = Config::load(CONFIG_SOURCE, Some(&taler_cfg_path)).unwrap();
- let worker_cfg = WorkerCfg::parse(&cfg).unwrap();
- let mut rpc = retry_opt!(rpc_common(&worker_cfg.rpc_cfg));
-
- for wallet in ["wire", "client"] {
- loop {
- let res = rpc.load_wallet(wallet).await;
- if let Err(Error::RPC { code, .. }) = res {
- match code {
- ErrorCode::RpcInWarmup => continue,
- ErrorCode::RpcWalletNotFound => {
- rpc.create_wallet(wallet, "").await.unwrap();
- break;
- }
- _ => {
- res.unwrap();
- }
- }
- } else {
- break;
- }
- }
- }
- drop(rpc);
- let mut wire_rpc = rpc_wallet(
- &worker_cfg.rpc_cfg,
- &WalletCfg {
- name: "wire".to_string(),
- password: None,
- },
- )
- .await
- .unwrap();
- let mut client_rpc = rpc_wallet(
- &worker_cfg.rpc_cfg,
- &WalletCfg {
- name: "client".to_string(),
- password: None,
- },
- )
- .await
- .unwrap();
- let wire_addr = wire_rpc.gen_addr().await.unwrap();
- let client_addr = client_rpc.gen_addr().await.unwrap();
- patch_config(&taler_cfg_path, &taler_cfg_path, |ini| {
- ini.with_section(Some("depolymerizer-bitcoin"))
- .set("WALLET", wire_addr.to_string());
- });
- let cfg = Config::load(CONFIG_SOURCE, Some(taler_cfg_path)).unwrap();
-
- // Wait for db to start
- db.wait_running();
- db.create_db("depolymerizer_bitcoin");
-
- // Dbinit & setup
- run(Command::Dbinit { reset: false }, &cfg).await.unwrap();
- run(Command::Setup { reset: false }, &cfg).await.unwrap();
-
- let db_cg = parse_db_cfg(&cfg).unwrap();
- Self {
- pool: pool(db_cg.cfg, DB_SCHEMA).await.unwrap(),
- wire_addr,
- client_addr,
- network,
- db,
- bitcoind,
- cfg,
- wire_rpc,
- client_rpc,
- tracked: Vec::new(),
- }
- }
-}
-
-#[derive(clap::Parser, Debug)]
-#[command(no_binary_name = true)]
-enum Shell {
- Setup,
- Reset,
- ResetDb,
- Sync,
- Credit,
- Debit,
- Mine {
- amount: Option<u16>,
- addr: Option<Address<NetworkUnchecked>>,
- },
- Exit,
- Tx {
- txid: Txid,
- },
- Track {
- txid: Txid,
- },
- Untrack {
- txid: Txid,
- },
-}
-
-async fn bitcoin(network: Network) {
- let mut env = BtcEnv::init(network).await;
-
- let mut repl = Repl::new(".history");
- loop {
- let info = env.client_rpc.get_blockchain_info().await.unwrap();
- let wire_balance = env.wire_rpc.get_balance().await.unwrap();
- println!("wire {} {wire_balance}", env.wire_addr);
- let client_balance = env.client_rpc.get_balance().await.unwrap();
- println!("client {} {client_balance}", env.client_addr);
- for txid in &env.tracked {
- match env.client_rpc.get_tx(txid).await {
- Ok(info) => println!(
- "{} {txid} {} {}",
- "tx".cyan(),
- info.amount,
- info.confirmations
- ),
- Err(e) => println!("{} {txid} {}", "tx".cyan(), e.red()),
- }
- }
- if let Some(cmd) =
- repl.read_line(&info.chain, &format!("{:.6}", info.verification_progress))
- {
- match cmd {
- Shell::Setup => run(Command::Setup { reset: false }, &env.cfg)
- .await
- .unwrap(),
- Shell::Reset => run(Command::Setup { reset: true }, &env.cfg).await.unwrap(),
- Shell::ResetDb => {
- run(Command::Dbinit { reset: true }, &env.cfg)
- .await
- .unwrap();
- run(Command::Setup { reset: false }, &env.cfg)
- .await
- .unwrap();
- }
- Shell::Sync => run(Command::Worker { transient: true }, &env.cfg)
- .await
- .unwrap(),
- Shell::Credit => {
- let reserve_pub = EddsaPublicKey::rand();
- let amount = amount("DEVBTC:0.00011");
- let txid = env
- .client_rpc
- .send_segwit_key(&env.wire_addr, &taler_to_btc(&amount), &reserve_pub)
- .await
- .unwrap();
- env.tracked.push(txid);
- info!(target: "testbench", "Credit {reserve_pub} {amount} {txid} to {}", env.wire_addr);
- }
- Shell::Debit => {
- let creditor = FullPayto::new(BtcWallet(env.client_addr.clone()), "client");
- let wtid = ShortHashCode::rand();
- let amount = amount("DEVBTC:0.0001");
- let transfer = TransferRequest {
- request_uid: HashCode::rand(),
- amount,
- exchange_base_url: Url::parse("https://test.com/").unwrap(),
- wtid: wtid.clone(),
- credit_account: creditor.as_uri(),
- metadata: None,
- };
- db::transfer(&env.pool, &creditor, &transfer).await.unwrap();
- info!(target: "testbench", "Debit {wtid} {amount} to {}", env.wire_addr);
- }
- Shell::Mine { amount, addr } => {
- let amount = amount.unwrap_or(1);
- let addr = addr.map(|a| a.assume_checked());
- let addr = addr.as_ref().unwrap_or(&env.client_addr);
- env.client_rpc.mine(amount, addr).await.unwrap();
- }
- Shell::Exit => break,
- Shell::Tx { txid } => {
- let info = env.client_rpc.get_tx(&txid).await.unwrap();
- info!(target: "testbench", "{txid} {} {}", info.amount, info.confirmations);
- }
- Shell::Track { txid } => {
- env.tracked.push(txid);
- }
- Shell::Untrack { txid } => {
- env.tracked.retain(|id| *id != txid);
- }
- }
- } else {
- break;
- }
- }
-}
-
-pub async fn instrumentation(filters: Vec<String>) {
- let root = PathBuf::from_str("testbench/instrumentation")
- .unwrap()
- .canonicalize()
- .unwrap();
- std::fs::remove_dir_all(&root).ok();
- std::fs::create_dir_all(root.join("bin")).unwrap();
-
- // Set panic hook
- let failures = Arc::new(Mutex::new(Vec::new()));
- {
- let failures = failures.clone();
- std::panic::set_hook(Box::new(move |e| {
- let backtrace = std::backtrace::Backtrace::force_capture();
- let info = format!("{e}\n{backtrace}");
- if let Some(id) = tokio::task::try_id() {
- failures.lock().unwrap().push((id, info));
- } else {
- eprintln!("Failed outside of a task:\n{info}")
- }
- }));
- }
-
- // Build binaries
- let p = ProgressBar::new_spinner();
- p.set_style(ProgressStyle::with_template("building {msg} {elapsed:.dim}").unwrap());
- p.enable_steady_tick(Duration::from_millis(1000));
- for name in ["depolymerizer-bitcoin"] {
- build_bin(&root, &p, name, None, name);
- build_bin(&root, &p, name, Some("fail"), &format!("{name}-fail"));
- }
- p.finish_and_clear();
-
- // Run tests
- let m = MultiProgress::new();
- let start_style =
- ProgressStyle::with_template("{prefix:.magenta} {msg} {elapsed:.dim}").unwrap();
- let ok_style =
- ProgressStyle::with_template("{prefix:.magenta} {msg:.green} {elapsed:.dim}").unwrap();
- let err_style =
- ProgressStyle::with_template("{prefix:.magenta} {msg:.red} {elapsed:.dim}").unwrap();
-
- let start = Instant::now();
- let db = Arc::new(LocalDb::new(&root));
- let mut tmp = Tmp {
- root: &root,
- filters: filters.as_slice(),
- m,
- start_style,
- ok_style,
- err_style,
- tasks: Vec::new(),
- db,
- start,
- };
- tmp.check("btc_wire", btc::wire).await;
- tmp.check("btc_lifetime", btc::lifetime).await;
- tmp.check("btc_reconnect", btc::reconnect).await;
- tmp.check("btc_stress", btc::stress).await;
- tmp.check("btc_conflict", btc::conflict).await;
- tmp.check("btc_reorg", btc::reorg).await;
- tmp.check("btc_hell", btc::hell).await;
- tmp.check("btc_analysis", btc::analysis).await;
- tmp.check("btc_bumpfee", btc::bumpfee).await;
- tmp.check("btc_maxfee", btc::maxfee).await;
- tmp.check("btc_config", btc::config).await;
- let mut results = Vec::new();
- for (name, task) in tmp.tasks {
- results.push((name, task.await.unwrap()));
- }
-
- let len = results.len();
-
- tmp.m.clear().unwrap();
- let failures = failures.lock().unwrap();
- for (name, (result, _, msg)) in &results {
- if let Err(e) = result
- && let Some((_, err)) = failures.iter().find(|(id, _)| *id == e.id())
- {
- println!("{} {}\n{}", name.magenta(), msg.red(), err.bright_black());
- }
- }
- for (name, (result, time, msg)) in results {
- match result {
- Ok(_) => {
- println!(
- "{} {} {}",
- name.magenta(),
- "OK".green(),
- format_args!("{}s", time.as_secs()).bright_black()
- );
- }
- Err(_) => {
- println!(
- "{} {} {}",
- name.magenta(),
- msg.red(),
- format_args!("{}s", time.as_secs()).bright_black()
- );
- }
- }
- }
- println!("{} tests in {}s", len, start.elapsed().as_secs());
-}
-
-pub fn build_bin(root: &Path, p: &ProgressBar, name: &str, features: Option<&str>, bin_name: &str) {
- p.set_message(bin_name.to_string());
- let mut args = vec!["build", "--bin", name, "--release"];
- if let Some(features) = features {
- args.extend_from_slice(&["--features", features]);
- }
- let result = try_cmd_redirect("cargo", &args, root.join("bin/build"))
- .unwrap()
- .0
- .wait()
- .unwrap();
- assert!(result.success());
- std::fs::rename(
- format!("target/release/{name}"),
- root.join("bin").join(bin_name),
- )
- .unwrap();
-}
diff --git a/testbench/src/utils.rs b/testbench/src/utils.rs
@@ -1,571 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2022-2025, 2026 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU Affero General Public License as published by the Free Software
- Foundation; either version 3, or (at your option) any later version.
-
- TALER is distributed in the hope that it will be useful, but WITHOUT ANY
- WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
- A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
-
- You should have received a copy of the GNU Affero General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
-*/
-
-use std::{
- fmt::Display,
- io::Write as _,
- net::{Ipv4Addr, SocketAddrV4, TcpListener},
- ops::{Deref, DerefMut},
- path::{Path, PathBuf},
- process::{Child, Command, Stdio},
- str::FromStr,
- sync::Arc,
- time::Duration,
-};
-
-use indicatif::ProgressBar;
-use ini::Ini;
-use taler_common::{
- api::{
- EddsaPublicKey, ShortHashCode,
- wire::{IncomingBankTransaction, IncomingHistory, OutgoingHistory, TransferRequest},
- },
- types::{amount::Amount, base32::Base32, payto::PaytoURI},
-};
-use url::Url;
-
-const LOG: &str = "DEBUG";
-
-#[must_use]
-pub async fn check_incoming(base_url: &str, txs: &[(EddsaPublicKey, Amount)]) -> bool {
- let mut res = ureq::get(&format!("{base_url}history/incoming"))
- .query("delta", format!("-{}", 100))
- .call()
- .unwrap();
- if txs.is_empty() {
- res.status() == 204
- } else {
- if res.status() != 200 {
- return false;
- }
- let history: IncomingHistory = res.body_mut().read_json().unwrap();
-
- history.incoming_transactions.len() == txs.len()
- && txs.iter().all(|(reserve_pub_key, taler_amount)| {
- history.incoming_transactions.iter().any(|h| {
- matches!(
- h,
- IncomingBankTransaction::Reserve {
- reserve_pub,
- amount,
- ..
- } if reserve_pub == reserve_pub_key && amount == taler_amount
- )
- })
- })
- }
-}
-
-#[must_use]
-pub async fn check_gateway_down(base_url: &str) -> bool {
- matches!(
- ureq::get(&format!("{base_url}history/incoming"))
- .query("delta", "-5")
- .call(),
- Err(ureq::Error::StatusCode(504 | 502))
- )
-}
-
-#[must_use]
-pub async fn check_gateway_up(base_url: &str) -> bool {
- ureq::get(&format!("{base_url}config")).call().is_ok()
-}
-
-pub async fn transfer(base_url: &str, wtid: &[u8; 32], credit_account: PaytoURI, amount: &Amount) {
- loop {
- let res = ureq::post(&format!("{base_url}transfer")).send_json(TransferRequest {
- request_uid: Base32::rand(),
- amount: *amount,
- exchange_base_url: Url::parse("https://exchange.test/").unwrap(),
- wtid: Base32::from(*wtid),
- credit_account: credit_account.clone(),
- metadata: None,
- });
- if !matches!(res, Err(ureq::Error::StatusCode(502))) {
- res.unwrap();
- break;
- }
- }
-}
-
-#[must_use]
-pub async fn check_outgoing(base_url: &str, txs: &[(ShortHashCode, Amount)]) -> bool {
- let mut res = ureq::get(format!("{base_url}history/outgoing"))
- .query("delta", format!("-{}", txs.len()))
- .call()
- .unwrap();
- if txs.is_empty() {
- res.status() == 204
- } else {
- if res.status() != 200 {
- return false;
- }
- let history: OutgoingHistory = res.body_mut().read_json().unwrap();
-
- history.outgoing_transactions.len() == txs.len()
- && txs.iter().all(|(wtid, amount)| {
- history
- .outgoing_transactions
- .iter()
- .any(|h| h.wtid == *wtid && &h.amount == amount)
- })
- }
-}
-pub struct ChildGuard(pub Child);
-
-impl Drop for ChildGuard {
- fn drop(&mut self) {
- self.0.kill().ok();
- }
-}
-
-#[track_caller]
-pub fn try_cmd_redirect(
- cmd: &str,
- args: &[&str],
- path: impl AsRef<Path>,
-) -> std::io::Result<ChildGuard> {
- let log_file = std::fs::OpenOptions::new()
- .create(true)
- .append(true)
- .open(path)?;
-
- let child = Command::new(cmd)
- .args(args)
- .stderr(log_file.try_clone()?)
- .stdout(log_file)
- .stdin(Stdio::null())
- .spawn()?;
- Ok(ChildGuard(child))
-}
-
-#[track_caller]
-pub fn cmd_redirect(cmd: &str, args: &[&str], path: impl AsRef<Path>) -> ChildGuard {
- try_cmd_redirect(cmd, args, path).unwrap()
-}
-
-#[track_caller]
-pub fn cmd_ok(mut child: ChildGuard, name: &str) {
- let result = child.0.wait().unwrap();
- if !result.success() {
- panic!("cmd {name} failed");
- }
-}
-
-#[track_caller]
-pub fn cmd_redirect_ok(cmd: &str, args: &[&str], path: impl AsRef<Path>, name: &str) {
- cmd_ok(cmd_redirect(cmd, args, path), name)
-}
-
-#[macro_export]
-macro_rules! retry_opt {
- ($expr:expr) => {
- async {
- let start = std::time::Instant::now();
- loop {
- let result = $expr.await;
- if result.is_err() && start.elapsed() < std::time::Duration::from_secs(30) {
- tokio::time::sleep(std::time::Duration::from_millis(300)).await;
- } else {
- return result.unwrap();
- }
- }
- }
- .await
- };
-}
-
-#[macro_export]
-macro_rules! retry {
- ($expr:expr) => {
- $crate::retry_opt! {
- async { $expr.await.then_some(()).ok_or("failure") }
- }
- };
-}
-
-#[derive(Clone)]
-pub struct TestCtx {
- pub name: String,
- pub root: PathBuf,
- pub dir: PathBuf,
- pub pb: ProgressBar,
- pub db: Arc<LocalDb>,
-}
-
-impl TestCtx {
- pub fn new(
- root: impl Into<PathBuf>,
- name: impl Into<String>,
- pb: ProgressBar,
- db: Arc<LocalDb>,
- ) -> Self {
- let root = root.into();
- let name = name.into();
- // Create log dir
- let dir = root.join(&name);
- std::fs::create_dir_all(&dir).unwrap();
-
- Self {
- name,
- dir,
- pb,
- db,
- root,
- }
- }
-
- pub fn log(&self, name: &str) -> PathBuf {
- self.dir.join(format!("{name}.log"))
- }
-
- pub fn step(&self, disp: impl Display) {
- self.pb.set_message(format!("{disp}"))
- }
-
- /* ----- Database ----- */
-
- pub fn stop_db(&mut self) {
- self.db.stop_db(&self.name);
- }
-
- pub fn resume_db(&mut self) {
- self.db.resume_db(&self.name);
- }
-}
-
-pub struct TalerCtx {
- pub wire_dir: PathBuf,
- pub wire2_dir: PathBuf,
- pub conf: PathBuf,
- ctx: TestCtx,
- pub wire_bin_path: String,
- stressed: bool,
- gateway: Option<ChildGuard>,
- pub gateway_url: String,
- wire: Option<ChildGuard>,
- wire2: Option<ChildGuard>,
- pub gateway_port: u16,
-}
-
-impl TalerCtx {
- pub fn new(ctx: &TestCtx, wire_name: impl Into<String>, config: &str, stressed: bool) -> Self {
- // Create temporary dir
- let dir = ctx.dir.clone();
- let conf = dir.join("taler.conf");
-
- // Create common dirs
- let wire_dir = dir.join("wire");
- let wire2_dir = dir.join("wire2");
- for dir in [&wire_dir, &wire2_dir] {
- std::fs::create_dir_all(dir).unwrap();
- }
-
- // Find unused port
- let gateway_port = unused_port();
- let gateway_url = format!("http://localhost:{gateway_port}/taler-wire-gateway/");
-
- // Generate taler config from base
- let wire_name = wire_name.into();
- let config = PathBuf::from_str("testbench/conf").unwrap().join(config);
- let mut cfg = ini::Ini::load_from_file(config).unwrap();
- cfg.with_section(Some("exchange-accountcredentials-admin"))
- .set("WIRE_GATEWAY_URL", &gateway_url);
- cfg.with_section(Some(format!("{wire_name}db-postgres")))
- .set("CONFIG", ctx.db.postgres_uri(&ctx.name));
- cfg.with_section(Some(format!("{wire_name}-httpd")))
- .set("PORT", gateway_port.to_string());
-
- cfg.write_to_file(&conf).unwrap();
-
- Self {
- ctx: ctx.clone(),
- gateway_url,
- wire_dir,
- wire2_dir,
- conf,
- wire_bin_path: if stressed {
- ctx.root.join(format!("bin/{wire_name}-fail"))
- } else {
- ctx.root.join(format!("bin/{wire_name}"))
- }
- .to_string_lossy()
- .to_string(),
- stressed,
- gateway: None,
- wire: None,
- wire2: None,
- gateway_port,
- }
- }
-
- pub fn dbinit(&self) {
- self.db.wait_running();
- self.db.create_db(&self.ctx.name);
- // Init db
- cmd_redirect_ok(
- &self.wire_bin_path,
- &["-c", self.conf.to_string_lossy().as_ref(), "dbinit"],
- self.log("cmd"),
- "wire dbinit",
- );
- }
-
- pub fn reset_db(&self) {
- // Reset db
- self.db.execute_sql(
- "
- FOR r IN (
- SELECT schemaname, tablename
- FROM pg_tables
- WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
- ) LOOP
- EXECUTE format(
- 'TRUNCATE TABLE %I.%I RESTART IDENTITY CASCADE',
- r.schemaname,
- r.tablename
- );
- END LOOP;
- ",
- );
- }
-
- pub fn setup(&self) {
- // Init db
- cmd_redirect_ok(
- &self.wire_bin_path,
- &["-c", self.conf.to_string_lossy().as_ref(), "setup", "-r"],
- self.log("cmd"),
- "wire setup",
- );
- }
-
- pub async fn run(&mut self) {
- // Run gateway
- self.gateway = Some(cmd_redirect(
- &self.wire_bin_path,
- &[
- "-c",
- self.conf.to_string_lossy().as_ref(),
- "-L",
- LOG,
- "serve",
- ],
- self.log("gateway"),
- ));
-
- // Start wires
- self.wire = Some(cmd_redirect(
- &self.wire_bin_path,
- &[
- "-c",
- self.conf.to_string_lossy().as_ref(),
- "-L",
- LOG,
- "worker",
- ],
- self.log("worker"),
- ));
- self.wire2 = self.stressed.then(|| {
- cmd_redirect(
- &self.wire_bin_path,
- &[
- "-c",
- self.conf.to_string_lossy().as_ref(),
- "-L",
- LOG,
- "worker",
- ],
- self.log("worker+"),
- )
- });
-
- // Wait for gateway to be up
- retry_opt! {
- tokio::net::TcpStream::connect(SocketAddrV4::new(Ipv4Addr::LOCALHOST, self.gateway_port))
- };
- }
-
- /* ----- Process ----- */
-
- #[must_use]
- pub fn wire_running(&mut self) -> bool {
- self.wire.as_mut().unwrap().0.try_wait().unwrap().is_none()
- }
-
- #[must_use]
- pub fn gateway_running(&mut self) -> bool {
- self.gateway
- .as_mut()
- .unwrap()
- .0
- .try_wait()
- .unwrap()
- .is_none()
- }
-
- /* ----- Wire Gateway -----*/
-
- pub async fn expect_credits(&self, txs: &[(EddsaPublicKey, Amount)]) -> bool {
- check_incoming(&self.gateway_url, txs).await
- }
-
- pub async fn expect_debits(&self, txs: &[(ShortHashCode, Amount)]) -> bool {
- check_outgoing(&self.gateway_url, txs).await
- }
-
- pub async fn expect_gateway_up(&self) {
- retry!(check_gateway_up(&self.gateway_url))
- }
-
- pub async fn expect_gateway_down(&self) {
- retry!(check_gateway_down(&self.gateway_url))
- }
-}
-
-impl Deref for TalerCtx {
- type Target = TestCtx;
-
- fn deref(&self) -> &Self::Target {
- &self.ctx
- }
-}
-
-impl DerefMut for TalerCtx {
- fn deref_mut(&mut self) -> &mut Self::Target {
- &mut self.ctx
- }
-}
-
-pub fn unused_port() -> u16 {
- TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 0))
- .unwrap()
- .local_addr()
- .unwrap()
- .port()
-}
-
-pub struct LocalDb {
- cluster_dir: String,
- cmd_log: String,
- cluster: ChildGuard,
-}
-
-impl LocalDb {
- pub fn new(root: &Path) -> Self {
- let cluster_dir = root.join("db");
- let cluster_log = root.join("postgres.log").to_string_lossy().to_string();
- let cmd_log = root.join("postgres-cmd.log").to_string_lossy().to_string();
- if !std::fs::exists(&cluster_dir).unwrap() {
- // Init databases files
- cmd_redirect_ok(
- "initdb",
- &[cluster_dir.to_string_lossy().as_ref()],
- &cluster_log,
- "init_db",
- );
- }
- // Generate database config
- std::fs::write(
- cluster_dir.join("postgresql.conf"),
- format!(
- "
- listen_addresses=''
- unix_socket_directories='{}'
- fsync=off
- synchronous_commit=off
- full_page_writes=off
- ",
- cluster_dir.to_string_lossy().as_ref()
- ),
- )
- .unwrap();
- let cluster = cmd_redirect(
- "postgres",
- &["-D", cluster_dir.to_string_lossy().as_ref()],
- &cmd_log,
- );
- Self {
- cluster_dir: cluster_dir.to_string_lossy().to_string(),
- cmd_log,
- cluster,
- }
- }
-
- pub fn postgres_uri(&self, database: &str) -> String {
- format!("postgres:///{database}?host={}", self.cluster_dir)
- }
-
- pub fn execute_sql(&self, sql: &str) -> bool {
- let mut psql = ChildGuard(
- Command::new("psql")
- .arg(self.postgres_uri("postgres"))
- .stderr(Stdio::null())
- .stdout(
- std::fs::File::options()
- .append(true)
- .create(true)
- .open(&self.cmd_log)
- .unwrap(),
- )
- .stdin(Stdio::piped())
- .spawn()
- .unwrap(),
- );
- psql.0
- .stdin
- .as_mut()
- .unwrap()
- .write_all(sql.as_bytes())
- .unwrap();
- psql.0.wait().unwrap().success()
- }
-
- pub fn create_db(&self, name: &str) {
- self.execute_sql(&format!("CREATE DATABASE {name};"));
- }
-
- pub fn stop_db(&self, name: &str) {
- self.execute_sql(&format!(
- "
- UPDATE pg_database SET datallowconn=false WHERE datname='{name}';
- SELECT pg_terminate_backend(pid)
- FROM pg_stat_activity
- WHERE datname='{name}' AND pid <> pg_backend_pid();
- "
- ));
- }
- pub fn resume_db(&self, name: &str) {
- self.execute_sql(&format!(
- "UPDATE pg_database SET datallowconn=true WHERE datname='{name}';"
- ));
- }
-
- pub fn wait_running(&self) {
- for _ in 0..10 {
- if self.execute_sql("SELECT true") {
- break;
- }
- std::thread::sleep(Duration::from_millis(500))
- }
- }
-}
-
-pub fn patch_config(from: impl AsRef<Path>, to: impl AsRef<Path>, patch: impl FnOnce(&mut Ini)) {
- let mut cfg = ini::Ini::load_from_file(from).unwrap();
- patch(&mut cfg);
- cfg.write_to_file(to).unwrap();
-}
