fix: arm the connection reset from the flow args path

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Qzqmh52NU8fB9RBQNNkJGt

backend/windmill-common/src/db.rs

@@ -331,12 +331,14 @@ pub mod connection_reset {
     }
 
     fn arm() {
-        let previous = RESET_UNTIL_MS.fetch_max(
-            now_ms() + RESET_WINDOW.as_millis() as u64,
-            Ordering::Relaxed,
-        );
-        // Entering a window costs throughput, so it must be attributable in the logs.
-        if previous == 0 {
+        let now = now_ms();
+        let previous =
+            RESET_UNTIL_MS.fetch_max(now + RESET_WINDOW.as_millis() as u64, Ordering::Relaxed);
+        // Every window costs throughput, so each one must be attributable in the logs —
+        // hence the comparison against `now` rather than against zero, which would report
+        // only the first incident a process ever sees. Re-arming inside a live window is
+        // the common case and stays quiet.
+        if previous <= now {
             tracing::warn!(
                 "a pooled connection was found in an aborted transaction; rolling back on \
                  checkout for the next {}s",
@@ -366,9 +368,11 @@ pub mod connection_reset {
     }
 
     /// Arms the reset when an error proves a session is stuck in an aborted transaction.
-    /// Reached from `Error`'s `From<sqlx::Error>` and from `to_anyhow`, which between them
-    /// cover the paths a `sqlx::Error` takes on its way out of a query.
-    pub(crate) fn note_sqlx_error(err: &sqlx::Error) {
+    /// `Error`'s `From<sqlx::Error>` and `to_anyhow` call this, which covers a query whose
+    /// error is converted or propagated with `?`. A caller that instead inspects the
+    /// `sqlx::Error` in place — formatting it into a message, matching on it — has to call
+    /// this itself, or a poisoned connection reported only there goes unnoticed.
+    pub fn note_sqlx_error(err: &sqlx::Error) {
         let sqlx::Error::Database(db_err) = err else {
             return;
         };

backend/windmill-common/src/error.rs

@@ -245,9 +245,10 @@ pub fn relocate_internal(loc: &'static Location<'static>) -> impl FnOnce(Error)
 }
 
 pub fn to_anyhow<T: 'static + std::error::Error + Send + Sync>(e: T) -> anyhow::Error {
-    // The other way a `sqlx::Error` leaves a query without becoming an `Error`, and so the
-    // other place a poisoned connection can announce itself. The downcast is a type-id
-    // comparison, and misses only errors handled entirely in place.
+    // Callers that hand a `sqlx::Error` to anyhow explicitly rather than converting it to
+    // `Error`. The downcast is a type-id comparison, and this covers only the explicit
+    // `map_err(to_anyhow)` form — `?` into an `anyhow::Result` goes straight through
+    // anyhow's own `From`.
     if let Some(sqlx_err) = (&e as &dyn std::any::Any).downcast_ref::<sqlx::Error>() {
         crate::db::connection_reset::note_sqlx_error(sqlx_err);
     }

backend/windmill-worker/src/worker_flow.rs

@@ -634,9 +634,12 @@ pub async fn update_flow_status_after_job_completion_internal(
         }));
 
         let from_result_to_args = |args: &Result<HashMap<String, Box<RawValue>>, sqlx::Error>| {
-            let args = args
-                .as_ref()
-                .map_err(|e| Error::internal_err(format!("retrieval of args from state: {e:#}")))?;
+            let args = args.as_ref().map_err(|e| {
+                // The error is only borrowed here, so it never reaches the conversions that
+                // would otherwise report it.
+                windmill_common::db::connection_reset::note_sqlx_error(e);
+                Error::internal_err(format!("retrieval of args from state: {e:#}"))
+            })?;
 
             Ok::<_, Error>(args.clone())
         };
@@ -4421,8 +4424,7 @@ async fn push_next_flow_job(
             .as_deref()
             .filter(|t| !t.is_empty() && *t != flow_job.tag.as_str())
         {
-            let is_super_admin =
-                windmill_common::auth::is_super_admin_email(db, email).await?;
+            let is_super_admin = windmill_common::auth::is_super_admin_email(db, email).await?;
             check_tag_available_for_workspace_internal(
                 db,
                 &flow_job.workspace_id,
@@ -6155,9 +6157,7 @@ pub async fn script_to_payload(
                 .await?
                 .prefetch_cached(&db)
                 .await?;
-            let on_behalf_of = script_info
-                .on_behalf_of(&flow_job.workspace_id, db)
-                .await?;
+            let on_behalf_of = script_info.on_behalf_of(&flow_job.workspace_id, db).await?;
             let ScriptHashInfo {
                 tag,
                 cache_ttl,