From 78633937c059a0784fab4821e23ef1850e748e03 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@windmill.dev>
Date: Thu, 25 Jun 2026 16:06:25 +0000
Subject: [PATCH] fix(debounce): emit accumulation log after committing the
 claim transaction

append_logs opened a second pool connection while the claim transaction (and its
batch row locks) were still held; under concurrent debounced pulls that risks
pool-exhaustion stalls/timeouts. Defer the log line until after tx.commit().

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 backend/windmill-queue/src/jobs.rs | 22 ++++++++++++----------
 1 file changed, 12 insertions(+), 10 deletions(-)

diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs
index f6fe92c148..8ef38e4f7e 100644
--- a/backend/windmill-queue/src/jobs.rs
+++ b/backend/windmill-queue/src/jobs.rs
@@ -3201,6 +3201,10 @@ impl PulledJobResult {
                 // its own args (dropping the siblings it had claimed). One transaction
                 // makes the claim and the merged-args write commit together (or neither).
                 let mut tx = db.begin().await?;
+                // Emitted AFTER the transaction commits — writing logs via a second pool
+                // connection while the claim tx + row locks are held risks pool-exhaustion
+                // stalls under concurrent debounced pulls.
+                let mut accumulation_log: Option<String> = None;
                 let claim = sqlx::query!(
                     "WITH mine AS (
                         SELECT debounce_batch, consumed_by FROM v2_job_debounce_batch WHERE id = $1
@@ -3311,16 +3315,10 @@ impl PulledJobResult {
                             .map(|v| v.get().to_string())
                             .unwrap_or_else(|| "null".to_string());
 
-                        append_logs(
-                            &j_id,
-                            &j.workspace_id,
-                            format!(
-                                "Accumulating debounced argument `{arg_name_to_accumulate}`:\n  original: {original_value}\n  accumulated: {}\n\n",
-                                &new_value
-                            ),
-                            &(db.into()),
-                        )
-                        .await;
+                        accumulation_log = Some(format!(
+                            "Accumulating debounced argument `{arg_name_to_accumulate}`:\n  original: {original_value}\n  accumulated: {}\n\n",
+                            &new_value
+                        ));
 
                         j.args
                             .get_or_insert(Json(Default::default()))
@@ -3343,6 +3341,10 @@ impl PulledJobResult {
                 // else: claimed_batch is None but it was our own prior claim (re-pull) —
                 // keep the args we already persisted on the first pull.
                 tx.commit().await?;
+
+                if let Some(msg) = accumulation_log {
+                    append_logs(&j_id, &j.workspace_id, msg, &(db.into())).await;
+                }
             } else {
                 // Debounced but no args to accumulate (plain debounce / dependency job):
                 // consume the batch by removing this job's rows.