commit 7253adbca4107f013c3c2340a016dc516af6b07b
parent 83bcc2d972deb6c68d5813243a6647f2d0abc930
Author: Christian Grothoff <christian@grothoff.org>
Date: Thu, 6 Aug 2026 20:18:12 +0200
minimize round trips and use arrays when importing incoming wire transfers
Diffstat:
30 files changed, 1667 insertions(+), 1528 deletions(-)
diff --git a/src/auditor/taler-auditor-sync.c b/src/auditor/taler-auditor-sync.c
@@ -29,7 +29,7 @@
#include "exchange-database/begin_shard.h"
#include "exchange-database/update_to_aml_unlocked.h"
#include "exchange-database/commit.h"
-#include "exchange-database/complete_shard.h"
+#include "exchange-database/update_shard_progress.h"
#include "exchange-database/compute_shard.h"
#include "exchange-database/get_count_known_coins.h"
#include "exchange-database/insert_aggregation_transient.h"
@@ -120,7 +120,6 @@ struct InsertContext;
#include "exchange-database/iterate_denominations.h"
#include "exchange-database/iterate_kyc_references.h"
#include "exchange-database/iterate_reserve_close_info.h"
-#include "exchange-database/insert_kycauth_in.h"
#include "exchange-database/get_kyc_provider_account.h"
#include "exchange-database/get_active_legitimization.h"
#include "exchange-database/get_aml_file_number.h"
@@ -154,7 +153,7 @@ struct InsertContext;
#include "exchange-database/release_revolving_shard.h"
#include "exchange-database/get_reserve.h"
#include "exchange-database/get_reserve_origin.h"
-#include "exchange-database/do_insert_reserve_in.h"
+#include "exchange-database/do_import_credits.h"
#include "exchange-database/update_reserve.h"
#include "exchange-database/rollback.h"
#include "exchange-database/iterate_account_merges_above_serial_id.h"
@@ -214,7 +213,6 @@ struct InsertContext;
#include "exchange-database/update_auditor.h"
#include "exchange-database/update_legitimization_process_by_row.h"
#include "exchange-database/update_wire.h"
-#include "exchange-database/insert_wad_in.h"
#include "exchange-database/iterate_prewires.h"
#include "exchange-database/insert_prewire.h"
#include "exchange-database/update_to_prewire_failed.h"
diff --git a/src/exchange/taler-exchange-expire.c b/src/exchange/taler-exchange-expire.c
@@ -30,7 +30,7 @@
#include "exchange-database/preflight.h"
#include "exchange-database/commit.h"
#include "exchange-database/rollback.h"
-#include "exchange-database/complete_shard.h"
+#include "exchange-database/update_shard_progress.h"
#include "exchange-database/abort_shard.h"
#include "exchange-database/begin_shard.h"
#include "exchange-database/do_expire_purse.h"
@@ -59,6 +59,18 @@ struct Shard
struct GNUNET_TIME_Absolute shard_end;
/**
+ * How far into the shard we have come. Purses that expired before this
+ * are dealt with and committed; a worker taking the shard over resumes
+ * here instead of at @e shard_start.
+ */
+ struct GNUNET_TIME_Absolute progress;
+
+ /**
+ * For how long we hold the shard. Renewed whenever we record progress.
+ */
+ struct GNUNET_TIME_Relative lease;
+
+ /**
* Number of starting points found in the shard.
*/
uint64_t work_counter;
@@ -177,42 +189,44 @@ commit_or_warn (void)
/**
- * Release lock on shard @a s in the database.
- * On error, terminates this process.
+ * Record how far we have come within shard @a s, and renew our lease on it.
+ * Must be called from inside the transaction that persists the work being
+ * reported, so that the two land together. Marking the shard completed is
+ * the same statement, so there is no separate "shard done" write.
+ *
+ * @param s shard we are working on
+ * @param progress point in the shard we have reached
+ * @return transaction status code
+ */
+static enum GNUNET_DB_QueryStatus
+record_progress (const struct Shard *s,
+ struct GNUNET_TIME_Absolute progress)
+{
+ return TALER_EXCHANGEDB_update_shard_progress (
+ pg,
+ "expire",
+ s->shard_start.abs_value_us,
+ s->shard_end.abs_value_us,
+ progress.abs_value_us,
+ s->lease);
+}
+
+
+/**
+ * We are done with shard @a s. Its completion has already been committed by
+ * #record_progress(), so this only reports and releases memory.
*
* @param[in] s shard to free (and memory to release)
*/
static void
release_shard (struct Shard *s)
{
- enum GNUNET_DB_QueryStatus qs;
unsigned long long wc = (unsigned long long) s->work_counter;
- qs = TALER_EXCHANGEDB_complete_shard (
- pg,
- "expire",
- s->shard_start.abs_value_us,
- s->shard_end.abs_value_us);
GNUNET_free (s);
- switch (qs)
- {
- case GNUNET_DB_STATUS_HARD_ERROR:
- case GNUNET_DB_STATUS_SOFT_ERROR:
- GNUNET_break (GNUNET_DB_STATUS_SOFT_ERROR != qs);
- GNUNET_break (0);
- global_ret = EXIT_FAILURE;
- GNUNET_SCHEDULER_shutdown ();
- return;
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- /* Strange, but let's just continue */
- break;
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Purse expiration shard completed with %llu purses\n",
- wc);
- /* normal case */
- break;
- }
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Purse expiration shard completed with %llu purses\n",
+ wc);
if ( (0 == wc) &&
(test_mode) &&
(! jump_mode) )
@@ -263,6 +277,7 @@ run_expire (void *cls)
{
struct Shard *s = cls;
enum GNUNET_DB_QueryStatus qs;
+ struct GNUNET_TIME_Absolute purse_expiration;
task = NULL;
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
@@ -289,8 +304,9 @@ run_expire (void *cls)
return;
}
qs = TALER_EXCHANGEDB_do_expire_purse (pg,
- s->shard_start,
- s->shard_end);
+ s->progress,
+ s->shard_end,
+ &purse_expiration);
switch (qs)
{
case GNUNET_DB_STATUS_HARD_ERROR:
@@ -308,7 +324,11 @@ run_expire (void *cls)
NULL);
return;
case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- if (0 > commit_or_warn ())
+ /* Nothing left in this shard: record it as done in the very transaction
+ we are about to commit, instead of in a second one afterwards. */
+ if ( (0 > record_progress (s,
+ s->shard_end)) ||
+ (0 > commit_or_warn ()) )
{
TALER_EXCHANGEDB_rollback (pg);
abort_shard (s);
@@ -324,7 +344,23 @@ run_expire (void *cls)
case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
/* commit, and go again immediately */
s->work_counter++;
- (void) commit_or_warn ();
+ if ( (0 > record_progress (s,
+ purse_expiration)) ||
+ (0 > commit_or_warn ()) )
+ {
+ /* Nothing landed, so leave our resume point where it was: moving it
+ here would skip the purse we just failed to decide. */
+ TALER_EXCHANGEDB_rollback (pg);
+ abort_shard (s);
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&run_shard,
+ NULL);
+ return;
+ }
+ /* Purses expiring in the same microsecond are looked at again next
+ round; deciding one twice is a no-op, so resuming at (not after)
+ this timestamp is the safe choice. */
+ s->progress = purse_expiration;
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_expire,
s);
@@ -356,6 +392,7 @@ run_shard (void *cls)
}
s = GNUNET_new (struct Shard);
s->start_time = GNUNET_TIME_timestamp_get ();
+ s->lease = shard_size;
qs = TALER_EXCHANGEDB_begin_shard (pg,
"expire",
shard_size,
@@ -366,7 +403,8 @@ run_shard (void *cls)
abs_value_us
: shard_size.rel_value_us,
&s->shard_start.abs_value_us,
- &s->shard_end.abs_value_us);
+ &s->shard_end.abs_value_us,
+ &s->progress.abs_value_us);
jump_mode = false;
if (0 >= qs)
{
diff --git a/src/exchange/taler-exchange-transfer.c b/src/exchange/taler-exchange-transfer.c
@@ -26,7 +26,7 @@
#include "taler/taler_json_lib.h"
#include "taler/taler_bank_service.h"
#include "exchange-database/begin_shard.h"
-#include "exchange-database/complete_shard.h"
+#include "exchange-database/update_shard_progress.h"
#include "exchange-database/commit.h"
#include "exchange-database/preflight.h"
#include "exchange-database/insert_sanction_list_hit.h"
@@ -83,6 +83,13 @@ struct Shard
*/
uint64_t batch_end;
+ /**
+ * For how long we hold this shard. Renewed whenever we record progress,
+ * so that a worker that is slowly but steadily working through a shard
+ * keeps it.
+ */
+ struct GNUNET_TIME_Relative lease;
+
};
@@ -376,8 +383,38 @@ select_shard (void *cls);
static void
batch_done (void)
{
+ enum GNUNET_DB_QueryStatus qs;
+
/* batch done */
GNUNET_assert (NULL == wpd_head);
+ /* Rides along with the transaction that persists this batch, so the
+ transfers and the record of how far the shard has come become visible
+ together. Once batch_end+1 reaches the end of the shard this same
+ statement marks it completed, which is why there is no separate
+ "shard done" write any more. */
+ qs = TALER_EXCHANGEDB_update_shard_progress (pg,
+ "transfer",
+ shard->shard_start,
+ shard->shard_end,
+ shard->batch_end + 1,
+ shard->lease);
+ if (0 > qs)
+ {
+ TALER_EXCHANGEDB_rollback (pg);
+ if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Serialization failure recording shard progress, trying again immediately!\n");
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&run_transfers,
+ NULL);
+ return;
+ }
+ GNUNET_break (0);
+ global_ret = EXIT_FAILURE;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
switch (commit_or_warn ())
{
case GNUNET_DB_STATUS_SOFT_ERROR:
@@ -627,36 +664,13 @@ run_transfers (void *cls)
limit = shard->shard_end - shard->batch_start;
if (0 >= limit)
{
+ /* The last batch we committed already carried this shard's progress past
+ its end, and was therefore also what marked it completed. Nothing left
+ to write here. */
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Shard [%llu,%llu) completed\n",
(unsigned long long) shard->shard_start,
(unsigned long long) shard->batch_end);
- qs = TALER_EXCHANGEDB_complete_shard (pg,
- "transfer",
- shard->shard_start,
- shard->batch_end + 1);
- switch (qs)
- {
- case GNUNET_DB_STATUS_HARD_ERROR:
- GNUNET_break (0);
- GNUNET_free (shard);
- GNUNET_SCHEDULER_shutdown ();
- return;
- case GNUNET_DB_STATUS_SOFT_ERROR:
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Got DB soft error for complete_shard. Rolling back.\n");
- GNUNET_free (shard);
- GNUNET_assert (NULL == task);
- task = GNUNET_SCHEDULER_add_now (&select_shard,
- NULL);
- return;
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- /* already existed, ok, let's just continue */
- break;
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- /* normal case */
- break;
- }
shard_delay = GNUNET_TIME_absolute_get_duration (
shard->shard_start_time);
GNUNET_free (shard);
@@ -748,6 +762,7 @@ select_shard (void *cls)
struct GNUNET_TIME_Relative delay;
uint64_t start;
uint64_t end;
+ uint64_t progress;
(void) cls;
task = NULL;
@@ -774,7 +789,8 @@ select_shard (void *cls)
delay,
shard_size,
&start,
- &end);
+ &end,
+ &progress);
switch (qs)
{
case GNUNET_DB_STATUS_HARD_ERROR:
@@ -817,7 +833,11 @@ select_shard (void *cls)
shard->shard_start_time = GNUNET_TIME_absolute_get ();
shard->shard_start = start;
shard->shard_end = end;
- shard->batch_start = start;
+ shard->lease = delay;
+ /* Resume where the shard says it got to, which is @a start for a fresh
+ shard and further along for one abandoned mid-way. */
+ shard->batch_start = progress;
+ shard->batch_end = (0 == progress) ? 0 : progress - 1;
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_transfers,
NULL);
diff --git a/src/exchange/taler-exchange-wirewatch.c b/src/exchange/taler-exchange-wirewatch.c
@@ -28,16 +28,10 @@
#include "taler/taler_bank_service.h"
#include "exchange-database/begin_shard.h"
#include "exchange-database/abort_shard.h"
-#include "exchange-database/complete_shard.h"
-#include "exchange-database/commit.h"
#include "exchange-database/preflight.h"
-#include "exchange-database/insert_kycauth_in.h"
-#include "exchange-database/insert_wad_in.h"
-#include "exchange-database/do_insert_reserve_in.h"
-#include "exchange-database/rollback.h"
+#include "exchange-database/do_import_credits.h"
#include "exchange-database/event_listen.h"
#include "exchange-database/event_listen_cancel.h"
-#include "exchange-database/insert_wad_in.h"
/**
* How long to wait for an HTTP reply if there
@@ -96,8 +90,8 @@ static struct GNUNET_TIME_Absolute delayed_until;
static uint64_t batch_start;
/**
- * Latest row offset seen in this transaction, becomes
- * the new #batch_start upon commit.
+ * Latest row offset we have imported and committed. Mirrors the
+ * @c progress_row of our shard in the database.
*/
static uint64_t latest_row_off;
@@ -112,6 +106,13 @@ static uint64_t shard_start;
static uint64_t shard_end;
/**
+ * For how long do we hold the shard? Renewed on every batch we import, so
+ * that a worker whose account trickles in slowly does not keep losing its
+ * shard to a second worker that then has nothing to do either.
+ */
+static struct GNUNET_TIME_Relative shard_lease;
+
+/**
* When did we start with the shard?
*/
static struct GNUNET_TIME_Absolute shard_start_time;
@@ -165,11 +166,6 @@ static unsigned int batch_thresh;
static bool progress;
/**
- * Did we start a transaction yet?
- */
-static bool started_transaction;
-
-/**
* Is this shard still open for processing.
*/
static bool shard_open;
@@ -269,13 +265,11 @@ shutdown_task (void *cls)
TALER_BANK_credit_history_cancel (hh);
hh = NULL;
}
- if (started_transaction)
- {
- TALER_EXCHANGEDB_rollback (pg);
- started_transaction = false;
- }
if (shard_open)
{
+ /* Everything we imported is committed, and #latest_row_off is recorded in
+ the shard, so releasing the lease loses no work: whoever picks the
+ shard up next resumes where we stopped. */
qs = TALER_EXCHANGEDB_abort_shard (pg,
job_name,
shard_start,
@@ -413,14 +407,14 @@ continue_with_shard (void *cls);
/**
- * We encountered a serialization error. Rollback the transaction and try
- * again.
+ * We encountered a serialization error. The batch that hit it was a single
+ * statement, so the database has already discarded all of it; there is
+ * nothing to roll back. Shrink the batch and ask the bank again from the
+ * last point we committed.
*/
static void
handle_soft_error (void)
{
- TALER_EXCHANGEDB_rollback (pg);
- started_transaction = false;
if (1 < batch_size)
{
batch_thresh = batch_size;
@@ -429,8 +423,6 @@ handle_soft_error (void)
"Reduced batch size to %llu due to serialization issue\n",
(unsigned long long) batch_size);
}
- /* Reset to beginning of transaction, and go again
- from there. */
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Encountered soft error, resetting start point to batch start\n");
latest_row_off = batch_start;
@@ -582,7 +574,6 @@ process_reply (const struct TALER_BANK_CreditDetails *details,
"Serial ID %llu not monotonic (got %llu before). Failing!\n",
(unsigned long long) cd->serial_id,
(unsigned long long) lroff);
- TALER_EXCHANGEDB_rollback (pg);
GNUNET_SCHEDULER_shutdown ();
return;
}
@@ -602,19 +593,32 @@ process_reply (const struct TALER_BANK_CreditDetails *details,
}
lroff = cd->serial_id;
}
- if (0 != details_length)
{
- enum GNUNET_DB_QueryStatus qss[details_length];
- struct TALER_EXCHANGEDB_ReserveInInfo reserves[details_length];
- unsigned int j = 0;
+ enum GNUNET_DB_QueryStatus qss[GNUNET_NZL (details_length)];
+ struct TALER_EXCHANGEDB_ReserveInInfo reserves[GNUNET_NZL (details_length)];
+ struct TALER_EXCHANGEDB_KycauthInInfo kycauths[GNUNET_NZL (details_length)];
+ struct TALER_EXCHANGEDB_WadInInfo wads[GNUNET_NZL (details_length)];
+ struct TALER_EXCHANGEDB_CreditBatch batch = {
+ .exchange_account_name = ai->section_name,
+ .reserves = reserves,
+ .kycauths = kycauths,
+ .wads = wads,
+ .job_name = job_name,
+ .shard_start = shard_start,
+ .shard_end = shard_end,
+ .progress_row = lroff,
+ .lease = shard_lease
+ };
+ unsigned int j;
/* make compiler happy */
memset (qss,
0,
sizeof (qss));
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Importing %u transactions\n",
- details_length);
+ if (0 != details_length)
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Importing %u transactions\n",
+ details_length);
for (unsigned int i = 0; i<details_length; i++)
{
const struct TALER_BANK_CreditDetails *cd = &details[i];
@@ -623,9 +627,9 @@ process_reply (const struct TALER_BANK_CreditDetails *details,
{
case TALER_BANK_CT_RESERVE:
{
- struct TALER_EXCHANGEDB_ReserveInInfo *res = &reserves[j++];
+ struct TALER_EXCHANGEDB_ReserveInInfo *res
+ = &reserves[batch.reserves_length++];
- /* add to batch, do later */
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Importing reserve transfer over %s\n",
TALER_amount2s (&cd->amount));
@@ -633,94 +637,62 @@ process_reply (const struct TALER_BANK_CreditDetails *details,
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;
}
break;
case TALER_BANK_CT_KYCAUTH:
{
+ struct TALER_EXCHANGEDB_KycauthInInfo *ka
+ = &kycauths[batch.kycauths_length++];
+
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Importing KYC auth transfer over %s\n",
TALER_amount2s (&cd->amount));
- qs = TALER_EXCHANGEDB_insert_kycauth_in (
- pg,
- &cd->details.kycauth.account_pub,
- &cd->amount,
- cd->execution_date,
- cd->debit_account_uri,
- ai->section_name,
- cd->serial_id);
- switch (qs)
- {
- 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 kycauth_in_insert (%u). Rolling back.\n",
- i);
- handle_soft_error ();
- return;
- default:
- break;
- }
- break;
+ ka->account_pub = &cd->details.kycauth.account_pub;
+ ka->balance = &cd->amount;
+ ka->execution_time = cd->execution_date;
+ ka->sender_account_details = cd->debit_account_uri;
+ ka->wire_reference = cd->serial_id;
}
+ break;
case TALER_BANK_CT_WAD:
{
+ struct TALER_EXCHANGEDB_WadInInfo *wad
+ = &wads[batch.wads_length++];
+
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Importing WAD transfer over %s\n",
TALER_amount2s (&cd->amount));
- qs = TALER_EXCHANGEDB_insert_wad_in (
- pg,
- &cd->details.wad.wad_id,
- cd->details.wad.origin_exchange_url,
- &cd->amount,
- cd->execution_date,
- cd->debit_account_uri,
- ai->section_name,
- cd->serial_id);
- switch (qs)
- {
- 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 wad_in_insert (%u). Rolling back.\n",
- i);
- handle_soft_error ();
- return;
- default:
- break;
- }
-
+ wad->wad_id = &cd->details.wad.wad_id;
+ wad->origin_exchange_url = cd->details.wad.origin_exchange_url;
+ wad->balance = &cd->amount;
+ wad->execution_time = cd->execution_date;
}
+ break;
}
}
- if (j > 0)
+ /* One statement, hence one transaction: these transfers and the record of
+ how far this shard has come become visible together. That is what lets
+ us hand the money to the wallets now instead of at the end of the
+ shard, without risking a crash that leaves the shard claiming work it
+ never did. */
+ qs = TALER_EXCHANGEDB_do_import_credits (pg,
+ &batch,
+ qss);
+ switch (qs)
{
- qs = TALER_EXCHANGEDB_do_insert_reserve_in (pg,
- reserves,
- j,
- qss);
- switch (qs)
- {
- 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;
- }
+ 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 importing %u transactions. Retrying.\n",
+ details_length);
+ handle_soft_error ();
+ return;
+ default:
+ break;
}
j = 0;
for (unsigned int i = 0; i<details_length; i++)
@@ -728,18 +700,13 @@ process_reply (const struct TALER_BANK_CreditDetails *details,
const struct TALER_BANK_CreditDetails *cd = &details[i];
if (TALER_BANK_CT_RESERVE != cd->type)
- continue;
+ continue; /* only reserve transfers report a per-row status */
switch (qss[j++])
{
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 ();
+ GNUNET_break (0); /* handled above, for the batch as a whole */
+ GNUNET_SCHEDULER_shutdown ();
return;
case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
/* Either wirewatch was freshly started after the system was
@@ -766,42 +733,23 @@ process_reply (const struct TALER_BANK_CreditDetails *details,
}
latest_row_off = lroff;
+ /* The same statement renewed our lease in the database, so keep the local
+ deadline in step or we would go re-acquire a shard we still hold. */
+ shard_end_time = GNUNET_TIME_relative_to_absolute (shard_lease);
shard_done = (shard_end <= latest_row_off);
if (shard_done)
{
- /* shard is complete, mark this as well */
- qs = TALER_EXCHANGEDB_complete_shard (pg,
- job_name,
- shard_start,
- shard_end);
- switch (qs)
- {
- 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 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;
- }
+ /* The shard was marked completed by the very statement that imported the
+ last of its transfers; there is nothing left to write. */
+ 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));
shard_delay = GNUNET_TIME_absolute_get_duration (shard_start_time);
shard_open = false;
transaction_completed ();
@@ -874,6 +822,10 @@ continue_with_shard (void *cls)
GNUNET_assert (shard_end > latest_row_off);
limit = GNUNET_MIN (batch_size,
shard_end - latest_row_off);
+ /* Where this batch starts out: the point we last committed, and thus where
+ #handle_soft_error() rewinds to and what #transaction_completed() measures
+ the batch against. */
+ batch_start = latest_row_off;
GNUNET_assert (NULL == hh);
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Requesting credit history starting from %llu\n",
@@ -912,8 +864,7 @@ lock_shard (void *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;
+ uint64_t progress_row;
(void) cls;
task = NULL;
@@ -930,7 +881,6 @@ lock_shard (void *cls)
(GNUNET_TIME_absolute_is_future (shard_end_time)) )
{
progress = false;
- batch_start = latest_row_off;
task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
NULL);
return;
@@ -950,12 +900,14 @@ lock_shard (void *cls)
GNUNET_TIME_relative_multiply (shard_delay,
max_workers)).rel_value_us);
shard_start_time = GNUNET_TIME_absolute_get ();
+ shard_lease = delay;
qs = TALER_EXCHANGEDB_begin_shard (pg,
job_name,
delay,
shard_size,
&shard_start,
- &shard_end);
+ &shard_end,
+ &progress_row);
switch (qs)
{
case GNUNET_DB_STATUS_HARD_ERROR:
@@ -1021,23 +973,16 @@ lock_shard (void *cls)
GNUNET_STRINGS_relative_time_to_string (delay,
true));
progress = false;
- batch_start = shard_start;
- if ( (shard_open) &&
- (shard_start == last_shard_start) &&
- (shard_end == last_shard_end) )
- {
+ /* The shard itself says where to resume. Whether this is a shard we had
+ before, one abandoned by another worker, or a brand new one no longer
+ matters: everything below progress_row is imported and committed, and
+ nothing above it is. */
+ if (progress_row != shard_start)
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;
- }
+ (unsigned long long) progress_row);
+ latest_row_off = progress_row;
+ batch_start = progress_row;
shard_open = true;
task = GNUNET_SCHEDULER_add_now (&continue_with_shard,
NULL);
diff --git a/src/exchangedb/begin_shard.c b/src/exchangedb/begin_shard.c
@@ -1,6 +1,6 @@
/*
This file is part of TALER
- Copyright (C) 2022 Taler Systems SA
+ Copyright (C) 2022, 2026 Taler Systems SA
TALER is free software; you can redistribute it and/or modify it under the
terms of the GNU General Public License as published by the Free Software
@@ -21,9 +21,14 @@
#include "taler/taler_pq_lib.h"
#include "exchange-database/begin_shard.h"
#include "helper.h"
-#include "exchange-database/start.h"
-#include "exchange-database/rollback.h"
-#include "exchange-database/commit.h"
+
+
+/**
+ * How often do we retry when another worker opened the very shard we were
+ * about to open? Each round is one round-trip and the loser of the race can
+ * simply take the shard after it, so this converges quickly.
+ */
+#define MAX_RETRIES 10
enum GNUNET_DB_QueryStatus
@@ -32,231 +37,76 @@ TALER_EXCHANGEDB_begin_shard (struct TALER_EXCHANGEDB_PostgresContext *pg,
struct GNUNET_TIME_Relative delay,
uint64_t shard_size,
uint64_t *start_row,
- uint64_t *end_row)
+ uint64_t *end_row,
+ uint64_t *progress_row)
{
-
- for (unsigned int retries = 0; retries<10; retries++)
+ PREPARE (pg,
+ "begin_shard",
+ "SELECT"
+ " out_start_row"
+ ",out_end_row"
+ ",out_progress_row"
+ " FROM exchange_do_begin_shard"
+ " ($1,$2,$3,$4);");
+ for (unsigned int retries = 0; retries<MAX_RETRIES; retries++)
{
- if (GNUNET_OK !=
- TALER_EXCHANGEDB_start (pg,
- "begin_shard"))
- {
- GNUNET_break (0);
- return GNUNET_DB_STATUS_HARD_ERROR;
- }
-
- {
- struct GNUNET_TIME_Absolute past;
- enum GNUNET_DB_QueryStatus qs;
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_string (job_name),
- GNUNET_PQ_query_param_absolute_time (&past),
- GNUNET_PQ_query_param_end
- };
- struct GNUNET_PQ_ResultSpec rs[] = {
- GNUNET_PQ_result_spec_uint64 ("start_row",
+ struct GNUNET_TIME_Absolute now
+ = GNUNET_TIME_absolute_get ();
+ struct GNUNET_TIME_Absolute lease_until
+ = GNUNET_TIME_relative_to_absolute (delay);
+ struct GNUNET_PQ_QueryParam params[] = {
+ GNUNET_PQ_query_param_string (job_name),
+ GNUNET_PQ_query_param_absolute_time (&now),
+ GNUNET_PQ_query_param_absolute_time (&lease_until),
+ GNUNET_PQ_query_param_uint64 (&shard_size),
+ GNUNET_PQ_query_param_end
+ };
+ bool lost_race = false;
+ struct GNUNET_PQ_ResultSpec rs[] = {
+ GNUNET_PQ_result_spec_allow_null (
+ GNUNET_PQ_result_spec_uint64 ("out_start_row",
start_row),
- GNUNET_PQ_result_spec_uint64 ("end_row",
+ &lost_race),
+ GNUNET_PQ_result_spec_allow_null (
+ GNUNET_PQ_result_spec_uint64 ("out_end_row",
end_row),
- GNUNET_PQ_result_spec_end
- };
-
- past = GNUNET_TIME_absolute_get ();
- PREPARE (pg,
- "begin_shard_open_shard",
- "SELECT"
- " start_row"
- ",end_row"
- " FROM work_shards"
- " WHERE job_name=$1"
- " AND completed=FALSE"
- " AND last_attempt<$2"
- " ORDER BY last_attempt ASC"
- " LIMIT 1;");
- qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
- "begin_shard_open_shard",
- params,
- rs);
- switch (qs)
- {
- case GNUNET_DB_STATUS_HARD_ERROR:
- GNUNET_break (0);
- TALER_EXCHANGEDB_rollback (pg);
- return qs;
- case GNUNET_DB_STATUS_SOFT_ERROR:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Serialization error on getting open shard\n");
- TALER_EXCHANGEDB_rollback (pg);
- continue;
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- {
- enum GNUNET_DB_QueryStatus qsz;
- struct GNUNET_TIME_Absolute now;
- struct GNUNET_PQ_QueryParam iparams[] = {
- GNUNET_PQ_query_param_string (job_name),
- GNUNET_PQ_query_param_absolute_time (&now),
- GNUNET_PQ_query_param_uint64 (start_row),
- GNUNET_PQ_query_param_uint64 (end_row),
- GNUNET_PQ_query_param_end
- };
-
- now = GNUNET_TIME_relative_to_absolute (delay);
- PREPARE (pg,
- "begin_shard_reclaim_shard",
- "UPDATE work_shards"
- " SET last_attempt=$2"
- " WHERE job_name=$1"
- " AND start_row=$3"
- " AND end_row=$4");
- qsz = GNUNET_PQ_eval_prepared_non_select (pg->conn,
- "begin_shard_reclaim_shard",
- iparams);
- switch (qsz)
- {
- case GNUNET_DB_STATUS_HARD_ERROR:
- GNUNET_break (0);
- TALER_EXCHANGEDB_rollback (pg);
- return qsz;
- case GNUNET_DB_STATUS_SOFT_ERROR:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Serialization error on claiming open shard\n");
- TALER_EXCHANGEDB_rollback (pg);
- continue;
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- goto commit;
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- GNUNET_break (0); /* logic error, should be impossible */
- TALER_EXCHANGEDB_rollback (pg);
- return GNUNET_DB_STATUS_HARD_ERROR;
- }
- }
- break; /* actually unreachable */
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- break; /* continued below */
- }
- } /* get_open_shard */
-
- /* No open shard, find last 'end_row' */
+ NULL),
+ GNUNET_PQ_result_spec_allow_null (
+ GNUNET_PQ_result_spec_uint64 ("out_progress_row",
+ progress_row),
+ NULL),
+ GNUNET_PQ_result_spec_end
+ };
+ enum GNUNET_DB_QueryStatus qs;
+
+ qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
+ "begin_shard",
+ params,
+ rs);
+ if (0 > qs)
+ return qs;
+ if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs)
{
- enum GNUNET_DB_QueryStatus qs;
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_string (job_name),
- GNUNET_PQ_query_param_end
- };
- struct GNUNET_PQ_ResultSpec rs[] = {
- GNUNET_PQ_result_spec_uint64 ("end_row",
- start_row),
- GNUNET_PQ_result_spec_end
- };
-
- PREPARE (pg,
- "begin_shard_last_shard",
- "SELECT"
- " end_row"
- " FROM work_shards"
- " WHERE job_name=$1"
- " ORDER BY end_row DESC"
- " LIMIT 1;");
- qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
- "begin_shard_last_shard",
- params,
- rs);
- switch (qs)
- {
- case GNUNET_DB_STATUS_HARD_ERROR:
- GNUNET_break (0);
- TALER_EXCHANGEDB_rollback (pg);
- return qs;
- case GNUNET_DB_STATUS_SOFT_ERROR:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Serialization error on getting last shard\n");
- TALER_EXCHANGEDB_rollback (pg);
- continue;
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- break;
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- *start_row = 0; /* base-case: no shards yet */
- break; /* continued below */
- }
- *end_row = *start_row + shard_size;
- } /* get_last_shard */
-
- /* Claim fresh shard */
+ GNUNET_break (0); /* the function always returns exactly one row */
+ return GNUNET_DB_STATUS_HARD_ERROR;
+ }
+ if (lost_race)
{
- enum GNUNET_DB_QueryStatus qs;
- struct GNUNET_TIME_Absolute now;
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_string (job_name),
- GNUNET_PQ_query_param_absolute_time (&now),
- GNUNET_PQ_query_param_uint64 (start_row),
- GNUNET_PQ_query_param_uint64 (end_row),
- GNUNET_PQ_query_param_end
- };
-
- now = GNUNET_TIME_relative_to_absolute (delay);
+ /* Someone else opened the shard we picked. Go around: either take over
+ an even older one, or open the one after theirs. */
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Trying to claim shard (%llu-%llu]\n",
- (unsigned long long) *start_row,
- (unsigned long long) *end_row);
-
- PREPARE (pg,
- "begin_shard_claim_next_shard",
- "INSERT INTO work_shards"
- "(job_name"
- ",last_attempt"
- ",start_row"
- ",end_row"
- ") VALUES "
- "($1, $2, $3, $4);");
- qs = GNUNET_PQ_eval_prepared_non_select (pg->conn,
- "begin_shard_claim_next_shard",
- params);
- switch (qs)
- {
- case GNUNET_DB_STATUS_HARD_ERROR:
- GNUNET_break (0);
- TALER_EXCHANGEDB_rollback (pg);
- return qs;
- case GNUNET_DB_STATUS_SOFT_ERROR:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Serialization error on claiming next shard\n");
- TALER_EXCHANGEDB_rollback (pg);
- continue;
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- /* continued below */
- break;
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- /* someone else got this shard already,
- try again */
- TALER_EXCHANGEDB_rollback (pg);
- continue;
- }
- } /* claim_next_shard */
-
- /* commit */
-commit:
- {
- enum GNUNET_DB_QueryStatus qs;
-
- qs = TALER_EXCHANGEDB_commit (pg);
- switch (qs)
- {
- case GNUNET_DB_STATUS_HARD_ERROR:
- GNUNET_break (0);
- TALER_EXCHANGEDB_rollback (pg);
- return qs;
- case GNUNET_DB_STATUS_SOFT_ERROR:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Serialization error on commit for beginning shard\n");
- TALER_EXCHANGEDB_rollback (pg);
- continue;
- case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Claimed new shard\n");
- return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
- }
+ "Lost the race to open a shard of %s, trying again\n",
+ job_name);
+ continue;
}
- } /* retry 'for' loop */
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Acquired shard %s (%llu,%llu], resuming at %llu\n",
+ job_name,
+ (unsigned long long) *start_row,
+ (unsigned long long) *end_row,
+ (unsigned long long) *progress_row);
+ return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
+ }
+ /* Persistent contention. Not an error, but the caller should back off. */
return GNUNET_DB_STATUS_SOFT_ERROR;
}
diff --git a/src/exchangedb/begin_shard.sql b/src/exchangedb/begin_shard.sql
@@ -0,0 +1,99 @@
+--
+-- This file is part of TALER
+-- Copyright (C) 2022--2026 Taler Systems SA
+--
+-- TALER is free software; you can redistribute it and/or modify it under the
+-- terms of the GNU 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 General Public License for more details.
+--
+-- You should have received a copy of the GNU General Public License along with
+-- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+--
+
+DROP FUNCTION IF EXISTS exchange_do_begin_shard;
+CREATE FUNCTION exchange_do_begin_shard(
+ IN in_job_name TEXT,
+ IN in_now INT8,
+ IN in_lease_until INT8,
+ IN in_shard_size INT8,
+ OUT out_start_row INT8,
+ OUT out_end_row INT8,
+ OUT out_progress_row INT8)
+LANGUAGE plpgsql
+AS $$
+BEGIN
+ -- Prefer an existing shard nobody is working on any more. FOR UPDATE SKIP
+ -- LOCKED hands concurrent workers *different* shards instead of letting
+ -- them collide and retry, which is what the explicit serializable
+ -- transaction this function replaced used to do.
+ SELECT start_row
+ ,end_row
+ ,progress_row
+ INTO out_start_row
+ ,out_end_row
+ ,out_progress_row
+ FROM work_shards
+ WHERE job_name=in_job_name
+ AND completed=FALSE
+ AND last_attempt<in_now
+ ORDER BY last_attempt ASC
+ LIMIT 1
+ FOR UPDATE SKIP LOCKED;
+
+ IF FOUND
+ THEN
+ -- Take over the lease. Whatever progress the previous worker committed
+ -- stands, and we return it so the caller resumes instead of restarting.
+ UPDATE work_shards
+ SET last_attempt=in_lease_until
+ WHERE job_name=in_job_name
+ AND start_row=out_start_row
+ AND end_row=out_end_row;
+ RETURN;
+ END IF;
+
+ -- No shard to take over, so open the next one past the last one known.
+ SELECT end_row
+ INTO out_start_row
+ FROM work_shards
+ WHERE job_name=in_job_name
+ ORDER BY end_row DESC
+ LIMIT 1;
+ IF NOT FOUND
+ THEN
+ out_start_row = 0; -- base case: no shards for this job yet
+ END IF;
+ out_end_row = out_start_row + in_shard_size;
+ out_progress_row = out_start_row;
+
+ INSERT INTO work_shards
+ (job_name
+ ,last_attempt
+ ,start_row
+ ,end_row
+ ,progress_row
+ ) VALUES (
+ in_job_name
+ ,in_lease_until
+ ,out_start_row
+ ,out_end_row
+ ,out_start_row
+ )
+ ON CONFLICT DO NOTHING;
+
+ IF NOT FOUND
+ THEN
+ -- Someone else opened this very shard between our two statements. Report
+ -- "nothing acquired" and let the caller come back around.
+ out_start_row = NULL;
+ out_end_row = NULL;
+ out_progress_row = NULL;
+ END IF;
+END $$;
+
+COMMENT ON FUNCTION exchange_do_begin_shard
+ IS 'Acquires a lease on a work shard for in_job_name, resuming an abandoned shard if there is one and opening a fresh shard of in_shard_size rows otherwise. Returns NULL columns if another worker won the race to open the same shard.';
diff --git a/src/exchangedb/complete_shard.c b/src/exchangedb/complete_shard.c
@@ -1,52 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2022 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU 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 General Public License for more details.
-
- You should have received a copy of the GNU General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
- */
-/**
- * @file exchangedb/complete_shard.c
- * @brief Implementation of the complete_shard function for Postgres
- * @author Christian Grothoff
- */
-#include "taler/taler_pq_lib.h"
-#include "exchange-database/complete_shard.h"
-#include "helper.h"
-
-enum GNUNET_DB_QueryStatus
-TALER_EXCHANGEDB_complete_shard (struct TALER_EXCHANGEDB_PostgresContext *pg,
- const char *job_name,
- uint64_t start_row,
- uint64_t end_row)
-{
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_string (job_name),
- GNUNET_PQ_query_param_uint64 (&start_row),
- GNUNET_PQ_query_param_uint64 (&end_row),
- GNUNET_PQ_query_param_end
- };
-
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Completing shard %llu-%llu\n",
- (unsigned long long) start_row,
- (unsigned long long) end_row);
- PREPARE (pg,
- "complete_shard",
- "UPDATE work_shards"
- " SET completed=TRUE"
- " WHERE job_name=$1"
- " AND start_row=$2"
- " AND end_row=$3");
- return GNUNET_PQ_eval_prepared_non_select (pg->conn,
- "complete_shard",
- params);
-}
diff --git a/src/exchangedb/do_expire_purse.c b/src/exchangedb/do_expire_purse.c
@@ -27,7 +27,8 @@ enum GNUNET_DB_QueryStatus
TALER_EXCHANGEDB_do_expire_purse (
struct TALER_EXCHANGEDB_PostgresContext *pg,
struct GNUNET_TIME_Absolute start_time,
- struct GNUNET_TIME_Absolute end_time)
+ struct GNUNET_TIME_Absolute end_time,
+ struct GNUNET_TIME_Absolute *purse_expiration)
{
struct GNUNET_TIME_Absolute now = GNUNET_TIME_absolute_get ();
struct GNUNET_PQ_QueryParam params[] = {
@@ -37,9 +38,14 @@ TALER_EXCHANGEDB_do_expire_purse (
GNUNET_PQ_query_param_end
};
bool found = false;
+ bool no_purse = false;
struct GNUNET_PQ_ResultSpec rs[] = {
GNUNET_PQ_result_spec_bool ("found",
&found),
+ GNUNET_PQ_result_spec_allow_null (
+ GNUNET_PQ_result_spec_absolute_time ("purse_expiration",
+ purse_expiration),
+ &no_purse),
GNUNET_PQ_result_spec_end
};
enum GNUNET_DB_QueryStatus qs;
@@ -49,6 +55,7 @@ TALER_EXCHANGEDB_do_expire_purse (
"do_expire_purse",
"SELECT "
" out_found AS found"
+ ",out_purse_expiration AS purse_expiration"
" FROM exchange_do_expire_purse"
" ($1,$2,$3);");
@@ -59,6 +66,8 @@ TALER_EXCHANGEDB_do_expire_purse (
if (qs < 0)
return qs;
GNUNET_assert (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT == qs);
+ if (no_purse)
+ *purse_expiration = end_time;
return found
? GNUNET_DB_STATUS_SUCCESS_ONE_RESULT
: GNUNET_DB_STATUS_SUCCESS_NO_RESULTS;
diff --git a/src/exchangedb/do_expire_purse.sql b/src/exchangedb/do_expire_purse.sql
@@ -14,11 +14,13 @@
-- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
--
-CREATE OR REPLACE FUNCTION exchange_do_expire_purse(
+DROP FUNCTION IF EXISTS exchange_do_expire_purse;
+CREATE FUNCTION exchange_do_expire_purse(
IN in_start_time INT8,
IN in_end_time INT8,
IN in_now INT8,
- OUT out_found BOOLEAN)
+ OUT out_found BOOLEAN,
+ OUT out_purse_expiration INT8)
LANGUAGE plpgsql
AS $$
DECLARE
@@ -33,8 +35,10 @@ BEGIN
-- and expire all at once, instead of one per query
SELECT purse_pub
,in_reserve_quota
+ ,purse_expiration
INTO my_purse_pub
,my_in_reserve_quota
+ ,out_purse_expiration
FROM purse_requests
WHERE (purse_expiration >= in_start_time) AND
(purse_expiration < in_end_time) AND
diff --git a/src/exchangedb/do_import_credits.c b/src/exchangedb/do_import_credits.c
@@ -0,0 +1,343 @@
+/*
+ This file is part of TALER
+ Copyright (C) 2022--2026 Taler Systems SA
+
+ TALER is free software; you can redistribute it and/or modify it under the
+ terms of the GNU 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 General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file exchangedb/do_import_credits.c
+ * @brief Implementation of the do_import_credits function for Postgres
+ * @author Christian Grothoff
+ * @author Joseph Xu
+ */
+#include "taler/taler_pq_lib.h"
+#include "exchange-database/do_import_credits.h"
+#include "helper.h"
+
+
+/**
+ * Compute the notification channel for a reserve that was credited.
+ *
+ * @param reserve_pub reserve to notify on
+ * @return string to pass to postgres for the notification
+ */
+static char *
+compute_notify_on_reserve (const struct TALER_ReservePublicKeyP *reserve_pub)
+{
+ struct TALER_EXCHANGEDB_ReserveEventP rep = {
+ .header.size = htons (sizeof (rep)),
+ .header.type = htons (TALER_DBEVENT_EXCHANGE_RESERVE_INCOMING),
+ .reserve_pub = *reserve_pub
+ };
+
+ return GNUNET_PQ_get_event_notify_channel (&rep.header);
+}
+
+
+/**
+ * Compute the notification channel for an account that authenticated itself.
+ *
+ * @param h_payto normalized hash of the account that was authenticated
+ * @return string to pass to postgres for the notification
+ */
+static char *
+compute_notify_on_kycauth (const struct TALER_NormalizedPaytoHashP *h_payto)
+{
+ struct TALER_EXCHANGEDB_KycCompletedEventP rep = {
+ .header.size = htons (sizeof (rep)),
+ .header.type = htons (TALER_DBEVENT_EXCHANGE_KYC_COMPLETED),
+ .h_payto = *h_payto
+ };
+
+ return GNUNET_PQ_get_event_notify_channel (&rep.header);
+}
+
+
+/**
+ * Closure for #helper_cb().
+ */
+struct Context
+{
+ /**
+ * Array with entries set to 'true' for transfers we had already imported.
+ */
+ bool *transaction_duplicates;
+
+ /**
+ * Length of the @e transaction_duplicates array.
+ */
+ unsigned int length;
+
+ /**
+ * Set to #GNUNET_SYSERR on failures.
+ */
+ enum GNUNET_GenericReturnValue status;
+};
+
+
+/**
+ * Function called with the rows the stored procedure returned, one per
+ * reserve transfer, in input order.
+ *
+ * @param cls closure of type `struct Context *`
+ * @param result the postgres result
+ * @param num_results the number of results in @a result
+ */
+static void
+helper_cb (void *cls,
+ PGresult *result,
+ unsigned int num_results)
+{
+ struct Context *ctx = cls;
+
+ if (num_results > ctx->length)
+ {
+ /* The procedure returns one row per reserve transfer we passed in; a
+ different count means the two have drifted apart, and writing past the
+ caller's array is not the way to find that out. */
+ GNUNET_break (0);
+ ctx->status = GNUNET_SYSERR;
+ return;
+ }
+ for (unsigned int i = 0; i<num_results; i++)
+ {
+ struct GNUNET_PQ_ResultSpec rs[] = {
+ GNUNET_PQ_result_spec_bool ("out_duplicate",
+ &ctx->transaction_duplicates[i]),
+ GNUNET_PQ_result_spec_end
+ };
+
+ if (GNUNET_OK !=
+ GNUNET_PQ_extract_result (result,
+ rs,
+ i))
+ {
+ GNUNET_break (0);
+ ctx->status = GNUNET_SYSERR;
+ return;
+ }
+ }
+}
+
+
+enum GNUNET_DB_QueryStatus
+TALER_EXCHANGEDB_do_import_credits (
+ struct TALER_EXCHANGEDB_PostgresContext *pg,
+ const struct TALER_EXCHANGEDB_CreditBatch *batch,
+ enum GNUNET_DB_QueryStatus *reserve_results)
+{
+ unsigned int rlen = batch->reserves_length;
+ unsigned int klen = batch->kycauths_length;
+ unsigned int wlen = batch->wads_length;
+ unsigned int dups = 0;
+
+ /* reserve transfers */
+ struct TALER_ReservePublicKeyP reserve_pubs[GNUNET_NZL (rlen)];
+ uint64_t wire_references[GNUNET_NZL (rlen)];
+ struct TALER_Amount balances[GNUNET_NZL (rlen)];
+ struct GNUNET_TIME_Timestamp execution_times[GNUNET_NZL (rlen)];
+ struct TALER_FullPaytoHashP h_full_paytos[GNUNET_NZL (rlen)];
+ struct TALER_NormalizedPaytoHashP h_normalized_paytos[GNUNET_NZL (rlen)];
+ const char *payto_uris[GNUNET_NZL (rlen)];
+ char *notify_s[GNUNET_NZL (rlen)];
+ bool transaction_duplicates[GNUNET_NZL (rlen)];
+
+ /* KYC authentication transfers */
+ union TALER_AccountPublicKeyP ka_account_pubs[GNUNET_NZL (klen)];
+ uint64_t ka_wire_references[GNUNET_NZL (klen)];
+ struct TALER_Amount ka_balances[GNUNET_NZL (klen)];
+ struct GNUNET_TIME_Timestamp ka_execution_times[GNUNET_NZL (klen)];
+ struct TALER_FullPaytoHashP ka_h_full_paytos[GNUNET_NZL (klen)];
+ struct TALER_NormalizedPaytoHashP ka_h_normalized_paytos[GNUNET_NZL (klen)];
+ const char *ka_payto_uris[GNUNET_NZL (klen)];
+ char *ka_notify_s[GNUNET_NZL (klen)];
+
+ /* WAD transfers */
+ struct TALER_WadIdentifierP wad_ids[GNUNET_NZL (wlen)];
+ const char *wad_origin_exchange_urls[GNUNET_NZL (wlen)];
+ struct TALER_Amount wad_balances[GNUNET_NZL (wlen)];
+ struct GNUNET_TIME_Timestamp wad_execution_times[GNUNET_NZL (wlen)];
+
+ struct GNUNET_TIME_Timestamp reserve_expiration
+ = GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time);
+ struct GNUNET_TIME_Timestamp gc
+ = GNUNET_TIME_relative_to_timestamp (pg->legal_reserve_expiration_time);
+ struct GNUNET_TIME_Absolute lease_until
+ = GNUNET_TIME_relative_to_absolute (batch->lease);
+ enum GNUNET_DB_QueryStatus qs;
+
+ for (unsigned int i = 0; i<rlen; i++)
+ {
+ const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &batch->reserves[i];
+
+ TALER_full_payto_hash (reserve->sender_account_details,
+ &h_full_paytos[i]);
+ TALER_full_payto_normalize_and_hash (reserve->sender_account_details,
+ &h_normalized_paytos[i]);
+ notify_s[i] = compute_notify_on_reserve (reserve->reserve_pub);
+ reserve_pubs[i] = *reserve->reserve_pub;
+ balances[i] = *reserve->balance;
+ execution_times[i] = reserve->execution_time;
+ payto_uris[i] = reserve->sender_account_details.full_payto;
+ wire_references[i] = reserve->wire_reference;
+ }
+ for (unsigned int i = 0; i<klen; i++)
+ {
+ const struct TALER_EXCHANGEDB_KycauthInInfo *ka = &batch->kycauths[i];
+
+ TALER_full_payto_hash (ka->sender_account_details,
+ &ka_h_full_paytos[i]);
+ TALER_full_payto_normalize_and_hash (ka->sender_account_details,
+ &ka_h_normalized_paytos[i]);
+ ka_notify_s[i] = compute_notify_on_kycauth (&ka_h_normalized_paytos[i]);
+ ka_account_pubs[i] = *ka->account_pub;
+ ka_balances[i] = *ka->balance;
+ ka_execution_times[i] = ka->execution_time;
+ ka_payto_uris[i] = ka->sender_account_details.full_payto;
+ ka_wire_references[i] = ka->wire_reference;
+ }
+ for (unsigned int i = 0; i<wlen; i++)
+ {
+ const struct TALER_EXCHANGEDB_WadInInfo *wad = &batch->wads[i];
+
+ wad_ids[i] = *wad->wad_id;
+ wad_origin_exchange_urls[i] = wad->origin_exchange_url;
+ wad_balances[i] = *wad->balance;
+ wad_execution_times[i] = wad->execution_time;
+ }
+
+ PREPARE (pg,
+ "do_import_credits",
+ "SELECT"
+ " out_duplicate"
+ " FROM exchange_do_import_credits"
+ " ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17"
+ " ,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28);");
+ {
+ struct GNUNET_PQ_QueryParam params[] = {
+ GNUNET_PQ_query_param_string (batch->exchange_account_name),
+ GNUNET_PQ_query_param_timestamp (&gc),
+ GNUNET_PQ_query_param_timestamp (&reserve_expiration),
+ /* reserve transfers */
+ GNUNET_PQ_query_param_array_auto_from_type (rlen,
+ reserve_pubs,
+ pg->conn),
+ GNUNET_PQ_query_param_array_uint64 (rlen,
+ wire_references,
+ pg->conn),
+ TALER_PQ_query_param_array_amount (rlen,
+ balances,
+ pg->conn),
+ GNUNET_PQ_query_param_array_timestamp (rlen,
+ execution_times,
+ pg->conn),
+ GNUNET_PQ_query_param_array_auto_from_type (rlen,
+ h_full_paytos,
+ pg->conn),
+ GNUNET_PQ_query_param_array_auto_from_type (rlen,
+ h_normalized_paytos,
+ pg->conn),
+ GNUNET_PQ_query_param_array_ptrs_string (rlen,
+ payto_uris,
+ pg->conn),
+ GNUNET_PQ_query_param_array_ptrs_string (rlen,
+ (const char **) notify_s,
+ pg->conn),
+ /* KYC authentication transfers */
+ GNUNET_PQ_query_param_array_auto_from_type (klen,
+ ka_account_pubs,
+ pg->conn),
+ GNUNET_PQ_query_param_array_uint64 (klen,
+ ka_wire_references,
+ pg->conn),
+ TALER_PQ_query_param_array_amount (klen,
+ ka_balances,
+ pg->conn),
+ GNUNET_PQ_query_param_array_timestamp (klen,
+ ka_execution_times,
+ pg->conn),
+ GNUNET_PQ_query_param_array_auto_from_type (klen,
+ ka_h_full_paytos,
+ pg->conn),
+ GNUNET_PQ_query_param_array_auto_from_type (klen,
+ ka_h_normalized_paytos,
+ pg->conn),
+ GNUNET_PQ_query_param_array_ptrs_string (klen,
+ ka_payto_uris,
+ pg->conn),
+ GNUNET_PQ_query_param_array_ptrs_string (klen,
+ (const char **) ka_notify_s,
+ pg->conn),
+ /* WAD transfers */
+ GNUNET_PQ_query_param_array_auto_from_type (wlen,
+ wad_ids,
+ pg->conn),
+ GNUNET_PQ_query_param_array_ptrs_string (wlen,
+ wad_origin_exchange_urls,
+ pg->conn),
+ TALER_PQ_query_param_array_amount (wlen,
+ wad_balances,
+ pg->conn),
+ GNUNET_PQ_query_param_array_timestamp (wlen,
+ wad_execution_times,
+ pg->conn),
+ /* shard bookkeeping */
+ GNUNET_PQ_query_param_string (batch->job_name),
+ GNUNET_PQ_query_param_uint64 (&batch->shard_start),
+ GNUNET_PQ_query_param_uint64 (&batch->shard_end),
+ GNUNET_PQ_query_param_uint64 (&batch->progress_row),
+ GNUNET_PQ_query_param_absolute_time (&lease_until),
+ GNUNET_PQ_query_param_end
+ };
+ struct Context ctx = {
+ .transaction_duplicates = transaction_duplicates,
+ .length = rlen,
+ .status = GNUNET_OK
+ };
+
+ qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
+ "do_import_credits",
+ params,
+ &helper_cb,
+ &ctx);
+ GNUNET_PQ_cleanup_query_params_closures (params);
+ if ( (qs >= 0) &&
+ (GNUNET_OK != ctx.status) )
+ qs = GNUNET_DB_STATUS_HARD_ERROR;
+ if (qs < 0)
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Failed to import credit batch (%d)\n",
+ qs);
+ }
+
+ for (unsigned int i = 0; i<rlen; i++)
+ GNUNET_free (notify_s[i]);
+ for (unsigned int i = 0; i<klen; i++)
+ GNUNET_free (ka_notify_s[i]);
+ if (qs < 0)
+ return qs;
+ for (unsigned int i = 0; i<rlen; i++)
+ {
+ if (transaction_duplicates[i])
+ dups++;
+ reserve_results[i] = transaction_duplicates[i]
+ ? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS
+ : GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
+ }
+ GNUNET_PQ_event_do_poll (pg->conn);
+ if (0 != dups)
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%u/%u duplicates among incoming transactions. Try increasing WIREWATCH_IDLE_SLEEP_INTERVAL in the [exchange] configuration section (if this happens a lot).\n",
+ dups,
+ rlen);
+ return qs;
+}
diff --git a/src/exchangedb/do_import_credits.sql b/src/exchangedb/do_import_credits.sql
@@ -0,0 +1,301 @@
+--
+-- This file is part of TALER
+-- Copyright (C) 2014--2026 Taler Systems SA
+--
+-- TALER is free software; you can redistribute it and/or modify it under the
+-- terms of the GNU 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 General Public License for more details.
+--
+-- You should have received a copy of the GNU General Public License along with
+-- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+--
+
+DROP FUNCTION IF EXISTS exchange_do_import_credits;
+CREATE FUNCTION exchange_do_import_credits(
+ -- All credits in one call come from one bank account and one HTTP
+ -- response, so the account and the reserve lifetimes are scalars.
+ IN in_exchange_account_name TEXT,
+ IN in_gc_date INT8,
+ IN in_reserve_expiration INT8,
+ -- Transfers into reserves.
+ IN ina_reserve_pub BYTEA[],
+ IN ina_wire_ref INT8[],
+ IN ina_credit taler_amount[],
+ IN ina_execution_date INT8[],
+ IN ina_wire_source_h_payto BYTEA[],
+ IN ina_h_normalized_payto BYTEA[],
+ IN ina_payto_uri TEXT[],
+ IN ina_notify TEXT[],
+ -- KYC authentication transfers.
+ IN ina_ka_account_pub BYTEA[],
+ IN ina_ka_wire_ref INT8[],
+ IN ina_ka_credit taler_amount[],
+ IN ina_ka_execution_date INT8[],
+ IN ina_ka_wire_source_h_payto BYTEA[],
+ IN ina_ka_h_normalized_payto BYTEA[],
+ IN ina_ka_payto_uri TEXT[],
+ IN ina_ka_notify TEXT[],
+ -- WAD transfers from other exchanges.
+ IN ina_wad_id BYTEA[],
+ IN ina_wad_origin_exchange_url TEXT[],
+ IN ina_wad_amount taler_amount[],
+ IN ina_wad_execution_date INT8[],
+ -- Which shard this batch belongs to, and how far it takes us.
+ IN in_job_name TEXT,
+ IN in_shard_start INT8,
+ IN in_shard_end INT8,
+ IN in_progress_row INT8,
+ IN in_lease_until INT8)
+RETURNS TABLE (out_duplicate BOOLEAN)
+LANGUAGE plpgsql
+AS $$
+DECLARE
+ conflict BOOL;
+ dup BOOL;
+ uuid INT8;
+ i INT4;
+ my_is_wallet BOOL;
+BEGIN
+
+ -- === transfers into reserves ============================================
+ FOR i IN 1..COALESCE(array_length(ina_reserve_pub,1),0)
+ LOOP
+ my_is_wallet
+ = (LOWER (SUBSTRING (ina_payto_uri[i], 0, 23)) =
+ 'payto://taler-reserve/') OR
+ (LOWER (SUBSTRING (ina_payto_uri[i], 0, 28)) =
+ 'payto://taler-reserve-http/');
+ INSERT INTO kyc_targets
+ (h_normalized_payto
+ ,is_wallet
+ ) VALUES (
+ ina_h_normalized_payto[i]
+ ,my_is_wallet
+ )
+ ON CONFLICT DO NOTHING;
+ INSERT INTO wire_targets
+ (wire_target_h_payto
+ ,h_normalized_payto
+ ,payto_uri
+ ) VALUES (
+ ina_wire_source_h_payto[i]
+ ,ina_h_normalized_payto[i]
+ ,ina_payto_uri[i]
+ )
+ ON CONFLICT DO NOTHING;
+
+ INSERT INTO reserves
+ (reserve_pub
+ ,current_balance
+ ,expiration_date
+ ,gc_date
+ ) VALUES (
+ ina_reserve_pub[i]
+ ,ina_credit[i]
+ ,in_reserve_expiration
+ ,in_gc_date
+ )
+ ON CONFLICT DO NOTHING
+ RETURNING reserve_uuid
+ INTO uuid;
+ conflict = NOT FOUND;
+
+ INSERT INTO reserves_in
+ (reserve_pub
+ ,wire_reference
+ ,credit
+ ,exchange_account_section
+ ,wire_source_h_payto
+ ,execution_date
+ ) VALUES (
+ ina_reserve_pub[i]
+ ,ina_wire_ref[i]
+ ,ina_credit[i]
+ ,in_exchange_account_name
+ ,ina_wire_source_h_payto[i]
+ ,ina_execution_date[i]
+ )
+ ON CONFLICT DO NOTHING;
+
+ IF NOT FOUND
+ THEN
+ IF conflict
+ THEN
+ dup = TRUE;
+ ELSE
+ dup = FALSE;
+ END IF;
+ ELSE
+ IF NOT conflict
+ THEN
+ EXECUTE FORMAT (
+ 'NOTIFY %s'
+ ,ina_notify[i]);
+ END IF;
+ dup = FALSE;
+ END IF;
+
+ IF (conflict AND NOT dup)
+ THEN
+ -- The reserve already existed, so the INSERT above did not credit it.
+ -- This is the body of the former exchange_do_batch_reserves_update(),
+ -- which the caller used to run in a *second* transaction, one
+ -- round-trip per affected reserve. It is reproduced here verbatim so
+ -- that this function behaves exactly as the sequence it replaces --
+ -- including the fact that its INSERT can never fire, because the
+ -- statement above already put that very row into reserves_in. See
+ -- the separate report on the lost credit; fixing it here would have
+ -- buried a change of behaviour inside a refactoring.
+ INSERT INTO reserves_in
+ (reserve_pub
+ ,wire_reference
+ ,credit
+ ,exchange_account_section
+ ,wire_source_h_payto
+ ,execution_date
+ ) VALUES (
+ ina_reserve_pub[i]
+ ,ina_wire_ref[i]
+ ,ina_credit[i]
+ ,in_exchange_account_name
+ ,ina_wire_source_h_payto[i]
+ ,in_reserve_expiration
+ )
+ ON CONFLICT DO NOTHING;
+ IF FOUND
+ THEN
+ dup = FALSE;
+ UPDATE reserves rs
+ SET
+ current_balance.frac = (rs.current_balance).frac+(ina_credit[i]).frac
+ - CASE
+ WHEN (rs.current_balance).frac + (ina_credit[i]).frac >= 100000000
+ THEN 100000000
+ ELSE 0
+ END
+ ,current_balance.val = (rs.current_balance).val+(ina_credit[i]).val
+ + CASE
+ WHEN (rs.current_balance).frac + (ina_credit[i]).frac >= 100000000
+ THEN 1
+ ELSE 0
+ END
+ ,expiration_date=GREATEST(expiration_date,in_reserve_expiration)
+ ,gc_date=GREATEST(gc_date,in_reserve_expiration)
+ WHERE reserve_pub=ina_reserve_pub[i];
+ EXECUTE FORMAT (
+ 'NOTIFY %s'
+ ,ina_notify[i]);
+ ELSE
+ dup = TRUE;
+ END IF;
+ END IF;
+
+ out_duplicate = dup;
+ RETURN NEXT;
+ END LOOP;
+
+ -- === KYC authentication transfers =======================================
+ FOR i IN 1..COALESCE(array_length(ina_ka_account_pub,1),0)
+ LOOP
+ INSERT INTO kycauths_in
+ (account_pub
+ ,wire_reference
+ ,credit
+ ,wire_source_h_payto
+ ,exchange_account_section
+ ,execution_date
+ ) VALUES (
+ ina_ka_account_pub[i]
+ ,ina_ka_wire_ref[i]
+ ,ina_ka_credit[i]
+ ,ina_ka_wire_source_h_payto[i]
+ ,in_exchange_account_name
+ ,ina_ka_execution_date[i]
+ )
+ ON CONFLICT DO NOTHING;
+
+ IF NOT FOUND
+ THEN
+ -- presumably already done
+ CONTINUE;
+ END IF;
+
+ UPDATE kyc_targets
+ SET target_pub=ina_ka_account_pub[i]
+ WHERE h_normalized_payto=ina_ka_h_normalized_payto[i];
+
+ IF NOT FOUND
+ THEN
+ -- First time we see this account, setup everything.
+ my_is_wallet
+ = (LOWER (SUBSTRING (ina_ka_payto_uri[i], 0, 23)) =
+ 'payto://taler-reserve/') OR
+ (LOWER (SUBSTRING (ina_ka_payto_uri[i], 0, 28)) =
+ 'payto://taler-reserve-http/');
+ INSERT INTO kyc_targets
+ (h_normalized_payto
+ ,is_wallet
+ ,target_pub
+ ) VALUES (
+ ina_ka_h_normalized_payto[i]
+ ,my_is_wallet
+ ,ina_ka_account_pub[i]);
+ INSERT INTO wire_targets
+ (wire_target_h_payto
+ ,h_normalized_payto
+ ,payto_uri
+ ) VALUES (
+ ina_ka_wire_source_h_payto[i]
+ ,ina_ka_h_normalized_payto[i]
+ ,ina_ka_payto_uri[i]);
+ END IF;
+
+ EXECUTE FORMAT (
+ 'NOTIFY %s'
+ ,ina_ka_notify[i]);
+ END LOOP;
+
+ -- === WAD transfers ======================================================
+ FOR i IN 1..COALESCE(array_length(ina_wad_id,1),0)
+ LOOP
+ -- ON CONFLICT DO NOTHING is what the per-row INSERT this replaces
+ -- effectively did: it let the unique violation through to libgnunetpq,
+ -- which reports 23505 as "no results" and the caller shrugged it off.
+ -- Inside a function an unhandled violation would take the whole batch
+ -- down instead, and re-importing a shard has to stay harmless.
+ INSERT INTO wads_in
+ (wad_id
+ ,origin_exchange_url
+ ,amount
+ ,arrival_time
+ ) VALUES (
+ ina_wad_id[i]
+ ,ina_wad_origin_exchange_url[i]
+ ,ina_wad_amount[i]
+ ,ina_wad_execution_date[i]
+ )
+ ON CONFLICT DO NOTHING;
+ END LOOP;
+
+ -- === shard bookkeeping ==================================================
+ -- Deliberately part of this statement rather than a follow-up transaction:
+ -- the imports above and the record of how far we got have to become
+ -- visible together. Marking the shard completed is the same update, so
+ -- finishing a shard costs nothing extra either.
+ UPDATE work_shards
+ SET progress_row=GREATEST(progress_row,in_progress_row)
+ ,completed=(GREATEST(progress_row,in_progress_row) >= end_row)
+ ,last_attempt=in_lease_until
+ WHERE job_name=in_job_name
+ AND start_row=in_shard_start
+ AND end_row=in_shard_end;
+
+ RETURN;
+END $$;
+
+COMMENT ON FUNCTION exchange_do_import_credits
+ IS 'Imports one bank credit history response (reserve, KYC authentication and WAD transfers alike) and advances the work shard it belongs to, all in one statement. Returns one row per entry of ina_reserve_pub, in order, saying whether that transfer had already been imported.';
diff --git a/src/exchangedb/do_insert_reserve_in.c b/src/exchangedb/do_insert_reserve_in.c
@@ -1,374 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2022-2024 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU 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 General Public License for more details.
-
- You should have received a copy of the GNU General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
- */
-/**
- * @file exchangedb/do_insert_reserve_in.c
- * @brief Implementation of the do_insert_reserve_in function for Postgres
- * @author Christian Grothoff
- * @author Joseph Xu
- */
-#include "taler/taler_pq_lib.h"
-#include "exchange-database/do_insert_reserve_in.h"
-#include "helper.h"
-#include "exchange-database/start.h"
-#include "exchange-database/start_read_committed.h"
-#include "exchange-database/commit.h"
-#include "exchange-database/preflight.h"
-#include "exchange-database/rollback.h"
-
-
-/**
- * Generate event notification for the reserve change.
- *
- * @param reserve_pub reserve to notfiy on
- * @return string to pass to postgres for the notification
- */
-static char *
-compute_notify_on_reserve (const struct TALER_ReservePublicKeyP *reserve_pub)
-{
- struct TALER_EXCHANGEDB_ReserveEventP rep = {
- .header.size = htons (sizeof (rep)),
- .header.type = htons (TALER_DBEVENT_EXCHANGE_RESERVE_INCOMING),
- .reserve_pub = *reserve_pub
- };
-
- return GNUNET_PQ_get_event_notify_channel (&rep.header);
-}
-
-
-/**
- * Closure for our helper_cb()
- */
-struct Context
-{
- /**
- * Array of reserve UUIDs to initialize.
- */
- uint64_t *reserve_uuids;
-
- /**
- * Array with entries set to 'true' for duplicate transactions.
- */
- bool *transaction_duplicates;
-
- /**
- * Array with entries set to 'true' for rows with conflicts.
- */
- bool *conflicts;
-
- /**
- * Set to #GNUNET_SYSERR on failures.
- */
- enum GNUNET_GenericReturnValue status;
-
- /**
- * Single value (no array) set to true if we need
- * to follow-up with an update.
- */
- bool needs_update;
-};
-
-
-/**
- * Helper function to be called with the results of a SELECT statement
- * that has returned @a num_results results.
- *
- * @param cls closure of type `struct Context *`
- * @param result the postgres result
- * @param num_results the number of results in @a result
- */
-static void
-helper_cb (void *cls,
- PGresult *result,
- unsigned int num_results)
-{
- struct Context *ctx = cls;
-
- for (unsigned int i = 0; i<num_results; i++)
- {
- struct GNUNET_PQ_ResultSpec rs[] = {
- GNUNET_PQ_result_spec_bool (
- "transaction_duplicate",
- &ctx->transaction_duplicates[i]),
- GNUNET_PQ_result_spec_allow_null (
- GNUNET_PQ_result_spec_uint64 ("ruuid",
- &ctx->reserve_uuids[i]),
- &ctx->conflicts[i]),
- GNUNET_PQ_result_spec_end
- };
-
- if (GNUNET_OK !=
- GNUNET_PQ_extract_result (result,
- rs,
- i))
- {
- GNUNET_break (0);
- ctx->status = GNUNET_SYSERR;
- return;
- }
- if (! ctx->transaction_duplicates[i])
- ctx->needs_update |= ctx->conflicts[i];
- }
-}
-
-
-enum GNUNET_DB_QueryStatus
-TALER_EXCHANGEDB_do_insert_reserve_in (
- struct TALER_EXCHANGEDB_PostgresContext *pg,
- const struct TALER_EXCHANGEDB_ReserveInInfo *reserves,
- unsigned int reserves_length,
- enum GNUNET_DB_QueryStatus *results)
-{
- unsigned int dups = 0;
-
- struct TALER_FullPaytoHashP h_full_paytos[
- GNUNET_NZL (reserves_length)];
- struct TALER_NormalizedPaytoHashP h_normalized_paytos[
- GNUNET_NZL (reserves_length)];
- char *notify_s[GNUNET_NZL (reserves_length)];
- struct TALER_ReservePublicKeyP reserve_pubs[GNUNET_NZL (reserves_length)];
- struct TALER_Amount balances[GNUNET_NZL (reserves_length)];
- struct GNUNET_TIME_Timestamp execution_times[GNUNET_NZL (reserves_length)];
- const char *sender_account_details[GNUNET_NZL (reserves_length)];
- const char *exchange_account_names[GNUNET_NZL (reserves_length)];
- uint64_t wire_references[GNUNET_NZL (reserves_length)];
- uint64_t reserve_uuids[GNUNET_NZL (reserves_length)];
- bool transaction_duplicates[GNUNET_NZL (reserves_length)];
- bool conflicts[GNUNET_NZL (reserves_length)];
- struct GNUNET_TIME_Timestamp reserve_expiration
- = GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time);
- struct GNUNET_TIME_Timestamp gc
- = GNUNET_TIME_relative_to_timestamp (pg->legal_reserve_expiration_time);
- enum GNUNET_DB_QueryStatus qs;
- bool need_update;
-
- for (unsigned int i = 0; i<reserves_length; i++)
- {
- const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
-
- TALER_full_payto_hash (reserve->sender_account_details,
- &h_full_paytos[i]);
- TALER_full_payto_normalize_and_hash (reserve->sender_account_details,
- &h_normalized_paytos[i]);
- notify_s[i] = compute_notify_on_reserve (reserve->reserve_pub);
- reserve_pubs[i] = *reserve->reserve_pub;
- balances[i] = *reserve->balance;
- execution_times[i] = reserve->execution_time;
- sender_account_details[i] = reserve->sender_account_details.full_payto;
- exchange_account_names[i] = reserve->exchange_account_name;
- wire_references[i] = reserve->wire_reference;
- }
-
- /* NOTE: kind-of pointless to explicitly start a transaction here... */
- if (GNUNET_OK !=
- TALER_EXCHANGEDB_preflight (pg))
- {
- GNUNET_break (0);
- qs = GNUNET_DB_STATUS_HARD_ERROR;
- goto finished;
- }
- if (GNUNET_OK !=
- TALER_EXCHANGEDB_start_read_committed (pg,
- "READ_COMMITED"))
- {
- GNUNET_break (0);
- qs = GNUNET_DB_STATUS_HARD_ERROR;
- goto finished;
- }
- PREPARE (pg,
- "do_insert_reserve_in_reserves_insert_with_array",
- "SELECT"
- " transaction_duplicate"
- ",ruuid"
- " FROM exchange_do_array_reserves_insert"
- " ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11);");
- {
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_timestamp (&gc),
- GNUNET_PQ_query_param_timestamp (&reserve_expiration),
- GNUNET_PQ_query_param_array_auto_from_type (reserves_length,
- reserve_pubs,
- pg->conn),
- GNUNET_PQ_query_param_array_uint64 (reserves_length,
- wire_references,
- pg->conn),
- TALER_PQ_query_param_array_amount (
- reserves_length,
- balances,
- pg->conn),
- GNUNET_PQ_query_param_array_ptrs_string (
- reserves_length,
- (const char **) exchange_account_names,
- pg->conn),
- GNUNET_PQ_query_param_array_timestamp (
- reserves_length,
- execution_times,
- pg->conn),
- GNUNET_PQ_query_param_array_auto_from_type (
- reserves_length,
- h_full_paytos,
- pg->conn),
- GNUNET_PQ_query_param_array_auto_from_type (
- reserves_length,
- h_normalized_paytos,
- pg->conn),
- GNUNET_PQ_query_param_array_ptrs_string (
- reserves_length,
- (const char **) sender_account_details,
- pg->conn),
- GNUNET_PQ_query_param_array_ptrs_string (
- reserves_length,
- (const char **) notify_s,
- pg->conn),
- GNUNET_PQ_query_param_end
- };
- struct Context ctx = {
- .reserve_uuids = reserve_uuids,
- .transaction_duplicates = transaction_duplicates,
- .conflicts = conflicts,
- .needs_update = false,
- .status = GNUNET_OK
- };
-
- qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
- "do_insert_reserve_in_reserves_insert_with_array",
- params,
- &helper_cb,
- &ctx);
- GNUNET_PQ_cleanup_query_params_closures (params);
- if ( (qs < 0) ||
- (GNUNET_OK != ctx.status) )
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Failed to insert into reserves (%d)\n",
- qs);
- goto finished;
- }
- need_update = ctx.needs_update;
- }
-
- {
- enum GNUNET_DB_QueryStatus cs;
-
- cs = TALER_EXCHANGEDB_commit (pg);
- if (cs < 0)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Failed to commit\n");
- qs = cs;
- goto finished;
- }
- }
-
- for (unsigned int i = 0; i<reserves_length; i++)
- {
- if (transaction_duplicates[i])
- dups++;
- results[i] = transaction_duplicates[i]
- ? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS
- : GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
- }
-
- if (! need_update)
- {
- qs = reserves_length;
- goto finished;
- }
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "Reserve update needed for some reserves in the batch\n");
- PREPARE (pg,
- "do_insert_reserve_in_reserves_update",
- "SELECT"
- " out_duplicate AS duplicate "
- "FROM exchange_do_batch_reserves_update"
- " ($1,$2,$3,$4,$5,$6,$7);");
-
- if (GNUNET_OK !=
- TALER_EXCHANGEDB_start (pg,
- "reserve-insert-continued"))
- {
- GNUNET_break (0);
- qs = GNUNET_DB_STATUS_HARD_ERROR;
- goto finished;
- }
-
- for (unsigned int i = 0; i<reserves_length; i++)
- {
- if (transaction_duplicates[i])
- continue;
- if (! conflicts[i])
- continue;
- {
- bool duplicate;
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_auto_from_type (&reserve_pubs[i]),
- GNUNET_PQ_query_param_timestamp (&reserve_expiration),
- GNUNET_PQ_query_param_uint64 (&wire_references[i]),
- TALER_PQ_query_param_amount (pg->conn,
- &balances[i]),
- GNUNET_PQ_query_param_string (exchange_account_names[i]),
- GNUNET_PQ_query_param_auto_from_type (&h_full_paytos[i]),
- GNUNET_PQ_query_param_string (notify_s[i]),
- GNUNET_PQ_query_param_end
- };
- struct GNUNET_PQ_ResultSpec rs[] = {
- GNUNET_PQ_result_spec_bool ("duplicate",
- &duplicate),
- GNUNET_PQ_result_spec_end
- };
- enum GNUNET_DB_QueryStatus qsi;
-
- qsi = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
- "do_insert_reserve_in_reserves_update",
- params,
- rs);
- if (qsi < 0)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Failed to update reserves (%d)\n",
- qsi);
- results[i] = qsi;
- goto finished;
- }
- results[i] = duplicate
- ? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS
- : GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
- }
- }
- {
- enum GNUNET_DB_QueryStatus cs;
-
- cs = TALER_EXCHANGEDB_commit (pg);
- if (cs < 0)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Failed to commit\n");
- qs = cs;
- goto finished;
- }
- }
-finished:
- for (unsigned int i = 0; i<reserves_length; i++)
- GNUNET_free (notify_s[i]);
- if (qs < 0)
- return qs;
- GNUNET_PQ_event_do_poll (pg->conn);
- if (0 != dups)
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "%u/%u duplicates among incoming transactions. Try increasing WIREWATCH_IDLE_SLEEP_INTERVAL in the [exchange] configuration section (if this happens a lot).\n",
- dups,
- reserves_length);
- return qs;
-}
diff --git a/src/exchangedb/do_insert_reserve_in.sql b/src/exchangedb/do_insert_reserve_in.sql
@@ -1,200 +0,0 @@
---
--- This file is part of TALER
--- Copyright (C) 2014--2024 Taler Systems SA
---
--- TALER is free software; you can redistribute it and/or modify it under the
--- terms of the GNU 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 General Public License for more details.
---
--- You should have received a copy of the GNU General Public License along with
--- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
---
-
-
-DROP FUNCTION IF EXISTS exchange_do_array_reserves_insert;
-CREATE FUNCTION exchange_do_array_reserves_insert(
- IN in_gc_date INT8,
- IN in_reserve_expiration INT8,
- IN ina_reserve_pub BYTEA[],
- IN ina_wire_ref INT8[],
- IN ina_credit taler_amount[],
- IN ina_exchange_account_name TEXT[],
- IN ina_execution_date INT8[],
- IN ina_wire_source_h_payto BYTEA[],
- IN ina_h_normalized_payto BYTEA[],
- IN ina_payto_uri TEXT[],
- IN ina_notify TEXT[])
-RETURNS SETOF exchange_do_array_reserve_insert_return_type
-LANGUAGE plpgsql
-AS $$
-DECLARE
- conflict BOOL;
- dup BOOL;
- uuid INT8;
- i INT4;
- my_is_wallet BOOL;
- ini_reserve_pub BYTEA;
- ini_wire_ref INT8;
- ini_credit taler_amount;
- ini_exchange_account_name TEXT;
- ini_execution_date INT8;
- ini_wire_source_h_payto BYTEA;
- ini_h_normalized_payto BYTEA;
- ini_payto_uri TEXT;
- ini_notify TEXT;
-BEGIN
-
- FOR i IN 1..array_length(ina_reserve_pub,1)
- LOOP
- ini_reserve_pub = ina_reserve_pub[i];
- ini_wire_ref = ina_wire_ref[i];
- ini_credit = ina_credit[i];
- ini_exchange_account_name = ina_exchange_account_name[i];
- ini_execution_date = ina_execution_date[i];
- ini_wire_source_h_payto = ina_wire_source_h_payto[i];
- ini_h_normalized_payto = ina_h_normalized_payto[i];
- ini_payto_uri = ina_payto_uri[i];
- ini_notify = ina_notify[i];
-
--- RAISE WARNING 'Starting loop on %', ini_notify;
-
- my_is_wallet
- = (LOWER (SUBSTRING (ini_payto_uri, 0, 23)) =
- 'payto://taler-reserve/') OR
- (LOWER (SUBSTRING (ini_payto_uri, 0, 28)) =
- 'payto://taler-reserve-http/');
- INSERT INTO kyc_targets
- (h_normalized_payto
- ,is_wallet
- ) VALUES (
- ini_h_normalized_payto
- ,my_is_wallet
- )
- ON CONFLICT DO NOTHING;
- INSERT INTO wire_targets
- (wire_target_h_payto
- ,h_normalized_payto
- ,payto_uri
- ) VALUES (
- ini_wire_source_h_payto
- ,ini_h_normalized_payto
- ,ini_payto_uri
- )
- ON CONFLICT DO NOTHING;
-
- INSERT INTO reserves
- (reserve_pub
- ,current_balance
- ,expiration_date
- ,gc_date
- ) VALUES (
- ini_reserve_pub
- ,ini_credit
- ,in_reserve_expiration
- ,in_gc_date
- )
- ON CONFLICT DO NOTHING
- RETURNING reserve_uuid
- INTO uuid;
- conflict = NOT FOUND;
-
- INSERT INTO reserves_in
- (reserve_pub
- ,wire_reference
- ,credit
- ,exchange_account_section
- ,wire_source_h_payto
- ,execution_date
- ) VALUES (
- ini_reserve_pub
- ,ini_wire_ref
- ,ini_credit
- ,ini_exchange_account_name
- ,ini_wire_source_h_payto
- ,ini_execution_date
- )
- ON CONFLICT DO NOTHING;
-
- IF NOT FOUND
- THEN
- IF conflict
- THEN
- dup = TRUE;
- else
- dup = FALSE;
- END IF;
- ELSE
- IF NOT conflict
- THEN
- EXECUTE FORMAT (
- 'NOTIFY %s'
- ,ini_notify);
- END IF;
- dup = FALSE;
- END IF;
- RETURN NEXT (dup,uuid);
- END LOOP;
- RETURN;
-END $$;
-
-
-CREATE OR REPLACE FUNCTION exchange_do_batch_reserves_update(
- IN in_reserve_pub BYTEA,
- IN in_expiration_date INT8,
- IN in_wire_ref INT8,
- IN in_credit taler_amount,
- IN in_exchange_account_name TEXT,
- IN in_wire_source_h_payto BYTEA,
- IN in_notify text,
- OUT out_duplicate BOOLEAN)
-LANGUAGE plpgsql
-AS $$
-BEGIN
- INSERT INTO reserves_in
- (reserve_pub
- ,wire_reference
- ,credit
- ,exchange_account_section
- ,wire_source_h_payto
- ,execution_date)
- VALUES
- (in_reserve_pub
- ,in_wire_ref
- ,in_credit
- ,in_exchange_account_name
- ,in_wire_source_h_payto
- ,in_expiration_date)
- ON CONFLICT DO NOTHING;
- IF FOUND
- THEN
- --IF THE INSERTION WAS A SUCCESS IT MEANS NO DUPLICATED TRANSACTION
- out_duplicate = FALSE;
- UPDATE reserves rs
- SET
- current_balance.frac = (rs.current_balance).frac+in_credit.frac
- - CASE
- WHEN (rs.current_balance).frac + in_credit.frac >= 100000000
- THEN 100000000
- ELSE 0
- END
- ,current_balance.val = (rs.current_balance).val+in_credit.val
- + CASE
- WHEN (rs.current_balance).frac + in_credit.frac >= 100000000
- THEN 1
- ELSE 0
- END
- ,expiration_date=GREATEST(expiration_date,in_expiration_date)
- ,gc_date=GREATEST(gc_date,in_expiration_date)
- WHERE reserve_pub=in_reserve_pub;
- EXECUTE FORMAT (
- 'NOTIFY %s'
- ,in_notify);
- ELSE
- out_duplicate = TRUE;
- END IF;
- RETURN;
-END $$;
diff --git a/src/exchangedb/insert_kycauth_in.c b/src/exchangedb/insert_kycauth_in.c
@@ -1,78 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2024 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU 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 General Public License for more details.
-
- You should have received a copy of the GNU General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
- */
-/**
- * @file exchangedb/insert_kycauth_in.c
- * @brief Implementation of the insert_kycauth_in function for Postgres
- * @author Christian Grothoff
- */
-#include "taler/taler_pq_lib.h"
-#include "exchange-database/insert_kycauth_in.h"
-#include "helper.h"
-
-
-enum GNUNET_DB_QueryStatus
-TALER_EXCHANGEDB_insert_kycauth_in (
- struct TALER_EXCHANGEDB_PostgresContext *pg,
- const union TALER_AccountPublicKeyP *account_pub,
- const struct TALER_Amount *credit_amount,
- struct GNUNET_TIME_Timestamp execution_date,
- const struct TALER_FullPayto debit_account_uri,
- const char *section_name,
- uint64_t serial_id)
-{
- struct TALER_NormalizedPaytoHashP h_normalized_payto;
- struct TALER_FullPaytoHashP h_full_payto;
-
- TALER_full_payto_hash (debit_account_uri,
- &h_full_payto);
- TALER_full_payto_normalize_and_hash (debit_account_uri,
- &h_normalized_payto);
- {
- struct TALER_EXCHANGEDB_KycCompletedEventP rep = {
- .header.size = htons (sizeof (rep)),
- .header.type = htons (TALER_DBEVENT_EXCHANGE_KYC_COMPLETED),
- .h_payto = h_normalized_payto
- };
- char *notify_s
- = GNUNET_PQ_get_event_notify_channel (&rep.header);
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_auto_from_type (account_pub),
- GNUNET_PQ_query_param_uint64 (&serial_id),
- TALER_PQ_query_param_amount (pg->conn,
- credit_amount),
- GNUNET_PQ_query_param_auto_from_type (&h_full_payto),
- GNUNET_PQ_query_param_auto_from_type (&h_normalized_payto),
- GNUNET_PQ_query_param_string (debit_account_uri.full_payto),
- GNUNET_PQ_query_param_string (section_name),
- GNUNET_PQ_query_param_timestamp (&execution_date),
- GNUNET_PQ_query_param_string (notify_s),
- GNUNET_PQ_query_param_end
- };
- enum GNUNET_DB_QueryStatus qs;
-
- PREPARE (pg,
- "insert_kycauth_in",
- "CALL"
- " exchange_do_kycauth_in_insert"
- " ($1,$2,$3,$4,$5,$6,$7,$8,$9);");
- qs = GNUNET_PQ_eval_prepared_non_select (
- pg->conn,
- "insert_kycauth_in",
- params);
- GNUNET_free (notify_s);
- return qs;
- }
-}
diff --git a/src/exchangedb/insert_kycauth_in.sql b/src/exchangedb/insert_kycauth_in.sql
@@ -1,92 +0,0 @@
---
--- This file is part of TALER
--- Copyright (C) 2024 Taler Systems SA
---
--- TALER is free software; you can redistribute it and/or modify it under the
--- terms of the GNU 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 General Public License for more details.
---
--- You should have received a copy of the GNU General Public License along with
--- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
---
-
-
-DROP PROCEDURE IF EXISTS exchange_do_kycauth_in_insert;
-CREATE PROCEDURE exchange_do_kycauth_in_insert(
- IN in_account_pub BYTEA,
- IN in_wire_reference INT8,
- IN in_credit taler_amount,
- IN in_wire_source_h_payto BYTEA,
- IN in_h_normalized_payto BYTEA,
- IN in_payto_uri TEXT,
- IN in_exchange_account_name TEXT,
- IN in_execution_date INT8,
- IN in_notify_s TEXT)
-LANGUAGE plpgsql
-AS $$
-DECLARE
- my_is_wallet BOOL;
-BEGIN
-
- INSERT INTO kycauths_in
- (account_pub
- ,wire_reference
- ,credit
- ,wire_source_h_payto
- ,exchange_account_section
- ,execution_date
- ) VALUES (
- in_account_pub
- ,in_wire_reference
- ,in_credit
- ,in_wire_source_h_payto
- ,in_exchange_account_name
- ,in_execution_date
- )
- ON CONFLICT DO NOTHING;
-
- IF NOT FOUND
- THEN
- -- presumably already done
- RETURN;
- END IF;
-
- UPDATE kyc_targets
- SET target_pub=in_account_pub
- WHERE h_normalized_payto=in_h_normalized_payto;
-
- IF NOT FOUND
- THEN
- -- First time we see this account, setup everything.
- my_is_wallet
- = (LOWER (SUBSTRING (in_payto_uri, 0, 23)) =
- 'payto://taler-reserve/') OR
- (LOWER (SUBSTRING (in_payto_uri, 0, 28)) =
- 'payto://taler-reserve-http/');
- INSERT INTO kyc_targets
- (h_normalized_payto
- ,is_wallet
- ,target_pub
- ) VALUES (
- in_h_normalized_payto
- ,my_is_wallet
- ,in_account_pub);
- INSERT INTO wire_targets
- (wire_target_h_payto
- ,h_normalized_payto
- ,payto_uri
- ) VALUES (
- in_wire_source_h_payto
- ,in_h_normalized_payto
- ,in_payto_uri);
- END IF;
-
- EXECUTE FORMAT (
- 'NOTIFY %s'
- ,in_notify_s);
-
-END $$;
diff --git a/src/exchangedb/insert_wad_in.c b/src/exchangedb/insert_wad_in.c
@@ -1,58 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2024 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU 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 General Public License for more details.
-
- You should have received a copy of the GNU General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
- */
-/**
- * @file exchangedb/insert_wad_in.c
- * @brief Implementation of the insert_wad_in function for Postgres
- * @author Christian Grothoff
- */
-#include "taler/taler_pq_lib.h"
-#include "exchange-database/insert_wad_in.h"
-#include "helper.h"
-
-enum GNUNET_DB_QueryStatus
-TALER_EXCHANGEDB_insert_wad_in (
- struct TALER_EXCHANGEDB_PostgresContext *pg,
- const struct TALER_WadIdentifierP *wad_id,
- const char *origin_exchange_url,
- const struct TALER_Amount *amount,
- struct GNUNET_TIME_Timestamp execution_date,
- const struct TALER_FullPayto debit_account_uri,
- const char *section_name,
- uint64_t serial_id)
-{
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_auto_from_type (wad_id),
- GNUNET_PQ_query_param_string (origin_exchange_url),
- TALER_PQ_query_param_amount (pg->conn,
- amount),
- GNUNET_PQ_query_param_timestamp (&execution_date),
- GNUNET_PQ_query_param_end
- };
-
- // FIXME-#7271: should we keep the account data + serial_id?
- PREPARE (pg,
- "insert_wad_in",
- "INSERT INTO wads_in "
- "(wad_id"
- ",origin_exchange_url"
- ",amount"
- ",arrival_time"
- ") VALUES "
- "($1, $2, $3, $4);");
- return GNUNET_PQ_eval_prepared_non_select (pg->conn,
- "insert_wad_in",
- params);
-}
diff --git a/src/exchangedb/meson.build b/src/exchangedb/meson.build
@@ -58,7 +58,7 @@ libtalerexchangedb = library(
'pg.c',
'update_to_aml_unlocked.c',
'commit.c',
- 'complete_shard.c',
+ 'update_shard_progress.c',
'compute_shard.c',
'get_count_known_coins.c',
'insert_aggregation_deferral.c',
@@ -150,7 +150,6 @@ libtalerexchangedb = library(
'iterate_denominations.c',
'iterate_kyc_references.c',
'iterate_reserve_close_info.c',
- 'insert_kycauth_in.c',
'get_kyc_provider_account.c',
'get_active_legitimization.c',
'get_aml_file_number.c',
@@ -187,7 +186,7 @@ libtalerexchangedb = library(
'release_revolving_shard.c',
'get_reserve.c',
'get_reserve_origin.c',
- 'do_insert_reserve_in.c',
+ 'do_import_credits.c',
'update_reserve.c',
'rollback.c',
'iterate_account_merges_above_serial_id.c',
@@ -257,7 +256,6 @@ libtalerexchangedb = library(
'update_auditor.c',
'update_legitimization_process_by_row.c',
'update_wire.c',
- 'insert_wad_in.c',
'iterate_prewires.c',
'insert_prewire.c',
'update_to_prewire_failed.c',
diff --git a/src/exchangedb/sql-schema/0013-preamble.sql b/src/exchangedb/sql-schema/0013-preamble.sql
@@ -0,0 +1,20 @@
+--
+-- This file is part of TALER
+-- Copyright (C) 2026 Taler Systems SA
+--
+-- TALER is free software; you can redistribute it and/or modify it under the
+-- terms of the GNU 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 General Public License for more details.
+--
+-- You should have received a copy of the GNU General Public License along with
+-- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+--
+
+BEGIN;
+
+SELECT _v.register_patch('exchange-0013', NULL, NULL);
+SET search_path TO exchange;
diff --git a/src/exchangedb/sql-schema/0013-work_shards.sql b/src/exchangedb/sql-schema/0013-work_shards.sql
@@ -0,0 +1,33 @@
+--
+-- This file is part of TALER
+-- Copyright (C) 2026 Taler Systems SA
+--
+-- TALER is free software; you can redistribute it and/or modify it under the
+-- terms of the GNU 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 General Public License for more details.
+--
+-- You should have received a copy of the GNU General Public License along with
+-- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+--
+
+-- Track how far a worker has come *within* a shard, so that the work it
+-- committed can be recorded long before the shard as a whole is finished, and
+-- so that a worker picking the shard up later resumes instead of restarting.
+ALTER TABLE work_shards
+ ADD COLUMN progress_row INT8;
+
+-- Existing rows have no recorded progress. Starting them over is always safe
+-- (every job using this table is idempotent, it just costs extra work), so
+-- point them at the beginning of their shard.
+UPDATE work_shards
+ SET progress_row=start_row;
+
+ALTER TABLE work_shards
+ ALTER COLUMN progress_row SET NOT NULL;
+
+COMMENT ON COLUMN work_shards.progress_row
+ IS 'rows in [start_row,progress_row) of this shard have been processed and committed; equal to start_row while nothing has been done yet and to end_row once the shard is complete. Only ever moves forward, so a second worker that concurrently grabbed the same shard cannot rewind it.';
diff --git a/src/exchangedb/sql-schema/meson.build b/src/exchangedb/sql-schema/meson.build
@@ -32,10 +32,10 @@ procedures_sql = [
'../insert_aml_decision.sql',
'../insert_successor_measure.sql',
'../insert_aml_officer.sql',
- '../do_insert_reserve_in.sql',
+ '../begin_shard.sql',
+ '../do_import_credits.sql',
# semi-dead...
'exchange_do_get_link_data.sql',
- '../insert_kycauth_in.sql',
'../do_trigger_kyc_rule_for_account.sql',
'../get_legitimization_requirement_by_row.sql',
'../insert_active_legitimization_measure.sql',
@@ -167,6 +167,8 @@ exchange_0012_sql = [
'commit.sql',
]
+exchange_0013_sql = ['0013-preamble.sql', '0013-work_shards.sql', 'commit.sql']
+
generated_sql = [
['auditor-triggers-0001.sql', ['auditor-triggers-0001.sql']],
['auditor-triggers-0002.sql', ['auditor-triggers-0002.sql']],
@@ -186,6 +188,7 @@ generated_sql = [
['exchange-0010.sql', exchange_0010_sql],
['exchange-0011.sql', exchange_0011_sql],
['exchange-0012.sql', exchange_0012_sql],
+ ['exchange-0013.sql', exchange_0013_sql],
['tops-0001.sql', ['tops-0001.sql']],
]
diff --git a/src/exchangedb/test_regressions.c b/src/exchangedb/test_regressions.c
@@ -34,6 +34,10 @@
#include "exchange-database/do_reserve_open.h"
#include "exchange-database/get_purse_deposit.h"
#include "exchange-database/get_reserve_close_info.h"
+#include "exchange-database/begin_shard.h"
+#include "exchange-database/abort_shard.h"
+#include "exchange-database/update_shard_progress.h"
+#include "exchange-database/do_import_credits.h"
/**
@@ -610,6 +614,237 @@ check_reserve_close_info_without_origin (void)
/**
+ * Read back the bookkeeping of one work shard.
+ *
+ * @param job_name job the shard belongs to
+ * @param start_row inclusive start row of the shard
+ * @param end_row exclusive end row of the shard
+ * @param[out] progress_row how far the shard has come
+ * @param[out] completed whether the shard is done
+ * @return transaction status code
+ */
+static enum GNUNET_DB_QueryStatus
+get_shard_state (const char *job_name,
+ uint64_t start_row,
+ uint64_t end_row,
+ uint64_t *progress_row,
+ bool *completed)
+{
+ struct GNUNET_PQ_QueryParam params[] = {
+ GNUNET_PQ_query_param_string (job_name),
+ GNUNET_PQ_query_param_uint64 (&start_row),
+ GNUNET_PQ_query_param_uint64 (&end_row),
+ GNUNET_PQ_query_param_end
+ };
+ struct GNUNET_PQ_ResultSpec rs[] = {
+ GNUNET_PQ_result_spec_uint64 ("progress_row",
+ progress_row),
+ GNUNET_PQ_result_spec_bool ("completed",
+ completed),
+ GNUNET_PQ_result_spec_end
+ };
+
+ PREPARE (pg,
+ "test_get_shard_state",
+ "SELECT"
+ " progress_row"
+ ",completed"
+ " FROM work_shards"
+ " WHERE job_name=$1"
+ " AND start_row=$2"
+ " AND end_row=$3;");
+ return GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
+ "test_get_shard_state",
+ params,
+ rs);
+}
+
+
+/**
+ * A worker that gets part of the way through a shard and then stops used to
+ * leave nothing behind: the shard was either untouched or completed, so the
+ * next worker to pick it up redid all of it, and nothing it had imported was
+ * visible until the whole shard was done. Check that the progress marker is
+ * kept, that it survives releasing the shard, that it never moves backwards,
+ * and that reaching the end of the shard is what completes it.
+ */
+static int
+check_shard_progress_survives_abort (void)
+{
+ const char *job = "test-shard-progress";
+ uint64_t start;
+ uint64_t end;
+ uint64_t progress;
+ uint64_t start2;
+ uint64_t end2;
+ uint64_t progress2;
+ bool completed;
+
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ TALER_EXCHANGEDB_begin_shard (pg,
+ job,
+ GNUNET_TIME_UNIT_HOURS,
+ 1024,
+ &start,
+ &end,
+ &progress));
+ /* A fresh shard has nothing done yet. */
+ FAILIF (progress != start);
+
+ /* Get half way, then let go of the shard. */
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ TALER_EXCHANGEDB_update_shard_progress (pg,
+ job,
+ start,
+ end,
+ start + 512,
+ GNUNET_TIME_UNIT_HOURS));
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ TALER_EXCHANGEDB_abort_shard (pg,
+ job,
+ start,
+ end));
+
+ /* The next worker gets the same shard back, but resumes in the middle. */
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ TALER_EXCHANGEDB_begin_shard (pg,
+ job,
+ GNUNET_TIME_UNIT_HOURS,
+ 1024,
+ &start2,
+ &end2,
+ &progress2));
+ FAILIF (start2 != start);
+ FAILIF (end2 != end);
+ FAILIF (progress2 != start + 512);
+
+ /* A straggler reporting older progress must not rewind the shard. */
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ TALER_EXCHANGEDB_update_shard_progress (pg,
+ job,
+ start,
+ end,
+ start + 1,
+ GNUNET_TIME_UNIT_HOURS));
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ get_shard_state (job,
+ start,
+ end,
+ &progress,
+ &completed));
+ FAILIF (progress != start + 512);
+ FAILIF (completed);
+
+ /* Reaching the end completes the shard; there is no second statement for
+ the caller to forget, or to crash before. */
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ TALER_EXCHANGEDB_update_shard_progress (pg,
+ job,
+ start,
+ end,
+ end,
+ GNUNET_TIME_UNIT_HOURS));
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ get_shard_state (job,
+ start,
+ end,
+ &progress,
+ &completed));
+ FAILIF (progress != end);
+ FAILIF (! completed);
+ return 0;
+}
+
+
+/**
+ * Importing a batch of incoming wire transfers and recording how far the
+ * shard has come is one statement, so a crash cannot land one without the
+ * other. Check that a batch moves both, that importing it again is harmless
+ * and reported as a duplicate, and that the shard completes as part of the
+ * import rather than in a transaction of its own.
+ */
+static int
+check_import_credits_advances_shard (void)
+{
+ const char *job = "test-import-credits";
+ struct TALER_ReservePublicKeyP reserve_pub;
+ struct TALER_Amount balance;
+ struct TALER_EXCHANGEDB_ReserveInInfo reserve;
+ struct TALER_EXCHANGEDB_CreditBatch batch;
+ enum GNUNET_DB_QueryStatus results[1];
+ uint64_t start;
+ uint64_t end;
+ uint64_t progress;
+ bool completed;
+
+ memset (&reserve_pub,
+ 0x51,
+ sizeof (reserve_pub));
+ parse_amount ("4.00",
+ &balance);
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ TALER_EXCHANGEDB_begin_shard (pg,
+ job,
+ GNUNET_TIME_UNIT_HOURS,
+ 1024,
+ &start,
+ &end,
+ &progress));
+ reserve.reserve_pub = &reserve_pub;
+ reserve.balance = &balance;
+ reserve.execution_time = GNUNET_TIME_timestamp_get ();
+ reserve.sender_account_details.full_payto
+ = (char *) "payto://x-taler-bank/localhost/shard-test?receiver-name=Shard";
+ reserve.wire_reference = start + 1;
+ memset (&batch,
+ 0,
+ sizeof (batch));
+ batch.exchange_account_name = "exchange-account-test";
+ batch.reserves = &reserve;
+ batch.reserves_length = 1;
+ batch.job_name = job;
+ batch.shard_start = start;
+ batch.shard_end = end;
+ batch.progress_row = start + 1;
+ batch.lease = GNUNET_TIME_UNIT_HOURS;
+
+ FAILIF (0 >
+ TALER_EXCHANGEDB_do_import_credits (pg,
+ &batch,
+ results));
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT != results[0]);
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ get_shard_state (job,
+ start,
+ end,
+ &progress,
+ &completed));
+ /* The transfer is committed and so is the fact that we consumed its row --
+ long before the rest of the shard has been looked at. */
+ FAILIF (progress != start + 1);
+ FAILIF (completed);
+
+ /* Re-importing the same batch is what happens whenever a shard is picked up
+ twice. It has to be harmless. */
+ batch.progress_row = end;
+ FAILIF (0 >
+ TALER_EXCHANGEDB_do_import_credits (pg,
+ &batch,
+ results));
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != results[0]);
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ get_shard_state (job,
+ start,
+ end,
+ &progress,
+ &completed));
+ FAILIF (progress != end);
+ FAILIF (! completed);
+ return 0;
+}
+
+
+/**
* All checks we know about.
*/
static const struct
@@ -629,6 +864,10 @@ static const struct
&check_purse_deposit_without_age_commitment },
{ "reserve-close-info-without-origin",
&check_reserve_close_info_without_origin },
+ { "shard-progress-survives-abort",
+ &check_shard_progress_survives_abort },
+ { "import-credits-advances-shard",
+ &check_import_credits_advances_shard },
{ NULL, NULL }
};
diff --git a/src/exchangedb/update_shard_progress.c b/src/exchangedb/update_shard_progress.c
@@ -0,0 +1,69 @@
+/*
+ This file is part of TALER
+ Copyright (C) 2022, 2026 Taler Systems SA
+
+ TALER is free software; you can redistribute it and/or modify it under the
+ terms of the GNU 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 General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file exchangedb/update_shard_progress.c
+ * @brief Implementation of the update_shard_progress function for Postgres
+ * @author Christian Grothoff
+ */
+#include "taler/taler_pq_lib.h"
+#include "exchange-database/update_shard_progress.h"
+#include "helper.h"
+
+
+enum GNUNET_DB_QueryStatus
+TALER_EXCHANGEDB_update_shard_progress (
+ struct TALER_EXCHANGEDB_PostgresContext *pg,
+ const char *job_name,
+ uint64_t start_row,
+ uint64_t end_row,
+ uint64_t progress_row,
+ struct GNUNET_TIME_Relative lease)
+{
+ struct GNUNET_TIME_Absolute lease_until
+ = GNUNET_TIME_relative_to_absolute (lease);
+ struct GNUNET_PQ_QueryParam params[] = {
+ GNUNET_PQ_query_param_string (job_name),
+ GNUNET_PQ_query_param_uint64 (&start_row),
+ GNUNET_PQ_query_param_uint64 (&end_row),
+ GNUNET_PQ_query_param_uint64 (&progress_row),
+ GNUNET_PQ_query_param_absolute_time (&lease_until),
+ GNUNET_PQ_query_param_end
+ };
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Shard %s (%llu,%llu] progressed to %llu\n",
+ job_name,
+ (unsigned long long) start_row,
+ (unsigned long long) end_row,
+ (unsigned long long) progress_row);
+ /* GREATEST() keeps this monotonic: a worker that lost the shard to a second
+ one and only notices later must not push the marker back. Both the
+ 'completed' flag and the lease renewal ride along on this one statement,
+ so a caller never needs a second round-trip (let alone a second
+ transaction) to say "and that was the last of it". */
+ PREPARE (pg,
+ "update_shard_progress",
+ "UPDATE work_shards"
+ " SET progress_row=GREATEST(progress_row,$4)"
+ " ,completed=(GREATEST(progress_row,$4) >= end_row)"
+ " ,last_attempt=$5"
+ " WHERE job_name=$1"
+ " AND start_row=$2"
+ " AND end_row=$3;");
+ return GNUNET_PQ_eval_prepared_non_select (pg->conn,
+ "update_shard_progress",
+ params);
+}
diff --git a/src/include/exchange-database/begin_shard.h b/src/include/exchange-database/begin_shard.h
@@ -24,15 +24,20 @@
#include "exchangedb_lib.h"
/**
- * Function called to grab a work shard on an operation @a op. Runs in its
- * own transaction.
+ * Function called to grab a work shard on an operation @a op. Takes over an
+ * abandoned shard if there is one, and opens a fresh one otherwise. Runs as
+ * a single statement, so it needs no transaction of its own.
*
* @param pg the database context
* @param job_name name of the operation to grab a word shard for
- * @param delay minimum age of a shard to grab
+ * @param delay how long we intend to hold the shard; also the minimum age of
+ * a shard held by someone else before we may take it over
* @param shard_size desired shard size
* @param[out] start_row inclusive start row of the shard (returned)
* @param[out] end_row exclusive end row of the shard (returned)
+ * @param[out] progress_row row to resume at; equal to @a start_row for a
+ * fresh shard, and further along for a shard taken over from a
+ * worker that got some of it done (returned)
* @return transaction status code
*/
enum GNUNET_DB_QueryStatus
@@ -41,6 +46,7 @@ TALER_EXCHANGEDB_begin_shard (struct TALER_EXCHANGEDB_PostgresContext *pg,
struct GNUNET_TIME_Relative delay,
uint64_t shard_size,
uint64_t *start_row,
- uint64_t *end_row);
+ uint64_t *end_row,
+ uint64_t *progress_row);
#endif
diff --git a/src/include/exchange-database/complete_shard.h b/src/include/exchange-database/complete_shard.h
@@ -1,42 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2022 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU 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 General Public License for more details.
-
- You should have received a copy of the GNU General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
- */
-/**
- * @file src/include/exchange-database/complete_shard.h
- * @brief implementation of the complete_shard function for Postgres
- * @author Christian Grothoff
- */
-#ifndef EXCHANGE_DATABASE_COMPLETE_SHARD_H
-#define EXCHANGE_DATABASE_COMPLETE_SHARD_H
-
-#include "exchangedb_lib.h"
-
-
-/**
- * Function called to persist that work on a shard was completed.
- *
- * @param pg the database context
- * @param job_name name of the operation to grab a word shard for
- * @param start_row inclusive start row of the shard
- * @param end_row exclusive end row of the shard
- * @return transaction status code
- */
-enum GNUNET_DB_QueryStatus
-TALER_EXCHANGEDB_complete_shard (struct TALER_EXCHANGEDB_PostgresContext *pg,
- const char *job_name,
- uint64_t start_row,
- uint64_t end_row);
-
-#endif
diff --git a/src/include/exchange-database/do_expire_purse.h b/src/include/exchange-database/do_expire_purse.h
@@ -30,11 +30,15 @@
* @param pg the database context
* @param start_time select purse expired after this time
* @param end_time select purse expired before this time
+ * @param[out] expiration expiration time of the purse that was cleaned
+ * up, so that the caller can record how far it has come; set to
+ * @a end_time if there was no purse left to expire
* @return transaction status code (#GNUNET_DB_STATUS_SUCCESS_NO_RESULTS if no purse expired in the given time interval).
*/
enum GNUNET_DB_QueryStatus
TALER_EXCHANGEDB_do_expire_purse (struct TALER_EXCHANGEDB_PostgresContext *pg,
struct GNUNET_TIME_Absolute start_time,
- struct GNUNET_TIME_Absolute end_time);
+ struct GNUNET_TIME_Absolute end_time,
+ struct GNUNET_TIME_Absolute *expiration);
#endif
diff --git a/src/include/exchange-database/do_import_credits.h b/src/include/exchange-database/do_import_credits.h
@@ -0,0 +1,161 @@
+/*
+ This file is part of TALER
+ Copyright (C) 2022--2026 Taler Systems SA
+
+ TALER is free software; you can redistribute it and/or modify it under the
+ terms of the GNU 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 General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file src/include/exchange-database/do_import_credits.h
+ * @brief implementation of the do_import_credits function for Postgres
+ * @author Christian Grothoff
+ */
+#ifndef EXCHANGE_DATABASE_DO_IMPORT_CREDITS_H
+#define EXCHANGE_DATABASE_DO_IMPORT_CREDITS_H
+
+#include "exchangedb_lib.h"
+
+
+/**
+ * Incoming wire transfer that credits a reserve.
+ */
+struct TALER_EXCHANGEDB_ReserveInInfo
+{
+ const struct TALER_ReservePublicKeyP *reserve_pub;
+ const struct TALER_Amount *balance;
+ struct GNUNET_TIME_Timestamp execution_time;
+ struct TALER_FullPayto sender_account_details;
+ uint64_t wire_reference;
+};
+
+
+/**
+ * Incoming wire transfer that authenticates an account by carrying an
+ * account public key in its subject.
+ */
+struct TALER_EXCHANGEDB_KycauthInInfo
+{
+ const union TALER_AccountPublicKeyP *account_pub;
+ const struct TALER_Amount *balance;
+ struct GNUNET_TIME_Timestamp execution_time;
+ struct TALER_FullPayto sender_account_details;
+ uint64_t wire_reference;
+};
+
+
+/**
+ * Incoming WAD transfer from another exchange.
+ */
+struct TALER_EXCHANGEDB_WadInInfo
+{
+ const struct TALER_WadIdentifierP *wad_id;
+ const char *origin_exchange_url;
+ const struct TALER_Amount *balance;
+ struct GNUNET_TIME_Timestamp execution_time;
+};
+
+
+/**
+ * One batch of incoming wire transfers to import, together with the work
+ * shard they were read under. All transfers in a batch come from the same
+ * bank account, which is why @e exchange_account_name is not per transfer.
+ */
+struct TALER_EXCHANGEDB_CreditBatch
+{
+ /**
+ * Configuration section name of the exchange bank account the transfers
+ * arrived at.
+ */
+ const char *exchange_account_name;
+
+ /**
+ * Transfers into reserves, may be NULL if @e reserves_length is 0.
+ */
+ const struct TALER_EXCHANGEDB_ReserveInInfo *reserves;
+
+ /**
+ * Length of the @e reserves array.
+ */
+ unsigned int reserves_length;
+
+ /**
+ * KYC authentication transfers, may be NULL if @e kycauths_length is 0.
+ */
+ const struct TALER_EXCHANGEDB_KycauthInInfo *kycauths;
+
+ /**
+ * Length of the @e kycauths array.
+ */
+ unsigned int kycauths_length;
+
+ /**
+ * WAD transfers, may be NULL if @e wads_length is 0.
+ */
+ const struct TALER_EXCHANGEDB_WadInInfo *wads;
+
+ /**
+ * Length of the @e wads array.
+ */
+ unsigned int wads_length;
+
+ /**
+ * Name of the job whose shard this batch was read under.
+ */
+ const char *job_name;
+
+ /**
+ * Inclusive start row of that shard.
+ */
+ uint64_t shard_start;
+
+ /**
+ * Exclusive end row of that shard.
+ */
+ uint64_t shard_end;
+
+ /**
+ * Row up to which (exclusive) the shard is done once this batch is in.
+ * Pass @e shard_end to complete the shard.
+ */
+ uint64_t progress_row;
+
+ /**
+ * For how much longer we intend to hold the shard.
+ */
+ struct GNUNET_TIME_Relative lease;
+};
+
+
+/**
+ * Import one batch of incoming wire transfers and advance the work shard it
+ * belongs to.
+ *
+ * This is a single statement, so it needs no transaction of its own: the
+ * transfers and the record of how far the shard has come either both land or
+ * neither does. That is what lets a caller commit as soon as the bank
+ * answers, rather than holding everything open until the shard is finished.
+ *
+ * @param pg the database context
+ * @param batch the transfers to import and the shard they belong to
+ * @param[out] reserve_results set to the query status per entry of
+ * @a batch->reserves, must be of length @a batch->reserves_length;
+ * #GNUNET_DB_STATUS_SUCCESS_NO_RESULTS means the transfer had
+ * already been imported
+ * @return transaction status code; on success, the number of reserve
+ * transfers that were looked at
+ */
+enum GNUNET_DB_QueryStatus
+TALER_EXCHANGEDB_do_import_credits (
+ struct TALER_EXCHANGEDB_PostgresContext *pg,
+ const struct TALER_EXCHANGEDB_CreditBatch *batch,
+ enum GNUNET_DB_QueryStatus *reserve_results);
+
+#endif
diff --git a/src/include/exchange-database/do_insert_reserve_in.h b/src/include/exchange-database/do_insert_reserve_in.h
@@ -1,59 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2022 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU 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 General Public License for more details.
-
- You should have received a copy of the GNU General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
- */
-/**
- * @file src/include/exchange-database/do_insert_reserve_in.h
- * @brief implementation of the do_insert_reserve_in function for Postgres
- * @author Christian Grothoff
- */
-#ifndef EXCHANGE_DATABASE_DO_INSERT_RESERVE_IN_H
-#define EXCHANGE_DATABASE_DO_INSERT_RESERVE_IN_H
-
-#include "exchangedb_lib.h"
-
-
-struct TALER_EXCHANGEDB_ReserveInInfo
-{
- const struct TALER_ReservePublicKeyP *reserve_pub;
- const struct TALER_Amount *balance;
- struct GNUNET_TIME_Timestamp execution_time;
- struct TALER_FullPayto sender_account_details;
- const char *exchange_account_name;
- uint64_t wire_reference;
-};
-
-
-/**
- * Insert an incoming transaction into reserves. New reserves are also
- * created through this function. Runs its own transaction(s).
- *
- * @param pg the database context
- * @param reserves array of reserves to insert
- * @param reserves_length length of the @a reserves array
- * @param[out] results set to query status per reserve, must be of length @a reserves_length
- * @return transaction status code
- */
-enum GNUNET_DB_QueryStatus
-TALER_EXCHANGEDB_do_insert_reserve_in (struct TALER_EXCHANGEDB_PostgresContext *
- pg
- ,
- const struct
- TALER_EXCHANGEDB_ReserveInInfo *
- reserves,
- unsigned int reserves_length,
- enum GNUNET_DB_QueryStatus *results);
-
-
-#endif
diff --git a/src/include/exchange-database/insert_kycauth_in.h b/src/include/exchange-database/insert_kycauth_in.h
@@ -1,54 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2024 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU 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 General Public License for more details.
-
- You should have received a copy of the GNU General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
- */
-/**
- * @file src/include/exchange-database/insert_kycauth_in.h
- * @brief implementation of the insert_kycauth_in function for Postgres
- * @author Christian Grothoff
- */
-#ifndef EXCHANGE_DATABASE_INSERT_KYCAUTH_IN_H
-#define EXCHANGE_DATABASE_INSERT_KYCAUTH_IN_H
-
-#include "taler/taler_util.h"
-#include "taler/taler_json_lib.h"
-#include "exchangedb_lib.h"
-
-
-/**
- * Insert an incoming KCYAUTH wire transfer into
- * the database and update the authentication key
- * for the origin account.
- *
- * @param pg the database context
- * @param account_pub public key of the account
- * @param credit_amount amount we were credited
- * @param execution_date when was the transfer made
- * @param debit_account_uri URI of the debit account
- * @param section_name section of the exchange bank account that received the transfer
- * @param serial_id bank-specific row identifying the transfer
- */
-enum GNUNET_DB_QueryStatus
-TALER_EXCHANGEDB_insert_kycauth_in (struct TALER_EXCHANGEDB_PostgresContext *pg,
- const union TALER_AccountPublicKeyP *
- account_pub,
- const struct TALER_Amount *credit_amount,
- struct GNUNET_TIME_Timestamp execution_date,
- const struct TALER_FullPayto
- debit_account_uri,
- const char *section_name,
- uint64_t serial_id);
-
-
-#endif
diff --git a/src/include/exchange-database/insert_wad_in.h b/src/include/exchange-database/insert_wad_in.h
@@ -1,52 +0,0 @@
-/*
- This file is part of TALER
- Copyright (C) 2024 Taler Systems SA
-
- TALER is free software; you can redistribute it and/or modify it under the
- terms of the GNU 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 General Public License for more details.
-
- You should have received a copy of the GNU General Public License along with
- TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
- */
-/**
- * @file src/include/exchange-database/insert_wad_in.h
- * @brief implementation of the insert_wad_in function for Postgres
- * @author Christian Grothoff
- */
-#ifndef EXCHANGE_DATABASE_INSERT_WAD_IN_H
-#define EXCHANGE_DATABASE_INSERT_WAD_IN_H
-
-#include "taler/taler_util.h"
-#include "taler/taler_json_lib.h"
-#include "exchangedb_lib.h"
-
-
-/**
- * Insert an incoming WAD wire transfer into the database.
- *
- * @param pg the database context
- * @param wad_id WAD identifier
- * @param origin_exchange_url exchange base URL originating the transfer
- * @param amount the amount that was transferred
- * @param execution_date when was the transfer made
- * @param debit_account_uri URI of the debit account
- * @param section_name section of the exchange bank account that received the transfer
- * @param serial_id bank-specific row identifying the transfer
- */
-enum GNUNET_DB_QueryStatus
-TALER_EXCHANGEDB_insert_wad_in (struct TALER_EXCHANGEDB_PostgresContext *pg,
- const struct TALER_WadIdentifierP *wad_id,
- const char *origin_exchange_url,
- const struct TALER_Amount *amount,
- struct GNUNET_TIME_Timestamp execution_date,
- const struct TALER_FullPayto debit_account_uri,
- const char *section_name,
- uint64_t serial_id);
-
-
-#endif
diff --git a/src/include/exchange-database/update_shard_progress.h b/src/include/exchange-database/update_shard_progress.h
@@ -0,0 +1,60 @@
+/*
+ This file is part of TALER
+ Copyright (C) 2022, 2026 Taler Systems SA
+
+ TALER is free software; you can redistribute it and/or modify it under the
+ terms of the GNU 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 General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along with
+ TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
+ */
+/**
+ * @file src/include/exchange-database/update_shard_progress.h
+ * @brief implementation of the update_shard_progress function for Postgres
+ * @author Christian Grothoff
+ */
+#ifndef EXCHANGE_DATABASE_UPDATE_SHARD_PROGRESS_H
+#define EXCHANGE_DATABASE_UPDATE_SHARD_PROGRESS_H
+
+#include "exchangedb_lib.h"
+
+
+/**
+ * Record how far a worker has come within a shard, and renew its lease on
+ * that shard. The shard is marked completed if @a progress_row reached
+ * @a end_row.
+ *
+ * Call this from within the same transaction that persists the work being
+ * reported. That is the point of the function: the progress marker and the
+ * work it describes have to become visible together, or a crash in between
+ * leaves the two disagreeing.
+ *
+ * Progress only ever moves forward. A second worker that grabbed the same
+ * shard (because our lease ran out while we were still going) therefore
+ * cannot rewind what we already recorded.
+ *
+ * @param pg the database context
+ * @param job_name name of the job the shard belongs to
+ * @param start_row inclusive start row of the shard
+ * @param end_row exclusive end row of the shard
+ * @param progress_row row up to which (exclusive) the work is done and
+ * committed; pass @a end_row to complete the shard
+ * @param lease for how much longer we intend to hold the shard
+ * @return transaction status code, #GNUNET_DB_STATUS_SUCCESS_NO_RESULTS if
+ * the shard is not in the table (someone else finished it)
+ */
+enum GNUNET_DB_QueryStatus
+TALER_EXCHANGEDB_update_shard_progress (
+ struct TALER_EXCHANGEDB_PostgresContext *pg,
+ const char *job_name,
+ uint64_t start_row,
+ uint64_t end_row,
+ uint64_t progress_row,
+ struct GNUNET_TIME_Relative lease);
+
+#endif
