commit 74add07b81510d98e76ac33353304ddf05500781
parent 4c57cbe8d93896185f2efd8f5d9b8df2d8b04c30
Author: Christian Grothoff <grothoff@gnunet.org>
Date: Tue, 14 Jul 2026 23:35:05 +0200
bound concurrently executed webhooks
Diffstat:
4 files changed, 47 insertions(+), 3 deletions(-)
diff --git a/src/backend/taler-merchant-webhook.c b/src/backend/taler-merchant-webhook.c
@@ -34,6 +34,12 @@
#include "merchant-database/preflight.h"
+/**
+ * Maximum number of webhooks we execute concurrently.
+ */
+#define CONCURRENCY_LIMIT 32
+
+
struct WorkResponse
{
struct WorkResponse *next;
@@ -49,6 +55,13 @@ static struct WorkResponse *w_head;
static struct WorkResponse *w_tail;
+/**
+ * Number of entries in the @e w_head DLL, that is the number
+ * of webhooks currently in flight. Never exceeds
+ * #CONCURRENCY_LIMIT.
+ */
+static uint64_t w_count;
+
static struct GNUNET_DB_EventHandler *event_handler;
/**
@@ -115,6 +128,7 @@ shutdown_task (void *cls)
GNUNET_CONTAINER_DLL_remove (w_head,
w_tail,
w);
+ w_count--;
GNUNET_CURL_job_cancel (w->job);
curl_slist_free_all (w->job_headers);
GNUNET_free (w->body);
@@ -171,11 +185,21 @@ handle_webhook_response (void *cls,
GNUNET_CONTAINER_DLL_remove (w_head,
w_tail,
w);
+ w_count--;
GNUNET_free (w->body);
curl_slist_free_all (w->job_headers);
- if (NULL == w_head)
+ if (0 == w_count)
+ {
+ /* We only SELECT() again after having finished all requests of
+ the current batch: the rows we are working on are only updated
+ (or deleted) once their request completed, so selecting earlier
+ would simply return the very same webhooks again and run them
+ a second time. */
+ if (NULL != task)
+ GNUNET_SCHEDULER_cancel (task);
task = GNUNET_SCHEDULER_add_now (&select_work,
NULL);
+ }
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Webhook %llu returned with status %ld\n",
(unsigned long long) w->webhook_pending_serial,
@@ -296,6 +320,8 @@ pending_webhooks_cb (void *cls,
GNUNET_CONTAINER_DLL_insert (w_head,
w_tail,
w);
+ w_count++;
+ GNUNET_assert (w_count <= CONCURRENCY_LIMIT);
w->webhook_pending_serial = webhook_pending_serial;
eh = curl_easy_init ();
GNUNET_assert (NULL != eh);
@@ -365,6 +391,7 @@ pending_webhooks_cb (void *cls,
GNUNET_CONTAINER_DLL_remove (w_head,
w_tail,
w);
+ w_count--;
GNUNET_free (w);
GNUNET_SCHEDULER_shutdown ();
return;
@@ -438,11 +465,22 @@ select_work (void *cls)
{
enum GNUNET_DB_QueryStatus qs;
struct GNUNET_TIME_Relative rel;
+ uint64_t limit;
(void) cls;
task = NULL;
+ GNUNET_assert (w_count <= CONCURRENCY_LIMIT);
+ limit = CONCURRENCY_LIMIT - w_count;
+ if (0 == limit)
+ {
+ /* All slots busy; handle_webhook_response() will select
+ more work once the batch completed. */
+ GNUNET_break (0);
+ return;
+ }
TALER_MERCHANTDB_preflight (pg);
qs = TALER_MERCHANTDB_lookup_pending_webhooks (pg,
+ limit,
&pending_webhooks_cb,
NULL);
switch (qs)
diff --git a/src/backenddb/lookup_pending_webhooks.c b/src/backenddb/lookup_pending_webhooks.c
@@ -116,6 +116,7 @@ lookup_pending_webhooks_cb (void *cls,
enum GNUNET_DB_QueryStatus
TALER_MERCHANTDB_lookup_pending_webhooks (
struct TALER_MERCHANTDB_PostgresContext *pg,
+ uint64_t limit,
TALER_MERCHANTDB_PendingWebhooksCallback cb,
void *cb_cls)
{
@@ -127,6 +128,7 @@ TALER_MERCHANTDB_lookup_pending_webhooks (
struct GNUNET_TIME_Absolute now = GNUNET_TIME_absolute_get ();
struct GNUNET_PQ_QueryParam params[] = {
GNUNET_PQ_query_param_absolute_time (&now),
+ GNUNET_PQ_query_param_uint64 (&limit),
GNUNET_PQ_query_param_end
};
enum GNUNET_DB_QueryStatus qs;
@@ -143,7 +145,8 @@ TALER_MERCHANTDB_lookup_pending_webhooks (
" ,body"
" FROM merchant.merchant_pending_webhooks"
" WHERE next_attempt <= $1"
- " ORDER BY next_attempt ASC");
+ " ORDER BY next_attempt ASC"
+ " LIMIT $2");
qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
"lookup_pending_webhooks",
params,
diff --git a/src/backenddb/test_merchantdb.c b/src/backenddb/test_merchantdb.c
@@ -7221,6 +7221,7 @@ test_lookup_pending_webhooks (const struct InstanceData *instance,
memset (results_matching, 0, sizeof (results_matching));
if (0 > TALER_MERCHANTDB_lookup_pending_webhooks (pg,
+ 1024, /* limit */
&lookup_pending_webhooks_cb,
&cls))
{
diff --git a/src/include/merchant-database/lookup_pending_webhooks.h b/src/include/merchant-database/lookup_pending_webhooks.h
@@ -57,12 +57,14 @@ typedef void
* send.
*
* @param pg database context
+ * @param limit maximum number of webhooks to return
* @param cb pending webhook callback
* @param cb_cls callback closure
*/
-// WHERE next_attempt <= now ORDER BY next_attempt ASC
+// WHERE next_attempt <= now ORDER BY next_attempt ASC LIMIT limit
enum GNUNET_DB_QueryStatus
TALER_MERCHANTDB_lookup_pending_webhooks (struct TALER_MERCHANTDB_PostgresContext *pg,
+ uint64_t limit,
TALER_MERCHANTDB_PendingWebhooksCallback cb,
void *cb_cls);
