summaryrefslogtreecommitdiff
path: root/src/exchange/taler-exchange-closer.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/exchange/taler-exchange-closer.c')
-rw-r--r--src/exchange/taler-exchange-closer.c601
1 files changed, 601 insertions, 0 deletions
diff --git a/src/exchange/taler-exchange-closer.c b/src/exchange/taler-exchange-closer.c
new file mode 100644
index 00000000..181b231b
--- /dev/null
+++ b/src/exchange/taler-exchange-closer.c
@@ -0,0 +1,601 @@
+/*
+ This file is part of TALER
+ Copyright (C) 2016-2020 Taler Systems SA
+
+ TALER is free software; you can redistribute it and/or modify it under the
+ terms of the GNU Affero General Public License as published by the Free Software
+ Foundation; either version 3, or (at your option) any later version.
+
+ TALER is distributed in the hope that it will be useful, but WITHOUT ANY
+ WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+ A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details.
+
+ You should have received a copy of the GNU Affero General Public License along with
+ TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+*/
+
+/**
+ * @file taler-exchange-closer.c
+ * @brief Process that closes expired reserves
+ * @author Christian Grothoff
+ */
+#include "platform.h"
+#include <gnunet/gnunet_util_lib.h>
+#include <jansson.h>
+#include <pthread.h>
+#include "taler_exchangedb_lib.h"
+#include "taler_exchangedb_plugin.h"
+#include "taler_json_lib.h"
+#include "taler_bank_service.h"
+
+
+/**
+ * Which currency is used by this exchange?
+ */
+static char *exchange_currency_string;
+
+/**
+ * What is the smallest unit we support for wire transfers?
+ * We will need to round down to a multiple of this amount.
+ */
+static struct TALER_Amount currency_round_unit;
+
+/**
+ * What is the base URL of this exchange? Used in the
+ * wire transfer subjects to that merchants and governments
+ * can ask for the list of aggregated deposits.
+ */
+static char *exchange_base_url;
+
+/**
+ * The exchange's configuration.
+ */
+static const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+/**
+ * Our database plugin.
+ */
+static struct TALER_EXCHANGEDB_Plugin *db_plugin;
+
+/**
+ * Next task to run, if any.
+ */
+static struct GNUNET_SCHEDULER_Task *task;
+
+/**
+ * How long should we sleep when idle before trying to find more work?
+ */
+static struct GNUNET_TIME_Relative aggregator_idle_sleep_interval;
+
+/**
+ * Value to return from main(). #GNUNET_OK on success, #GNUNET_SYSERR
+ * on serious errors.
+ */
+static int global_ret;
+
+/**
+ * #GNUNET_YES if we are in test mode and should exit when idle.
+ */
+static int test_mode;
+
+
+/**
+ * Main work function that finds and triggers transfers for reserves
+ * closures.
+ *
+ * @param cls closure
+ */
+static void
+run_reserve_closures (void *cls);
+
+
+/**
+ * We're being aborted with CTRL-C (or SIGTERM). Shut down.
+ *
+ * @param cls closure
+ */
+static void
+shutdown_task (void *cls)
+{
+ (void) cls;
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Running shutdown\n");
+ if (NULL != task)
+ {
+ GNUNET_SCHEDULER_cancel (task);
+ task = NULL;
+ }
+ TALER_EXCHANGEDB_plugin_unload (db_plugin);
+ db_plugin = NULL;
+ TALER_EXCHANGEDB_unload_accounts ();
+ cfg = NULL;
+}
+
+
+/**
+ * Parse the configuration for wirewatch.
+ *
+ * @return #GNUNET_OK on success
+ */
+static int
+parse_wirewatch_config ()
+{
+ if (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_string (cfg,
+ "exchange",
+ "BASE_URL",
+ &exchange_base_url))
+ {
+ GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
+ "exchange",
+ "BASE_URL");
+ return GNUNET_SYSERR;
+ }
+ if (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_time (cfg,
+ "exchange",
+ "AGGREGATOR_IDLE_SLEEP_INTERVAL",
+ &aggregator_idle_sleep_interval))
+ {
+ GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
+ "exchange",
+ "AGGREGATOR_IDLE_SLEEP_INTERVAL");
+ return GNUNET_SYSERR;
+ }
+ if (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_string (cfg,
+ "taler",
+ "CURRENCY",
+ &exchange_currency_string))
+ {
+ GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
+ "taler",
+ "CURRENCY");
+ return GNUNET_SYSERR;
+ }
+ if (strlen (exchange_currency_string) >= TALER_CURRENCY_LEN)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Currency `%s' longer than the allowed limit of %u characters.",
+ exchange_currency_string,
+ (unsigned int) TALER_CURRENCY_LEN);
+ return GNUNET_SYSERR;
+ }
+
+ if ( (GNUNET_OK !=
+ TALER_config_get_amount (cfg,
+ "taler",
+ "CURRENCY_ROUND_UNIT",
+ &currency_round_unit)) ||
+ (0 != strcasecmp (exchange_currency_string,
+ currency_round_unit.currency)) ||
+ ( (0 != currency_round_unit.fraction) &&
+ (0 != currency_round_unit.value) ) )
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Invalid value specified in section `TALER' under `CURRENCY_ROUND_UNIT'\n");
+ return GNUNET_SYSERR;
+ }
+
+ if (NULL ==
+ (db_plugin = TALER_EXCHANGEDB_plugin_load (cfg)))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Failed to initialize DB subsystem\n");
+ return GNUNET_SYSERR;
+ }
+ if (GNUNET_OK !=
+ TALER_EXCHANGEDB_load_accounts (cfg))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "No wire accounts configured for debit!\n");
+ TALER_EXCHANGEDB_plugin_unload (db_plugin);
+ db_plugin = NULL;
+ return GNUNET_SYSERR;
+ }
+ return GNUNET_OK;
+}
+
+
+/**
+ * Perform a database commit. If it fails, print a warning.
+ *
+ * @param session session to perform the commit for.
+ * @return status of commit
+ */
+static enum GNUNET_DB_QueryStatus
+commit_or_warn (struct TALER_EXCHANGEDB_Session *session)
+{
+ enum GNUNET_DB_QueryStatus qs;
+
+ qs = db_plugin->commit (db_plugin->cls,
+ session);
+ if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs)
+ return qs;
+ GNUNET_log ((GNUNET_DB_STATUS_SOFT_ERROR == qs)
+ ? GNUNET_ERROR_TYPE_INFO
+ : GNUNET_ERROR_TYPE_ERROR,
+ "Failed to commit database transaction!\n");
+ return qs;
+}
+
+
+/**
+ * Closure for #expired_reserve_cb().
+ */
+struct ExpiredReserveContext
+{
+
+ /**
+ * Database session we are using.
+ */
+ struct TALER_EXCHANGEDB_Session *session;
+
+ /**
+ * Set to #GNUNET_YES if the transaction continues
+ * asynchronously.
+ */
+ int async_cont;
+};
+
+
+/**
+ * Function called with details about expired reserves.
+ * We trigger the reserve closure by inserting the respective
+ * closing record and prewire instructions into the respective
+ * tables.
+ *
+ * @param cls a `struct ExpiredReserveContext *`
+ * @param reserve_pub public key of the reserve
+ * @param left amount left in the reserve
+ * @param account_payto_uri information about the bank account that initially
+ * caused the reserve to be created
+ * @param expiration_date when did the reserve expire
+ * @return transaction status code
+ */
+static enum GNUNET_DB_QueryStatus
+expired_reserve_cb (void *cls,
+ const struct TALER_ReservePublicKeyP *reserve_pub,
+ const struct TALER_Amount *left,
+ const char *account_payto_uri,
+ struct GNUNET_TIME_Absolute expiration_date)
+{
+ struct ExpiredReserveContext *erc = cls;
+ struct TALER_EXCHANGEDB_Session *session = erc->session;
+ struct GNUNET_TIME_Absolute now;
+ struct TALER_WireTransferIdentifierRawP wtid;
+ struct TALER_Amount amount_without_fee;
+ const struct TALER_Amount *closing_fee;
+ int ret;
+ enum GNUNET_DB_QueryStatus qs;
+ struct TALER_EXCHANGEDB_WireAccount *wa;
+
+ /* NOTE: potential optimization: use custom SQL API to not
+ fetch this: */
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Processing reserve closure at %s\n",
+ GNUNET_STRINGS_absolute_time_to_string (expiration_date));
+ now = GNUNET_TIME_absolute_get ();
+ (void) GNUNET_TIME_round_abs (&now);
+
+ /* lookup account we should use */
+ wa = TALER_EXCHANGEDB_find_account_by_payto_uri (account_payto_uri);
+ if (NULL == wa)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "No wire account configured to deal with target URI `%s'\n",
+ account_payto_uri);
+ global_ret = GNUNET_SYSERR;
+ GNUNET_SCHEDULER_shutdown ();
+ return GNUNET_DB_STATUS_HARD_ERROR;
+ }
+
+ /* lookup `closing_fee` from time of actual reserve expiration
+ (we may be lagging behind!) */
+ {
+ struct TALER_EXCHANGEDB_AggregateFees *af;
+
+ af = TALER_EXCHANGEDB_update_fees (cfg,
+ db_plugin,
+ wa,
+ expiration_date,
+ session);
+ if (NULL == af)
+ {
+ global_ret = GNUNET_SYSERR;
+ GNUNET_SCHEDULER_shutdown ();
+ return GNUNET_DB_STATUS_HARD_ERROR;
+ }
+ closing_fee = &af->closing_fee;
+ }
+
+ /* calculate transfer amount */
+ ret = TALER_amount_subtract (&amount_without_fee,
+ left,
+ closing_fee);
+ if ( (GNUNET_SYSERR == ret) ||
+ (GNUNET_NO == ret) )
+ {
+ /* Closing fee higher than or equal to remaining balance, close
+ without wire transfer. */
+ closing_fee = left;
+ GNUNET_assert (GNUNET_OK ==
+ TALER_amount_get_zero (left->currency,
+ &amount_without_fee));
+ }
+ /* round down to enable transfer */
+ if (GNUNET_SYSERR ==
+ TALER_amount_round_down (&amount_without_fee,
+ &currency_round_unit))
+ {
+ GNUNET_break (0);
+ global_ret = GNUNET_SYSERR;
+ GNUNET_SCHEDULER_shutdown ();
+ return GNUNET_DB_STATUS_HARD_ERROR;
+ }
+ if ( (0 == amount_without_fee.value) &&
+ (0 == amount_without_fee.fraction) )
+ ret = GNUNET_NO;
+
+ /* NOTE: sizeof (*reserve_pub) == sizeof (wtid) right now, but to
+ be future-compatible, we use the memset + min construction */
+ memset (&wtid,
+ 0,
+ sizeof (wtid));
+ memcpy (&wtid,
+ reserve_pub,
+ GNUNET_MIN (sizeof (wtid),
+ sizeof (*reserve_pub)));
+ if (GNUNET_SYSERR != ret)
+ qs = db_plugin->insert_reserve_closed (db_plugin->cls,
+ session,
+ reserve_pub,
+ now,
+ account_payto_uri,
+ &wtid,
+ left,
+ closing_fee);
+ else
+ qs = GNUNET_DB_STATUS_HARD_ERROR;
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Closing reserve %s over %s (%d, %d)\n",
+ TALER_B2S (reserve_pub),
+ TALER_amount2s (left),
+ ret,
+ qs);
+ /* Check for hard failure */
+ if ( (GNUNET_SYSERR == ret) ||
+ (GNUNET_DB_STATUS_HARD_ERROR == qs) )
+ {
+ GNUNET_break (0);
+ global_ret = GNUNET_SYSERR;
+ GNUNET_SCHEDULER_shutdown ();
+ return GNUNET_DB_STATUS_HARD_ERROR;
+ }
+ if ( (GNUNET_OK != ret) ||
+ (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != qs) )
+ {
+ /* Reserve balance was almost zero OR soft error */
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Reserve was virtually empty, moving on\n");
+ (void) commit_or_warn (session);
+ erc->async_cont = GNUNET_YES;
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
+ NULL);
+ return qs;
+ }
+
+ /* success, perform wire transfer */
+ {
+ char *method;
+ void *buf;
+ size_t buf_size;
+
+ method = TALER_payto_get_method (account_payto_uri);
+ TALER_BANK_prepare_transfer (account_payto_uri,
+ &amount_without_fee,
+ exchange_base_url,
+ &wtid,
+ &buf,
+ &buf_size);
+ /* Commit our intention to execute the wire transfer! */
+ qs = db_plugin->wire_prepare_data_insert (db_plugin->cls,
+ session,
+ method,
+ buf,
+ buf_size);
+ GNUNET_free (buf);
+ GNUNET_free (method);
+ }
+ if (GNUNET_DB_STATUS_HARD_ERROR == qs)
+ {
+ GNUNET_break (0);
+ return GNUNET_DB_STATUS_HARD_ERROR;
+ }
+ if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
+ {
+ /* start again */
+ return GNUNET_DB_STATUS_SUCCESS_NO_RESULTS;
+ }
+ erc->async_cont = GNUNET_YES;
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
+ NULL);
+ return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
+}
+
+
+/**
+ * Main work function that finds and triggers transfers for reserves
+ * closures.
+ *
+ * @param cls closure
+ */
+static void
+run_reserve_closures (void *cls)
+{
+ struct TALER_EXCHANGEDB_Session *session;
+ enum GNUNET_DB_QueryStatus qs;
+ const struct GNUNET_SCHEDULER_TaskContext *tc;
+ struct ExpiredReserveContext erc;
+ struct GNUNET_TIME_Absolute now;
+
+ (void) cls;
+ task = NULL;
+ tc = GNUNET_SCHEDULER_get_task_context ();
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
+ return;
+ if (NULL == (session = db_plugin->get_session (db_plugin->cls)))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Failed to obtain database session!\n");
+ global_ret = GNUNET_SYSERR;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+
+ if (GNUNET_OK !=
+ db_plugin->start (db_plugin->cls,
+ session,
+ "aggregator reserve closures"))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Failed to start database transaction!\n");
+ global_ret = GNUNET_SYSERR;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ erc.session = session;
+ erc.async_cont = GNUNET_NO;
+ now = GNUNET_TIME_absolute_get ();
+ (void) GNUNET_TIME_round_abs (&now);
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Checking for reserves to close by date %s\n",
+ GNUNET_STRINGS_absolute_time_to_string (now));
+ qs = db_plugin->get_expired_reserves (db_plugin->cls,
+ session,
+ now,
+ &expired_reserve_cb,
+ &erc);
+ GNUNET_assert (1 >= qs);
+ switch (qs)
+ {
+ case GNUNET_DB_STATUS_HARD_ERROR:
+ GNUNET_break (0);
+ db_plugin->rollback (db_plugin->cls,
+ session);
+ global_ret = GNUNET_SYSERR;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ case GNUNET_DB_STATUS_SOFT_ERROR:
+ db_plugin->rollback (db_plugin->cls,
+ session);
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
+ NULL);
+ return;
+ case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "No more idle reserves, going back to aggregation\n");
+ db_plugin->rollback (db_plugin->cls,
+ session);
+ GNUNET_assert (NULL == task);
+ if (GNUNET_YES == test_mode)
+ {
+ GNUNET_SCHEDULER_shutdown ();
+ }
+ else
+ {
+ task = GNUNET_SCHEDULER_add_delayed (aggregator_idle_sleep_interval,
+ &run_reserve_closures,
+ NULL);
+ }
+ return;
+ case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+ (void) commit_or_warn (session);
+ if (GNUNET_YES == erc.async_cont)
+ break;
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
+ NULL);
+ return;
+ }
+}
+
+
+/**
+ * First task.
+ *
+ * @param cls closure, NULL
+ * @param args remaining command-line arguments
+ * @param cfgfile name of the configuration file used (for saving, can be NULL!)
+ * @param c configuration
+ */
+static void
+run (void *cls,
+ char *const *args,
+ const char *cfgfile,
+ const struct GNUNET_CONFIGURATION_Handle *c)
+{
+ (void) cls;
+ (void) args;
+ (void) cfgfile;
+
+ cfg = c;
+ if (GNUNET_OK != parse_wirewatch_config ())
+ {
+ cfg = NULL;
+ global_ret = 1;
+ return;
+ }
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
+ NULL);
+ GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
+ cls);
+}
+
+
+/**
+ * The main function of the taler-exchange-closer.
+ *
+ * @param argc number of arguments from the command line
+ * @param argv command line arguments
+ * @return 0 ok, 1 on error
+ */
+int
+main (int argc,
+ char *const *argv)
+{
+ struct GNUNET_GETOPT_CommandLineOption options[] = {
+ GNUNET_GETOPT_option_timetravel ('T',
+ "timetravel"),
+ GNUNET_GETOPT_option_flag ('t',
+ "test",
+ "run in test mode and exit when idle",
+ &test_mode),
+ GNUNET_GETOPT_option_version (VERSION "-" VCS_VERSION),
+ GNUNET_GETOPT_OPTION_END
+ };
+
+ if (GNUNET_OK != GNUNET_STRINGS_get_utf8_args (argc, argv,
+ &argc, &argv))
+ return 2;
+ if (GNUNET_OK !=
+ GNUNET_PROGRAM_run (argc, argv,
+ "taler-exchange-closer",
+ gettext_noop (
+ "background process that closes expired reserves"),
+ options,
+ &run, NULL))
+ {
+ GNUNET_free ((void *) argv);
+ return 1;
+ }
+ GNUNET_free ((void *) argv);
+ return global_ret;
+}
+
+
+/* end of taler-exchange-closer.c */