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