commit 36929158c3b344c4f807401436a9424f686a8855
parent 67e4c8b94aa7198cc6a98eb4cf67ef0d02222b20
Author: Christian Grothoff <christian@grothoff.org>
Date: Sat, 10 Oct 2026 23:49:43 +0200
fix issue where taler-exchange-transfer would miss outgoing transfers if the serial numbers were out-of-order; patch auto-fixes issue on running exchanges once they upgrade; only plausible if concurrent aggregators were used, so should not have yet mattered in practice
Diffstat:
5 files changed, 532 insertions(+), 35 deletions(-)
diff --git a/src/exchange/taler-exchange-transfer.c b/src/exchange/taler-exchange-transfer.c
@@ -35,6 +35,7 @@
#include "exchange-database/update_to_prewire_finished.h"
#include "exchange-database/update_to_prewire_failed.h"
#include "exchange-database/iterate_prewires.h"
+#include "exchange-database/iterate_prewires_below_open_shards.h"
#include "exchange-database/event_listen.h"
#include "exchange-database/event_listen_cancel.h"
#include "exchange-database/update_to_prewire_finished.h"
@@ -176,6 +177,12 @@ static struct WirePrepareData *wpd_tail;
static struct Shard *shard;
/**
+ * True while we execute transfers that #sweep_transfers() found below the
+ * open shards, rather than transfers of #shard.
+ */
+static bool sweeping;
+
+/**
* Handle to the context for interacting with the bank / wire gateway.
*/
static struct GNUNET_CURL_Context *ctx;
@@ -359,6 +366,15 @@ static void
run_transfers (void *cls);
+/**
+ * Execute the wire transfers that were left behind in completed shards.
+ *
+ * @param cls NULL
+ */
+static void
+sweep_transfers (void *cls);
+
+
static void
run_transfers_delayed (void *cls)
{
@@ -385,6 +401,8 @@ static void
batch_done (void)
{
enum GNUNET_DB_QueryStatus qs;
+ GNUNET_SCHEDULER_TaskCallback next
+ = sweeping ? &sweep_transfers : &run_transfers;
/* batch done */
GNUNET_assert (NULL == wpd_head);
@@ -392,13 +410,17 @@ batch_done (void)
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);
+ "shard done" write any more. A sweep works below the open shards
+ and has no progress to record. */
+ if (sweeping)
+ qs = GNUNET_DB_STATUS_SUCCESS_NO_RESULTS;
+ else
+ 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);
@@ -423,7 +445,7 @@ batch_done (void)
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Serialization failure, trying again immediately!\n");
GNUNET_assert (NULL == task);
- task = GNUNET_SCHEDULER_add_now (&run_transfers,
+ task = GNUNET_SCHEDULER_add_now (next,
NULL);
return;
case GNUNET_DB_STATUS_HARD_ERROR:
@@ -432,13 +454,14 @@ batch_done (void)
GNUNET_SCHEDULER_shutdown ();
return;
case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- shard->batch_start = shard->batch_end + 1;
+ if (! sweeping)
+ shard->batch_start = shard->batch_end + 1;
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Batch complete\n");
- /* continue with #run_transfers(), just to guard
- against the unlikely case that there are more. */
+ /* continue with #run_transfers() or #sweep_transfers(),
+ just to guard against the unlikely case that there are more. */
GNUNET_assert (NULL == task);
- task = GNUNET_SCHEDULER_add_now (&run_transfers,
+ task = GNUNET_SCHEDULER_add_now (next,
NULL);
return;
default:
@@ -531,8 +554,9 @@ wire_confirm_cb (void *cls,
GNUNET_SCHEDULER_shutdown ();
return;
}
- shard->batch_end = GNUNET_MAX (wpd->row_id,
- shard->batch_end);
+ if (! sweeping)
+ shard->batch_end = GNUNET_MAX (wpd->row_id,
+ shard->batch_end);
switch (qs)
{
case GNUNET_DB_STATUS_SOFT_ERROR:
@@ -541,7 +565,9 @@ wire_confirm_cb (void *cls,
GNUNET_assert (NULL == task);
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Serialization failure, trying again immediately!\n");
- task = GNUNET_SCHEDULER_add_now (&run_transfers,
+ task = GNUNET_SCHEDULER_add_now (sweeping
+ ? &sweep_transfers
+ : &run_transfers,
NULL);
return;
case GNUNET_DB_STATUS_HARD_ERROR:
@@ -587,7 +613,8 @@ wire_prepare_cb (void *cls,
if ( (NULL != task) ||
(EXIT_SUCCESS != global_ret) )
return; /* current transaction was aborted */
- if (rowid >= shard->shard_end)
+ if ( (! sweeping) &&
+ (rowid >= shard->shard_end) )
{
/* skip */
shard->batch_end = shard->shard_end - 1;
@@ -596,6 +623,16 @@ wire_prepare_cb (void *cls,
batch_done ();
return;
}
+ /* prewire_uuid is assigned at INSERT time, but the inserting transactions
+ can commit in a different order. Such a late row is only visible now. */
+ if (sweeping)
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Wire transfer %llu committed after its shard was completed, executing it now\n",
+ (unsigned long long) rowid);
+ else if (rowid < shard->batch_start)
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Wire transfer %llu committed after the shard moved past it, executing it now\n",
+ (unsigned long long) rowid);
if ( (NULL == wire_method) ||
(NULL == buf) )
{
@@ -662,12 +699,14 @@ run_transfers (void *cls)
(void) cls;
task = NULL;
+ sweeping = false;
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. */
+ to write here. Rows of the shard that commit only now are left to
+ #sweep_transfers(), which we run before taking the next shard. */
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Shard [%llu,%llu) completed\n",
(unsigned long long) shard->shard_start,
@@ -676,7 +715,7 @@ run_transfers (void *cls)
shard->shard_start_time);
GNUNET_free (shard);
GNUNET_assert (NULL == task);
- task = GNUNET_SCHEDULER_add_now (&select_shard,
+ task = GNUNET_SCHEDULER_add_now (&sweep_transfers,
NULL);
return;
}
@@ -687,7 +726,7 @@ run_transfers (void *cls)
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Checking for %lld pending wire transfers [%llu-...)\n",
(long long) limit,
- (unsigned long long) shard->batch_start);
+ (unsigned long long) shard->shard_start);
if (GNUNET_OK !=
TALER_EXCHANGEDB_start_read_committed (pg,
"aggregator run transfer"))
@@ -699,8 +738,12 @@ run_transfers (void *cls)
return;
}
GNUNET_assert (NULL == task);
+ /* Start at the beginning of the shard, not at batch_start: a row below
+ batch_start may have committed after we passed it. Only unfinished
+ rows are returned (from an index of just those), so this costs
+ nothing when there is none. */
qs = TALER_EXCHANGEDB_iterate_prewires (pg,
- shard->batch_start,
+ shard->shard_start,
limit,
&wire_prepare_cb,
NULL);
@@ -724,25 +767,99 @@ run_transfers (void *cls)
NULL);
return;
case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
- /* no more prepared wire transfers, go sleep a bit! */
+ /* no more prepared wire transfers in our shard; check below it,
+ then go sleep a bit! */
TALER_EXCHANGEDB_rollback (pg);
GNUNET_assert (NULL == wpd_head);
GNUNET_assert (NULL == task);
- if (GNUNET_YES == test_mode)
- {
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "No more pending wire transfers, shutting down (because we are in test mode)\n");
- GNUNET_SCHEDULER_shutdown ();
- }
- else
+ task = GNUNET_SCHEDULER_add_now (&sweep_transfers,
+ NULL);
+ return;
+ default:
+ /* continued in wire_prepare_cb() */
+ return;
+ }
+}
+
+
+/**
+ * There is nothing to do in our shard or below it. Sleep a bit, or, in
+ * test mode, shut down.
+ */
+static void
+go_idle (void)
+{
+ GNUNET_assert (NULL == task);
+ if (GNUNET_YES == test_mode)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "No more pending wire transfers, shutting down (because we are in test mode)\n");
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "No more pending wire transfers, going idle\n");
+ task = GNUNET_SCHEDULER_add_delayed (transfer_idle_sleep_interval,
+ &run_transfers_delayed,
+ NULL);
+}
+
+
+static void
+sweep_transfers (void *cls)
+{
+ enum GNUNET_DB_QueryStatus qs;
+
+ (void) cls;
+ task = NULL;
+ GNUNET_assert (NULL == wpd_head);
+ sweeping = true;
+ if (GNUNET_OK !=
+ TALER_EXCHANGEDB_start_read_committed (pg,
+ "transfer sweep"))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "Failed to start database transaction!\n");
+ global_ret = EXIT_FAILURE;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ }
+ /* Same cap on parallel requests to the bank as in #run_transfers(). */
+ qs = TALER_EXCHANGEDB_iterate_prewires_below_open_shards (pg,
+ "transfer",
+ 256,
+ &wire_prepare_cb,
+ NULL);
+ switch (qs)
+ {
+ case GNUNET_DB_STATUS_HARD_ERROR:
+ cleanup_wpd ();
+ TALER_EXCHANGEDB_rollback (pg);
+ GNUNET_break (0);
+ global_ret = EXIT_FAILURE;
+ GNUNET_SCHEDULER_shutdown ();
+ return;
+ case GNUNET_DB_STATUS_SOFT_ERROR:
+ TALER_EXCHANGEDB_rollback (pg);
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ "Serialization failure, trying again immediately!\n");
+ cleanup_wpd ();
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&sweep_transfers,
+ NULL);
+ return;
+ case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+ TALER_EXCHANGEDB_rollback (pg);
+ GNUNET_assert (NULL == wpd_head);
+ sweeping = false;
+ if (NULL != shard)
{
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- "No more pending wire transfers, going idle\n");
- GNUNET_assert (NULL == task);
- task = GNUNET_SCHEDULER_add_delayed (transfer_idle_sleep_interval,
- &run_transfers_delayed,
- NULL);
+ go_idle ();
+ return;
}
+ GNUNET_assert (NULL == task);
+ task = GNUNET_SCHEDULER_add_now (&select_shard,
+ NULL);
return;
default:
/* continued in wire_prepare_cb() */
@@ -897,7 +1014,8 @@ run (void *cls,
return;
}
GNUNET_assert (NULL == task);
- task = GNUNET_SCHEDULER_add_now (&select_shard,
+ /* Pick up what earlier runs left behind before taking a shard. */
+ task = GNUNET_SCHEDULER_add_now (&sweep_transfers,
NULL);
GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
cls);
diff --git a/src/exchangedb/iterate_prewires_below_open_shards.c b/src/exchangedb/iterate_prewires_below_open_shards.c
@@ -0,0 +1,155 @@
+/*
+ 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/>
+ */
+/**
+ * @file exchangedb/iterate_prewires_below_open_shards.c
+ * @brief Implementation of the iterate_prewires_below_open_shards function for Postgres
+ * @author Christian Grothoff
+ */
+#include "taler/taler_pq_lib.h"
+#include "exchange-database/iterate_prewires_below_open_shards.h"
+#include "helper.h"
+
+/**
+ * Closure for #prewire_cb().
+ */
+struct PrewireContext
+{
+ /**
+ * Function to call on each result.
+ */
+ TALER_EXCHANGEDB_WirePreparationIterator cb;
+
+ /**
+ * Closure for @a cb.
+ */
+ void *cb_cls;
+
+ /**
+ * #GNUNET_OK if everything went fine.
+ */
+ enum GNUNET_GenericReturnValue status;
+};
+
+
+/**
+ * Invoke the callback for each result.
+ *
+ * @param cls a `struct PrewireContext *`
+ * @param result SQL result
+ * @param num_results number of rows in @a result
+ */
+static void
+prewire_cb (void *cls,
+ PGresult *result,
+ unsigned int num_results)
+{
+ struct PrewireContext *pc = cls;
+
+ for (unsigned int i = 0; i < num_results; i++)
+ {
+ uint64_t prewire_uuid;
+ char *wire_method;
+ void *buf = NULL;
+ size_t buf_size;
+ struct GNUNET_PQ_ResultSpec rs[] = {
+ GNUNET_PQ_result_spec_uint64 ("prewire_uuid",
+ &prewire_uuid),
+ GNUNET_PQ_result_spec_string ("wire_method",
+ &wire_method),
+ GNUNET_PQ_result_spec_variable_size ("buf",
+ &buf,
+ &buf_size),
+ GNUNET_PQ_result_spec_end
+ };
+
+ if (GNUNET_OK !=
+ GNUNET_PQ_extract_result (result,
+ rs,
+ i))
+ {
+ GNUNET_break (0);
+ pc->status = GNUNET_SYSERR;
+ return;
+ }
+ pc->cb (pc->cb_cls,
+ prewire_uuid,
+ wire_method,
+ buf,
+ buf_size);
+ GNUNET_PQ_cleanup_result (rs);
+ }
+}
+
+
+enum GNUNET_DB_QueryStatus
+TALER_EXCHANGEDB_iterate_prewires_below_open_shards (
+ struct TALER_EXCHANGEDB_PostgresContext *pg,
+ const char *job_name,
+ uint64_t limit,
+ TALER_EXCHANGEDB_WirePreparationIterator cb,
+ void *cb_cls)
+{
+ struct GNUNET_PQ_QueryParam params[] = {
+ GNUNET_PQ_query_param_string (job_name),
+ GNUNET_PQ_query_param_uint64 (&limit),
+ GNUNET_PQ_query_param_end
+ };
+ struct PrewireContext pc = {
+ .cb = cb,
+ .cb_cls = cb_cls,
+ .status = GNUNET_OK
+ };
+ enum GNUNET_DB_QueryStatus qs;
+
+ /* The bound is the start of the lowest shard still open, or, with none
+ open, the end of the highest shard: new shards are only ever opened
+ above that, so every row below the bound belongs to a completed shard.
+ No lease covers those rows, hence SKIP LOCKED keeps two workers from
+ executing the same one. */
+ PREPARE (pg,
+ "iterate_prewires_below_open_shards",
+ "SELECT"
+ " prewire_uuid"
+ ",wire_method"
+ ",buf"
+ " FROM prewire"
+ " WHERE prewire_uuid <"
+ " COALESCE ("
+ " (SELECT MIN(start_row)"
+ " FROM work_shards"
+ " WHERE job_name=$1"
+ " AND completed=FALSE),"
+ " (SELECT end_row"
+ " FROM work_shards"
+ " WHERE job_name=$1"
+ " ORDER BY end_row DESC"
+ " LIMIT 1),"
+ " 0)"
+ " AND finished=FALSE"
+ " AND failed=FALSE"
+ " ORDER BY prewire_uuid ASC"
+ " LIMIT $2"
+ " FOR UPDATE SKIP LOCKED;");
+ qs = GNUNET_PQ_eval_prepared_multi_select (
+ pg->conn,
+ "iterate_prewires_below_open_shards",
+ params,
+ &prewire_cb,
+ &pc);
+ if (GNUNET_OK != pc.status)
+ return GNUNET_DB_STATUS_HARD_ERROR;
+ return qs;
+}
diff --git a/src/exchangedb/meson.build b/src/exchangedb/meson.build
@@ -257,6 +257,7 @@ libtalerexchangedb = library(
'update_legitimization_process_by_row.c',
'update_wire.c',
'iterate_prewires.c',
+ 'iterate_prewires_below_open_shards.c',
'insert_prewire.c',
'update_to_prewire_failed.c',
'update_to_prewire_finished.c',
diff --git a/src/exchangedb/test_prewire.c b/src/exchangedb/test_prewire.c
@@ -30,6 +30,7 @@
#include "test_common.h"
#include "exchange-database/insert_prewire.h"
#include "exchange-database/iterate_prewires.h"
+#include "exchange-database/iterate_prewires_below_open_shards.h"
#include "exchange-database/update_to_prewire_finished.h"
#include "exchange-database/update_to_prewire_failed.h"
@@ -109,6 +110,41 @@ ctx_free (struct PrewireContext *ctx)
/**
+ * Closure for #rows_cb().
+ */
+struct RowsContext
+{
+ /**
+ * Rows seen, in the order the callback saw them.
+ */
+ uint64_t rows[8];
+
+ /**
+ * How many rows did the callback see?
+ */
+ unsigned int total;
+};
+
+
+static void
+rows_cb (void *cls,
+ uint64_t rowid,
+ const char *wire_method,
+ const char *buf,
+ size_t buf_size)
+{
+ struct RowsContext *rc = cls;
+
+ (void) wire_method;
+ (void) buf;
+ (void) buf_size;
+ if (rc->total < sizeof (rc->rows) / sizeof (rc->rows[0]))
+ rc->rows[rc->total] = rowid;
+ rc->total++;
+}
+
+
+/**
* An empty queue yields nothing, and marking rows that do not exist does
* nothing.
*
@@ -295,6 +331,138 @@ check_finished_and_failed (struct TALER_EXCHANGEDB_PostgresContext *pg)
/**
+ * Unfinished rows below every open shard of the job are found; rows in
+ * or above an open shard, finished and failed rows, and the shards of
+ * other jobs do not count.
+ *
+ * @param pg the database context
+ * @return 0 on success
+ */
+static int
+check_below_open_shards (struct TALER_EXCHANGEDB_PostgresContext *pg)
+{
+ static const char buf[] = "late-transfer";
+ struct RowsContext rc;
+ uint64_t r[4];
+
+ for (unsigned int i = 0; i < 4; i++)
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ TALER_EXCHANGEDB_insert_prewire (pg,
+ "iban",
+ buf,
+ sizeof (buf)));
+ /* the rows of the earlier checks are finished or failed */
+ memset (&rc,
+ 0,
+ sizeof (rc));
+ FAILIF (0 >=
+ TALER_EXCHANGEDB_iterate_prewires (pg,
+ 0,
+ 10,
+ &rows_cb,
+ &rc));
+ FAILIF (4 != rc.total);
+ memcpy (r,
+ rc.rows,
+ sizeof (r));
+
+ /* without shards, nothing is below them */
+ memset (&rc,
+ 0,
+ sizeof (rc));
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
+ TALER_EXCHANGEDB_iterate_prewires_below_open_shards (pg,
+ "transfer",
+ 10,
+ &rows_cb,
+ &rc));
+ FAILIF (0 != rc.total);
+
+ /* r[0] and r[1] lie in a completed shard, r[2] and r[3] in an open
+ one; an open shard of another job covers all of them */
+ FAILIF (GNUNET_OK !=
+ TDB_exec (pg,
+ "INSERT INTO work_shards"
+ " (job_name,last_attempt,start_row,end_row"
+ " ,completed,progress_row)"
+ " VALUES"
+ " ('transfer',0,0,%llu,TRUE,%llu)"
+ ",('transfer',0,%llu,%llu,FALSE,%llu)"
+ ",('other',0,0,%llu,FALSE,0);",
+ (unsigned long long) r[2],
+ (unsigned long long) r[2],
+ (unsigned long long) r[2],
+ (unsigned long long) (r[3] + 10),
+ (unsigned long long) (r[3] + 1),
+ (unsigned long long) (r[3] + 10)));
+ memset (&rc,
+ 0,
+ sizeof (rc));
+ FAILIF (0 >=
+ TALER_EXCHANGEDB_iterate_prewires_below_open_shards (pg,
+ "transfer",
+ 10,
+ &rows_cb,
+ &rc));
+ FAILIF (2 != rc.total);
+ FAILIF (r[0] != rc.rows[0]);
+ FAILIF (r[1] != rc.rows[1]);
+
+ /* the limit is honoured */
+ memset (&rc,
+ 0,
+ sizeof (rc));
+ FAILIF (0 >=
+ TALER_EXCHANGEDB_iterate_prewires_below_open_shards (pg,
+ "transfer",
+ 1,
+ &rows_cb,
+ &rc));
+ FAILIF (1 != rc.total);
+ FAILIF (r[0] != rc.rows[0]);
+
+ /* finished rows drop out */
+ FAILIF (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
+ TALER_EXCHANGEDB_update_to_prewire_finished (pg,
+ r[0]));
+ memset (&rc,
+ 0,
+ sizeof (rc));
+ FAILIF (0 >=
+ TALER_EXCHANGEDB_iterate_prewires_below_open_shards (pg,
+ "transfer",
+ 10,
+ &rows_cb,
+ &rc));
+ FAILIF (1 != rc.total);
+ FAILIF (r[1] != rc.rows[0]);
+
+ /* with no shard of the job open, everything below the highest shard's
+ end counts */
+ FAILIF (GNUNET_OK !=
+ TDB_exec (pg,
+ "UPDATE work_shards"
+ " SET completed=TRUE"
+ " ,progress_row=end_row"
+ " WHERE job_name='transfer';"));
+ memset (&rc,
+ 0,
+ sizeof (rc));
+ FAILIF (0 >=
+ TALER_EXCHANGEDB_iterate_prewires_below_open_shards (pg,
+ "transfer",
+ 10,
+ &rows_cb,
+ &rc));
+ FAILIF (3 != rc.total);
+ FAILIF (r[1] != rc.rows[0]);
+ FAILIF (r[2] != rc.rows[1]);
+ FAILIF (r[3] != rc.rows[2]);
+ return 0;
+}
+
+
+/**
* The checks to run, in order.
*/
static const struct TDB_Test tests[] = {
@@ -304,6 +472,8 @@ static const struct TDB_Test tests[] = {
&check_insert_and_iterate },
{ "prewire-finished-and-failed",
&check_finished_and_failed },
+ { "prewire-below-open-shards",
+ &check_below_open_shards },
{ NULL, NULL }
};
diff --git a/src/include/exchange-database/iterate_prewires_below_open_shards.h b/src/include/exchange-database/iterate_prewires_below_open_shards.h
@@ -0,0 +1,53 @@
+/*
+ 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/>
+ */
+/**
+ * @file src/include/exchange-database/iterate_prewires_below_open_shards.h
+ * @brief implementation of the iterate_prewires_below_open_shards function for Postgres
+ * @author Christian Grothoff
+ */
+#ifndef EXCHANGE_DATABASE_ITERATE_PREWIRES_BELOW_OPEN_SHARDS_H
+#define EXCHANGE_DATABASE_ITERATE_PREWIRES_BELOW_OPEN_SHARDS_H
+
+#include "taler/taler_json_lib.h"
+#include "exchangedb_lib.h"
+#include "exchange-database/iterate_prewires.h"
+
+
+/**
+ * Function called to get unfinished wire transfer preparations that lie
+ * below every shard of @a job_name that is still open, that is, in shards
+ * that were completed while the transaction inserting the row had not yet
+ * committed. Locks the rows returned until the end of the transaction and
+ * skips rows another transaction holds locked.
+ *
+ * Primary test table: `prewire` (see test_prewire.c).
+ *
+ * @param pg the database context
+ * @param job_name name of the work shard job that processes `prewire`
+ * @param limit maximum number of results to return
+ * @param cb function to call for each unfinished item
+ * @param cb_cls closure for @a cb
+ * @return transaction status code
+ */
+enum GNUNET_DB_QueryStatus
+TALER_EXCHANGEDB_iterate_prewires_below_open_shards (
+ struct TALER_EXCHANGEDB_PostgresContext *pg,
+ const char *job_name,
+ uint64_t limit,
+ TALER_EXCHANGEDB_WirePreparationIterator cb,
+ void *cb_cls);
+
+#endif
