worker.rs (30048B)
1 /* 2 This file is part of TALER 3 Copyright (C) 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 17 use std::{num::ParseIntError, time::Duration}; 18 19 use aws_lc_rs::signature::EcdsaKeyPair; 20 use failure_injection::{InjectedErr, fail_point}; 21 use http_client::ApiErr; 22 use jiff::{Timestamp, Zoned, civil::Date}; 23 use sqlx::{Acquire as _, PgConnection, PgPool, postgres::PgListener}; 24 use taler_api::subject::{self, IncomingSubject, parse_incoming_unstructured}; 25 use taler_common::{ 26 ExpoBackoffDecorr, 27 config::Config, 28 types::{ 29 amount::{self}, 30 iban::IBAN, 31 }, 32 }; 33 use tracing::{debug, error, info, trace, warn}; 34 35 use crate::{ 36 FullHuPayto, HuIban, 37 config::{AccountType, WorkerCfg}, 38 db::{self, AddIncomingResult, Initiated, RegisterResult, TxIn, TxOut, TxOutKind}, 39 magnet_api::{ 40 api::MagnetErr, 41 client::{ApiClient, AuthClient}, 42 types::{Direction, Next, Order, TxDto, TxStatus}, 43 }, 44 setup, 45 }; 46 47 // const TXS_CURSOR_KEY: &str = "txs_cursor"; TODO cursor is broken 48 49 #[derive(Debug, thiserror::Error)] 50 pub enum WorkerError { 51 #[error(transparent)] 52 Db(#[from] sqlx::Error), 53 #[error(transparent)] 54 Api(#[from] ApiErr<MagnetErr>), 55 #[error("Another worker is running concurrently")] 56 Concurrency, 57 #[error(transparent)] 58 Injected(#[from] InjectedErr), 59 } 60 61 pub type WorkerResult = Result<(), WorkerError>; 62 63 /// Retry an operation, not a terminated task. A closed pool cannot recover, 64 /// and injected interruptions must remain visible to the caller/supervisor. 65 fn retry_delay(err: &WorkerError, jitter: &mut ExpoBackoffDecorr) -> Option<Duration> { 66 match err { 67 WorkerError::Db(sqlx::Error::PoolClosed) | WorkerError::Injected(_) => None, 68 WorkerError::Concurrency => Some(jitter.backoff().max(Duration::from_secs(15))), 69 _ => Some(jitter.backoff()), 70 } 71 } 72 73 pub async fn run_worker( 74 cfg: &Config, 75 pool: &PgPool, 76 client: &http_client::Client, 77 transient: bool, 78 ) -> anyhow::Result<()> { 79 let cfg = WorkerCfg::parse(cfg)?; 80 let keys = setup::load(&cfg).map_err(|err| taler_common::config::ValueErr::Invalid { 81 ty: "keys file".into(), 82 section: "magnet-bank-worker".into(), 83 option: "KEYS_FILE".into(), 84 err: err.to_string(), 85 })?; 86 let client = AuthClient::new(client, &cfg.api_url, &cfg.consumer).upgrade(&keys.access_token); 87 88 if transient { 89 let mut conn = pool.acquire().await?; 90 let account = client.account(cfg.payto.bban()).await?; 91 Worker { 92 client: &client, 93 db: &mut conn, 94 account_number: &account.number, 95 account_code: account.code, 96 key: &keys.signing_key, 97 account_type: cfg.account_type, 98 ignore_tx_before: cfg.ignore_tx_before, 99 ignore_bounces_before: cfg.ignore_bounces_before, 100 } 101 .run() 102 .await?; 103 return Ok(()); 104 } 105 106 let mut jitter = ExpoBackoffDecorr::default(); 107 // Account discovery is an API operation; retry it without reloading keys 108 // or rebuilding the HTTP client. 109 let account = loop { 110 match client.account(cfg.payto.bban()).await { 111 Ok(account) => break account, 112 Err(err) => { 113 error!(target: "worker", "account lookup failed: {err}"); 114 tokio::time::sleep(jitter.backoff()).await; 115 } 116 } 117 }; 118 jitter.reset(); 119 120 loop { 121 // Only database/session failures reconstruct the listener. API failures 122 // retry a pass with the existing client and database session. 123 let res: WorkerResult = async { 124 let db = &mut PgListener::connect_with(pool).await?; 125 db.listen_all(["transfer"]).await?; 126 info!(target: "worker", "database listener connected"); 127 loop { 128 let result = Worker { 129 client: &client, 130 db: db.acquire().await?, 131 account_number: &account.number, 132 account_code: account.code, 133 key: &keys.signing_key, 134 account_type: cfg.account_type, 135 ignore_tx_before: cfg.ignore_tx_before, 136 ignore_bounces_before: cfg.ignore_bounces_before, 137 } 138 .run() 139 .await; 140 match result { 141 Ok(()) => jitter.reset(), 142 Err(err @ WorkerError::Db(_)) => return Err(err), 143 Err(err) => { 144 let Some(delay) = retry_delay(&err, &mut jitter) else { 145 return Err(err); 146 }; 147 error!(target: "worker", ?delay, "synchronization pass failed: {err}"); 148 tokio::time::sleep(delay).await; 149 continue; 150 } 151 } 152 // Notifications may accelerate successful polling, but never 153 // bypass the backoff after a failed operation. 154 match tokio::time::timeout(cfg.frequency, db.try_recv()).await { 155 Ok(res) => { 156 let mut ntf = res?; 157 while let Some(n) = ntf { 158 debug!(target: "worker", "notification from {}", n.channel()); 159 ntf = db.next_buffered(); 160 } 161 } 162 Err(_) => info!(target: "worker", "running at frequency"), 163 } 164 } 165 } 166 .await; 167 let err = res.unwrap_err(); 168 let Some(delay) = retry_delay(&err, &mut jitter) else { 169 return Err(err.into()); 170 }; 171 error!(target: "worker", ?delay, "database session failed: {err}"); 172 tokio::time::sleep(delay).await; 173 } 174 } 175 176 pub struct Worker<'a> { 177 pub client: &'a ApiClient<'a>, 178 pub db: &'a mut PgConnection, 179 pub account_number: &'a str, 180 pub account_code: u64, 181 pub key: &'a EcdsaKeyPair, 182 pub account_type: AccountType, 183 pub ignore_tx_before: Option<Date>, 184 pub ignore_bounces_before: Option<Date>, 185 } 186 187 impl Worker<'_> { 188 /// Run a single worker pass 189 pub async fn run(&mut self) -> WorkerResult { 190 // Some worker operations are not idempotent, therefore it's not safe to have multiple worker 191 // running concurrently. We use a global Postgres advisory lock to prevent it. 192 if !db::worker_lock(self.db).await? { 193 return Err(WorkerError::Concurrency); 194 }; 195 196 // Sync transactions 197 let mut next: Option<Next> = None; //kv_get(&mut *self.db, TXS_CURSOR_KEY).await?; TODO cursor logic is broken and cannot be stored & reused 198 let mut all_final = true; 199 let mut first = true; 200 loop { 201 let page = self 202 .client 203 .page_tx( 204 Direction::Both, 205 Order::Ascending, 206 100, 207 self.account_number, 208 &next, 209 first, 210 ) 211 .await?; 212 first = false; 213 next = page.next; 214 for item in page.list { 215 all_final &= item.tx.status.is_final(); 216 let tx = extract_tx_info(item.tx); 217 match tx { 218 Tx::In(tx_in) => { 219 // We only register final successful incoming transactions 220 if tx_in.status != TxStatus::Completed { 221 debug!(target: "worker", "pending or failed in {tx_in}"); 222 continue; 223 } 224 225 if let Some(before) = self.ignore_tx_before 226 && tx_in.value_date < before 227 { 228 debug!(target: "worker", "ignore in {tx_in}"); 229 continue; 230 } 231 let bounce = async |db: &mut PgConnection, 232 reason: &str| 233 -> Result<(), WorkerError> { 234 if let Some(before) = self.ignore_bounces_before 235 && tx_in.value_date < before 236 { 237 match db::register_tx_in(db, &tx_in, &None, &Timestamp::now()) 238 .await? 239 { 240 AddIncomingResult::Success { new, .. } => { 241 if new { 242 info!(target: "worker", "in {tx_in} skip bounce: {reason}"); 243 } else { 244 trace!(target: "worker", "in {tx_in} already skip bounce "); 245 } 246 } 247 AddIncomingResult::ReservePubReuse 248 | AddIncomingResult::UnknownMapping 249 | AddIncomingResult::MappingReuse => unreachable!(), 250 } 251 } else { 252 let res = db::register_bounce_tx_in( 253 db, 254 &tx_in, 255 reason, 256 &Timestamp::now(), 257 ) 258 .await?; 259 260 if res.tx_new { 261 info!(target: "worker", 262 "in {tx_in} bounced in {}: {reason}", 263 res.bounce_id 264 ); 265 } else { 266 trace!(target: "worker", 267 "in {tx_in} already seen and bounced in {}: {reason}", 268 res.bounce_id 269 ); 270 } 271 } 272 Ok(()) 273 }; 274 match self.account_type { 275 AccountType::Exchange => { 276 match parse_incoming_unstructured(&tx_in.subject) { 277 Ok(subject) => match subject { 278 IncomingSubject::Key(subject) => { 279 match db::register_tx_in( 280 self.db, 281 &tx_in, 282 &Some(subject), 283 &Timestamp::now(), 284 ) 285 .await? 286 { 287 AddIncomingResult::Success { new, .. } => { 288 if new { 289 info!(target: "worker", "in {tx_in}"); 290 } else { 291 trace!(target: "worker", "in {tx_in} already seen"); 292 } 293 } 294 AddIncomingResult::ReservePubReuse => { 295 bounce(self.db, "reserve pub reuse").await? 296 } 297 AddIncomingResult::UnknownMapping => { 298 bounce(self.db, "unknown mapping").await? 299 } 300 AddIncomingResult::MappingReuse => { 301 bounce(self.db, "mapping reuse").await? 302 } 303 } 304 } 305 IncomingSubject::AdminBalanceAdjust => { 306 // TODO bounce or skip ? 307 } 308 }, 309 Err(e) => bounce(self.db, &e.to_string()).await?, 310 } 311 } 312 AccountType::Normal => { 313 match db::register_tx_in(self.db, &tx_in, &None, &Timestamp::now()) 314 .await? 315 { 316 AddIncomingResult::Success { new, .. } => { 317 if new { 318 info!(target: "worker", "in {tx_in}"); 319 } else { 320 trace!(target: "worker", "in {tx_in} already seen"); 321 } 322 } 323 AddIncomingResult::ReservePubReuse 324 | AddIncomingResult::UnknownMapping 325 | AddIncomingResult::MappingReuse => unreachable!(), 326 } 327 } 328 } 329 } 330 Tx::Out(tx_out) => { 331 match tx_out.status { 332 TxStatus::ToBeRecorded => { 333 self.recover_tx(&tx_out).await?; 334 continue; 335 } 336 TxStatus::PendingFirstSignature 337 | TxStatus::PendingSecondSignature 338 | TxStatus::PendingProcessing 339 | TxStatus::Verified 340 | TxStatus::PartiallyCompleted 341 | TxStatus::UnderReview => { 342 // Still pending 343 debug!(target: "worker", "pending out {tx_out}"); 344 continue; 345 } 346 TxStatus::Rejected | TxStatus::Canceled | TxStatus::Completed => {} 347 } 348 match self.account_type { 349 AccountType::Exchange => { 350 let kind = if let Ok(subject) = 351 subject::parse_outgoing(&tx_out.subject) 352 { 353 TxOutKind::Talerable(subject) 354 } else if let Ok(bounced) = parse_bounce_outgoing(&tx_out.subject) { 355 TxOutKind::Bounce(bounced) 356 } else { 357 TxOutKind::Simple 358 }; 359 if tx_out.status == TxStatus::Completed { 360 let res = db::register_tx_out( 361 self.db, 362 &tx_out, 363 &kind, 364 &Timestamp::now(), 365 ) 366 .await?; 367 match res.result { 368 RegisterResult::idempotent => match kind { 369 TxOutKind::Simple => { 370 trace!(target: "worker", "out malformed {tx_out} already seen") 371 } 372 TxOutKind::Bounce(_) => { 373 trace!(target: "worker", "out bounce {tx_out} already seen") 374 } 375 TxOutKind::Talerable(_) => { 376 trace!(target: "worker", "out {tx_out} already seen") 377 } 378 }, 379 RegisterResult::known => match kind { 380 TxOutKind::Simple => { 381 warn!(target: "worker", "out malformed {tx_out}") 382 } 383 TxOutKind::Bounce(_) => { 384 info!(target: "worker", "out bounce {tx_out}") 385 } 386 TxOutKind::Talerable(_) => { 387 info!(target: "worker", "out {tx_out}") 388 } 389 }, 390 RegisterResult::recovered => match kind { 391 TxOutKind::Simple => { 392 warn!(target: "worker", "out malformed (recovered) {tx_out}") 393 } 394 TxOutKind::Bounce(_) => { 395 warn!(target: "worker", "out bounce (recovered) {tx_out}") 396 } 397 TxOutKind::Talerable(_) => { 398 warn!(target: "worker", "out (recovered) {tx_out}") 399 } 400 }, 401 } 402 } else { 403 let bounced = match kind { 404 TxOutKind::Simple => None, 405 TxOutKind::Bounce(bounced) => Some(bounced), 406 TxOutKind::Talerable(_) => None, 407 }; 408 let res = db::register_tx_out_failure( 409 self.db, 410 tx_out.code, 411 bounced, 412 &Timestamp::now(), 413 ) 414 .await?; 415 if let Some(id) = res.initiated_id { 416 if res.new { 417 error!(target: "worker", "out failure {id} {tx_out}"); 418 } else { 419 trace!(target: "worker", "out failure {id} {tx_out} already seen"); 420 } 421 } 422 } 423 } 424 AccountType::Normal => { 425 if tx_out.status == TxStatus::Completed { 426 let res = db::register_tx_out( 427 self.db, 428 &tx_out, 429 &TxOutKind::Simple, 430 &Timestamp::now(), 431 ) 432 .await?; 433 match res.result { 434 RegisterResult::idempotent => { 435 trace!(target: "worker", "out {tx_out} already seen"); 436 } 437 RegisterResult::known => { 438 info!(target: "worker", "out {tx_out}"); 439 } 440 RegisterResult::recovered => { 441 warn!(target: "worker", "out (recovered) {tx_out}"); 442 } 443 } 444 } else { 445 let res = db::register_tx_out_failure( 446 self.db, 447 tx_out.code, 448 None, 449 &Timestamp::now(), 450 ) 451 .await?; 452 if let Some(id) = res.initiated_id { 453 if res.new { 454 error!(target: "worker", "out failure {id} {tx_out}"); 455 } else { 456 trace!(target: "worker", "out failure {id} {tx_out} already seen"); 457 } 458 } 459 } 460 } 461 } 462 } 463 } 464 } 465 466 if let Some(_next) = &next { 467 // Update in db cursor only if all previous transactions where final 468 if all_final { 469 // debug!(target: "worker", "advance cursor {next:?}"); 470 // kv_set(&mut *self.db, TXS_CURSOR_KEY, &next).await?; TODO cursor is broken 471 } 472 } else { 473 break; 474 } 475 } 476 477 // Send transactions 478 let start = Timestamp::now(); 479 let now = Zoned::now(); 480 loop { 481 let batch = db::pending_batch(&mut *self.db, &start).await?; 482 if batch.is_empty() { 483 break; 484 } 485 for tx in batch { 486 debug!(target: "worker", "send tx {tx}"); 487 self.init_tx(&tx, &now).await?; 488 } 489 } 490 Ok(()) 491 } 492 493 /// Try to sign an unsigned initiated transaction 494 pub async fn recover_tx(&mut self, tx: &TxOut) -> WorkerResult { 495 if db::initiated_exists_for_code(&mut *self.db, tx.code) 496 .await? 497 .is_some() 498 { 499 // Known initiated we submit it 500 assert_eq!(tx.amount.frac, 0); 501 self.submit_tx( 502 tx.code, 503 -(tx.amount.val as f64), 504 &tx.value_date, 505 tx.creditor.bban(), 506 ) 507 .await?; 508 } else { 509 // The transaction is unknown (we failed after creating it and before storing it in the db) 510 // we delete it 511 self.client.delete_tx(tx.code).await?; 512 debug!(target: "worker", "out {}: delete uncompleted orphan", tx.code); 513 } 514 515 Ok(()) 516 } 517 518 /// Create and sign a forint transfer 519 pub async fn init_tx(&mut self, tx: &Initiated, now: &Zoned) -> WorkerResult { 520 trace!(target: "worker", "init tx {tx}"); 521 assert_eq!(tx.amount.frac, 0); 522 let date = now.date(); 523 // Initialize the new transaction, on failure an orphan initiated transaction can be created 524 let res = self 525 .client 526 .init_tx( 527 self.account_code, 528 tx.amount.val as f64, 529 &tx.subject, 530 &date, 531 &tx.creditor.name, 532 tx.creditor.bban(), 533 ) 534 .await; 535 fail_point("init-tx")?; 536 let info = match res { 537 // Check if succeeded 538 Ok(info) => { 539 // Update transaction status, on failure the initiated transaction will be orphan 540 db::initiated_submit_success(&mut *self.db, tx.id, &Timestamp::now(), info.code) 541 .await?; 542 info 543 } 544 Err(e) => { 545 if let MagnetErr::Magnet(e) = &*e.err { 546 // Check if error is permanent 547 if matches!( 548 (e.error_code, e.short_message.as_str()), 549 (404, "BSZLA_NEM_TALALHATO") // Unknown account 550 | (409, "FORRAS_SZAMLA_ESZAMLA_EGYEZIK") // Same account 551 ) { 552 db::initiated_submit_permanent_failure( 553 &mut *self.db, 554 tx.id, 555 &Timestamp::now(), 556 &e.to_string(), 557 ) 558 .await?; 559 error!(target: "worker", "initiated failure {tx}: {e}"); 560 return WorkerResult::Ok(()); 561 } 562 } 563 return Err(e.into()); 564 } 565 }; 566 trace!(target: "worker", "init tx {}", info.code); 567 568 // Sign transaction 569 self.submit_tx(info.code, info.amount, &date, tx.creditor.bban()) 570 .await?; 571 Ok(()) 572 } 573 574 /** Submit an initiated forint transfer */ 575 pub async fn submit_tx( 576 &mut self, 577 tx_code: u64, 578 amount: f64, 579 date: &Date, 580 creditor: &str, 581 ) -> WorkerResult { 582 debug!(target: "worker", "submit tx {tx_code}"); 583 fail_point("submit-tx")?; 584 // Submit an initiated transaction, on failure we will retry 585 match self 586 .client 587 .submit_tx( 588 self.key, 589 self.account_number, 590 tx_code, 591 amount, 592 date, 593 creditor, 594 ) 595 .await 596 { 597 Ok(_) => Ok(()), 598 Err(e) => { 599 if let MagnetErr::Magnet(e) = &*e.err { 600 // Check if soft failure 601 if matches!( 602 (e.error_code, e.short_message.as_str()), 603 (409, "TRANZAKCIO_ROSSZ_STATUS") // Already summited or cannot be signed 604 ) { 605 warn!(target: "worker", "submit tx {tx_code}: {e}"); 606 return Ok(()); 607 } 608 } 609 Err(e.into()) 610 } 611 } 612 } 613 } 614 615 pub enum Tx { 616 In(TxIn), 617 Out(TxOut), 618 } 619 620 pub fn extract_tx_info(tx: TxDto) -> Tx { 621 // TODO amount from f64 without allocations 622 let amount = amount::amount(format!("{}:{}", tx.currency, tx.amount.abs())); 623 // TODO we should support non hungarian account and error handling 624 let iban = if tx.counter_account.starts_with("HU") { 625 let iban: IBAN = tx.counter_account.parse().unwrap(); 626 HuIban::try_from(iban).unwrap() 627 } else { 628 HuIban::from_bban(&tx.counter_account).unwrap() 629 }; 630 let counter_account = FullHuPayto::new(iban, &tx.counter_name); 631 if tx.amount.is_sign_positive() { 632 Tx::In(TxIn { 633 code: tx.code, 634 amount, 635 subject: tx.subject.unwrap_or_default(), 636 debtor: counter_account, 637 value_date: tx.value_date, 638 status: tx.status, 639 }) 640 } else { 641 Tx::Out(TxOut { 642 code: tx.code, 643 amount, 644 subject: tx.subject.unwrap_or_default(), 645 creditor: counter_account, 646 value_date: tx.value_date, 647 status: tx.status, 648 }) 649 } 650 } 651 652 #[derive(Debug, thiserror::Error)] 653 pub enum BounceSubjectErr { 654 #[error("missing parts")] 655 MissingParts, 656 #[error("not a bounce")] 657 NotBounce, 658 #[error("malformed bounced id: {0}")] 659 Id(#[from] ParseIntError), 660 } 661 662 pub fn parse_bounce_outgoing(subject: &str) -> Result<u32, BounceSubjectErr> { 663 let (prefix, id) = subject 664 .rsplit_once(" ") 665 .ok_or(BounceSubjectErr::MissingParts)?; 666 if !prefix.starts_with("bounce") { 667 return Err(BounceSubjectErr::NotBounce); 668 } 669 let id: u32 = id.parse()?; 670 Ok(id) 671 } 672 673 #[cfg(test)] 674 mod retry_tests { 675 use super::*; 676 677 #[test] 678 fn database_outages_retry_but_terminal_interruptions_propagate() { 679 let mut jitter = ExpoBackoffDecorr::default(); 680 let err = WorkerError::Db(sqlx::Error::Io( 681 std::io::ErrorKind::ConnectionRefused.into(), 682 )); 683 let delay = retry_delay(&err, &mut jitter).unwrap(); 684 assert!(delay >= Duration::from_millis(400)); 685 assert!(delay <= Duration::from_secs(30)); 686 assert!(retry_delay(&WorkerError::Db(sqlx::Error::PoolClosed), &mut jitter).is_none()); 687 assert!( 688 retry_delay( 689 &WorkerError::Injected(InjectedErr("worker interrupted")), 690 &mut jitter 691 ) 692 .is_none() 693 ); 694 assert!( 695 retry_delay(&WorkerError::Concurrency, &mut jitter).unwrap() >= Duration::from_secs(15) 696 ); 697 } 698 }