taler-rust

worker.rs (23549B)

---

 /*
   This file is part of TALER
   Copyright (C) 2025, 2026 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::time::Duration;
 
 use failure_injection::{InjectedErr, fail_point};
 use http_client::ApiErr;
 use jiff::Timestamp;
 use sqlx::{Acquire as _, PgConnection, PgPool, postgres::PgListener};
 use taler_api::subject::{self, IncomingSubject, parse_incoming_unstructured};
 use taler_common::{
     ExpoBackoffDecorr,
     config::Config,
     types::amount::{self, Currency},
 };
 use tokio::sync::Notify;
 use tracing::{debug, error, info, trace, warn};
 
 use crate::{
     config::{AccountType, WorkerCfg},
     constants::SYNC_CURSOR_KEY,
     cyclos_api::{
         api::{CyclosAuth, CyclosErr},
         client::Client,
         types::{AccountKind, HistoryItem, InputError, NotFoundError, OrderBy},
     },
     db::{
         self, AddIncomingResult, ChargebackFailureResult, RegisterResult, TxIn, TxOut, TxOutKind,
         kv_get, kv_set,
     },
     notification::watch_notification,
 };
 
 #[derive(Debug, thiserror::Error)]
 pub enum WorkerError {
     #[error(transparent)]
     Db(#[from] sqlx::Error),
     #[error(transparent)]
     Api(#[from] ApiErr<CyclosErr>),
     #[error("Another worker is running concurrently")]
     Concurrency,
     #[error(transparent)]
     Injected(#[from] InjectedErr),
 }
 
 pub type WorkerResult = Result<(), WorkerError>;
 
 /// Retry an operation, not a terminated task. A closed pool cannot recover,
 /// and injected interruptions must remain visible to the caller/supervisor.
 fn retry_delay(
     err: &WorkerError,
     jitter: &mut ExpoBackoffDecorr,
     frequency: Duration,
 ) -> Option<Duration> {
     match err {
         WorkerError::Db(sqlx::Error::PoolClosed) | WorkerError::Injected(_) => None,
         WorkerError::Concurrency => Some(jitter.backoff().max(Duration::from_secs(15))),
         WorkerError::Api(e)
             if matches!(*e.err, CyclosErr::Input(InputError::Validation { .. })) =>
         {
             // Invalid input must not be resubmitted at notification speed.
             Some(jitter.backoff().max(frequency))
         }
         _ => Some(jitter.backoff()),
     }
 }
 
 pub async fn run_worker(
     cfg: &Config,
     pool: &PgPool,
     client: &http_client::Client,
     transient: bool,
 ) -> anyhow::Result<()> {
     let cfg = WorkerCfg::parse(cfg)?;
     let client = Client {
         client,
         api_url: &cfg.host.api_url,
         auth: &CyclosAuth::Basic {
             username: cfg.host.username,
             password: cfg.host.password,
         },
     };
     if transient {
         let mut conn = pool.acquire().await?;
         Worker {
             client: &client,
             db: &mut conn,
             account_type_id: *cfg.account_type_id,
             payment_type_id: *cfg.payment_type_id,
             account_type: cfg.account_type,
             currency: cfg.currency,
         }
         .run()
         .await?;
         return Ok(());
     }
 
     let notification = Notify::new();
 
     let watcher = async {
         watch_notification(&client, &notification).await;
     };
     let worker = async {
         let mut jitter = ExpoBackoffDecorr::default();
         loop {
             let res: WorkerResult = async {
                 let db = &mut PgListener::connect_with(pool).await?;
                 db.listen_all(["transfer"]).await?;
                 info!(target: "worker", "database listener connected");
                 loop {
                     let result = Worker {
                         client: &client,
                         db: db.acquire().await?,
                         account_type_id: *cfg.account_type_id,
                         payment_type_id: *cfg.payment_type_id,
                         account_type: cfg.account_type,
                         currency: cfg.currency,
                     }
                     .run()
                     .await;
                     match result {
                         Ok(()) => jitter.reset(),
                         Err(err @ WorkerError::Db(_)) => return Err(err),
                         Err(err) => {
                             let Some(delay) = retry_delay(&err, &mut jitter, cfg.frequency) else {
                                 return Err(err);
                             };
                             error!(target: "worker", ?delay, "synchronization pass failed: {err}");
                             tokio::time::sleep(delay).await;
                             continue;
                         }
                     }
                     tokio::select! {
                         _ = tokio::time::sleep(cfg.frequency) => {
                             info!(target: "worker", "running at frequency");
                         }
                         res = db.try_recv() => {
                             let mut ntf = res?;
                             while let Some(n) = ntf {
                                 debug!(target: "worker", "notification from {}", n.channel());
                                 ntf = db.next_buffered();
                             }
                         }
                         _ = notification.notified() => {
                             info!(target: "worker", "running at notification trigger");
                         }
                     }
                 }
             }
             .await;
             let err = res.unwrap_err();
             let Some(delay) = retry_delay(&err, &mut jitter, cfg.frequency) else {
                 break Err(err);
             };
             error!(target: "worker", ?delay, "database session failed: {err}");
             tokio::time::sleep(delay).await;
         }
     };
     // The optional notification listener reconnects independently. Polling
     // continues during its outages; a terminal worker result is not swallowed.
     tokio::select! {
         result = worker => result?,
         _ = watcher => unreachable!("notification listener does not return"),
     }
 }
 
 pub struct Worker<'a> {
     pub client: &'a Client<'a>,
     pub db: &'a mut PgConnection,
     pub currency: Currency,
     pub account_type_id: i64,
     pub payment_type_id: i64,
     pub account_type: AccountType,
 }
 
 impl Worker<'_> {
     /// Run a single worker pass
     pub async fn run(&mut self) -> WorkerResult {
         // Some worker operations are not idempotent, therefore it's not safe to have multiple worker
         // running concurrently. We use a global Postgres advisory lock to prevent it.
         if !db::worker_lock(self.db).await? {
             return Err(WorkerError::Concurrency);
         }
 
         db::task_start(&mut *self.db, "sync").await?;
 
         // Sync transactions
         let mut cursor: Timestamp = kv_get(&mut *self.db, SYNC_CURSOR_KEY)
             .await?
             .unwrap_or_default();
 
         loop {
             let page = self
                 .client
                 .history(self.account_type_id, OrderBy::DateAsc, 0, Some(cursor))
                 .await?;
             for transfer in page.page {
                 if transfer.date > cursor {
                     cursor = transfer.date;
                 }
                 let tx = extract_tx_info(transfer);
                 match tx {
                     Tx::In(tx_in) => self.ingest_in(tx_in).await?,
                     Tx::Out(tx_out) => self.ingest_out(tx_out).await?,
                 }
             }
 
             kv_set(&mut *self.db, SYNC_CURSOR_KEY, &cursor).await?;
 
             if !page.has_next_page {
                 break;
             }
         }
 
         db::task_success(&mut *self.db, "sync").await?;
         db::task_start(&mut *self.db, "submit").await?;
 
         // Send transactions
         let start = Timestamp::now();
         loop {
             let batch = db::pending_batch(&mut *self.db, &start).await?;
             if batch.is_empty() {
                 break;
             }
             for initiated in batch {
                 debug!(target: "worker", "send tx {initiated}");
                 let res = self
                     .client
                     .direct_payment(
                         initiated.creditor_id,
                         self.payment_type_id,
                         initiated.amount,
                         &initiated.subject,
                     )
                     .await;
                 fail_point("direct-payment")?;
                 match res {
                     Ok(tx) => {
                         // Update transaction status, on failure the initiated transaction will be orphan
                         db::initiated_submit_success(
                             &mut *self.db,
                             initiated.id,
                             &tx.date,
                             tx.id.0,
                         )
                         .await?;
                         trace!(target: "worker", "init tx {}", tx.id);
                     }
                     Err(e) => {
                         let msg = match &*e.err {
                             CyclosErr::Unknown(NotFoundError { entity_type, key }) => {
                                 format!("unknown {entity_type} {key}")
                             }
                             CyclosErr::Forbidden(err) => err.to_string(),
                             _ => return Err(e.into()),
                         };
                         // TODO is permission should be considered are hard or soft failure ?
                         db::initiated_submit_permanent_failure(&mut *self.db, initiated.id, &msg)
                             .await?;
                         error!(target: "worker", "initiated failure {initiated}: {msg}");
                     }
                 }
             }
         }
         db::task_success(&mut *self.db, "submit").await?;
         Ok(())
     }
 
     /// Ingest an incoming transaction
     async fn ingest_in(&mut self, tx: TxIn) -> WorkerResult {
         match self.account_type {
             AccountType::Exchange => {
                 let transfer = self.client.transfer(tx.transfer_id).await?;
                 let bounce = async |db: &mut PgConnection,
                                     reason: &str|
                        -> Result<(), WorkerError> {
                     // Fetch existing transaction
                     if let Some(chargeback) = transfer.charged_back_by {
                         let res = db::register_bounced_tx_in(
                             db,
                             &tx,
                             *chargeback.id,
                             reason,
                             &Timestamp::now(),
                         )
                         .await?;
                         if res.tx_new {
                             info!(target: "worker",
                                 "in {tx} bounced (recovered) in {}: {reason}", chargeback.id
                             );
                         } else {
                             trace!(target: "worker",
                                 "in {tx} already seen and bounced in {}: {reason}",chargeback.id
                             );
                         }
                     } else if !transfer.can_chargeback {
                         match db::register_tx_in(db, &tx, &None, &Timestamp::now()).await? {
                             AddIncomingResult::Success { new, .. } => {
                                 if new {
                                     warn!(target: "worker", "in {tx} cannot bounce: {reason}");
                                 } else {
                                     trace!(target: "worker", "in {tx} already seen and cannot bounce ");
                                 }
                             }
                             AddIncomingResult::ReservePubReuse
                             | AddIncomingResult::UnknownMapping
                             | AddIncomingResult::MappingReuse => unreachable!(),
                         }
                     } else {
                         let chargeback_id = self.client.chargeback(*transfer.id).await?;
                         fail_point("chargeback")?;
                         let res = db::register_bounced_tx_in(
                             db,
                             &tx,
                             chargeback_id,
                             reason,
                             &Timestamp::now(),
                         )
                         .await?;
                         if res.tx_new {
                             info!(target: "worker", "in {tx} bounced in {chargeback_id}: {reason}");
                         } else {
                             trace!(target: "worker", "in {tx} already seen and bounced in {chargeback_id}: {reason}");
                         }
                     }
                     Ok(())
                 };
                 if let Some(chargeback) = transfer.chargeback_of {
                     // This a chargeback of one of our transaction, if we bounce we might enter a loop
                     match db::initiated_chargeback_failure(&mut *self.db, *chargeback.id).await? {
                         ChargebackFailureResult::Unknown => {
                             trace!(target: "worker", "initiated failure unknown: charged back")
                         }
                         ChargebackFailureResult::Known(initiated) => {
                             error!(target: "worker", "initiated failure {initiated}: charged back")
                         }
                         ChargebackFailureResult::Idempotent(initiated) => {
                             trace!(target: "worker", "initiated failure {initiated} already seen: charged back")
                         }
                     }
                     // Sill register the incoming transaction as an incoming one
                     match db::register_tx_in(self.db, &tx, &None, &Timestamp::now()).await? {
                         AddIncomingResult::Success { new, .. } => {
                             if new {
                                 info!(target: "worker", "in {tx} chargeback");
                             } else {
                                 trace!(target: "worker", "in {tx} chargeback already seen");
                             }
                         }
                         AddIncomingResult::ReservePubReuse
                         | AddIncomingResult::UnknownMapping
                         | AddIncomingResult::MappingReuse => unreachable!(),
                     }
 
                     return Ok(());
                 }
                 match parse_incoming_unstructured(&tx.subject) {
                     Ok(subject) => {
                         match subject {
                             IncomingSubject::Key(subject) => {
                                 match db::register_tx_in(
                                     self.db,
                                     &tx,
                                     &Some(subject),
                                     &Timestamp::now(),
                                 )
                                 .await?
                                 {
                                     AddIncomingResult::Success { new, .. } => {
                                         if new {
                                             info!(target: "worker", "in {tx}");
                                         } else {
                                             trace!(target: "worker", "in {tx} already seen");
                                         }
                                     }
                                     AddIncomingResult::ReservePubReuse => {
                                         bounce(self.db, "reserve pub reuse").await?
                                     }
                                     AddIncomingResult::UnknownMapping => {
                                         bounce(self.db, "unknown mapping").await?
                                     }
                                     AddIncomingResult::MappingReuse => {
                                         bounce(self.db, "mapping reuse").await?
                                     }
                                 }
                             }
                             IncomingSubject::AdminBalanceAdjust => {
                                 // TODO bounce or skip ?
                             }
                         }
                     }
                     Err(e) => bounce(self.db, &e.to_string()).await?,
                 }
             }
             AccountType::Normal => {
                 match db::register_tx_in(self.db, &tx, &None, &Timestamp::now()).await? {
                     AddIncomingResult::Success { new, .. } => {
                         if new {
                             info!(target: "worker", "in {tx}");
                         } else {
                             trace!(target: "worker", "in {tx} already seen");
                         }
                     }
                     AddIncomingResult::ReservePubReuse
                     | AddIncomingResult::UnknownMapping
                     | AddIncomingResult::MappingReuse => unreachable!(),
                 }
             }
         }
         Ok(())
     }
 
     async fn ingest_out(&mut self, tx: TxOut) -> WorkerResult {
         match self.account_type {
             AccountType::Exchange => {
                 let transfer = self.client.transfer(tx.transfer_id).await?;
 
                 if transfer.charged_back_by.is_some() {
                     match db::initiated_chargeback_failure(&mut *self.db, *transfer.id).await? {
                         ChargebackFailureResult::Unknown => {
                             trace!(target: "worker", "initiated failure unknown: charged back")
                         }
                         ChargebackFailureResult::Known(initiated) => {
                             error!(target: "worker", "initiated failure {initiated}: charged back")
                         }
                         ChargebackFailureResult::Idempotent(initiated) => {
                             trace!(target: "worker", "initiated failure {initiated} already seen: charged back")
                         }
                     }
                 }
 
                 let kind = if let Ok(subject) = subject::parse_outgoing(&tx.subject) {
                     TxOutKind::Talerable(subject)
                 } else if let Some(chargeback) = &transfer.chargeback_of {
                     TxOutKind::Bounce(*chargeback.id)
                 } else {
                     TxOutKind::Simple
                 };
 
                 let res = db::register_tx_out(self.db, &tx, &kind, &Timestamp::now()).await?;
                 match res.result {
                     RegisterResult::idempotent => match kind {
                         TxOutKind::Simple => {
                             trace!(target: "worker", "out malformed {tx} already seen")
                         }
                         TxOutKind::Bounce(_) => {
                             trace!(target: "worker", "out bounce {tx} already seen")
                         }
                         TxOutKind::Talerable(_) => {
                             trace!(target: "worker", "out {tx} already seen")
                         }
                     },
                     RegisterResult::known => match kind {
                         TxOutKind::Simple => {
                             warn!(target: "worker", "out malformed {tx}")
                         }
                         TxOutKind::Bounce(_) => {
                             info!(target: "worker", "out bounce {tx}")
                         }
                         TxOutKind::Talerable(_) => {
                             info!(target: "worker", "out {tx}")
                         }
                     },
                     RegisterResult::recovered => match kind {
                         TxOutKind::Simple => {
                             warn!(target: "worker", "out malformed (recovered) {tx}")
                         }
                         TxOutKind::Bounce(_) => {
                             warn!(target: "worker", "out bounce (recovered) {tx}")
                         }
                         TxOutKind::Talerable(_) => {
                             warn!(target: "worker", "out (recovered) {tx}")
                         }
                     },
                 }
             }
             AccountType::Normal => {
                 let res = db::register_tx_out(self.db, &tx, &TxOutKind::Simple, &Timestamp::now())
                     .await?;
                 match res.result {
                     RegisterResult::idempotent => {
                         trace!(target: "worker", "out {tx} already seen");
                     }
                     RegisterResult::known => {
                         info!(target: "worker", "out {tx}");
                     }
                     RegisterResult::recovered => {
                         warn!(target: "worker", "out (recovered) {tx}");
                     }
                 }
             }
         }
         Ok(())
     }
 }
 
 pub enum Tx {
     In(TxIn),
     Out(TxOut),
 }
 
 pub fn extract_tx_info(tx: HistoryItem) -> Tx {
     let amount = amount::decimal(tx.amount.trim_start_matches('-'));
     let (id, name) = match tx.related_account.kind {
         AccountKind::System => (tx.related_account.ty.id, tx.related_account.ty.name),
         AccountKind::User { user } => (user.id, user.display),
     };
     if tx.amount.starts_with('-') {
         Tx::Out(TxOut {
             transfer_id: *tx.id,
             tx_id: tx.transaction.map(|it| *it.id),
             amount,
             subject: tx.description.unwrap_or_default(),
             creditor_id: *id,
             creditor_name: name,
             valued_at: tx.date,
         })
     } else {
         Tx::In(TxIn {
             transfer_id: *tx.id,
             tx_id: tx.transaction.map(|it| *it.id),
             amount,
             subject: tx.description.unwrap_or_default(),
             debtor_id: *id,
             debtor_name: name,
             valued_at: tx.date,
         })
     }
 }
 
 #[cfg(test)]
 mod retry_tests {
     use super::*;
 
     #[test]
     fn database_outages_retry_but_terminal_interruptions_propagate() {
         let mut jitter = ExpoBackoffDecorr::default();
         let err = WorkerError::Db(sqlx::Error::Io(
             std::io::ErrorKind::ConnectionRefused.into(),
         ));
         let delay = retry_delay(&err, &mut jitter, Duration::from_secs(60)).unwrap();
         assert!(delay >= Duration::from_millis(400));
         assert!(delay <= Duration::from_secs(30));
         assert!(
             retry_delay(
                 &WorkerError::Db(sqlx::Error::PoolClosed),
                 &mut jitter,
                 Duration::ZERO
             )
             .is_none()
         );
         assert!(
             retry_delay(
                 &WorkerError::Injected(InjectedErr("worker interrupted")),
                 &mut jitter,
                 Duration::ZERO
             )
             .is_none()
         );
         assert!(
             retry_delay(&WorkerError::Concurrency, &mut jitter, Duration::ZERO).unwrap()
                 >= Duration::from_secs(15)
         );
     }
 }