taler-exchange-router.c (12751B)
1 /* 2 This file is part of TALER 3 Copyright (C) 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 Affero General Public License as published by the Free Software 7 Foundation; either version 3, or (at your option) any later version. 8 9 TALER is distributed in the hope that it will be useful, but WITHOUT ANY 10 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 11 A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. 12 13 You should have received a copy of the GNU Affero General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 17 /** 18 * @file taler-exchange-router.c 19 * @brief Process that routes P2P payments. Responsible for 20 * aggregating remote payments into the respective wad transfers. 21 * Execution of actual wad transfers is still to be done by taler-exchange-transfer, 22 * and watching for incoming wad transfers is done by taler-exchange-wirewatch. 23 * @author Christian Grothoff 24 */ 25 #include "platform.h" 26 #include <gnunet/gnunet_util_lib.h> 27 #include <jansson.h> 28 #include <pthread.h> 29 #include "exchangedb_lib.h" 30 #include "taler/taler_json_lib.h" 31 #include "taler/taler_bank_service.h" 32 #include "exchange-database/release_revolving_shard.h" 33 #include "exchange-database/start.h" 34 #include "exchange-database/commit.h" 35 #include "exchange-database/preflight.h" 36 #include "exchange-database/begin_revolving_shard.h" 37 #include "exchange-database/release_revolving_shard.h" 38 #include "exchange-database/release_revolving_shard.h" 39 40 41 // FIXME #7271: revisit how (and if) we do sharding! 42 // Maybe use different helpers for wads than 43 // for local purses?! 44 /** 45 * Work shard we are processing. 46 */ 47 struct Shard 48 { 49 50 /** 51 * When did we start processing the shard? 52 */ 53 struct GNUNET_TIME_Timestamp start_time; 54 55 /** 56 * Starting row of the shard. 57 */ 58 uint32_t shard_start; 59 60 /** 61 * Inclusive end row of the shard. 62 */ 63 uint32_t shard_end; 64 65 /** 66 * Number of starting points found in the shard. 67 */ 68 uint64_t work_counter; 69 70 }; 71 72 73 /** 74 * What is the smallest unit we support for wire transfers? 75 * We will need to round down to a multiple of this amount. 76 */ 77 static struct TALER_Amount currency_round_unit; 78 79 /** 80 * What is the base URL of this exchange? Used in the 81 * wire transfer subjects so that merchants and governments 82 * can ask for the list of aggregated deposits. 83 */ 84 static char *exchange_base_url; 85 86 /** 87 * Set to #GNUNET_YES if this exchange does not support KYC checks 88 * and thus P2P transfers are to be made regardless of the 89 * KYC status of the target reserve. 90 */ 91 static int kyc_off; 92 93 /** 94 * The exchange's configuration. 95 */ 96 static const struct GNUNET_CONFIGURATION_Handle *cfg; 97 98 /** 99 * Our database plugin. 100 */ 101 static struct TALER_EXCHANGEDB_PostgresContext *pg; 102 103 /** 104 * Next task to run, if any. 105 */ 106 static struct GNUNET_SCHEDULER_Task *task; 107 108 /** 109 * How long should we sleep when idle before trying to find more work? 110 */ 111 static struct GNUNET_TIME_Relative router_idle_sleep_interval; 112 113 /** 114 * How big are the shards we are processing? Is an inclusive offset, so every 115 * shard ranges from [X,X+shard_size) exclusive. So a shard covers 116 * shard_size slots. The maximum value for shard_size is INT32_MAX+1. 117 */ 118 static uint32_t shard_size; 119 120 /** 121 * Value to return from main(). 0 on success, non-zero on errors. 122 */ 123 static int global_ret; 124 125 /** 126 * #GNUNET_YES if we are in test mode and should exit when idle. 127 */ 128 static int test_mode; 129 130 131 /** 132 * Select a shard to work on. 133 * 134 * @param cls NULL 135 */ 136 static void 137 run_shard (void *cls); 138 139 140 /** 141 * We're being aborted with CTRL-C (or SIGTERM). Shut down. 142 * 143 * @param cls closure 144 */ 145 static void 146 shutdown_task (void *cls) 147 { 148 (void) cls; 149 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 150 "Running shutdown\n"); 151 if (NULL != task) 152 { 153 GNUNET_SCHEDULER_cancel (task); 154 task = NULL; 155 } 156 TALER_EXCHANGEDB_disconnect (pg); 157 pg = NULL; 158 TALER_EXCHANGEDB_unload_accounts (); 159 cfg = NULL; 160 } 161 162 163 /** 164 * Parse the configuration for wirewatch. 165 * 166 * @return #GNUNET_OK on success 167 */ 168 static enum GNUNET_GenericReturnValue 169 parse_wirewatch_config (void) 170 { 171 if (GNUNET_OK != 172 GNUNET_CONFIGURATION_get_value_string (cfg, 173 "exchange", 174 "BASE_URL", 175 &exchange_base_url)) 176 { 177 GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR, 178 "exchange", 179 "BASE_URL"); 180 return GNUNET_SYSERR; 181 } 182 if (GNUNET_OK != 183 GNUNET_CONFIGURATION_get_value_time (cfg, 184 "exchange", 185 "ROUTER_IDLE_SLEEP_INTERVAL", 186 &router_idle_sleep_interval)) 187 { 188 GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR, 189 "exchange", 190 "ROUTER_IDLE_SLEEP_INTERVAL"); 191 return GNUNET_SYSERR; 192 } 193 if ( (GNUNET_OK != 194 TALER_config_get_amount (cfg, 195 "exchange", 196 "CURRENCY_ROUND_UNIT", 197 ¤cy_round_unit)) || 198 (TALER_amount_is_zero (¤cy_round_unit)) ) 199 { 200 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 201 "Need non-zero value in section `exchange' under `CURRENCY_ROUND_UNIT'\n"); 202 return GNUNET_SYSERR; 203 } 204 205 if (NULL == 206 (pg = TALER_EXCHANGEDB_connect (cfg))) 207 { 208 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 209 "Failed to initialize DB subsystem\n"); 210 return GNUNET_SYSERR; 211 } 212 if (GNUNET_OK != 213 TALER_EXCHANGEDB_load_accounts (cfg, 214 TALER_EXCHANGEDB_ALO_DEBIT)) 215 { 216 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 217 "No wire accounts configured for debit!\n"); 218 TALER_EXCHANGEDB_disconnect (pg); 219 pg = NULL; 220 return GNUNET_SYSERR; 221 } 222 return GNUNET_OK; 223 } 224 225 226 /** 227 * Perform a database commit. If it fails, print a warning. 228 * 229 * @return status of commit 230 */ 231 static enum GNUNET_DB_QueryStatus 232 commit_or_warn (void) 233 { 234 enum GNUNET_DB_QueryStatus qs; 235 236 qs = TALER_EXCHANGEDB_commit (pg); 237 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 238 return qs; 239 GNUNET_log ((GNUNET_DB_STATUS_SOFT_ERROR == qs) 240 ? GNUNET_ERROR_TYPE_INFO 241 : GNUNET_ERROR_TYPE_ERROR, 242 "Failed to commit database transaction!\n"); 243 return qs; 244 } 245 246 247 /** 248 * Release lock on shard @a s in the database. 249 * On error, terminates this process. 250 * 251 * @param[in] s shard to free (and memory to release) 252 */ 253 static void 254 release_shard (struct Shard *s) 255 { 256 enum GNUNET_DB_QueryStatus qs; 257 258 qs = TALER_EXCHANGEDB_release_revolving_shard ( 259 pg, 260 "router", 261 s->shard_start, 262 s->shard_end); 263 GNUNET_free (s); 264 switch (qs) 265 { 266 case GNUNET_DB_STATUS_HARD_ERROR: 267 case GNUNET_DB_STATUS_SOFT_ERROR: 268 GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR != qs); 269 GNUNET_break (0); 270 global_ret = EXIT_FAILURE; 271 GNUNET_SCHEDULER_shutdown (); 272 return; 273 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 274 /* Strange, but let's just continue */ 275 break; 276 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 277 /* normal case */ 278 break; 279 } 280 } 281 282 283 static void 284 run_routing (void *cls) 285 { 286 struct Shard *s = cls; 287 288 task = NULL; 289 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 290 "Checking for ready P2P transfers to route\n"); 291 (void) TALER_EXCHANGEDB_start (pg, 292 "taler-exchange-router: main"); 293 // FIXME #7271: do actual work here! 294 commit_or_warn (); 295 release_shard (s); 296 /* The routing logic above is an unfinished stub (#7271) that currently 297 performs no work. Until it does, do not busy-loop hammering the 298 database: honor test_mode (exit once caught up) and otherwise wait 299 for the configured idle interval before acquiring the next shard. */ 300 if (test_mode) 301 { 302 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 303 "Transfers done, shutting down (test mode)\n"); 304 GNUNET_SCHEDULER_shutdown (); 305 return; 306 } 307 task = GNUNET_SCHEDULER_add_delayed (router_idle_sleep_interval, 308 &run_shard, 309 NULL); 310 } 311 312 313 /** 314 * Select a shard to work on. 315 * 316 * @param cls NULL 317 */ 318 static void 319 run_shard (void *cls) 320 { 321 struct Shard *s; 322 enum GNUNET_DB_QueryStatus qs; 323 324 (void) cls; 325 task = NULL; 326 if (GNUNET_SYSERR == 327 TALER_EXCHANGEDB_preflight (pg)) 328 { 329 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 330 "Failed to obtain database connection!\n"); 331 global_ret = EXIT_FAILURE; 332 GNUNET_SCHEDULER_shutdown (); 333 return; 334 } 335 s = GNUNET_new (struct Shard); 336 s->start_time = GNUNET_TIME_timestamp_get (); 337 qs = TALER_EXCHANGEDB_begin_revolving_shard (pg, 338 "router", 339 shard_size, 340 1U + INT32_MAX, 341 &s->shard_start, 342 &s->shard_end); 343 if (0 >= qs) 344 { 345 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 346 { 347 static struct GNUNET_TIME_Relative delay; 348 349 GNUNET_free (s); 350 delay = GNUNET_TIME_randomized_backoff (delay, 351 GNUNET_TIME_UNIT_SECONDS); 352 task = GNUNET_SCHEDULER_add_delayed (delay, 353 &run_shard, 354 NULL); 355 return; 356 } 357 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 358 "Failed to begin shard (%d)!\n", 359 qs); 360 GNUNET_break (GNUNET_DB_STATUS_HARD_ERROR != qs); 361 global_ret = EXIT_FAILURE; 362 GNUNET_SCHEDULER_shutdown (); 363 return; 364 } 365 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 366 "Starting shard [%u:%u]!\n", 367 (unsigned int) s->shard_start, 368 (unsigned int) s->shard_end); 369 task = GNUNET_SCHEDULER_add_now (&run_routing, 370 s); 371 } 372 373 374 /** 375 * First task. 376 * 377 * @param cls closure, NULL 378 * @param args remaining command-line arguments 379 * @param cfgfile name of the configuration file used (for saving, can be NULL!) 380 * @param c configuration 381 */ 382 static void 383 run (void *cls, 384 char *const *args, 385 const char *cfgfile, 386 const struct GNUNET_CONFIGURATION_Handle *c) 387 { 388 unsigned long long ass; 389 (void) cls; 390 (void) args; 391 (void) cfgfile; 392 393 cfg = c; 394 if (GNUNET_OK != parse_wirewatch_config ()) 395 { 396 cfg = NULL; 397 global_ret = EXIT_NOTCONFIGURED; 398 return; 399 } 400 if (GNUNET_OK != 401 GNUNET_CONFIGURATION_get_value_number (cfg, 402 "exchange", 403 "ROUTER_SHARD_SIZE", 404 &ass)) 405 { 406 cfg = NULL; 407 global_ret = EXIT_NOTCONFIGURED; 408 return; 409 } 410 if ( (0 == ass) || 411 (ass > INT32_MAX) ) 412 shard_size = 1U + INT32_MAX; 413 else 414 shard_size = (uint32_t) ass; 415 GNUNET_assert (NULL == task); 416 if (0) 417 { 418 /* FIXME: router is not implemented, remove this once it makes 419 sense to start the DB workload logic... */ 420 task = GNUNET_SCHEDULER_add_now (&run_shard, 421 NULL); 422 } 423 GNUNET_SCHEDULER_add_shutdown (&shutdown_task, 424 cls); 425 } 426 427 428 /** 429 * The main function of the taler-exchange-router. 430 * 431 * @param argc number of arguments from the command line 432 * @param argv command line arguments 433 * @return 0 ok, non-zero on error, see #global_ret 434 */ 435 int 436 main (int argc, 437 char *const *argv) 438 { 439 struct GNUNET_GETOPT_CommandLineOption options[] = { 440 GNUNET_GETOPT_option_timetravel ('T', 441 "timetravel"), 442 GNUNET_GETOPT_option_flag ('t', 443 "test", 444 "run in test mode and exit when idle", 445 &test_mode), 446 GNUNET_GETOPT_option_flag ('y', 447 "kyc-off", 448 "perform wire transfers without KYC checks", 449 &kyc_off), 450 GNUNET_GETOPT_OPTION_END 451 }; 452 enum GNUNET_GenericReturnValue ret; 453 454 ret = GNUNET_PROGRAM_run ( 455 TALER_EXCHANGE_project_data (), 456 argc, argv, 457 "taler-exchange-router", 458 gettext_noop ( 459 "background process that routes P2P transfers"), 460 options, 461 &run, NULL); 462 if (GNUNET_SYSERR == ret) 463 return EXIT_INVALIDARGUMENT; 464 if (GNUNET_NO == ret) 465 return EXIT_SUCCESS; 466 return global_ret; 467 } 468 469 470 /* end of taler-exchange-router.c */