From 65bcc00cd9b289193e27a6f74b053e71f90be698 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@windmill.dev>
Date: Thu, 14 Aug 2025 17:31:38 +0000
Subject: [PATCH] fix: add timeouts to more queries to prevent some rare
 deadlocks scnarios

---
 backend/src/monitor.rs                | 26 ++++++++++++++++----------
 backend/windmill-common/src/worker.rs | 17 ++++++++++++-----
 2 files changed, 28 insertions(+), 15 deletions(-)

diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs
index 379ec89f97..2668934c8a 100644
--- a/backend/src/monitor.rs
+++ b/backend/src/monitor.rs
@@ -18,6 +18,7 @@ use sqlx::{Pool, Postgres};
 use tokio::{
     join,
     sync::{mpsc, RwLock},
+    time::timeout,
 };
 use uuid::Uuid;
 
@@ -684,20 +685,25 @@ async fn send_log_file_to_object_store(
         let (ok_lines, err_lines) = read_log_counters(ts_str);
 
         if let Some(db) = conn.as_sql() {
-            if let Err(e) = sqlx::query!("INSERT INTO log_file (hostname, mode, worker_group, log_ts, file_path, ok_lines, err_lines, json_fmt)
+            match timeout(Duration::from_secs(10), sqlx::query!("INSERT INTO log_file (hostname, mode, worker_group, log_ts, file_path, ok_lines, err_lines, json_fmt)
              VALUES ($1, $2::text::LOG_MODE, $3, $4, $5, $6, $7, $8)
              ON CONFLICT (hostname, log_ts) DO UPDATE SET ok_lines = log_file.ok_lines + $6, err_lines = log_file.err_lines + $7",
                 hostname, mode.to_string(), worker_group.clone(), ts, highest_file, ok_lines as i64, err_lines as i64, *JSON_FMT)
-                .execute(db)
-                .await {
-                tracing::error!("Error inserting log file: {:?}", e);
-            } else {
-                if let Err(e) = LAST_LOG_FILE_SENT.lock().map(|mut last_log_file_sent| {
-                    last_log_file_sent.replace(ts);
-                }) {
-                    tracing::error!("Error updating last log file sent: {:?}", e);
+                .execute(db)).await {
+                Ok(Ok(_)) => {
+                    if let Err(e) = LAST_LOG_FILE_SENT.lock().map(|mut last_log_file_sent| {
+                        last_log_file_sent.replace(ts);
+                    }) {
+                        tracing::error!("Error updating last log file sent: {:?}", e);
+                    }
+                    tracing::info!("Log file sent: {}", highest_file);
+                }
+                Ok(Err(e)) => {
+                    tracing::error!("Error inserting log file: {:?}", e);
+                }
+                Err(e) => {
+                    tracing::error!("Error inserting log file, timeout elapsed: {:?}", e);
                 }
-                tracing::info!("Log file sent: {}", highest_file);
             }
         } else {
             // tracing::warn!("Not sending log file to object store in agent mode");
diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs
index 5236bb8f75..c34b9959f1 100644
--- a/backend/windmill-common/src/worker.rs
+++ b/backend/windmill-common/src/worker.rs
@@ -18,15 +18,22 @@ use std::{
     path::{Component, Path, PathBuf},
     str::FromStr,
     sync::{atomic::AtomicBool, Arc},
+    time::Duration,
 };
 #[cfg(windows)]
 use sysinfo::System;
-use tokio::sync::RwLock;
+use tokio::{sync::RwLock, time::timeout};
 use uuid::Uuid;
 use windmill_macros::annotations;
 
 use crate::{
-    agent_workers::PingJobStatusResponse, cache::{unwrap_or_error, RawNode, RawScript}, error::{self, to_anyhow}, global_settings::CUSTOM_TAGS_SETTING, indexer::TantivyIndexerSettings, server::Smtp, KillpillSender, BASE_INTERNAL_URL, DB
+    agent_workers::PingJobStatusResponse,
+    cache::{unwrap_or_error, RawNode, RawScript},
+    error::{self, to_anyhow},
+    global_settings::CUSTOM_TAGS_SETTING,
+    indexer::TantivyIndexerSettings,
+    server::Smtp,
+    KillpillSender, BASE_INTERNAL_URL, DB,
 };
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
@@ -1353,7 +1360,7 @@ pub async fn update_worker_ping_main_loop_query(
     occupancy_rate_30m: Option<f32>,
     db: &DB,
 ) -> anyhow::Result<()> {
-    sqlx::query!(
+    timeout(Duration::from_secs(10), sqlx::query!(
         "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2,
          occupancy_rate = $3, memory_usage = $4, wm_memory_usage = $5, vcpus = COALESCE($7, vcpus),
          memory = COALESCE($8, memory), occupancy_rate_15s = $9, occupancy_rate_5m = $10, occupancy_rate_30m = $11 WHERE worker = $6",
@@ -1369,8 +1376,8 @@ pub async fn update_worker_ping_main_loop_query(
         occupancy_rate_5m,
         occupancy_rate_30m,
     )
-    .execute(db)
-    .await?;
+        .execute(db))
+    .await??;
     Ok(())
 }