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