starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-05 16:03:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: better timeouts on job pull (#7434)

Browse Source 
* fix: better timeouts on job pull

Signed-off-by: pyranota <pyra@duck.com>

* increase timeouts

Signed-off-by: pyranota <pyra@duck.com>

---------

Signed-off-by: pyranota <pyra@duck.com>
This commit is contained in:
Pyra
2025-12-22 21:44:35 +00:00
committed by GitHub
parent f83ceb2057
commit b96bf83a45
2 changed files with 49 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-queue/src/jobs.rs
+47 -29
View File
@@ -7,6 +7,7 @@
*/
use std::future::Future;
use std::time::Duration;
use std::{collections::HashMap, sync::Arc, vec};
use anyhow::Context;
@@ -27,6 +28,7 @@ use sqlx::{Encode, PgExecutor};
use tokio::sync::mpsc::Sender;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tokio::time::timeout;
use tokio::{sync::RwLock, time::sleep};
use ulid::Ulid;
use uuid::Uuid;
@@ -2878,19 +2880,26 @@ pub async fn pull(
let job = if query_suspended.is_empty() {
None
} else {
sqlx::query_as::<_, PulledJob>(query_suspended)
.bind(worker_name)
.fetch_optional(db)
.await?
timeout(
Duration::from_secs(15),
sqlx::query_as::<_, PulledJob>(query_suspended)
.bind(worker_name)
.fetch_optional(db),
)
.await??
};
let (job, suspended) = if let Some(job) = job {
(Some(job), true)
} else {
let job = sqlx::query_as::<_, PulledJob>(query_no_suspend)
.bind(worker_name)
.fetch_optional(db)
.await?;
let job = timeout(
Duration::from_secs(15),
sqlx::query_as::<_, PulledJob>(query_no_suspend)
.bind(worker_name)
.fetch_optional(db),
)
.await??;
(job, false)
};
@@ -2937,14 +2946,17 @@ pub async fn pull(
// Concurrency limit is available for either enterprise job or dependency job
&& (cfg!(feature = "enterprise") || (job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING)) =>
{
crate::jobs_ee::apply_concurrency_limit(
db,
pull_loop_count,
suspended,
job,
&concurrency_settings,
timeout(
Duration::from_secs(15),
crate::jobs_ee::apply_concurrency_limit(
db,
pull_loop_count,
suspended,
job,
&concurrency_settings,
),
)
.await?
.await??
.unwrap_or(PulledJobResult {
job: None,
suspended,
@@ -2966,14 +2978,17 @@ pub async fn pull(
return Ok(njob);
};
let (job, suspended) = pull_single_job_and_mark_as_running_no_concurrency_limit(
db,
suspend_first,
worker_name,
#[cfg(feature = "benchmark")]
bench,
let (job, suspended) = timeout(
Duration::from_secs(15),
pull_single_job_and_mark_as_running_no_concurrency_limit(
db,
suspend_first,
worker_name,
#[cfg(feature = "benchmark")]
bench,
),
)
.await?;
.await??;
let Some(job) = job else {
return Ok(PulledJobResult {
job: None,
@@ -3027,14 +3042,17 @@ pub async fn pull(
if cfg!(feature = "enterprise")
|| (pulled_job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING)
{
if let Some(pulled_job_res) = crate::jobs_ee::apply_concurrency_limit(
db,
pull_loop_count,
suspended,
pulled_job,
&concurrency_settings,
if let Some(pulled_job_res) = timeout(
Duration::from_secs(15),
crate::jobs_ee::apply_concurrency_limit(
db,
pull_loop_count,
suspended,
pulled_job,
&concurrency_settings,
),
)
.await?
.await??
{
return Ok(pulled_job_res);
}
backend/windmill-worker/src/worker.rs
+2 -2
View File
@@ -1766,7 +1766,7 @@ pub async fn run_worker(
last_suspend_first = Instant::now();
}
let mut job = match timeout(
Duration::from_secs(10),
Duration::from_secs(30),
pull(
&db,
suspend_first,
@@ -1781,7 +1781,7 @@ pub async fn run_worker(
{
Ok(job) => job,
Err(e) => {
tracing::error!(worker = %worker_name, hostname = %hostname, "pull timed out after 10s, sleeping for 30s: {e:?}");
tracing::error!(worker = %worker_name, hostname = %hostname, "pull timed out after 20s, sleeping for 30s: {e:?}");
tokio::time::sleep(Duration::from_secs(30)).await;
continue;
}