diff --git a/backend/windmill-common/src/db.rs b/backend/windmill-common/src/db.rs index 0a6ceba37d..30e3303ab8 100644 --- a/backend/windmill-common/src/db.rs +++ b/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` 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` 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; }; diff --git a/backend/windmill-common/src/error.rs b/backend/windmill-common/src/error.rs index c2d6640746..843005f2be 100644 --- a/backend/windmill-common/src/error.rs +++ b/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(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::() { crate::db::connection_reset::note_sqlx_error(sqlx_err); } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index bd39afa8c7..eb74984a61 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/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>, 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,