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