Line data Source code
1 : /*
2 : This file is part of TALER
3 : Copyright (C) 2016--2023 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 : * @file taler-exchange-wirewatch.c
18 : * @brief Process that watches for wire transfers to the exchange's bank account
19 : * @author Christian Grothoff
20 : */
21 : #include "platform.h"
22 : #include <gnunet/gnunet_util_lib.h>
23 : #include <jansson.h>
24 : #include <pthread.h>
25 : #include <microhttpd.h>
26 : #include "exchangedb_lib.h"
27 : #include "taler/taler_json_lib.h"
28 : #include "taler/taler_bank_service.h"
29 : #include "exchange-database/begin_shard.h"
30 : #include "exchange-database/abort_shard.h"
31 : #include "exchange-database/preflight.h"
32 : #include "exchange-database/do_import_credits.h"
33 : #include "exchange-database/event_listen.h"
34 : #include "exchange-database/event_listen_cancel.h"
35 :
36 : /**
37 : * How long to wait for an HTTP reply if there
38 : * are no transactions pending at the server?
39 : */
40 : #define LONGPOLL_TIMEOUT GNUNET_TIME_UNIT_MINUTES
41 :
42 : /**
43 : * What is the maximum batch size we use for credit history
44 : * requests with the bank. See `batch_size` below.
45 : */
46 : #define MAXIMUM_BATCH_SIZE 1024
47 :
48 : /**
49 : * Information about our account.
50 : */
51 : static const struct TALER_EXCHANGEDB_AccountInfo *ai;
52 :
53 : /**
54 : * Active request for history.
55 : */
56 : static struct TALER_BANK_CreditHistoryHandle *hh;
57 :
58 : /**
59 : * Set to true if the request for history did actually
60 : * return transaction items.
61 : */
62 : static bool hh_returned_data;
63 :
64 : /**
65 : * Set to true if the request for history did not
66 : * succeed because the account was unknown.
67 : */
68 : static bool hh_account_404;
69 :
70 : /**
71 : * Set to true if the request for history did not
72 : * succeed because of some unexpected HTTP request error.
73 : */
74 : static bool hh_error;
75 :
76 : /**
77 : * When did we start the last @e hh request?
78 : */
79 : static struct GNUNET_TIME_Absolute hh_start_time;
80 :
81 : /**
82 : * Until when is processing this wire plugin delayed?
83 : */
84 : static struct GNUNET_TIME_Absolute delayed_until;
85 :
86 : /**
87 : * Encoded offset in the wire transfer list from where
88 : * to start the next query with the bank.
89 : */
90 : static uint64_t batch_start;
91 :
92 : /**
93 : * Latest row offset we have imported and committed. Mirrors the
94 : * @c progress_row of our shard in the database.
95 : */
96 : static uint64_t latest_row_off;
97 :
98 : /**
99 : * Offset where our current shard begins (inclusive).
100 : */
101 : static uint64_t shard_start;
102 :
103 : /**
104 : * Offset where our current shard ends (exclusive).
105 : */
106 : static uint64_t shard_end;
107 :
108 : /**
109 : * For how long do we hold the shard? Renewed on every batch we import, so
110 : * that a worker whose account trickles in slowly does not keep losing its
111 : * shard to a second worker that then has nothing to do either.
112 : */
113 : static struct GNUNET_TIME_Relative shard_lease;
114 :
115 : /**
116 : * When did we start with the shard?
117 : */
118 : static struct GNUNET_TIME_Absolute shard_start_time;
119 :
120 : /**
121 : * For how long did we lock the shard?
122 : */
123 : static struct GNUNET_TIME_Absolute shard_end_time;
124 :
125 : /**
126 : * How long did we take to finish the last shard
127 : * for this account?
128 : */
129 : static struct GNUNET_TIME_Relative shard_delay;
130 :
131 : /**
132 : * How long did we take to finish the last shard
133 : * for this account?
134 : */
135 : static struct GNUNET_TIME_Relative longpoll_timeout;
136 :
137 : /**
138 : * How long do we wait on 404.
139 : */
140 : static struct GNUNET_TIME_Relative h404_backoff;
141 :
142 : /**
143 : * How long do we wait on HTTP history request errors.
144 : */
145 : static struct GNUNET_TIME_Relative hh_error_backoff;
146 :
147 : /**
148 : * Name of our job in the shard table.
149 : */
150 : static char *job_name;
151 :
152 : /**
153 : * How many transactions do we retrieve per batch?
154 : */
155 : static unsigned int batch_size;
156 :
157 : /**
158 : * How much do we increment @e batch_size on success?
159 : */
160 : static unsigned int batch_thresh;
161 :
162 : /**
163 : * Did work remain in the transaction queue? Set to true
164 : * if we did some work and thus there might be more.
165 : */
166 : static bool progress;
167 :
168 : /**
169 : * Is this shard still open for processing.
170 : */
171 : static bool shard_open;
172 :
173 : /**
174 : * Handle to the context for interacting with the bank.
175 : */
176 : static struct GNUNET_CURL_Context *ctx;
177 :
178 : /**
179 : * Scheduler context for running the @e ctx.
180 : */
181 : static struct GNUNET_CURL_RescheduleContext *rc;
182 :
183 : /**
184 : * The exchange's configuration (global)
185 : */
186 : static const struct GNUNET_CONFIGURATION_Handle *cfg;
187 :
188 : /**
189 : * Our DB plugin.
190 : */
191 : static struct TALER_EXCHANGEDB_PostgresContext *pg;
192 :
193 : /**
194 : * How long should we sleep when idle before trying to find more work?
195 : * Also used for how long we wait to grab a shard before trying it again.
196 : * The value should be set to a bit above the average time it takes to
197 : * process a shard.
198 : */
199 : static struct GNUNET_TIME_Relative wirewatch_idle_sleep_interval;
200 :
201 : /**
202 : * How long do we sleep on serialization conflicts?
203 : */
204 : static struct GNUNET_TIME_Relative wirewatch_conflict_sleep_interval;
205 :
206 : /**
207 : * Modulus to apply to group shards. The shard size must ultimately be a
208 : * multiple of the batch size. Thus, if this is not a multiple of the
209 : * #MAXIMUM_BATCH_SIZE, the batch size will be set to the #shard_size.
210 : */
211 : static unsigned int shard_size = MAXIMUM_BATCH_SIZE;
212 :
213 : /**
214 : * How many workers should we plan our scheduling with?
215 : */
216 : static unsigned int max_workers = 16;
217 :
218 : /**
219 : * -e command-line option: exit on errors talking to the bank?
220 : */
221 : static int exit_on_error;
222 :
223 : /**
224 : * Value to return from main(). 0 on success, non-zero on
225 : * on serious errors.
226 : */
227 : static int global_ret;
228 :
229 : /**
230 : * Are we run in testing mode and should only do one pass?
231 : */
232 : static int test_mode;
233 :
234 : /**
235 : * Should we ignore if the bank does not know our bank
236 : * account?
237 : */
238 : static int ignore_account_404;
239 :
240 : /**
241 : * Current task waiting for execution, if any.
242 : */
243 : static struct GNUNET_SCHEDULER_Task *task;
244 :
245 : /**
246 : * Name of the configuration section with the account we should watch.
247 : */
248 : static char *account_section;
249 :
250 : /**
251 : * We're being aborted with CTRL-C (or SIGTERM). Shut down.
252 : *
253 : * @param cls closure
254 : */
255 : static void
256 62 : shutdown_task (void *cls)
257 : {
258 : enum GNUNET_DB_QueryStatus qs;
259 : (void) cls;
260 :
261 62 : if (NULL != hh)
262 : {
263 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
264 : "History request cancelled on shutdown\n");
265 0 : TALER_BANK_credit_history_cancel (hh);
266 0 : hh = NULL;
267 : }
268 62 : if (shard_open)
269 : {
270 : /* Everything we imported is committed, and #latest_row_off is recorded in
271 : the shard, so releasing the lease loses no work: whoever picks the
272 : shard up next resumes where we stopped. */
273 62 : qs = TALER_EXCHANGEDB_abort_shard (pg,
274 : job_name,
275 : shard_start,
276 : shard_end);
277 62 : if (qs <= 0)
278 0 : GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
279 : "Failed to abort work shard on shutdown\n");
280 : }
281 62 : GNUNET_free (job_name);
282 62 : if (NULL != ctx)
283 : {
284 62 : GNUNET_CURL_fini (ctx);
285 62 : ctx = NULL;
286 : }
287 62 : if (NULL != rc)
288 : {
289 62 : GNUNET_CURL_gnunet_rc_destroy (rc);
290 62 : rc = NULL;
291 : }
292 62 : if (NULL != task)
293 : {
294 0 : GNUNET_SCHEDULER_cancel (task);
295 0 : task = NULL;
296 : }
297 62 : TALER_EXCHANGEDB_disconnect (pg);
298 62 : pg = NULL;
299 62 : TALER_EXCHANGEDB_unload_accounts ();
300 62 : cfg = NULL;
301 62 : }
302 :
303 :
304 : /**
305 : * Function called with information about a wire account. Adds the
306 : * account to our list (if it is enabled and we can load the plugin).
307 : *
308 : * @param cls closure, NULL
309 : * @param in_ai account information
310 : */
311 : static void
312 172 : add_account_cb (void *cls,
313 : const struct TALER_EXCHANGEDB_AccountInfo *in_ai)
314 : {
315 : (void) cls;
316 172 : if (! in_ai->credit_enabled)
317 0 : return; /* not enabled for us, skip */
318 172 : if ( (NULL != account_section) &&
319 170 : (0 != strcasecmp (in_ai->section_name,
320 : account_section)) )
321 110 : return; /* not enabled for us, skip */
322 62 : if (NULL != ai)
323 : {
324 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
325 : "Multiple accounts enabled (%s and %s), use '-a' command-line option to select one!\n",
326 : ai->section_name,
327 : in_ai->section_name);
328 0 : GNUNET_SCHEDULER_shutdown ();
329 0 : global_ret = EXIT_INVALIDARGUMENT;
330 0 : return;
331 : }
332 62 : ai = in_ai;
333 62 : GNUNET_asprintf (&job_name,
334 : "wirewatch-%s",
335 62 : ai->section_name);
336 62 : batch_size = MAXIMUM_BATCH_SIZE;
337 62 : if (0 != shard_size % batch_size)
338 62 : batch_size = shard_size;
339 : }
340 :
341 :
342 : /**
343 : * Parse configuration parameters for the exchange server into the
344 : * corresponding global variables.
345 : *
346 : * @return EXIT_SUCCESS on success, EXIT_NOTCONFIGURED for invalid settings,
347 : * EXIT_FAILURE for an unavailable dependency
348 : */
349 : static int
350 62 : exchange_serve_process_config (void)
351 : {
352 62 : if (GNUNET_OK !=
353 62 : GNUNET_CONFIGURATION_get_value_time (cfg,
354 : "exchange",
355 : "WIREWATCH_IDLE_SLEEP_INTERVAL",
356 : &wirewatch_idle_sleep_interval))
357 : {
358 0 : GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
359 : "exchange",
360 : "WIREWATCH_IDLE_SLEEP_INTERVAL");
361 0 : return EXIT_NOTCONFIGURED;
362 : }
363 62 : if (NULL ==
364 62 : (pg = TALER_EXCHANGEDB_connect (cfg)))
365 : {
366 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
367 : "Failed to initialize DB subsystem\n");
368 0 : return EXIT_FAILURE;
369 : }
370 62 : if (GNUNET_OK !=
371 62 : TALER_EXCHANGEDB_load_accounts (cfg,
372 : TALER_EXCHANGEDB_ALO_CREDIT
373 : | TALER_EXCHANGEDB_ALO_AUTHDATA))
374 : {
375 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
376 : "No wire accounts configured for credit!\n");
377 0 : return EXIT_NOTCONFIGURED;
378 : }
379 62 : TALER_EXCHANGEDB_find_accounts (&add_account_cb,
380 : NULL);
381 62 : if (NULL == ai)
382 : {
383 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
384 : "No accounts enabled for credit!\n");
385 0 : GNUNET_SCHEDULER_shutdown ();
386 0 : return EXIT_NOTCONFIGURED;
387 : }
388 62 : return EXIT_SUCCESS;
389 : }
390 :
391 :
392 : /**
393 : * Lock a shard and then begin to query for incoming wire transfers.
394 : *
395 : * @param cls NULL
396 : */
397 : static void
398 : lock_shard (void *cls);
399 :
400 :
401 : /**
402 : * Continue with the credit history of the shard.
403 : *
404 : * @param cls NULL
405 : */
406 : static void
407 : continue_with_shard (void *cls);
408 :
409 :
410 : /**
411 : * We encountered a serialization error. The batch that hit it was a single
412 : * statement, so the database has already discarded all of it; there is
413 : * nothing to roll back. Shrink the batch and ask the bank again from the
414 : * last point we committed.
415 : */
416 : static void
417 0 : handle_soft_error (void)
418 : {
419 0 : if (1 < batch_size)
420 : {
421 0 : batch_thresh = batch_size;
422 0 : batch_size /= 2;
423 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
424 : "Reduced batch size to %llu due to serialization issue\n",
425 : (unsigned long long) batch_size);
426 : }
427 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
428 : "Encountered soft error, resetting start point to batch start\n");
429 0 : latest_row_off = batch_start;
430 0 : GNUNET_assert (NULL == task);
431 0 : task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
432 : NULL);
433 0 : }
434 :
435 :
436 : /**
437 : * Schedule the #lock_shard() operation.
438 : */
439 : static void
440 167 : schedule_transfers (void)
441 : {
442 167 : if (shard_open)
443 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
444 : "Will retry my shard (%llu,%llu] of %s in %s\n",
445 : (unsigned long long) shard_start,
446 : (unsigned long long) shard_end,
447 : job_name,
448 : GNUNET_STRINGS_relative_time_to_string (
449 : GNUNET_TIME_absolute_get_remaining (delayed_until),
450 : true));
451 : else
452 167 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
453 : "Will try to lock next shard of %s in %s\n",
454 : job_name,
455 : GNUNET_STRINGS_relative_time_to_string (
456 : GNUNET_TIME_absolute_get_remaining (delayed_until),
457 : true));
458 167 : GNUNET_assert (NULL == task);
459 167 : task = GNUNET_SCHEDULER_add_at (delayed_until,
460 : &lock_shard,
461 : NULL);
462 167 : }
463 :
464 :
465 : /**
466 : * We are done with the work that is possible right now (and the transaction
467 : * was committed, if there was one to commit). Move on to the next shard.
468 : */
469 : static void
470 167 : transaction_completed (void)
471 : {
472 167 : if ( (batch_start + batch_size ==
473 61 : latest_row_off) &&
474 61 : (batch_size < MAXIMUM_BATCH_SIZE) )
475 : {
476 : /* The current batch size worked without serialization
477 : issues, and we are allowed to grow. Do so slowly. */
478 : int delta;
479 :
480 61 : delta = ((int) batch_thresh - (int) batch_size) / 4;
481 61 : if (delta < 0)
482 0 : delta = -delta;
483 61 : batch_size = GNUNET_MIN (MAXIMUM_BATCH_SIZE,
484 : batch_size + delta + 1);
485 61 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
486 : "Increasing batch size to %llu\n",
487 : (unsigned long long) batch_size);
488 : }
489 :
490 167 : if ( (! progress) && test_mode)
491 : {
492 : /* Transaction list was drained and we are in
493 : test mode. So we are done. */
494 62 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
495 : "Transaction list drained and in test mode. Exiting\n");
496 62 : GNUNET_SCHEDULER_shutdown ();
497 62 : return;
498 : }
499 105 : if (! (hh_returned_data || hh_account_404 || hh_error) )
500 : {
501 : /* Enforce long-polling delay even if the server ignored it
502 : and returned earlier */
503 : struct GNUNET_TIME_Relative latency;
504 : struct GNUNET_TIME_Relative left;
505 :
506 0 : latency = GNUNET_TIME_absolute_get_duration (hh_start_time);
507 0 : left = GNUNET_TIME_relative_subtract (longpoll_timeout,
508 : latency);
509 0 : if (! (test_mode ||
510 0 : GNUNET_TIME_relative_is_zero (left)) )
511 0 : GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
512 : "Server did not respect long-polling, enforcing client-side by sleeping for %s\n",
513 : GNUNET_TIME_relative2s (left,
514 : true));
515 0 : delayed_until = GNUNET_TIME_relative_to_absolute (left);
516 : }
517 105 : if (hh_account_404)
518 : {
519 0 : h404_backoff = GNUNET_TIME_STD_BACKOFF (h404_backoff);
520 0 : delayed_until = GNUNET_TIME_relative_to_absolute (
521 : h404_backoff);
522 : }
523 : else
524 : {
525 105 : h404_backoff = GNUNET_TIME_UNIT_ZERO;
526 : }
527 105 : if (hh_error)
528 : {
529 0 : hh_error_backoff = GNUNET_TIME_STD_BACKOFF (hh_error_backoff);
530 0 : delayed_until = GNUNET_TIME_relative_to_absolute (
531 : hh_error_backoff);
532 : }
533 : else
534 : {
535 105 : hh_error_backoff = GNUNET_TIME_UNIT_ZERO;
536 : }
537 105 : if (test_mode)
538 105 : delayed_until = GNUNET_TIME_UNIT_ZERO_ABS;
539 105 : GNUNET_assert (NULL == task);
540 105 : schedule_transfers ();
541 : }
542 :
543 :
544 : /**
545 : * We got incoming transaction details from the bank. Add them
546 : * to the database.
547 : *
548 : * @param details array of transaction details
549 : * @param details_length length of the @a details array
550 : */
551 : static void
552 105 : process_reply (const struct TALER_BANK_CreditDetails *details,
553 : unsigned int details_length)
554 : {
555 : enum GNUNET_DB_QueryStatus qs;
556 : bool shard_done;
557 105 : uint64_t lroff = latest_row_off;
558 :
559 105 : if (0 == details_length)
560 : {
561 : /* Server should have used 204, not 200! */
562 0 : GNUNET_break_op (0);
563 0 : transaction_completed ();
564 0 : return;
565 : }
566 105 : if (details_length > MAXIMUM_BATCH_SIZE)
567 : {
568 : /* We never ask for more than #MAXIMUM_BATCH_SIZE transactions; a bank
569 : returning more would make us allocate unbounded amounts of stack */
570 0 : GNUNET_break_op (0);
571 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
572 : "Bank returned %u transactions, but we asked for at most %u!\n",
573 : details_length,
574 : (unsigned int) MAXIMUM_BATCH_SIZE);
575 0 : GNUNET_SCHEDULER_shutdown ();
576 0 : return;
577 : }
578 105 : hh_returned_data = true;
579 : /* check serial IDs for range constraints */
580 175 : for (unsigned int i = 0; i<details_length; i++)
581 : {
582 105 : const struct TALER_BANK_CreditDetails *cd = &details[i];
583 :
584 105 : if (cd->serial_id < lroff)
585 : {
586 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
587 : "Serial ID %llu not monotonic (got %llu before). Failing!\n",
588 : (unsigned long long) cd->serial_id,
589 : (unsigned long long) lroff);
590 0 : GNUNET_SCHEDULER_shutdown ();
591 0 : return;
592 : }
593 105 : if (cd->serial_id > shard_end)
594 : {
595 : /* we are *past* the current shard (likely because the serial_id of the
596 : shard_end happens to not exist in the DB). So commit and stop this
597 : iteration! */
598 35 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
599 : "Serial ID %llu past shard end at %llu, ending iteration early!\n",
600 : (unsigned long long) cd->serial_id,
601 : (unsigned long long) shard_end);
602 35 : details_length = i;
603 35 : progress = true;
604 35 : lroff = cd->serial_id - 1;
605 35 : break;
606 : }
607 70 : lroff = cd->serial_id;
608 : }
609 105 : {
610 105 : enum GNUNET_DB_QueryStatus qss[GNUNET_NZL (details_length)];
611 105 : struct TALER_EXCHANGEDB_ReserveInInfo reserves[GNUNET_NZL (details_length)];
612 105 : struct TALER_EXCHANGEDB_KycauthInInfo kycauths[GNUNET_NZL (details_length)];
613 105 : struct TALER_EXCHANGEDB_WadInInfo wads[GNUNET_NZL (details_length)];
614 105 : struct TALER_EXCHANGEDB_CreditBatch batch = {
615 105 : .exchange_account_name = ai->section_name,
616 : .reserves = reserves,
617 : .kycauths = kycauths,
618 : .wads = wads,
619 : .job_name = job_name,
620 : .shard_start = shard_start,
621 : .shard_end = shard_end,
622 : .progress_row = lroff,
623 : .lease = shard_lease
624 : };
625 : unsigned int j;
626 :
627 : /* make compiler happy */
628 105 : memset (qss,
629 : 0,
630 : sizeof (qss));
631 105 : if (0 != details_length)
632 70 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
633 : "Importing %u transactions\n",
634 : details_length);
635 175 : for (unsigned int i = 0; i<details_length; i++)
636 : {
637 70 : const struct TALER_BANK_CreditDetails *cd = &details[i];
638 :
639 70 : switch (cd->type)
640 : {
641 54 : case TALER_BANK_CT_RESERVE:
642 : {
643 54 : struct TALER_EXCHANGEDB_ReserveInInfo *res
644 54 : = &reserves[batch.reserves_length++];
645 :
646 54 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
647 : "Importing reserve transfer over %s\n",
648 : TALER_amount2s (&cd->amount));
649 54 : res->reserve_pub = &cd->details.reserve.reserve_pub;
650 54 : res->balance = &cd->amount;
651 54 : res->execution_time = cd->execution_date;
652 54 : res->sender_account_details = cd->debit_account_uri;
653 54 : res->wire_reference = cd->serial_id;
654 : }
655 54 : break;
656 16 : case TALER_BANK_CT_KYCAUTH:
657 : {
658 16 : struct TALER_EXCHANGEDB_KycauthInInfo *ka
659 16 : = &kycauths[batch.kycauths_length++];
660 :
661 16 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
662 : "Importing KYC auth transfer over %s\n",
663 : TALER_amount2s (&cd->amount));
664 16 : ka->account_pub = &cd->details.kycauth.account_pub;
665 16 : ka->balance = &cd->amount;
666 16 : ka->execution_time = cd->execution_date;
667 16 : ka->sender_account_details = cd->debit_account_uri;
668 16 : ka->wire_reference = cd->serial_id;
669 : }
670 16 : break;
671 0 : case TALER_BANK_CT_WAD:
672 : {
673 0 : struct TALER_EXCHANGEDB_WadInInfo *wad
674 0 : = &wads[batch.wads_length++];
675 :
676 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
677 : "Importing WAD transfer over %s\n",
678 : TALER_amount2s (&cd->amount));
679 0 : wad->wad_id = &cd->details.wad.wad_id;
680 0 : wad->origin_exchange_url = cd->details.wad.origin_exchange_url;
681 0 : wad->balance = &cd->amount;
682 0 : wad->execution_time = cd->execution_date;
683 : }
684 0 : break;
685 : }
686 : }
687 : /* One statement, hence one transaction: these transfers and the record of
688 : how far this shard has come become visible together. That is what lets
689 : us hand the money to the wallets now instead of at the end of the
690 : shard, without risking a crash that leaves the shard claiming work it
691 : never did. */
692 105 : qs = TALER_EXCHANGEDB_do_import_credits (pg,
693 : &batch,
694 : qss);
695 105 : switch (qs)
696 : {
697 0 : case GNUNET_DB_STATUS_HARD_ERROR:
698 0 : GNUNET_break (0);
699 0 : GNUNET_SCHEDULER_shutdown ();
700 0 : return;
701 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
702 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
703 : "Got DB soft error importing %u transactions. Retrying.\n",
704 : details_length);
705 0 : handle_soft_error ();
706 0 : return;
707 105 : default:
708 105 : break;
709 : }
710 105 : j = 0;
711 175 : for (unsigned int i = 0; i<details_length; i++)
712 : {
713 70 : const struct TALER_BANK_CreditDetails *cd = &details[i];
714 :
715 70 : if (TALER_BANK_CT_RESERVE != cd->type)
716 16 : continue; /* only reserve transfers report a per-row status */
717 54 : switch (qss[j++])
718 : {
719 0 : case GNUNET_DB_STATUS_HARD_ERROR:
720 : case GNUNET_DB_STATUS_SOFT_ERROR:
721 0 : GNUNET_break (0); /* handled above, for the batch as a whole */
722 0 : GNUNET_SCHEDULER_shutdown ();
723 0 : return;
724 0 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
725 : /* Either wirewatch was freshly started after the system was
726 : shutdown and we're going over an incomplete shard again
727 : after being restarted, or the shard lock period was too
728 : short (number of workers set incorrectly?) and a 2nd
729 : wirewatcher has been stealing our work while we are still
730 : at it. */
731 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
732 : "Attempted to import transaction %llu (%s) twice. "
733 : "This should happen rarely (if not, ask for support).\n",
734 : (unsigned long long) cd->serial_id,
735 : job_name);
736 0 : break;
737 54 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
738 54 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
739 : "Imported transaction %llu.\n",
740 : (unsigned long long) cd->serial_id);
741 : /* normal case */
742 54 : progress = true;
743 54 : break;
744 : }
745 : }
746 : }
747 :
748 105 : latest_row_off = lroff;
749 : /* The same statement renewed our lease in the database, so keep the local
750 : deadline in step or we would go re-acquire a shard we still hold. */
751 105 : shard_end_time = GNUNET_TIME_relative_to_absolute (shard_lease);
752 105 : shard_done = (shard_end <= latest_row_off);
753 105 : if (shard_done)
754 : {
755 : /* The shard was marked completed by the very statement that imported the
756 : last of its transfers; there is nothing left to write. */
757 105 : progress = true;
758 105 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
759 : "Completed shard %s (%llu,%llu] after %s\n",
760 : job_name,
761 : (unsigned long long) shard_start,
762 : (unsigned long long) shard_end,
763 : GNUNET_STRINGS_relative_time_to_string (
764 : GNUNET_TIME_absolute_get_duration (shard_start_time),
765 : true));
766 105 : shard_delay = GNUNET_TIME_absolute_get_duration (shard_start_time);
767 105 : shard_open = false;
768 105 : transaction_completed ();
769 105 : return;
770 : }
771 0 : GNUNET_assert (NULL == task);
772 0 : task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
773 : NULL);
774 : }
775 :
776 :
777 : /**
778 : * Callbacks of this type are used to serve the result of asking
779 : * the bank for the transaction history.
780 : *
781 : * @param cls NULL
782 : * @param reply response we got from the bank
783 : */
784 : static void
785 167 : history_cb (void *cls,
786 : const struct TALER_BANK_CreditHistoryResponse *reply)
787 : {
788 : (void) cls;
789 167 : GNUNET_assert (NULL == task);
790 167 : hh = NULL;
791 167 : GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
792 : "History request returned with HTTP status %u\n",
793 : reply->http_status);
794 167 : switch (reply->http_status)
795 : {
796 105 : case MHD_HTTP_OK:
797 105 : process_reply (reply->details.ok.details,
798 105 : reply->details.ok.details_length);
799 105 : return;
800 61 : case MHD_HTTP_NO_CONTENT:
801 61 : transaction_completed ();
802 61 : return;
803 1 : case MHD_HTTP_NOT_FOUND:
804 1 : hh_account_404 = true;
805 1 : if (ignore_account_404)
806 : {
807 0 : transaction_completed ();
808 0 : return;
809 : }
810 1 : break;
811 0 : default:
812 0 : hh_error = true;
813 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
814 : "Error fetching history: %s (%u)\n",
815 : TALER_ErrorCode_get_hint (reply->ec),
816 : reply->http_status);
817 0 : break;
818 : }
819 1 : if (! exit_on_error)
820 : {
821 1 : transaction_completed ();
822 1 : return;
823 : }
824 0 : GNUNET_SCHEDULER_shutdown ();
825 : }
826 :
827 :
828 : static void
829 167 : continue_with_shard (void *cls)
830 : {
831 : unsigned int limit;
832 :
833 : (void) cls;
834 167 : task = NULL;
835 167 : GNUNET_assert (shard_end > latest_row_off);
836 167 : limit = GNUNET_MIN (batch_size,
837 : shard_end - latest_row_off);
838 : /* Where this batch starts out: the point we last committed, and thus where
839 : #handle_soft_error() rewinds to and what #transaction_completed() measures
840 : the batch against. */
841 167 : batch_start = latest_row_off;
842 167 : GNUNET_assert (NULL == hh);
843 167 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
844 : "Requesting credit history starting from %llu\n",
845 : (unsigned long long) latest_row_off);
846 167 : hh_start_time = GNUNET_TIME_absolute_get ();
847 167 : hh_returned_data = false;
848 167 : hh_account_404 = false;
849 167 : hh_error = false;
850 167 : hh = TALER_BANK_credit_history (ctx,
851 167 : ai->auth,
852 : latest_row_off,
853 : limit,
854 : test_mode
855 167 : ? GNUNET_TIME_UNIT_ZERO
856 : : longpoll_timeout,
857 : &history_cb,
858 : NULL);
859 167 : if (NULL == hh)
860 : {
861 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
862 : "Failed to start request for account history!\n");
863 0 : global_ret = EXIT_FAILURE;
864 0 : GNUNET_SCHEDULER_shutdown ();
865 0 : return;
866 : }
867 : }
868 :
869 :
870 : /**
871 : * Reserve a shard for us to work on.
872 : *
873 : * @param cls NULL
874 : */
875 : static void
876 167 : lock_shard (void *cls)
877 : {
878 : enum GNUNET_DB_QueryStatus qs;
879 : struct GNUNET_TIME_Relative delay;
880 : uint64_t progress_row;
881 :
882 : (void) cls;
883 167 : task = NULL;
884 167 : if (GNUNET_SYSERR ==
885 167 : TALER_EXCHANGEDB_preflight (pg))
886 : {
887 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
888 : "Failed to obtain database connection!\n");
889 0 : global_ret = EXIT_FAILURE;
890 0 : GNUNET_SCHEDULER_shutdown ();
891 0 : return;
892 : }
893 167 : if ( (shard_open) &&
894 0 : (GNUNET_TIME_absolute_is_future (shard_end_time)) )
895 : {
896 0 : progress = false;
897 0 : task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
898 : NULL);
899 0 : return;
900 : }
901 167 : if (shard_open)
902 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
903 : "Shard not completed in time, will try to re-acquire\n");
904 : /* How long we lock a shard depends on the number of
905 : workers expected, and how long we usually took to
906 : process a shard. */
907 167 : if (0 == max_workers)
908 167 : delay = GNUNET_TIME_UNIT_ZERO;
909 : else
910 0 : delay.rel_value_us = GNUNET_CRYPTO_random_u64 (
911 0 : 4 * GNUNET_TIME_relative_max (
912 : wirewatch_idle_sleep_interval,
913 : GNUNET_TIME_relative_multiply (shard_delay,
914 0 : max_workers)).rel_value_us);
915 167 : shard_start_time = GNUNET_TIME_absolute_get ();
916 167 : shard_lease = delay;
917 167 : qs = TALER_EXCHANGEDB_begin_shard (pg,
918 : job_name,
919 : delay,
920 : shard_size,
921 : &shard_start,
922 : &shard_end,
923 : &progress_row);
924 167 : switch (qs)
925 : {
926 0 : case GNUNET_DB_STATUS_HARD_ERROR:
927 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
928 : "Failed to obtain starting point for monitoring from database!\n");
929 0 : global_ret = EXIT_FAILURE;
930 0 : GNUNET_SCHEDULER_shutdown ();
931 0 : return;
932 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
933 : /* try again */
934 : {
935 : struct GNUNET_TIME_Relative rdelay;
936 :
937 : wirewatch_conflict_sleep_interval
938 0 : = GNUNET_TIME_STD_BACKOFF (wirewatch_conflict_sleep_interval);
939 0 : rdelay = GNUNET_TIME_randomize (wirewatch_conflict_sleep_interval);
940 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
941 : "Serialization error tying to obtain shard %s, will try again in %s!\n",
942 : job_name,
943 : GNUNET_STRINGS_relative_time_to_string (rdelay,
944 : true));
945 : #if 1
946 0 : if (GNUNET_TIME_relative_cmp (rdelay,
947 : >,
948 : GNUNET_TIME_UNIT_SECONDS))
949 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
950 : "Delay would have been for %s\n",
951 : GNUNET_TIME_relative2s (rdelay,
952 : true));
953 0 : rdelay = GNUNET_TIME_relative_min (rdelay,
954 : GNUNET_TIME_UNIT_SECONDS);
955 : #endif
956 0 : delayed_until = GNUNET_TIME_relative_to_absolute (rdelay);
957 : }
958 0 : GNUNET_assert (NULL == task);
959 0 : schedule_transfers ();
960 0 : return;
961 0 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
962 0 : GNUNET_break (0);
963 0 : GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
964 : "No shard available, will try again for %s in %s!\n",
965 : job_name,
966 : GNUNET_STRINGS_relative_time_to_string (
967 : wirewatch_idle_sleep_interval,
968 : true));
969 0 : delayed_until = GNUNET_TIME_relative_to_absolute (
970 : wirewatch_idle_sleep_interval);
971 0 : shard_open = false;
972 0 : GNUNET_assert (NULL == task);
973 0 : schedule_transfers ();
974 0 : return;
975 167 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
976 : /* continued below */
977 167 : wirewatch_conflict_sleep_interval = GNUNET_TIME_UNIT_ZERO;
978 167 : break;
979 : }
980 167 : shard_end_time = GNUNET_TIME_relative_to_absolute (delay);
981 167 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
982 : "Starting with shard %s at (%llu,%llu] locked for %s\n",
983 : job_name,
984 : (unsigned long long) shard_start,
985 : (unsigned long long) shard_end,
986 : GNUNET_STRINGS_relative_time_to_string (delay,
987 : true));
988 167 : progress = false;
989 : /* The shard itself says where to resume. Whether this is a shard we had
990 : before, one abandoned by another worker, or a brand new one no longer
991 : matters: everything below progress_row is imported and committed, and
992 : nothing above it is. */
993 167 : if (progress_row != shard_start)
994 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
995 : "Continuing from %llu\n",
996 : (unsigned long long) progress_row);
997 167 : latest_row_off = progress_row;
998 167 : batch_start = progress_row;
999 167 : shard_open = true;
1000 167 : task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
1001 : NULL);
1002 : }
1003 :
1004 :
1005 : /**
1006 : * First task.
1007 : *
1008 : * @param cls closure, NULL
1009 : * @param args remaining command-line arguments
1010 : * @param cfgfile name of the configuration file used (for saving, can be NULL!)
1011 : * @param c configuration
1012 : */
1013 : static void
1014 62 : run (void *cls,
1015 : char *const *args,
1016 : const char *cfgfile,
1017 : const struct GNUNET_CONFIGURATION_Handle *c)
1018 : {
1019 : (void) cls;
1020 : (void) args;
1021 : (void) cfgfile;
1022 :
1023 62 : cfg = c;
1024 62 : GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
1025 : cls);
1026 62 : if (0 == shard_size)
1027 : {
1028 : /* Would give us empty shards, and thus fail the assertion in
1029 : #continue_with_shard(). */
1030 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1031 : "Shard size (-S) must be positive\n");
1032 0 : global_ret = EXIT_INVALIDARGUMENT;
1033 0 : GNUNET_SCHEDULER_shutdown ();
1034 0 : return;
1035 : }
1036 62 : if (EXIT_SUCCESS !=
1037 62 : (global_ret = exchange_serve_process_config ()))
1038 : {
1039 0 : GNUNET_SCHEDULER_shutdown ();
1040 0 : return;
1041 : }
1042 62 : ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule,
1043 : &rc);
1044 62 : if (NULL == ctx)
1045 : {
1046 0 : GNUNET_break (0);
1047 0 : GNUNET_SCHEDULER_shutdown ();
1048 0 : global_ret = EXIT_FAILURE;
1049 0 : return;
1050 : }
1051 62 : rc = GNUNET_CURL_gnunet_rc_create (ctx);
1052 62 : schedule_transfers ();
1053 : }
1054 :
1055 :
1056 : /**
1057 : * The main function of taler-exchange-wirewatch
1058 : *
1059 : * @param argc number of arguments from the command line
1060 : * @param argv command line arguments
1061 : * @return 0 ok, non-zero on error
1062 : */
1063 : int
1064 62 : main (int argc,
1065 : char *const *argv)
1066 : {
1067 62 : struct GNUNET_GETOPT_CommandLineOption options[] = {
1068 62 : GNUNET_GETOPT_option_string ('a',
1069 : "account",
1070 : "SECTION_NAME",
1071 : "name of the configuration section with the account we should watch (needed if more than one is enabled for crediting)",
1072 : &account_section),
1073 62 : GNUNET_GETOPT_option_flag ('e',
1074 : "exit-on-error",
1075 : "terminate wirewatch if we failed to download information from the bank",
1076 : &exit_on_error),
1077 62 : GNUNET_GETOPT_option_relative_time ('f',
1078 : "longpoll-timeout",
1079 : "DELAY",
1080 : "what is the timeout when asking the bank about new transactions, specify with unit (e.g. --longpoll-timeout=30s)",
1081 : &longpoll_timeout),
1082 62 : GNUNET_GETOPT_option_flag ('I',
1083 : "ignore-not-found",
1084 : "continue, even if the bank account of the exchange was not found",
1085 : &ignore_account_404),
1086 62 : GNUNET_GETOPT_option_uint ('S',
1087 : "size",
1088 : "SIZE",
1089 : "Size to process per shard (default: 1024)",
1090 : &shard_size),
1091 62 : GNUNET_GETOPT_option_timetravel ('T',
1092 : "timetravel"),
1093 62 : GNUNET_GETOPT_option_flag ('t',
1094 : "test",
1095 : "run in test mode and exit when idle",
1096 : &test_mode),
1097 62 : GNUNET_GETOPT_option_uint ('w',
1098 : "workers",
1099 : "COUNT",
1100 : "Plan work load with up to COUNT worker processes (default: 16)",
1101 : &max_workers),
1102 62 : GNUNET_GETOPT_option_version (VERSION),
1103 : GNUNET_GETOPT_OPTION_END
1104 : };
1105 : enum GNUNET_GenericReturnValue ret;
1106 :
1107 62 : longpoll_timeout = LONGPOLL_TIMEOUT;
1108 62 : ret = GNUNET_PROGRAM_run (
1109 : TALER_EXCHANGE_project_data (),
1110 : argc, argv,
1111 : "taler-exchange-wirewatch",
1112 : gettext_noop (
1113 : "background process that watches for incoming wire transfers from customers"),
1114 : options,
1115 : &run, NULL);
1116 62 : if (GNUNET_SYSERR == ret)
1117 0 : return EXIT_NOTCONFIGURED;
1118 62 : if (GNUNET_NO == ret)
1119 0 : return EXIT_SUCCESS;
1120 62 : return global_ret;
1121 : }
1122 :
1123 :
1124 : /* end of taler-exchange-wirewatch.c */
|
