bitcoinafterlife/bal-server
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

fix: resolve PostgreSQL type mismatches and add willexecutor debug logging

Browse Source 
- Fix get_next_address_index: use try_get::<i32> for PG SERIAL/INTEGER columns
- Fix search_tx: use try_get::<i32> for PG status column
- Fix execute_insert: parse locktime (String→i64) and in_vout (String→i32)
  before binding to PG INTEGER columns
- Fix execute_insert: bind tbl_out vout as i32, amount as String for PG
- Fix save_new_address: cast xpub i64 to i32 for PG INTEGER column
- Fix get_pending_txs: cast i64 bind params to i32, use try_get::<i32> for reads
- Fix get_stats: use try_get::<i32> for all numeric PG INTEGER columns
- Add trace logging in parse_request_transactions for xpub address matching
- Add trace logging in get_all_addresses_by_xpub for query debugging
This commit is contained in:
svatantrya
2026-08-18 03:14:03 -04:00
parent 90ae3b6510
commit b277d894dd
65 changed files with 4879 additions and 1188 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
src/bin/bal-pusher.rs
View File

@@ -10,7 +10,6 @@ use log::{debug, error, info, trace, warn};
use serde::Deserialize;
use serde::Serialize;
use serde_json::json;
use sqlite::{Connection, Value};
use std::collections::HashMap;
use std::env;
use std::error::Error as StdError;
@@ -18,7 +17,7 @@ use std::str;
use std::{thread, time::Duration};
use zmq::{Context, Socket};
use bal_server::db::open_db;
use bal_server::db::{calculate_and_upsert_stats, get_pending_txs, open_database};
use bal_server::validation::is_valid_welist_url;
use base64::{Engine as _, engine::general_purpose};
use reqwest::Client as rClient;
@@ -29,7 +28,9 @@ const LOCKTIME_THRESHOLD: i64 = 5000000;
const VERSION: &str = env!("CARGO_PKG_VERSION");
#[derive(Debug, Clone, Serialize, Deserialize)]
struct MyConfig {
db_backend: String,
db_file: String,
pg_dsn: String,
bitcoin_dir: String,
regtest: NetworkParams,
testnet: NetworkParams,
@@ -44,7 +45,9 @@ struct MyConfig {
impl Default for MyConfig {
fn default() -> Self {
MyConfig {
db_backend: env::var("BAL_PUSHER_DB_BACKEND").unwrap_or("sqlite".to_string()),
db_file: env::var("BAL_PUSHER_DB_FILE").unwrap_or("bal.db".to_string()),
pg_dsn: env::var("BAL_PUSHER_PG_DSN").unwrap_or_default(),
bitcoin_dir: env::var("BAL_PUSHER_BITCOIN_DIR").unwrap_or("".to_string()),
regtest: get_network_params_default(Network::Regtest),
testnet: get_network_params_default(Network::Testnet),
@@ -206,114 +209,79 @@ async fn main_result(cfg: &MyConfig, network_params: &NetworkParams) -> Result<(
Ok((rpc, bcinfo)) => {
info!("connected");
info!("median time: {}", bcinfo.median_time);
//info!("height time: {}",bcinfo.median_time);
info!("blocks: {}", bcinfo.blocks);
debug!("best block hash: {}", bcinfo.best_block_hash);
let average_time = bcinfo.median_time;
let db = match open_db(&cfg.db_file) {
Ok(c) => c,
let connection_string = match cfg.db_backend.as_str() {
"sqlite" => cfg.db_file.clone(),
"postgresql" => cfg.pg_dsn.clone(),
other => {
error!("Unknown DB backend: {}", other);
return Ok(());
}
};
let db = match open_database(&cfg.db_backend, &connection_string).await {
Ok(pool) => pool,
Err(e) => {
error!("Fatal: {}", e);
std::process::exit(1);
}
};
info!("db open {}", &cfg.db_file);
info!("db open {}", &connection_string);
let sqlquery = "SELECT * FROM tbl_tx WHERE network = :network AND status = :status AND ( locktime < :bestblock_height OR locktime > :locktime_threshold AND locktime < :bestblock_time);";
let query_tx = match db.prepare(sqlquery) {
Ok(q) => q.into_iter(),
let pending_txs = match get_pending_txs(
&db,
&network_params.db_field,
LOCKTIME_THRESHOLD,
bcinfo.blocks as i64,
average_time as i64,
)
.await
{
Ok(txs) => txs,
Err(e) => {
warn!("tbl_tx not ready yet (tables may not exist): {}", e);
warn!("Failed to query pending transactions: {}", e);
return Ok(());
}
};
trace!("query_tx: {}", sqlquery);
trace!(":locktime_threshold: {}", LOCKTIME_THRESHOLD);
trace!(":bestblock_time: {}", average_time);
trace!(":bestblock_height: {}", bcinfo.blocks);
trace!(":network: {}", network_params.db_field.clone());
trace!(":status: {}", 0);
//let query_tx = db.prepare("SELECT * FROM tbl_tx where status = :status").unwrap().into_iter();
let mut pushed_txs: Vec<String> = Vec::new();
let mut invalid_txs: std::collections::HashMap<String, String> = HashMap::new();
for row_result in match query_tx.bind::<&[(_, Value)]>(
&[
(":locktime_threshold", LOCKTIME_THRESHOLD.into()),
(":bestblock_time", (average_time as i64).into()),
(":bestblock_height", (bcinfo.blocks as i64).into()),
(":network", network_params.db_field.clone().into()),
(":status", 0.into()),
][..],
) {
Ok(bound) => bound,
Err(e) => {
error!("Failed to bind query parameters: {}", e);
return Ok(());
}
} {
let row = match row_result {
Ok(r) => r,
Err(e) => {
warn!("Failed to read row: {}", e);
continue;
}
};
let tx = row.read::<&str, _>("tx");
let txid = row.read::<&str, _>("txid");
let locktime = row.read::<i64, _>("locktime");
let mut invalid_txs: HashMap<String, String> = HashMap::new();
for row in &pending_txs {
let txid = &row.txid;
let tx = &row.tx;
let locktime = row.locktime;
info!("to be pushed: {}: {}", txid, locktime);
match rpc.send_raw_transaction(tx) {
match rpc.send_raw_transaction(tx.as_str()) {
Ok(o) => {
info!("tx: {} pusshata PUSHED\n{}", txid, o);
pushed_txs.push(txid.to_string());
pushed_txs.push(txid.clone());
}
Err(err) => {
warn!("Error: {}\n{}", err, txid);
//store err in invalid_txs
invalid_txs.insert(txid.to_string(), err.to_string());
invalid_txs.insert(txid.clone(), err.to_string());
}
};
}
for txid in &pushed_txs {
let sql = "UPDATE tbl_tx SET status = 1 WHERE txid = ?";
match db.prepare(sql) {
Ok(mut stmt) => {
if let Err(e) = stmt.bind((1, Value::String(txid.clone()))) {
error!("Failed to bind txid for status update: {}", e);
continue;
}
let _ = stmt.next();
}
Err(e) => {
error!("Failed to prepare status update: {}", e);
}
if let Err(e) = bal_server::db::update_tx_status(&db, txid, 1, None).await {
error!("Failed to update tx status: {}", e);
}
}
for (txid, txerr) in &invalid_txs {
let sql = "UPDATE tbl_tx SET status = 2, push_err = ? WHERE txid = ?";
match db.prepare(sql) {
Ok(mut stmt) => {
if let Err(e) = stmt.bind((1, Value::String(txerr.clone()))) {
error!("Failed to bind txerr for error update: {}", e);
continue;
}
if let Err(e) = stmt.bind((2, Value::String(txid.clone()))) {
error!("Failed to bind txid for error update: {}", e);
continue;
}
let _ = stmt.next();
}
Err(e) => {
error!("Failed to prepare error update: {}", e);
}
if let Err(e) = bal_server::db::update_tx_status(&db, txid, 2, Some(txerr)).await {
error!("Failed to update tx status: {}", e);
}
}
if let Err(e) = send_stats_report(cfg, bcinfo).await {
error!("send_stats_report failed: {}", e);
}
if let Err(e) = calculate_stats(&db, network_params.db_field.clone()).await {
if let Err(e) = calculate_and_upsert_stats(&db, &network_params.db_field).await {
warn!("calculate_stats failed: {e}");
}
}
@@ -325,65 +293,8 @@ async fn main_result(cfg: &MyConfig, network_params: &NetworkParams) -> Result<(
}
Ok(())
}
async fn calculate_stats(db: &Connection, chain: String) -> Result<(), reqwest::Error> {
// Validate chain to prevent SQL injection via environment variable tampering
if !chain
.chars()
.all(|c| c.is_alphanumeric() || c == '-' || c == '_')
|| chain.is_empty()
{
error!("Invalid chain name: {chain}");
return Ok(());
}
//let sql = "drop table if exists tbl_stats;";
let sql = format!("DELETE FROM tbl_stats WHERE chain = '{chain}';");
if let Err(err) = db.execute(&sql) {
error!("error deleting from tbl_stats where chain:{chain} error: {err}");
}
let sql = format!(
"INSERT INTO tbl_stats (
report_date, chain, totals, waiting, sent, failed,
waiting_profit, sent_profit, missed_profit, unique_inputs
)
VALUES (
CURRENT_TIMESTAMP,
'{chain}',
(SELECT COUNT(*) FROM tbl_tx WHERE network = '{chain}'),
(SELECT COUNT(*) FROM tbl_tx WHERE status = 0 AND network = '{chain}'),
(SELECT COUNT(*) FROM tbl_tx WHERE status = 1 AND network = '{chain}'),
(SELECT COUNT(*) FROM tbl_tx WHERE status = 2 AND network = '{chain}'),
(SELECT IFNULL(SUM(our_fees),0) FROM tbl_tx WHERE status = 0 AND network = '{chain}'),
(SELECT IFNULL(SUM(our_fees),0) FROM tbl_tx WHERE status = 1 AND network = '{chain}'),
(SELECT IFNULL(SUM(our_fees),0) FROM tbl_tx WHERE status = 2 AND network = '{chain}'),
(SELECT COUNT(DISTINCT tbl_inp.in_txid)
FROM tbl_inp
JOIN tbl_tx ON tbl_inp.txid = tbl_tx.txid
WHERE tbl_tx.status = 0 AND tbl_tx.network = '{chain}')
)
ON CONFLICT(chain) DO UPDATE SET
report_date = excluded.report_date,
totals = excluded.totals,
waiting = excluded.waiting,
sent = excluded.sent,
failed = excluded.failed,
waiting_profit = excluded.waiting_profit,
sent_profit = excluded.sent_profit,
missed_profit = excluded.missed_profit,
unique_inputs = excluded.unique_inputs;
"
);
if let Err(err) = db.execute(&sql) {
error!("error inserting creating stats table {err}");
} else {
info!("tbl_stats creation success");
}
Ok(())
}
/// Parse the `(host, port)` pair from a base URL like `https://host[:port]`.
///
/// Falls back to the scheme's well-known default port (443 for `https`,
/// 80 for plain `http`), or to 443 when the scheme is unknown.
fn parse_host_port(base_url: &str) -> Option<(String, u16)> {
let url = Url::parse(base_url).ok()?;
let host = url
@@ -396,8 +307,6 @@ fn parse_host_port(base_url: &str) -> Option<(String, u16)> {
}
/// Resolve `host:port` and return the first IPv6 (AAAA) address, if any.
///
/// Returns `None` when the host has no IPv6 address.
async fn resolve_first_ipv6(host: &str, port: u16) -> Option<SocketAddr> {
use std::net::ToSocketAddrs;
let host = host.to_string();
@@ -412,14 +321,6 @@ async fn resolve_first_ipv6(host: &str, port: u16) -> Option<SocketAddr> {
.flatten()
}
/// Build the HTTP client used for welist reports.
///
/// When `BAL_PUSHER_PREFER_IPV6` is truthy, the welist host is resolved and
/// the client is pinned to its first IPv6 address (the original hostname is
/// still used for the `Host` header and TLS SNI). This works around networks
/// where the IPv4 route to the welist host is broken while IPv6 works: the
/// default connector may otherwise pick the broken family and the request
/// stalls. When the variable is unset (the default), behavior is unchanged.
async fn welist_http_client(welist_url: &str) -> rClient {
let prefer_ipv6 = env::var("BAL_PUSHER_PREFER_IPV6")
.unwrap_or("false".to_string())
@@ -521,6 +422,15 @@ fn sign_message(private_key_path: &str, message: &str) -> String {
}
fn parse_env(cfg: &mut MyConfig) {
if let Ok(value) = env::var("BAL_PUSHER_DB_BACKEND") {
cfg.db_backend = value;
}
if let Ok(value) = env::var("BAL_PUSHER_DB_FILE") {
cfg.db_file = value;
}
if let Ok(value) = env::var("BAL_PUSHER_PG_DSN") {
cfg.pg_dsn = value;
}
cfg.regtest = parse_env_netconfig(cfg, "regtest");
cfg.signet = parse_env_netconfig(cfg, "signet");
cfg.testnet = parse_env_netconfig(cfg, "testnet");
@@ -528,7 +438,6 @@ fn parse_env(cfg: &mut MyConfig) {
drop(parse_env_netconfig(cfg, "bitcoin"));
}
fn parse_env_netconfig(cfg_lock: &mut MyConfig, chain: &str) -> NetworkParams {
//fn parse_env_netconfig(cfg_lock: &MutexGuard<MyConfig>, chain: &str) -> &NetworkParams{
let cfg = match chain {
"regtest" => &mut cfg_lock.regtest,
"signet" => &mut cfg_lock.signet,

518
src/bin/bal-server.rs
View File

@@ -7,16 +7,14 @@ use chrono::Utc;
use hex_conservative::FromHex;
use log::{debug, error, info, trace};
use serde::{Deserialize, Serialize};
use sqlite::State;
use sqlite::{Connection, Value};
use std::collections::{HashMap, HashSet};
use std::env;
use std::fs;
use std::sync::Mutex;
use bal_server::db::{
check_duplicate_txids, create_database, execute_insert, get_all_addresses_by_xpub,
get_last_used_address_by_ip, get_next_address_index, insert_xpub, open_db, save_new_address,
DatabasePool, InsertInpData, InsertOutData, InsertTxData, check_duplicate_txids,
create_database, get_all_addresses_by_xpub, get_last_used_address_by_ip,
get_next_address_index, get_stats, insert_xpub, open_database, save_new_address, search_tx,
};
use bal_server::xpub::new_address_from_xpub;
@@ -56,7 +54,9 @@ struct MyConfig {
info: String,
bind_address: String,
bind_port: u16,
db_backend: String,
db_file: String,
pg_dsn: String,
pub_key_path: String,
expose_stats: bool,
}
@@ -71,7 +71,9 @@ impl Default for MyConfig {
mainnet: NetConfig::default_network("bitcoin".to_string(), Network::Bitcoin),
bind_address: "127.0.0.1".to_string(),
bind_port: 9137,
db_backend: "sqlite".to_string(),
db_file: "bal.db".to_string(),
pg_dsn: String::new(),
info: "Will Executor Server".to_string(),
pub_key_path: "public_key.pem".to_string(),
expose_stats: env::var("BAL_SERVER_EXPOSE_STATS")
@@ -192,7 +194,7 @@ fn parse_actix_config() -> ActixConfig {
}
struct AppState {
db: Mutex<Connection>,
db: DatabasePool,
cfg: MyConfig,
}
@@ -272,35 +274,26 @@ async fn echo_info(
address
}
true => {
// Lock #1: fetch existing address OR atomically claim next index
let next_idx = {
let db = match data.db.lock() {
Ok(g) => g,
Err(_p) => {
error!("DB mutex poisoned in echo_info (lookup phase)");
return HttpResponse::InternalServerError().body("error");
}
};
match get_last_used_address_by_ip(
&db,
&netconfig.name,
&netconfig.address,
&remote_addr,
) {
Some(address) => {
return HttpResponse::Ok().json(InfoResponse {
address,
base_fee: netconfig.fixed_fee,
chain: netconfig.network.to_string(),
info: data.cfg.info.to_string(),
version: VERSION.to_string(),
});
}
None => get_next_address_index(&db, &netconfig.name, &netconfig.address),
}
}; // lock released
if let Some(address) = get_last_used_address_by_ip(
&data.db,
&netconfig.name,
&netconfig.address,
&remote_addr,
)
.await
{
return HttpResponse::Ok().json(InfoResponse {
address,
base_fee: netconfig.fixed_fee,
chain: netconfig.network.to_string(),
info: data.cfg.info.to_string(),
version: VERSION.to_string(),
});
}
let next_idx =
get_next_address_index(&data.db, &netconfig.name, &netconfig.address).await;
// Derive address (CPU-bound, no lock held)
let derived =
match new_address_from_xpub(&netconfig.address, next_idx.1, netconfig.network) {
Ok(address) => address,
@@ -310,20 +303,10 @@ async fn echo_info(
}
};
// Lock #2: save the newly derived address
{
let db = match data.db.lock() {
Ok(g) => g,
Err(_p) => {
error!("DB mutex poisoned in echo_info (save phase)");
return HttpResponse::InternalServerError().body("error");
}
};
save_new_address(&db, next_idx.0, &derived.0, &derived.1, &remote_addr);
debug!("save new address {} {}", derived.0, derived.1);
trace!("next {} {}", next_idx.0, next_idx.1);
derived.0
} // lock released
save_new_address(&data.db, next_idx.0, &derived.0, &derived.1, &remote_addr).await;
debug!("save new address {} {}", derived.0, derived.1);
trace!("next {} {}", next_idx.0, next_idx.1);
derived.0
}
};
let info = InfoResponse {
@@ -357,83 +340,31 @@ async fn echo_stats(path: web::Path<String>, data: web::Data<AppState>) -> impl
if !data.cfg.expose_stats {
return HttpResponse::Forbidden().body("error");
}
let mut stats: Vec<StatsResponse> = vec![];
let db = match data.db.lock() {
Ok(g) => g,
Err(_p) => {
error!("DB mutex poisoned in echo_stats");
return HttpResponse::InternalServerError().body("error");
}
};
let mut stmt = match db.prepare(
"SELECT report_date, chain, totals, waiting, sent, failed, waiting_profit, sent_profit, missed_profit, unique_inputs FROM tbl_stats WHERE chain = ?"
) {
Ok(s) => s,
let stats_rows = match get_stats(&data.db, &netconfig.name).await {
Ok(rows) => rows,
Err(e) => {
error!("Failed to prepare stats query: {}", e);
error!("Failed to query stats: {}", e);
return HttpResponse::InternalServerError().body("error");
}
};
if let Err(e) = stmt.bind((1, Value::String(netconfig.name.clone()))) {
error!("Failed to bind chain in stats query: {}", e);
return HttpResponse::InternalServerError().body("error");
}
while let Ok(State::Row) = stmt.next() {
let report_date = stmt.read("report_date").unwrap_or("0".to_string());
let chain = stmt.read("chain").unwrap_or("?".to_string());
let totals = stmt
.read("totals")
.unwrap_or("0".to_string())
.parse::<i64>()
.unwrap_or(0);
let waiting = stmt
.read("waiting")
.unwrap_or("0".to_string())
.parse::<i64>()
.unwrap_or(0);
let sent = stmt
.read("sent")
.unwrap_or("0".to_string())
.parse::<i64>()
.unwrap_or(0);
let failed = stmt
.read("failed")
.unwrap_or("0".to_string())
.parse::<i64>()
.unwrap_or(0);
let waiting_profit = stmt
.read("waiting_profit")
.unwrap_or("0".to_string())
.parse::<i64>()
.unwrap_or(0);
let sent_profit = stmt
.read("sent_profit")
.unwrap_or("0".to_string())
.parse::<i64>()
.unwrap_or(0);
let missed_profit = stmt
.read("missed_profit")
.unwrap_or("0".to_string())
.parse::<i64>()
.unwrap_or(0);
let unique_inputs = stmt
.read("unique_inputs")
.unwrap_or("0".to_string())
.parse::<i64>()
.unwrap_or(0);
stats.push(StatsResponse {
report_date,
chain,
totals,
waiting,
sent,
failed,
waiting_profit,
sent_profit,
missed_profit,
unique_inputs,
});
}
let stats: Vec<StatsResponse> = stats_rows
.into_iter()
.map(|row| StatsResponse {
report_date: row.report_date,
chain: row.chain,
totals: row.totals,
waiting: row.waiting,
sent: row.sent,
failed: row.failed,
waiting_profit: row.waiting_profit,
sent_profit: row.sent_profit,
missed_profit: row.missed_profit,
unique_inputs: row.unique_inputs,
})
.collect();
debug!("echo stats reply for chain: {}", netconfig.name);
HttpResponse::Ok().json(stats)
}
@@ -453,93 +384,42 @@ async fn echo_search(body: Bytes, data: web::Data<AppState>) -> impl Responder {
return HttpResponse::BadRequest().body("error");
}
let db = match data.db.lock() {
Ok(g) => g,
Err(_p) => {
error!("DB mutex poisoned in echo_search");
return HttpResponse::InternalServerError().body("error");
}
};
let mut statement = match db.prepare("SELECT * FROM tbl_tx WHERE txid = ? LIMIT 1") {
Ok(s) => s,
Err(e) => {
error!("Failed to prepare statement: {}", e);
return HttpResponse::InternalServerError().body("error");
}
};
if let Err(e) = statement.bind((1, strbody)) {
error!("Failed to bind parameter: {}", e);
return HttpResponse::InternalServerError().body("error");
}
match search_tx(&data.db, strbody).await {
Ok(Some(row)) => {
let mut response_data = HashMap::new();
response_data.insert("status", row.status);
response_data.insert("tx", row.tx);
response_data.insert("our_address", row.our_address);
response_data.insert("our_fees", row.our_fees);
response_data.insert("time", row.reqid);
if let Ok(State::Row) = statement.next() {
let mut response_data = HashMap::new();
match statement.read::<String, _>("status") {
Ok(value) => {
response_data.insert("status", value);
}
Err(e) => {
error!("Error reading status: {}", e);
match serde_json::to_string(&response_data) {
Ok(json_data) => {
debug!("echo search reply: {}", json_data);
HttpResponse::Ok().json(&response_data)
}
Err(_) => HttpResponse::BadRequest().body("error"),
}
}
match statement.read::<String, _>("tx") {
Ok(value) => {
response_data.insert("tx", value);
}
Err(e) => {
error!("Error reading tx: {}", e);
}
Ok(None) => HttpResponse::BadRequest().body("error"),
Err(e) => {
error!("Failed to search tx: {}", e);
HttpResponse::InternalServerError().body("error")
}
match statement.read::<String, _>("our_address") {
Ok(value) => {
response_data.insert("our_address", value);
}
Err(e) => {
error!("Error reading address: {}", e);
}
}
match statement.read::<String, _>("our_fees") {
Ok(value) => {
response_data.insert("our_fees", value);
}
Err(e) => {
error!("Error reading fees: {}", e);
}
}
match statement.read::<String, _>("reqid") {
Ok(value) => {
response_data.insert("time", value);
}
Err(e) => {
error!("Error reading reqid: {}", e);
}
}
match serde_json::to_string(&response_data) {
Ok(json_data) => {
debug!("echo search reply: {}", json_data);
HttpResponse::Ok().json(&response_data)
}
Err(_) => HttpResponse::BadRequest().body("error"),
}
} else {
HttpResponse::BadRequest().body("error")
}
}
/// Holds a transaction that has already been parsed and validated outside the DB lock.
#[derive(Clone)]
struct ParsedTx {
txid: String,
wtxid: String,
ntxid: String,
raw_hex: String, // the original line
raw_hex: String,
locktime: String,
inputs: Vec<(String, String)>, // (in_txid, in_vout)
outputs: Vec<(usize, String, u64)>, // (idx, script_pubkey, amount_sat)
inputs: Vec<(String, String)>,
outputs: Vec<(usize, String, u64)>,
}
/// Parse all transactions from the request body **without** needing the DB lock.
/// Skips transactions that don't have a valid willexecutor output.
fn parse_request_transactions(
strbody: &str,
_req_time: i64,
@@ -576,7 +456,6 @@ fn parse_request_transactions(
let wtxid = tx.compute_wtxid();
let locktime = tx.lock_time.to_string();
// Collect inputs
let mut inputs: Vec<(String, String)> = Vec::with_capacity(tx.input.len());
for input in tx.input {
inputs.push((
@@ -585,7 +464,6 @@ fn parse_request_transactions(
));
}
// Collect outputs and find which one is ours + its amount
let mut outputs: Vec<(usize, String, u64)> = Vec::with_capacity(tx.output.len());
let mut found = false;
let mut our_address = String::new();
@@ -601,13 +479,25 @@ fn parse_request_transactions(
netconfig.network,
) {
Ok(addr) => addr.to_string(),
Err(_) => continue, // skip un-decodable outputs
Err(_) => continue,
};
let expected_ours = if netconfig.xpub {
if known_addresses.contains(&address) {
trace!(
"output {} address {} found in known_addresses (total: {})",
idx,
&address,
known_addresses.len()
);
address.clone()
} else {
trace!(
"output {} address {} NOT in known_addresses (total: {}), skipping",
idx,
&address,
known_addresses.len()
);
continue;
}
} else {
@@ -619,6 +509,11 @@ fn parse_request_transactions(
our_fees = amount;
found = true;
trace!("address and fees are correct {}: {}", our_address, our_fees);
} else if address == expected_ours {
trace!(
"output {} address matches but amount {} < fixed_fee {}, skipping",
idx, amount, netconfig.fixed_fee
);
}
}
@@ -679,26 +574,17 @@ async fn echo_push(
};
// ===== PHASE 1: parse all transactions WITHOUT the DB lock =====
let known_addresses: HashSet<String> = {
let db = match data.db.lock() {
Ok(g) => g,
Err(_p) => {
error!("DB mutex poisoned acquiring addresses in echo_push");
let known_addresses: HashSet<String> = if netconfig.xpub {
match get_all_addresses_by_xpub(&data.db, &netconfig.address).await {
Ok(addrs) => addrs,
Err(e) => {
error!("Failed to load addresses from xpub: {}", e);
return HttpResponse::InternalServerError().body("error");
}
};
if netconfig.xpub {
match get_all_addresses_by_xpub(&db, &netconfig.address) {
Ok(addrs) => addrs,
Err(e) => {
error!("Failed to load addresses from xpub: {}", e);
return HttpResponse::InternalServerError().body("error");
}
}
} else {
HashSet::new()
}
}; // lock released here
} else {
HashSet::new()
};
// Parse all transactions (CPU-bound, no DB needed)
let parsed = parse_request_transactions(strbody, req_time, netconfig, &known_addresses);
@@ -709,134 +595,86 @@ async fn echo_push(
let all_txids: Vec<String> = parsed.iter().map(|(p, _, _)| p.txid.clone()).collect();
// ===== PHASE 2: check duplicates in a single batch query =====
let duplicates = {
let db = match data.db.lock() {
Ok(g) => g,
Err(_p) => {
error!("DB mutex poisoned in echo_push duplicate check");
return HttpResponse::InternalServerError().body("error");
}
};
match check_duplicate_txids(&db, &all_txids) {
Ok(dups) => dups,
Err(e) => {
error!("Duplicate check failed: {}", e);
return HttpResponse::InternalServerError().body("error");
}
let duplicates = match check_duplicate_txids(&data.db, &all_txids).await {
Ok(dups) => dups,
Err(e) => {
error!("Duplicate check failed: {}", e);
return HttpResponse::InternalServerError().body("error");
}
}; // lock released here
};
let all_present = all_txids.iter().all(|t| duplicates.contains(t));
if all_present {
return HttpResponse::Ok().body("already present");
}
// ===== PHASE 3: build insert statements and execute (single DB lock, minimal time) =====
// ===== PHASE 3: build insert data and execute (single transaction, minimal time) =====
let mut tx_data = Vec::new();
let mut inp_data = Vec::new();
let mut out_data = Vec::new();
for (parsed, our_address, our_fees) in &parsed {
if duplicates.contains(&parsed.txid) {
continue;
}
tx_data.push(InsertTxData {
txid: parsed.txid.clone(),
wtxid: parsed.wtxid.clone(),
ntxid: parsed.ntxid.clone(),
raw_hex: parsed.raw_hex.clone(),
locktime: parsed.locktime.clone(),
reqid: req_time.to_string(),
network: netconfig.name.clone(),
our_address: our_address.clone(),
our_fees: our_fees.to_string(),
});
for (in_txid, in_vout) in &parsed.inputs {
inp_data.push(InsertInpData {
txid: parsed.txid.clone(),
in_txid: in_txid.clone(),
in_vout: in_vout.clone(),
});
}
for (idx, script, amount) in &parsed.outputs {
out_data.push(InsertOutData {
txid: parsed.txid.clone(),
vout: i64::try_from(*idx).unwrap_or(-1),
script_pubkey: script.clone(),
amount: i64::try_from(*amount).unwrap_or(0),
});
}
}
if tx_data.is_empty() {
return HttpResponse::Ok().body("already present");
}
if let Err(err) = bal_server::db::execute_insert(&data.db, &tx_data, &inp_data, &out_data).await
{
let db = match data.db.lock() {
Ok(g) => g,
Err(_p) => {
error!("DB mutex poisoned in echo_push insert phase");
return HttpResponse::InternalServerError().body("error");
}
};
let sqltxshead = "INSERT INTO tbl_tx (txid, wtxid, ntxid, tx, locktime, reqid, network, our_address, our_fees)".to_string();
let mut sqltxs = String::new();
let sqlinpshead = "INSERT INTO tbl_inp (txid, in_txid, in_vout )".to_string();
let mut sqlinps = String::new();
let sqloutshead = "INSERT INTO tbl_out (txid, vout, script_pubkey, amount )".to_string();
let mut sqlouts = String::new();
let mut union_tx = true;
let mut union_inps = true;
let mut union_outs = true;
let mut ptx: Vec<(usize, Value)> = vec![];
let mut pinps: Vec<(usize, Value)> = vec![];
let mut pouts: Vec<(usize, Value)> = vec![];
let mut linenum = 1usize;
let mut lineinp = 1usize;
let mut lineout = 1usize;
for (parsed, our_address, our_fees) in &parsed {
if duplicates.contains(&parsed.txid) {
continue;
}
if !union_tx {
sqltxs.push_str(" UNION ALL");
} else {
union_tx = false;
}
sqltxs.push_str(" SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?");
ptx.push((linenum, Value::String(parsed.txid.clone())));
ptx.push((linenum + 1, Value::String(parsed.wtxid.clone())));
ptx.push((linenum + 2, Value::String(parsed.ntxid.clone())));
ptx.push((linenum + 3, Value::String(parsed.raw_hex.clone())));
ptx.push((linenum + 4, Value::String(parsed.locktime.clone())));
ptx.push((linenum + 5, Value::String(req_time.to_string())));
ptx.push((linenum + 6, Value::String(netconfig.name.clone())));
ptx.push((linenum + 7, Value::String(our_address.clone())));
ptx.push((linenum + 8, Value::String(our_fees.to_string())));
linenum += 9;
for (in_txid, in_vout) in &parsed.inputs {
if !union_inps {
sqlinps.push_str(" UNION ALL");
} else {
union_inps = false;
}
sqlinps.push_str(" SELECT ?, ?, ?");
pinps.push((lineinp, Value::String(parsed.txid.clone())));
pinps.push((lineinp + 1, Value::String(in_txid.clone())));
pinps.push((lineinp + 2, Value::String(in_vout.clone())));
lineinp += 3;
}
for (idx, script, amount) in &parsed.outputs {
if !union_outs {
sqlouts.push_str(" UNION ALL");
} else {
union_outs = false;
}
sqlouts.push_str(" SELECT ?, ?, ?, ?");
pouts.push((lineout, Value::String(parsed.txid.clone())));
pouts.push((
lineout + 1,
Value::Integer(i64::try_from(*idx).unwrap_or(-1)),
));
pouts.push((lineout + 2, Value::String(script.clone())));
pouts.push((
lineout + 3,
Value::Integer(i64::try_from(*amount).unwrap_or(0)),
));
lineout += 4;
}
}
if sqltxs.is_empty() {
return HttpResponse::Ok().body("already present");
}
let sqltxs = format!("{}{};", sqltxshead, sqltxs);
let sqlinps = format!("{}{};", sqlinpshead, sqlinps);
let sqlouts = format!("{}{};", sqloutshead, sqlouts);
if let Err(err) = execute_insert(&db, sqltxs, ptx, sqlinps, pinps, sqlouts, pouts) {
error!("execute_insert failed: {}", err);
return HttpResponse::BadRequest().body("error");
}
} // lock released
error!("execute_insert failed: {}", err);
return HttpResponse::BadRequest().body("error");
}
HttpResponse::Ok().body("thx")
}
fn parse_env(data: &MyConfig) -> MyConfig {
let mut cfg = data.clone();
if let Ok(value) = env::var("BAL_SERVER_DB_BACKEND") {
debug!("BAL_SERVER_DB_BACKEND: {}", value);
cfg.db_backend = value;
}
if let Ok(value) = env::var("BAL_SERVER_DB_FILE") {
debug!("BAL_SERVER_DB_FILE: {}", value);
cfg.db_file = value;
}
if let Ok(value) = env::var("BAL_SERVER_PG_DSN") {
debug!("BAL_SERVER_PG_DSN: {}", value);
cfg.pg_dsn = value;
}
if let Ok(value) = env::var("BAL_SERVER_BIND_ADDRESS") {
debug!("BAL_SERVER_BIND_ADDRESS: {}", value);
cfg.bind_address = value;
@@ -889,10 +727,10 @@ fn parse_env_netconfig(cfg: &mut MyConfig, chain: &str) {
}
}
fn init_network(db: &Connection, cfg: &MyConfig) {
async fn init_network(pool: &DatabasePool, cfg: &MyConfig) {
for network in NETWORKS {
let netconfig = cfg.get_net_config(network);
insert_xpub(db, &netconfig.name.to_string(), &netconfig.address);
insert_xpub(pool, &netconfig.name.to_string(), &netconfig.address).await;
}
}
@@ -903,32 +741,44 @@ async fn main() -> std::io::Result<()> {
let actix_cfg = parse_actix_config();
let cfg = parse_env(&cfg);
let db = match open_db(&cfg.db_file) {
Ok(c) => c,
let connection_string = match cfg.db_backend.as_str() {
"sqlite" => cfg.db_file.clone(),
"postgresql" => cfg.pg_dsn.clone(),
other => {
return Err(std::io::Error::other(format!(
"Unknown DB backend: {}",
other
)));
}
};
let db = match open_database(&cfg.db_backend, &connection_string).await {
Ok(pool) => pool,
Err(e) => {
return Err(std::io::Error::other(e));
}
};
// Create database tables
create_database(&db);
create_database(&db)
.await
.map_err(|e| std::io::Error::other(format!("Failed to create database: {}", e)))?;
// Initialize networks
init_network(&db, &cfg);
init_network(&db, &cfg).await;
let data = web::Data::new(AppState {
db: Mutex::new(db),
db,
cfg: cfg.clone(),
});
let bind_address = data.cfg.bind_address.clone();
let bind_port = data.cfg.bind_port;
// Use a single global rate limiter with the most conservative settings (1 req/sec)
// Per-endpoint rate limiting requires advanced configuration with explicit types
let governor_conf = GovernorConfigBuilder::const_default()
.seconds_per_request(actix_cfg.rate_limit_pushtxs.0) // Most restrictive: 1 req/sec
.burst_size(actix_cfg.rate_limit_pushtxs.1) // Burst: 3
.seconds_per_request(actix_cfg.rate_limit_pushtxs.0)
.burst_size(actix_cfg.rate_limit_pushtxs.1)
.finish()
.unwrap();