9 Commits

Author SHA1 Message Date
fe1c4ee2c8 bal-pusher.env 2025-07-19 07:15:53 -04:00
ca10284479 logs about current block 2025-07-19 07:12:18 -04:00
afd21a5f2a bal-pusher readme 2025-07-16 04:50:40 -04:00
d154567aeb release: bal-server-0.2.1 2025-07-16 04:26:02 -04:00
8965a06dbe release: bal-server-0.2.1 2025-07-16 04:25:17 -04:00
134504e870 release: bal-server-0.2.1 2025-07-16 03:58:42 -04:00
66f34cb29f pusher fix 2025-07-16 03:47:28 -04:00
36edfcd073 fixed network db_field 2025-07-15 15:18:37 -04:00
9e89ae884e fixed db field network 2025-07-15 15:12:21 -04:00
5 changed files with 125 additions and 24 deletions

View File

@@ -28,3 +28,71 @@ The `bal-server` application can be configured using environment variables. The
| `BAL_SERVER_TESTNET_FIXED_FEE` | Fixed fee for the testnet environment. | 50000 |
| `BAL_SERVER_BITCOIN_ADDRESS` | Bitcoin address for the mainnet environment. | - |
| `BAL_SERVER_BITCOIN_FIXED_FEE` | Fixed fee for the mainnet environment. | 50000 |
# bal-pusher
`bal-pusher` is a tool that retrieves Bitcoin transactions from a database and pushes them to the Bitcoin network when their **locktime** exceeds the **median time past** (MTP). It listens for Bitcoin block updates via ZMQ.
## Installation
To use `bal-pusher`, you need to compile and install Bitcoin with ZMQ (ZeroMQ) support enabled. Then, configure your Bitcoin node and `bal-pusher` to push the transactions.
### Prerequisites
1. **Bitcoin with ZMQ Support**:
Ensure that Bitcoin is compiled with ZMQ support. Add the following line to your `bitcoin.conf` file:
```
zmqpubhashblock=tcp://127.0.0.1:28332
```
2. **Install Rust and Cargo**:
If you haven't already installed Rust and Cargo, you can follow the official instructions to do so: [Rust Installation](https://www.rust-lang.org/tools/install).
### Installation Steps
1. Clone the repository:
```bash
git clone <repo-url>
cd bal-pusher
```
2. Build the project:
```bash
cargo build --release
```
3. Install the binary:
```bash
sudo cp target/release/bal-pusher /usr/local/bin
```
## Configuration
`bal-pusher` can be configured using environment variables. If no configuration file is provided, a default configuration file will be created.
### Available Configuration Variables
| Variable | Description | Default |
|---------------------------------------|------------------------------------------|----------------------------------------------|
| `BAL_PUSHER_CONFIG_FILE` | Path to the configuration file. If the file does not exist, it will be created. | `$HOME/.config/bal-pusher/default-config.toml` |
| `BAL_PUSHER_DB_FILE` | Path to the SQLite3 database file. If the file does not exist, it will be created. | `bal.db` |
| `BAL_PUSHER_ZMQ_LISTENER` | ZMQ listener for Bitcoin updates. | `tcp://127.0.0.1:28332` |
| `BAL_PUSHER_BITCOIN_HOST` | Bitcoin server host for RPC connections. | `http://127.0.0.1` |
| `BAL_PUSHER_BITCOIN_PORT` | Bitcoin RPC server port. | `8332` |
| `BAL_PUSHER_BITCOIN_COOKIE_FILE` | Path to Bitcoin RPC cookie file. | `$HOME/.bitcoin/.cookie` |
| `BAL_PUSHER_BITCOIN_RPC_USER` | Bitcoin RPC username. | - |
| `BAL_PUSHER_BITCOIN_RPC_PASSWORD` | Bitcoin RPC password. | - |
## Running `bal-pusher`
Once the application is installed and configured, you can start `bal-pusher` by running the following command:
```bash
$ bal-pusher
```
This will start the service, which will listen for Bitcoin blocks via ZMQ and push transactions from the database when their locktime exceeds the median time past.

11
bal-pusher.env Normal file
View File

@@ -0,0 +1,11 @@
RUST_LOG=info
BAL_PUSHER_DB_FILE=/home/bal/bal.db
BAL_PUSHER_BITCOIN_COOKIE_FILE=/home/bitcoin/.bitcoin/.cookie
BAL_PUSHER_REGTEST_COOKIE_FILE=/home/bitcoin/.bitcoin/regtest/.cookie
BAL_PUSHER_TESTNET_COOKIE_FILE=/home/bitcoin/.bitcoin/testnet3/.cookie
BAL_PUSHER_SIGNET_COOKIE_FILE=/home/bitcoin/.bitcoin/signet/.cookie
BAL_PUSHER_ZMQ_LISTENER=tcp://127.0.0.1:28332

View File

