taler-exchange-aggregator.c (47667B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2016-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 /** 18 * @file taler-exchange-aggregator.c 19 * @brief Process that aggregates outgoing transactions and prepares their execution 20 * @author Christian Grothoff 21 */ 22 #include "platform.h" 23 #include <gnunet/gnunet_util_lib.h> 24 #include <jansson.h> 25 #include <pthread.h> 26 #include "exchangedb_lib.h" 27 #include "taler/taler_json_lib.h" 28 #include "taler/taler_kyclogic_lib.h" 29 #include "taler/taler_bank_service.h" 30 #include "taler/taler_dbevents.h" 31 #include "exchange-database/release_revolving_shard.h" 32 #include "exchange-database/begin_revolving_shard.h" 33 #include "exchange-database/start.h" 34 #include "exchange-database/preflight.h" 35 #include "exchange-database/commit.h" 36 #include "exchange-database/rollback.h" 37 #include "exchange-database/do_drain_kyc_alert.h" 38 #include "exchange-database/get_ready_deposit.h" 39 #include "exchange-database/start_deferred_wire_out.h" 40 #include "exchange-database/do_aggregate.h" 41 #include "exchange-database/insert_aggregation_transient.h" 42 #include "exchange-database/get_aggregation_transient.h" 43 #include "exchange-database/get_aggregation_transient_by_normalized_payto.h" 44 #include "exchange-database/get_wire_fee.h" 45 #include "exchange-database/do_trigger_kyc_rule_for_account.h" 46 #include "exchange-database/iterate_aggregation_amounts_for_kyc_check.h" 47 #include "exchange-database/insert_prewire.h" 48 #include "exchange-database/insert_wire_out.h" 49 #include "exchange-database/delete_aggregation_transient.h" 50 #include "exchange-database/insert_aggregation_deferral.h" 51 #include "exchange-database/update_aggregation_transient.h" 52 #include "exchange-database/event_notify.h" 53 #include "exchange-database/insert_prewire.h" 54 struct AggregationUnit; 55 #define TALER_EXCHANGEDB_CURRENT_RULES_RESULT_CLOSURE struct AggregationUnit 56 #include "exchange-database/begin_rule_update.h" 57 58 /** 59 * How often do we retry after serialization failures? 60 */ 61 #define MAX_RETRIES 5 62 63 /** 64 * Information about one aggregation process to be executed. There is 65 * at most one of these around at any given point in time. 66 * Note that this limits parallelism, and we might want 67 * to revise this decision at a later point. 68 */ 69 struct AggregationUnit 70 { 71 /** 72 * Public key of the merchant. 73 */ 74 struct TALER_MerchantPublicKeyP merchant_pub; 75 76 /** 77 * Transient amount already found aggregated, 78 * set only if @e have_transient is true. 79 */ 80 struct TALER_Amount trans; 81 82 /** 83 * Total amount to be transferred, before subtraction of @e fees.wire and rounding down. 84 */ 85 struct TALER_Amount total_amount; 86 87 /** 88 * Final amount to be transferred (after fee and rounding down). 89 */ 90 struct TALER_Amount final_amount; 91 92 /** 93 * Wire fee we charge for @e wp at @e execution_time. 94 */ 95 struct TALER_WireFeeSet fees; 96 97 /** 98 * Wire transfer identifier we use. 99 */ 100 struct TALER_WireTransferIdentifierRawP wtid; 101 102 /** 103 * The current time (which triggered the aggregation and 104 * defines the wire fee). 105 */ 106 struct GNUNET_TIME_Timestamp execution_time; 107 108 /** 109 * Wire details of the merchant. 110 */ 111 struct TALER_FullPayto payto_uri; 112 113 /** 114 * Selected wire target for the aggregation. 115 */ 116 struct TALER_FullPaytoHashP h_full_payto; 117 118 /** 119 * Selected wire target for KYC checks. 120 */ 121 struct TALER_NormalizedPaytoHashP h_normalized_payto; 122 123 /** 124 * Exchange wire account to be used for the preparation and 125 * eventual execution of the aggregate wire transfer. 126 */ 127 const struct TALER_EXCHANGEDB_AccountInfo *wa; 128 129 /** 130 * Additional metadata to include in the wire transfer subject, 131 * can be NULL. 132 */ 133 char *extra_wire_subject_metadata; 134 135 /** 136 * Shard this aggregation unit is part of. 137 */ 138 struct Shard *shard; 139 140 /** 141 * Handle to async process to obtain the legitimization rules. 142 */ 143 struct TALER_EXCHANGEDB_RuleUpdater *ru; 144 145 /** 146 * Row in KYC table for legitimization requirements 147 * that are pending for this aggregation, or 0 if none. 148 */ 149 uint64_t requirement_row; 150 151 /** 152 * How often did we retry the transaction? 153 */ 154 unsigned int retries; 155 156 /** 157 * Should we run a follow-up transaction with a legitimization 158 * check? 159 */ 160 bool legi_check; 161 162 /** 163 * Do we have an entry in the transient table for 164 * this aggregation? 165 */ 166 bool have_transient; 167 168 /** 169 * Is the wrong merchant public key associated with 170 * the KYC data? 171 */ 172 bool bad_kyc_auth; 173 174 }; 175 176 177 /** 178 * Work shard we are processing. 179 */ 180 struct Shard 181 { 182 183 /** 184 * When did we start processing the shard? 185 */ 186 struct GNUNET_TIME_Timestamp start_time; 187 188 /** 189 * Starting row of the shard. 190 */ 191 uint32_t shard_start; 192 193 /** 194 * Inclusive end row of the shard. 195 */ 196 uint32_t shard_end; 197 198 /** 199 * Number of starting points found in the shard. 200 */ 201 uint64_t work_counter; 202 203 }; 204 205 206 /** 207 * What is the smallest unit we support for wire transfers? 208 * We will need to round down to a multiple of this amount. 209 */ 210 static struct TALER_Amount currency_round_unit; 211 212 /** 213 * What is the base URL of this exchange? Used in the 214 * wire transfer subjects so that merchants and governments 215 * can ask for the list of aggregated deposits. 216 */ 217 static char *exchange_base_url; 218 219 /** 220 * Set to #GNUNET_YES if this exchange does not support KYC checks 221 * and thus deposits are to be aggregated regardless of the 222 * KYC status of the target account. 223 */ 224 static int kyc_off; 225 226 /** 227 * The exchange's configuration. 228 */ 229 static const struct GNUNET_CONFIGURATION_Handle *cfg; 230 231 /** 232 * Key used to encrypt KYC attribute data in our database. 233 */ 234 static struct TALER_AttributeEncryptionKeyP attribute_key; 235 236 /** 237 * Our database connection. 238 */ 239 static struct TALER_EXCHANGEDB_PostgresContext *pg; 240 241 /** 242 * Next task to run, if any. 243 */ 244 static struct GNUNET_SCHEDULER_Task *task; 245 246 /** 247 * How long should we sleep when idle before trying to find more work? 248 */ 249 static struct GNUNET_TIME_Relative aggregator_idle_sleep_interval; 250 251 /** 252 * How big are the shards we are processing? Is an inclusive offset, so every 253 * shard ranges from [X,X+shard_size) exclusive. So a shard covers 254 * shard_size slots. The maximum value for shard_size is INT32_MAX+1. 255 */ 256 static uint32_t shard_size; 257 258 /** 259 * Value to return from main(). 0 on success, non-zero on errors. 260 */ 261 static int global_ret; 262 263 /** 264 * #GNUNET_YES if we are in test mode and should exit when idle. 265 */ 266 static int test_mode; 267 268 269 /** 270 * Main work function that queries the DB and aggregates transactions 271 * into larger wire transfers. 272 * 273 * @param cls a `struct Shard *` 274 */ 275 static void 276 run_aggregation (void *cls); 277 278 279 /** 280 * Work on transactions unlocked by KYC. 281 * 282 * @param cls NULL 283 */ 284 static void 285 drain_kyc_alerts (void *cls); 286 287 288 /** 289 * Free data stored in @a au, including @a au itself. 290 * 291 * @param[in] au aggregation unit to clean up 292 */ 293 static void 294 cleanup_au (struct AggregationUnit *au) 295 { 296 GNUNET_assert (NULL != au); 297 GNUNET_free (au->extra_wire_subject_metadata); 298 if (NULL != au->ru) 299 { 300 GNUNET_break (0); 301 TALER_EXCHANGEDB_begin_rule_update_cancel (au->ru); 302 au->ru = NULL; 303 } 304 GNUNET_free (au->payto_uri.full_payto); 305 GNUNET_free (au); 306 } 307 308 309 /** 310 * Perform a database commit. If it fails, print a warning. 311 * 312 * @return status of commit 313 */ 314 static enum GNUNET_DB_QueryStatus 315 commit_or_warn (void) 316 { 317 enum GNUNET_DB_QueryStatus qs; 318 319 qs = TALER_EXCHANGEDB_commit (pg); 320 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 321 return qs; 322 GNUNET_log ((GNUNET_DB_STATUS_SOFT_ERROR == qs) 323 ? GNUNET_ERROR_TYPE_INFO 324 : GNUNET_ERROR_TYPE_ERROR, 325 "Failed to commit database transaction!\n"); 326 return qs; 327 } 328 329 330 /** 331 * Release lock on shard @a s in the database. 332 * On error, terminates this process. 333 * 334 * @param[in] s shard to free (and memory to release) 335 */ 336 static void 337 release_shard (struct Shard *s) 338 { 339 enum GNUNET_DB_QueryStatus qs; 340 341 if (NULL == s) 342 return; /* aggregation units from drain_kyc_alerts have a NULL shard */ 343 qs = TALER_EXCHANGEDB_release_revolving_shard ( 344 pg, 345 "aggregator", 346 s->shard_start, 347 s->shard_end); 348 GNUNET_free (s); 349 switch (qs) 350 { 351 case GNUNET_DB_STATUS_HARD_ERROR: 352 case GNUNET_DB_STATUS_SOFT_ERROR: 353 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR != qs); 354 GNUNET_break (0); 355 global_ret = EXIT_FAILURE; 356 GNUNET_SCHEDULER_shutdown (); 357 return; 358 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 359 /* Strange, but let's just continue */ 360 break; 361 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 362 /* normal case */ 363 break; 364 } 365 } 366 367 368 /** 369 * Schedule the next major task, or exit depending on mode. 370 */ 371 static void 372 next_task (uint64_t counter) 373 { 374 if ( (GNUNET_YES == test_mode) && 375 (0 == counter) ) 376 { 377 /* in test mode, shutdown after a shard is done with 0 work */ 378 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 379 "No work done and in test mode, shutting down\n"); 380 GNUNET_SCHEDULER_shutdown (); 381 return; 382 } 383 GNUNET_assert (NULL == task); 384 /* If we ended up doing zero work, sleep a bit */ 385 if (0 == counter) 386 { 387 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 388 "Going to sleep for %s before trying again\n", 389 GNUNET_TIME_relative2s (aggregator_idle_sleep_interval, 390 true)); 391 task = GNUNET_SCHEDULER_add_delayed (aggregator_idle_sleep_interval, 392 &drain_kyc_alerts, 393 NULL); 394 } 395 else 396 { 397 task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts, 398 NULL); 399 } 400 } 401 402 403 /** 404 * Rollback the current transaction (if any), 405 * then free data stored in @a au, including @a au itself, and then 406 * run the next aggregation task. 407 * 408 * @param[in] au aggregation unit to clean up 409 */ 410 static void 411 cleanup_and_next (struct AggregationUnit *au) 412 { 413 struct Shard *s = au->shard; 414 uint64_t counter = (NULL == s) ? 0 : s->work_counter; 415 416 /* just in case, often no transaction is running here anymore */ 417 TALER_EXCHANGEDB_rollback (pg); 418 cleanup_au (au); 419 if (NULL != s) 420 release_shard (s); 421 if (EXIT_SUCCESS == global_ret) 422 next_task (counter); 423 } 424 425 426 /** 427 * We're being aborted with CTRL-C (or SIGTERM). Shut down. 428 * 429 * @param cls closure 430 */ 431 static void 432 shutdown_task (void *cls) 433 { 434 (void) cls; 435 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 436 "Running shutdown\n"); 437 if (NULL != task) 438 { 439 GNUNET_SCHEDULER_cancel (task); 440 task = NULL; 441 } 442 TALER_KYCLOGIC_kyc_done (); 443 TALER_EXCHANGEDB_disconnect (pg); 444 pg = NULL; 445 TALER_EXCHANGEDB_unload_accounts (); 446 cfg = NULL; 447 } 448 449 450 /** 451 * Parse the configuration for aggregator. 452 * 453 * @return #GNUNET_OK on success 454 */ 455 static enum GNUNET_GenericReturnValue 456 parse_aggregator_config (void) 457 { 458 enum GNUNET_GenericReturnValue enable_kyc; 459 460 enable_kyc 461 = GNUNET_CONFIGURATION_get_value_yesno ( 462 cfg, 463 "exchange", 464 "ENABLE_KYC"); 465 if (GNUNET_SYSERR == enable_kyc) 466 { 467 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 468 "Need YES or NO in section `exchange' under `ENABLE_KYC'\n"); 469 return GNUNET_SYSERR; 470 } 471 if (GNUNET_NO == enable_kyc) 472 { 473 kyc_off = true; 474 } 475 else 476 { 477 char *attr_enc_key_str; 478 479 if (GNUNET_OK != 480 GNUNET_CONFIGURATION_get_value_string (cfg, 481 "exchange", 482 "ATTRIBUTE_ENCRYPTION_KEY", 483 &attr_enc_key_str)) 484 { 485 GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR, 486 "exchange", 487 "ATTRIBUTE_ENCRYPTION_KEY"); 488 return GNUNET_SYSERR; 489 } 490 GNUNET_CRYPTO_hash (attr_enc_key_str, 491 strlen (attr_enc_key_str), 492 &attribute_key.hash); 493 GNUNET_free (attr_enc_key_str); 494 } 495 if (GNUNET_OK != 496 GNUNET_CONFIGURATION_get_value_string (cfg, 497 "exchange", 498 "BASE_URL", 499 &exchange_base_url)) 500 { 501 GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR, 502 "exchange", 503 "BASE_URL"); 504 return GNUNET_SYSERR; 505 } 506 if (GNUNET_OK != 507 GNUNET_CONFIGURATION_get_value_time (cfg, 508 "exchange", 509 "AGGREGATOR_IDLE_SLEEP_INTERVAL", 510 &aggregator_idle_sleep_interval)) 511 { 512 GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR, 513 "exchange", 514 "AGGREGATOR_IDLE_SLEEP_INTERVAL"); 515 return GNUNET_SYSERR; 516 } 517 if ( (GNUNET_OK != 518 TALER_config_get_amount (cfg, 519 "exchange", 520 "CURRENCY_ROUND_UNIT", 521 ¤cy_round_unit)) || 522 (TALER_amount_is_zero (¤cy_round_unit)) ) 523 { 524 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 525 "Need non-zero amount in section `exchange' under `CURRENCY_ROUND_UNIT'\n"); 526 return GNUNET_SYSERR; 527 } 528 529 if (NULL == 530 (pg = TALER_EXCHANGEDB_connect (cfg))) 531 { 532 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 533 "Failed to initialize DB subsystem\n"); 534 return GNUNET_SYSERR; 535 } 536 if (GNUNET_OK != 537 TALER_EXCHANGEDB_load_accounts (cfg, 538 TALER_EXCHANGEDB_ALO_DEBIT)) 539 { 540 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 541 "No wire accounts configured for debit!\n"); 542 TALER_EXCHANGEDB_disconnect (pg); 543 pg = NULL; 544 return GNUNET_SYSERR; 545 } 546 return GNUNET_OK; 547 } 548 549 550 /** 551 * Callback to return all applicable amounts for the KYC 552 * decision to @ a cb. 553 * 554 * @param cls a `struct AggregationUnit *` 555 * @param limit time limit for the iteration 556 * @param cb function to call with the amounts 557 * @param cb_cls closure for @a cb 558 * @return transaction status 559 */ 560 static enum GNUNET_DB_QueryStatus 561 return_relevant_amounts (void *cls, 562 struct GNUNET_TIME_Absolute limit, 563 TALER_KYCLOGIC_KycAmountCallback cb, 564 void *cb_cls) 565 { 566 const struct AggregationUnit *au = cls; 567 enum GNUNET_DB_QueryStatus qs; 568 569 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 570 "Returning amount %s in KYC check\n", 571 TALER_amount2s (&au->total_amount)); 572 if (GNUNET_OK != 573 cb (cb_cls, 574 &au->total_amount, 575 GNUNET_TIME_absolute_get ())) 576 return GNUNET_DB_STATUS_SUCCESS_NO_RESULTS; 577 qs = TALER_EXCHANGEDB_iterate_aggregation_amounts_for_kyc_check ( 578 pg, 579 &au->h_normalized_payto, 580 limit, 581 cb, 582 cb_cls); 583 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 584 { 585 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 586 "Failed to select aggregation amounts for KYC limit check!\n"); 587 } 588 return qs; 589 } 590 591 592 /** 593 * The aggregation process failed hard, shut down the program. 594 * 595 * @param[in] au aggregation that failed hard 596 */ 597 static void 598 fail_aggregation (struct AggregationUnit *au) 599 { 600 struct Shard *s = au->shard; 601 602 cleanup_au (au); 603 global_ret = EXIT_FAILURE; 604 GNUNET_SCHEDULER_shutdown (); 605 TALER_EXCHANGEDB_rollback (pg); 606 release_shard (s); 607 } 608 609 610 /** 611 * Run the next task with the given shard @a s. 612 * 613 * @param s shard to run, NULL to run more drain jobs 614 */ 615 static void 616 run_task_with_shard (struct Shard *s) 617 { 618 GNUNET_assert (NULL == task); 619 if (NULL == s) 620 task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts, 621 NULL); 622 else 623 task = GNUNET_SCHEDULER_add_now (&run_aggregation, 624 s); 625 } 626 627 628 /** 629 * The aggregation process failed with a serialization 630 * issue. Rollback the transaction and try again. 631 * 632 * @param[in] au aggregation that needs to be rolled back 633 */ 634 static void 635 rollback_aggregation (struct AggregationUnit *au) 636 { 637 struct Shard *s = au->shard; 638 639 cleanup_au (au); 640 TALER_EXCHANGEDB_rollback (pg); 641 run_task_with_shard (s); 642 } 643 644 645 /** 646 * Function called with legitimization rule set. Check 647 * how that affects the aggregation process. 648 * 649 * @param[in] au a `struct AggregationUnit *` 650 * @param[in] rur new legitimization rule set to evaluate 651 */ 652 static void 653 evaluate_rules ( 654 struct AggregationUnit *au, 655 struct TALER_EXCHANGEDB_RuleUpdaterResult *rur); 656 657 658 /** 659 * The aggregation process succeeded and should be finally committed. 660 * 661 * @param[in] au aggregation that needs to be committed 662 */ 663 static void 664 commit_aggregation (struct AggregationUnit *au) 665 { 666 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 667 "Committing aggregation result over %s to %s\n", 668 TALER_amount2s (&au->final_amount), 669 au->payto_uri.full_payto); 670 /* Now we can finally commit the overall transaction, as we are 671 again consistent if all of this passes. */ 672 switch (commit_or_warn ()) 673 { 674 case GNUNET_DB_STATUS_SOFT_ERROR: 675 /* try again */ 676 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 677 "Serialization issue on commit; trying again later!\n"); 678 cleanup_and_next (au); 679 return; 680 case GNUNET_DB_STATUS_HARD_ERROR: 681 GNUNET_break (0); 682 global_ret = EXIT_FAILURE; 683 GNUNET_SCHEDULER_shutdown (); 684 cleanup_and_next (au); 685 return; 686 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 687 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 688 "Commit complete, going again\n"); 689 if (au->legi_check) 690 { 691 au->legi_check = false; 692 au->ru = TALER_EXCHANGEDB_begin_rule_update ( 693 pg, 694 &attribute_key, 695 &au->h_normalized_payto, 696 false, /* aggregation doesn't apply to wallets */ 697 &evaluate_rules, 698 au); 699 if (NULL != au->ru) 700 return; 701 } 702 cleanup_and_next (au); 703 return; 704 default: 705 GNUNET_break (0); 706 global_ret = EXIT_FAILURE; 707 GNUNET_SCHEDULER_shutdown (); 708 cleanup_and_next (au); 709 return; 710 } 711 } 712 713 714 /** 715 * Trigger the wire transfer for the @a au 716 * and delete the record of the aggregation. 717 * 718 * @param[in] au information about the aggregation 719 */ 720 static void 721 trigger_wire_transfer (struct AggregationUnit *au) 722 { 723 enum GNUNET_DB_QueryStatus qs; 724 725 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 726 "Preparing wire transfer of %s to %s\n", 727 TALER_amount2s (&au->final_amount), 728 TALER_B2S (&au->merchant_pub)); 729 { 730 void *buf; 731 size_t buf_size; 732 733 TALER_BANK_prepare_transfer (au->payto_uri, 734 &au->final_amount, 735 exchange_base_url, 736 &au->wtid, 737 au->extra_wire_subject_metadata, 738 &buf, 739 &buf_size); 740 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 741 "Storing %u bytes of wire prepare data\n", 742 (unsigned int) buf_size); 743 /* Commit our intention to execute the wire transfer! */ 744 qs = TALER_EXCHANGEDB_insert_prewire (pg, 745 au->wa->method, 746 buf, 747 buf_size); 748 GNUNET_log (qs >= 0 749 ? GNUNET_ERROR_TYPE_DEBUG 750 : GNUNET_ERROR_TYPE_WARNING, 751 "wire_prepare_data_insert returned %d\n", 752 (int) qs); 753 GNUNET_free (buf); 754 } 755 /* Commit the WTID data to 'wire_out' */ 756 if (qs >= 0) 757 { 758 qs = TALER_EXCHANGEDB_insert_wire_out ( 759 pg, 760 au->execution_time, 761 &au->wtid, 762 &au->h_full_payto, 763 au->wa->section_name, 764 &au->final_amount, 765 au->extra_wire_subject_metadata); 766 GNUNET_log (qs >= 0 767 ? GNUNET_ERROR_TYPE_DEBUG 768 : GNUNET_ERROR_TYPE_WARNING, 769 "store_wire_transfer_out returned %d\n", 770 (int) qs); 771 } 772 if ( (qs >= 0) && 773 au->have_transient) 774 qs = TALER_EXCHANGEDB_delete_aggregation_transient ( 775 pg, 776 &au->h_full_payto, 777 &au->wtid); 778 779 switch (qs) 780 { 781 case GNUNET_DB_STATUS_SOFT_ERROR: 782 GNUNET_log ( 783 GNUNET_ERROR_TYPE_INFO, 784 "Serialization issue during aggregation; trying again later!\n"); 785 rollback_aggregation (au); 786 return; 787 case GNUNET_DB_STATUS_HARD_ERROR: 788 GNUNET_break (0); 789 fail_aggregation (au); 790 return; 791 default: 792 break; 793 } 794 { 795 struct TALER_EXCHANGEDB_CoinDepositEventP rep = { 796 .header.size = htons (sizeof (rep)), 797 .header.type = htons (TALER_DBEVENT_EXCHANGE_DEPOSIT_STATUS_CHANGED), 798 .merchant_pub = au->merchant_pub 799 }; 800 801 TALER_EXCHANGEDB_event_notify (pg, 802 &rep.header, 803 NULL, 804 0); 805 } 806 commit_aggregation (au); 807 } 808 809 810 /** 811 * A step of #evaluate_rules() failed. If that was a serialization failure 812 * and we still have retries left, roll the transaction back and evaluate the 813 * rules again from the top; the KYC measure, the transient's requirement_row 814 * and the append-only deferral record are written in one transaction and 815 * nothing else ever rescans transients, so dropping them here loses them for 816 * good. 817 * 818 * @param[in,out] au aggregation we are working on 819 * @param qs status that made the caller stop 820 * @return true if a retry was scheduled and the caller must return at once 821 */ 822 static bool 823 retry_rule_evaluation (struct AggregationUnit *au, 824 enum GNUNET_DB_QueryStatus qs) 825 { 826 if (GNUNET_DB_STATUS_SOFT_ERROR != qs) 827 return false; 828 if (au->retries++ >= MAX_RETRIES) 829 return false; 830 TALER_EXCHANGEDB_rollback (pg); 831 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 832 "Serialization failure evaluating KYC rules, trying again!\n"); 833 au->ru = TALER_EXCHANGEDB_begin_rule_update ( 834 pg, 835 &attribute_key, 836 &au->h_normalized_payto, 837 false, /* aggregation does not apply to wallets */ 838 &evaluate_rules, 839 au); 840 return (NULL != au->ru); 841 } 842 843 844 static void 845 evaluate_rules ( 846 struct AggregationUnit *au, 847 struct TALER_EXCHANGEDB_RuleUpdaterResult *rur) 848 { 849 struct TALER_KYCLOGIC_LegitimizationRuleSet *lrs = rur->lrs; 850 enum GNUNET_DB_QueryStatus qs; 851 const struct TALER_KYCLOGIC_KycRule *requirement; 852 853 au->ru = NULL; 854 if (TALER_EC_NONE != rur->ec) 855 { 856 if (NULL != lrs) 857 { 858 /* strange, but whatever */ 859 TALER_KYCLOGIC_rules_free (lrs); 860 } 861 /* Rollback just in case, should have already been done 862 before by the TALER_EXCHANGEDB_begin_rule_update() logic. */ 863 TALER_EXCHANGEDB_rollback (pg); 864 if ( (TALER_EC_GENERIC_DB_SOFT_FAILURE == rur->ec) && 865 (au->retries++ < MAX_RETRIES) ) 866 { 867 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 868 "Serialization failure, trying again!\n"); 869 au->ru = TALER_EXCHANGEDB_begin_rule_update ( 870 pg, 871 &attribute_key, 872 &au->h_normalized_payto, 873 false, /* aggregation does not apply to wallets */ 874 &evaluate_rules, 875 au); 876 if (NULL != au->ru) 877 return; 878 } 879 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 880 "KYC rule evaluation failed hard: %s (%d, %s)\n", 881 TALER_ErrorCode_get_hint (rur->ec), 882 (int) rur->ec, 883 rur->hint); 884 cleanup_and_next (au); 885 return; 886 } 887 888 /* Note that here we are in an open transaction that fetched 889 (or updated) the current set of legitimization rules. So 890 we must properly commit at the end! */ 891 { 892 struct TALER_Amount next_threshold; 893 894 qs = TALER_KYCLOGIC_kyc_test_required ( 895 TALER_KYCLOGIC_KYC_TRIGGER_AGGREGATE, 896 lrs, 897 &return_relevant_amounts, 898 (void *) au, 899 &requirement, 900 &next_threshold); 901 } 902 if (qs < 0) 903 { 904 TALER_KYCLOGIC_rules_free (lrs); 905 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 906 if (retry_rule_evaluation (au, 907 qs)) 908 return; 909 cleanup_and_next (au); 910 return; 911 } 912 if (NULL == requirement) 913 { 914 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 915 "KYC check clear, proceeding with wire transfer\n"); 916 TALER_KYCLOGIC_rules_free (lrs); 917 trigger_wire_transfer (au); 918 return; 919 } 920 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 921 "KYC requirement for %s is %s\n", 922 TALER_amount2s (&au->total_amount), 923 TALER_KYCLOGIC_rule2s (requirement)); 924 { 925 json_t *jrule; 926 927 jrule = TALER_KYCLOGIC_rule_to_measures (requirement); 928 qs = TALER_EXCHANGEDB_do_trigger_kyc_rule_for_account ( 929 pg, 930 au->payto_uri, 931 &au->h_normalized_payto, 932 NULL, 933 &au->merchant_pub, 934 jrule, 935 TALER_KYCLOGIC_rule2priority (requirement), 936 &au->requirement_row, 937 &au->bad_kyc_auth); 938 json_decref (jrule); 939 } 940 if (qs < 0) 941 { 942 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 943 "Failed to persist KYC requirement `%s' in DB!\n", 944 TALER_KYCLOGIC_rule2s (requirement)); 945 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 946 TALER_KYCLOGIC_rules_free (lrs); 947 if (retry_rule_evaluation (au, 948 qs)) 949 return; 950 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 951 global_ret = EXIT_FAILURE; 952 cleanup_and_next (au); 953 return; 954 } 955 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 956 "Legitimization process %llu started\n", 957 (unsigned long long) au->requirement_row); 958 TALER_KYCLOGIC_rules_free (lrs); 959 960 qs = TALER_EXCHANGEDB_update_aggregation_transient (pg, 961 &au->h_full_payto, 962 &au->wtid, 963 au->requirement_row, 964 &au->total_amount); 965 966 967 if (qs < 0) 968 { 969 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 970 "Failed to persist updated transient in in DB!\n"); 971 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 972 if (retry_rule_evaluation (au, 973 qs)) 974 return; 975 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 976 global_ret = EXIT_FAILURE; 977 cleanup_and_next (au); 978 return; 979 } 980 981 /* Say so on the record as well: the transient above is invisible to the 982 auditor (no serial ID, updated in place, deleted on payout), so without 983 this the auditor sees money aggregated and not sent and has to guess 984 whether we are withholding it lawfully. */ 985 qs = TALER_EXCHANGEDB_insert_aggregation_deferral ( 986 pg, 987 &au->wtid, 988 &au->h_full_payto, 989 &au->total_amount, 990 TALER_EXCHANGEDB_DR_KYC, 991 au->requirement_row, 992 au->execution_time); 993 if (qs < 0) 994 { 995 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 996 "Failed to persist deferral reason in DB!\n"); 997 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR == qs); 998 if (retry_rule_evaluation (au, 999 qs)) 1000 return; 1001 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 1002 global_ret = EXIT_FAILURE; 1003 cleanup_and_next (au); 1004 return; 1005 } 1006 1007 { 1008 struct TALER_EXCHANGEDB_CoinDepositEventP rep = { 1009 .header.size = htons (sizeof (rep)), 1010 .header.type = htons (TALER_DBEVENT_EXCHANGE_DEPOSIT_STATUS_CHANGED), 1011 .merchant_pub = au->merchant_pub 1012 }; 1013 1014 TALER_EXCHANGEDB_event_notify (pg, 1015 &rep.header, 1016 NULL, 1017 0); 1018 } 1019 1020 /* First commit, turns the rollback in cleanup into a NOP! */ 1021 qs = commit_or_warn (); 1022 if (0 > qs) 1023 { 1024 if (retry_rule_evaluation (au, 1025 qs)) 1026 return; 1027 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 1028 global_ret = EXIT_FAILURE; 1029 } 1030 cleanup_and_next (au); 1031 } 1032 1033 1034 /** 1035 * The aggregation process could not be concluded and its progress state 1036 * should be remembered in a transient aggregation. 1037 * 1038 * @param[in] au aggregation that needs to be committed 1039 * into a transient aggregation 1040 */ 1041 static void 1042 commit_to_transient (struct AggregationUnit *au) 1043 { 1044 enum GNUNET_DB_QueryStatus qs; 1045 1046 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1047 "Not ready for wire transfer (%s)\n", 1048 TALER_amount2s (&au->final_amount)); 1049 if (au->have_transient) 1050 qs = TALER_EXCHANGEDB_update_aggregation_transient (pg, 1051 &au->h_full_payto, 1052 &au->wtid, 1053 au->requirement_row, 1054 &au->total_amount); 1055 else 1056 qs = TALER_EXCHANGEDB_insert_aggregation_transient (pg, 1057 &au->h_full_payto, 1058 au->wa->section_name, 1059 &au->merchant_pub, 1060 &au->wtid, 1061 au->requirement_row, 1062 &au->total_amount); 1063 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 1064 { 1065 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1066 "Serialization issue, trying again later!\n"); 1067 rollback_aggregation (au); 1068 return; 1069 } 1070 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 1071 { 1072 GNUNET_break (0); 1073 fail_aggregation (au); 1074 return; 1075 } 1076 if (! au->legi_check) 1077 { 1078 /* Nothing is pending on our side: we have decided that the aggregate is 1079 not worth its wire fee yet, and that is the whole reason the money is 1080 still here. Put that on the record for the auditor. 1081 1082 When @e legi_check is set we are instead merely parking the aggregate 1083 so that the KYC rules can be evaluated in a transaction of their own, 1084 and we do not yet know whether we will be paying out or withholding -- 1085 evaluate_rules() records the deferral in the latter case. */ 1086 qs = TALER_EXCHANGEDB_insert_aggregation_deferral ( 1087 pg, 1088 &au->wtid, 1089 &au->h_full_payto, 1090 &au->total_amount, 1091 TALER_EXCHANGEDB_DR_AMOUNT_TOO_SMALL, 1092 0, 1093 au->execution_time); 1094 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 1095 { 1096 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1097 "Serialization issue, trying again later!\n"); 1098 rollback_aggregation (au); 1099 return; 1100 } 1101 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 1102 { 1103 GNUNET_break (0); 1104 fail_aggregation (au); 1105 return; 1106 } 1107 } 1108 au->have_transient = true; 1109 /* commit */ 1110 commit_aggregation (au); 1111 } 1112 1113 1114 /** 1115 * Test if legitimization rules are satisfied for a transfer to @a h_payto. 1116 * 1117 * @param[in] au aggregation unit to check for 1118 */ 1119 static void 1120 check_legitimization_satisfied (struct AggregationUnit *au) 1121 { 1122 if (kyc_off) 1123 { 1124 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1125 "KYC checks are off, legitimization satisfied\n"); 1126 trigger_wire_transfer (au); 1127 return; 1128 } 1129 /* get legi rules *after* committing, as the legi check 1130 should run in a separate transaction! */ 1131 au->legi_check = true; 1132 commit_to_transient (au); 1133 } 1134 1135 1136 /** 1137 * Perform the main aggregation work for @a au. Expects to be in 1138 * a working transaction, which the caller must also ultimately commit 1139 * (or rollback) depending on our return value. 1140 * 1141 * @param[in,out] au aggregation unit to work on 1142 */ 1143 static void 1144 do_aggregate (struct AggregationUnit *au) 1145 { 1146 enum GNUNET_DB_QueryStatus qs; 1147 1148 au->wa = TALER_EXCHANGEDB_find_account_by_payto_uri ( 1149 au->payto_uri); 1150 if (NULL == au->wa) 1151 { 1152 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1153 "No exchange account configured for `%s', please fix your setup to continue!\n", 1154 au->payto_uri.full_payto); 1155 global_ret = EXIT_FAILURE; 1156 fail_aggregation (au); 1157 return; 1158 } 1159 1160 { 1161 struct GNUNET_TIME_Timestamp start_date; 1162 struct GNUNET_TIME_Timestamp end_date; 1163 struct TALER_MasterSignatureP master_sig; 1164 uint64_t rowid; 1165 1166 qs = TALER_EXCHANGEDB_get_wire_fee (pg, 1167 au->wa->method, 1168 au->execution_time, 1169 &rowid, 1170 &start_date, 1171 &end_date, 1172 &au->fees, 1173 &master_sig); 1174 switch (qs) 1175 { 1176 case GNUNET_DB_STATUS_HARD_ERROR: 1177 GNUNET_break (0); 1178 fail_aggregation (au); 1179 return; 1180 case GNUNET_DB_STATUS_SOFT_ERROR: 1181 /* serializiability issue, try again */ 1182 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 1183 "Serialization issue, trying again later!\n"); 1184 rollback_aggregation (au); 1185 return; 1186 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 1187 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1188 "Could not get wire fees for %s at %s. Aborting run.\n", 1189 au->wa->method, 1190 GNUNET_TIME_timestamp2s (au->execution_time)); 1191 fail_aggregation (au); 1192 return; 1193 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 1194 /* continued below */ 1195 break; 1196 } 1197 } 1198 1199 /* Now try to find other deposits to aggregate */ 1200 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 1201 "Found ready deposit for %s, aggregating by target %s\n", 1202 TALER_B2S (&au->merchant_pub), 1203 au->payto_uri.full_payto); 1204 qs = TALER_EXCHANGEDB_get_aggregation_transient (pg, 1205 &au->h_full_payto, 1206 &au->merchant_pub, 1207 au->wa->section_name, 1208 &au->wtid, 1209 &au->trans); 1210 switch (qs) 1211 { 1212 case GNUNET_DB_STATUS_HARD_ERROR: 1213 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1214 "Failed to lookup transient aggregates!\n"); 1215 fail_aggregation (au); 1216 return; 1217 case GNUNET_DB_STATUS_SOFT_ERROR: 1218 /* serializiability issue, try again */ 1219 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 1220 "Serialization issue, trying again later!\n"); 1221 rollback_aggregation (au); 1222 return; 1223 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 1224 GNUNET_CRYPTO_random_block (&au->wtid, 1225 sizeof (au->wtid)); 1226 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1227 "No transient aggregation found, starting %s\n", 1228 TALER_B2S (&au->wtid)); 1229 au->have_transient = false; 1230 break; 1231 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 1232 au->have_transient = true; 1233 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1234 "Transient aggregation found, resuming %s\n", 1235 TALER_B2S (&au->wtid)); 1236 break; 1237 } 1238 qs = TALER_EXCHANGEDB_do_aggregate (pg, 1239 &au->h_full_payto, 1240 &au->merchant_pub, 1241 &au->wtid, 1242 &au->total_amount); 1243 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 1244 { 1245 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1246 "Failed to execute aggregation!\n"); 1247 fail_aggregation (au); 1248 return; 1249 } 1250 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 1251 { 1252 /* serializiability issue, try again */ 1253 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 1254 "Serialization issue, trying again later!\n"); 1255 rollback_aggregation (au); 1256 return; 1257 } 1258 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1259 "Aggregation total is %s.\n", 1260 TALER_amount2s (&au->total_amount)); 1261 /* Subtract wire transfer fee and round to the unit supported by the 1262 wire transfer method; Check if after rounding down, we still have 1263 an amount to transfer, and if not mark as 'tiny'. */ 1264 if (au->have_transient) 1265 GNUNET_assert (0 <= 1266 TALER_amount_add (&au->total_amount, 1267 &au->total_amount, 1268 &au->trans)); 1269 1270 1271 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 1272 "Rounding aggregate of %s\n", 1273 TALER_amount2s (&au->total_amount)); 1274 if ( (0 >= 1275 TALER_amount_subtract (&au->final_amount, 1276 &au->total_amount, 1277 &au->fees.wire)) || 1278 (GNUNET_SYSERR == 1279 TALER_amount_round_down (&au->final_amount, 1280 ¤cy_round_unit)) || 1281 (TALER_amount_is_zero (&au->final_amount)) ) 1282 { 1283 commit_to_transient (au); 1284 return; 1285 } 1286 check_legitimization_satisfied (au); 1287 } 1288 1289 1290 static void 1291 run_aggregation (void *cls) 1292 { 1293 struct Shard *s = cls; 1294 struct AggregationUnit *au; 1295 enum GNUNET_DB_QueryStatus qs; 1296 1297 task = NULL; 1298 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1299 "Checking for ready deposits to aggregate\n"); 1300 /* make sure we have current fees */ 1301 au = GNUNET_new (struct AggregationUnit); 1302 au->execution_time = GNUNET_TIME_timestamp_get (); 1303 au->shard = s; 1304 if (GNUNET_OK != 1305 TALER_EXCHANGEDB_start_deferred_wire_out (pg)) 1306 { 1307 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1308 "Failed to start database transaction!\n"); 1309 global_ret = EXIT_FAILURE; 1310 GNUNET_SCHEDULER_shutdown (); 1311 release_shard (s); 1312 return; 1313 } 1314 qs = TALER_EXCHANGEDB_get_ready_deposit ( 1315 pg, 1316 s->shard_start, 1317 s->shard_end, 1318 &au->merchant_pub, 1319 &au->payto_uri, 1320 &au->extra_wire_subject_metadata); 1321 switch (qs) 1322 { 1323 case GNUNET_DB_STATUS_HARD_ERROR: 1324 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1325 "Failed to begin deposit iteration!\n"); 1326 global_ret = EXIT_FAILURE; 1327 GNUNET_SCHEDULER_shutdown (); 1328 cleanup_and_next (au); 1329 return; 1330 case GNUNET_DB_STATUS_SOFT_ERROR: 1331 cleanup_au (au); 1332 TALER_EXCHANGEDB_rollback (pg); 1333 run_task_with_shard (s); 1334 return; 1335 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 1336 { 1337 struct GNUNET_TIME_Relative duration 1338 = GNUNET_TIME_absolute_get_duration (s->start_time.abs_time); 1339 1340 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1341 "Completed shard [%u,%u] after %s with %llu deposits\n", 1342 (unsigned int) s->shard_start, 1343 (unsigned int) s->shard_end, 1344 GNUNET_TIME_relative2s (duration, 1345 true), 1346 (unsigned long long) s->work_counter); 1347 cleanup_and_next (au); 1348 return; 1349 } 1350 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 1351 s->work_counter++; 1352 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1353 "Found ready deposit!\n"); 1354 /* continued below */ 1355 break; 1356 } 1357 1358 TALER_full_payto_hash (au->payto_uri, 1359 &au->h_full_payto); 1360 TALER_full_payto_normalize_and_hash (au->payto_uri, 1361 &au->h_normalized_payto); 1362 GNUNET_break (! TALER_payto_is_wallet (au->payto_uri.full_payto)); 1363 do_aggregate (au); 1364 } 1365 1366 1367 /** 1368 * Select a shard to work on. 1369 * 1370 * @param cls NULL 1371 */ 1372 static void 1373 run_shard (void *cls) 1374 { 1375 struct Shard *s; 1376 enum GNUNET_DB_QueryStatus qs; 1377 1378 (void) cls; 1379 task = NULL; 1380 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1381 "Running aggregation shard\n"); 1382 if (GNUNET_SYSERR == 1383 TALER_EXCHANGEDB_preflight (pg)) 1384 { 1385 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1386 "Failed to obtain database connection!\n"); 1387 global_ret = EXIT_FAILURE; 1388 GNUNET_SCHEDULER_shutdown (); 1389 return; 1390 } 1391 s = GNUNET_new (struct Shard); 1392 s->start_time = GNUNET_TIME_timestamp_get (); 1393 qs = TALER_EXCHANGEDB_begin_revolving_shard (pg, 1394 "aggregator", 1395 shard_size, 1396 1U + INT32_MAX, 1397 &s->shard_start, 1398 &s->shard_end); 1399 switch (qs) 1400 { 1401 case GNUNET_DB_STATUS_SOFT_ERROR: 1402 { 1403 static struct GNUNET_TIME_Relative delay; 1404 1405 GNUNET_free (s); 1406 delay = GNUNET_TIME_randomized_backoff (delay, 1407 GNUNET_TIME_UNIT_SECONDS); 1408 GNUNET_assert (NULL == task); 1409 task = GNUNET_SCHEDULER_add_delayed (delay, 1410 &run_shard, 1411 NULL); 1412 return; 1413 } 1414 case GNUNET_DB_STATUS_HARD_ERROR: 1415 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1416 "Failed to begin shard (%d)!\n", 1417 qs); 1418 global_ret = EXIT_FAILURE; 1419 GNUNET_SCHEDULER_shutdown (); 1420 GNUNET_free (s); 1421 return; 1422 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 1423 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1424 "No open shard available\n"); 1425 GNUNET_SCHEDULER_shutdown (); 1426 GNUNET_free (s); 1427 return; 1428 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 1429 break; 1430 } 1431 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1432 "Starting shard [%u:%u]!\n", 1433 (unsigned int) s->shard_start, 1434 (unsigned int) s->shard_end); 1435 GNUNET_assert (NULL == task); 1436 task = GNUNET_SCHEDULER_add_now (&run_aggregation, 1437 s); 1438 } 1439 1440 1441 static void 1442 drain_kyc_alerts (void *cls) 1443 { 1444 enum GNUNET_DB_QueryStatus qs; 1445 struct AggregationUnit *au; 1446 1447 (void) cls; 1448 task = NULL; 1449 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1450 "Draining KYC alerts\n"); 1451 au = GNUNET_new (struct AggregationUnit); 1452 au->execution_time = GNUNET_TIME_timestamp_get (); 1453 if (GNUNET_SYSERR == 1454 TALER_EXCHANGEDB_preflight (pg)) 1455 { 1456 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1457 "Failed to obtain database connection!\n"); 1458 GNUNET_free (au); 1459 global_ret = EXIT_FAILURE; 1460 GNUNET_SCHEDULER_shutdown (); 1461 return; 1462 } 1463 if (GNUNET_OK != 1464 TALER_EXCHANGEDB_start (pg, 1465 "handle kyc alerts")) 1466 { 1467 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1468 "Failed to start database transaction!\n"); 1469 GNUNET_free (au); 1470 global_ret = EXIT_FAILURE; 1471 GNUNET_SCHEDULER_shutdown (); 1472 return; 1473 } 1474 while (1) 1475 { 1476 qs = TALER_EXCHANGEDB_do_drain_kyc_alert (pg, 1477 1, 1478 &au->h_normalized_payto); 1479 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 1480 "Found %d KYC alerts\n", 1481 (int) qs); 1482 switch (qs) 1483 { 1484 case GNUNET_DB_STATUS_HARD_ERROR: 1485 GNUNET_break (0); 1486 TALER_EXCHANGEDB_rollback (pg); 1487 GNUNET_free (au); 1488 GNUNET_assert (NULL == task); 1489 global_ret = EXIT_FAILURE; 1490 GNUNET_SCHEDULER_shutdown (); 1491 return; 1492 case GNUNET_DB_STATUS_SOFT_ERROR: 1493 TALER_EXCHANGEDB_rollback (pg); 1494 GNUNET_assert (NULL == task); 1495 GNUNET_free (au); 1496 task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts, 1497 NULL); 1498 return; 1499 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 1500 GNUNET_free (au); 1501 TALER_EXCHANGEDB_rollback (pg); 1502 GNUNET_assert (NULL == task); 1503 task = GNUNET_SCHEDULER_add_now (&run_shard, 1504 NULL); 1505 return; 1506 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 1507 /* handled below */ 1508 break; 1509 } 1510 qs = TALER_EXCHANGEDB_get_aggregation_transient_by_normalized_payto ( 1511 pg, 1512 &au->h_normalized_payto, 1513 &au->payto_uri, 1514 &au->wtid, 1515 &au->merchant_pub, 1516 &au->trans); 1517 switch (qs) 1518 { 1519 case GNUNET_DB_STATUS_HARD_ERROR: 1520 GNUNET_break (0); 1521 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1522 "Failed to lookup transient aggregates!\n"); 1523 TALER_EXCHANGEDB_rollback (pg); 1524 GNUNET_free (au); 1525 global_ret = EXIT_FAILURE; 1526 GNUNET_SCHEDULER_shutdown (); 1527 return; 1528 case GNUNET_DB_STATUS_SOFT_ERROR: 1529 /* serializiability issue, try again */ 1530 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 1531 "Serialization issue, trying again later!\n"); 1532 TALER_EXCHANGEDB_rollback (pg); 1533 GNUNET_assert (NULL == task); 1534 task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts, 1535 NULL); 1536 return; 1537 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 1538 continue; /* while (1) */ 1539 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 1540 TALER_full_payto_hash (au->payto_uri, 1541 &au->h_full_payto); 1542 au->have_transient = true; 1543 do_aggregate (au); 1544 return; 1545 } 1546 GNUNET_assert (0); 1547 } /* while(1) */ 1548 } 1549 1550 1551 /** 1552 * First task. 1553 * 1554 * @param cls closure, NULL 1555 * @param args remaining command-line arguments 1556 * @param cfgfile name of the configuration file used (for saving, can be NULL!) 1557 * @param c configuration 1558 */ 1559 static void 1560 run (void *cls, 1561 char *const *args, 1562 const char *cfgfile, 1563 const struct GNUNET_CONFIGURATION_Handle *c) 1564 { 1565 unsigned long long ass; 1566 (void) cls; 1567 (void) args; 1568 (void) cfgfile; 1569 1570 cfg = c; 1571 if (GNUNET_OK != 1572 parse_aggregator_config ()) 1573 { 1574 cfg = NULL; 1575 global_ret = EXIT_NOTCONFIGURED; 1576 return; 1577 } 1578 if (GNUNET_OK != 1579 GNUNET_CONFIGURATION_get_value_number (cfg, 1580 "exchange", 1581 "AGGREGATOR_SHARD_SIZE", 1582 &ass)) 1583 { 1584 cfg = NULL; 1585 global_ret = EXIT_NOTCONFIGURED; 1586 return; 1587 } 1588 if ( (0 == ass) || 1589 (ass > INT32_MAX) ) 1590 shard_size = 1U + INT32_MAX; 1591 else 1592 shard_size = (uint32_t) ass; 1593 if (GNUNET_OK != 1594 TALER_KYCLOGIC_kyc_init (cfg, 1595 cfgfile)) 1596 { 1597 cfg = NULL; 1598 global_ret = EXIT_NOTCONFIGURED; 1599 return; 1600 } 1601 GNUNET_SCHEDULER_add_shutdown (&shutdown_task, 1602 NULL); 1603 GNUNET_assert (NULL == task); 1604 task = GNUNET_SCHEDULER_add_now (&drain_kyc_alerts, 1605 NULL); 1606 } 1607 1608 1609 /** 1610 * The main function of the taler-exchange-aggregator. 1611 * 1612 * @param argc number of arguments from the command line 1613 * @param argv command line arguments 1614 * @return 0 ok, non-zero on error, see #global_ret 1615 */ 1616 int 1617 main (int argc, 1618 char *const *argv) 1619 { 1620 struct GNUNET_GETOPT_CommandLineOption options[] = { 1621 GNUNET_GETOPT_option_timetravel ('T', 1622 "timetravel"), 1623 GNUNET_GETOPT_option_flag ('t', 1624 "test", 1625 "run in test mode and exit when idle", 1626 &test_mode), 1627 GNUNET_GETOPT_option_flag ('y', 1628 "kyc-off", 1629 "perform wire transfers without KYC checks", 1630 &kyc_off), 1631 GNUNET_GETOPT_OPTION_END 1632 }; 1633 enum GNUNET_GenericReturnValue ret; 1634 1635 ret = GNUNET_PROGRAM_run ( 1636 TALER_EXCHANGE_project_data (), 1637 argc, argv, 1638 "taler-exchange-aggregator", 1639 gettext_noop ( 1640 "background process that aggregates and executes wire transfers"), 1641 options, 1642 &run, NULL); 1643 if (GNUNET_SYSERR == ret) 1644 return EXIT_INVALIDARGUMENT; 1645 if (GNUNET_NO == ret) 1646 return EXIT_SUCCESS; 1647 return global_ret; 1648 } 1649 1650 1651 /* end of taler-exchange-aggregator.c */