taler-merchant-reconciliation.c (39534B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2023-2025 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 * @file src/backend/taler-merchant-reconciliation.c 18 * @brief Process that reconciles information about incoming bank transfers with orders by asking the exchange 19 * @author Christian Grothoff 20 */ 21 #include "platform.h" 22 struct Inquiry; 23 #define TALER_EXCHANGE_GET_TRANSFERS_RESULT_CLOSURE struct Inquiry 24 #include "microhttpd.h" 25 #include <gnunet/gnunet_util_lib.h> 26 #include <jansson.h> 27 #include <pthread.h> 28 #include <taler/taler_dbevents.h> 29 #include "taler/taler_merchant_util.h" 30 #include "taler/taler_merchant_bank_lib.h" 31 #include "merchantdb_lib.h" 32 #include "merchantdb_lib.h" 33 #include "merchant-database/update_to_expected_transfer_finalized.h" 34 #include "merchant-database/insert_transfer_details.h" 35 #include "merchant-database/iterate_deposits_by_contract_and_coin.h" 36 #include "merchant-database/get_exchange_wire_fee.h" 37 #include "merchant-database/get_exchange_keys.h" 38 #include "merchant-database/iterate_open_expected_transfers.h" 39 #include "merchant-database/set_instance.h" 40 #include "merchant-database/update_expected_transfer_status.h" 41 #include "merchant-database/event_listen.h" 42 #include "merchant-database/preflight.h" 43 44 /** 45 * Timeout for the exchange interaction. Rather long as we should do 46 * long-polling and do not want to wake up too often. 47 */ 48 #define EXCHANGE_TIMEOUT GNUNET_TIME_relative_multiply ( \ 49 GNUNET_TIME_UNIT_MINUTES, \ 50 30) 51 52 /** 53 * How many inquiries do we process concurrently at most. 54 */ 55 #define OPEN_INQUIRY_LIMIT 1024 56 57 /** 58 * How many inquiries do we process concurrently per exchange at most. 59 */ 60 #define EXCHANGE_INQUIRY_LIMIT 16 61 62 63 /** 64 * Information about an inquiry job. 65 */ 66 struct Inquiry; 67 68 69 /** 70 * Information about an exchange. 71 */ 72 struct Exchange 73 { 74 /** 75 * Kept in a DLL. 76 */ 77 struct Exchange *next; 78 79 /** 80 * Kept in a DLL. 81 */ 82 struct Exchange *prev; 83 84 /** 85 * Head of active inquiries. 86 */ 87 struct Inquiry *w_head; 88 89 /** 90 * Tail of active inquiries. 91 */ 92 struct Inquiry *w_tail; 93 94 /** 95 * Which exchange are we tracking here. 96 */ 97 char *exchange_url; 98 99 /** 100 * The keys of this exchange 101 */ 102 struct TALER_EXCHANGE_Keys *keys; 103 104 /** 105 * How many active inquiries do we have right now with this exchange. 106 */ 107 unsigned int exchange_inquiries; 108 109 /** 110 * How long should we wait between requests 111 * for transfer details? 112 */ 113 struct GNUNET_TIME_Relative transfer_delay; 114 115 }; 116 117 118 /** 119 * Information about an inquiry job. 120 */ 121 struct Inquiry 122 { 123 /** 124 * Kept in a DLL. 125 */ 126 struct Inquiry *next; 127 128 /** 129 * Kept in a DLL. 130 */ 131 struct Inquiry *prev; 132 133 /** 134 * Handle to the exchange that made the transfer. 135 */ 136 struct Exchange *exchange; 137 138 /** 139 * Task where we retry fetching transfer details from the exchange. 140 */ 141 struct GNUNET_SCHEDULER_Task *task; 142 143 /** 144 * For which merchant instance is this tracking request? 145 */ 146 char *instance_id; 147 148 /** 149 * payto:// URI used for the transfer. 150 */ 151 struct TALER_FullPayto payto_uri; 152 153 /** 154 * Handle for the GET /transfers request. 155 */ 156 struct TALER_EXCHANGE_GetTransfersHandle *wdh; 157 158 /** 159 * When did the transfer happen? 160 */ 161 struct GNUNET_TIME_Timestamp execution_time; 162 163 /** 164 * Argument for the /wire/transfers request. 165 */ 166 struct TALER_WireTransferIdentifierRawP wtid; 167 168 /** 169 * Row of the wire transfer in our database. 170 */ 171 uint64_t rowid; 172 173 }; 174 175 176 /** 177 * Head of known exchanges. 178 */ 179 static struct Exchange *e_head; 180 181 /** 182 * Tail of known exchanges. 183 */ 184 static struct Exchange *e_tail; 185 186 /** 187 * The merchant's configuration. 188 */ 189 static const struct GNUNET_CONFIGURATION_Handle *cfg; 190 191 /** 192 * Our database connection. 193 */ 194 static struct TALER_MERCHANTDB_PostgresContext *pg; 195 196 /** 197 * Handle to the context for interacting with the bank. 198 */ 199 static struct GNUNET_CURL_Context *ctx; 200 201 /** 202 * Scheduler context for running the @e ctx. 203 */ 204 static struct GNUNET_CURL_RescheduleContext *rc; 205 206 /** 207 * Main task for #find_work(). 208 */ 209 static struct GNUNET_SCHEDULER_Task *task; 210 211 /** 212 * Event handler to learn that there are new transfers 213 * to check. 214 */ 215 static struct GNUNET_DB_EventHandler *eh; 216 217 /** 218 * Event handler to learn that there may be new exchange 219 * keys to check. 220 */ 221 static struct GNUNET_DB_EventHandler *eh_keys; 222 223 /** 224 * How many active inquiries do we have right now. 225 */ 226 static unsigned int active_inquiries; 227 228 /** 229 * Set to true if we ever encountered any problem. 230 */ 231 static bool found_problem; 232 233 /** 234 * Value to return from main(). 0 on success, non-zero on errors. 235 */ 236 static int global_ret; 237 238 /** 239 * Should we enable HTTP/2 and HTTP/3 when talking to the exchange? 240 * Those are not expected to be terribly beneficial for a client with 241 * stable connections to a few servers, but they could cause stability 242 * issues with libcurl. Hence we *default* to HTTP/1.1-only, as that 243 * is the conservative and most tested code path. 244 */ 245 static int enable_h3; 246 247 /** 248 * #GNUNET_YES if we are in test mode and should exit when idle. 249 */ 250 static int test_mode; 251 252 /** 253 * True if the last DB query was limited by the 254 * #OPEN_INQUIRY_LIMIT and we thus should check again 255 * as soon as we are substantially below that limit, 256 * and not only when we get a DB notification. 257 */ 258 static bool at_limit; 259 260 261 /** 262 * Initiate download from exchange. 263 * 264 * @param cls a `struct Inquiry *` 265 */ 266 static void 267 exchange_request (void *cls); 268 269 270 /** 271 * The exchange @a e is ready to handle more inquiries, 272 * prepare to launch them. 273 * 274 * @param[in,out] e exchange to potentially launch inquiries on 275 */ 276 static void 277 launch_inquiries_at_exchange (struct Exchange *e) 278 { 279 for (struct Inquiry *w = e->w_head; 280 NULL != w; 281 w = w->next) 282 { 283 if (e->exchange_inquiries >= EXCHANGE_INQUIRY_LIMIT) 284 break; 285 if ( (NULL == w->task) && 286 (NULL == w->wdh) ) 287 { 288 e->exchange_inquiries++; 289 w->task = GNUNET_SCHEDULER_add_now (&exchange_request, 290 w); 291 } 292 } 293 } 294 295 296 /** 297 * Updates the transaction status for inquiry @a w to the given values. 298 * 299 * @param w inquiry to update status for 300 * @param next_attempt when should we retry @a w (if ever) 301 * @param http_status HTTP status of the response 302 * @param ec error code to use (if any) 303 * @param last_hint hint delivered with the response (if any, possibly NULL) 304 * @param needs_retry true if we should try the HTTP request again 305 */ 306 static void 307 update_transaction_status (const struct Inquiry *w, 308 struct GNUNET_TIME_Absolute next_attempt, 309 unsigned int http_status, 310 enum TALER_ErrorCode ec, 311 const char *last_hint, 312 bool needs_retry) 313 { 314 enum GNUNET_DB_QueryStatus qs; 315 316 qs = TALER_MERCHANTDB_set_instance (pg, 317 w->instance_id); 318 if (qs <= 0) 319 { 320 GNUNET_break (0); 321 global_ret = EXIT_FAILURE; 322 GNUNET_SCHEDULER_shutdown (); 323 return; 324 } 325 qs = TALER_MERCHANTDB_update_expected_transfer_status (pg, 326 w->exchange->exchange_url, 327 &w->wtid, 328 next_attempt, 329 http_status, 330 ec, 331 last_hint, 332 needs_retry); 333 GNUNET_break (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == 334 TALER_MERCHANTDB_set_instance ( 335 pg, 336 NULL)); 337 if (qs < 0) 338 { 339 GNUNET_break (0); 340 global_ret = EXIT_FAILURE; 341 GNUNET_SCHEDULER_shutdown (); 342 return; 343 } 344 } 345 346 347 /** 348 * Interact with the database to get the current set 349 * of exchange keys known to us. 350 * 351 * @param e the exchange to check 352 */ 353 static void 354 sync_keys (struct Exchange *e) 355 { 356 enum GNUNET_DB_QueryStatus qs; 357 struct TALER_EXCHANGE_Keys *keys; 358 struct GNUNET_TIME_Absolute first_retry; 359 360 qs = TALER_MERCHANTDB_get_exchange_keys (pg, 361 e->exchange_url, 362 &first_retry, 363 &keys); 364 if (qs < 0) 365 { 366 GNUNET_break (0); 367 return; 368 } 369 if ( (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) || 370 (NULL == keys) ) 371 { 372 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 373 "Cannot launch inquiries at `%s': lacking /keys response\n", 374 e->exchange_url); 375 return; 376 } 377 TALER_EXCHANGE_keys_decref (e->keys); 378 e->keys = keys; 379 launch_inquiries_at_exchange (e); 380 } 381 382 383 /** 384 * Lookup our internal data structure for the given 385 * @a exchange_url or create one if we do not yet have 386 * one. 387 * 388 * @param exchange_url base URL of the exchange 389 * @return our state for this exchange 390 */ 391 static struct Exchange * 392 find_exchange (const char *exchange_url) 393 { 394 struct Exchange *e; 395 396 for (e = e_head; NULL != e; e = e->next) 397 if (0 == strcmp (exchange_url, 398 e->exchange_url)) 399 return e; 400 e = GNUNET_new (struct Exchange); 401 e->exchange_url = GNUNET_strdup (exchange_url); 402 GNUNET_CONTAINER_DLL_insert (e_head, 403 e_tail, 404 e); 405 sync_keys (e); 406 return e; 407 } 408 409 410 /** 411 * Finds new transfers that require work in the merchant database. 412 * 413 * @param cls NULL 414 */ 415 static void 416 find_work (void *cls); 417 418 419 /** 420 * Free resources of @a w. 421 * 422 * @param[in] w inquiry job to terminate 423 */ 424 static void 425 end_inquiry (struct Inquiry *w) 426 { 427 struct Exchange *e = w->exchange; 428 429 GNUNET_assert (active_inquiries > 0); 430 active_inquiries--; 431 /* The exchange-request slot (e->exchange_inquiries) is taken when the 432 exchange_request task is scheduled (w->task) and held until either that 433 task fails to create the request or the request completes. Reclaim it 434 here for whichever of the (mutually exclusive) states is active, and 435 cancel a still-queued task to avoid a use-after-free on the freed w. */ 436 if (NULL != w->task) 437 { 438 GNUNET_SCHEDULER_cancel (w->task); 439 w->task = NULL; 440 GNUNET_assert (e->exchange_inquiries > 0); 441 e->exchange_inquiries--; 442 } 443 if (NULL != w->wdh) 444 { 445 TALER_EXCHANGE_get_transfers_cancel (w->wdh); 446 w->wdh = NULL; 447 GNUNET_assert (e->exchange_inquiries > 0); 448 e->exchange_inquiries--; 449 } 450 GNUNET_free (w->instance_id); 451 GNUNET_free (w->payto_uri.full_payto); 452 GNUNET_CONTAINER_DLL_remove (e->w_head, 453 e->w_tail, 454 w); 455 GNUNET_free (w); 456 if ( (active_inquiries < OPEN_INQUIRY_LIMIT / 2) && 457 (NULL == task) && 458 (at_limit) ) 459 { 460 at_limit = false; 461 GNUNET_assert (NULL == task); 462 task = GNUNET_SCHEDULER_add_now (&find_work, 463 NULL); 464 } 465 if ( (NULL == task) && 466 (! at_limit) && 467 (0 == active_inquiries) && 468 (test_mode) ) 469 { 470 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 471 "No more open inquiries and in test mode. Exiting.\n"); 472 GNUNET_SCHEDULER_shutdown (); 473 return; 474 } 475 } 476 477 478 /** 479 * We're being aborted with CTRL-C (or SIGTERM). Shut down. 480 * 481 * @param cls closure (NULL) 482 */ 483 static void 484 shutdown_task (void *cls) 485 { 486 (void) cls; 487 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 488 "Running shutdown\n"); 489 while (NULL != e_head) 490 { 491 struct Exchange *e = e_head; 492 493 while (NULL != e->w_head) 494 { 495 struct Inquiry *w = e->w_head; 496 497 end_inquiry (w); 498 } 499 GNUNET_free (e->exchange_url); 500 if (NULL != e->keys) 501 { 502 TALER_EXCHANGE_keys_decref (e->keys); 503 e->keys = NULL; 504 } 505 GNUNET_CONTAINER_DLL_remove (e_head, 506 e_tail, 507 e); 508 GNUNET_free (e); 509 } 510 if (NULL != eh) 511 { 512 TALER_MERCHANTDB_event_listen_cancel (eh); 513 eh = NULL; 514 } 515 if (NULL != eh_keys) 516 { 517 TALER_MERCHANTDB_event_listen_cancel (eh_keys); 518 eh_keys = NULL; 519 } 520 if (NULL != task) 521 { 522 GNUNET_SCHEDULER_cancel (task); 523 task = NULL; 524 } 525 if (NULL != pg) 526 { 527 TALER_MERCHANTDB_disconnect (pg); 528 pg = NULL; 529 } 530 cfg = NULL; 531 if (NULL != ctx) 532 { 533 GNUNET_CURL_fini (ctx); 534 ctx = NULL; 535 } 536 if (NULL != rc) 537 { 538 GNUNET_CURL_gnunet_rc_destroy (rc); 539 rc = NULL; 540 } 541 } 542 543 544 /** 545 * Check that the given @a wire_fee is what the @a e should charge 546 * at the @a execution_time. If the fee is correct (according to our 547 * database), return #GNUNET_OK. If we do not have the fee structure in our 548 * DB, we just accept it and return #GNUNET_NO; if we have proof that the fee 549 * is bogus, we respond with the proof to the client and return 550 * #GNUNET_SYSERR. 551 * 552 * @param w inquiry to check fees of 553 * @param execution_time time of the wire transfer 554 * @param wire_fee fee claimed by the exchange 555 * @return #GNUNET_SYSERR if we returned hard proof of 556 * missbehavior from the exchange to the client 557 */ 558 static enum GNUNET_GenericReturnValue 559 check_wire_fee (struct Inquiry *w, 560 struct GNUNET_TIME_Timestamp execution_time, 561 const struct TALER_Amount *wire_fee) 562 { 563 struct Exchange *e = w->exchange; 564 const struct TALER_EXCHANGE_Keys *keys = e->keys; 565 struct TALER_WireFeeSet fees; 566 struct TALER_MasterSignatureP master_sig; 567 struct GNUNET_TIME_Timestamp start_date; 568 struct GNUNET_TIME_Timestamp end_date; 569 enum GNUNET_DB_QueryStatus qs; 570 char *wire_method; 571 572 if (NULL == keys) 573 { 574 GNUNET_break (0); 575 return GNUNET_NO; 576 } 577 wire_method = TALER_payto_get_method (w->payto_uri.full_payto); 578 qs = TALER_MERCHANTDB_get_exchange_wire_fee (pg, 579 &keys->master_pub, 580 wire_method, 581 execution_time, 582 &fees, 583 &start_date, 584 &end_date, 585 &master_sig); 586 switch (qs) 587 { 588 case GNUNET_DB_STATUS_HARD_ERROR: 589 GNUNET_break (0); 590 GNUNET_free (wire_method); 591 return GNUNET_SYSERR; 592 case GNUNET_DB_STATUS_SOFT_ERROR: 593 GNUNET_free (wire_method); 594 return GNUNET_NO; 595 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 596 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 597 "Failed to find wire fee for `%s' and method `%s' at %s in DB, accepting blindly that the fee is %s\n", 598 TALER_B2S (&keys->master_pub), 599 wire_method, 600 GNUNET_TIME_timestamp2s (execution_time), 601 TALER_amount2s (wire_fee)); 602 GNUNET_free (wire_method); 603 return GNUNET_OK; 604 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 605 break; 606 } 607 if ( (GNUNET_OK != 608 TALER_amount_cmp_currency (&fees.wire, 609 wire_fee)) || 610 (0 > TALER_amount_cmp (&fees.wire, 611 wire_fee)) ) 612 { 613 GNUNET_break_op (0); 614 GNUNET_free (wire_method); 615 return GNUNET_SYSERR; /* expected_fee >= wire_fee */ 616 } 617 GNUNET_free (wire_method); 618 return GNUNET_OK; 619 } 620 621 622 /** 623 * Closure for #check_transfer() 624 */ 625 struct CheckTransferContext 626 { 627 628 /** 629 * Pointer to the detail that we are currently 630 * checking in #check_transfer(). 631 */ 632 const struct TALER_TrackTransferDetails *current_detail; 633 634 /** 635 * Which transaction detail are we currently looking at? 636 */ 637 unsigned int current_offset; 638 639 /** 640 * #GNUNET_NO if we did not find a matching coin. 641 * #GNUNET_SYSERR if we found a matching coin, but the amounts do not match. 642 * #GNUNET_OK if we did find a matching coin. 643 */ 644 enum GNUNET_GenericReturnValue check_transfer_result; 645 646 /** 647 * Set to error code, if any. 648 */ 649 enum TALER_ErrorCode ec; 650 651 /** 652 * Set to true if @e ec indicates a permanent failure. 653 */ 654 bool failure; 655 }; 656 657 658 /** 659 * This function checks that the information about the coin which 660 * was paid back by _this_ wire transfer matches what _we_ (the merchant) 661 * knew about this coin. 662 * 663 * @param cls closure with our `struct CheckTransferContext *` 664 * @param exchange_url URL of the exchange that issued @a coin_pub 665 * @param amount_with_fee amount the exchange will transfer for this coin 666 * @param deposit_fee fee the exchange will charge for this coin 667 * @param refund_fee fee the exchange will charge for refunding this coin 668 * @param wire_fee paid wire fee 669 * @param h_wire hash of merchant's wire details 670 * @param deposit_timestamp when did the exchange receive the deposit 671 * @param refund_deadline until when are refunds allowed 672 * @param exchange_sig signature by the exchange 673 * @param exchange_pub exchange signing key used for @a exchange_sig 674 */ 675 static void 676 check_transfer (void *cls, 677 const char *exchange_url, 678 const struct TALER_Amount *amount_with_fee, 679 const struct TALER_Amount *deposit_fee, 680 const struct TALER_Amount *refund_fee, 681 const struct TALER_Amount *wire_fee, 682 const struct TALER_MerchantWireHashP *h_wire, 683 struct GNUNET_TIME_Timestamp deposit_timestamp, 684 struct GNUNET_TIME_Timestamp refund_deadline, 685 const struct TALER_ExchangeSignatureP *exchange_sig, 686 const struct TALER_ExchangePublicKeyP *exchange_pub) 687 { 688 struct CheckTransferContext *ctc = cls; 689 const struct TALER_TrackTransferDetails *ttd = ctc->current_detail; 690 691 if (GNUNET_SYSERR == ctc->check_transfer_result) 692 { 693 GNUNET_break (0); 694 return; /* already had a serious issue; odd that we're called more than once as well... */ 695 } 696 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 697 "Checking coin with value %s\n", 698 TALER_amount2s (amount_with_fee)); 699 if ( (GNUNET_OK != 700 TALER_amount_cmp_currency (amount_with_fee, 701 &ttd->coin_value)) || 702 (0 != TALER_amount_cmp (amount_with_fee, 703 &ttd->coin_value)) ) 704 { 705 /* Disagreement between the exchange and us about how much this 706 coin is worth! */ 707 GNUNET_break_op (0); 708 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 709 "Disagreement about coin value %s\n", 710 TALER_amount2s (amount_with_fee)); 711 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 712 "Exchange gave it a value of %s\n", 713 TALER_amount2s (&ttd->coin_value)); 714 ctc->check_transfer_result = GNUNET_SYSERR; 715 /* Build the `TrackTransferConflictDetails` */ 716 ctc->ec = TALER_EC_MERCHANT_PRIVATE_POST_TRANSFERS_CONFLICTING_REPORTS; 717 ctc->failure = true; 718 /* FIXME-#9426: this should be reported to the auditor (once the auditor has an API for this) */ 719 return; 720 } 721 if ( (GNUNET_OK != 722 TALER_amount_cmp_currency (deposit_fee, 723 &ttd->coin_fee)) || 724 (0 != TALER_amount_cmp (deposit_fee, 725 &ttd->coin_fee)) ) 726 { 727 /* Disagreement between the exchange and us about how much this 728 coin is worth! */ 729 GNUNET_break_op (0); 730 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 731 "Expected fee is %s\n", 732 TALER_amount2s (&ttd->coin_fee)); 733 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 734 "Fee claimed by exchange is %s\n", 735 TALER_amount2s (deposit_fee)); 736 ctc->check_transfer_result = GNUNET_SYSERR; 737 /* Build the `TrackTransferConflictDetails` */ 738 ctc->ec = TALER_EC_MERCHANT_PRIVATE_POST_TRANSFERS_CONFLICTING_REPORTS; 739 ctc->failure = true; 740 /* FIXME-#9426: this should be reported to the auditor (once the auditor has an API for this) */ 741 return; 742 } 743 ctc->check_transfer_result = GNUNET_OK; 744 } 745 746 747 /** 748 * Function called with detailed wire transfer data, including all 749 * of the coin transactions that were combined into the wire transfer. 750 * 751 * @param cls closure a `struct Inquiry *` 752 * @param tgr response details 753 */ 754 static void 755 wire_transfer_cb (struct Inquiry *w, 756 const struct TALER_EXCHANGE_GetTransfersResponse *tgr) 757 { 758 struct Exchange *e = w->exchange; 759 const struct TALER_EXCHANGE_TransferData *td = NULL; 760 761 GNUNET_assert (e->exchange_inquiries > 0); 762 e->exchange_inquiries--; 763 w->wdh = NULL; 764 if (EXCHANGE_INQUIRY_LIMIT - 1 == e->exchange_inquiries) 765 launch_inquiries_at_exchange (e); 766 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 767 "Got response code %u from exchange for GET /transfers/$WTID\n", 768 tgr->hr.http_status); 769 switch (tgr->hr.http_status) 770 { 771 case MHD_HTTP_OK: 772 td = &tgr->details.ok.td; 773 w->execution_time = td->execution_time; 774 e->transfer_delay = GNUNET_TIME_UNIT_ZERO; 775 break; 776 case MHD_HTTP_BAD_REQUEST: 777 case MHD_HTTP_FORBIDDEN: 778 case MHD_HTTP_NOT_FOUND: 779 found_problem = true; 780 update_transaction_status (w, 781 GNUNET_TIME_UNIT_FOREVER_ABS, 782 tgr->hr.http_status, 783 tgr->hr.ec, 784 tgr->hr.hint, 785 false); 786 end_inquiry (w); 787 return; 788 case MHD_HTTP_INTERNAL_SERVER_ERROR: 789 case MHD_HTTP_BAD_GATEWAY: 790 case MHD_HTTP_GATEWAY_TIMEOUT: 791 e->transfer_delay = GNUNET_TIME_STD_BACKOFF (e->transfer_delay); 792 update_transaction_status (w, 793 GNUNET_TIME_relative_to_absolute ( 794 e->transfer_delay), 795 tgr->hr.http_status, 796 tgr->hr.ec, 797 tgr->hr.hint, 798 true); 799 end_inquiry (w); 800 return; 801 default: 802 found_problem = true; 803 e->transfer_delay = GNUNET_TIME_STD_BACKOFF (e->transfer_delay); 804 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 805 "Unexpected HTTP status %u\n", 806 tgr->hr.http_status); 807 update_transaction_status (w, 808 GNUNET_TIME_relative_to_absolute ( 809 e->transfer_delay), 810 tgr->hr.http_status, 811 tgr->hr.ec, 812 tgr->hr.hint, 813 true); 814 end_inquiry (w); 815 return; 816 } 817 TALER_MERCHANTDB_preflight (pg); 818 819 { 820 enum GNUNET_DB_QueryStatus qs; 821 822 qs = TALER_MERCHANTDB_set_instance (pg, 823 w->instance_id); 824 if (0 > qs) 825 { 826 /* Always report on DB error as well to enable diagnostics */ 827 GNUNET_break (GNUNET_DB_STATUS_HARD_ERROR == qs); 828 end_inquiry (w); 829 global_ret = EXIT_FAILURE; 830 GNUNET_SCHEDULER_shutdown (); 831 return; 832 } 833 qs = TALER_MERCHANTDB_insert_transfer_details (pg, 834 w->instance_id, 835 w->exchange->exchange_url, 836 w->payto_uri, 837 &w->wtid, 838 td, 839 NULL); 840 if (0 > qs) 841 { 842 /* Always report on DB error as well to enable diagnostics */ 843 GNUNET_break (GNUNET_DB_STATUS_HARD_ERROR == qs); 844 end_inquiry (w); 845 global_ret = EXIT_FAILURE; 846 GNUNET_SCHEDULER_shutdown (); 847 return; 848 } 849 // FIXME: insert_transfer_details has more complex 850 // error possibilities inside, expose them here 851 // and persist them with the transaction status 852 // if they arise (especially no_account, no_exchange, conflict) 853 // -- not sure how no_instance could happen... 854 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 855 { 856 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 857 "Transfer already known. Ignoring duplicate.\n"); 858 GNUNET_break (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == 859 TALER_MERCHANTDB_set_instance ( 860 pg, 861 NULL)); 862 end_inquiry (w); 863 return; 864 } 865 } 866 867 { 868 struct CheckTransferContext ctc = { 869 .ec = TALER_EC_NONE, 870 .failure = false 871 }; 872 873 for (unsigned int i = 0; i<td->details_length; i++) 874 { 875 const struct TALER_TrackTransferDetails *ttd = &td->details[i]; 876 enum GNUNET_DB_QueryStatus qs; 877 878 if (TALER_EC_NONE != ctc.ec) 879 break; /* already encountered an error */ 880 ctc.current_offset = i; 881 ctc.current_detail = ttd; 882 /* Set the coin as "never seen" before. */ 883 ctc.check_transfer_result = GNUNET_NO; 884 qs = TALER_MERCHANTDB_iterate_deposits_by_contract_and_coin ( 885 pg, 886 w->instance_id, 887 &ttd->h_contract_terms, 888 &ttd->coin_pub, 889 &check_transfer, 890 &ctc); 891 switch (qs) 892 { 893 case GNUNET_DB_STATUS_SOFT_ERROR: 894 GNUNET_break (0); 895 ctc.ec = TALER_EC_GENERIC_DB_FETCH_FAILED; 896 break; 897 case GNUNET_DB_STATUS_HARD_ERROR: 898 GNUNET_break (0); 899 ctc.ec = TALER_EC_GENERIC_DB_FETCH_FAILED; 900 break; 901 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 902 /* The exchange says we made this deposit, but WE do not 903 recall making it (corrupted / unreliable database?)! 904 Well, let's say thanks and accept the money! */ 905 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 906 "Failed to find payment data in DB\n"); 907 ctc.check_transfer_result = GNUNET_OK; 908 break; 909 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 910 break; 911 } 912 switch (ctc.check_transfer_result) 913 { 914 case GNUNET_NO: 915 /* Internal error: how can we have called #check_transfer() 916 but still have no result? */ 917 GNUNET_break (0); 918 ctc.ec = TALER_EC_GENERIC_INTERNAL_INVARIANT_FAILURE; 919 GNUNET_break (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == 920 TALER_MERCHANTDB_set_instance ( 921 pg, 922 NULL)); 923 end_inquiry (w); 924 return; 925 case GNUNET_SYSERR: 926 /* #check_transfer() failed, report conflict! */ 927 GNUNET_break_op (0); 928 GNUNET_assert (TALER_EC_NONE != ctc.ec); 929 break; 930 case GNUNET_OK: 931 break; 932 } 933 } 934 if (TALER_EC_NONE != ctc.ec) 935 { 936 update_transaction_status ( 937 w, 938 ctc.failure 939 ? GNUNET_TIME_UNIT_FOREVER_ABS 940 : GNUNET_TIME_relative_to_absolute ( 941 GNUNET_TIME_UNIT_MINUTES), 942 MHD_HTTP_OK, 943 ctc.ec, 944 NULL /* no hint */, 945 ! ctc.failure); 946 GNUNET_break (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == 947 TALER_MERCHANTDB_set_instance ( 948 pg, 949 NULL)); 950 end_inquiry (w); 951 return; 952 } 953 } 954 955 if (GNUNET_SYSERR == 956 check_wire_fee (w, 957 td->execution_time, 958 &td->wire_fee)) 959 { 960 GNUNET_break_op (0); 961 update_transaction_status (w, 962 GNUNET_TIME_UNIT_FOREVER_ABS, 963 MHD_HTTP_OK, 964 TALER_EC_MERCHANT_PRIVATE_POST_TRANSFERS_BAD_WIRE_FEE, 965 TALER_amount2s (&td->wire_fee), 966 false); 967 GNUNET_break (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == 968 TALER_MERCHANTDB_set_instance ( 969 pg, 970 NULL)); 971 end_inquiry (w); 972 return; 973 } 974 975 { 976 enum GNUNET_DB_QueryStatus qs; 977 978 qs = TALER_MERCHANTDB_update_to_expected_transfer_finalized (pg, 979 w->exchange->exchange_url, 980 &w->wtid, 981 &td->h_details, 982 &td->total_amount, 983 &td->wire_fee, 984 &td->exchange_pub, 985 &td->exchange_sig, 986 td->exchange_payto_uri); 987 if (qs < 0) 988 { 989 GNUNET_break (0); 990 global_ret = EXIT_FAILURE; 991 GNUNET_SCHEDULER_shutdown (); 992 return; 993 } 994 } 995 GNUNET_break (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == 996 TALER_MERCHANTDB_set_instance ( 997 pg, 998 NULL)); 999 end_inquiry (w); 1000 } 1001 1002 1003 /** 1004 * Initiate download from an exchange for a given inquiry. 1005 * 1006 * @param cls a `struct Inquiry *` 1007 */ 1008 static void 1009 exchange_request (void *cls) 1010 { 1011 struct Inquiry *w = cls; 1012 struct Exchange *e = w->exchange; 1013 1014 w->task = NULL; 1015 if (NULL == e->keys) 1016 return; 1017 w->wdh = TALER_EXCHANGE_get_transfers_create ( 1018 ctx, 1019 e->exchange_url, 1020 e->keys, 1021 &w->wtid); 1022 if (NULL == w->wdh) 1023 { 1024 GNUNET_break (0); 1025 e->exchange_inquiries--; 1026 e->transfer_delay = GNUNET_TIME_STD_BACKOFF (e->transfer_delay); 1027 update_transaction_status (w, 1028 GNUNET_TIME_relative_to_absolute ( 1029 e->transfer_delay), 1030 0 /* failed to begin */, 1031 TALER_EC_MERCHANT_EXCHANGE_TRANSFERS_TRANSIENT_FAILURE, 1032 "Failed to initiate GET request at exchange", 1033 true); 1034 end_inquiry (w); 1035 return; 1036 } 1037 GNUNET_assert (TALER_EC_NONE == 1038 TALER_EXCHANGE_get_transfers_start (w->wdh, 1039 &wire_transfer_cb, 1040 w)); 1041 1042 /* Wait at least 1m for the network transfer */ 1043 update_transaction_status (w, 1044 GNUNET_TIME_relative_to_absolute ( 1045 GNUNET_TIME_UNIT_MINUTES), 1046 0 /* timeout */, 1047 TALER_EC_MERCHANT_EXCHANGE_TRANSFERS_AWAITING_LIST, 1048 "Initiated GET with exchange", 1049 true); 1050 } 1051 1052 1053 /** 1054 * Function called with information about a transfer we 1055 * should ask the exchange about. 1056 * 1057 * @param cls closure (NULL) 1058 * @param rowid row of the transfer in the merchant database 1059 * @param instance_id instance that received the transfer 1060 * @param exchange_url base URL of the exchange that initiated the transfer 1061 * @param payto_uri account of the merchant that received the transfer 1062 * @param wtid wire transfer subject identifying the aggregation 1063 * @param next_attempt when should we next try to interact with the exchange 1064 */ 1065 static void 1066 start_inquiry ( 1067 void *cls, 1068 uint64_t rowid, 1069 const char *instance_id, 1070 const char *exchange_url, 1071 struct TALER_FullPayto payto_uri, 1072 const struct TALER_WireTransferIdentifierRawP *wtid, 1073 struct GNUNET_TIME_Absolute next_attempt) 1074 { 1075 struct Exchange *e; 1076 struct Inquiry *w; 1077 1078 (void) cls; 1079 if (GNUNET_TIME_absolute_is_future (next_attempt)) 1080 { 1081 if (NULL == task) 1082 task = GNUNET_SCHEDULER_add_at (next_attempt, 1083 &find_work, 1084 NULL); 1085 return; 1086 } 1087 active_inquiries++; 1088 1089 e = find_exchange (exchange_url); 1090 for (w = e->w_head; NULL != w; w = w->next) 1091 { 1092 if (0 == GNUNET_memcmp (&w->wtid, 1093 wtid)) 1094 { 1095 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 1096 "Already processing inquiry. Aborting ongoing inquiry\n"); 1097 end_inquiry (w); 1098 break; 1099 } 1100 } 1101 1102 w = GNUNET_new (struct Inquiry); 1103 w->payto_uri.full_payto = GNUNET_strdup (payto_uri.full_payto); 1104 w->instance_id = GNUNET_strdup (instance_id); 1105 w->rowid = rowid; 1106 w->wtid = *wtid; 1107 GNUNET_CONTAINER_DLL_insert (e->w_head, 1108 e->w_tail, 1109 w); 1110 w->exchange = e; 1111 if (NULL != w->exchange->keys) 1112 { 1113 e->exchange_inquiries++; 1114 w->task = GNUNET_SCHEDULER_add_now (&exchange_request, 1115 w); 1116 } 1117 /* Wait at least 1 minute for /keys */ 1118 update_transaction_status (w, 1119 GNUNET_TIME_relative_to_absolute ( 1120 GNUNET_TIME_UNIT_MINUTES), 1121 0 /* timeout */, 1122 TALER_EC_MERCHANT_EXCHANGE_TRANSFERS_AWAITING_KEYS, 1123 exchange_url, 1124 true); 1125 } 1126 1127 1128 static void 1129 find_work (void *cls) 1130 { 1131 enum GNUNET_DB_QueryStatus qs; 1132 int limit; 1133 1134 (void) cls; 1135 task = NULL; 1136 GNUNET_assert (OPEN_INQUIRY_LIMIT >= active_inquiries); 1137 limit = OPEN_INQUIRY_LIMIT - active_inquiries; 1138 if (0 == limit) 1139 { 1140 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1141 "Not looking for work: at limit\n"); 1142 at_limit = true; 1143 return; 1144 } 1145 at_limit = false; 1146 qs = TALER_MERCHANTDB_iterate_open_expected_transfers (pg, 1147 limit, 1148 &start_inquiry, 1149 NULL); 1150 if (qs < 0) 1151 { 1152 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1153 "Failed to obtain open transfers from database\n"); 1154 GNUNET_SCHEDULER_shutdown (); 1155 return; 1156 } 1157 if (qs >= limit) 1158 { 1159 /* DB limited response, re-trigger DB interaction 1160 the moment we significantly fall below the 1161 limit */ 1162 at_limit = true; 1163 } 1164 if (0 == active_inquiries) 1165 { 1166 if (test_mode) 1167 { 1168 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1169 "No more open inquiries and in test mode. Existing.\n"); 1170 GNUNET_SCHEDULER_shutdown (); 1171 return; 1172 } 1173 GNUNET_log ( 1174 GNUNET_ERROR_TYPE_INFO, 1175 "No open inquiries found, waiting for notification to resume\n"); 1176 } 1177 } 1178 1179 1180 /** 1181 * Function called when transfers are added to the merchant database. We look 1182 * for more work. 1183 * 1184 * @param cls closure (NULL) 1185 * @param extra additional event data provided 1186 * @param extra_size number of bytes in @a extra 1187 */ 1188 static void 1189 transfer_added (void *cls, 1190 const void *extra, 1191 size_t extra_size) 1192 { 1193 (void) cls; 1194 (void) extra; 1195 (void) extra_size; 1196 if (active_inquiries > OPEN_INQUIRY_LIMIT / 2) 1197 { 1198 /* Trigger DB only once we are substantially below the limit */ 1199 at_limit = true; 1200 return; 1201 } 1202 if (NULL != task) 1203 return; 1204 task = GNUNET_SCHEDULER_add_now (&find_work, 1205 NULL); 1206 } 1207 1208 1209 /** 1210 * Function called when keys were changed in the 1211 * merchant database. Updates ours. 1212 * 1213 * @param cls closure (NULL) 1214 * @param extra additional event data provided 1215 * @param extra_size number of bytes in @a extra 1216 */ 1217 static void 1218 keys_changed (void *cls, 1219 const void *extra, 1220 size_t extra_size) 1221 { 1222 const char *url = extra; 1223 struct Exchange *e; 1224 1225 (void) cls; 1226 if ( (NULL == extra) || 1227 (0 == extra_size) ) 1228 { 1229 GNUNET_break (0); 1230 return; 1231 } 1232 if ('\0' != url[extra_size - 1]) 1233 { 1234 GNUNET_break (0); 1235 return; 1236 } 1237 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1238 "Received keys change notification: reload `%s'\n", 1239 url); 1240 e = find_exchange (url); 1241 sync_keys (e); 1242 } 1243 1244 1245 /** 1246 * First task. 1247 * 1248 * @param cls closure, NULL 1249 * @param args remaining command-line arguments 1250 * @param cfgfile name of the configuration file used (for saving, can be NULL!) 1251 * @param c configuration 1252 */ 1253 static void 1254 run (void *cls, 1255 char *const *args, 1256 const char *cfgfile, 1257 const struct GNUNET_CONFIGURATION_Handle *c) 1258 { 1259 (void) args; 1260 (void) cfgfile; 1261 1262 cfg = c; 1263 TALER_EXCHANGE_setup (enable_h3 1264 ? TALER_EXCHANGE_GO_ENABLE_HTTP3 1265 : TALER_EXCHANGE_GO_FORCE_HTTP1_1); 1266 GNUNET_SCHEDULER_add_shutdown (&shutdown_task, 1267 NULL); 1268 ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule, 1269 &rc); 1270 rc = GNUNET_CURL_gnunet_rc_create (ctx); 1271 if (NULL == ctx) 1272 { 1273 GNUNET_break (0); 1274 GNUNET_SCHEDULER_shutdown (); 1275 global_ret = EXIT_FAILURE; 1276 return; 1277 } 1278 if (NULL == 1279 (pg = TALER_MERCHANTDB_connect (cfg))) 1280 { 1281 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1282 "Failed to initialize DB subsystem. Consider running taler-merchant-dbconfig!\n"); 1283 GNUNET_SCHEDULER_shutdown (); 1284 global_ret = EXIT_FAILURE; 1285 return; 1286 } 1287 { 1288 struct GNUNET_DB_EventHeaderP es = { 1289 .size = htons (sizeof (es)), 1290 .type = htons (TALER_DBEVENT_MERCHANT_WIRE_TRANSFER_EXPECTED) 1291 }; 1292 1293 eh = TALER_MERCHANTDB_event_listen (pg, 1294 &es, 1295 GNUNET_TIME_UNIT_FOREVER_REL, 1296 &transfer_added, 1297 NULL); 1298 } 1299 { 1300 struct GNUNET_DB_EventHeaderP es = { 1301 .size = htons (sizeof (es)), 1302 .type = htons (TALER_DBEVENT_MERCHANT_EXCHANGE_KEYS) 1303 }; 1304 1305 eh_keys 1306 = TALER_MERCHANTDB_event_listen (pg, 1307 &es, 1308 GNUNET_TIME_UNIT_FOREVER_REL, 1309 &keys_changed, 1310 NULL); 1311 } 1312 1313 GNUNET_assert (NULL == task); 1314 task = GNUNET_SCHEDULER_add_now (&find_work, 1315 NULL); 1316 } 1317 1318 1319 /** 1320 * The main function of taler-merchant-reconciliation 1321 * 1322 * @param argc number of arguments from the command line 1323 * @param argv command line arguments 1324 * @return 0 ok, 1 on error 1325 */ 1326 int 1327 main (int argc, 1328 char *const *argv) 1329 { 1330 struct GNUNET_GETOPT_CommandLineOption options[] = { 1331 GNUNET_GETOPT_option_flag ('3', 1332 "http3", 1333 "enable support for HTTP/2 and HTTP/3", 1334 &enable_h3), 1335 GNUNET_GETOPT_option_timetravel ('T', 1336 "timetravel"), 1337 GNUNET_GETOPT_option_flag ('t', 1338 "test", 1339 "run in test mode and exit when idle", 1340 &test_mode), 1341 GNUNET_GETOPT_option_version (VERSION), 1342 GNUNET_GETOPT_OPTION_END 1343 }; 1344 enum GNUNET_GenericReturnValue ret; 1345 1346 ret = GNUNET_PROGRAM_run ( 1347 TALER_MERCHANT_project_data (), 1348 argc, argv, 1349 "taler-merchant-reconciliation", 1350 gettext_noop ( 1351 "background process that reconciles bank transfers with orders by asking the exchange"), 1352 options, 1353 &run, NULL); 1354 if (GNUNET_SYSERR == ret) 1355 return EXIT_INVALIDARGUMENT; 1356 if (GNUNET_NO == ret) 1357 return EXIT_SUCCESS; 1358 if ( (found_problem) && 1359 (0 == global_ret) ) 1360 global_ret = 7; 1361 return global_ret; 1362 } 1363 1364 1365 /* end of taler-merchant-reconciliation.c */