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