worker.rs (23261B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2022-2025, 2026 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU Affero General Public License as published by the Free Software 7 Foundation; either version 3, or (at your option) any later version. 8 9 TALER is distributed in the hope that it will be useful, but WITHOUT ANY 10 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 11 A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. 12 13 You should have received a copy of the GNU Affero General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 use std::{fmt::Write, time::SystemTime}; 17 18 use bitcoin::{Amount as BtcAmount, Txid, hashes::Hash}; 19 use depolymerizer_common::metadata::OutMetadata; 20 use failure_injection::fail_point; 21 use jiff::Timestamp; 22 use sqlx::{ 23 Acquire, Either, PgConnection, PgPool, 24 postgres::{PgAdvisoryLock, PgAdvisoryLockKey, PgListener}, 25 }; 26 use taler_api::subject::IncomingSubject; 27 use taler_common::{ExpoBackoffDecorr, api::ShortHashCode, db::IncomingType}; 28 use tokio::time::sleep; 29 use tracing::{debug, error, info, trace, warn}; 30 31 use super::{LoopError, LoopResult, analysis::analysis}; 32 use crate::{ 33 GetOpReturnErr, RpcApiExtended as _, 34 config::WorkerCfg, 35 db::{self, AddIncomingResult, SyncOutState, TxOut, TxOutKind}, 36 rpc::{ 37 self, Category, 38 ErrorCode::{self, RpcInvalidAddressOrKey}, 39 ListSinceBlock, ListTransaction, Rpc, RpcApi, 40 }, 41 taler_utils::btc_to_taler, 42 }; 43 44 pub async fn worker_loop(mut state: WorkerCfg, pool: PgPool) { 45 let mut jitter = ExpoBackoffDecorr::default(); 46 let mut lifetime = state.lifetime; 47 let mut status = true; 48 let mut skip_notification = true; 49 50 loop { 51 let result: LoopResult<()> = async { 52 // Connect 53 let rpc = &mut Rpc::new(&state.rpc_cfg).await?; 54 rpc.load_wallet(&state.wallet_cfg.name).await?; 55 if let Some(password) = &state.wallet_cfg.password { 56 rpc.unlock_wallet(password).await?; 57 } 58 let rpc = &mut rpc.wallet(&state.wallet_cfg.name); 59 let db = &mut PgListener::connect_with(&pool).await?; 60 61 // Listen to all channels 62 db.listen_all(["new_block", "transfer"]).await?; 63 64 loop { 65 // Wait for the next notification 66 { 67 let ntf = db.next_buffered(); 68 if let Some(ntf) = &ntf { 69 trace!(target: "worker", "notification from {}", ntf.channel()) 70 } 71 if !skip_notification && ntf.is_none() { 72 debug!(target: "worker", "waiting for notifications"); 73 // Block until next notification 74 if let Some(ntf) = db.try_recv().await? { 75 trace!(target: "worker", "notification from {}", ntf.channel()) 76 } 77 } 78 // Conflate all notifications 79 while let Some(ntf) = db.next_buffered() { 80 trace!(target: "worker", "notification from {}", ntf.channel()) 81 } 82 } 83 84 // Check lifetime 85 if let Some(nb) = lifetime.as_mut() { 86 if *nb == 0 { 87 info!(target: "worker", "Reach end of lifetime"); 88 return Ok(()); 89 } else { 90 *nb -= 1; 91 } 92 } 93 94 debug!(target: "worker", "syncing blockchain"); 95 96 let mut db = db.acquire().await?; 97 98 // It is not possible to atomically update the blockchain and the database. 99 // When we failed to sync the database and the blockchain state we rely on 100 // sync_chain to recover the lost updates. 101 // When this function is running concurrently, it not possible to known another 102 // execution has failed, and this can lead to a transaction being sent multiple time. 103 // To ensure only a single version of this function is running at a given time we rely 104 // on postgres advisory lock 105 106 // Take the lock 107 let lock = PgAdvisoryLock::with_key(PgAdvisoryLockKey::BigInt(42)); 108 let Either::Left(mut lock) = lock.try_acquire(&mut db).await? else { 109 return Err(LoopError::Concurrency); 110 }; 111 112 // Perform analysis 113 state.conf = analysis(rpc, state.conf, state.max_conf).await?; 114 115 worker_step(lock.as_mut(), rpc, &mut state, &mut status).await?; 116 117 skip_notification = false; 118 jitter.reset(); 119 } 120 } 121 .await; 122 if let Err(e) = result { 123 error!(target: "worker", "{e}"); 124 // When we catch an error, we sometimes want to retry immediately (eg. reconnect to RPC or DB). 125 // Bitcoin error codes are generic. We need to match the msg to get precise ones. Some errors 126 // can resolve themselves when a new block is mined (new fees, new transactions). Our simple 127 // approach is to wait for the next loop when an RPC error is caught to prevent endless logged errors. 128 skip_notification = match e { 129 LoopError::DB(_) | LoopError::Injected(_) | LoopError::Concurrency => true, 130 LoopError::Rpc(e) => match e { 131 rpc::Error::Transport(_) 132 | rpc::Error::Cookie(_, _) 133 | rpc::Error::Tcp(_, _) 134 | rpc::Error::Elapsed(_, _) => true, 135 rpc::Error::RPC { code, .. } => code == ErrorCode::RpcWalletError, 136 rpc::Error::Bitcoin(_) | rpc::Error::Json { .. } | rpc::Error::Null => false, 137 }, 138 }; 139 sleep(jitter.backoff()).await; 140 } else { 141 return; 142 } 143 } 144 } 145 146 pub async fn worker_transient(mut state: WorkerCfg, pool: PgPool) -> LoopResult<()> { 147 let mut status = true; 148 149 // Connect 150 let rpc = &mut Rpc::new(&state.rpc_cfg).await?; 151 rpc.load_wallet(&state.wallet_cfg.name).await?; 152 if let Some(password) = &state.wallet_cfg.password { 153 rpc.unlock_wallet(password).await?; 154 } 155 let rpc = &mut rpc.wallet(&state.wallet_cfg.name); 156 let mut db = pool.acquire().await?; 157 158 // It is not possible to atomically update the blockchain and the database. 159 // When we failed to sync the database and the blockchain state we rely on 160 // sync_chain to recover the lost updates. 161 // When this function is running concurrently, it not possible to known another 162 // execution has failed, and this can lead to a transaction being sent multiple time. 163 // To ensure only a single version of this function is running at a given time we rely 164 // on postgres advisory lock 165 166 // Take the lock 167 let lock = PgAdvisoryLock::with_key(PgAdvisoryLockKey::BigInt(42)); 168 let Either::Left(mut lock) = lock.try_acquire(&mut db).await? else { 169 return Err(LoopError::Concurrency); 170 }; 171 172 worker_step(lock.as_mut(), rpc, &mut state, &mut status).await?; 173 Ok(()) 174 } 175 176 /// Synchronize local db with blockchain and perform transactions 177 pub async fn worker_step( 178 db: &mut PgConnection, 179 rpc: &mut impl RpcApi, 180 state: &mut WorkerCfg, 181 status: &mut bool, 182 ) -> LoopResult<()> { 183 // Sync chain 184 if let Some(stuck) = sync_chain(db, rpc, state, status).await? { 185 // As we are now in sync with the blockchain if a transaction has Requested status it have not been sent 186 187 // Send requested debits 188 while debit(db, rpc, state).await? {} 189 190 // Bump stuck transactions 191 for (txid, wtid) in stuck { 192 let bump = rpc.bump_fee(&txid).await?; 193 failure_injection::fail_point("bumpfee")?; 194 db::transfer_bumpfee(&mut *db, &bump.txid, &wtid).await?; 195 info!(target: "worker", ">> bump {wtid} {txid} -> {}", bump.txid); 196 } 197 198 // Send requested bounce 199 while bounce(db, rpc, &state.bounce_fee).await? {} 200 } 201 Ok(()) 202 } 203 204 /// Parse new transactions, return stuck transactions if the database is up to date with the latest mined block 205 async fn sync_chain( 206 db: &mut PgConnection, 207 rpc: &mut impl RpcApi, 208 state: &WorkerCfg, 209 status: &mut bool, 210 ) -> LoopResult<Option<Vec<(Txid, ShortHashCode)>>> { 211 // Get stored last_hash 212 let sync_state = db::get_sync_state(&mut *db).await?; 213 214 // Get all transactions made since this block 215 let ListSinceBlock { 216 mut transactions, 217 mut removed, 218 lastblock, 219 } = rpc.list_since_block(Some(&sync_state), state.conf).await?; 220 transactions.sort_unstable_by_key(|it| (it.confirmations, it.txid)); 221 transactions.dedup_by_key(|it| it.txid); 222 removed.sort_unstable_by_key(|it| (it.confirmations, it.txid)); 223 removed.dedup_by_key(|it| it.txid); 224 225 // Check if a confirmed incoming transaction have been removed by a blockchain reorganization 226 let conflict = sync_chain_removed(removed, db, state.conf as i32).await?; 227 228 // Sync server status with database 229 let new_status = !conflict.stop_server(); 230 if *status != new_status { 231 db::update_status(db, new_status).await?; 232 *status = new_status; 233 if new_status { 234 info!(target: "worker", "Recovered lost transactions"); 235 } 236 } 237 238 if conflict.stop_worker() { 239 return Ok(None); 240 } 241 242 let mut stuck = vec![]; 243 244 for tx in transactions { 245 match tx.category { 246 Category::Send => { 247 if let Some(wtid) = 248 sync_chain_outgoing(&tx.txid, tx.confirmations, rpc, db, state).await? 249 { 250 stuck.push((tx.txid, wtid)); 251 } 252 } 253 Category::Receive if tx.confirmations >= state.conf as i32 => { 254 sync_chain_incoming_confirmed(&tx.txid, rpc, db, state).await? 255 } 256 _ => { 257 // Ignore coinbase and unconfirmed send transactions 258 } 259 } 260 } 261 262 // Move last_hash forward 263 db::swap_sync_state(db, &sync_state, &lastblock).await?; 264 265 Ok(Some(stuck)) 266 } 267 268 #[derive(Debug, Clone, Copy)] 269 enum ReorgConflict { 270 BackingCompromised, 271 IncomingCompromised, 272 Ok, 273 } 274 275 impl ReorgConflict { 276 pub fn stop_server(&self) -> bool { 277 matches!(self, Self::BackingCompromised) 278 } 279 280 pub fn stop_worker(&self) -> bool { 281 matches!(self, Self::BackingCompromised | Self::IncomingCompromised) 282 } 283 } 284 285 /// Sync database with removed transactions, return false if bitcoin backing is compromised 286 async fn sync_chain_removed( 287 removed: Vec<ListTransaction>, 288 db: &mut PgConnection, 289 min_confirmations: i32, 290 ) -> LoopResult<ReorgConflict> { 291 let potential_problematic_ids: Vec<Txid> = removed 292 .into_iter() 293 .filter_map(|tx| { 294 (tx.category == Category::Receive && tx.confirmations < min_confirmations) 295 .then_some(tx.txid) 296 }) 297 .collect(); 298 299 // Only keep incoming transaction that are not reconfirmed 300 let problematic_tx = db::reorg(&mut *db, &potential_problematic_ids).await?; 301 if problematic_tx.is_empty() { 302 return Ok(ReorgConflict::Ok); 303 } 304 // Bitcoin backing can be compromised in only two cases: 305 // - a confirmed reserve 306 // - a confirmed bounced 307 308 // TODO use partition_in_place when stable 309 let (compromise, problematic): (Vec<_>, Vec<_>) = 310 problematic_tx.iter().partition(|it| match it { 311 db::ProblematicTx::Taler { ty, .. } => *ty == IncomingType::reserve, 312 db::ProblematicTx::Bounce { .. } => true, 313 db::ProblematicTx::Simple { .. } => false, 314 }); 315 let mut buf = "The following transaction have been removed from the blockchain, ".to_string(); 316 let (txs, state) = if compromise.is_empty() { 317 buf.push_str("waiting until they reappear:"); 318 (problematic, ReorgConflict::IncomingCompromised) 319 } else { 320 buf.push_str("bitcoin backing is compromised until they reappear:"); 321 (compromise, ReorgConflict::BackingCompromised) 322 }; 323 for tx in txs { 324 match tx { 325 db::ProblematicTx::Taler { 326 txid, 327 addr, 328 ty, 329 metadata, 330 } => { 331 write!(&mut buf, "\n\t{txid} {ty} {metadata} from {addr}",).unwrap(); 332 } 333 db::ProblematicTx::Bounce { txid, bounced_in } => { 334 write!(&mut buf, "\n\t{txid} bounced in {bounced_in}").unwrap(); 335 } 336 db::ProblematicTx::Simple { txid } => { 337 write!(&mut buf, "\n\t{txid}").unwrap(); 338 } 339 } 340 } 341 error!(target: "worker", "{buf}"); 342 Ok(state) 343 } 344 345 /// Sync database with an incoming confirmed transaction 346 async fn sync_chain_incoming_confirmed( 347 txid: &Txid, 348 rpc: &mut impl RpcApi, 349 db: &mut PgConnection, 350 state: &WorkerCfg, 351 ) -> Result<(), LoopError> { 352 let (tx, metadata) = rpc.get_tx_segwit_key(txid).await?; 353 // Store transactions in database 354 let debit_addr = rpc.sender_address(&tx).await?; 355 let amount = btc_to_taler(&tx.amount, &state.currency); 356 let time = Timestamp::from_second(tx.time as i64).unwrap(); 357 let ty = IncomingType::reserve; 358 match metadata { 359 Ok(reserve_pub) => { 360 match db::register_tx_in( 361 &mut *db, 362 txid, 363 &amount, 364 &debit_addr, 365 &Timestamp::from_second(tx.time as i64).unwrap(), 366 &Some(IncomingSubject::Reserve(reserve_pub)), 367 ) 368 .await? 369 { 370 AddIncomingResult::Success { 371 new, 372 row_id: _, 373 valued_at: _, 374 pending: _, 375 } => { 376 if new { 377 info!(target: "worker", "<< {ty} {reserve_pub} {txid} {debit_addr} {amount}"); 378 } 379 } 380 AddIncomingResult::ReservePubReuse => { 381 db::bounce(db, txid, &amount, &debit_addr, &time, "reserve_pub reuse").await? 382 } 383 AddIncomingResult::UnknownMapping => todo!(), 384 AddIncomingResult::MappingReuse => todo!(), 385 } 386 } 387 Err(e) => db::bounce(db, txid, &amount, &debit_addr, &time, &e.to_string()).await?, 388 } 389 Ok(()) 390 } 391 392 /// Sync database with an outgoing transaction, return true if stuck 393 async fn sync_chain_outgoing( 394 txid: &Txid, 395 confirmations: i32, 396 rpc: &mut impl RpcApi, 397 db: &mut PgConnection, 398 state: &WorkerCfg, 399 ) -> LoopResult<Option<ShortHashCode>> { 400 let confirmed = confirmations >= state.conf as i32; 401 let chain = if confirmed { "onchain" } else { "sent" }; 402 match rpc 403 .get_tx_op_return(txid) 404 .await 405 .map(|(tx, bytes)| (tx, OutMetadata::decode(&bytes))) 406 { 407 Ok((tx, Ok(info))) => { 408 let credit_addr = tx.details[0].address.clone().unwrap().assume_checked(); 409 let amount = btc_to_taler(&tx.amount, &state.currency); 410 let created_at = Timestamp::from_second(tx.time as i64).unwrap(); 411 match info { 412 OutMetadata::Debit { 413 wtid, 414 url, 415 metadata, 416 } => { 417 if confirmations < 0 { 418 // Handle conflict 419 if db::transfer_conflict(db, txid).await? { 420 warn!(target: "worker", ">> conflict {wtid} {txid} {credit_addr} {amount}"); 421 } 422 } else { 423 // Sync db state 424 match db::sync_out( 425 db, 426 &TxOut { 427 id: *txid, 428 replaces_txid: tx.replaces_txid, 429 amount, 430 credit_acc: &credit_addr, 431 block_time: created_at, 432 }, 433 &TxOutKind::Talerable { 434 wtid: &wtid, 435 url: &url, 436 metadata: metadata.as_deref(), 437 }, 438 confirmed, 439 ) 440 .await? 441 .state 442 { 443 SyncOutState::New => { 444 info!(target: "worker", ">> {chain} {wtid} {txid} {credit_addr} {amount}"); 445 } 446 SyncOutState::Replaced => { 447 warn!( 448 target: "worker", 449 ">> recovered {chain} {wtid} {txid} -> {} {credit_addr} {amount}", 450 tx.replaces_txid.unwrap() 451 ) 452 } 453 SyncOutState::Recovered => { 454 warn!(target: "worker", ">> recovered {chain} {wtid} {txid} {credit_addr} {amount}") 455 } 456 SyncOutState::None => {} 457 } 458 if let Some(delay) = state.bump_delay 459 && confirmations == 0 460 && tx.replaced_by_txid.is_none() 461 { 462 // Check stuck 463 let now = SystemTime::now() 464 .duration_since(SystemTime::UNIX_EPOCH) 465 .unwrap() 466 .as_secs(); 467 if now - tx.time >= delay as u64 { 468 return Ok(Some(wtid)); 469 } 470 } 471 } 472 } 473 OutMetadata::Bounce { bounced } => { 474 let bounced = Txid::from_byte_array(bounced); 475 if confirmations < 0 { 476 // Handle conflict 477 if db::bounce_conflict(db, txid).await? { 478 warn!(target: "worker", "|| conflict {bounced} {txid}"); 479 } 480 } else { 481 // Sync db state 482 match db::sync_out( 483 db, 484 &TxOut { 485 id: *txid, 486 replaces_txid: tx.replaces_txid, 487 amount, 488 credit_acc: &credit_addr, 489 block_time: created_at, 490 }, 491 &TxOutKind::Bounce(bounced), 492 confirmed, 493 ) 494 .await? 495 .state 496 { 497 SyncOutState::New => { 498 info!(target: "worker", "|| {chain} {bounced} {txid}") 499 } 500 SyncOutState::Replaced => { 501 warn!( 502 target: "worker", 503 "|| recovered {chain} {bounced} {txid} -> {}", 504 tx.replaced_by_txid.unwrap() 505 ) 506 } 507 SyncOutState::Recovered => { 508 warn!(target: "worker", "|| recovered {chain} {bounced} {txid}") 509 } 510 SyncOutState::None => {} 511 } 512 } 513 } 514 } 515 } 516 Ok((_, Err(e))) => warn!(target: "worker", "send: decode-info {txid} - {e}"), 517 Err(e) => match e { 518 GetOpReturnErr::MissingOpReturn => { /* Ignore */ } 519 GetOpReturnErr::RPC(e) => return Err(e)?, 520 }, 521 } 522 Ok(None) 523 } 524 525 /// Send a debit transaction on the blockchain, return false if no more requested transactions are found 526 async fn debit( 527 db: &mut PgConnection, 528 rpc: &mut impl RpcApi, 529 state: &WorkerCfg, 530 ) -> LoopResult<bool> { 531 // We rely on the advisory lock to ensure we are the only one sending transactions 532 if let Some((id, amount, wtid, addr, url, metadata)) = 533 db::pending_transfer(&mut *db, &state.currency).await? 534 { 535 let metadata = OutMetadata::Debit { 536 wtid, 537 url, 538 metadata, 539 }; 540 541 let txid = match rpc 542 .send(&addr, amount, Some(&metadata.encode().unwrap()), false) 543 .await 544 { 545 Ok(id) => id, 546 547 Err(e) => match &e { 548 rpc::Error::RPC { code, msg, .. } if *code == RpcInvalidAddressOrKey => { 549 let reason = format!("{code:?} {msg}"); 550 db::transfer_ignored(db, id, &reason).await?; 551 warn!(target: "worker", ">> ignored {wtid} {addr} {amount}: {reason}"); 552 return Ok(true); 553 } 554 _ => return Err(e.into()), 555 }, 556 }; 557 fail_point("debit")?; 558 db::transfer_sent(db, id, &txid).await?; 559 let amount = btc_to_taler(&amount.to_signed().unwrap(), &state.currency); 560 info!(target: "worker", ">> sent {wtid} {txid} {addr} {amount}"); 561 Ok(true) 562 } else { 563 Ok(false) 564 } 565 } 566 567 /// Bounce a transaction on the blockchain, return false if no more requested transactions are found 568 async fn bounce(db: &mut PgConnection, rpc: &mut impl RpcApi, fee: &BtcAmount) -> LoopResult<bool> { 569 // We rely on the advisory lock to ensure we are the only one sending transactions 570 if let Some((id, bounced, reason)) = db::pending_bounce(&mut *db).await? { 571 let metadata = OutMetadata::Bounce { 572 bounced: *bounced.as_byte_array(), 573 }; 574 575 match rpc 576 .bounce(&bounced, fee, Some(&metadata.encode().unwrap())) 577 .await 578 { 579 Ok(txid) => { 580 fail_point("bounce")?; 581 db::bounce_sent(db, id, &txid).await?; 582 info!(target: "worker", "|| sent {bounced} {txid}: {reason}"); 583 } 584 Err(err) => match err { 585 rpc::Error::RPC { 586 code: ErrorCode::RpcWalletInsufficientFunds | ErrorCode::RpcWalletError, 587 .. 588 } => { 589 warn!(target: "worker", "{err}"); 590 return Ok(false); 591 } 592 e => Err(e)?, 593 }, 594 } 595 Ok(true) 596 } else { 597 Ok(false) 598 } 599 }