SAFE21.io/bal-server
1
0
Fork 0
forked from bitcoinafterlife/bal-server
Code Pull Requests Activity

fix: Docker support, WAL race condition, pusher panic fixes

Browse Source 
- Add multi-stage Dockerfile with tini, non-root user, healthcheck
- Fix SQLite WAL mode race between bal-server and bal-pusher (busy_timeout + retry)
- Fix tbl_stats missing UNIQUE index for ON CONFLICT clause
- Replace unwrap() panics in bal-pusher with graceful error handling
- Add docker/entrypoint.sh with BAL_PUSHER_NETWORK support
- cargo fmt across all files
This commit is contained in:
svatantrya
2026-07-16 21:24:09 -04:00
parent 56696050ec
commit fbe61a5862
8 changed files with 355 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
src/db.rs
View File

@@ -1,7 +1,9 @@
use log::{error, info, trace};
use log::{error, info, trace, warn};
use sqlite::{Connection, Error, State, Value};
use std::collections::HashSet;
use std::path::Path;
use std::thread;
use std::time::Duration;
/// Check which txids are already present in the database in a single batch query.
/// Returns a HashSet of txids that already exist (duplicates).
@@ -10,28 +12,28 @@ pub fn check_duplicate_txids(db: &Connection, txids: &[String]) -> Result<HashSe
if txids.is_empty() {
return Ok(HashSet::new());
}
// Build a single query with all txids using IN clause placeholders
// SQLite supports up to 1000 parameters per statement, so we chunk for safety
let mut duplicates = HashSet::new();
let chunk_size = 500; // Safe chunk size for SQLite parameters
for chunk in txids.chunks(chunk_size) {
let placeholders = chunk.iter().map(|_| "?").collect::<Vec<_>>().join(",");
let sql = format!("SELECT txid FROM tbl_tx WHERE txid IN ({})", placeholders);
let mut stmt = db.prepare(sql)?;
for (i, txid) in chunk.iter().enumerate() {
stmt.bind((i + 1, Value::String(txid.clone())))?;
}
while let Ok(State::Row) = stmt.next() {
if let Ok(txid) = stmt.read::<String, _>("txid") {
duplicates.insert(txid);
}
}
}
Ok(duplicates)
}
@@ -61,12 +63,10 @@ pub fn open_db(path: &str) -> Result<Connection, String> {
];
for prefix in &forbidden {
if path_str.starts_with(prefix) {
return Err(
format!(
"Absolute database path under {} is forbidden",
prefix
)
);
return Err(format!(
"Absolute database path under {} is forbidden",
prefix
));
}
}
}
@@ -74,24 +74,48 @@ pub fn open_db(path: &str) -> Result<Connection, String> {
// If file exists, must be a regular file (not a symlink, device, etc.)
if p.exists() {
if p.is_symlink() {
return Err(
"Database path must not be a symlink".to_string()
);
return Err("Database path must not be a symlink".to_string());
}
let metadata = std::fs::metadata(p)
.map_err(|e| format!("Cannot access database file metadata: {}", e))?;
if !metadata.is_file() {
return Err(
"Database path must point to a regular file, not a directory or device".to_string()
"Database path must point to a regular file, not a directory or device".to_string(),
);
}
}
let conn = sqlite::open(path)
.map_err(|e| format!("Failed to open SQLite database: {}", e))?;
let conn = sqlite::open(path).map_err(|e| format!("Failed to open SQLite database: {}", e))?;
// Set busy timeout BEFORE WAL mode so SQLite waits instead of failing immediately.
// This handles the race where two processes (server + pusher) open the same DB
// and both try to enable WAL mode concurrently.
conn.execute("PRAGMA busy_timeout = 5000;")
.map_err(|e| format!("Failed to set busy_timeout: {}", e))?;
// Retry WAL mode up to 5 times (handles concurrent open from bal-pusher).
let mut wal_ok = false;
for attempt in 0..5 {
match conn.execute("PRAGMA journal_mode = WAL;") {
Ok(_) => {
wal_ok = true;
break;
}
Err(e) => {
warn!(
"WAL mode attempt {}/5 failed: {}, retrying in 100ms...",
attempt + 1,
e
);
thread::sleep(Duration::from_millis(100));
}
}
}
if !wal_ok {
// WAL might already be enabled by another process; this is not fatal.
warn!("Could not set WAL mode after retries — may already be enabled by another process");
}
conn.execute("PRAGMA journal_mode = WAL;")
.map_err(|e| format!("Failed to enable WAL mode: {}", e))?;
conn.execute("PRAGMA synchronous = NORMAL;")
.map_err(|e| format!("Failed to set synchronous NORMAL: {}", e))?;
@@ -102,13 +126,15 @@ pub fn open_db(path: &str) -> Result<Connection, String> {
/// in-memory lookup during transaction validation (replaces N+1 query).
pub fn get_all_addresses_by_xpub(db: &Connection, xpub: &str) -> Result<HashSet<String>, Error> {
let mut stmt = db.prepare(
"SELECT a.address FROM tbl_address a JOIN tbl_xpub x ON a.xpub = x.id WHERE x.xpub = ?"
"SELECT a.address FROM tbl_address a JOIN tbl_xpub x ON a.xpub = x.id WHERE x.xpub = ?",
)?;
stmt.bind((1, Value::String(xpub.to_string())))?;
let mut addresses = HashSet::new();
while let Ok(State::Row) = stmt.next() {
match stmt.read::<String, _>("address") {
Ok(addr) => { addresses.insert(addr); }
Ok(addr) => {
addresses.insert(addr);
}
Err(e) => {
error!("Failed to read address column: {}", e);
}
@@ -134,7 +160,9 @@ pub fn create_database(db: &Connection) {
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("CREATE TABLE IF NOT EXISTS tbl_stats (report_date TEXT, chain TEXT, totals INTEGER, waiting INTEGER, sent INTEGER, failed INTEGER, waiting_profit INTEGER, sent_profit INTEGER, missed_profit INTEGER, unique_inputs INTEGER);");
let _ = db.execute("CREATE INDEX IF NOT EXISTS idx_stats_chain ON tbl_stats(chain);");
// UNIQUE index required for ON CONFLICT(chain) DO UPDATE in calculate_stats
let _ = db.execute("DROP INDEX IF EXISTS idx_stats_chain;");
let _ = db.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_stats_chain ON tbl_stats(chain);");
let _ = db.execute("UPDATE tbl_tx set network='bitcoin' where network='mainnet');");
}