merchant

Merchant backend to process payments, run by merchants
Log | Files | Refs | Submodules | README | LICENSE

taler-merchant-depositcheck.c (30908B)


      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/lookup_pending_deposits.h"
     34 #include "merchant-database/select_exchange_keys.h"
     35 #include "merchant-database/preflight.h"
     36 #include "merchant-database/account_kyc_set_failed.h"
     37 #include "merchant-database/set_instance.h"
     38 #include "merchant-database/insert_deposit_to_transfer.h"
     39 #include "merchant-database/update_deposit_confirmation_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  * #GNUNET_YES if we are in test mode and should exit when idle.
    265  */
    266 static int test_mode;
    267 
    268 
    269 /**
    270  * We're being aborted with CTRL-C (or SIGTERM). Shut down.
    271  *
    272  * @param cls closure
    273  */
    274 static void
    275 shutdown_task (void *cls)
    276 {
    277   struct Child *c;
    278   struct ExchangeInteraction *w;
    279 
    280   (void) cls;
    281   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    282               "Running shutdown\n");
    283   if (NULL != eh)
    284   {
    285     TALER_MERCHANTDB_event_listen_cancel (eh);
    286     eh = NULL;
    287   }
    288   if (NULL != keys_eh)
    289   {
    290     TALER_MERCHANTDB_event_listen_cancel (keys_eh);
    291     keys_eh = NULL;
    292   }
    293   if (NULL != task)
    294   {
    295     GNUNET_SCHEDULER_cancel (task);
    296     task = NULL;
    297   }
    298   while (NULL != (w = w_head))
    299   {
    300     GNUNET_CONTAINER_DLL_remove (w_head,
    301                                  w_tail,
    302                                  w);
    303     if (NULL != w->dgh)
    304     {
    305       TALER_EXCHANGE_get_deposits_cancel (w->dgh);
    306       w->dgh = NULL;
    307     }
    308     w_count--;
    309     GNUNET_free (w->instance_id);
    310     GNUNET_free (w);
    311   }
    312   while (NULL != (c = c_head))
    313   {
    314     GNUNET_CONTAINER_DLL_remove (c_head,
    315                                  c_tail,
    316                                  c);
    317     if (NULL != c->rt)
    318     {
    319       GNUNET_SCHEDULER_cancel (c->rt);
    320       c->rt = NULL;
    321     }
    322     if (NULL != c->cwh)
    323     {
    324       GNUNET_wait_child_cancel (c->cwh);
    325       c->cwh = NULL;
    326     }
    327     if (NULL != c->process)
    328     {
    329       enum GNUNET_OS_ProcessStatusType type
    330         = GNUNET_OS_PROCESS_UNKNOWN;
    331       unsigned long code = 0;
    332 
    333       GNUNET_break (GNUNET_OK ==
    334                     GNUNET_process_kill (c->process,
    335                                          SIGTERM));
    336       GNUNET_break (GNUNET_OK ==
    337                     GNUNET_process_wait (c->process,
    338                                          true,
    339                                          &type,
    340                                          &code));
    341       if ( (GNUNET_OS_PROCESS_EXITED != type) ||
    342            (0 != code) )
    343         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
    344                     "Process for exchange %s had trouble (%d/%d)\n",
    345                     c->base_url,
    346                     (int) type,
    347                     (int) code);
    348       GNUNET_process_destroy (c->process);
    349     }
    350     GNUNET_free (c->base_url);
    351     GNUNET_free (c);
    352   }
    353   if (NULL != pg)
    354   {
    355     TALER_MERCHANTDB_rollback (pg); /* just in case */
    356     TALER_MERCHANTDB_disconnect (pg);
    357     pg = NULL;
    358   }
    359   cfg = NULL;
    360   if (NULL != ctx)
    361   {
    362     GNUNET_CURL_fini (ctx);
    363     ctx = NULL;
    364   }
    365   if (NULL != rc)
    366   {
    367     GNUNET_CURL_gnunet_rc_destroy (rc);
    368     rc = NULL;
    369   }
    370 }
    371 
    372 
    373 /**
    374  * Task to get more deposits to work on from the database.
    375  *
    376  * @param cls NULL
    377  */
    378 static void
    379 select_work (void *cls);
    380 
    381 
    382 /**
    383  * Make sure to run the select_work() task at
    384  * the @a next_deadline.
    385  *
    386  * @param deadline time when work becomes ready
    387  */
    388 static void
    389 run_at (struct GNUNET_TIME_Absolute deadline)
    390 {
    391   if ( (NULL != task) &&
    392        (GNUNET_TIME_absolute_cmp (deadline,
    393                                   >,
    394                                   next_deadline)) )
    395   {
    396     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    397                 "Not scheduling for %s yet, already have earlier task pending\n",
    398                 GNUNET_TIME_absolute2s (deadline));
    399     return;
    400   }
    401   if (NULL == keys)
    402   {
    403     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    404                 "Not scheduling for %s yet, no /keys available\n",
    405                 GNUNET_TIME_absolute2s (deadline));
    406     return; /* too early */
    407   }
    408   next_deadline = deadline;
    409   if (NULL != task)
    410     GNUNET_SCHEDULER_cancel (task);
    411   task = GNUNET_SCHEDULER_add_at (deadline,
    412                                   &select_work,
    413                                   NULL);
    414 }
    415 
    416 
    417 /**
    418  * Function called with detailed wire transfer data.
    419  *
    420  * @param cls closure with a `struct ExchangeInteraction *`
    421  * @param dr HTTP response data
    422  */
    423 static void
    424 deposit_get_cb (
    425   struct ExchangeInteraction *w,
    426   const struct TALER_EXCHANGE_GetDepositsResponse *dr)
    427 {
    428   struct GNUNET_TIME_Absolute future_retry;
    429 
    430   w->dgh = NULL;
    431   future_retry
    432     = GNUNET_TIME_relative_to_absolute (w->retry_backoff);
    433   switch (dr->hr.http_status)
    434   {
    435   case MHD_HTTP_OK:
    436     {
    437       enum GNUNET_DB_QueryStatus qs;
    438 
    439       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    440                   "Exchange returned wire transfer over %s for deposited coin %s\n",
    441                   TALER_amount2s (&dr->details.ok.coin_contribution),
    442                   TALER_B2S (&w->coin_pub));
    443       qs = TALER_MERCHANTDB_set_instance (
    444         pg,
    445         w->instance_id);
    446       if (qs <= 0)
    447       {
    448         GNUNET_break (0);
    449         global_ret = EXIT_FAILURE;
    450         GNUNET_SCHEDULER_shutdown ();
    451         return;
    452       }
    453       qs = TALER_MERCHANTDB_insert_deposit_to_transfer (
    454         pg,
    455         w->deposit_serial,
    456         &w->h_wire,
    457         exchange_url,
    458         &dr->details.ok);
    459       if (qs <= 0)
    460       {
    461         GNUNET_break (0);
    462         global_ret = EXIT_FAILURE;
    463         GNUNET_SCHEDULER_shutdown ();
    464         return;
    465       }
    466       GNUNET_break (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT ==
    467                     TALER_MERCHANTDB_set_instance (
    468                       pg,
    469                       NULL));
    470       break;
    471     }
    472   case MHD_HTTP_ACCEPTED:
    473     {
    474       /* got a 'preliminary' reply from the exchange,
    475          remember our target UUID */
    476       enum GNUNET_DB_QueryStatus qs;
    477       struct GNUNET_TIME_Timestamp now;
    478 
    479       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    480                   "Exchange returned KYC requirement (%d) for deposited coin %s\n",
    481                   dr->details.accepted.kyc_ok,
    482                   TALER_B2S (&w->coin_pub));
    483       now = GNUNET_TIME_timestamp_get ();
    484       qs = TALER_MERCHANTDB_account_kyc_set_failed (
    485         pg,
    486         w->instance_id,
    487         &w->h_wire,
    488         exchange_url,
    489         now,
    490         MHD_HTTP_ACCEPTED,
    491         dr->details.accepted.kyc_ok);
    492       if (qs < 0)
    493       {
    494         GNUNET_break (0);
    495         global_ret = EXIT_FAILURE;
    496         GNUNET_SCHEDULER_shutdown ();
    497         return;
    498       }
    499       if (dr->details.accepted.kyc_ok)
    500       {
    501         GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    502                     "Bumping wire transfer deadline in DB to %s as that is when we will retry\n",
    503                     GNUNET_TIME_absolute2s (future_retry));
    504         qs = TALER_MERCHANTDB_update_deposit_confirmation_status (
    505           pg,
    506           w->deposit_serial,
    507           true, /* need to try again in the future! */
    508           GNUNET_TIME_absolute_to_timestamp (future_retry),
    509           MHD_HTTP_ACCEPTED,
    510           TALER_EC_NONE,
    511           "Exchange reported 202 Accepted but no KYC block");
    512         if (qs < 0)
    513         {
    514           GNUNET_break (0);
    515           global_ret = EXIT_FAILURE;
    516           GNUNET_SCHEDULER_shutdown ();
    517           return;
    518         }
    519       }
    520       else
    521       {
    522         future_retry
    523           = GNUNET_TIME_absolute_max (
    524               future_retry,
    525               GNUNET_TIME_relative_to_absolute (
    526                 KYC_RETRY_DELAY));
    527         GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    528                     "Bumping wire transfer deadline in DB to %s as that is when we will retry\n",
    529                     GNUNET_TIME_absolute2s (future_retry));
    530         qs = TALER_MERCHANTDB_update_deposit_confirmation_status (
    531           pg,
    532           w->deposit_serial,
    533           true /* need to try again in the future */,
    534           GNUNET_TIME_absolute_to_timestamp (future_retry),
    535           MHD_HTTP_ACCEPTED,
    536           TALER_EC_NONE,
    537           "Exchange reported 202 Accepted due to KYC/AML block");
    538         if (qs < 0)
    539         {
    540           GNUNET_break (0);
    541           global_ret = EXIT_FAILURE;
    542           GNUNET_SCHEDULER_shutdown ();
    543           return;
    544         }
    545       }
    546       break;
    547     }
    548   default:
    549     {
    550       enum GNUNET_DB_QueryStatus qs;
    551       bool retry_needed = false;
    552 
    553       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    554                   "Exchange %s returned tracking failure for deposited coin %s: %u\n",
    555                   exchange_url,
    556                   TALER_B2S (&w->coin_pub),
    557                   dr->hr.http_status);
    558       /* rough classification by HTTP status group */
    559       switch (dr->hr.http_status / 100)
    560       {
    561       case 0:
    562         /* timeout */
    563         retry_needed = true;
    564         break;
    565       case 1:
    566       case 2:
    567       case 3:
    568         /* very strange */
    569         retry_needed = false;
    570         break;
    571       case 4:
    572         /* likely fatal */
    573         retry_needed = false;
    574         break;
    575       case 5:
    576         /* likely transient */
    577         retry_needed = true;
    578         break;
    579       }
    580       qs = TALER_MERCHANTDB_update_deposit_confirmation_status (
    581         pg,
    582         w->deposit_serial,
    583         retry_needed,
    584         GNUNET_TIME_absolute_to_timestamp (future_retry),
    585         (uint32_t) dr->hr.http_status,
    586         dr->hr.ec,
    587         dr->hr.hint);
    588       if (qs < 0)
    589       {
    590         GNUNET_break (0);
    591         global_ret = EXIT_FAILURE;
    592         GNUNET_SCHEDULER_shutdown ();
    593         return;
    594       }
    595       break;
    596     }
    597   } /* end switch */
    598 
    599   GNUNET_CONTAINER_DLL_remove (w_head,
    600                                w_tail,
    601                                w);
    602   w_count--;
    603   GNUNET_free (w->instance_id);
    604   GNUNET_free (w);
    605   GNUNET_assert (NULL != keys);
    606   if (0 == w_count)
    607   {
    608     /* We only SELECT() again after having finished
    609        all requests, as otherwise we'll most like
    610        just SELECT() those again that are already
    611        being requested; alternatively, we could
    612        update the retry_time already on SELECT(),
    613        but this should be easier on the DB. */
    614     if (NULL != task)
    615       GNUNET_SCHEDULER_cancel (task);
    616     task = GNUNET_SCHEDULER_add_now (&select_work,
    617                                      NULL);
    618   }
    619 }
    620 
    621 
    622 /**
    623  * Typically called by `select_work`.
    624  *
    625  * @param cls NULL
    626  * @param deposit_serial identifies the deposit operation
    627  * @param wire_deadline when is the wire due
    628  * @param retry_time current value for the retry backoff
    629  * @param h_contract_terms hash of the contract terms
    630  * @param merchant_priv private key of the merchant
    631  * @param instance_id row ID of the instance
    632  * @param h_wire hash of the merchant's wire account into
    633  * @param amount_with_fee amount the exchange will deposit for this coin
    634  * @param deposit_fee fee the exchange will charge for this coin which the deposit was made
    635  * @param coin_pub public key of the deposited coin
    636  */
    637 static void
    638 pending_deposits_cb (
    639   void *cls,
    640   uint64_t deposit_serial,
    641   struct GNUNET_TIME_Absolute wire_deadline,
    642   struct GNUNET_TIME_Absolute retry_time,
    643   const struct TALER_PrivateContractHashP *h_contract_terms,
    644   const struct TALER_MerchantPrivateKeyP *merchant_priv,
    645   const char *instance_id,
    646   const struct TALER_MerchantWireHashP *h_wire,
    647   const struct TALER_Amount *amount_with_fee,
    648   const struct TALER_Amount *deposit_fee,
    649   const struct TALER_CoinSpendPublicKeyP *coin_pub)
    650 {
    651   struct ExchangeInteraction *w;
    652   struct GNUNET_TIME_Absolute mx
    653     = GNUNET_TIME_absolute_max (wire_deadline,
    654                                 retry_time);
    655   struct GNUNET_TIME_Relative retry_backoff;
    656 
    657   (void) cls;
    658   if (GNUNET_TIME_absolute_is_future (mx))
    659   {
    660     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    661                 "Pending deposit should be checked next at %s\n",
    662                 GNUNET_TIME_absolute2s (mx));
    663     run_at (mx);
    664     return;
    665   }
    666   if (GNUNET_TIME_absolute_is_zero (retry_time))
    667     retry_backoff = GNUNET_TIME_absolute_get_duration (wire_deadline);
    668   else
    669     retry_backoff = GNUNET_TIME_absolute_get_difference (wire_deadline,
    670                                                          retry_time);
    671   w = GNUNET_new (struct ExchangeInteraction);
    672   w->deposit_serial = deposit_serial;
    673   w->wire_deadline = wire_deadline;
    674   w->retry_backoff = GNUNET_TIME_randomized_backoff (retry_backoff,
    675                                                      GNUNET_TIME_UNIT_DAYS);
    676   w->h_contract_terms = *h_contract_terms;
    677   w->merchant_priv = *merchant_priv;
    678   w->h_wire = *h_wire;
    679   w->amount_with_fee = *amount_with_fee;
    680   w->deposit_fee = *deposit_fee;
    681   w->coin_pub = *coin_pub;
    682   w->instance_id = GNUNET_strdup (instance_id);
    683   GNUNET_CONTAINER_DLL_insert (w_head,
    684                                w_tail,
    685                                w);
    686   w_count++;
    687   GNUNET_assert (NULL != keys);
    688   if (GNUNET_TIME_absolute_is_past (
    689         keys->key_data_expiration.abs_time))
    690   {
    691     /* Parent should re-start us, then we will re-fetch /keys */
    692     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    693                 "/keys expired, shutting down\n");
    694     GNUNET_SCHEDULER_shutdown ();
    695     return;
    696   }
    697   GNUNET_assert (NULL == w->dgh);
    698   w->dgh = TALER_EXCHANGE_get_deposits_create (
    699     ctx,
    700     exchange_url,
    701     keys,
    702     &w->merchant_priv,
    703     &w->h_wire,
    704     &w->h_contract_terms,
    705     &w->coin_pub);
    706   if (NULL == w->dgh)
    707   {
    708     GNUNET_break (0);
    709     GNUNET_SCHEDULER_shutdown ();
    710     return;
    711   }
    712   if (TALER_EC_NONE !=
    713       TALER_EXCHANGE_get_deposits_start (w->dgh,
    714                                          &deposit_get_cb,
    715                                          w))
    716   {
    717     GNUNET_break (0);
    718     TALER_EXCHANGE_get_deposits_cancel (w->dgh);
    719     w->dgh = NULL;
    720     GNUNET_SCHEDULER_shutdown ();
    721     return;
    722   }
    723 }
    724 
    725 
    726 /**
    727  * Function called on events received from Postgres.
    728  *
    729  * @param cls closure, NULL
    730  * @param extra additional event data provided, timestamp with wire deadline
    731  * @param extra_size number of bytes in @a extra
    732  */
    733 static void
    734 db_notify (void *cls,
    735            const void *extra,
    736            size_t extra_size)
    737 {
    738   struct GNUNET_TIME_Absolute deadline;
    739   struct GNUNET_TIME_AbsoluteNBO nbo_deadline;
    740 
    741   (void) cls;
    742   if (sizeof (nbo_deadline) != extra_size)
    743   {
    744     GNUNET_break (0);
    745     return;
    746   }
    747   if (0 != w_count)
    748     return; /* already at work! */
    749   memcpy (&nbo_deadline,
    750           extra,
    751           extra_size);
    752   deadline = GNUNET_TIME_absolute_ntoh (nbo_deadline);
    753   run_at (deadline);
    754 }
    755 
    756 
    757 static void
    758 select_work (void *cls)
    759 {
    760   bool retry = false;
    761   uint64_t limit = CONCURRENCY_LIMIT - w_count;
    762 
    763   (void) cls;
    764   task = NULL;
    765   GNUNET_assert (w_count <= CONCURRENCY_LIMIT);
    766   GNUNET_assert (NULL != keys);
    767   if (0 == limit)
    768   {
    769     GNUNET_break (0);
    770     return;
    771   }
    772   if (GNUNET_TIME_absolute_is_past (
    773         keys->key_data_expiration.abs_time))
    774   {
    775     /* Parent should re-start us, then we will re-fetch /keys */
    776     GNUNET_SCHEDULER_shutdown ();
    777     return;
    778   }
    779   while (1)
    780   {
    781     enum GNUNET_DB_QueryStatus qs;
    782 
    783     TALER_MERCHANTDB_preflight (pg);
    784     if (retry)
    785       limit = 1;
    786     qs = TALER_MERCHANTDB_lookup_pending_deposits (
    787       pg,
    788       exchange_url,
    789       limit,
    790       retry,
    791       &pending_deposits_cb,
    792       NULL);
    793     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    794                 "Looking up pending deposits query status was %d\n",
    795                 (int) qs);
    796     switch (qs)
    797     {
    798     case GNUNET_DB_STATUS_HARD_ERROR:
    799     case GNUNET_DB_STATUS_SOFT_ERROR:
    800       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
    801                   "Transaction failed!\n");
    802       global_ret = EXIT_FAILURE;
    803       GNUNET_SCHEDULER_shutdown ();
    804       return;
    805     case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
    806       if (test_mode)
    807       {
    808         GNUNET_SCHEDULER_shutdown ();
    809         return;
    810       }
    811       if (retry)
    812         return; /* nothing left */
    813       retry = true;
    814       continue;
    815     case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
    816     default:
    817       /* wait for async completion, then select more work. */
    818       return;
    819     }
    820   }
    821 }
    822 
    823 
    824 /**
    825  * Start a copy of this process with the exchange URL
    826  * set to the given @a base_url
    827  *
    828  * @param base_url base URL to run with
    829  */
    830 static struct GNUNET_Process *
    831 start_worker (const char *base_url)
    832 {
    833   struct GNUNET_Process *p;
    834   char toff[30];
    835   long long zo;
    836   enum GNUNET_GenericReturnValue ret;
    837 
    838   zo = GNUNET_TIME_get_offset ();
    839   GNUNET_snprintf (toff,
    840                    sizeof (toff),
    841                    "%lld",
    842                    zo);
    843   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
    844               "Launching worker for exchange `%s' using `%s`\n",
    845               base_url,
    846               NULL == cfg_filename
    847               ? "<default>"
    848               : cfg_filename);
    849   p = GNUNET_process_create (GNUNET_OS_INHERIT_STD_ERR);
    850 
    851   if (NULL == cfg_filename)
    852     ret = GNUNET_process_run_command_va (
    853       p,
    854       "taler-merchant-depositcheck",
    855       "taler-merchant-depositcheck",
    856       "-e", base_url,
    857       "-L", "INFO",
    858       "-T", toff,
    859       test_mode ? "-t" : NULL,
    860       NULL);
    861   else
    862     ret = GNUNET_process_run_command_va (
    863       p,
    864       "taler-merchant-depositcheck",
    865       "taler-merchant-depositcheck",
    866       "-c", cfg_filename,
    867       "-e", base_url,
    868       "-L", "INFO",
    869       "-T", toff,
    870       test_mode ? "-t" : NULL,
    871       NULL);
    872   if (GNUNET_OK != ret)
    873   {
    874     GNUNET_process_destroy (p);
    875     return NULL;
    876   }
    877   return p;
    878 }
    879 
    880 
    881 /**
    882  * Restart worker process for the given child.
    883  *
    884  * @param cls a `struct Child *` that needs a worker.
    885  */
    886 static void
    887 restart_child (void *cls);
    888 
    889 
    890 /**
    891  * Function called upon death or completion of a child process.
    892  *
    893  * @param cls a `struct Child *`
    894  * @param type type of the process
    895  * @param exit_code status code of the process
    896  */
    897 static void
    898 child_done_cb (void *cls,
    899                enum GNUNET_OS_ProcessStatusType type,
    900                long unsigned int exit_code)
    901 {
    902   struct Child *c = cls;
    903 
    904   c->cwh = NULL;
    905   GNUNET_process_destroy (c->process);
    906   c->process = NULL;
    907   if ( (GNUNET_OS_PROCESS_EXITED != type) ||
    908        (0 != exit_code) )
    909   {
    910     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
    911                 "Process for exchange %s had trouble (%d/%d)\n",
    912                 c->base_url,
    913                 (int) type,
    914                 (int) exit_code);
    915     GNUNET_SCHEDULER_shutdown ();
    916     global_ret = EXIT_NOTINSTALLED;
    917     return;
    918   }
    919   if (test_mode &&
    920       (! GNUNET_TIME_relative_is_zero (c->rd)) )
    921   {
    922     return;
    923   }
    924   if (GNUNET_TIME_absolute_is_future (c->next_start))
    925     c->rd = GNUNET_TIME_STD_BACKOFF (c->rd);
    926   else
    927     c->rd = GNUNET_TIME_UNIT_SECONDS;
    928   c->rt = GNUNET_SCHEDULER_add_at (c->next_start,
    929                                    &restart_child,
    930                                    c);
    931 }
    932 
    933 
    934 static void
    935 restart_child (void *cls)
    936 {
    937   struct Child *c = cls;
    938 
    939   c->rt = NULL;
    940   c->next_start = GNUNET_TIME_relative_to_absolute (c->rd);
    941   c->process = start_worker (c->base_url);
    942   if (NULL == c->process)
    943   {
    944     GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR,
    945                          "exec");
    946     global_ret = EXIT_NO_RESTART;
    947     GNUNET_SCHEDULER_shutdown ();
    948     return;
    949   }
    950   c->cwh = GNUNET_wait_child (c->process,
    951                               &child_done_cb,
    952                               c);
    953 }
    954 
    955 
    956 /**
    957  * Function to iterate over section.
    958  *
    959  * @param cls closure
    960  * @param section name of the section
    961  */
    962 static void
    963 cfg_iter_cb (void *cls,
    964              const char *section)
    965 {
    966   char *base_url;
    967   struct Child *c;
    968 
    969   if (0 !=
    970       strncasecmp (section,
    971                    "merchant-exchange-",
    972                    strlen ("merchant-exchange-")))
    973     return;
    974   if (GNUNET_YES ==
    975       GNUNET_CONFIGURATION_get_value_yesno (cfg,
    976                                             section,
    977                                             "DISABLED"))
    978     return;
    979   if (GNUNET_OK !=
    980       GNUNET_CONFIGURATION_get_value_string (cfg,
    981                                              section,
    982                                              "EXCHANGE_BASE_URL",
    983                                              &base_url))
    984   {
    985     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_WARNING,
    986                                section,
    987                                "EXCHANGE_BASE_URL");
    988     return;
    989   }
    990   c = GNUNET_new (struct Child);
    991   c->rd = GNUNET_TIME_UNIT_SECONDS;
    992   c->base_url = base_url;
    993   GNUNET_CONTAINER_DLL_insert (c_head,
    994                                c_tail,
    995                                c);
    996   c->rt = GNUNET_SCHEDULER_add_now (&restart_child,
    997                                     c);
    998 }
    999 
   1000 
   1001 /**
   1002  * Trigger (re)loading of keys from DB.
   1003  *
   1004  * @param cls NULL
   1005  * @param extra base URL of the exchange that changed
   1006  * @param extra_len number of bytes in @a extra
   1007  */
   1008 static void
   1009 update_exchange_keys (void *cls,
   1010                       const void *extra,
   1011                       size_t extra_len)
   1012 {
   1013   const char *url = extra;
   1014 
   1015   if ( (NULL == extra) ||
   1016        (0 == extra_len) )
   1017   {
   1018     GNUNET_break (0);
   1019     return;
   1020   }
   1021   if ('\0' != url[extra_len - 1])
   1022   {
   1023     GNUNET_break (0);
   1024     return;
   1025   }
   1026   if (0 != strcmp (url,
   1027                    exchange_url))
   1028     return; /* not relevant for us */
   1029 
   1030   {
   1031     enum GNUNET_DB_QueryStatus qs;
   1032     struct GNUNET_TIME_Absolute earliest_retry;
   1033 
   1034     if (NULL != keys)
   1035     {
   1036       TALER_EXCHANGE_keys_decref (keys);
   1037       keys = NULL;
   1038     }
   1039     qs = TALER_MERCHANTDB_select_exchange_keys (pg,
   1040                                                 exchange_url,
   1041                                                 &earliest_retry,
   1042                                                 &keys);
   1043     if (qs < 0)
   1044     {
   1045       GNUNET_break (0);
   1046       global_ret = EXIT_FAILURE;
   1047       GNUNET_SCHEDULER_shutdown ();
   1048       return;
   1049     }
   1050     if ( (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) ||
   1051          (NULL == keys) )
   1052     {
   1053       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
   1054                   "No keys yet for `%s'\n",
   1055                   exchange_url);
   1056     }
   1057   }
   1058   if (NULL == keys)
   1059   {
   1060     if (NULL != task)
   1061     {
   1062       GNUNET_SCHEDULER_cancel (task);
   1063       task = NULL;
   1064     }
   1065   }
   1066   else
   1067   {
   1068     if (NULL == task)
   1069       task = GNUNET_SCHEDULER_add_now (&select_work,
   1070                                        NULL);
   1071   }
   1072 }
   1073 
   1074 
   1075 /**
   1076  * First task.
   1077  *
   1078  * @param cls closure, NULL
   1079  * @param args remaining command-line arguments
   1080  * @param cfgfile name of the configuration file used (for saving, can be NULL!)
   1081  * @param c configuration
   1082  */
   1083 static void
   1084 run (void *cls,
   1085      char *const *args,
   1086      const char *cfgfile,
   1087      const struct GNUNET_CONFIGURATION_Handle *c)
   1088 {
   1089   (void) args;
   1090 
   1091   cfg = c;
   1092   if (NULL != cfgfile)
   1093     cfg_filename = GNUNET_strdup (cfgfile);
   1094   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
   1095               "Running with configuration %s\n",
   1096               cfgfile);
   1097   GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
   1098                                  NULL);
   1099   if (NULL == exchange_url)
   1100   {
   1101     GNUNET_CONFIGURATION_iterate_sections (c,
   1102                                            &cfg_iter_cb,
   1103                                            NULL);
   1104     if (NULL == c_head)
   1105     {
   1106       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
   1107                   "No exchanges found in configuration\n");
   1108       global_ret = EXIT_NOTCONFIGURED;
   1109       GNUNET_SCHEDULER_shutdown ();
   1110       return;
   1111     }
   1112     return;
   1113   }
   1114 
   1115   ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule,
   1116                           &rc);
   1117   rc = GNUNET_CURL_gnunet_rc_create (ctx);
   1118   if (NULL == ctx)
   1119   {
   1120     GNUNET_break (0);
   1121     GNUNET_SCHEDULER_shutdown ();
   1122     global_ret = EXIT_NO_RESTART;
   1123     return;
   1124   }
   1125   if (NULL ==
   1126       (pg = TALER_MERCHANTDB_connect (cfg)))
   1127   {
   1128     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
   1129                 "Failed to initialize DB subsystem\n");
   1130     GNUNET_SCHEDULER_shutdown ();
   1131     global_ret = EXIT_NOTCONFIGURED;
   1132     return;
   1133   }
   1134   {
   1135     struct GNUNET_DB_EventHeaderP es = {
   1136       .size = htons (sizeof (es)),
   1137       .type = htons (TALER_DBEVENT_MERCHANT_NEW_WIRE_DEADLINE)
   1138     };
   1139 
   1140     eh = TALER_MERCHANTDB_event_listen (pg,
   1141                                         &es,
   1142                                         GNUNET_TIME_UNIT_FOREVER_REL,
   1143                                         &db_notify,
   1144                                         NULL);
   1145   }
   1146   {
   1147     struct GNUNET_DB_EventHeaderP es = {
   1148       .size = htons (sizeof (es)),
   1149       .type = htons (TALER_DBEVENT_MERCHANT_EXCHANGE_KEYS)
   1150     };
   1151 
   1152     keys_eh = TALER_MERCHANTDB_event_listen (pg,
   1153                                              &es,
   1154                                              GNUNET_TIME_UNIT_FOREVER_REL,
   1155                                              &update_exchange_keys,
   1156                                              NULL);
   1157   }
   1158 
   1159   update_exchange_keys (NULL,
   1160                         exchange_url,
   1161                         strlen (exchange_url) + 1);
   1162 }
   1163 
   1164 
   1165 /**
   1166  * The main function of the taler-merchant-depositcheck
   1167  *
   1168  * @param argc number of arguments from the command line
   1169  * @param argv command line arguments
   1170  * @return 0 ok, 1 on error
   1171  */
   1172 int
   1173 main (int argc,
   1174       char *const *argv)
   1175 {
   1176   struct GNUNET_GETOPT_CommandLineOption options[] = {
   1177     GNUNET_GETOPT_option_string ('e',
   1178                                  "exchange",
   1179                                  "BASE_URL",
   1180                                  "limit us to checking deposits of this exchange",
   1181                                  &exchange_url),
   1182     GNUNET_GETOPT_option_timetravel ('T',
   1183                                      "timetravel"),
   1184     GNUNET_GETOPT_option_flag ('t',
   1185                                "test",
   1186                                "run in test mode and exit when idle",
   1187                                &test_mode),
   1188     GNUNET_GETOPT_option_version (VERSION),
   1189     GNUNET_GETOPT_OPTION_END
   1190   };
   1191   enum GNUNET_GenericReturnValue ret;
   1192 
   1193   ret = GNUNET_PROGRAM_run (
   1194     TALER_MERCHANT_project_data (),
   1195     argc, argv,
   1196     "taler-merchant-depositcheck",
   1197     gettext_noop (
   1198       "background process that checks with the exchange on deposits that are past the wire deadline"),
   1199     options,
   1200     &run, NULL);
   1201   if (GNUNET_SYSERR == ret)
   1202     return EXIT_INVALIDARGUMENT;
   1203   if (GNUNET_NO == ret)
   1204     return EXIT_SUCCESS;
   1205   return global_ret;
   1206 }
   1207 
   1208 
   1209 /* end of taler-merchant-depositcheck.c */