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