worker.rs (9949B)
1 /* 2 This file is part of TALER 3 Copyright (C) 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 17 use std::{sync::LazyLock, time::Duration}; 18 19 use futures_util::future::{join_all, try_join_all}; 20 use http_client::ApiErr; 21 use jiff::Timestamp; 22 use regex::Regex; 23 use sqlx::PgPool; 24 use taler_api::subject::{IncomingSubject, parse_incoming_unstructured}; 25 use taler_common::{ExpoBackoffDecorr, config::Config, types::payto::BankID}; 26 use tracing::{error, info, trace, warn}; 27 28 use crate::{ 29 config::{WiseBalance, WorkerCfg}, 30 db::{AddIncomingResult, TxIn, register_tx_in}, 31 payto::WiseAccount, 32 wise_api::{ 33 client::{Client, WiseErr}, 34 types::Direction, 35 }, 36 }; 37 38 #[derive(Debug, thiserror::Error)] 39 pub enum WorkerError { 40 #[error(transparent)] 41 Db(#[from] sqlx::Error), 42 #[error(transparent)] 43 Api(#[from] ApiErr<WiseErr>), 44 } 45 46 pub type WorkerResult = Result<(), WorkerError>; 47 48 fn parse_account(account_str: &str) -> Option<WiseAccount> { 49 static IBAN_BIC_PATTERN: LazyLock<Regex> = 50 LazyLock::new(|| Regex::new(r"^\(([A-Z0-9]{8,11})\) ([A-Z0-9]+)$").unwrap()); 51 52 if let Some(caps) = IBAN_BIC_PATTERN.captures(account_str) { 53 let bic = caps[1].parse().ok()?; 54 let iban = caps[2].parse().ok()?; 55 56 return Some(WiseAccount::IBAN(BankID { 57 iban, 58 bic: Some(bic), 59 })); 60 } 61 62 None 63 } 64 65 /// Each balance owns its retry schedule. A slow or failing balance must not 66 /// delay other balances, and success elsewhere must not reset its backoff. 67 async fn poll_balance( 68 balance_id: u32, 69 frequency: Duration, 70 transient: bool, 71 mut sync: impl AsyncFnMut() -> WorkerResult, 72 ) -> WorkerResult { 73 let mut jitter = ExpoBackoffDecorr::default(); 74 loop { 75 let result = sync().await; 76 if transient { 77 return result; 78 } 79 let delay = match result { 80 Ok(()) => { 81 jitter.reset(); 82 frequency 83 } 84 Err(err @ WorkerError::Db(sqlx::Error::PoolClosed)) => return Err(err), 85 Err(err) => { 86 let delay = jitter.backoff(); 87 error!(target: "worker", balance_id, ?delay, "balance synchronization failed: {err}"); 88 delay 89 } 90 }; 91 tokio::time::sleep(delay).await; 92 } 93 } 94 95 pub async fn run_worker( 96 cfg: &Config, 97 pool: &PgPool, 98 client: &http_client::Client, 99 transient: bool, 100 ) -> anyhow::Result<()> { 101 let cfg = WorkerCfg::parse(cfg)?; 102 let client = Client::new(client, &cfg.token); 103 let cfg = &cfg; 104 let client = &client; 105 let workers = cfg.balances.iter().map(|balance| { 106 poll_balance(balance.id, cfg.frequency, transient, async move || { 107 sync_balance(cfg, balance, pool, client).await 108 }) 109 }); 110 if transient { 111 // Attempt every balance once, even if another balance fails. 112 for result in join_all(workers).await { 113 result?; 114 } 115 } else { 116 // These are operation loops, not detached tasks: panics and terminal 117 // failures still reach the process, cancelling the remaining loops. 118 try_join_all(workers).await?; 119 } 120 Ok(()) 121 } 122 123 async fn sync_balance( 124 cfg: &WorkerCfg, 125 balance: &WiseBalance, 126 pool: &PgPool, 127 client: &Client<'_>, 128 ) -> WorkerResult { 129 let stmt = client 130 .balance_statement( 131 cfg.profile_id, 132 balance.id, 133 &balance.currency, 134 "2026-07-16T00:00:00.000Z".parse().unwrap(), 135 Timestamp::now(), 136 ) 137 .await?; 138 let now = Timestamp::now(); 139 for tx in stmt.transactions { 140 match tx.direction { 141 Direction::Debit => { 142 // TODO support outgoing transaction 143 } 144 Direction::Credit => { 145 let subject = parse_incoming_unstructured(&tx.details.payment_reference); 146 // Parse sender account 147 let payto = parse_account(&tx.details.sender_account); 148 // 149 let t = TxIn { 150 balance_id: balance.id, 151 wise_ref: Some(tx.reference_number), 152 amount: tx.amount.into(), 153 subject: tx.details.payment_reference, 154 name: tx.details.sender_name, 155 debtor: payto, 156 value_at: tx.date, 157 }; 158 let subject = &match subject { 159 Ok(IncomingSubject::Key(key)) => Some(key), 160 Ok(IncomingSubject::AdminBalanceAdjust) | Err(_) => None, 161 }; 162 let failure = match register_tx_in(pool, &t, subject, &now).await? { 163 AddIncomingResult::Success { new, .. } => { 164 if new { 165 info!(target: "worker", "in {t}"); 166 if t.debtor.is_none() { 167 warn!(target: "worker", "Couldn't parse creditor account from '{}'", tx.details.sender_account) 168 } 169 } else { 170 trace!(target: "worker", "in {t} already seen"); 171 } 172 continue; 173 } 174 AddIncomingResult::ReservePubReuse => "reserve pub reuse", 175 AddIncomingResult::UnknownMapping => "unknown mapping", 176 AddIncomingResult::MappingReuse => "mapping reuse", 177 }; 178 179 match register_tx_in(pool, &t, &None, &now).await? { 180 AddIncomingResult::Success { new, .. } => { 181 if new { 182 info!(target: "worker", "in {t}: {failure}"); 183 if t.debtor.is_none() { 184 warn!(target: "worker", "Couldn't parse creditor account from '{}'", tx.details.sender_account) 185 } 186 } else { 187 trace!(target: "worker", "in {t} already seen: {failure}"); 188 } 189 continue; 190 } 191 AddIncomingResult::ReservePubReuse 192 | AddIncomingResult::UnknownMapping 193 | AddIncomingResult::MappingReuse => unreachable!(), 194 }; 195 } 196 } 197 } 198 Ok(()) 199 } 200 201 #[cfg(test)] 202 mod tests { 203 use std::cell::Cell; 204 205 use futures_util::FutureExt; 206 use tokio::time::Instant; 207 208 use super::*; 209 210 fn unavailable() -> WorkerError { 211 sqlx::Error::Io(std::io::ErrorKind::ConnectionRefused.into()).into() 212 } 213 214 #[tokio::test(start_paused = true)] 215 async fn failed_balance_retries_without_stalling_other_balances() { 216 let attempts = Cell::new(0); 217 let healthy = Cell::new(0); 218 let start = Instant::now(); 219 let bad = poll_balance(1, Duration::from_secs(60), false, async || { 220 attempts.set(attempts.get() + 1); 221 if attempts.get() == 3 { 222 Err(sqlx::Error::PoolClosed.into()) 223 } else { 224 Err(unavailable()) 225 } 226 }); 227 let good = poll_balance(2, Duration::from_millis(100), false, async || { 228 healthy.set(healthy.get() + 1); 229 Ok(()) 230 }); 231 let result = try_join_all([bad.boxed_local(), good.boxed_local()]).await; 232 assert!(matches!( 233 result, 234 Err(WorkerError::Db(sqlx::Error::PoolClosed)) 235 )); 236 assert_eq!(attempts.get(), 3); 237 assert!(healthy.get() > attempts.get()); 238 assert!(start.elapsed() >= Duration::from_millis(800)); 239 } 240 241 #[tokio::test(start_paused = true)] 242 async fn success_resets_only_this_balances_backoff() { 243 let attempts = Cell::new(0); 244 let failed_at = Cell::new(Instant::now()); 245 let result = poll_balance(1, Duration::from_millis(10), false, async || { 246 attempts.set(attempts.get() + 1); 247 match attempts.get() { 248 11 => Ok(()), 249 12 => { 250 failed_at.set(Instant::now()); 251 Err(unavailable()) 252 } 253 13 => Err(sqlx::Error::PoolClosed.into()), 254 _ => Err(unavailable()), 255 } 256 }) 257 .await; 258 assert!(result.is_err()); 259 let delay = failed_at.get().elapsed(); 260 assert!(delay >= Duration::from_millis(400)); 261 assert!(delay < Duration::from_secs(1)); 262 } 263 264 #[tokio::test(start_paused = true)] 265 async fn transient_attempts_each_balance_once_and_reports_failure() { 266 let attempts = Cell::new(0); 267 let results = join_all([ 268 poll_balance( 269 1, 270 Duration::from_secs(60), 271 true, 272 async || Err(unavailable()), 273 ) 274 .boxed_local(), 275 poll_balance(2, Duration::from_secs(60), true, async || { 276 tokio::time::sleep(Duration::from_millis(100)).await; 277 attempts.set(attempts.get() + 1); 278 Ok(()) 279 }) 280 .boxed_local(), 281 ]) 282 .await; 283 assert!(results[0].is_err()); 284 assert!(results[1].is_ok()); 285 assert_eq!(attempts.get(), 1); 286 } 287 }