taler-auditor-sync.c (37252B)
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/start_deferred_wire_out.h" 205 #include "exchange-database/start.h" 206 #include "exchange-database/start_read_committed.h" 207 #include "exchange-database/start_read_only.h" 208 #include "exchange-database/insert_wire_out.h" 209 #include "exchange-database/get_exists_aml_officer.h" 210 #include "exchange-database/do_trigger_kyc_rule_for_account.h" 211 #include "exchange-database/update_aggregation_transient.h" 212 #include "exchange-database/update_auditor.h" 213 #include "exchange-database/update_legitimization_process_by_row.h" 214 #include "exchange-database/update_wire.h" 215 #include "exchange-database/iterate_prewires.h" 216 #include "exchange-database/insert_prewire.h" 217 #include "exchange-database/update_to_prewire_failed.h" 218 #include "exchange-database/update_to_prewire_finished.h" 219 220 221 /** 222 * Handle to access the exchange's source database. 223 */ 224 static struct TALER_EXCHANGEDB_PostgresContext *src; 225 226 /** 227 * Handle to access the exchange's destination database. 228 */ 229 static struct TALER_EXCHANGEDB_PostgresContext *dst; 230 231 /** 232 * Return value from #main(). 233 */ 234 static int global_ret; 235 236 /** 237 * Main task to do synchronization. 238 */ 239 static struct GNUNET_SCHEDULER_Task *sync_task; 240 241 /** 242 * What is our target transaction size (number of records)? 243 */ 244 static unsigned int transaction_size = 512; 245 246 /** 247 * Number of records copied in this transaction. 248 */ 249 static unsigned long long actual_size; 250 251 /** 252 * Terminate once synchronization is achieved. 253 */ 254 static int exit_if_synced; 255 256 257 /** 258 * Highest value in `enum TALER_EXCHANGEDB_ReplicatedTable`. Update 259 * together with #rt_name() and #tables when the exchange gains another 260 * replicated table. 261 */ 262 #define RT_MAX TALER_EXCHANGEDB_RT_KYCAUTHS_IN 263 264 265 /** 266 * Return the name of the database table @a rt lives in. 267 * 268 * The switch is exhaustive and deliberately has no @e default case: when 269 * a table is added to `enum TALER_EXCHANGEDB_ReplicatedTable`, the 270 * compiler points at this function (-Wswitch), which is the reminder 271 * that #RT_MAX and #tables below must be extended as well. #check_tables() 272 * catches it at runtime if they were not. 273 * 274 * @param rt table to name 275 * @return name of the table in the database 276 */ 277 static const char * 278 rt_name (enum TALER_EXCHANGEDB_ReplicatedTable rt) 279 { 280 switch (rt) 281 { 282 case TALER_EXCHANGEDB_RT_DENOMINATIONS: 283 return "denominations"; 284 case TALER_EXCHANGEDB_RT_DENOMINATION_REVOCATIONS: 285 return "denomination_revocations"; 286 case TALER_EXCHANGEDB_RT_KYC_TARGETS: 287 return "kyc_targets"; 288 case TALER_EXCHANGEDB_RT_WIRE_TARGETS: 289 return "wire_targets"; 290 case TALER_EXCHANGEDB_RT_RESERVES: 291 return "reserves"; 292 case TALER_EXCHANGEDB_RT_RESERVES_IN: 293 return "reserves_in"; 294 case TALER_EXCHANGEDB_RT_RESERVES_CLOSE: 295 return "reserves_close"; 296 case TALER_EXCHANGEDB_RT_RESERVES_OPEN_REQUESTS: 297 return "reserves_open_requests"; 298 case TALER_EXCHANGEDB_RT_RESERVES_OPEN_DEPOSITS: 299 return "reserves_open_deposits"; 300 case TALER_EXCHANGEDB_RT_AUDITORS: 301 return "auditors"; 302 case TALER_EXCHANGEDB_RT_AUDITOR_DENOM_SIGS: 303 return "auditor_denom_sigs"; 304 case TALER_EXCHANGEDB_RT_EXCHANGE_SIGN_KEYS: 305 return "exchange_sign_keys"; 306 case TALER_EXCHANGEDB_RT_SIGNKEY_REVOCATIONS: 307 return "signkey_revocations"; 308 case TALER_EXCHANGEDB_RT_KNOWN_COINS: 309 return "known_coins"; 310 case TALER_EXCHANGEDB_RT_REFRESH: 311 return "refresh"; 312 case TALER_EXCHANGEDB_RT_BATCH_DEPOSITS: 313 return "batch_deposits"; 314 case TALER_EXCHANGEDB_RT_COIN_DEPOSITS: 315 return "coin_deposits"; 316 case TALER_EXCHANGEDB_RT_REFUNDS: 317 return "refunds"; 318 case TALER_EXCHANGEDB_RT_WIRE_OUT: 319 return "wire_out"; 320 case TALER_EXCHANGEDB_RT_AGGREGATION_TRACKING: 321 return "aggregation_tracking"; 322 case TALER_EXCHANGEDB_RT_AGGREGATION_DEFERRALS: 323 return "aggregation_deferrals"; 324 case TALER_EXCHANGEDB_RT_WIRE_FEE: 325 return "wire_fee"; 326 case TALER_EXCHANGEDB_RT_GLOBAL_FEE: 327 return "global_fee"; 328 case TALER_EXCHANGEDB_RT_RECOUP: 329 return "recoup"; 330 case TALER_EXCHANGEDB_RT_RECOUP_REFRESH: 331 return "recoup_refresh"; 332 case TALER_EXCHANGEDB_RT_PURSE_REQUESTS: 333 return "purse_requests"; 334 case TALER_EXCHANGEDB_RT_PURSE_DECISION: 335 return "purse_decision"; 336 case TALER_EXCHANGEDB_RT_PURSE_MERGES: 337 return "purse_merges"; 338 case TALER_EXCHANGEDB_RT_PURSE_DEPOSITS: 339 return "purse_deposits"; 340 case TALER_EXCHANGEDB_RT_ACCOUNT_MERGES: 341 return "account_merges"; 342 case TALER_EXCHANGEDB_RT_HISTORY_REQUESTS: 343 return "history_requests"; 344 case TALER_EXCHANGEDB_RT_CLOSE_REQUESTS: 345 return "close_requests"; 346 case TALER_EXCHANGEDB_RT_WADS_OUT: 347 return "wads_out"; 348 case TALER_EXCHANGEDB_RT_WADS_OUT_ENTRIES: 349 return "wad_out_entries"; 350 case TALER_EXCHANGEDB_RT_WADS_IN: 351 return "wads_in"; 352 case TALER_EXCHANGEDB_RT_WADS_IN_ENTRIES: 353 return "wad_in_entries"; 354 case TALER_EXCHANGEDB_RT_PROFIT_DRAINS: 355 return "profit_drains"; 356 case TALER_EXCHANGEDB_RT_AML_STAFF: 357 return "aml_staff"; 358 case TALER_EXCHANGEDB_RT_PURSE_DELETION: 359 return "purse_deletion"; 360 case TALER_EXCHANGEDB_RT_WITHDRAW: 361 return "withdraw"; 362 case TALER_EXCHANGEDB_RT_LEGITIMIZATION_MEASURES: 363 return "legitimization_measures"; 364 case TALER_EXCHANGEDB_RT_LEGITIMIZATION_OUTCOMES: 365 return "legitimization_outcomes"; 366 case TALER_EXCHANGEDB_RT_LEGITIMIZATION_PROCESSES: 367 return "legitimization_processes"; 368 case TALER_EXCHANGEDB_RT_KYC_ATTRIBUTES: 369 return "kyc_attributes"; 370 case TALER_EXCHANGEDB_RT_AML_HISTORY: 371 return "aml_history"; 372 case TALER_EXCHANGEDB_RT_KYC_EVENTS: 373 return "kyc_events"; 374 case TALER_EXCHANGEDB_RT_KYCAUTHS_IN: 375 return "kycauths_in"; 376 } 377 GNUNET_break (0); 378 return "<unknown>"; 379 } 380 381 382 /** 383 * Information we track per replicated table. 384 */ 385 struct Table 386 { 387 /** 388 * Which table is this record about? 389 */ 390 enum TALER_EXCHANGEDB_ReplicatedTable rt; 391 392 /** 393 * Up to which record is the destination table synchronized. 394 */ 395 uint64_t start_serial; 396 397 /** 398 * Highest serial in the source table. 399 */ 400 uint64_t end_serial; 401 402 /** 403 * Marker for the end of the list of #tables. 404 */ 405 bool end; 406 }; 407 408 409 /** 410 * Information about replicated tables. 411 * 412 * The order matters: a table must be listed after every table its 413 * foreign keys point to, as the destination database enforces those 414 * constraints just like the source does. Must cover every value of 415 * `enum TALER_EXCHANGEDB_ReplicatedTable`, which #check_tables() verifies 416 * at startup. 417 */ 418 static struct Table tables[] = { 419 { .rt = TALER_EXCHANGEDB_RT_DENOMINATIONS}, 420 { .rt = TALER_EXCHANGEDB_RT_DENOMINATION_REVOCATIONS}, 421 { .rt = TALER_EXCHANGEDB_RT_KYC_TARGETS}, 422 { .rt = TALER_EXCHANGEDB_RT_WIRE_TARGETS}, 423 { .rt = TALER_EXCHANGEDB_RT_LEGITIMIZATION_MEASURES}, 424 { .rt = TALER_EXCHANGEDB_RT_LEGITIMIZATION_OUTCOMES}, 425 { .rt = TALER_EXCHANGEDB_RT_LEGITIMIZATION_PROCESSES}, 426 /* kyc_attributes.legitimization_serial references 427 legitimization_processes */ 428 { .rt = TALER_EXCHANGEDB_RT_KYC_ATTRIBUTES}, 429 { .rt = TALER_EXCHANGEDB_RT_AML_STAFF}, 430 /* aml_history.outcome_serial_id references legitimization_outcomes, 431 aml_history.kyc_attributes_serial_id references kyc_attributes */ 432 { .rt = TALER_EXCHANGEDB_RT_AML_HISTORY}, 433 { .rt = TALER_EXCHANGEDB_RT_KYC_EVENTS}, 434 { .rt = TALER_EXCHANGEDB_RT_RESERVES}, 435 { .rt = TALER_EXCHANGEDB_RT_RESERVES_IN}, 436 { .rt = TALER_EXCHANGEDB_RT_KYCAUTHS_IN}, 437 { .rt = TALER_EXCHANGEDB_RT_RESERVES_CLOSE}, 438 { .rt = TALER_EXCHANGEDB_RT_RESERVES_OPEN_REQUESTS}, 439 { .rt = TALER_EXCHANGEDB_RT_RESERVES_OPEN_DEPOSITS}, 440 { .rt = TALER_EXCHANGEDB_RT_WITHDRAW}, 441 { .rt = TALER_EXCHANGEDB_RT_AUDITORS}, 442 { .rt = TALER_EXCHANGEDB_RT_AUDITOR_DENOM_SIGS}, 443 { .rt = TALER_EXCHANGEDB_RT_EXCHANGE_SIGN_KEYS}, 444 { .rt = TALER_EXCHANGEDB_RT_SIGNKEY_REVOCATIONS}, 445 { .rt = TALER_EXCHANGEDB_RT_KNOWN_COINS}, 446 { .rt = TALER_EXCHANGEDB_RT_REFRESH}, 447 { .rt = TALER_EXCHANGEDB_RT_BATCH_DEPOSITS}, 448 { .rt = TALER_EXCHANGEDB_RT_COIN_DEPOSITS}, 449 { .rt = TALER_EXCHANGEDB_RT_REFUNDS}, 450 { .rt = TALER_EXCHANGEDB_RT_WIRE_OUT}, 451 { .rt = TALER_EXCHANGEDB_RT_AGGREGATION_TRACKING}, 452 /* after batch_deposits: the foreign key that ties a deferral's lifetime to 453 the deposits it is about is enforced here too */ 454 { .rt = TALER_EXCHANGEDB_RT_AGGREGATION_DEFERRALS}, 455 { .rt = TALER_EXCHANGEDB_RT_WIRE_FEE}, 456 { .rt = TALER_EXCHANGEDB_RT_GLOBAL_FEE}, 457 { .rt = TALER_EXCHANGEDB_RT_RECOUP}, 458 { .rt = TALER_EXCHANGEDB_RT_RECOUP_REFRESH }, 459 { .rt = TALER_EXCHANGEDB_RT_PURSE_REQUESTS}, 460 { .rt = TALER_EXCHANGEDB_RT_PURSE_DECISION}, 461 { .rt = TALER_EXCHANGEDB_RT_PURSE_MERGES}, 462 { .rt = TALER_EXCHANGEDB_RT_PURSE_DEPOSITS}, 463 { .rt = TALER_EXCHANGEDB_RT_PURSE_DELETION}, 464 { .rt = TALER_EXCHANGEDB_RT_ACCOUNT_MERGES}, 465 { .rt = TALER_EXCHANGEDB_RT_HISTORY_REQUESTS}, 466 { .rt = TALER_EXCHANGEDB_RT_CLOSE_REQUESTS}, 467 { .rt = TALER_EXCHANGEDB_RT_WADS_OUT}, 468 { .rt = TALER_EXCHANGEDB_RT_WADS_OUT_ENTRIES}, 469 { .rt = TALER_EXCHANGEDB_RT_WADS_IN}, 470 { .rt = TALER_EXCHANGEDB_RT_WADS_IN_ENTRIES}, 471 { .rt = TALER_EXCHANGEDB_RT_PROFIT_DRAINS}, 472 { .end = true } 473 }; 474 475 476 /** 477 * Check that #tables lists every replicated table exactly once. A table 478 * missing here is not an error the operator could ever notice: the sync 479 * would simply run to completion with that table left empty. 480 * 481 * @return #GNUNET_OK if #tables is complete 482 */ 483 static enum GNUNET_GenericReturnValue 484 check_tables (void) 485 { 486 bool seen[RT_MAX + 1] = { false }; 487 enum GNUNET_GenericReturnValue ret = GNUNET_OK; 488 489 for (unsigned int i = 0; ! tables[i].end; i++) 490 { 491 enum TALER_EXCHANGEDB_ReplicatedTable rt = tables[i].rt; 492 493 if (seen[rt]) 494 { 495 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 496 "Table `%s' is listed for replication more than once\n", 497 rt_name (rt)); 498 ret = GNUNET_SYSERR; 499 } 500 seen[rt] = true; 501 } 502 for (unsigned int i = 0; i <= RT_MAX; i++) 503 { 504 if (seen[i]) 505 continue; 506 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 507 "Table `%s' would not be replicated\n", 508 rt_name ((enum TALER_EXCHANGEDB_ReplicatedTable) i)); 509 ret = GNUNET_SYSERR; 510 } 511 return ret; 512 } 513 514 515 /** 516 * Closure for #do_insert. 517 */ 518 struct InsertContext 519 { 520 /** 521 * Table we are replicating. 522 */ 523 struct Table *table; 524 525 /** 526 * Set to error if insertion created an error. 527 */ 528 enum GNUNET_DB_QueryStatus qs; 529 }; 530 531 532 /** 533 * Function called on data to replicate in the auditor's database. 534 * 535 * @param ctx closure, a `struct InsertContext` 536 * @param td record from an exchange table 537 * @return #GNUNET_OK to continue to iterate, 538 * #GNUNET_SYSERR to fail with an error 539 */ 540 static enum GNUNET_GenericReturnValue 541 do_insert (struct InsertContext *ctx, 542 const struct TALER_EXCHANGEDB_TableData *td) 543 { 544 enum GNUNET_DB_QueryStatus qs; 545 546 if (0 >= ctx->qs) 547 return GNUNET_SYSERR; 548 qs = TALER_EXCHANGEDB_insert_records_by_table (dst, 549 td); 550 if (0 >= qs) 551 { 552 switch (qs) 553 { 554 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 555 GNUNET_assert (0); 556 break; 557 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 558 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 559 "Failed to insert record into table `%s': no change\n", 560 rt_name (td->table)); 561 break; 562 case GNUNET_DB_STATUS_SOFT_ERROR: 563 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 564 "Serialization error inserting record into table `%s' (will retry)\n", 565 rt_name (td->table)); 566 break; 567 case GNUNET_DB_STATUS_HARD_ERROR: 568 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 569 "Failed to insert record into table `%s': hard error\n", 570 rt_name (td->table)); 571 break; 572 } 573 ctx->qs = qs; 574 return GNUNET_SYSERR; 575 } 576 actual_size++; 577 ctx->table->start_serial = td->serial; 578 return GNUNET_OK; 579 } 580 581 582 /** 583 * Determine the last serial number of each table in @a db. 584 * 585 * Failing to determine a serial must abort the pass: the previous 586 * value of the field is stale (#tables is file-scope), and continuing 587 * with it would silently skip records or entire tables. 588 * 589 * @param db database to query 590 * @param source true to store the result in the `end_serial` of each 591 * table (data source), false for `start_serial` (destination) 592 * @return #GNUNET_OK on success, #GNUNET_SYSERR to rollback 593 */ 594 static enum GNUNET_GenericReturnValue 595 lookup_serials (struct TALER_EXCHANGEDB_PostgresContext *db, 596 bool source) 597 { 598 for (unsigned int i = 0; ! tables[i].end; i++) 599 { 600 uint64_t *serial = source 601 ? &tables[i].end_serial 602 : &tables[i].start_serial; 603 enum GNUNET_DB_QueryStatus qs; 604 605 qs = TALER_EXCHANGEDB_get_serial_by_table (db, 606 tables[i].rt, 607 serial); 608 switch (qs) 609 { 610 case GNUNET_DB_STATUS_HARD_ERROR: 611 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 612 "Failed to determine last serial of table `%s': hard error\n", 613 rt_name (tables[i].rt)); 614 global_ret = EXIT_FAILURE; 615 return GNUNET_SYSERR; 616 case GNUNET_DB_STATUS_SOFT_ERROR: 617 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 618 "Serialization error determining last serial of table `%s' (will retry)\n", 619 rt_name (tables[i].rt)); 620 return GNUNET_SYSERR; 621 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 622 /* Table is empty; the statement did not write *serial. */ 623 *serial = 0; 624 break; 625 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 626 break; 627 } 628 } 629 return GNUNET_OK; 630 } 631 632 633 /** 634 * Run one replication transaction. 635 * 636 * @return #GNUNET_OK on success, #GNUNET_SYSERR to rollback 637 */ 638 static enum GNUNET_GenericReturnValue 639 transact (void) 640 { 641 struct InsertContext ctx = { 642 .qs = GNUNET_DB_STATUS_SUCCESS_ONE_RESULT 643 }; 644 645 if (0 > 646 TALER_EXCHANGEDB_start (src, 647 "lookup src serials")) 648 return GNUNET_SYSERR; 649 if (GNUNET_OK != 650 lookup_serials (src, 651 true)) 652 return GNUNET_SYSERR; 653 TALER_EXCHANGEDB_rollback (src); 654 if (GNUNET_OK != 655 TALER_EXCHANGEDB_start (dst, 656 "lookup dst serials")) 657 return GNUNET_SYSERR; 658 if (GNUNET_OK != 659 lookup_serials (dst, 660 false)) 661 return GNUNET_SYSERR; 662 TALER_EXCHANGEDB_rollback (dst); 663 for (unsigned int i = 0; ! tables[i].end; i++) 664 { 665 struct Table *table = &tables[i]; 666 667 if (table->start_serial == table->end_serial) 668 continue; 669 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 670 "Replicating table `%s' from %llu to %llu\n", 671 rt_name (table->rt), 672 (unsigned long long) table->start_serial, 673 (unsigned long long) table->end_serial); 674 ctx.table = table; 675 while (table->start_serial < table->end_serial) 676 { 677 enum GNUNET_DB_QueryStatus qs; 678 679 if (GNUNET_OK != 680 TALER_EXCHANGEDB_start (src, 681 "copy table (src)")) 682 return GNUNET_SYSERR; 683 if (GNUNET_OK != 684 TALER_EXCHANGEDB_start (dst, 685 "copy table (dst)")) 686 return GNUNET_SYSERR; 687 qs = TALER_EXCHANGEDB_iterate_records_by_table (src, 688 table->rt, 689 table->start_serial, 690 &do_insert, 691 &ctx); 692 if (ctx.qs < 0) 693 qs = ctx.qs; 694 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 695 { 696 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 697 "Failed to lookup records from table `%s': hard error\n", 698 rt_name (table->rt)); 699 global_ret = EXIT_FAILURE; 700 return GNUNET_SYSERR; 701 } 702 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 703 { 704 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 705 "Serialization error looking up records from table `%s' (will retry)\n", 706 rt_name (table->rt)); 707 return GNUNET_SYSERR; /* will retry */ 708 } 709 if (0 == qs) 710 { 711 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 712 "Failed to lookup records from table `%s': no results\n", 713 rt_name (table->rt)); 714 GNUNET_break (0); /* should be impossible */ 715 global_ret = EXIT_FAILURE; 716 return GNUNET_SYSERR; 717 } 718 if (0 == ctx.qs) 719 return GNUNET_SYSERR; /* insertion failed, maybe record existed? try again */ 720 TALER_EXCHANGEDB_rollback (src); 721 qs = TALER_EXCHANGEDB_commit (dst); 722 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 723 { 724 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 725 "Serialization error committing transaction on table `%s' (will retry)\n", 726 rt_name (table->rt)); 727 /* The rows of the failed commit are gone, but start_serial was 728 already advanced past them by do_insert(); continuing here 729 would leave a permanent hole. Restart the pass instead, which 730 re-reads the serials actually present in the destination. */ 731 return GNUNET_SYSERR; 732 } 733 if (GNUNET_DB_STATUS_HARD_ERROR == qs) 734 { 735 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 736 "Hard error committing transaction on table `%s'\n", 737 rt_name (table->rt)); 738 global_ret = EXIT_FAILURE; 739 return GNUNET_SYSERR; 740 } 741 } 742 } 743 /* we do not care about conflicting UPDATEs to src table, so safe to just rollback */ 744 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 745 "Sync pass completed successfully with %llu updates\n", 746 actual_size); 747 return GNUNET_OK; 748 } 749 750 751 /** 752 * Task to do the actual synchronization work. 753 * 754 * @param cls NULL, unused 755 */ 756 static void 757 do_sync (void *cls) 758 { 759 static struct GNUNET_TIME_Relative delay; 760 enum GNUNET_GenericReturnValue ret; 761 762 (void) cls; 763 sync_task = NULL; 764 actual_size = 0; 765 if (GNUNET_SYSERR == 766 TALER_EXCHANGEDB_preflight (src)) 767 { 768 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 769 "Failed to begin transaction with data source. Exiting\n"); 770 global_ret = EXIT_FAILURE; 771 return; 772 } 773 if (GNUNET_SYSERR == 774 TALER_EXCHANGEDB_preflight (dst)) 775 { 776 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 777 "Failed to begin transaction with data destination. Exiting\n"); 778 global_ret = EXIT_FAILURE; 779 return; 780 } 781 ret = transact (); 782 if (GNUNET_OK != ret) 783 { 784 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 785 "Transaction failed, rolling back\n"); 786 TALER_EXCHANGEDB_rollback (src); 787 TALER_EXCHANGEDB_rollback (dst); 788 } 789 if (0 != global_ret) 790 { 791 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 792 "Transaction failed permanently, exiting\n"); 793 return; 794 } 795 /* Only a pass that ran to completion proves the databases agree: 796 a rolled-back pass copies nothing, and reporting that as "in sync" 797 would let -t exit successfully having replicated nothing. */ 798 if ( (GNUNET_OK == ret) && 799 (0 == actual_size) && 800 (exit_if_synced) ) 801 { 802 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 803 "Databases are synchronized. Exiting\n"); 804 return; 805 } 806 if (actual_size < transaction_size / 2) 807 { 808 delay = GNUNET_TIME_STD_BACKOFF (delay); 809 } 810 else if (actual_size >= transaction_size) 811 { 812 delay = GNUNET_TIME_UNIT_ZERO; 813 } 814 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 815 "Next sync pass in %s\n", 816 GNUNET_STRINGS_relative_time_to_string (delay, 817 GNUNET_YES)); 818 sync_task = GNUNET_SCHEDULER_add_delayed (delay, 819 &do_sync, 820 NULL); 821 } 822 823 824 /** 825 * Set an option of type 'char *' from the command line with 826 * filename expansion a la #GNUNET_STRINGS_filename_expand(). 827 * 828 * @param ctx command line processing context 829 * @param scls additional closure (will point to the `char *`, 830 * which will be allocated) 831 * @param option name of the option 832 * @param value actual value of the option (a string) 833 * @return #GNUNET_OK 834 */ 835 static enum GNUNET_GenericReturnValue 836 set_filename (struct GNUNET_GETOPT_CommandLineProcessorContext *ctx, 837 void *scls, 838 const char *option, 839 const char *value) 840 { 841 char **val = scls; 842 843 (void) ctx; 844 (void) option; 845 GNUNET_assert (NULL != value); 846 GNUNET_free (*val); 847 *val = GNUNET_STRINGS_filename_expand (value); 848 return GNUNET_OK; 849 } 850 851 852 /** 853 * Allow user to specify configuration file name (-s option) 854 * 855 * @param[out] fn set to the name of the configuration file 856 */ 857 static struct GNUNET_GETOPT_CommandLineOption 858 option_cfgfile_src (char **fn) 859 { 860 struct GNUNET_GETOPT_CommandLineOption clo = { 861 .shortName = 's', 862 .name = "source-configuration", 863 .argumentHelp = "FILENAME", 864 .description = gettext_noop ( 865 "use configuration file FILENAME for the SOURCE database"), 866 .require_argument = 1, 867 .processor = &set_filename, 868 .scls = (void *) fn 869 }; 870 871 return clo; 872 } 873 874 875 /** 876 * Allow user to specify configuration file name (-d option) 877 * 878 * @param[out] fn set to the name of the configuration file 879 */ 880 static struct GNUNET_GETOPT_CommandLineOption 881 option_cfgfile_dst (char **fn) 882 { 883 struct GNUNET_GETOPT_CommandLineOption clo = { 884 .shortName = 'd', 885 .name = "destination-configuration", 886 .argumentHelp = "FILENAME", 887 .description = gettext_noop ( 888 "use configuration file FILENAME for the DESTINATION database"), 889 .require_argument = 1, 890 .processor = &set_filename, 891 .scls = (void *) fn 892 }; 893 894 return clo; 895 } 896 897 898 static struct GNUNET_CONFIGURATION_Handle * 899 load_config (const char *cfgfile) 900 { 901 struct GNUNET_CONFIGURATION_Handle *cfg; 902 903 cfg = GNUNET_CONFIGURATION_create (TALER_AUDITOR_project_data ()); 904 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 905 "Loading config file: %s\n", 906 cfgfile); 907 if (GNUNET_SYSERR == 908 GNUNET_CONFIGURATION_load (cfg, 909 cfgfile)) 910 { 911 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 912 "Malformed configuration file `%s', exit ...\n", 913 cfgfile); 914 GNUNET_CONFIGURATION_destroy (cfg); 915 return NULL; 916 } 917 return cfg; 918 } 919 920 921 /** 922 * Shutdown task. 923 * 924 * @param cls NULL, unused 925 */ 926 static void 927 do_shutdown (void *cls) 928 { 929 (void) cls; 930 if (NULL != sync_task) 931 { 932 GNUNET_SCHEDULER_cancel (sync_task); 933 sync_task = NULL; 934 } 935 } 936 937 938 /** 939 * Initial task. 940 * 941 * @param cls NULL, unused 942 */ 943 static void 944 run (void *cls) 945 { 946 (void) cls; 947 948 GNUNET_SCHEDULER_add_shutdown (&do_shutdown, 949 NULL); 950 sync_task = GNUNET_SCHEDULER_add_now (&do_sync, 951 NULL); 952 } 953 954 955 /** 956 * Setup plugins in #src and #dst and #run() the main 957 * logic with those plugins. 958 */ 959 static void 960 setup (struct GNUNET_CONFIGURATION_Handle *src_cfg, 961 struct GNUNET_CONFIGURATION_Handle *dst_cfg) 962 { 963 src = TALER_EXCHANGEDB_connect (src_cfg); 964 if (NULL == src) 965 { 966 global_ret = EXIT_NOTINSTALLED; 967 return; 968 } 969 dst = TALER_EXCHANGEDB_connect (dst_cfg); 970 if (NULL == dst) 971 { 972 global_ret = EXIT_NOTINSTALLED; 973 TALER_EXCHANGEDB_disconnect (src); 974 src = NULL; 975 return; 976 } 977 GNUNET_SCHEDULER_run (&run, 978 NULL); 979 TALER_EXCHANGEDB_disconnect (src); 980 src = NULL; 981 TALER_EXCHANGEDB_disconnect (dst); 982 dst = NULL; 983 } 984 985 986 /** 987 * The main function of the taler-auditor-exchange tool. This tool is used 988 * to add (or remove) an exchange's master key and base URL to the auditor's 989 * database. 990 * 991 * @param argc number of arguments from the command line 992 * @param argv command line arguments 993 * @return 0 ok, non-zero on error 994 */ 995 int 996 main (int argc, 997 char *const *argv) 998 { 999 char *src_cfgfile = NULL; 1000 char *dst_cfgfile = NULL; 1001 char *level = GNUNET_strdup ("WARNING"); 1002 struct GNUNET_CONFIGURATION_Handle *src_cfg; 1003 struct GNUNET_CONFIGURATION_Handle *dst_cfg; 1004 const struct GNUNET_GETOPT_CommandLineOption options[] = { 1005 GNUNET_GETOPT_option_mandatory ( 1006 option_cfgfile_src (&src_cfgfile)), 1007 GNUNET_GETOPT_option_mandatory ( 1008 option_cfgfile_dst (&dst_cfgfile)), 1009 GNUNET_GETOPT_option_help ( 1010 TALER_AUDITOR_project_data (), 1011 gettext_noop ("Make a safe copy of an exchange database")), 1012 GNUNET_GETOPT_option_uint ( 1013 'b', 1014 "batch", 1015 "SIZE", 1016 gettext_noop ( 1017 "target SIZE for a the number of records to copy in one transaction"), 1018 &transaction_size), 1019 GNUNET_GETOPT_option_flag ( 1020 't', 1021 "terminate-when-synchronized", 1022 gettext_noop ( 1023 "terminate as soon as the databases are synchronized"), 1024 &exit_if_synced), 1025 GNUNET_GETOPT_option_version (VERSION), 1026 GNUNET_GETOPT_option_loglevel (&level), 1027 GNUNET_GETOPT_OPTION_END 1028 }; 1029 1030 TALER_gcrypt_init (); /* must trigger initialization manually at this point! */ 1031 { 1032 int ret; 1033 1034 ret = GNUNET_GETOPT_run ("taler-auditor-sync", 1035 options, 1036 argc, argv); 1037 if (GNUNET_NO == ret) 1038 return EXIT_SUCCESS; 1039 if (GNUNET_SYSERR == ret) 1040 return EXIT_INVALIDARGUMENT; 1041 } 1042 GNUNET_assert (GNUNET_OK == 1043 GNUNET_log_setup ("taler-auditor-sync", 1044 level, 1045 NULL)); 1046 GNUNET_free (level); 1047 /* suppress compiler warnings... */ 1048 GNUNET_assert (NULL != src_cfgfile); 1049 GNUNET_assert (NULL != dst_cfgfile); 1050 if (GNUNET_OK != 1051 check_tables ()) 1052 { 1053 GNUNET_break (0); 1054 GNUNET_free (src_cfgfile); 1055 GNUNET_free (dst_cfgfile); 1056 return EXIT_FAILURE; 1057 } 1058 if (0 == strcmp (src_cfgfile, 1059 dst_cfgfile)) 1060 { 1061 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 1062 "Source and destination configuration files must differ!\n"); 1063 return EXIT_INVALIDARGUMENT; 1064 } 1065 src_cfg = load_config (src_cfgfile); 1066 if (NULL == src_cfg) 1067 { 1068 GNUNET_free (src_cfgfile); 1069 GNUNET_free (dst_cfgfile); 1070 return EXIT_NOTCONFIGURED; 1071 } 1072 dst_cfg = load_config (dst_cfgfile); 1073 if (NULL == dst_cfg) 1074 { 1075 GNUNET_CONFIGURATION_destroy (src_cfg); 1076 GNUNET_free (src_cfgfile); 1077 GNUNET_free (dst_cfgfile); 1078 return EXIT_NOTCONFIGURED; 1079 } 1080 setup (src_cfg, 1081 dst_cfg); 1082 GNUNET_CONFIGURATION_destroy (src_cfg); 1083 GNUNET_CONFIGURATION_destroy (dst_cfg); 1084 GNUNET_free (src_cfgfile); 1085 GNUNET_free (dst_cfgfile); 1086 1087 return global_ret; 1088 } 1089 1090 1091 /* end of taler-auditor-sync.c */