taler-auditor-sync.c (35425B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2020-2022 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU 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 General Public License for more details. 12 13 You should have received a copy of the GNU General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 /** 17 * @file taler-auditor-sync.c 18 * @brief Tool used by the auditor to make a 'safe' copy of the exchanges' database. 19 * @author Christian Grothoff 20 */ 21 #include "platform.h" 22 #include "exchangedb_lib.h" 23 #include "exchange-database/preflight.h" 24 #include "exchange-database/get_exists_aml_officer.h" 25 #include "exchange-database/abort_shard.h" 26 #include "exchange-database/insert_signkey.h" 27 #include "exchange-database/do_aggregate.h" 28 #include "exchange-database/begin_revolving_shard.h" 29 #include "exchange-database/begin_shard.h" 30 #include "exchange-database/update_to_aml_unlocked.h" 31 #include "exchange-database/commit.h" 32 #include "exchange-database/update_shard_progress.h" 33 #include "exchange-database/compute_shard.h" 34 #include "exchange-database/get_count_known_coins.h" 35 #include "exchange-database/insert_aggregation_transient.h" 36 #include "exchange-database/create_tables.h" 37 #include "exchange-database/delete_aggregation_transient.h" 38 #include "exchange-database/delete_shard_locks.h" 39 #include "exchange-database/disable_rules.h" 40 #include "exchange-database/do_check_deposit_idempotent.h" 41 #include "exchange-database/do_deposit.h" 42 #include "exchange-database/do_purse_delete.h" 43 #include "exchange-database/do_purse_deposit.h" 44 #include "exchange-database/do_purse_merge.h" 45 #include "exchange-database/do_recoup.h" 46 #include "exchange-database/do_recoup_refresh.h" 47 #include "exchange-database/do_refresh.h" 48 #include "exchange-database/do_refund.h" 49 #include "exchange-database/do_reserve_open.h" 50 #include "exchange-database/do_reserve_purse.h" 51 #include "exchange-database/do_withdraw.h" 52 #include "exchange-database/do_drain_kyc_alert.h" 53 #include "exchange-database/drop_tables.h" 54 #include "exchange-database/enable_rules.h" 55 #include "exchange-database/do_insert_known_coin.h" 56 #include "exchange-database/event_listen_cancel.h" 57 #include "exchange-database/event_listen.h" 58 #include "exchange-database/do_expire_purse.h" 59 #include "exchange-database/get_aggregation_transient_by_normalized_payto.h" 60 #include "exchange-database/gc.h" 61 #include "exchange-database/get_coin_denomination.h" 62 #include "exchange-database/get_coin_transactions.h" 63 #include "exchange-database/get_denomination_by_serial.h" 64 #include "exchange-database/get_denomination_info.h" 65 #include "exchange-database/get_denomination_revocation.h" 66 #include "exchange-database/get_profit_drain.h" 67 #include "exchange-database/iterate_expired_reserves.h" 68 #include "exchange-database/get_global_fee.h" 69 #include "exchange-database/iterate_global_fees.h" 70 #include "exchange-database/get_known_coin.h" 71 #include "exchange-database/get_kyc_rules.h" 72 #include "exchange-database/get_old_coin_by_h_blind.h" 73 #include "exchange-database/get_pending_legitimization_process.h" 74 #include "exchange-database/get_purse_deposit.h" 75 #include "exchange-database/get_purse_request.h" 76 #include "exchange-database/get_ready_deposit.h" 77 #include "exchange-database/get_refresh.h" 78 #include "exchange-database/get_reserve_balance.h" 79 #include "exchange-database/get_reserve_by_h_planchets.h" 80 #include "exchange-database/get_reserve_history.h" 81 #include "exchange-database/get_signature_for_known_coin.h" 82 #include "exchange-database/iterate_unfinished_close_requests.h" 83 #include "exchange-database/iterate_wire_accounts.h" 84 #include "exchange-database/get_wire_fee.h" 85 #include "exchange-database/iterate_wire_fees.h" 86 #include "exchange-database/get_wire_hash_for_contract.h" 87 #include "exchange-database/get_withdraw.h" 88 #include "exchange-database/get_exists_deposit.h" 89 #include "exchange-database/inject_auditor_triggers.h" 90 #include "exchange-database/insert_active_legitimization_measure.h" 91 #include "exchange-database/insert_aml_decision.h" 92 #include "exchange-database/insert_aml_officer.h" 93 #include "exchange-database/insert_aml_program_failure.h" 94 #include "exchange-database/insert_auditor_denom_sig.h" 95 #include "exchange-database/insert_auditor.h" 96 #include "exchange-database/insert_close_request.h" 97 #include "exchange-database/insert_contract.h" 98 #include "exchange-database/insert_denomination_info.h" 99 #include "exchange-database/insert_denomination_revocation.h" 100 #include "exchange-database/insert_profit_drain.h" 101 #include "exchange-database/insert_global_fee.h" 102 #include "exchange-database/insert_kyc_failure.h" 103 #include "exchange-database/insert_legitimization_process.h" 104 #include "exchange-database/insert_partner.h" 105 #include "exchange-database/insert_purse_request.h" 106 struct InsertContext; 107 #define TALER_EXCHANGEDB_REPLICATION_RESULT_CLOSURE struct InsertContext 108 #include "exchange-database/insert_records_by_table.h" 109 #include "exchange-database/insert_reserve_closed.h" 110 #include "exchange-database/insert_reserve_open_deposit.h" 111 #include "exchange-database/insert_sanction_list_hit.h" 112 #include "exchange-database/insert_signkey_revocation.h" 113 #include "exchange-database/insert_successor_measure.h" 114 #include "exchange-database/insert_wire_fee.h" 115 #include "exchange-database/insert_wire.h" 116 #include "exchange-database/iterate_active_auditors.h" 117 #include "exchange-database/iterate_active_signkeys.h" 118 #include "exchange-database/iterate_auditor_denominations.h" 119 #include "exchange-database/iterate_denomination_info.h" 120 #include "exchange-database/iterate_denominations.h" 121 #include "exchange-database/iterate_kyc_references.h" 122 #include "exchange-database/iterate_reserve_close_info.h" 123 #include "exchange-database/get_kyc_provider_account.h" 124 #include "exchange-database/get_active_legitimization.h" 125 #include "exchange-database/get_aml_file_number.h" 126 #include "exchange-database/iterate_aml_history.h" 127 #include "exchange-database/get_aml_officer.h" 128 #include "exchange-database/get_auditor_status.h" 129 #include "exchange-database/get_auditor_timestamp.h" 130 #include "exchange-database/get_completed_legitimization.h" 131 #include "exchange-database/get_denomination_meta.h" 132 #include "exchange-database/get_global_fee_by_time.h" 133 #include "exchange-database/get_h_payto_by_access_token.h" 134 #include "exchange-database/iterate_kyc_history.h" 135 #include "exchange-database/get_legitimization_process_by_account.h" 136 #include "exchange-database/get_legitimization_requirement_by_row.h" 137 #include "exchange-database/get_kyc_status_by_token.h" 138 #include "exchange-database/get_pending_legitimization.h" 139 #include "exchange-database/iterate_records_by_table.h" 140 #include "exchange-database/get_rules_by_access_token.h" 141 #include "exchange-database/get_serial_by_table.h" 142 #include "exchange-database/get_signkey.h" 143 #include "exchange-database/get_signkey_revocation.h" 144 #include "exchange-database/get_transfer_by_deposit.h" 145 #include "exchange-database/get_wire_fee_by_time.h" 146 #include "exchange-database/get_wire_timestamp.h" 147 #include "exchange-database/iterate_wire_transfers.h" 148 #include "exchange-database/update_to_refresh_revealed.h" 149 #include "exchange-database/do_insert_kyc_attributes.h" 150 #include "exchange-database/preflight.h" 151 #include "exchange-database/get_pending_profit_drain.h" 152 #include "exchange-database/update_to_profit_drain_finished.h" 153 #include "exchange-database/release_revolving_shard.h" 154 #include "exchange-database/get_reserve.h" 155 #include "exchange-database/get_reserve_origin.h" 156 #include "exchange-database/do_import_credits.h" 157 #include "exchange-database/update_reserve.h" 158 #include "exchange-database/rollback.h" 159 #include "exchange-database/iterate_account_merges_above_serial_id.h" 160 #include "exchange-database/iterate_aggregation_amounts_for_kyc_check.h" 161 #include "exchange-database/iterate_aggregations_above_serial_id.h" 162 #include "exchange-database/get_aggregation_transient.h" 163 #include "exchange-database/iterate_all_kyc_attributes.h" 164 #include "exchange-database/iterate_all_purse_decisions_above_serial_id.h" 165 #include "exchange-database/iterate_all_purse_deletions_above_serial_id.h" 166 #include "exchange-database/iterate_aml_attributes.h" 167 #include "exchange-database/iterate_aml_decisions.h" 168 #include "exchange-database/iterate_aml_measures.h" 169 #include "exchange-database/iterate_aml_statistics.h" 170 #include "exchange-database/get_auditor_denom_sig.h" 171 #include "exchange-database/iterate_batch_deposits_missing_wire.h" 172 #include "exchange-database/iterate_coin_deposits_above_serial_id.h" 173 #include "exchange-database/get_contract_by_purse.h" 174 #include "exchange-database/get_contract.h" 175 #include "exchange-database/iterate_deposit_amounts_for_kyc_check.h" 176 #include "exchange-database/iterate_exchange_credit_transfers.h" 177 #include "exchange-database/iterate_exchange_debit_transfers.h" 178 #include "exchange-database/iterate_exchange_kycauth_transfers.h" 179 #include "exchange-database/iterate_kyc_accounts.h" 180 #include "exchange-database/iterate_kyc_attributes.h" 181 #include "exchange-database/iterate_merge_amounts_for_kyc_check.h" 182 #include "exchange-database/get_purse_by_merge_pub.h" 183 #include "exchange-database/iterate_purse_decisions_above_serial_id.h" 184 #include "exchange-database/iterate_purse_deposits_above_serial_id.h" 185 #include "exchange-database/iterate_purse_deposits_by_purse.h" 186 #include "exchange-database/get_purse.h" 187 #include "exchange-database/get_purse_merge.h" 188 #include "exchange-database/iterate_purse_merges_above_serial_id.h" 189 #include "exchange-database/iterate_purse_requests_above_serial_id.h" 190 #include "exchange-database/iterate_recoups_above_serial_id.h" 191 #include "exchange-database/iterate_recoup_refreshes_above_serial_id.h" 192 #include "exchange-database/iterate_refreshes_above_serial_id.h" 193 #include "exchange-database/iterate_refunds_above_serial_id.h" 194 #include "exchange-database/iterate_refunds_by_coin.h" 195 #include "exchange-database/iterate_reserve_closed_above_serial_id.h" 196 #include "exchange-database/get_reserve_close_info.h" 197 #include "exchange-database/iterate_reserve_open_above_serial_id.h" 198 #include "exchange-database/iterate_reserves_in_above_serial_id.h" 199 #include "exchange-database/iterate_wire_outs_above_serial_id_by_account.h" 200 #include "exchange-database/iterate_wire_outs_above_serial_id.h" 201 #include "exchange-database/iterate_withdrawals_above_serial_id.h" 202 #include "exchange-database/iterate_withdraw_amounts_for_kyc_check.h" 203 #include "exchange-database/update_to_aml_locked.h" 204 #include "exchange-database/update_purse_balance.h" 205 #include "exchange-database/start_deferred_wire_out.h" 206 #include "exchange-database/start.h" 207 #include "exchange-database/start_read_committed.h" 208 #include "exchange-database/start_read_only.h" 209 #include "exchange-database/insert_wire_out.h" 210 #include "exchange-database/get_exists_aml_officer.h" 211 #include "exchange-database/do_trigger_kyc_rule_for_account.h" 212 #include "exchange-database/update_aggregation_transient.h" 213 #include "exchange-database/update_auditor.h" 214 #include "exchange-database/update_legitimization_process_by_row.h" 215 #include "exchange-database/update_wire.h" 216 #include "exchange-database/iterate_prewires.h" 217 #include "exchange-database/insert_prewire.h" 218 #include "exchange-database/update_to_prewire_failed.h" 219 #include "exchange-database/update_to_prewire_finished.h" 220 221 222 /** 223 * Handle to access the exchange's source database. 224 */ 225 static struct TALER_EXCHANGEDB_PostgresContext *src; 226 227 /** 228 * Handle to access the exchange's destination database. 229 */ 230 static struct TALER_EXCHANGEDB_PostgresContext *dst; 231 232 /** 233 * Return value from #main(). 234 */ 235 static int global_ret; 236 237 /** 238 * Main task to do synchronization. 239 */ 240 static struct GNUNET_SCHEDULER_Task *sync_task; 241 242 /** 243 * What is our target transaction size (number of records)? 244 */ 245 static unsigned int transaction_size = 512; 246 247 /** 248 * Number of records copied in this transaction. 249 */ 250 static unsigned long long actual_size; 251 252 /** 253 * Terminate once synchronization is achieved. 254 */ 255 static int exit_if_synced; 256 257 258 /** 259 * Highest value in `enum TALER_EXCHANGEDB_ReplicatedTable`. Update 260 * together with #rt_name() and #tables when the exchange gains another 261 * replicated table. 262 */ 263 #define RT_MAX TALER_EXCHANGEDB_RT_KYCAUTHS_IN 264 265 266 /** 267 * Return the name of the database table @a rt lives in. 268 * 269 * The switch is exhaustive and deliberately has no @e default case: when 270 * a table is added to `enum TALER_EXCHANGEDB_ReplicatedTable`, the 271 * compiler points at this function (-Wswitch), which is the reminder 272 * that #RT_MAX and #tables below must be extended as well. #check_tables() 273 * catches it at runtime if they were not. 274 * 275 * @param rt table to name 276 * @return name of the table in the database 277 */ 278 static const char * 279 rt_name (enum TALER_EXCHANGEDB_ReplicatedTable rt) 280 { 281 switch (rt) 282 { 283 case TALER_EXCHANGEDB_RT_DENOMINATIONS: 284 return "denominations"; 285 case TALER_EXCHANGEDB_RT_DENOMINATION_REVOCATIONS: 286 return "denomination_revocations"; 287 case TALER_EXCHANGEDB_RT_KYC_TARGETS: 288 return "kyc_targets"; 289 case TALER_EXCHANGEDB_RT_WIRE_TARGETS: 290 return "wire_targets"; 291 case TALER_EXCHANGEDB_RT_RESERVES: 292 return "reserves"; 293 case TALER_EXCHANGEDB_RT_RESERVES_IN: 294 return "reserves_in"; 295 case TALER_EXCHANGEDB_RT_RESERVES_CLOSE: 296 return "reserves_close"; 297 case TALER_EXCHANGEDB_RT_RESERVES_OPEN_REQUESTS: 298 return "reserves_open_requests"; 299 case TALER_EXCHANGEDB_RT_RESERVES_OPEN_DEPOSITS: 300 return "reserves_open_deposits"; 301 case TALER_EXCHANGEDB_RT_AUDITORS: 302 return "auditors"; 303 case TALER_EXCHANGEDB_RT_AUDITOR_DENOM_SIGS: 304 return "auditor_denom_sigs"; 305 case TALER_EXCHANGEDB_RT_EXCHANGE_SIGN_KEYS: 306 return "exchange_sign_keys"; 307 case TALER_EXCHANGEDB_RT_SIGNKEY_REVOCATIONS: 308 return "signkey_revocations"; 309 case TALER_EXCHANGEDB_RT_KNOWN_COINS: 310 return "known_coins"; 311 case TALER_EXCHANGEDB_RT_REFRESH: 312 return "refresh"; 313 case TALER_EXCHANGEDB_RT_BATCH_DEPOSITS: 314 return "batch_deposits"; 315 case TALER_EXCHANGEDB_RT_COIN_DEPOSITS: 316 return "coin_deposits"; 317 case TALER_EXCHANGEDB_RT_REFUNDS: 318 return "refunds"; 319 case TALER_EXCHANGEDB_RT_WIRE_OUT: 320 return "wire_out"; 321 case TALER_EXCHANGEDB_RT_AGGREGATION_TRACKING: 322 return "aggregation_tracking"; 323 case TALER_EXCHANGEDB_RT_AGGREGATION_DEFERRALS: 324 return "aggregation_deferrals"; 325 case TALER_EXCHANGEDB_RT_WIRE_FEE: 326 return "wire_fee"; 327 case TALER_EXCHANGEDB_RT_GLOBAL_FEE: 328 return "global_fee"; 329 case TALER_EXCHANGEDB_RT_RECOUP: 330 return "recoup"; 331 case TALER_EXCHANGEDB_RT_RECOUP_REFRESH: 332 return "recoup_refresh"; 333 case TALER_EXCHANGEDB_RT_PURSE_REQUESTS: 334 return "purse_requests"; 335 case TALER_EXCHANGEDB_RT_PURSE_DECISION: 336 return "purse_decision"; 337 case TALER_EXCHANGEDB_RT_PURSE_MERGES: 338 return "purse_merges"; 339 case TALER_EXCHANGEDB_RT_PURSE_DEPOSITS: 340 return "purse_deposits"; 341 case TALER_EXCHANGEDB_RT_ACCOUNT_MERGES: 342 return "account_merges"; 343 case TALER_EXCHANGEDB_RT_HISTORY_REQUESTS: 344 return "history_requests"; 345 case TALER_EXCHANGEDB_RT_CLOSE_REQUESTS: 346 return "close_requests"; 347 case TALER_EXCHANGEDB_RT_WADS_OUT: 348 return "wads_out"; 349 case TALER_EXCHANGEDB_RT_WADS_OUT_ENTRIES: 350 return "wad_out_entries"; 351 case TALER_EXCHANGEDB_RT_WADS_IN: 352 return "wads_in"; 353 case TALER_EXCHANGEDB_RT_WADS_IN_ENTRIES: 354 return "wad_in_entries"; 355 case TALER_EXCHANGEDB_RT_PROFIT_DRAINS: 356 return "profit_drains"; 357 case TALER_EXCHANGEDB_RT_AML_STAFF: 358 return "aml_staff"; 359 case TALER_EXCHANGEDB_RT_PURSE_DELETION: 360 return "purse_deletion"; 361 case TALER_EXCHANGEDB_RT_WITHDRAW: 362 return "withdraw"; 363 case TALER_EXCHANGEDB_RT_LEGITIMIZATION_MEASURES: 364 return "legitimization_measures"; 365 case TALER_EXCHANGEDB_RT_LEGITIMIZATION_OUTCOMES: 366 return "legitimization_outcomes"; 367 case TALER_EXCHANGEDB_RT_LEGITIMIZATION_PROCESSES: 368 return "legitimization_processes"; 369 case TALER_EXCHANGEDB_RT_KYC_ATTRIBUTES: 370 return "kyc_attributes"; 371 case TALER_EXCHANGEDB_RT_AML_HISTORY: 372 return "aml_history"; 373 case TALER_EXCHANGEDB_RT_KYC_EVENTS: 374 return "kyc_events"; 375 case TALER_EXCHANGEDB_RT_KYCAUTHS_IN: 376 return "kycauths_in"; 377 } 378 GNUNET_break (0); 379 return "<unknown>"; 380 } 381 382 383 /** 384 * Information we track per replicated table. 385 */ 386 struct Table 387 { 388 /** 389 * Which table is this record about? 390 */ 391 enum TALER_EXCHANGEDB_ReplicatedTable rt; 392 393 /** 394 * Up to which record is the destination table synchronized. 395 */ 396 uint64_t start_serial; 397 398 /** 399 * Highest serial in the source table. 400 */ 401 uint64_t end_serial; 402 403 /** 404 * Marker for the end of the list of #tables. 405 */ 406 bool end; 407 }; 408 409 410 /** 411 * Information about replicated tables. 412 * 413 * The order matters: a table must be listed after every table its 414 * foreign keys point to, as the destination database enforces those 415 * constraints just like the source does. Must cover every value of 416 * `enum TALER_EXCHANGEDB_ReplicatedTable`, which #check_tables() verifies 417 * at startup. 418 */ 419 static struct Table tables[] = { 420 { .rt = TALER_EXCHANGEDB_RT_DENOMINATIONS}, 421 { .rt = TALER_EXCHANGEDB_RT_DENOMINATION_REVOCATIONS}, 422 { .rt = TALER_EXCHANGEDB_RT_KYC_TARGETS}, 423 { .rt = TALER_EXCHANGEDB_RT_WIRE_TARGETS}, 424 { .rt = TALER_EXCHANGEDB_RT_LEGITIMIZATION_MEASURES}, 425 { .rt = TALER_EXCHANGEDB_RT_LEGITIMIZATION_OUTCOMES}, 426 { .rt = TALER_EXCHANGEDB_RT_LEGITIMIZATION_PROCESSES}, 427 /* kyc_attributes.legitimization_serial references 428 legitimization_processes */ 429 { .rt = TALER_EXCHANGEDB_RT_KYC_ATTRIBUTES}, 430 { .rt = TALER_EXCHANGEDB_RT_AML_STAFF}, 431 /* aml_history.outcome_serial_id references legitimization_outcomes, 432 aml_history.kyc_attributes_serial_id references kyc_attributes */ 433 { .rt = TALER_EXCHANGEDB_RT_AML_HISTORY}, 434 { .rt = TALER_EXCHANGEDB_RT_KYC_EVENTS}, 435 { .rt = TALER_EXCHANGEDB_RT_RESERVES}, 436 { .rt = TALER_EXCHANGEDB_RT_RESERVES_IN}, 437 { .rt = TALER_EXCHANGEDB_RT_KYCAUTHS_IN}, 438 { .rt = TALER_EXCHANGEDB_RT_RESERVES_CLOSE}, 439 { .rt = TALER_EXCHANGEDB_RT_RESERVES_OPEN_REQUESTS}, 440 { .rt = TALER_EXCHANGEDB_RT_RESERVES_OPEN_DEPOSITS}, 441 { .rt = TALER_EXCHANGEDB_RT_WITHDRAW}, 442 { .rt = TALER_EXCHANGEDB_RT_AUDITORS}, 443 { .rt = TALER_EXCHANGEDB_RT_AUDITOR_DENOM_SIGS}, 444 { .rt = TALER_EXCHANGEDB_RT_EXCHANGE_SIGN_KEYS}, 445 { .rt = TALER_EXCHANGEDB_RT_SIGNKEY_REVOCATIONS}, 446 { .rt = TALER_EXCHANGEDB_RT_KNOWN_COINS}, 447 { .rt = TALER_EXCHANGEDB_RT_REFRESH}, 448 { .rt = TALER_EXCHANGEDB_RT_BATCH_DEPOSITS}, 449 { .rt = TALER_EXCHANGEDB_RT_COIN_DEPOSITS}, 450 { .rt = TALER_EXCHANGEDB_RT_REFUNDS}, 451 { .rt = TALER_EXCHANGEDB_RT_WIRE_OUT}, 452 { .rt = TALER_EXCHANGEDB_RT_AGGREGATION_TRACKING}, 453 /* after batch_deposits: the foreign key that ties a deferral's lifetime to 454 the deposits it is about is enforced here too */ 455 { .rt = TALER_EXCHANGEDB_RT_AGGREGATION_DEFERRALS}, 456 { .rt = TALER_EXCHANGEDB_RT_WIRE_FEE}, 457 { .rt = TALER_EXCHANGEDB_RT_GLOBAL_FEE}, 458 { .rt = TALER_EXCHANGEDB_RT_RECOUP}, 459 { .rt = TALER_EXCHANGEDB_RT_RECOUP_REFRESH }, 460 { .rt = TALER_EXCHANGEDB_RT_PURSE_REQUESTS}, 461 { .rt = TALER_EXCHANGEDB_RT_PURSE_DECISION}, 462 { .rt = TALER_EXCHANGEDB_RT_PURSE_MERGES}, 463 { .rt = TALER_EXCHANGEDB_RT_PURSE_DEPOSITS}, 464 { .rt = TALER_EXCHANGEDB_RT_PURSE_DELETION}, 465 { .rt = TALER_EXCHANGEDB_RT_ACCOUNT_MERGES}, 466 { .rt = TALER_EXCHANGEDB_RT_HISTORY_REQUESTS}, 467 { .rt = TALER_EXCHANGEDB_RT_CLOSE_REQUESTS}, 468 { .rt = TALER_EXCHANGEDB_RT_WADS_OUT}, 469 { .rt = TALER_EXCHANGEDB_RT_WADS_OUT_ENTRIES}, 470 { .rt = TALER_EXCHANGEDB_RT_WADS_IN}, 471 { .rt = TALER_EXCHANGEDB_RT_WADS_IN_ENTRIES}, 472 { .rt = TALER_EXCHANGEDB_RT_PROFIT_DRAINS}, 473 { .end = true } 474 }; 475 476 477 /** 478 * Check that #tables lists every replicated table exactly once. A table 479 * missing here is not an error the operator could ever notice: the sync 480 * would simply run to completion with that table left empty. 481 * 482 * @return #GNUNET_OK if #tables is complete 483 */ 484 static enum GNUNET_GenericReturnValue 485 check_tables (void) 486 { 487 bool seen[RT_MAX + 1] = { false }; 488 enum GNUNET_GenericReturnValue ret = GNUNET_OK; 489 490 for (unsigned int i = 0; ! tables[i].end; i++) 491 { 492 enum TALER_EXCHANGEDB_ReplicatedTable rt = tables[i].rt; 493 494 if (seen[rt]) 495 { 496 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 497 "Table `%s' is listed for replication more than once\n", 498 rt_name (rt)); 499 ret = GNUNET_SYSERR; 500 } 501 seen[rt] = true; 502 } 503 for (unsigned int i = 0; i <= RT_MAX; i++) 504 { 505 if (seen[i]) 506 continue; 507 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 508 "Table `%s' would not be replicated\n", 509 rt_name ((enum TALER_EXCHANGEDB_ReplicatedTable) i)); 510 ret = GNUNET_SYSERR; 511 } 512 return ret; 513 } 514 515 516 /** 517 * Closure for #do_insert. 518 */ 519 struct InsertContext 520 { 521 /** 522 * Table we are replicating. 523 */ 524 struct Table *table; 525 526 /** 527 * Set to error if insertion created an error. 528 */ 529 enum GNUNET_DB_QueryStatus qs; 530 }; 531 532 533 /** 534 * Function called on data to replicate in the auditor's database. 535 * 536 * @param ctx closure, a `struct InsertContext` 537 * @param td record from an exchange table 538 * @return #GNUNET_OK to continue to iterate, 539 * #GNUNET_SYSERR to fail with an error 540 */ 541 static enum GNUNET_GenericReturnValue 542 do_insert (struct InsertContext *ctx, 543 const struct TALER_EXCHANGEDB_TableData *td) 544 { 545 enum GNUNET_DB_QueryStatus qs; 546 547 if (0 >= ctx->qs) 548 return GNUNET_SYSERR; 549 qs = TALER_EXCHANGEDB_insert_records_by_table (dst, 550 td); 551 if (0 >= qs) 552 { 553 switch (qs) 554 { 555 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 556 GNUNET_assert (0); 557 break; 558 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 559 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 560 "Failed to insert record into table `%s': no change\n", 561 rt_name (td->table)); 562 break; 563 case GNUNET_DB_STATUS_SOFT_ERROR: 564 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 565 "Serialization error inserting record into table `%s' (will retry)\n", 566 rt_name (td->table)); 567 break; 568 case GNUNET_DB_STATUS_HARD_ERROR: 569 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 570 "Failed to insert record into table `%s': hard error\n", 571 rt_name (td->table)); 572 break; 573 } 574 ctx->qs = qs; 575 return GNUNET_SYSERR; 576 } 577 actual_size++; 578 ctx->table->start_serial = td->serial; 579 return GNUNET_OK; 580 } 581 582 583 /** 584 * Run one replication transaction. 585 * 586 * @return #GNUNET_OK on success, #GNUNET_SYSERR to rollback 587 */ 588 static enum GNUNET_GenericReturnValue 589 transact (void) 590 { 591 struct InsertContext ctx = { 592 .qs = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT 593 }; 594 595 if (0 > 596 TALER_EXCHANGEDB_start (src, 597 "lookup src serials")) 598 return GNUNET_SYSERR; 599 for (unsigned int i = 0; ! tables[i].end; i++) 600 TALER_EXCHANGEDB_get_serial_by_table (src, 601 tables[i].rt, 602 &tables[i].end_serial); 603 TALER_EXCHANGEDB_rollback (src); 604 if (GNUNET_OK != 605 TALER_EXCHANGEDB_start (dst, 606 "lookup dst serials")) 607 return GNUNET_SYSERR; 608 for (unsigned int i = 0; ! tables[i].end; i++) 609 TALER_EXCHANGEDB_get_serial_by_table (dst, 610 tables[i].rt, 611 &tables[i].start_serial); 612 TALER_EXCHANGEDB_rollback (dst); 613 for (unsigned int i = 0; ! tables[i].end; i++) 614 { 615 struct Table *table = &tables[i]; 616 617 if (table->start_serial == table->end_serial) 618 continue; 619 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 620 "Replicating table `%s' from %llu to %llu\n", 621 rt_name (table->rt), 622 (unsigned long long) table->start_serial, 623 (unsigned long long) table->end_serial); 624 ctx.table = table; 625 while (table->start_serial < table->end_serial) 626 { 627 enum GNUNET_DB_QueryStatus qs; 628 629 if (GNUNET_OK != 630 TALER_EXCHANGEDB_start (src, 631 "copy table (src)")) 632 return GNUNET_SYSERR; 633 if (GNUNET_OK != 634 TALER_EXCHANGEDB_start (dst, 635 "copy table (dst)")) 636 return GNUNET_SYSERR; 637 qs = TALER_EXCHANGEDB_iterate_records_by_table (src, 638 table->rt, 639 table->start_serial, 640 &do_insert, 641 &ctx); 642 if (ctx.qs < 0) 643 qs = ctx.qs; 644 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 645 { 646 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 647 "Failed to lookup records from table `%s': hard error\n", 648 rt_name (table->rt)); 649 global_ret = EXIT_FAILURE; 650 return GNUNET_SYSERR; 651 } 652 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 653 { 654 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 655 "Serialization error looking up records from table `%s' (will retry)\n", 656 rt_name (table->rt)); 657 return GNUNET_SYSERR; /* will retry */ 658 } 659 if (0 == qs) 660 { 661 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 662 "Failed to lookup records from table `%s': no results\n", 663 rt_name (table->rt)); 664 GNUNET_break (0); /* should be impossible */ 665 global_ret = EXIT_FAILURE; 666 return GNUNET_SYSERR; 667 } 668 if (0 == ctx.qs) 669 return GNUNET_SYSERR; /* insertion failed, maybe record existed? try again */ 670 TALER_EXCHANGEDB_rollback (src); 671 qs = TALER_EXCHANGEDB_commit (dst); 672 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 673 { 674 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 675 "Serialization error committing transaction on table `%s' (will retry)\n", 676 rt_name (table->rt)); 677 continue; 678 } 679 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 680 { 681 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 682 "Hard error committing transaction on table `%s'\n", 683 rt_name (table->rt)); 684 global_ret = EXIT_FAILURE; 685 return GNUNET_SYSERR; 686 } 687 } 688 } 689 /* we do not care about conflicting UPDATEs to src table, so safe to just rollback */ 690 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 691 "Sync pass completed successfully with %llu updates\n", 692 actual_size); 693 return GNUNET_OK; 694 } 695 696 697 /** 698 * Task to do the actual synchronization work. 699 * 700 * @param cls NULL, unused 701 */ 702 static void 703 do_sync (void *cls) 704 { 705 static struct GNUNET_TIME_Relative delay; 706 enum GNUNET_GenericReturnValue ret; 707 708 (void) cls; 709 sync_task = NULL; 710 actual_size = 0; 711 if (GNUNET_SYSERR == 712 TALER_EXCHANGEDB_preflight (src)) 713 { 714 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 715 "Failed to begin transaction with data source. Exiting\n"); 716 return; 717 } 718 if (GNUNET_SYSERR == 719 TALER_EXCHANGEDB_preflight (dst)) 720 { 721 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 722 "Failed to begin transaction with data destination. Exiting\n"); 723 return; 724 } 725 ret = transact (); 726 if (GNUNET_OK != ret) 727 { 728 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 729 "Transaction failed, rolling back\n"); 730 TALER_EXCHANGEDB_rollback (src); 731 TALER_EXCHANGEDB_rollback (dst); 732 } 733 if (0 != global_ret) 734 { 735 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 736 "Transaction failed permanently, exiting\n"); 737 return; 738 } 739 /* Only a pass that ran to completion proves the databases agree: 740 a rolled-back pass copies nothing, and reporting that as "in sync" 741 would let -t exit successfully having replicated nothing. */ 742 if ( (GNUNET_OK == ret) && 743 (0 == actual_size) && 744 (exit_if_synced) ) 745 { 746 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 747 "Databases are synchronized. Exiting\n"); 748 return; 749 } 750 if (actual_size < transaction_size / 2) 751 { 752 delay = GNUNET_TIME_STD_BACKOFF (delay); 753 } 754 else if (actual_size >= transaction_size) 755 { 756 delay = GNUNET_TIME_UNIT_ZERO; 757 } 758 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 759 "Next sync pass in %s\n", 760 GNUNET_STRINGS_relative_time_to_string (delay, 761 GNUNET_YES)); 762 sync_task = GNUNET_SCHEDULER_add_delayed (delay, 763 &do_sync, 764 NULL); 765 } 766 767 768 /** 769 * Set an option of type 'char *' from the command line with 770 * filename expansion a la #GNUNET_STRINGS_filename_expand(). 771 * 772 * @param ctx command line processing context 773 * @param scls additional closure (will point to the `char *`, 774 * which will be allocated) 775 * @param option name of the option 776 * @param value actual value of the option (a string) 777 * @return #GNUNET_OK 778 */ 779 static enum GNUNET_GenericReturnValue 780 set_filename (struct GNUNET_GETOPT_CommandLineProcessorContext *ctx, 781 void *scls, 782 const char *option, 783 const char *value) 784 { 785 char **val = scls; 786 787 (void) ctx; 788 (void) option; 789 GNUNET_assert (NULL != value); 790 GNUNET_free (*val); 791 *val = GNUNET_STRINGS_filename_expand (value); 792 return GNUNET_OK; 793 } 794 795 796 /** 797 * Allow user to specify configuration file name (-s option) 798 * 799 * @param[out] fn set to the name of the configuration file 800 */ 801 static struct GNUNET_GETOPT_CommandLineOption 802 option_cfgfile_src (char **fn) 803 { 804 struct GNUNET_GETOPT_CommandLineOption clo = { 805 .shortName = 's', 806 .name = "source-configuration", 807 .argumentHelp = "FILENAME", 808 .description = gettext_noop ( 809 "use configuration file FILENAME for the SOURCE database"), 810 .require_argument = 1, 811 .processor = &set_filename, 812 .scls = (void *) fn 813 }; 814 815 return clo; 816 } 817 818 819 /** 820 * Allow user to specify configuration file name (-d option) 821 * 822 * @param[out] fn set to the name of the configuration file 823 */ 824 static struct GNUNET_GETOPT_CommandLineOption 825 option_cfgfile_dst (char **fn) 826 { 827 struct GNUNET_GETOPT_CommandLineOption clo = { 828 .shortName = 'd', 829 .name = "destination-configuration", 830 .argumentHelp = "FILENAME", 831 .description = gettext_noop ( 832 "use configuration file FILENAME for the DESTINATION database"), 833 .require_argument = 1, 834 .processor = &set_filename, 835 .scls = (void *) fn 836 }; 837 838 return clo; 839 } 840 841 842 static struct GNUNET_CONFIGURATION_Handle * 843 load_config (const char *cfgfile) 844 { 845 struct GNUNET_CONFIGURATION_Handle *cfg; 846 847 cfg = GNUNET_CONFIGURATION_create (TALER_AUDITOR_project_data ()); 848 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 849 "Loading config file: %s\n", 850 cfgfile); 851 if (GNUNET_SYSERR == 852 GNUNET_CONFIGURATION_load (cfg, 853 cfgfile)) 854 { 855 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 856 "Malformed configuration file `%s', exit ...\n", 857 cfgfile); 858 GNUNET_CONFIGURATION_destroy (cfg); 859 return NULL; 860 } 861 return cfg; 862 } 863 864 865 /** 866 * Shutdown task. 867 * 868 * @param cls NULL, unused 869 */ 870 static void 871 do_shutdown (void *cls) 872 { 873 (void) cls; 874 if (NULL != sync_task) 875 { 876 GNUNET_SCHEDULER_cancel (sync_task); 877 sync_task = NULL; 878 } 879 } 880 881 882 /** 883 * Initial task. 884 * 885 * @param cls NULL, unused 886 */ 887 static void 888 run (void *cls) 889 { 890 (void) cls; 891 892 GNUNET_SCHEDULER_add_shutdown (&do_shutdown, 893 NULL); 894 sync_task = GNUNET_SCHEDULER_add_now (&do_sync, 895 NULL); 896 } 897 898 899 /** 900 * Setup plugins in #src and #dst and #run() the main 901 * logic with those plugins. 902 */ 903 static void 904 setup (struct GNUNET_CONFIGURATION_Handle *src_cfg, 905 struct GNUNET_CONFIGURATION_Handle *dst_cfg) 906 { 907 src = TALER_EXCHANGEDB_connect (src_cfg); 908 if (NULL == src) 909 { 910 global_ret = EXIT_NOTINSTALLED; 911 return; 912 } 913 dst = TALER_EXCHANGEDB_connect (dst_cfg); 914 if (NULL == dst) 915 { 916 global_ret = EXIT_NOTINSTALLED; 917 TALER_EXCHANGEDB_disconnect (src); 918 src = NULL; 919 return; 920 } 921 GNUNET_SCHEDULER_run (&run, 922 NULL); 923 TALER_EXCHANGEDB_disconnect (src); 924 src = NULL; 925 TALER_EXCHANGEDB_disconnect (dst); 926 dst = NULL; 927 } 928 929 930 /** 931 * The main function of the taler-auditor-exchange tool. This tool is used 932 * to add (or remove) an exchange's master key and base URL to the auditor's 933 * database. 934 * 935 * @param argc number of arguments from the command line 936 * @param argv command line arguments 937 * @return 0 ok, non-zero on error 938 */ 939 int 940 main (int argc, 941 char *const *argv) 942 { 943 char *src_cfgfile = NULL; 944 char *dst_cfgfile = NULL; 945 char *level = GNUNET_strdup ("WARNING"); 946 struct GNUNET_CONFIGURATION_Handle *src_cfg; 947 struct GNUNET_CONFIGURATION_Handle *dst_cfg; 948 const struct GNUNET_GETOPT_CommandLineOption options[] = { 949 GNUNET_GETOPT_option_mandatory ( 950 option_cfgfile_src (&src_cfgfile)), 951 GNUNET_GETOPT_option_mandatory ( 952 option_cfgfile_dst (&dst_cfgfile)), 953 GNUNET_GETOPT_option_help ( 954 TALER_AUDITOR_project_data (), 955 gettext_noop ("Make a safe copy of an exchange database")), 956 GNUNET_GETOPT_option_uint ( 957 'b', 958 "batch", 959 "SIZE", 960 gettext_noop ( 961 "target SIZE for a the number of records to copy in one transaction"), 962 &transaction_size), 963 GNUNET_GETOPT_option_flag ( 964 't', 965 "terminate-when-synchronized", 966 gettext_noop ( 967 "terminate as soon as the databases are synchronized"), 968 &exit_if_synced), 969 GNUNET_GETOPT_option_version (VERSION), 970 GNUNET_GETOPT_option_loglevel (&level), 971 GNUNET_GETOPT_OPTION_END 972 }; 973 974 TALER_gcrypt_init (); /* must trigger initialization manually at this point! */ 975 { 976 int ret; 977 978 ret = GNUNET_GETOPT_run ("taler-auditor-sync", 979 options, 980 argc, argv); 981 if (GNUNET_NO == ret) 982 return EXIT_SUCCESS; 983 if (GNUNET_SYSERR == ret) 984 return EXIT_INVALIDARGUMENT; 985 } 986 GNUNET_assert (GNUNET_OK == 987 GNUNET_log_setup ("taler-auditor-sync", 988 level, 989 NULL)); 990 GNUNET_free (level); 991 /* suppress compiler warnings... */ 992 GNUNET_assert (NULL != src_cfgfile); 993 GNUNET_assert (NULL != dst_cfgfile); 994 if (GNUNET_OK != 995 check_tables ()) 996 { 997 GNUNET_break (0); 998 GNUNET_free (src_cfgfile); 999 GNUNET_free (dst_cfgfile); 1000 return EXIT_FAILURE; 1001 } 1002 if (0 == strcmp (src_cfgfile, 1003 dst_cfgfile)) 1004 { 1005 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1006 "Source and destination configuration files must differ!\n"); 1007 return EXIT_INVALIDARGUMENT; 1008 } 1009 src_cfg = load_config (src_cfgfile); 1010 if (NULL == src_cfg) 1011 { 1012 GNUNET_free (src_cfgfile); 1013 GNUNET_free (dst_cfgfile); 1014 return EXIT_NOTCONFIGURED; 1015 } 1016 dst_cfg = load_config (dst_cfgfile); 1017 if (NULL == dst_cfg) 1018 { 1019 GNUNET_CONFIGURATION_destroy (src_cfg); 1020 GNUNET_free (src_cfgfile); 1021 GNUNET_free (dst_cfgfile); 1022 return EXIT_NOTCONFIGURED; 1023 } 1024 setup (src_cfg, 1025 dst_cfg); 1026 GNUNET_CONFIGURATION_destroy (src_cfg); 1027 GNUNET_CONFIGURATION_destroy (dst_cfg); 1028 GNUNET_free (src_cfgfile); 1029 GNUNET_free (dst_cfgfile); 1030 1031 return global_ret; 1032 } 1033 1034 1035 /* end of taler-auditor-sync.c */