summaryrefslogtreecommitdiff
path: root/src/exchange/taler-exchange-wirewatch.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/exchange/taler-exchange-wirewatch.c')
-rw-r--r--src/exchange/taler-exchange-wirewatch.c1044
1 files changed, 521 insertions, 523 deletions
diff --git a/src/exchange/taler-exchange-wirewatch.c b/src/exchange/taler-exchange-wirewatch.c
index 61af32dce..da5d9c098 100644
--- a/src/exchange/taler-exchange-wirewatch.c
+++ b/src/exchange/taler-exchange-wirewatch.c
@@ -1,6 +1,6 @@
/*
This file is part of TALER
- Copyright (C) 2016--2022 Taler Systems SA
+ Copyright (C) 2016--2023 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
@@ -13,7 +13,6 @@
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-wirewatch.c
* @brief Process that watches for wire transfers to the exchange's bank account
@@ -43,122 +42,111 @@
#define MAXIMUM_BATCH_SIZE 1024
/**
- * Information we keep for each supported account.
+ * Information about our account.
*/
-struct WireAccount
-{
- /**
- * Accounts are kept in a DLL.
- */
- struct WireAccount *next;
-
- /**
- * Plugins are kept in a DLL.
- */
- struct WireAccount *prev;
-
- /**
- * Information about this account.
- */
- const struct TALER_EXCHANGEDB_AccountInfo *ai;
-
- /**
- * Active request for history.
- */
- struct TALER_BANK_CreditHistoryHandle *hh;
-
- /**
- * Until when is processing this wire plugin delayed?
- */
- struct GNUNET_TIME_Absolute delayed_until;
-
- /**
- * Encoded offset in the wire transfer list from where
- * to start the next query with the bank.
- */
- uint64_t batch_start;
-
- /**
- * Latest row offset seen in this transaction, becomes
- * the new #batch_start upon commit.
- */
- uint64_t latest_row_off;
-
- /**
- * Maximum row offset this transaction may yield. If we got the
- * maximum number of rows, we must not @e delay before running
- * the next transaction.
- */
- uint64_t max_row_off;
-
- /**
- * Offset where our current shard begins (inclusive).
- */
- uint64_t shard_start;
-
- /**
- * Offset where our current shard ends (exclusive).
- */
- uint64_t shard_end;
-
- /**
- * When did we start with the shard?
- */
- struct GNUNET_TIME_Absolute shard_start_time;
-
- /**
- * For how long did we lock the shard?
- */
- struct GNUNET_TIME_Absolute shard_end_time;
-
- /**
- * How long did we take to finish the last shard
- * for this account?
- */
- struct GNUNET_TIME_Relative shard_delay;
-
- /**
- * Name of our job in the shard table.
- */
- char *job_name;
-
- /**
- * How many transactions do we retrieve per batch?
- */
- unsigned int batch_size;
-
- /**
- * How much do we increment @e batch_size on success?
- */
- unsigned int batch_thresh;
-
- /**
- * Should we delay the next request to the wire plugin a bit? Set to
- * false if we actually did some work.
- */
- bool delay;
-
- /**
- * Did we start a transaction yet?
- */
- bool started_transaction;
-
- /**
- * Is this shard still open for processing.
- */
- bool shard_open;
-};
+static const struct TALER_EXCHANGEDB_AccountInfo *ai;
+
+/**
+ * Active request for history.
+ */
+static struct TALER_BANK_CreditHistoryHandle *hh;
+
+/**
+ * Set to true if the request for history did actually
+ * return transaction items.
+ */
+static bool hh_returned_data;
+
+/**
+ * Set to true if the request for history did not
+ * succeed because the account was unknown.
+ */
+static bool hh_account_404;
+
+/**
+ * When did we start the last @e hh request?
+ */
+static struct GNUNET_TIME_Absolute hh_start_time;
+
+/**
+ * Until when is processing this wire plugin delayed?
+ */
+static struct GNUNET_TIME_Absolute delayed_until;
+
+/**
+ * Encoded offset in the wire transfer list from where
+ * to start the next query with the bank.
+ */
+static uint64_t batch_start;
+
+/**
+ * Latest row offset seen in this transaction, becomes
+ * the new #batch_start upon commit.
+ */
+static uint64_t latest_row_off;
+
+/**
+ * Offset where our current shard begins (inclusive).
+ */
+static uint64_t shard_start;
+
+/**
+ * Offset where our current shard ends (exclusive).
+ */
+static uint64_t shard_end;
+
+/**
+ * When did we start with the shard?
+ */
+static struct GNUNET_TIME_Absolute shard_start_time;
+
+/**
+ * For how long did we lock the shard?
+ */
+static struct GNUNET_TIME_Absolute shard_end_time;
+
+/**
+ * How long did we take to finish the last shard
+ * for this account?
+ */
+static struct GNUNET_TIME_Relative shard_delay;
+
+/**
+ * How long did we take to finish the last shard
+ * for this account?
+ */
+static struct GNUNET_TIME_Relative longpoll_timeout;
+
+/**
+ * Name of our job in the shard table.
+ */
+static char *job_name;
+
+/**
+ * How many transactions do we retrieve per batch?
+ */
+static unsigned int batch_size;
+
+/**
+ * How much do we increment @e batch_size on success?
+ */
+static unsigned int batch_thresh;
+/**
+ * Did work remain in the transaction queue? Set to true
+ * if we did some work and thus there might be more.
+ */
+static bool progress;
/**
- * Head of list of loaded wire plugins.
+ * Did we start a transaction yet?
*/
-static struct WireAccount *wa_head;
+static bool started_transaction;
/**
- * Tail of list of loaded wire plugins.
+ * Is this shard still open for processing.
*/
-static struct WireAccount *wa_tail;
+static bool shard_open;
/**
* Handle to the context for interacting with the bank.
@@ -189,6 +177,11 @@ static struct TALER_EXCHANGEDB_Plugin *db_plugin;
static struct GNUNET_TIME_Relative wirewatch_idle_sleep_interval;
/**
+ * How long do we sleep on serialization conflicts?
+ */
+static struct GNUNET_TIME_Relative wirewatch_conflict_sleep_interval;
+
+/**
* Modulus to apply to group shards. The shard size must ultimately be a
* multiple of the batch size. Thus, if this is not a multiple of the
* #MAXIMUM_BATCH_SIZE, the batch size will be set to the #shard_size.
@@ -227,6 +220,10 @@ static int ignore_account_404;
*/
static struct GNUNET_SCHEDULER_Task *task;
+/**
+ * Name of the configuration section with the account we should watch.
+ */
+static char *account_section;
/**
* We're being aborted with CTRL-C (or SIGTERM). Shut down.
@@ -236,38 +233,32 @@ static struct GNUNET_SCHEDULER_Task *task;
static void
shutdown_task (void *cls)
{
+ enum GNUNET_DB_QueryStatus qs;
(void) cls;
- {
- struct WireAccount *wa;
- while (NULL != (wa = wa_head))
- {
- enum GNUNET_DB_QueryStatus qs;
-
- if (NULL != wa->hh)
- {
- TALER_BANK_credit_history_cancel (wa->hh);
- wa->hh = NULL;
- }
- GNUNET_CONTAINER_DLL_remove (wa_head,
- wa_tail,
- wa);
- if (wa->started_transaction)
- {
- db_plugin->rollback (db_plugin->cls);
- wa->started_transaction = false;
- }
- qs = db_plugin->abort_shard (db_plugin->cls,
- wa->job_name,
- wa->shard_start,
- wa->shard_end);
- if (qs <= 0)
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Failed to abort work shard on shutdown\n");
- GNUNET_free (wa->job_name);
- GNUNET_free (wa);
- }
+ if (NULL != hh)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "History request cancelled on shutdown\n");
+ TALER_BANK_credit_history_cancel (hh);
+ hh = NULL;
}
+ if (started_transaction)
+ {
+ db_plugin->rollback (db_plugin->cls);
+ started_transaction = false;
+ }
+ if (shard_open)
+ {
+ qs = db_plugin->abort_shard (db_plugin->cls,
+ job_name,
+ shard_start,
+ shard_end);
+ if (qs <= 0)
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Failed to abort work shard on shutdown\n");
+ }
+ GNUNET_free (job_name);
if (NULL != ctx)
{
GNUNET_CURL_fini (ctx);
@@ -295,28 +286,36 @@ shutdown_task (void *cls)
* account to our list (if it is enabled and we can load the plugin).
*
* @param cls closure, NULL
- * @param ai account information
+ * @param in_ai account information
*/
static void
add_account_cb (void *cls,
- const struct TALER_EXCHANGEDB_AccountInfo *ai)
+ const struct TALER_EXCHANGEDB_AccountInfo *in_ai)
{
- struct WireAccount *wa;
-
(void) cls;
- if (! ai->credit_enabled)
+ if (! in_ai->credit_enabled)
return; /* not enabled for us, skip */
- wa = GNUNET_new (struct WireAccount);
- wa->ai = ai;
- GNUNET_asprintf (&wa->job_name,
+ if ( (NULL != account_section) &&
+ (0 != strcasecmp (in_ai->section_name,
+ account_section)) )
+ return; /* not enabled for us, skip */
+ if (NULL != ai)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Multiple accounts enabled (%s and %s), use '-a' command-line option to select one!\n",
+ ai->section_name,
+ in_ai->section_name);
+ GNUNET_SCHEDULER_shutdown ();
+ global_ret = EXIT_INVALIDARGUMENT;
+ return;
+ }
+ ai = in_ai;
+ GNUNET_asprintf (&job_name,
"wirewatch-%s",
ai->section_name);
- wa->batch_size = MAXIMUM_BATCH_SIZE;
- if (0 != shard_size % wa->batch_size)
- wa->batch_size = shard_size;
- GNUNET_CONTAINER_DLL_insert (wa_head,
- wa_tail,
- wa);
+ batch_size = MAXIMUM_BATCH_SIZE;
+ if (0 != shard_size % batch_size)
+ batch_size = shard_size;
}
@@ -354,13 +353,17 @@ exchange_serve_process_config (void)
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"No wire accounts configured for credit!\n");
- TALER_EXCHANGEDB_plugin_unload (db_plugin);
- db_plugin = NULL;
return GNUNET_SYSERR;
}
TALER_EXCHANGEDB_find_accounts (&add_account_cb,
NULL);
- GNUNET_assert (NULL != wa_head);
+ if (NULL == ai)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "No accounts enabled for credit!\n");
+ GNUNET_SCHEDULER_shutdown ();
+ return GNUNET_SYSERR;
+ }
return GNUNET_OK;
}
@@ -368,449 +371,398 @@ exchange_serve_process_config (void)
/**
* Lock a shard and then begin to query for incoming wire transfers.
*
- * @param cls a `struct WireAccount` to operate on
+ * @param cls NULL
*/
static void
lock_shard (void *cls);
/**
- * Continue with the credit history of the shard
- * reserved as @a wa.
+ * Continue with the credit history of the shard.
*
- * @param[in,out] cls `struct WireAccount *` account with shard to continue processing
+ * @param cls NULL
*/
static void
continue_with_shard (void *cls);
/**
- * We encountered a serialization error.
- * Rollback the transaction and try again
- *
- * @param wa account we are transacting on
+ * We encountered a serialization error. Rollback the transaction and try
+ * again.
*/
static void
-handle_soft_error (struct WireAccount *wa)
+handle_soft_error (void)
{
db_plugin->rollback (db_plugin->cls);
- wa->started_transaction = false;
- if (1 < wa->batch_size)
+ started_transaction = false;
+ if (1 < batch_size)
{
- wa->batch_thresh = wa->batch_size;
- wa->batch_size /= 2;
+ batch_thresh = batch_size;
+ batch_size /= 2;
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Reduced batch size to %llu due to serialization issue\n",
- (unsigned long long) wa->batch_size);
+ (unsigned long long) batch_size);
}
/* Reset to beginning of transaction, and go again
from there. */
- wa->latest_row_off = wa->batch_start;
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Encountered soft error, resetting start point to batch start\n");
+ latest_row_off = batch_start;
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
- wa);
+ NULL);
}
/**
- * Schedule the #lock_shard() operation for
- * @a wa. If @a wa is NULL, start with #wa_head.
- *
- * @param wa account to schedule #lock_shard() for,
- * possibly NULL (!).
+ * Schedule the #lock_shard() operation.
*/
static void
-schedule_transfers (struct WireAccount *wa)
+schedule_transfers (void)
{
- if (NULL == wa)
- {
- wa = wa_head;
- GNUNET_assert (NULL != wa);
- }
- if (wa->shard_open)
+ if (shard_open)
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Will retry my shard (%llu,%llu] of %s in %s\n",
- (unsigned long long) wa->shard_start,
- (unsigned long long) wa->shard_end,
- wa->job_name,
+ (unsigned long long) shard_start,
+ (unsigned long long) shard_end,
+ job_name,
GNUNET_STRINGS_relative_time_to_string (
- GNUNET_TIME_absolute_get_remaining (wa->delayed_until),
- GNUNET_YES));
+ GNUNET_TIME_absolute_get_remaining (delayed_until),
+ true));
else
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Will try to lock next shard of %s in %s\n",
- wa->job_name,
+ job_name,
GNUNET_STRINGS_relative_time_to_string (
- GNUNET_TIME_absolute_get_remaining (wa->delayed_until),
- GNUNET_YES));
+ GNUNET_TIME_absolute_get_remaining (delayed_until),
+ true));
GNUNET_assert (NULL == task);
- task = GNUNET_SCHEDULER_add_at (wa->delayed_until,
+ task = GNUNET_SCHEDULER_add_at (delayed_until,
&lock_shard,
- wa);
+ NULL);
}
/**
- * We are done with the work that is possible on @a wa right now (and the
- * transaction was committed, if there was one to commit). Move on to the next
- * account.
- *
- * @param wa wire account for which we completed a shard
+ * We are done with the work that is possible right now (and the transaction
+ * was committed, if there was one to commit). Move on to the next shard.
*/
static void
-account_completed (struct WireAccount *wa)
+transaction_completed (void)
{
- GNUNET_assert (! wa->started_transaction);
- if ( (wa->batch_start + wa->batch_size ==
- wa->latest_row_off) &&
- (wa->batch_size < MAXIMUM_BATCH_SIZE) )
+ if ( (batch_start + batch_size ==
+ latest_row_off) &&
+ (batch_size < MAXIMUM_BATCH_SIZE) )
{
/* The current batch size worked without serialization
issues, and we are allowed to grow. Do so slowly. */
int delta;
- delta = ((int) wa->batch_thresh - (int) wa->batch_size) / 4;
+ delta = ((int) batch_thresh - (int) batch_size) / 4;
if (delta < 0)
delta = -delta;
- wa->batch_size = GNUNET_MIN (MAXIMUM_BATCH_SIZE,
- wa->batch_size + delta + 1);
+ batch_size = GNUNET_MIN (MAXIMUM_BATCH_SIZE,
+ batch_size + delta + 1);
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Increasing batch size to %llu\n",
- (unsigned long long) wa->batch_size);
- }
-
- if (wa->delay)
- {
- /* This account was finished, block this one for the
- #wirewatch_idle_sleep_interval and move on to the next one. */
- wa->delayed_until
- = GNUNET_TIME_relative_to_absolute (wirewatch_idle_sleep_interval);
- wa = wa->next;
+ (unsigned long long) batch_size);
}
- GNUNET_assert (NULL == task);
- schedule_transfers (wa);
-}
-
-
-/**
- * Check if we are finished with the current shard. If so, update the
- * database, marking the shard as finished.
- *
- * @param wa wire account to commit for
- * @return true if we were indeed done with the shard
- */
-static bool
-check_shard_done (struct WireAccount *wa)
-{
- enum GNUNET_DB_QueryStatus qs;
- if (wa->shard_end > wa->latest_row_off)
+ if ( (! progress) && test_mode)
{
+ /* Transaction list was drained and we are in
+ test mode. So we are done. */
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Shard %s (%llu,%llu] at %llu\n",
- wa->job_name,
- (unsigned long long) wa->shard_start,
- (unsigned long long) wa->shard_end,
- (unsigned long long) wa->latest_row_off);
- return false; /* actually, not done! */
+ "Transaction list drained and in test mode. Exiting\n");
+ GNUNET_SCHEDULER_shutdown ();
+ return;
}
- /* shard is complete, mark this as well */
- qs = db_plugin->complete_shard (db_plugin->cls,
- wa->job_name,
- wa->shard_start,
- wa->shard_end);
- switch (qs)
+ if (! (hh_returned_data || hh_account_404) )
{
- case GNUNET_DB_STATUS_HARD_ERROR:
- GNUNET_break (0);
- db_plugin->rollback (db_plugin->cls);
- GNUNET_SCHEDULER_shutdown ();
- return false;
- case GNUNET_DB_STATUS_SOFT_ERROR:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Got DB soft error for complete_shard. Rolling back.\n");
- handle_soft_error (wa);
- return false;
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- GNUNET_break (0);
- /* Not expected, but let's just continue */
- break;
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- /* normal case */
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Completed shard %s (%llu,%llu] after %s\n",
- wa->job_name,
- (unsigned long long) wa->shard_start,
- (unsigned long long) wa->shard_end,
- GNUNET_STRINGS_relative_time_to_string (
- GNUNET_TIME_absolute_get_duration (wa->shard_start_time),
- GNUNET_YES));
- break;
+ /* Enforce long-polling delay even if the server ignored it
+ and returned earlier */
+ struct GNUNET_TIME_Relative latency;
+ struct GNUNET_TIME_Relative left;
+
+ latency = GNUNET_TIME_absolute_get_duration (hh_start_time);
+ left = GNUNET_TIME_relative_subtract (longpoll_timeout,
+ latency);
+ if (! (test_mode ||
+ GNUNET_TIME_relative_is_zero (left)) )
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Server did not respect long-polling, enforcing client-side by sleeping for %s\n",
+ GNUNET_TIME_relative2s (left,
+ true));
+ delayed_until = GNUNET_TIME_relative_to_absolute (left);
}
- return true;
+ if (hh_account_404)
+ delayed_until = GNUNET_TIME_relative_to_absolute (
+ GNUNET_TIME_UNIT_MILLISECONDS);
+ if (test_mode)
+ delayed_until = GNUNET_TIME_UNIT_ZERO_ABS;
+ GNUNET_assert (NULL == task);
+ schedule_transfers ();
}
/**
- * We are finished with the current transaction, try
- * to commit and then schedule the next iteration.
+ * We got incoming transaction details from the bank. Add them
+ * to the database.
*
- * @param wa wire account to commit for
+ * @param details array of transaction details
+ * @param details_length length of the @a details array
*/
static void
-do_commit (struct WireAccount *wa)
+process_reply (const struct TALER_BANK_CreditDetails *details,
+ unsigned int details_length)
{
enum GNUNET_DB_QueryStatus qs;
bool shard_done;
+ uint64_t lroff = latest_row_off;
- GNUNET_assert (NULL == task);
- shard_done = check_shard_done (wa);
- wa->started_transaction = false;
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Committing %s progress (%llu,%llu] at %llu\n (%s)",
- wa->job_name,
- (unsigned long long) wa->shard_start,
- (unsigned long long) wa->shard_end,
- (unsigned long long) wa->latest_row_off,
- shard_done
- ? "shard done"
- : "shard incomplete");
- qs = db_plugin->commit (db_plugin->cls);
- switch (qs)
+ if (0 == details_length)
{
- case GNUNET_DB_STATUS_HARD_ERROR:
- GNUNET_break (0);
- GNUNET_SCHEDULER_shutdown ();
- return;
- case GNUNET_DB_STATUS_SOFT_ERROR:
- /* reduce transaction size to reduce rollback probability */
- handle_soft_error (wa);
+ /* Server should have used 204, not 200! */
+ GNUNET_break_op (0);
+ transaction_completed ();
return;
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- /* normal case */
- break;
}
- if (shard_done)
- {
- wa->shard_delay = GNUNET_TIME_absolute_get_duration (wa->shard_start_time);
- wa->shard_open = false;
- account_completed (wa);
- }
- else
+ hh_returned_data = true;
+ /* check serial IDs for range constraints */
+ for (unsigned int i = 0; i<details_length; i++)
{
- task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
- wa);
- }
-}
-
-
-/**
- * Callbacks of this type are used to serve the result of asking
- * the bank for the transaction history.
- *
- * @param cls closure with the `struct WioreAccount *` we are processing
- * @param http_status HTTP status code from the server
- * @param ec taler error code
- * @param serial_id identification of the position at which we are querying
- * @param details details about the wire transfer
- * @param json raw JSON response
- * @return #GNUNET_OK to continue, #GNUNET_SYSERR to abort iteration
- */
-static enum GNUNET_GenericReturnValue
-history_cb (void *cls,
- unsigned int http_status,
- enum TALER_ErrorCode ec,
- uint64_t serial_id,
- const struct TALER_BANK_CreditDetails *details,
- const json_t *json)
-{
- struct WireAccount *wa = cls;
- enum GNUNET_DB_QueryStatus qs;
+ const struct TALER_BANK_CreditDetails *cd = &details[i];
- (void) json;
- GNUNET_assert (NULL == task);
- if (NULL == details)
- {
- wa->hh = NULL;
- if ( (! ( (MHD_HTTP_NOT_FOUND == http_status) &&
- (ignore_account_404) ) ) &&
- ( (MHD_HTTP_NO_CONTENT != http_status) &&
- ( (TALER_EC_NONE != ec) ||
- (MHD_HTTP_OK != http_status) ) ) )
+ if (cd->serial_id < lroff)
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
- "Error fetching history: %s (%u)\n",
- TALER_ErrorCode_get_hint (ec),
- http_status);
- if (! (exit_on_error || test_mode) )
- {
- account_completed (wa);
- return GNUNET_OK;
- }
+ "Serial ID %llu not monotonic (got %llu before). Failing!\n",
+ (unsigned long long) cd->serial_id,
+ (unsigned long long) lroff);
+ db_plugin->rollback (db_plugin->cls);
GNUNET_SCHEDULER_shutdown ();
- return GNUNET_OK;
+ return;
}
- if (wa->started_transaction)
+ if (cd->serial_id > shard_end)
{
+ /* we are *past* the current shard (likely because the serial_id of the
+ shard_end happens to not exist in the DB). So commit and stop this
+ iteration! */
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "End of list. Committing progress on %s of (%llu,%llu]!\n",
- wa->job_name,
- (unsigned long long) wa->batch_start,
- (unsigned long long) wa->latest_row_off);
- do_commit (wa);
- return GNUNET_OK; /* will be ignored anyway */
+ "Serial ID %llu past shard end at %llu, ending iteration early!\n",
+ (unsigned long long) cd->serial_id,
+ (unsigned long long) shard_end);
+ details_length = i;
+ progress = true;
+ lroff = cd->serial_id - 1;
+ break;
}
- /* We did not even start a transaction. */
- if ( (wa->delay) &&
- (test_mode) &&
- (NULL == wa->next) )
- {
- /* We exit on idle */
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Shutdown due to test mode!\n");
- GNUNET_SCHEDULER_shutdown ();
- return GNUNET_OK;
- }
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "No transactions in history response, moving on.\n");
- account_completed (wa);
- return GNUNET_OK; /* will be ignored anyway */
- }
-
- /* We did get 'details' from the bank. Do sanity checks before inserting. */
- if (serial_id < wa->latest_row_off)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
- "Serial ID %llu not monotonic (got %llu before). Failing!\n",
- (unsigned long long) serial_id,
- (unsigned long long) wa->latest_row_off);
- GNUNET_SCHEDULER_shutdown ();
- wa->hh = NULL;
- return GNUNET_SYSERR;
+ lroff = cd->serial_id;
}
- /* If we got 'limit' transactions back from the bank,
- we should not introduce any delay before the next
- call. */
- if (serial_id >= wa->max_row_off)
- wa->delay = false;
- if (serial_id > wa->shard_end)
+ if (0 != details_length)
{
- /* we are *past* the current shard (likely because the serial_id of the
- shard_end happens to not exist in the DB). So commit and stop this
- iteration! */
+ enum GNUNET_DB_QueryStatus qss[details_length];
+ struct TALER_EXCHANGEDB_ReserveInInfo reserves[details_length];
+
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Serial ID %llu past shard end at %llu, ending iteration early!\n",
- (unsigned long long) serial_id,
- (unsigned long long) wa->shard_end);
- wa->latest_row_off = serial_id - 1; /* excluding serial_id! */
- wa->hh = NULL;
- if (wa->started_transaction)
+ "Importing %u transactions\n",
+ details_length);
+ for (unsigned int i = 0; i<details_length; i++)
{
- GNUNET_assert (NULL == task);
- do_commit (wa);
+ const struct TALER_BANK_CreditDetails *cd = &details[i];
+ struct TALER_EXCHANGEDB_ReserveInInfo *res = &reserves[i];
+
+ res->reserve_pub = &cd->reserve_pub;
+ res->balance = &cd->amount;
+ res->execution_time = cd->execution_date;
+ res->sender_account_details = cd->debit_account_uri;
+ res->exchange_account_name = ai->section_name;
+ res->wire_reference = cd->serial_id;
}
- else
+ qs = db_plugin->reserves_in_insert (db_plugin->cls,
+ reserves,
+ details_length,
+ qss);
+ switch (qs)
{
- GNUNET_assert (NULL == task);
- if (check_shard_done (wa))
- account_completed (wa);
- else
- task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
- wa);
+ case GNUNET_DB_STATUS_HARD_ERROR:
+ GNUNET_break (0);
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ case GNUNET_DB_STATUS_SOFT_ERROR:
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Got DB soft error for reserves_in_insert (%u). Rolling back.\n",
+ details_length);
+ handle_soft_error ();
+ return;
+ default:
+ break;
+ }
+ for (unsigned int i = 0; i<details_length; i++)
+ {
+ const struct TALER_BANK_CreditDetails *cd = &details[i];
+
+ switch (qss[i])
+ {
+ case GNUNET_DB_STATUS_HARD_ERROR:
+ GNUNET_break (0);
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ case GNUNET_DB_STATUS_SOFT_ERROR:
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Got DB soft error for batch_reserves_in_insert(%u). Rolling back.\n",
+ i);
+ handle_soft_error ();
+ return;
+ case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+ /* Either wirewatch was freshly started after the system was
+ shutdown and we're going over an incomplete shard again
+ after being restarted, or the shard lock period was too
+ short (number of workers set incorrectly?) and a 2nd
+ wirewatcher has been stealing our work while we are still
+ at it. */
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Attempted to import transaction %llu (%s) twice. "
+ "This should happen rarely (if not, ask for support).\n",
+ (unsigned long long) cd->serial_id,
+ job_name);
+ break;
+ case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Imported transaction %llu.\n",
+ (unsigned long long) cd->serial_id);
+ /* normal case */
+ progress = true;
+ break;
+ }
}
- return GNUNET_SYSERR;
}
- if (! wa->started_transaction)
+
+ latest_row_off = lroff;
+ shard_done = (shard_end <= latest_row_off);
+ if (shard_done)
{
- if (GNUNET_OK !=
- db_plugin->start_read_committed (db_plugin->cls,
- "wirewatch check for incoming wire transfers"))
+ /* shard is complete, mark this as well */
+ qs = db_plugin->complete_shard (db_plugin->cls,
+ job_name,
+ shard_start,
+ shard_end);
+ switch (qs)
{
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
- "Failed to start database transaction!\n");
- global_ret = EXIT_FAILURE;
+ case GNUNET_DB_STATUS_HARD_ERROR:
+ GNUNET_break (0);
GNUNET_SCHEDULER_shutdown ();
- wa->hh = NULL;
- return GNUNET_SYSERR;
+ return;
+ case GNUNET_DB_STATUS_SOFT_ERROR:
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Got DB soft error for complete_shard. Rolling back.\n");
+ handle_soft_error ();
+ return;
+ case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+ GNUNET_break (0);
+ /* Not expected, but let's just continue */
+ break;
+ case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+ /* normal case */
+ progress = true;
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Completed shard %s (%llu,%llu] after %s\n",
+ job_name,
+ (unsigned long long) shard_start,
+ (unsigned long long) shard_end,
+ GNUNET_STRINGS_relative_time_to_string (
+ GNUNET_TIME_absolute_get_duration (shard_start_time),
+ true));
+ break;
}
- wa->started_transaction = true;
+ shard_delay = GNUNET_TIME_absolute_get_duration (shard_start_time);
+ shard_open = false;
+ transaction_completed ();
+ return;
}
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
+ NULL);
+}
+
+
+/**
+ * Callbacks of this type are used to serve the result of asking
+ * the bank for the transaction history.
+ *
+ * @param cls NULL
+ * @param reply response we got from the bank
+ */
+static void
+history_cb (void *cls,
+ const struct TALER_BANK_CreditHistoryResponse *reply)
+{
+ (void) cls;
+ GNUNET_assert (NULL == task);
+ hh = NULL;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Adding wire transfer over %s with (hashed) subject `%s'\n",
- TALER_amount2s (&details->amount),
- TALER_B2S (&details->reserve_pub));
- /* FIXME #7276: Consider using Postgres multi-valued insert here,
- for up to 15x speed-up according to
- https://dba.stackexchange.com/questions/224989/multi-row-insert-vs-transactional-single-row-inserts#225006
- (Note: this may require changing both the
- plugin API as well as modifying how this function is called.) */
- qs = db_plugin->reserves_in_insert (db_plugin->cls,
- &details->reserve_pub,
- &details->amount,
- details->execution_date,
- details->debit_account_uri,
- wa->ai->section_name,
- serial_id);
- switch (qs)
+ "History request returned with HTTP status %u\n",
+ reply->http_status);
+ switch (reply->http_status)
{
- case GNUNET_DB_STATUS_HARD_ERROR:
- GNUNET_break (0);
- db_plugin->rollback (db_plugin->cls);
- wa->started_transaction = false;
- GNUNET_SCHEDULER_shutdown ();
- wa->hh = NULL;
- return GNUNET_SYSERR;
- case GNUNET_DB_STATUS_SOFT_ERROR:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Got DB soft error for reserves_in_insert. Rolling back.\n");
- handle_soft_error (wa);
- wa->hh = NULL;
- return GNUNET_SYSERR;
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- /* Either wirewatch was freshly started after the system was
- shutdown and we're going over an incomplete shard again
- after being restarted, or the shard lock period was too
- short (number of workers set incorrectly?) and a 2nd
- wirewatcher has been stealing our work while we are still
- at it. */
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Attempted to import transaction %llu (%s) twice. "
- "This should happen rarely (if not, ask for support).\n",
- (unsigned long long) serial_id,
- wa->job_name);
- /* already existed, ok, let's just continue */
+ case MHD_HTTP_OK:
+ process_reply (reply->details.ok.details,
+ reply->details.ok.details_length);
+ return;
+ case MHD_HTTP_NO_CONTENT:
+ transaction_completed ();
+ return;
+ case MHD_HTTP_NOT_FOUND:
+ hh_account_404 = true;
+ if (ignore_account_404)
+ {
+ transaction_completed ();
+ return;
+ }
break;
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- /* normal case */
+ default:
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Error fetching history: %s (%u)\n",
+ TALER_ErrorCode_get_hint (reply->ec),
+ reply->http_status);
break;
}
- wa->latest_row_off = serial_id;
- return GNUNET_OK;
+ if (! exit_on_error)
+ {
+ transaction_completed ();
+ return;
+ }
+ GNUNET_SCHEDULER_shutdown ();
}
static void
continue_with_shard (void *cls)
{
- struct WireAccount *wa = cls;
unsigned int limit;
+ (void) cls;
task = NULL;
- limit = GNUNET_MIN (wa->batch_size,
- wa->shard_end - wa->latest_row_off);
- wa->max_row_off = wa->latest_row_off + limit;
- GNUNET_assert (NULL == wa->hh);
- wa->hh = TALER_BANK_credit_history (ctx,
- wa->ai->auth,
- wa->latest_row_off,
- limit,
- test_mode
- ? GNUNET_TIME_UNIT_ZERO
- : LONGPOLL_TIMEOUT,
- &history_cb,
- wa);
- if (NULL == wa->hh)
+ GNUNET_assert (shard_end > latest_row_off);
+ limit = GNUNET_MIN (batch_size,
+ shard_end - latest_row_off);
+ GNUNET_assert (NULL == hh);
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Requesting credit history starting from %llu\n",
+ (unsigned long long) latest_row_off);
+ hh_start_time = GNUNET_TIME_absolute_get ();
+ hh_returned_data = false;
+ hh_account_404 = false;
+ hh = TALER_BANK_credit_history (ctx,
+ ai->auth,
+ latest_row_off,
+ limit,
+ test_mode
+ ? GNUNET_TIME_UNIT_ZERO
+ : longpoll_timeout,
+ &history_cb,
+ NULL);
+ if (NULL == hh)
{
GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to start request for account history!\n");
@@ -821,13 +773,20 @@ continue_with_shard (void *cls)
}
+/**
+ * Reserve a shard for us to work on.
+ *
+ * @param cls NULL
+ */
static void
lock_shard (void *cls)
{
- struct WireAccount *wa = cls;
enum GNUNET_DB_QueryStatus qs;
struct GNUNET_TIME_Relative delay;
+ uint64_t last_shard_start = shard_start;
+ uint64_t last_shard_end = shard_end;
+ (void) cls;
task = NULL;
if (GNUNET_SYSERR ==
db_plugin->preflight (db_plugin->cls))
@@ -838,17 +797,16 @@ lock_shard (void *cls)
GNUNET_SCHEDULER_shutdown ();
return;
}
- if ( (wa->shard_open) &&
- (GNUNET_TIME_absolute_is_future (wa->shard_end_time)) )
+ if ( (shard_open) &&
+ (GNUNET_TIME_absolute_is_future (shard_end_time)) )
{
- wa->delay = true; /* default is to delay, unless
- we find out that we're really busy */
- wa->batch_start = wa->latest_row_off;
+ progress = false;
+ batch_start = latest_row_off;
task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
- wa);
+ NULL);
return;
}
- if (wa->shard_open)
+ if (shard_open)
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Shard not completed in time, will try to re-acquire\n");
/* How long we lock a shard depends on the number of
@@ -861,15 +819,15 @@ lock_shard (void *cls)
GNUNET_CRYPTO_QUALITY_WEAK,
4 * GNUNET_TIME_relative_max (
wirewatch_idle_sleep_interval,
- GNUNET_TIME_relative_multiply (wa->shard_delay,
+ GNUNET_TIME_relative_multiply (shard_delay,
max_workers)).rel_value_us);
- wa->shard_start_time = GNUNET_TIME_absolute_get ();
+ shard_start_time = GNUNET_TIME_absolute_get ();
qs = db_plugin->begin_shard (db_plugin->cls,
- wa->job_name,
+ job_name,
delay,
shard_size,
- &wa->shard_start,
- &wa->shard_end);
+ &shard_start,
+ &shard_end);
switch (qs)
{
case GNUNET_DB_STATUS_HARD_ERROR:
@@ -883,49 +841,78 @@ lock_shard (void *cls)
{
struct GNUNET_TIME_Relative rdelay;
- rdelay = GNUNET_TIME_randomize (wirewatch_idle_sleep_interval);
+ wirewatch_conflict_sleep_interval
+ = GNUNET_TIME_STD_BACKOFF (wirewatch_conflict_sleep_interval);
+ rdelay = GNUNET_TIME_randomize (wirewatch_conflict_sleep_interval);
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
"Serialization error tying to obtain shard %s, will try again in %s!\n",
- wa->job_name,
+ job_name,
GNUNET_STRINGS_relative_time_to_string (rdelay,
- GNUNET_YES));
- wa->delayed_until = GNUNET_TIME_relative_to_absolute (rdelay);
+ true));
+#if 1
+ if (GNUNET_TIME_relative_cmp (rdelay,
+ >,
+ GNUNET_TIME_UNIT_SECONDS))
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Delay would have been for %s\n",
+ GNUNET_TIME_relative2s (rdelay,
+ true));
+ rdelay = GNUNET_TIME_relative_min (rdelay,
+ GNUNET_TIME_UNIT_SECONDS);
+#endif
+ delayed_until = GNUNET_TIME_relative_to_absolute (rdelay);
}
GNUNET_assert (NULL == task);
- schedule_transfers (wa->next);
+ schedule_transfers ();
return;
case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
GNUNET_break (0);
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
"No shard available, will try again for %s in %s!\n",
- wa->job_name,
+ job_name,
GNUNET_STRINGS_relative_time_to_string (
wirewatch_idle_sleep_interval,
- GNUNET_YES));
- wa->delayed_until = GNUNET_TIME_relative_to_absolute (
+ true));
+ delayed_until = GNUNET_TIME_relative_to_absolute (
wirewatch_idle_sleep_interval);
+ shard_open = false;
GNUNET_assert (NULL == task);
- schedule_transfers (wa->next);
+ schedule_transfers ();
return;
case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
/* continued below */
+ wirewatch_conflict_sleep_interval = GNUNET_TIME_UNIT_ZERO;
break;
}
- wa->shard_end_time = GNUNET_TIME_relative_to_absolute (delay);
- wa->shard_open = true;
+ shard_end_time = GNUNET_TIME_relative_to_absolute (delay);
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Starting with shard %s at (%llu,%llu] locked for %s\n",
- wa->job_name,
- (unsigned long long) wa->shard_start,
- (unsigned long long) wa->shard_end,
+ job_name,
+ (unsigned long long) shard_start,
+ (unsigned long long) shard_end,
GNUNET_STRINGS_relative_time_to_string (delay,
- GNUNET_YES));
- wa->delay = true; /* default is to delay, unless
- we find out that we're really busy */
- wa->batch_start = wa->shard_start;
- wa->latest_row_off = wa->batch_start;
+ true));
+ progress = false;
+ batch_start = shard_start;
+ if ( (shard_open) &&
+ (shard_start == last_shard_start) &&
+ (shard_end == last_shard_end) )
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Continuing from %llu\n",
+ (unsigned long long) latest_row_off);
+ GNUNET_break (latest_row_off >= batch_start); /* resume where we left things */
+ }
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Resetting shard start to original start point (%d)\n",
+ shard_open ? 1 : 0);
+ latest_row_off = batch_start;
+ }
+ shard_open = true;
task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
- wa);
+ NULL);
}
@@ -948,26 +935,26 @@ run (void *cls,
(void) cfgfile;
cfg = c;
+ GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
+ cls);
if (GNUNET_OK !=
exchange_serve_process_config ())
{
global_ret = EXIT_NOTCONFIGURED;
+ GNUNET_SCHEDULER_shutdown ();
return;
}
- GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
- cls);
ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule,
&rc);
if (NULL == ctx)
{
GNUNET_break (0);
GNUNET_SCHEDULER_shutdown ();
+ global_ret = EXIT_NO_RESTART;
return;
}
rc = GNUNET_CURL_gnunet_rc_create (ctx);
- GNUNET_assert (NULL == task);
- task = GNUNET_SCHEDULER_add_now (&lock_shard,
- wa_head);
+ schedule_transfers ();
}
@@ -983,10 +970,20 @@ main (int argc,
char *const *argv)
{
struct GNUNET_GETOPT_CommandLineOption options[] = {
+ GNUNET_GETOPT_option_string ('a',
+ "account",
+ "SECTION_NAME",
+ "name of the configuration section with the account we should watch (needed if more than one is enabled for crediting)",
+ &account_section),
GNUNET_GETOPT_option_flag ('e',
"exit-on-error",
"terminate wirewatch if we failed to download information from the bank",
&exit_on_error),
+ GNUNET_GETOPT_option_relative_time ('f',
+ "longpoll-timeout",
+ "DELAY",
+ "what is the timeout when asking the bank about new transactions, specify with unit (e.g. --longpoll-timeout=30s)",
+ &longpoll_timeout),
GNUNET_GETOPT_option_flag ('I',
"ignore-not-found",
"continue, even if the bank account of the exchange was not found",
@@ -1012,6 +1009,7 @@ main (int argc,
};
enum GNUNET_GenericReturnValue ret;
+ longpoll_timeout = LONGPOLL_TIMEOUT;
if (GNUNET_OK !=
GNUNET_STRINGS_get_utf8_args (argc, argv,
&argc, &argv))