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