Line data Source code
1 : /*
2 : This file is part of TALER
3 : Copyright (C) 2024, 2025 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 src/backend/taler-merchant-depositcheck.c
18 : * @brief Process that inquires with the exchange for deposits that should have been wired
19 : * @author Christian Grothoff
20 : */
21 : #include "platform.h"
22 : struct ExchangeInteraction;
23 : #define TALER_EXCHANGE_GET_DEPOSITS_RESULT_CLOSURE struct ExchangeInteraction
24 : #include "microhttpd.h"
25 : #include <gnunet/gnunet_util_lib.h>
26 : #include <jansson.h>
27 : #include <pthread.h>
28 : #include <taler/taler_dbevents.h>
29 : #include <taler/taler_exchange_service.h>
30 : #include "taler/taler_merchant_util.h"
31 : #include "merchantdb_lib.h"
32 : #include "merchant-database/event_listen.h"
33 : #include "merchant-database/iterate_pending_deposits.h"
34 : #include "merchant-database/get_exchange_keys.h"
35 : #include "merchant-database/preflight.h"
36 : #include "merchant-database/insert_kyc_failure.h"
37 : #include "merchant-database/set_instance.h"
38 : #include "merchant-database/insert_deposit_to_transfer.h"
39 : #include "merchant-database/update_deposit_settlement_status.h"
40 : #include "merchant-database/start.h"
41 :
42 : /**
43 : * How many requests do we make at most in parallel to the same exchange?
44 : */
45 : #define CONCURRENCY_LIMIT 32
46 :
47 : /**
48 : * How long do we not try a deposit check if the deposit
49 : * was put on hold due to a KYC/AML block?
50 : */
51 : #define KYC_RETRY_DELAY GNUNET_TIME_UNIT_HOURS
52 :
53 : /**
54 : * Information we keep per exchange.
55 : */
56 : struct Child
57 : {
58 :
59 : /**
60 : * Kept in a DLL.
61 : */
62 : struct Child *next;
63 :
64 : /**
65 : * Kept in a DLL.
66 : */
67 : struct Child *prev;
68 :
69 : /**
70 : * The child process.
71 : */
72 : struct GNUNET_Process *process;
73 :
74 : /**
75 : * Wait handle.
76 : */
77 : struct GNUNET_ChildWaitHandle *cwh;
78 :
79 : /**
80 : * Which exchange is this state for?
81 : */
82 : char *base_url;
83 :
84 : /**
85 : * Task to restart the child.
86 : */
87 : struct GNUNET_SCHEDULER_Task *rt;
88 :
89 : /**
90 : * When should the child be restarted at the earliest?
91 : */
92 : struct GNUNET_TIME_Absolute next_start;
93 :
94 : /**
95 : * Current minimum delay between restarts, grows
96 : * exponentially if child exits before this time.
97 : */
98 : struct GNUNET_TIME_Relative rd;
99 :
100 : };
101 :
102 :
103 : /**
104 : * Information we keep per exchange interaction.
105 : */
106 : struct ExchangeInteraction
107 : {
108 : /**
109 : * Kept in a DLL.
110 : */
111 : struct ExchangeInteraction *next;
112 :
113 : /**
114 : * Kept in a DLL.
115 : */
116 : struct ExchangeInteraction *prev;
117 :
118 : /**
119 : * Handle for exchange interaction.
120 : */
121 : struct TALER_EXCHANGE_GetDepositsHandle *dgh;
122 :
123 : /**
124 : * Wire deadline for the deposit.
125 : */
126 : struct GNUNET_TIME_Absolute wire_deadline;
127 :
128 : /**
129 : * Current value for the retry backoff
130 : */
131 : struct GNUNET_TIME_Relative retry_backoff;
132 :
133 : /**
134 : * Target account hash of the deposit.
135 : */
136 : struct TALER_MerchantWireHashP h_wire;
137 :
138 : /**
139 : * Deposited amount.
140 : */
141 : struct TALER_Amount amount_with_fee;
142 :
143 : /**
144 : * Deposit fee paid.
145 : */
146 : struct TALER_Amount deposit_fee;
147 :
148 : /**
149 : * Public key of the deposited coin.
150 : */
151 : struct TALER_CoinSpendPublicKeyP coin_pub;
152 :
153 : /**
154 : * Hash over the @e contract_terms.
155 : */
156 : struct TALER_PrivateContractHashP h_contract_terms;
157 :
158 : /**
159 : * Merchant instance's private key.
160 : */
161 : struct TALER_MerchantPrivateKeyP merchant_priv;
162 :
163 : /**
164 : * Serial number of the row in the deposits table
165 : * that we are processing.
166 : */
167 : uint64_t deposit_serial;
168 :
169 : /**
170 : * The instance the deposit belongs to.
171 : */
172 : char *instance_id;
173 :
174 : };
175 :
176 :
177 : /**
178 : * Head of list of children we forked.
179 : */
180 : static struct Child *c_head;
181 :
182 : /**
183 : * Tail of list of children we forked.
184 : */
185 : static struct Child *c_tail;
186 :
187 : /**
188 : * Key material of the exchange.
189 : */
190 : static struct TALER_EXCHANGE_Keys *keys;
191 :
192 : /**
193 : * Head of list of active exchange interactions.
194 : */
195 : static struct ExchangeInteraction *w_head;
196 :
197 : /**
198 : * Tail of list of active exchange interactions.
199 : */
200 : static struct ExchangeInteraction *w_tail;
201 :
202 : /**
203 : * Number of active entries in the @e w_head list.
204 : */
205 : static uint64_t w_count;
206 :
207 : /**
208 : * Notification handler from database on new work.
209 : */
210 : static struct GNUNET_DB_EventHandler *eh;
211 :
212 : /**
213 : * Notification handler from database on new keys.
214 : */
215 : static struct GNUNET_DB_EventHandler *keys_eh;
216 :
217 : /**
218 : * The merchant's configuration.
219 : */
220 : static const struct GNUNET_CONFIGURATION_Handle *cfg;
221 :
222 : /**
223 : * Name of the configuration file we use.
224 : */
225 : static char *cfg_filename;
226 :
227 : /**
228 : * Our database plugin.
229 : */
230 : static struct TALER_MERCHANTDB_PostgresContext *pg;
231 :
232 : /**
233 : * Next wire deadline that @e task is scheduled for.
234 : */
235 : static struct GNUNET_TIME_Absolute next_deadline;
236 :
237 : /**
238 : * Next task to run, if any.
239 : */
240 : static struct GNUNET_SCHEDULER_Task *task;
241 :
242 : /**
243 : * Handle to the context for interacting with the exchange.
244 : */
245 : static struct GNUNET_CURL_Context *ctx;
246 :
247 : /**
248 : * Scheduler context for running the @e ctx.
249 : */
250 : static struct GNUNET_CURL_RescheduleContext *rc;
251 :
252 : /**
253 : * Which exchange are we monitoring? NULL if we
254 : * are the parent of the workers.
255 : */
256 : static char *exchange_url;
257 :
258 : /**
259 : * Value to return from main(). 0 on success, non-zero on errors.
260 : */
261 : static int global_ret;
262 :
263 : /**
264 : * Should we enable HTTP/2 and HTTP/3 when talking to the exchange?
265 : * Those are not expected to be terribly beneficial for a client with
266 : * stable connections to a few servers, but they could cause stability
267 : * issues with libcurl. Hence we *default* to HTTP/1.1-only, as that
268 : * is the conservative and most tested code path.
269 : */
270 : static int enable_h3;
271 :
272 : /**
273 : * #GNUNET_YES if we are in test mode and should exit when idle.
274 : */
275 : static int test_mode;
276 :
277 :
278 : /**
279 : * We're being aborted with CTRL-C (or SIGTERM). Shut down.
280 : *
281 : * @param cls closure
282 : */
283 : static void
284 15 : shutdown_task (void *cls)
285 : {
286 : struct Child *c;
287 : struct ExchangeInteraction *w;
288 :
289 : (void) cls;
290 15 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
291 : "Running shutdown\n");
292 15 : if (NULL != eh)
293 : {
294 10 : TALER_MERCHANTDB_event_listen_cancel (eh);
295 10 : eh = NULL;
296 : }
297 15 : if (NULL != keys_eh)
298 : {
299 10 : TALER_MERCHANTDB_event_listen_cancel (keys_eh);
300 10 : keys_eh = NULL;
301 : }
302 15 : if (NULL != task)
303 : {
304 0 : GNUNET_SCHEDULER_cancel (task);
305 0 : task = NULL;
306 : }
307 15 : while (NULL != (w = w_head))
308 : {
309 0 : GNUNET_CONTAINER_DLL_remove (w_head,
310 : w_tail,
311 : w);
312 0 : if (NULL != w->dgh)
313 : {
314 0 : TALER_EXCHANGE_get_deposits_cancel (w->dgh);
315 0 : w->dgh = NULL;
316 : }
317 0 : w_count--;
318 0 : GNUNET_free (w->instance_id);
319 0 : GNUNET_free (w);
320 : }
321 20 : while (NULL != (c = c_head))
322 : {
323 5 : GNUNET_CONTAINER_DLL_remove (c_head,
324 : c_tail,
325 : c);
326 5 : if (NULL != c->rt)
327 : {
328 0 : GNUNET_SCHEDULER_cancel (c->rt);
329 0 : c->rt = NULL;
330 : }
331 5 : if (NULL != c->cwh)
332 : {
333 0 : GNUNET_wait_child_cancel (c->cwh);
334 0 : c->cwh = NULL;
335 : }
336 5 : if (NULL != c->process)
337 : {
338 0 : enum GNUNET_OS_ProcessStatusType type
339 : = GNUNET_OS_PROCESS_UNKNOWN;
340 0 : unsigned long code = 0;
341 :
342 0 : GNUNET_break (GNUNET_OK ==
343 : GNUNET_process_kill (c->process,
344 : SIGTERM));
345 0 : GNUNET_break (GNUNET_OK ==
346 : GNUNET_process_wait (c->process,
347 : true,
348 : &type,
349 : &code));
350 0 : if ( (GNUNET_OS_PROCESS_EXITED != type) ||
351 0 : (0 != code) )
352 0 : GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
353 : "Process for exchange %s had trouble (%d/%d)\n",
354 : c->base_url,
355 : (int) type,
356 : (int) code);
357 0 : GNUNET_process_destroy (c->process);
358 : }
359 5 : GNUNET_free (c->base_url);
360 5 : GNUNET_free (c);
361 : }
362 15 : if (NULL != pg)
363 : {
364 10 : TALER_MERCHANTDB_rollback (pg); /* just in case */
365 10 : TALER_MERCHANTDB_disconnect (pg);
366 10 : pg = NULL;
367 : }
368 15 : cfg = NULL;
369 15 : if (NULL != ctx)
370 : {
371 10 : GNUNET_CURL_fini (ctx);
372 10 : ctx = NULL;
373 : }
374 15 : if (NULL != rc)
375 : {
376 10 : GNUNET_CURL_gnunet_rc_destroy (rc);
377 10 : rc = NULL;
378 : }
379 15 : }
380 :
381 :
382 : /**
383 : * Task to get more deposits to work on from the database.
384 : *
385 : * @param cls NULL
386 : */
387 : static void
388 : select_work (void *cls);
389 :
390 :
391 : /**
392 : * Make sure to run the select_work() task at
393 : * the @a next_deadline.
394 : *
395 : * @param deadline time when work becomes ready
396 : */
397 : static void
398 0 : run_at (struct GNUNET_TIME_Absolute deadline)
399 : {
400 0 : if ( (NULL != task) &&
401 0 : (GNUNET_TIME_absolute_cmp (deadline,
402 : >,
403 : next_deadline)) )
404 : {
405 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
406 : "Not scheduling for %s yet, already have earlier task pending\n",
407 : GNUNET_TIME_absolute2s (deadline));
408 0 : return;
409 : }
410 0 : if (NULL == keys)
411 : {
412 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
413 : "Not scheduling for %s yet, no /keys available\n",
414 : GNUNET_TIME_absolute2s (deadline));
415 0 : return; /* too early */
416 : }
417 0 : next_deadline = deadline;
418 0 : if (NULL != task)
419 0 : GNUNET_SCHEDULER_cancel (task);
420 0 : task = GNUNET_SCHEDULER_add_at (deadline,
421 : &select_work,
422 : NULL);
423 : }
424 :
425 :
426 : /**
427 : * Function called with detailed wire transfer data.
428 : *
429 : * @param cls closure with a `struct ExchangeInteraction *`
430 : * @param dr HTTP response data
431 : */
432 : static void
433 8 : deposit_get_cb (
434 : struct ExchangeInteraction *w,
435 : const struct TALER_EXCHANGE_GetDepositsResponse *dr)
436 : {
437 : struct GNUNET_TIME_Absolute future_retry;
438 : enum GNUNET_DB_QueryStatus qs;
439 :
440 8 : w->dgh = NULL;
441 8 : qs = TALER_MERCHANTDB_set_instance (
442 : pg,
443 8 : w->instance_id);
444 8 : if (qs <= 0)
445 : {
446 0 : GNUNET_break (0);
447 0 : global_ret = EXIT_FAILURE;
448 0 : GNUNET_SCHEDULER_shutdown ();
449 0 : return;
450 : }
451 : future_retry
452 8 : = GNUNET_TIME_relative_to_absolute (w->retry_backoff);
453 8 : switch (dr->hr.http_status)
454 : {
455 8 : case MHD_HTTP_OK:
456 8 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
457 : "Exchange returned wire transfer over %s for deposited coin %s\n",
458 : TALER_amount2s (&dr->details.ok.coin_contribution),
459 : TALER_B2S (&w->coin_pub));
460 : {
461 : enum TALER_MERCHANTDB_DepositToTransferStatus dtts;
462 :
463 8 : dtts = TALER_MERCHANTDB_insert_deposit_to_transfer (
464 : pg,
465 : w->deposit_serial,
466 8 : &w->h_wire,
467 : exchange_url,
468 : &dr->details.ok);
469 8 : switch (dtts)
470 : {
471 0 : case TALER_MERCHANTDB_DTTS_HARD_ERROR:
472 : case TALER_MERCHANTDB_DTTS_SOFT_ERROR:
473 : case TALER_MERCHANTDB_DTTS_NO_RESULTS:
474 0 : GNUNET_break (0);
475 0 : global_ret = EXIT_FAILURE;
476 0 : GNUNET_SCHEDULER_shutdown ();
477 0 : return;
478 8 : case TALER_MERCHANTDB_DTTS_SETTLED:
479 8 : break;
480 0 : case TALER_MERCHANTDB_DTTS_SIGNKEY_UNKNOWN:
481 : /* transient, the DB scheduled a retry for us */
482 0 : break;
483 0 : case TALER_MERCHANTDB_DTTS_ACCOUNT_UNKNOWN:
484 : /* permanent failure, the operator has to look into this */
485 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
486 : "Exchange `%s' claims to have wired coin %s to an account we do not know; deposit will not settle\n",
487 : exchange_url,
488 : TALER_B2S (&w->coin_pub));
489 0 : break;
490 : }
491 : }
492 8 : break;
493 0 : case MHD_HTTP_ACCEPTED:
494 : {
495 : /* got a 'preliminary' reply from the exchange,
496 : remember our target UUID */
497 : struct GNUNET_TIME_Timestamp now;
498 :
499 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
500 : "Exchange returned KYC requirement (%d) for deposited coin %s\n",
501 : dr->details.accepted.kyc_ok,
502 : TALER_B2S (&w->coin_pub));
503 0 : now = GNUNET_TIME_timestamp_get ();
504 0 : qs = TALER_MERCHANTDB_insert_kyc_failure (
505 : pg,
506 0 : w->instance_id,
507 0 : &w->h_wire,
508 : exchange_url,
509 : now,
510 : MHD_HTTP_ACCEPTED,
511 0 : dr->details.accepted.kyc_ok);
512 0 : if (qs < 0)
513 : {
514 0 : GNUNET_break (0);
515 0 : global_ret = EXIT_FAILURE;
516 0 : GNUNET_SCHEDULER_shutdown ();
517 0 : return;
518 : }
519 0 : if (dr->details.accepted.kyc_ok)
520 : {
521 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
522 : "Bumping wire transfer deadline in DB to %s as that is when we will retry\n",
523 : GNUNET_TIME_absolute2s (future_retry));
524 0 : qs = TALER_MERCHANTDB_update_deposit_settlement_status (
525 : pg,
526 : w->deposit_serial,
527 : true, /* need to try again in the future! */
528 : GNUNET_TIME_absolute_to_timestamp (future_retry),
529 : MHD_HTTP_ACCEPTED,
530 : TALER_EC_NONE,
531 : "Exchange reported 202 Accepted but no KYC block");
532 0 : if (qs < 0)
533 : {
534 0 : GNUNET_break (0);
535 0 : global_ret = EXIT_FAILURE;
536 0 : GNUNET_SCHEDULER_shutdown ();
537 0 : return;
538 : }
539 : }
540 : else
541 : {
542 : future_retry
543 0 : = GNUNET_TIME_absolute_max (
544 : future_retry,
545 : GNUNET_TIME_relative_to_absolute (
546 : KYC_RETRY_DELAY));
547 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
548 : "Bumping wire transfer deadline in DB to %s as that is when we will retry\n",
549 : GNUNET_TIME_absolute2s (future_retry));
550 0 : qs = TALER_MERCHANTDB_update_deposit_settlement_status (
551 : pg,
552 : w->deposit_serial,
553 : true /* need to try again in the future */,
554 : GNUNET_TIME_absolute_to_timestamp (future_retry),
555 : MHD_HTTP_ACCEPTED,
556 : TALER_EC_NONE,
557 : "Exchange reported 202 Accepted due to KYC/AML block");
558 0 : if (qs < 0)
559 : {
560 0 : GNUNET_break (0);
561 0 : global_ret = EXIT_FAILURE;
562 0 : GNUNET_SCHEDULER_shutdown ();
563 0 : return;
564 : }
565 : }
566 0 : break;
567 : }
568 0 : default:
569 : {
570 0 : bool retry_needed = false;
571 :
572 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
573 : "Exchange %s returned tracking failure for deposited coin %s: %u\n",
574 : exchange_url,
575 : TALER_B2S (&w->coin_pub),
576 : dr->hr.http_status);
577 : /* rough classification by HTTP status group */
578 0 : switch (dr->hr.http_status / 100)
579 : {
580 0 : case 0:
581 : /* timeout */
582 0 : retry_needed = true;
583 0 : break;
584 0 : case 1:
585 : case 2:
586 : case 3:
587 : /* very strange */
588 0 : retry_needed = false;
589 0 : break;
590 0 : case 4:
591 : /* likely fatal */
592 0 : retry_needed = false;
593 0 : break;
594 0 : case 5:
595 : /* likely transient */
596 0 : retry_needed = true;
597 0 : break;
598 : }
599 0 : qs = TALER_MERCHANTDB_update_deposit_settlement_status (
600 : pg,
601 : w->deposit_serial,
602 : retry_needed,
603 : GNUNET_TIME_absolute_to_timestamp (future_retry),
604 0 : (uint32_t) dr->hr.http_status,
605 0 : dr->hr.ec,
606 0 : dr->hr.hint);
607 0 : if (qs < 0)
608 : {
609 0 : GNUNET_break (0);
610 0 : global_ret = EXIT_FAILURE;
611 0 : GNUNET_SCHEDULER_shutdown ();
612 0 : return;
613 : }
614 0 : break;
615 : }
616 : } /* end switch */
617 8 : GNUNET_break (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT ==
618 : TALER_MERCHANTDB_set_instance (pg,
619 : NULL));
620 :
621 8 : GNUNET_CONTAINER_DLL_remove (w_head,
622 : w_tail,
623 : w);
624 8 : w_count--;
625 8 : GNUNET_free (w->instance_id);
626 8 : GNUNET_free (w);
627 8 : GNUNET_assert (NULL != keys);
628 8 : if (0 == w_count)
629 : {
630 : /* We only SELECT() again after having finished
631 : all requests, as otherwise we'll most like
632 : just SELECT() those again that are already
633 : being requested; alternatively, we could
634 : update the retry_time already on SELECT(),
635 : but this should be easier on the DB. */
636 8 : if (NULL != task)
637 0 : GNUNET_SCHEDULER_cancel (task);
638 8 : task = GNUNET_SCHEDULER_add_now (&select_work,
639 : NULL);
640 : }
641 : }
642 :
643 :
644 : /**
645 : * Typically called by `select_work`.
646 : *
647 : * @param cls NULL
648 : * @param deposit_serial identifies the deposit operation
649 : * @param wire_deadline when is the wire due
650 : * @param retry_time current value for the retry backoff
651 : * @param h_contract_terms hash of the contract terms
652 : * @param merchant_priv private key of the merchant
653 : * @param instance_id row ID of the instance
654 : * @param h_wire hash of the merchant's wire account into
655 : * @param amount_with_fee amount the exchange will deposit for this coin
656 : * @param deposit_fee fee the exchange will charge for this coin which the deposit was made
657 : * @param coin_pub public key of the deposited coin
658 : */
659 : static void
660 8 : pending_deposits_cb (
661 : void *cls,
662 : uint64_t deposit_serial,
663 : struct GNUNET_TIME_Absolute wire_deadline,
664 : struct GNUNET_TIME_Absolute retry_time,
665 : const struct TALER_PrivateContractHashP *h_contract_terms,
666 : const struct TALER_MerchantPrivateKeyP *merchant_priv,
667 : const char *instance_id,
668 : const struct TALER_MerchantWireHashP *h_wire,
669 : const struct TALER_Amount *amount_with_fee,
670 : const struct TALER_Amount *deposit_fee,
671 : const struct TALER_CoinSpendPublicKeyP *coin_pub)
672 : {
673 : struct ExchangeInteraction *w;
674 : struct GNUNET_TIME_Absolute mx
675 8 : = GNUNET_TIME_absolute_max (wire_deadline,
676 : retry_time);
677 : struct GNUNET_TIME_Relative retry_backoff;
678 :
679 : (void) cls;
680 8 : if (GNUNET_TIME_absolute_is_future (mx))
681 : {
682 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
683 : "Pending deposit should be checked next at %s\n",
684 : GNUNET_TIME_absolute2s (mx));
685 0 : run_at (mx);
686 0 : return;
687 : }
688 8 : if (GNUNET_TIME_absolute_is_zero (retry_time))
689 0 : retry_backoff = GNUNET_TIME_absolute_get_duration (wire_deadline);
690 : else
691 8 : retry_backoff = GNUNET_TIME_absolute_get_difference (wire_deadline,
692 : retry_time);
693 8 : w = GNUNET_new (struct ExchangeInteraction);
694 8 : w->deposit_serial = deposit_serial;
695 8 : w->wire_deadline = wire_deadline;
696 8 : w->retry_backoff = GNUNET_TIME_randomized_backoff (retry_backoff,
697 : GNUNET_TIME_UNIT_DAYS);
698 8 : w->h_contract_terms = *h_contract_terms;
699 8 : w->merchant_priv = *merchant_priv;
700 8 : w->h_wire = *h_wire;
701 8 : w->amount_with_fee = *amount_with_fee;
702 8 : w->deposit_fee = *deposit_fee;
703 8 : w->coin_pub = *coin_pub;
704 8 : w->instance_id = GNUNET_strdup (instance_id);
705 8 : GNUNET_CONTAINER_DLL_insert (w_head,
706 : w_tail,
707 : w);
708 8 : w_count++;
709 8 : GNUNET_assert (NULL != keys);
710 8 : if (GNUNET_TIME_absolute_is_past (
711 8 : keys->key_data_expiration.abs_time))
712 : {
713 : /* Parent should re-start us, then we will re-fetch /keys */
714 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
715 : "/keys expired, shutting down\n");
716 0 : GNUNET_SCHEDULER_shutdown ();
717 0 : return;
718 : }
719 8 : GNUNET_assert (NULL == w->dgh);
720 16 : w->dgh = TALER_EXCHANGE_get_deposits_create (
721 : ctx,
722 : exchange_url,
723 : keys,
724 8 : &w->merchant_priv,
725 8 : &w->h_wire,
726 8 : &w->h_contract_terms,
727 8 : &w->coin_pub);
728 8 : if (NULL == w->dgh)
729 : {
730 0 : GNUNET_break (0);
731 0 : GNUNET_SCHEDULER_shutdown ();
732 0 : return;
733 : }
734 8 : if (TALER_EC_NONE !=
735 8 : TALER_EXCHANGE_get_deposits_start (w->dgh,
736 : &deposit_get_cb,
737 : w))
738 : {
739 0 : GNUNET_break (0);
740 0 : TALER_EXCHANGE_get_deposits_cancel (w->dgh);
741 0 : w->dgh = NULL;
742 0 : GNUNET_SCHEDULER_shutdown ();
743 0 : return;
744 : }
745 : }
746 :
747 :
748 : /**
749 : * Function called on events received from Postgres.
750 : *
751 : * @param cls closure, NULL
752 : * @param extra additional event data provided, timestamp with wire deadline
753 : * @param extra_size number of bytes in @a extra
754 : */
755 : static void
756 0 : db_notify (void *cls,
757 : const void *extra,
758 : size_t extra_size)
759 : {
760 : struct GNUNET_TIME_Absolute deadline;
761 : struct GNUNET_TIME_AbsoluteNBO nbo_deadline;
762 :
763 : (void) cls;
764 0 : if (sizeof (nbo_deadline) != extra_size)
765 : {
766 0 : GNUNET_break (0);
767 0 : return;
768 : }
769 0 : if (0 != w_count)
770 0 : return; /* already at work! */
771 0 : memcpy (&nbo_deadline,
772 : extra,
773 : extra_size);
774 0 : deadline = GNUNET_TIME_absolute_ntoh (nbo_deadline);
775 0 : run_at (deadline);
776 : }
777 :
778 :
779 : static void
780 18 : select_work (void *cls)
781 : {
782 18 : bool retry = false;
783 18 : uint64_t limit = CONCURRENCY_LIMIT - w_count;
784 :
785 : (void) cls;
786 18 : task = NULL;
787 18 : GNUNET_assert (w_count <= CONCURRENCY_LIMIT);
788 18 : GNUNET_assert (NULL != keys);
789 18 : if (0 == limit)
790 : {
791 0 : GNUNET_break (0);
792 0 : return;
793 : }
794 18 : if (GNUNET_TIME_absolute_is_past (
795 18 : keys->key_data_expiration.abs_time))
796 : {
797 : /* Parent should re-start us, then we will re-fetch /keys */
798 0 : GNUNET_SCHEDULER_shutdown ();
799 0 : return;
800 : }
801 : while (1)
802 0 : {
803 : enum GNUNET_DB_QueryStatus qs;
804 :
805 18 : TALER_MERCHANTDB_preflight (pg);
806 18 : if (retry)
807 0 : limit = 1;
808 18 : qs = TALER_MERCHANTDB_iterate_pending_deposits (
809 : pg,
810 : exchange_url,
811 : limit,
812 : retry,
813 : &pending_deposits_cb,
814 : NULL);
815 18 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
816 : "Looking up pending deposits query status was %d\n",
817 : (int) qs);
818 18 : switch (qs)
819 : {
820 0 : case GNUNET_DB_STATUS_HARD_ERROR:
821 : case GNUNET_DB_STATUS_SOFT_ERROR:
822 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
823 : "Transaction failed!\n");
824 0 : global_ret = EXIT_FAILURE;
825 0 : GNUNET_SCHEDULER_shutdown ();
826 0 : return;
827 10 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
828 10 : if (test_mode)
829 : {
830 10 : GNUNET_SCHEDULER_shutdown ();
831 10 : return;
832 : }
833 0 : if (retry)
834 0 : return; /* nothing left */
835 0 : retry = true;
836 0 : continue;
837 8 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
838 : default:
839 : /* wait for async completion, then select more work. */
840 8 : return;
841 : }
842 : }
843 : }
844 :
845 :
846 : /**
847 : * Start a copy of this process with the exchange URL
848 : * set to the given @a base_url
849 : *
850 : * @param base_url base URL to run with
851 : */
852 : static struct GNUNET_Process *
853 5 : start_worker (const char *base_url)
854 : {
855 : struct GNUNET_Process *p;
856 : char toff[30];
857 : long long zo;
858 : enum GNUNET_GenericReturnValue ret;
859 :
860 5 : zo = GNUNET_TIME_get_offset ();
861 5 : GNUNET_snprintf (toff,
862 : sizeof (toff),
863 : "%lld",
864 : zo);
865 5 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
866 : "Launching worker for exchange `%s' using `%s`\n",
867 : base_url,
868 : NULL == cfg_filename
869 : ? "<default>"
870 : : cfg_filename);
871 5 : p = GNUNET_process_create (GNUNET_OS_INHERIT_STD_ERR);
872 :
873 5 : if (NULL == cfg_filename)
874 0 : ret = GNUNET_process_run_command_va (
875 : p,
876 : "taler-merchant-depositcheck",
877 : "taler-merchant-depositcheck",
878 : "-e", base_url,
879 : "-L", "INFO",
880 : "-T", toff,
881 0 : test_mode ? "-t" : NULL,
882 : NULL);
883 : else
884 5 : ret = GNUNET_process_run_command_va (
885 : p,
886 : "taler-merchant-depositcheck",
887 : "taler-merchant-depositcheck",
888 : "-c", cfg_filename,
889 : "-e", base_url,
890 : "-L", "INFO",
891 : "-T", toff,
892 5 : test_mode ? "-t" : NULL,
893 : NULL);
894 5 : if (GNUNET_OK != ret)
895 : {
896 0 : GNUNET_process_destroy (p);
897 0 : return NULL;
898 : }
899 5 : return p;
900 : }
901 :
902 :
903 : /**
904 : * Restart worker process for the given child.
905 : *
906 : * @param cls a `struct Child *` that needs a worker.
907 : */
908 : static void
909 : restart_child (void *cls);
910 :
911 :
912 : /**
913 : * Function called upon death or completion of a child process.
914 : *
915 : * @param cls a `struct Child *`
916 : * @param type type of the process
917 : * @param exit_code status code of the process
918 : */
919 : static void
920 5 : child_done_cb (void *cls,
921 : enum GNUNET_OS_ProcessStatusType type,
922 : long unsigned int exit_code)
923 : {
924 5 : struct Child *c = cls;
925 :
926 5 : c->cwh = NULL;
927 5 : GNUNET_process_destroy (c->process);
928 5 : c->process = NULL;
929 5 : if ( (GNUNET_OS_PROCESS_EXITED != type) ||
930 : (0 != exit_code) )
931 : {
932 0 : GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
933 : "Process for exchange %s had trouble (%d/%d)\n",
934 : c->base_url,
935 : (int) type,
936 : (int) exit_code);
937 0 : GNUNET_SCHEDULER_shutdown ();
938 : /* Preserve explicit permanent failures; a signal is a transient failure,
939 : even when its number happens to be 6 or 9. */
940 0 : global_ret = (GNUNET_OS_PROCESS_EXITED == type)
941 : ? (int) exit_code
942 0 : : EXIT_FAILURE;
943 0 : return;
944 : }
945 : /* A successful child finishes when its exchange keys expire. Starting
946 : the next scan is normal work; failed children are handled above. */
947 5 : if (test_mode &&
948 5 : (! GNUNET_TIME_relative_is_zero (c->rd)) )
949 : {
950 5 : return;
951 : }
952 0 : if (GNUNET_TIME_absolute_is_future (c->next_start))
953 0 : c->rd = GNUNET_TIME_STD_BACKOFF (c->rd);
954 : else
955 0 : c->rd = GNUNET_TIME_UNIT_SECONDS;
956 0 : c->rt = GNUNET_SCHEDULER_add_at (c->next_start,
957 : &restart_child,
958 : c);
959 : }
960 :
961 :
962 : static void
963 5 : restart_child (void *cls)
964 : {
965 5 : struct Child *c = cls;
966 :
967 5 : c->rt = NULL;
968 5 : c->next_start = GNUNET_TIME_relative_to_absolute (c->rd);
969 5 : c->process = start_worker (c->base_url);
970 5 : if (NULL == c->process)
971 : {
972 0 : GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR,
973 : "exec");
974 0 : global_ret = EXIT_FAILURE;
975 0 : GNUNET_SCHEDULER_shutdown ();
976 0 : return;
977 : }
978 5 : c->cwh = GNUNET_wait_child (c->process,
979 : &child_done_cb,
980 : c);
981 : }
982 :
983 :
984 : /**
985 : * Function to iterate over section.
986 : *
987 : * @param cls closure
988 : * @param section name of the section
989 : */
990 : static void
991 223 : cfg_iter_cb (void *cls,
992 : const char *section)
993 : {
994 : char *base_url;
995 : struct Child *c;
996 :
997 223 : if (0 !=
998 223 : strncasecmp (section,
999 : "merchant-exchange-",
1000 : strlen ("merchant-exchange-")))
1001 218 : return;
1002 15 : if (GNUNET_YES ==
1003 15 : GNUNET_CONFIGURATION_get_value_yesno (cfg,
1004 : section,
1005 : "DISABLED"))
1006 10 : return;
1007 5 : if (GNUNET_OK !=
1008 5 : GNUNET_CONFIGURATION_get_value_string (cfg,
1009 : section,
1010 : "EXCHANGE_BASE_URL",
1011 : &base_url))
1012 : {
1013 0 : GNUNET_log_config_missing (GNUNET_ERROR_TYPE_WARNING,
1014 : section,
1015 : "EXCHANGE_BASE_URL");
1016 0 : return;
1017 : }
1018 5 : c = GNUNET_new (struct Child);
1019 5 : c->rd = GNUNET_TIME_UNIT_SECONDS;
1020 5 : c->base_url = base_url;
1021 5 : GNUNET_CONTAINER_DLL_insert (c_head,
1022 : c_tail,
1023 : c);
1024 5 : c->rt = GNUNET_SCHEDULER_add_now (&restart_child,
1025 : c);
1026 : }
1027 :
1028 :
1029 : /**
1030 : * Trigger (re)loading of keys from DB.
1031 : *
1032 : * @param cls NULL
1033 : * @param extra base URL of the exchange that changed
1034 : * @param extra_len number of bytes in @a extra
1035 : */
1036 : static void
1037 10 : update_exchange_keys (void *cls,
1038 : const void *extra,
1039 : size_t extra_len)
1040 : {
1041 10 : const char *url = extra;
1042 :
1043 10 : if ( (NULL == extra) ||
1044 : (0 == extra_len) )
1045 : {
1046 0 : GNUNET_break (0);
1047 0 : return;
1048 : }
1049 10 : if ('\0' != url[extra_len - 1])
1050 : {
1051 0 : GNUNET_break (0);
1052 0 : return;
1053 : }
1054 10 : if (0 != strcmp (url,
1055 : exchange_url))
1056 0 : return; /* not relevant for us */
1057 :
1058 : {
1059 : enum GNUNET_DB_QueryStatus qs;
1060 : struct GNUNET_TIME_Absolute earliest_retry;
1061 :
1062 10 : if (NULL != keys)
1063 : {
1064 0 : TALER_EXCHANGE_keys_decref (keys);
1065 0 : keys = NULL;
1066 : }
1067 10 : qs = TALER_MERCHANTDB_get_exchange_keys (pg,
1068 : exchange_url,
1069 : &earliest_retry,
1070 : &keys);
1071 10 : if (qs < 0)
1072 : {
1073 0 : GNUNET_break (0);
1074 0 : global_ret = EXIT_FAILURE;
1075 0 : GNUNET_SCHEDULER_shutdown ();
1076 0 : return;
1077 : }
1078 10 : if ( (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) ||
1079 10 : (NULL == keys) )
1080 : {
1081 0 : GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1082 : "No keys yet for `%s'\n",
1083 : exchange_url);
1084 : }
1085 : }
1086 10 : if (NULL == keys)
1087 : {
1088 0 : if (NULL != task)
1089 : {
1090 0 : GNUNET_SCHEDULER_cancel (task);
1091 0 : task = NULL;
1092 : }
1093 : }
1094 : else
1095 : {
1096 10 : if (NULL == task)
1097 10 : task = GNUNET_SCHEDULER_add_now (&select_work,
1098 : NULL);
1099 : }
1100 : }
1101 :
1102 :
1103 : /**
1104 : * First task.
1105 : *
1106 : * @param cls closure, NULL
1107 : * @param args remaining command-line arguments
1108 : * @param cfgfile name of the configuration file used (for saving, can be NULL!)
1109 : * @param c configuration
1110 : */
1111 : static void
1112 15 : run (void *cls,
1113 : char *const *args,
1114 : const char *cfgfile,
1115 : const struct GNUNET_CONFIGURATION_Handle *c)
1116 : {
1117 : (void) args;
1118 :
1119 15 : cfg = c;
1120 15 : TALER_EXCHANGE_setup (enable_h3
1121 15 : ? TALER_EXCHANGE_GO_ENABLE_HTTP3
1122 : : TALER_EXCHANGE_GO_FORCE_HTTP1_1);
1123 15 : if (NULL != cfgfile)
1124 15 : cfg_filename = GNUNET_strdup (cfgfile);
1125 15 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1126 : "Running with configuration %s\n",
1127 : cfgfile);
1128 15 : GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
1129 : NULL);
1130 15 : if (NULL == exchange_url)
1131 : {
1132 5 : GNUNET_CONFIGURATION_iterate_sections (c,
1133 : &cfg_iter_cb,
1134 : NULL);
1135 5 : if (NULL == c_head)
1136 : {
1137 0 : GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1138 : "No exchanges found in configuration\n");
1139 0 : global_ret = EXIT_NOTCONFIGURED;
1140 0 : GNUNET_SCHEDULER_shutdown ();
1141 0 : return;
1142 : }
1143 5 : return;
1144 : }
1145 :
1146 10 : ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule,
1147 : &rc);
1148 10 : rc = GNUNET_CURL_gnunet_rc_create (ctx);
1149 10 : if (NULL == ctx)
1150 : {
1151 0 : GNUNET_break (0);
1152 0 : GNUNET_SCHEDULER_shutdown ();
1153 0 : global_ret = EXIT_FAILURE;
1154 0 : return;
1155 : }
1156 10 : if (NULL ==
1157 10 : (pg = TALER_MERCHANTDB_connect (cfg)))
1158 : {
1159 0 : GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1160 : "Failed to initialize DB subsystem\n");
1161 0 : GNUNET_SCHEDULER_shutdown ();
1162 0 : global_ret = EXIT_FAILURE;
1163 0 : return;
1164 : }
1165 : {
1166 10 : struct GNUNET_DB_EventHeaderP es = {
1167 10 : .size = htons (sizeof (es)),
1168 10 : .type = htons (TALER_DBEVENT_MERCHANT_NEW_WIRE_DEADLINE)
1169 : };
1170 :
1171 20 : eh = TALER_MERCHANTDB_event_listen (pg,
1172 : &es,
1173 10 : GNUNET_TIME_UNIT_FOREVER_REL,
1174 : &db_notify,
1175 : NULL);
1176 : }
1177 : {
1178 10 : struct GNUNET_DB_EventHeaderP es = {
1179 10 : .size = htons (sizeof (es)),
1180 10 : .type = htons (TALER_DBEVENT_MERCHANT_EXCHANGE_KEYS)
1181 : };
1182 :
1183 20 : keys_eh = TALER_MERCHANTDB_event_listen (pg,
1184 : &es,
1185 10 : GNUNET_TIME_UNIT_FOREVER_REL,
1186 : &update_exchange_keys,
1187 : NULL);
1188 : }
1189 :
1190 10 : update_exchange_keys (NULL,
1191 : exchange_url,
1192 10 : strlen (exchange_url) + 1);
1193 : }
1194 :
1195 :
1196 : /**
1197 : * The main function of the taler-merchant-depositcheck
1198 : *
1199 : * @param argc number of arguments from the command line
1200 : * @param argv command line arguments
1201 : * @return 0 ok, 1 on error
1202 : */
1203 : int
1204 15 : main (int argc,
1205 : char *const *argv)
1206 : {
1207 15 : struct GNUNET_GETOPT_CommandLineOption options[] = {
1208 15 : GNUNET_GETOPT_option_string ('e',
1209 : "exchange",
1210 : "BASE_URL",
1211 : "limit us to checking deposits of this exchange",
1212 : &exchange_url),
1213 15 : GNUNET_GETOPT_option_flag ('3',
1214 : "http3",
1215 : "enable support for HTTP/2 and HTTP/3",
1216 : &enable_h3),
1217 15 : GNUNET_GETOPT_option_timetravel ('T',
1218 : "timetravel"),
1219 15 : GNUNET_GETOPT_option_flag ('t',
1220 : "test",
1221 : "run in test mode and exit when idle",
1222 : &test_mode),
1223 15 : GNUNET_GETOPT_option_version (VERSION),
1224 : GNUNET_GETOPT_OPTION_END
1225 : };
1226 : enum GNUNET_GenericReturnValue ret;
1227 :
1228 15 : ret = GNUNET_PROGRAM_run (
1229 : TALER_MERCHANT_project_data (),
1230 : argc, argv,
1231 : "taler-merchant-depositcheck",
1232 : gettext_noop (
1233 : "background process that checks with the exchange on deposits that are past the wire deadline"),
1234 : options,
1235 : &run, NULL);
1236 15 : if (GNUNET_SYSERR == ret)
1237 0 : return EXIT_NOTCONFIGURED;
1238 15 : if (GNUNET_NO == ret)
1239 0 : return EXIT_SUCCESS;
1240 15 : return global_ret;
1241 : }
1242 :
1243 :
1244 : /* end of taler-merchant-depositcheck.c */
|