@@ -9,9 +9,7 @@ use sqlite::{Value};
use serde::Serialize;
use serde::Deserialize;
use std::env;
use std::fs::OpenOptions;
use std::io::{ Write};
use log::{info,debug,warn,error};
use log::{info,warn,error,trace,debug};
use zmq::{Context, Socket};
use std::str;
use std::{thread, time::Duration};
@@ -73,7 +71,7 @@ fn get_network_params(cfg: &MyConfig,network:Network)-> &NetworkParams{
fn get_network_params_default(network:Network) -> NetworkParams{
match network {
Network::Testnet => NetworkParams{
host: "http://localhost".to_string(),
host: "http://i27.0.0.1".to_string(),
port: 18332,
dir_path: "testnet3/".to_string(),
db_field: "testnet".to_string(),
@@ -82,7 +80,7 @@ fn get_network_params_default(network:Network) -> NetworkParams{
rpc_pass: "".to_string(),
},
Network::Signet => NetworkParams{
host: "http://localhost".to_string(),
host: "http://127.0.0.1".to_string(),
port: 18332,
dir_path: "signet/".to_string(),
db_field: "signet".to_string(),
@@ -91,7 +89,7 @@ fn get_network_params_default(network:Network) -> NetworkParams{
rpc_pass: "".to_string(),
},
Network::Regtest => NetworkParams{
host: "http://localhost".to_string(),
host: "http://127.0.0.1".to_string(),
port: 18443,
dir_path: "regtest/".to_string(),
db_field: "regtest".to_string(),
@@ -100,7 +98,7 @@ fn get_network_params_default(network:Network) -> NetworkParams{
rpc_pass: "".to_string(),
},
_ => NetworkParams{
host: "http://localhost".to_string(),
host: "http://127.0.0.1".to_string(),
port: 8332,
dir_path: "".to_string(),
db_field: "bitcoin".to_string(),
@@ -117,7 +115,6 @@ fn get_cookie_filename(network: &NetworkParams) ->Result<String,Box<dyn StdError
}else{
match env::var_os("HOME") {
Some(home) => {
info!("some home {}",home.to_str().unwrap());
match home.to_str(){
Some(home_str) => {
let cookie_file_path = format!("{}/.bitcoin/{}.cookie",home_str, network.dir_path);
@@ -148,9 +145,13 @@ fn get_client_from_cookie(url: &String,network: &NetworkParams)->Result<(Client,
match get_cookie_filename(network){
Ok(cookie) => {
match Client::new(&url[..], Auth::CookieFile(cookie.into())) {
Ok(client) => match client.get_blockchain_info(){
Ok(bcinfo) => Ok((client,bcinfo)),
Err(err) => Err(err.into())
Ok(client) => {
match client.get_blockchain_info(){
Ok(bcinfo) => Ok((client,bcinfo)),
Err(err) => {
Err(err.into())
}
}
},
Err(err)=>Err(err.into())
@@ -160,7 +161,7 @@ fn get_client_from_cookie(url: &String,network: &NetworkParams)->Result<(Client,
}
}
fn get_client(network: &NetworkParams) -> Result<(Client,GetBlockchainInfoResult),Box<dyn StdError>>{
let url = format!("{}:{}",network.host,&network.port);
let url = format!("{}:{}/",network.host,&network.port);
match get_client_from_username(&url,network){
Ok(client) =>{Ok(client)},
Err(_) =>{
@@ -200,10 +201,21 @@ fn main_result(cfg: &MyConfig, network_params: &NetworkParams) -> Result<(), Err
//}
//let average_time = time_sum/11;
info!("median time: {}",bcinfo.median_time);
info!("blocks: {}",bcinfo.blocks);
debug!("best block hash: {}",bcinfo.best_block_hash);
let average_time = bcinfo.median_time;
let db = sqlite::open(&cfg.db_file).unwrap();
let query_tx = db.prepare("SELECT * FROM tbl_tx WHERE network = :network AND status = :status AND ( locktime < :bestblock_height OR locktime > :locktime_threshold AND locktime < :bestblock_time);").unwrap().into_iter();
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 = db.prepare(sqlquery).unwrap().into_iter();
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();
@@ -223,25 +235,26 @@ fn main_result(cfg: &MyConfig, network_params: &NetworkParams) -> Result<(), Err
info!("to be pushed: {}: {}",txid, locktime);
match rpc.send_raw_transaction(tx){
Ok(o) => {
let mut file = OpenOptions::new()
/*let mut file = OpenOptions::new()
.append(true) // Set the append option
.create(true) // Create the file if it doesn't exist
.open("valid_txs")?;
let data = format!("{}\t:\t{}\t:\t{}\n",txid,average_time,locktime);
file.write_all(data.as_bytes())?;
drop(file);
*/
info!("tx: {} pusshata PUSHED\n{}",txid,o);
pushed_txs.push(txid.to_string());
},
Err(err) => {
let mut file = OpenOptions::new()
/*let mut file = OpenOptions::new()
.append(true) // Set the append option
.create(true) // Create the file if it doesn't exist
.open("invalid_txs")?;
.open("/home/bal/invalid_txs")?;
let data = format!("{}:\t{}\t:\t{}\t:\t{}\n",txid,err,average_time,locktime);
file.write_all(data.as_bytes())?;
drop(file);
*/
warn!("Error: {}\n{}",err,txid);
//store err in invalid_txs
invalid_txs.insert(txid.to_string(), err.to_string());
@@ -251,12 +264,16 @@ fn main_result(cfg: &MyConfig, network_params: &NetworkParams) -> Result<(), Err
}
if pushed_txs.len() > 0 {
let _ = db.execute(format!("UPDATE tbl_tx SET status = 1 WHERE txid in ('{}');",pushed_txs.join("','")));
let sql = format!("UPDATE tbl_tx SET status = 1 WHERE txid in ('{}');",pushed_txs.join("','"));
trace!("sqlok: {}",&sql);
let _ = db.execute(&sql);
}
if invalid_txs.len() > 0 {
for (txid,txerr) in &invalid_txs{
//let _ = db.execute(format!("UPDATE tbl_tx SET status = 2 WHERE txid in ('{}'Yp);",invalid_txs.join("','")));
let _ = db.execute(format!("UPDATE tbl_tx SET status = 2, push_err='{txerr}' WHERE txid = '{txid}'"));
let sql = format!("UPDATE tbl_tx SET status = 2, push_err='{txerr}' WHERE txid = '{txid}'");
trace!("sqlerror: {}",&sql);
let _ = db.execute(&sql);
}
}
}
@@ -381,7 +398,7 @@ fn main(){
};
debug!("Network: {}",arg_network);
info!("Network: {}",arg_network);
let network_params = get_network_params(&cfg,network);
@@ -389,6 +406,7 @@ fn main(){
let socket: Socket = context.socket(zmq::SUB).unwrap();
let zmq_address = cfg.zmq_listener.clone();
info!("zmq listening on: {}",zmq_address);
socket.connect(&zmq_address).unwrap();
socket.set_subscribe(b"").unwrap();
@@ -411,8 +429,10 @@ fn main(){
let sequence = rdr.read_u32::<LittleEndian>().expect("Failed to read integer");
sequence_str = sequence.to_string();
}*/
debug!("ZMQ:GET TOPIC: {}", String::from_utf8(topic.clone()).expect("invalid topic"));
trace!("ZMQ:GET BODY: {}", hex::encode(&body));
if topic == b"hashblock" {
info!("NEW BLOCK{}", hex::encode(body));
info!("NEW BLOCK: {}", hex::encode(&body));
//let cfg = cfg.clone();
let _ = main_result(&cfg,network_params);
}

View File

@@ -33,7 +33,7 @@ use crate::db::{ create_database, get_next_address_index, insert_xpub, save_new_
#[path = "../xpub.rs"]
mod xpub;
use crate::xpub::new_address_from_xpub;
const VERSION:&str="0.2.0";
const VERSION:&str="0.2.1";
const NETWORKS : [&str; 4]= ["bitcoin","testnet","signet","regtest"];
#[derive(Debug, Clone,Serialize, Deserialize)]
struct NetConfig {
@@ -369,7 +369,7 @@ async fn echo_push(whole_body: &Bytes,
ptx.push((linenum+3,Value::String(line.to_string())));
ptx.push((linenum+4,Value::String(locktime.to_string())));
ptx.push((linenum+5,Value::String(req_time.to_string())));
ptx.push((linenum+6,Value::String(param.to_string())));
ptx.push((linenum+6,Value::String(netconfig.name.to_string())));
ptx.push((linenum+7,Value::String(our_address.to_string())));
ptx.push((linenum+8,Value::String(our_fees.to_string())));
linenum += 9;
@@ -499,7 +499,7 @@ fn parse_env_netconfig<'a>(mut cfg_lock: MutexGuard<'a, MyConfig>, chain: &'a st
}
}
if let Ok(value) = env::var(format!("BAL_SERVER_{}_FIXE_FEE",chain.to_uppercase())) {
if let Ok(value) = env::var(format!("BAL_SERVER_{}_FIXED_FEE",chain.to_uppercase())) {
if let Ok(v) = value.parse::<u64>(){
cfg.fixed_fee = v;
}

View File

@@ -17,6 +17,8 @@ pub fn create_database(db: &Connection){
let _ = db.execute("CREATE UNIQUE INDEX idx_xpub ON tbl_xpub (network, xpub)");
let _ = db.execute("CREATE TABLE IF NOT EXISTS tbl_address (address TEXT PRIMARY_KEY, path TEXT NOT NULL, date_create TIMESTAMP DEFAULT CURRENT_TIMESTAMP, xpub INTEGER,remote_address TEXT);");
let _ = db.execute("UPDATE tbl_tx set network='bitcoin' where network='mainnet');");
}
/*