depolymerization

worker.rs (18398B)

---

 /*
   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/>
 */
 use std::{fmt::Write, time::SystemTime};
 
 use bitcoin::{Amount as BtcAmount, Txid, hashes::Hash};
 use depolymerizer_common::{metadata::OutMetadata, status::BounceStatus};
 use sqlx::{PgPool, postgres::PgListener};
 use taler_common::{ExpoBackoffDecorr, api_common::ShortHashCode, types::timestamp::Timestamp};
 use tokio::time::sleep;
 use tracing::{debug, error, info, trace, warn};
 use url::Url;
 
 use crate::{
     GetOpReturnErr, GetSegwitErr,
     config::WorkerCfg,
     db::{self, AddIncomingResult, SyncBounceResult, SyncOutResult, worker_lock},
     fail_point::fail_point,
     rpc::{
         self, Category, ErrorCode, ListSinceBlock, ListTransaction, Rpc, Transaction, rpc_wallet,
     },
     rpc_utils::sender_address,
     taler_utils::btc_to_taler,
 };
 
 use super::{LoopError, LoopResult, analysis::analysis};
 
 pub async fn worker_loop(mut state: WorkerCfg, pool: PgPool) {
     let mut jitter = ExpoBackoffDecorr::default();
     let mut lifetime = state.lifetime;
     let mut status = true;
     let mut skip_notification = true;
 
     loop {
         let result: LoopResult<()> = async {
             // Connect
             let rpc = &mut rpc_wallet(&state.rpc_cfg, &state.wallet_cfg).await?;
             let db = &mut PgListener::connect_with(&pool).await?;
 
             // It is not possible to atomically update the blockchain and the database.
             // When we failed to sync the database and the blockchain state we rely on
             // sync_chain to recover the lost updates.
             // When this function is running concurrently, it not possible to known another
             // execution has failed, and this can lead to a transaction being sent multiple time.
             // To ensure only a single version of this function is running at a given time we rely
             // on postgres advisory lock
 
             // Take the lock
             if !worker_lock(&mut *db).await? {
                 return Err(LoopError::Concurrency);
             }
 
             // Listen to all channels
             db.listen_all(["new_block", "taler_out"]).await?;
 
             loop {
                 // Wait for the next notification
                 {
                     let ntf = db.next_buffered();
                     if let Some(ntf) = &ntf {
                         trace!(target: "worker", "notification from {}", ntf.channel())
                     }
                     if !skip_notification && ntf.is_none() {
                         debug!(target: "worker", "waiting for notifications");
                         // Block until next notification
                         if let Some(ntf) = db.try_recv().await? {
                             trace!(target: "worker", "notification from {}", ntf.channel())
                         }
                     }
                     // Conflate all notifications
                     while let Some(ntf) = db.next_buffered() {
                         trace!(target: "worker", "notification from {}", ntf.channel())
                     }
                 }
 
                 // Check lifetime
                 if let Some(nb) = lifetime.as_mut() {
                     if *nb == 0 {
                         info!(target: "worker", "Reach end of lifetime");
                         return Ok(());
                     } else {
                         *nb -= 1;
                     }
                 }
 
                 debug!(target: "worker", "syncing blockchain");
 
                 // Perform analysis
                 state.confirmation =
                     analysis(rpc, state.confirmation, state.max_confirmation).await?;
 
                 worker_step(rpc, db, &mut state, &mut status).await?;
 
                 skip_notification = false;
                 jitter.reset();
             }
         }
         .await;
         if let Err(e) = result {
             error!(target: "worker", "{e}");
             // When we catch an error, we sometimes want to retry immediately (eg. reconnect to RPC or DB).
             // Bitcoin error codes are generic. We need to match the msg to get precise ones. Some errors
             // can resolve themselves when a new block is mined (new fees, new transactions). Our simple
             // approach is to wait for the next loop when an RPC error is caught to prevent endless logged errors.
             skip_notification = matches!(
                 e,
                 LoopError::Rpc(rpc::Error::Transport(_))
                     | LoopError::DB(_)
                     | LoopError::Injected(_)
                     | LoopError::Concurrency
             );
             sleep(jitter.backoff()).await;
         } else {
             return;
         }
     }
 }
 
 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 db = &mut PgListener::connect_with(&pool).await?;
 
     // It is not possible to atomically update the blockchain and the database.
     // When we failed to sync the database and the blockchain state we rely on
     // sync_chain to recover the lost updates.
     // When this function is running concurrently, it not possible to known another
     // execution has failed, and this can lead to a transaction being sent multiple time.
     // To ensure only a single version of this function is running at a given time we rely
     // on postgres advisory lock
 
     // Take the lock
     if !worker_lock(&mut *db).await? {
         return Err(LoopError::Concurrency);
     }
 
     worker_step(rpc, db, &mut state, &mut status).await?;
     Ok(())
 }
 
 /// Synchronize local db with blockchain and perform transactions
 async fn worker_step(
     rpc: &mut Rpc,
     db: &mut PgListener,
     state: &mut WorkerCfg,
     status: &mut bool,
 ) -> LoopResult<()> {
     // Sync chain
     if let Some(stuck) = sync_chain(rpc, db, 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
         while debit(db, rpc, state).await? {}
 
         // Bump stuck transactions
         for id in stuck {
             let bump = rpc.bump_fee(&id).await?;
             fail_point("(injected) fail bump", 0.3)?;
             let wtid = db::bump_tx_id(&mut *db, &id, &bump.txid).await?;
             info!(target: "worker", ">> (bump) {wtid} replace {id} with {}", bump.txid);
         }
 
         // Send requested bounce
         while bounce(db, rpc, &state.bounce_fee).await? {}
     }
     Ok(())
 }
 
 /// 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 PgListener,
     state: &WorkerCfg,
     status: &mut bool,
 ) -> LoopResult<Option<Vec<Txid>>> {
     // Get stored last_hash
     let sync_state = db::get_sync_state(&mut *db).await?;
     // Get the current confirmation delay
     let conf_delay = state.confirmation;
 
     // Get all transactions made since this block
     let ListSinceBlock {
         transactions,
         removed,
         lastblock,
     } = rpc.list_since_block(Some(&sync_state), conf_delay).await?;
 
     // Check if a confirmed incoming transaction have been removed by a blockchain reorganization
     let new_status = sync_chain_removed(removed, db, conf_delay as i32).await?;
 
     // Sync status with database
     if *status != new_status {
         db::update_status(db, new_status).await?;
         *status = new_status;
         if new_status {
             info!(target: "worker", "Recovered lost transactions");
         }
     }
     if !new_status {
         return Ok(None);
     }
 
     let mut stuck = vec![];
 
     for tx in transactions {
         match tx.category {
             Category::Send => {
                 if sync_chain_outgoing(&tx.txid, tx.confirmations, rpc, db, state).await? {
                     stuck.push(tx.txid);
                 }
             }
             Category::Receive if tx.confirmations >= conf_delay as i32 => {
                 sync_chain_incoming_confirmed(&tx.txid, rpc, db, state).await?
             }
             _ => {
                 // Ignore coinbase and unconfirmed send transactions
             }
         }
     }
 
     // Move last_hash forward
     db::swap_sync_state(db, &sync_state, &lastblock).await?;
 
     Ok(Some(stuck))
 }
 
 /// Sync database with removed transactions, return false if bitcoin backing is compromised
 async fn sync_chain_removed(
     removed: Vec<ListTransaction>,
     db: &mut PgListener,
     min_confirmations: i32,
 ) -> LoopResult<bool> {
     // A removed incoming transaction is a correctness issues in only two cases:
     // - it is a confirmed credit registered in the database
     // - it is an invalid transactions already bounced
     // Those two cases can compromise bitcoin backing
     // Removed outgoing transactions will be retried automatically by the node
 
     let potential_problematic_ids: Vec<Txid> = removed
         .into_iter()
         .filter_map(|tx| {
             (tx.category == Category::Receive && tx.confirmations < min_confirmations)
                 .then_some(tx.txid)
         })
         .collect();
 
     // Only keep incoming transaction that are not reconfirmed
     let problematic_tx = db::reorg(&mut *db, &potential_problematic_ids).await?;
     if problematic_tx.is_empty() {
         return Ok(true);
     }
     let mut buf = "The following transaction have been removed from the blockchain, bitcoin backing is compromised until the transaction reappear:".to_string();
     for tx in problematic_tx {
         match tx {
             db::ProblematicTx::In {
                 txid,
                 addr,
                 reserve_pub,
             } => {
                 write!(&mut buf, "\n\tcredit {reserve_pub} in {txid} from {addr}",).unwrap();
             }
             db::ProblematicTx::Bounce { txid, bounced } => {
                 write!(&mut buf, "\n\tbounced {txid} in {bounced}").unwrap();
             }
         }
     }
     error!(target: "worker", "{buf}");
     Ok(false)
 }
 
 /// Sync database with an incoming confirmed transaction
 async fn sync_chain_incoming_confirmed(
     id: &Txid,
     rpc: &mut Rpc,
     db: &mut PgListener,
     state: &WorkerCfg,
 ) -> Result<(), LoopError> {
     match rpc.get_tx_segwit_key(id).await {
         Ok((full, reserve_pub)) => {
             // Store transactions in database
             let debit_addr = sender_address(rpc, &full).await?;
             let amount = btc_to_taler(&full.amount, &state.currency);
             match db::register_tx_in(
                 &mut *db,
                 id,
                 &amount,
                 &debit_addr,
                 &Timestamp::from_sql_micros((full.time * 1000000) as i64).unwrap(),
                 &reserve_pub,
             )
             .await?
             {
                 AddIncomingResult::Success {
                     new,
                     row_id: _,
                     valued_at: _,
                 } => {
                     if new {
                         info!(target: "worker", "<< {amount} {reserve_pub} in {id} from {debit_addr}");
                     }
                 }
                 AddIncomingResult::ReservePubReuse => {
                     db::bounce(db, id, "reserve_pub reuse").await?
                 }
             }
         }
         Err(err) => match err {
             GetSegwitErr::Decode(e) => db::bounce(db, id, &e.to_string()).await?,
             GetSegwitErr::RPC(e) => return Err(e.into()),
         },
     }
     Ok(())
 }
 
 /// Sync database with a debit transaction, return true if stuck
 async fn sync_chain_debit(
     txid: &Txid,
     full: &Transaction,
     wtid: &ShortHashCode,
     exchange_url: &Url,
     db: &mut PgListener,
     confirmations: i32,
     state: &WorkerCfg,
 ) -> LoopResult<bool> {
     let credit_addr = full.details[0].address.clone().unwrap().assume_checked();
     let amount = btc_to_taler(&full.amount, &state.currency);
 
     if confirmations < 0 {
         // Handle conflicting tx
         if full.replaced_by_txid.is_none() && db::conflict_tx_out(db, txid).await? {
             warn!(target: "worker", ">> (conflict) {wtid} in {txid} to {credit_addr}");
         }
     } else {
         match db::sync_out(
             db,
             txid,
             full.replaced_by_txid.as_ref(),
             &amount,
             exchange_url,
             &credit_addr,
             wtid,
             &Timestamp::from_sql_micros((full.time * 1000000) as i64).unwrap(),
         )
         .await?
         {
             SyncOutResult::New => {
                 warn!(target: "worker", ">> (onchain) {amount} {wtid} in {txid} to {credit_addr}");
             }
             SyncOutResult::Replaced => {
                 info!(
                     target: "worker",
                     ">> (recovered) {wtid} replace {txid} with {}",
                     full.replaced_by_txid.unwrap()
                 )
             }
             SyncOutResult::Recovered => {
                 warn!(target: "worker", ">> (recovered) {amount} {wtid} in {txid} to {credit_addr}")
             }
             SyncOutResult::None => {}
         }
 
         // Check if stuck
         if let Some(delay) = state.bump_delay
             && confirmations == 0
             && full.replaced_by_txid.is_none()
         {
             let now = SystemTime::now()
                 .duration_since(SystemTime::UNIX_EPOCH)
                 .unwrap()
                 .as_secs();
             if now - full.time > delay as u64 {
                 return Ok(true);
             }
         }
     }
     Ok(false)
 }
 
 /// Sync database with an outgoing bounce transaction
 async fn sync_chain_bounce(
     txid: &Txid,
     bounced: &Txid,
     db: &mut PgListener,
     confirmations: i32,
 ) -> LoopResult<()> {
     if confirmations < 0 {
         // Handle conflicting tx
         if db::conflict_bounce(db, txid).await? {
             warn!(target: "worker", "|| (conflict) {bounced} in {txid}");
         }
     } else {
         match db::sync_bounce(db, txid, bounced, &Timestamp::now()).await? {
             SyncBounceResult::New => warn!(target: "worker", "|| (onchain) {bounced} in {txid}"),
             SyncBounceResult::Recovered => {
                 warn!(target: "worker", "|| (recovered) {bounced} in {txid}")
             }
             SyncBounceResult::None => {}
         }
     }
 
     Ok(())
 }
 
 /// Sync database with an outgoing transaction, return true if stuck
 async fn sync_chain_outgoing(
     id: &Txid,
     confirmations: i32,
     rpc: &mut Rpc,
     db: &mut PgListener,
     state: &WorkerCfg,
 ) -> LoopResult<bool> {
     match rpc
         .get_tx_op_return(id)
         .await
         .map(|(full, bytes)| (full, OutMetadata::decode(&bytes)))
     {
         Ok((full, Ok(info))) => match info {
             OutMetadata::Debit { wtid, url } => {
                 return sync_chain_debit(id, &full, &wtid, &url, db, confirmations, state).await;
             }
             OutMetadata::Bounce { bounced } => {
                 sync_chain_bounce(id, &Txid::from_byte_array(bounced), db, confirmations).await?
             }
         },
         Ok((_, Err(e))) => warn!(target: "worker", "send: decode-info {id} - {e}"),
         Err(e) => match e {
             GetOpReturnErr::MissingOpReturn => { /* Ignore */ }
             GetOpReturnErr::RPC(e) => return Err(e)?,
         },
     }
     Ok(false)
 }
 
 /// Send a debit transaction on the blockchain, return false if no more requested transactions are found
 async fn debit(db: &mut PgListener, rpc: &mut Rpc, 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)) =
         db::pending_debit(&mut *db, &state.currency).await?
     {
         let metadata = OutMetadata::Debit {
             wtid: wtid.clone(),
             url,
         };
 
         let txid = rpc
             .send(&addr, &amount, Some(&metadata.encode().unwrap()), false)
             .await?;
         fail_point("(injected) fail debit", 0.3)?;
         db::debit_sent(db, id, &txid).await?;
         let amount = btc_to_taler(&amount.to_signed().unwrap(), &state.currency);
         info!(target: "worker", ">> {amount} {wtid} in {txid} to {addr}");
         Ok(true)
     } else {
         Ok(false)
     }
 }
 
 /// Bounce a transaction on the blockchain, return false if no more requested transactions are found
 async fn bounce(db: &mut PgListener, rpc: &mut Rpc, 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 {
             bounced: *bounced.as_byte_array(),
         };
 
         match rpc
             .bounce(&bounced, fee, Some(&metadata.encode().unwrap()))
             .await
         {
             Ok(txid) => {
                 fail_point("(injected) fail bounce", 0.3)?;
                 db::bounce_set_status(db, id, Some(&txid), &BounceStatus::sent).await?;
                 if let Some(reason) = reason {
                     info!(target: "worker", "|| {bounced} in {txid}: {reason}");
                 } else {
                     info!(target: "worker", "|| {bounced} in {txid}");
                 }
             }
             Err(err) => match err {
                 rpc::Error::RPC {
                     code: ErrorCode::RpcWalletInsufficientFunds | ErrorCode::RpcWalletError,
                     msg: _,
                 } => {
                     warn!(target: "worker", "{err}");
                     return Ok(false);
                 }
                 e => Err(e)?,
             },
         }
         Ok(true)
     } else {
         Ok(false)
     }
 }