taler-exchange-expire.c (15519B)
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-expire.c 19 * @brief Process that cleans up expired purses 20 * @author Christian Grothoff 21 */ 22 #include "platform.h" 23 #include <gnunet/gnunet_util_lib.h> 24 #include <jansson.h> 25 #include <pthread.h> 26 #include "exchangedb_lib.h" 27 #include "taler/taler_json_lib.h" 28 #include "taler/taler_bank_service.h" 29 #include "exchange-database/start.h" 30 #include "exchange-database/preflight.h" 31 #include "exchange-database/commit.h" 32 #include "exchange-database/rollback.h" 33 #include "exchange-database/update_shard_progress.h" 34 #include "exchange-database/abort_shard.h" 35 #include "exchange-database/begin_shard.h" 36 #include "exchange-database/do_expire_purse.h" 37 #include "exchange-database/start.h" 38 39 40 /** 41 * Work shard we are processing. 42 */ 43 struct Shard 44 { 45 46 /** 47 * When did we start processing the shard? 48 */ 49 struct GNUNET_TIME_Timestamp start_time; 50 51 /** 52 * Starting row of the shard. 53 */ 54 struct GNUNET_TIME_Absolute shard_start; 55 56 /** 57 * Inclusive end row of the shard. 58 */ 59 struct GNUNET_TIME_Absolute shard_end; 60 61 /** 62 * How far into the shard we have come. Purses that expired before this 63 * are dealt with and committed; a worker taking the shard over resumes 64 * here instead of at @e shard_start. 65 */ 66 struct GNUNET_TIME_Absolute progress; 67 68 /** 69 * For how long we hold the shard. Renewed whenever we record progress. 70 */ 71 struct GNUNET_TIME_Relative lease; 72 73 /** 74 * Number of starting points found in the shard. 75 */ 76 uint64_t work_counter; 77 78 }; 79 80 81 /** 82 * The exchange's configuration. 83 */ 84 static const struct GNUNET_CONFIGURATION_Handle *cfg; 85 86 /** 87 * Our database plugin. 88 */ 89 static struct TALER_EXCHANGEDB_PostgresContext *pg; 90 91 /** 92 * Next task to run, if any. 93 */ 94 static struct GNUNET_SCHEDULER_Task *task; 95 96 /** 97 * How big are the shards we are processing? Is an inclusive offset, so every 98 * shard ranges from [X,X+shard_size) exclusive. So a shard covers 99 * shard_size slots. 100 */ 101 static struct GNUNET_TIME_Relative shard_size; 102 103 /** 104 * Value to return from main(). 0 on success, non-zero on errors. 105 */ 106 static int global_ret; 107 108 /** 109 * #GNUNET_YES if we are in test mode and should exit when idle. 110 */ 111 static int test_mode; 112 113 /** 114 * If this is a first-time run, we immediately 115 * try to catch up with the present. 116 */ 117 static bool jump_mode; 118 119 120 /** 121 * Select a shard to work on. 122 * 123 * @param cls NULL 124 */ 125 static void 126 run_shard (void *cls); 127 128 129 /** 130 * We're being aborted with CTRL-C (or SIGTERM). Shut down. 131 * 132 * @param cls closure 133 */ 134 static void 135 shutdown_task (void *cls) 136 { 137 (void) cls; 138 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 139 "Running shutdown\n"); 140 if (NULL != task) 141 { 142 GNUNET_SCHEDULER_cancel (task); 143 task = NULL; 144 } 145 TALER_EXCHANGEDB_disconnect (pg); 146 pg = NULL; 147 cfg = NULL; 148 } 149 150 151 /** 152 * Parse the configuration for expire. 153 * 154 * @return #GNUNET_OK on success 155 */ 156 static enum GNUNET_GenericReturnValue 157 parse_expire_config (void) 158 { 159 if (NULL == 160 (pg = TALER_EXCHANGEDB_connect (cfg))) 161 { 162 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 163 "Failed to initialize DB subsystem\n"); 164 return GNUNET_SYSERR; 165 } 166 return GNUNET_OK; 167 } 168 169 170 /** 171 * Perform a database commit. If it fails, print a warning. 172 * 173 * @return status of commit 174 */ 175 static enum GNUNET_DB_QueryStatus 176 commit_or_warn (void) 177 { 178 enum GNUNET_DB_QueryStatus qs; 179 180 qs = TALER_EXCHANGEDB_commit (pg); 181 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 182 return qs; 183 GNUNET_log ((GNUNET_DB_STATUS_SOFT_ERROR == qs) 184 ? GNUNET_ERROR_TYPE_INFO 185 : GNUNET_ERROR_TYPE_ERROR, 186 "Failed to commit database transaction!\n"); 187 return qs; 188 } 189 190 191 /** 192 * Record how far we have come within shard @a s, and renew our lease on it. 193 * Must be called from inside the transaction that persists the work being 194 * reported, so that the two land together. Marking the shard completed is 195 * the same statement, so there is no separate "shard done" write. 196 * 197 * @param s shard we are working on 198 * @param progress point in the shard we have reached 199 * @return transaction status code 200 */ 201 static enum GNUNET_DB_QueryStatus 202 record_progress (const struct Shard *s, 203 struct GNUNET_TIME_Absolute progress) 204 { 205 return TALER_EXCHANGEDB_update_shard_progress ( 206 pg, 207 "expire", 208 s->shard_start.abs_value_us, 209 s->shard_end.abs_value_us, 210 progress.abs_value_us, 211 s->lease); 212 } 213 214 215 /** 216 * We are done with shard @a s. Its completion has already been committed by 217 * #record_progress(), so this only reports and releases memory. 218 * 219 * @param[in] s shard to free (and memory to release) 220 */ 221 static void 222 release_shard (struct Shard *s) 223 { 224 unsigned long long wc = (unsigned long long) s->work_counter; 225 226 GNUNET_free (s); 227 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 228 "Purse expiration shard completed with %llu purses\n", 229 wc); 230 if ( (0 == wc) && 231 (test_mode) && 232 (! jump_mode) ) 233 { 234 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 235 "In test-mode without work. Terminating.\n"); 236 GNUNET_SCHEDULER_shutdown (); 237 return; 238 } 239 } 240 241 242 /** 243 * Release lock on shard @a s in the database due to an abort of the 244 * operation. On error, terminates this process. 245 * 246 * @param[in] s shard to free (and memory to release) 247 */ 248 static void 249 abort_shard (struct Shard *s) 250 { 251 enum GNUNET_DB_QueryStatus qs; 252 253 qs = TALER_EXCHANGEDB_abort_shard (pg, 254 "expire", 255 s->shard_start.abs_value_us, 256 s->shard_end.abs_value_us); 257 switch (qs) 258 { 259 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 260 /* good case, handled below */ 261 break; 262 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 263 /* Somebody else already reclaimed the shard; not our problem, and 264 certainly not a reason to terminate. */ 265 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 266 "Shard to abort was already released\n"); 267 GNUNET_free (s); 268 return; 269 case GNUNET_DB_STATUS_SOFT_ERROR: 270 /* This is reached from the soft-error path of run_shard(), i.e. exactly 271 when contention is highest. The shard lock times out on its own, so 272 shutting the daemon down here does more harm than good, 273 let's just continue. */ 274 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 275 "Failed to abort shard due to serialization failure; the " 276 "lock will time out on its own\n"); 277 GNUNET_free (s); 278 return; 279 case GNUNET_DB_STATUS_HARD_ERROR: 280 /* Serious database issue, probably best to exit */ 281 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 282 "Failed to abort shard (%d)!\n", 283 qs); 284 global_ret = EXIT_FAILURE; 285 GNUNET_SCHEDULER_shutdown (); 286 return; 287 } 288 GNUNET_free (s); 289 } 290 291 292 /** 293 * Main function that processes the work in one shard. 294 * 295 * @param[in] cls a `struct Shard` to process 296 */ 297 static void 298 run_expire (void *cls) 299 { 300 struct Shard *s = cls; 301 enum GNUNET_DB_QueryStatus qs; 302 struct GNUNET_TIME_Absolute purse_expiration; 303 304 task = NULL; 305 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 306 "Checking for expired purses\n"); 307 if (GNUNET_SYSERR == 308 TALER_EXCHANGEDB_preflight (pg)) 309 { 310 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 311 "Failed to obtain database connection!\n"); 312 abort_shard (s); 313 global_ret = EXIT_FAILURE; 314 GNUNET_SCHEDULER_shutdown (); 315 return; 316 } 317 if (GNUNET_OK != 318 TALER_EXCHANGEDB_start (pg, 319 "expire-purse")) 320 { 321 GNUNET_break (0); 322 TALER_EXCHANGEDB_rollback (pg); 323 abort_shard (s); 324 global_ret = EXIT_FAILURE; 325 GNUNET_SCHEDULER_shutdown (); 326 return; 327 } 328 qs = TALER_EXCHANGEDB_do_expire_purse (pg, 329 s->progress, 330 s->shard_end, 331 &purse_expiration); 332 switch (qs) 333 { 334 case GNUNET_DB_STATUS_HARD_ERROR: 335 GNUNET_break (0); 336 TALER_EXCHANGEDB_rollback (pg); 337 abort_shard (s); 338 global_ret = EXIT_FAILURE; 339 GNUNET_SCHEDULER_shutdown (); 340 return; 341 case GNUNET_DB_STATUS_SOFT_ERROR: 342 TALER_EXCHANGEDB_rollback (pg); 343 abort_shard (s); 344 GNUNET_assert (NULL == task); 345 task = GNUNET_SCHEDULER_add_now (&run_shard, 346 NULL); 347 return; 348 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 349 /* Nothing left in this shard: record it as done in the very transaction 350 we are about to commit, instead of in a second one afterwards. */ 351 if ( (0 > record_progress (s, 352 s->shard_end)) || 353 (0 > commit_or_warn ()) ) 354 { 355 TALER_EXCHANGEDB_rollback (pg); 356 abort_shard (s); 357 } 358 else 359 { 360 release_shard (s); 361 } 362 GNUNET_assert (NULL == task); 363 task = GNUNET_SCHEDULER_add_now (&run_shard, 364 NULL); 365 return; 366 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 367 /* commit, and go again immediately */ 368 s->work_counter++; 369 if ( (0 > record_progress (s, 370 purse_expiration)) || 371 (0 > commit_or_warn ()) ) 372 { 373 /* Nothing landed, so leave our resume point where it was: moving it 374 here would skip the purse we just failed to decide. */ 375 TALER_EXCHANGEDB_rollback (pg); 376 abort_shard (s); 377 GNUNET_assert (NULL == task); 378 task = GNUNET_SCHEDULER_add_now (&run_shard, 379 NULL); 380 return; 381 } 382 /* Purses expiring in the same microsecond are looked at again next 383 round; deciding one twice is a no-op, so resuming at (not after) 384 this timestamp is the safe choice. */ 385 s->progress = purse_expiration; 386 GNUNET_assert (NULL == task); 387 task = GNUNET_SCHEDULER_add_now (&run_expire, 388 s); 389 } 390 } 391 392 393 /** 394 * Select a shard to work on. 395 * 396 * @param cls NULL 397 */ 398 static void 399 run_shard (void *cls) 400 { 401 struct Shard *s; 402 enum GNUNET_DB_QueryStatus qs; 403 404 (void) cls; 405 task = NULL; 406 if (GNUNET_SYSERR == 407 TALER_EXCHANGEDB_preflight (pg)) 408 { 409 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 410 "Failed to obtain database connection!\n"); 411 global_ret = EXIT_FAILURE; 412 GNUNET_SCHEDULER_shutdown (); 413 return; 414 } 415 s = GNUNET_new (struct Shard); 416 s->start_time = GNUNET_TIME_timestamp_get (); 417 s->lease = shard_size; 418 qs = TALER_EXCHANGEDB_begin_shard (pg, 419 "expire", 420 shard_size, 421 jump_mode 422 ? GNUNET_TIME_absolute_subtract ( 423 GNUNET_TIME_absolute_get (), 424 shard_size). 425 abs_value_us 426 : shard_size.rel_value_us, 427 &s->shard_start.abs_value_us, 428 &s->shard_end.abs_value_us, 429 &s->progress.abs_value_us); 430 jump_mode = false; 431 if (0 >= qs) 432 { 433 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 434 { 435 static struct GNUNET_TIME_Relative delay; 436 437 GNUNET_free (s); 438 delay = GNUNET_TIME_randomized_backoff (delay, 439 GNUNET_TIME_UNIT_SECONDS); 440 GNUNET_assert (NULL == task); 441 task = GNUNET_SCHEDULER_add_delayed (delay, 442 &run_shard, 443 NULL); 444 return; 445 } 446 GNUNET_free (s); 447 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 448 "Failed to begin shard (%d)!\n", 449 qs); 450 GNUNET_break (GNUNET_DB_STATUS_HARD_ERROR != qs); 451 global_ret = EXIT_FAILURE; 452 GNUNET_SCHEDULER_shutdown (); 453 return; 454 } 455 if (GNUNET_TIME_absolute_is_future (s->shard_end)) 456 { 457 abort_shard (s); 458 if (test_mode) 459 { 460 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 461 "In test-mode without work. Terminating.\n"); 462 GNUNET_SCHEDULER_shutdown (); 463 return; 464 } 465 GNUNET_assert (NULL == task); 466 task = GNUNET_SCHEDULER_add_at (s->shard_end, 467 &run_shard, 468 NULL); 469 return; 470 } 471 /* If this is a first-time run, we immediately 472 try to catch up with the present */ 473 if (GNUNET_TIME_absolute_is_zero (s->shard_start)) 474 jump_mode = true; 475 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 476 "Starting shard [%llu:%llu)!\n", 477 (unsigned long long) s->shard_start.abs_value_us, 478 (unsigned long long) s->shard_end.abs_value_us); 479 GNUNET_assert (NULL == task); 480 task = GNUNET_SCHEDULER_add_now (&run_expire, 481 s); 482 } 483 484 485 /** 486 * First task. 487 * 488 * @param cls closure, NULL 489 * @param args remaining command-line arguments 490 * @param cfgfile name of the configuration file used (for saving, can be NULL!) 491 * @param c configuration 492 */ 493 static void 494 run (void *cls, 495 char *const *args, 496 const char *cfgfile, 497 const struct GNUNET_CONFIGURATION_Handle *c) 498 { 499 (void) cls; 500 (void) args; 501 (void) cfgfile; 502 503 cfg = c; 504 if (GNUNET_OK != parse_expire_config ()) 505 { 506 cfg = NULL; 507 global_ret = EXIT_NOTCONFIGURED; 508 return; 509 } 510 if (GNUNET_OK != 511 GNUNET_CONFIGURATION_get_value_time (cfg, 512 "exchange", 513 "EXPIRE_SHARD_SIZE", 514 &shard_size)) 515 { 516 cfg = NULL; 517 global_ret = EXIT_NOTCONFIGURED; 518 return; 519 } 520 GNUNET_assert (NULL == task); 521 task = GNUNET_SCHEDULER_add_now (&run_shard, 522 NULL); 523 GNUNET_SCHEDULER_add_shutdown (&shutdown_task, 524 cls); 525 } 526 527 528 /** 529 * The main function of the taler-exchange-expire. 530 * 531 * @param argc number of arguments from the command line 532 * @param argv command line arguments 533 * @return 0 ok, non-zero on error, see #global_ret 534 */ 535 int 536 main (int argc, 537 char *const *argv) 538 { 539 struct GNUNET_GETOPT_CommandLineOption options[] = { 540 GNUNET_GETOPT_option_timetravel ('T', 541 "timetravel"), 542 GNUNET_GETOPT_option_flag ('t', 543 "test", 544 "run in test mode and exit when idle", 545 &test_mode), 546 GNUNET_GETOPT_OPTION_END 547 }; 548 enum GNUNET_GenericReturnValue ret; 549 550 ret = GNUNET_PROGRAM_run ( 551 TALER_EXCHANGE_project_data (), 552 argc, argv, 553 "taler-exchange-expire", 554 gettext_noop ( 555 "background process that expires purses"), 556 options, 557 &run, NULL); 558 if (GNUNET_SYSERR == ret) 559 return EXIT_INVALIDARGUMENT; 560 if (GNUNET_NO == ret) 561 return EXIT_SUCCESS; 562 return global_ret; 563 } 564 565 566 /* end of taler-exchange-expire.c */