kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: improve scheduling reliability in extreme pool contention conditions (#7825)

Browse Source 
* fix: reuse outer tx for schedule push in commit_completed_job

Instead of calling handle_maybe_scheduled_job(db) which opens its own
connections (peak=3), inline the schedule push using a savepoint on the
outer transaction. Auth is fetched via the tx connection using
fetch_authed_from_permissioned_as_conn, and push_scheduled_job runs
on a savepoint so failures roll back only the push, not the completion.

On push failure: savepoint rolls back, schedule is disabled on the outer
tx, and the zombie return path is preserved if disabling also fails.

Peak connections drop from 3 to 1 (or 2 on cold RunnableSettings cache).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* all

* fix: extract shared try_schedule_next_job to unify schedule push paths

Replace the two diverging schedule-push implementations (inlined in
commit_completed_job and standalone handle_maybe_scheduled_job) with a
single try_schedule_next_job that reuses the caller's transaction via
savepoints. This eliminates extra pool connection usage in the
worker_flow.rs path and ensures consistent retry/error semantics.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* test: add failpoint markers to try_schedule_next_job

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* chore: remove plan.md

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: remove inner retry loop from try_schedule_next_job, add caller-level retries

The 10-retry x 5s-sleep loop inside try_schedule_next_job held locks on
v2_job_completed/v2_job_queue for up to ~45s when running inside the
outer commit_completed_job transaction.

Now try_schedule_next_job makes a single attempt and returns errors to
the caller. Non-retryable errors (QuotaExceeded, NotFound) disable the
schedule immediately inside the function. Transient errors are returned
for the caller to retry:

- commit_completed_job path: outer backon retry (10x3s) retries the
  entire transaction including the schedule push, so no locks are held
  during sleep.
- handle_flow path: new backon retry (10x3s) wraps begin/push/commit
  with a fresh transaction per attempt.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: clear push_err after successful schedule disable to prevent stuck schedules

When try_schedule_next_job disables the schedule for non-retryable errors
(NotFound, QuotaExceeded), clear the error so the caller commits the tx
(persisting the disable). Previously, the error propagated up, causing the
tx to be dropped and rolling back the disable — leaving the schedule
permanently enabled but broken.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: add 5s timeout on push_scheduled_job, clean up handle_flow error handling

- Add tokio::time::timeout(5s) around push_scheduled_job inside
  try_schedule_next_job to bound worst-case lock holding per attempt
- Remove unreachable QuotaExceeded/NotFound match arms in handle_flow
  (these errors are handled internally by try_schedule_next_job)
- Add report_error_to_workspace_handler_or_critical_side_channel in
  handle_flow when post-exhaustion schedule disable fails

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: return SchedulePushZombieError when both schedule push and disable fail

When handle_flow cannot push the next scheduled job AND cannot disable the
schedule, return a SchedulePushZombieError so the worker leaves the flow job
in the queue for zombie detection to restart. This prevents stuck schedules
where neither the next tick was pushed nor the schedule was disabled.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-02-07 14:16:32 +01:00
committed by GitHub
parent 153dd32187
commit bbb397b6ad
6 changed files with 1318 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/Cargo.toml
View File

@@ -207,7 +207,7 @@ tikv-jemalloc-ctl = { optional = true, workspace = true }
[dev-dependencies]
serde_json.workspace = true
reqwest.workspace = true
windmill-queue.workspace = true
windmill-queue = { workspace = true, features = ["failpoints"] }
axum.workspace = true
serde.workspace = true
windmill-api-client.workspace = true

1072
backend/tests/schedule_push.rs
View File

File diff suppressed because it is too large Load Diff

1
backend/windmill-queue/Cargo.toml
View File

@@ -14,6 +14,7 @@ private = []
enterprise = ["windmill-common/enterprise"]
cloud = []
benchmark = ["windmill-common/benchmark"]
failpoints = []
prometheus = ["dep:prometheus"]
smtp = []

260
backend/windmill-queue/src/jobs.rs
View File

@@ -23,7 +23,7 @@ use reqwest::Client;
use serde::Deserialize;
use serde::{ser::SerializeMap, Serialize};
use serde_json::{json, value::RawValue};
use sqlx::{types::Json, Pool, Postgres, Transaction};
use sqlx::{types::Json, Acquire, Pool, Postgres, Transaction};
use sqlx::{Encode, PgExecutor};
use tokio::sync::mpsc::Sender;
use tokio::sync::oneshot;
@@ -56,7 +56,7 @@ use windmill_common::{
auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username},
cache::{self, FlowData},
db::{Authed, UserDB},
error::{self, to_anyhow, Error},
error::{self, Error},
flow_status::{
BranchAllStatus, FlowCleanupModule, FlowStatus, FlowStatusModule, FlowStatusModuleWParent,
Iterator as FlowIterator, JobResult, RestartedFrom, RetryStatus, MAX_RETRY_ATTEMPTS,
@@ -114,6 +114,26 @@ lazy_static::lazy_static! {
}
#[cfg(feature = "failpoints")]
pub mod schedule_failpoints {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ScheduleFailPoint {
SavepointCreate,
Push,
PushQuotaExceeded,
SavepointCommit,
ScheduleDisable,
}
tokio::task_local! {
pub static ACTIVE: ScheduleFailPoint;
}
pub fn is_active(point: ScheduleFailPoint) -> bool {
ACTIVE.try_with(|fp| *fp == point).unwrap_or(false)
}
}
lazy_static::lazy_static! {
pub static ref HTTP_CLIENT: Client = configure_client(reqwest::ClientBuilder::new()
.user_agent("windmill/beta")
@@ -832,13 +852,14 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
.retry(
ConstantBuilder::default()
.with_delay(std::time::Duration::from_secs(3))
.with_max_times(5)
.with_max_times(10)
.build(),
)
.when(|err| {
!matches!(err, Error::QuotaExceeded(_))
&& !matches!(err, Error::ResultTooLarge(_))
&& !matches!(err, Error::AlreadyCompleted(_))
&& !matches!(err, Error::NotFound(_))
})
.notify(|err, dur| {
tracing::error!("Could not insert completed job, retrying in {dur:#?}, err: {err:#?}");
@@ -1100,22 +1121,12 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
.unwrap_or(false);
if schedule_next_tick {
if let Err(err) = Box::pin(handle_maybe_scheduled_job(
db,
completed_job,
&schedule,
&script_path,
&completed_job.workspace_id,
))
.warn_after_seconds(10)
.await
{
match err {
Error::QuotaExceeded(_) => (),
// scheduling next job failed and could not disable schedule => make zombie job to retry
_ => return Ok((Some(job_id), 0, true)),
}
};
let (returned_tx, schedule_push_err) =
try_schedule_next_job(db, tx, completed_job, &schedule, &script_path).await;
tx = returned_tx;
if let Some(err) = schedule_push_err {
return Err(err);
}
}
#[cfg(all(feature = "enterprise", feature = "private"))]
@@ -1768,100 +1779,169 @@ pub async fn send_success_to_workspace_handler<'a, 'c, T: Serialize + Send + Syn
Ok(())
}
pub async fn handle_maybe_scheduled_job<'c>(
pub async fn try_schedule_next_job<'c>(
db: &Pool<Postgres>,
mut tx: Transaction<'c, Postgres>,
job: &MiniCompletedJob,
schedule: &Schedule,
script_path: &str,
w_id: &str,
) -> windmill_common::error::Result<()> {
) -> (Transaction<'c, Postgres>, Option<Error>) {
if !schedule.enabled {
tracing::info!(
"Schedule {} in {} is disabled. Not scheduling again.",
schedule.path,
&job.workspace_id
);
return (tx, None);
}
if script_path != schedule.script_path {
tracing::warn!(
"Schedule {} in {} has a different script path than the job. Not scheduling again",
schedule.path,
&job.workspace_id
);
return (tx, None);
}
tracing::info!(
"Schedule {} scheduling next job for {} in {w_id}",
"Schedule {} scheduling next job for {} in {}",
schedule.path,
schedule.script_path
schedule.script_path,
&job.workspace_id
);
if schedule.enabled && script_path == schedule.script_path {
let schedule_authed = windmill_common::auth::fetch_authed_from_permissioned_as(
windmill_common::users::username_to_permissioned_as(&schedule.edited_by),
schedule.email.clone(),
w_id,
db,
let schedule_authed =
windmill_common::auth::fetch_authed_from_permissioned_as_conn(
&windmill_common::users::username_to_permissioned_as(
&schedule.edited_by,
),
&schedule.email,
&job.workspace_id,
&mut *tx,
)
.await
.ok();
let push_next_job_future = (|| {
tokio::time::timeout(std::time::Duration::from_secs(5), async {
let mut tx = db.begin().await?;
tx = push_scheduled_job(db, tx, &schedule, schedule_authed.as_ref(), Some(job.scheduled_for)).await?;
tx.commit().await?;
Ok::<(), Error>(())
})
.map_err(|e| Error::internal_err(format!("Pushing next scheduled job timedout: {e:#}")))
.unwrap_or_else(|e| Err(e))
})
.retry(
ConstantBuilder::default()
.with_delay(std::time::Duration::from_secs(5))
.with_max_times(10)
.build(),
)
.when(|err| !matches!(err, Error::QuotaExceeded(_)))
.notify(|err, dur| {
tracing::error!(
"Could not push next scheduled job for schedule {}, retrying in {dur:#?}, err: {err:#?}", schedule.path
);
})
.sleep(tokio::time::sleep);
match push_next_job_future.await {
Ok(()) => Ok(()),
Err(err) => {
let update_schedule = sqlx::query!(
"UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
err.to_string(),
&schedule.workspace_id,
&schedule.path
let mut push_err = None;
#[cfg(feature = "failpoints")]
if schedule_failpoints::is_active(schedule_failpoints::ScheduleFailPoint::SavepointCreate) {
push_err = Some(Error::internal_err("failpoint: savepoint create".to_string()));
}
if push_err.is_none() {
let savepoint_result = tx.begin().await;
match savepoint_result {
Ok(savepoint) => {
let push_result = match tokio::time::timeout(
std::time::Duration::from_secs(5),
push_scheduled_job(
db,
savepoint,
schedule,
schedule_authed.as_ref(),
Some(job.scheduled_for),
),
)
.execute(db)
.await;
match update_schedule {
Ok(_) => {
match err {
Error::QuotaExceeded(_) => {}
_ => {
report_error_to_workspace_handler_or_critical_side_channel(job, db,
format!("Could not schedule next job for {} with err {}. Schedule disabled", schedule.path, err.to_string()),
).await;
.await
{
Ok(result) => result,
Err(_elapsed) => Err(Error::internal_err(
"push_scheduled_job timed out after 5s".to_string(),
)),
};
#[cfg(feature = "failpoints")]
let push_result = if schedule_failpoints::is_active(schedule_failpoints::ScheduleFailPoint::Push) {
if let Ok(sp) = push_result { sp.rollback().await.ok(); }
Err(Error::internal_err("failpoint: push".to_string()))
} else if schedule_failpoints::is_active(schedule_failpoints::ScheduleFailPoint::PushQuotaExceeded) {
if let Ok(sp) = push_result { sp.rollback().await.ok(); }
Err(Error::QuotaExceeded("failpoint: push quota exceeded".to_string()))
} else {
push_result
};
match push_result {
Ok(savepoint) => {
#[cfg(feature = "failpoints")]
let savepoint_commit_fail = schedule_failpoints::is_active(schedule_failpoints::ScheduleFailPoint::SavepointCommit);
#[cfg(not(feature = "failpoints"))]
let savepoint_commit_fail = false;
if savepoint_commit_fail {
savepoint.rollback().await.ok();
push_err = Some(Error::internal_err("failpoint: savepoint commit".to_string()));
} else {
match savepoint.commit().await {
Ok(()) => {}
Err(e) => {
push_err = Some(Error::internal_err(format!(
"Could not commit savepoint: {e:#}"
)));
}
}
}
Ok(())
}
Err(disable_err) => match err {
Error::QuotaExceeded(_) => Err(err),
_ => {
report_error_to_workspace_handler_or_critical_side_channel(job, db,
format!("Could not schedule next job for {} and could not disable schedule with err {}.", schedule.path, disable_err),
).await;
Err(to_anyhow(disable_err).into())
}
},
Err(err) if matches!(err, Error::QuotaExceeded(_)) => {
push_err = Some(err);
}
Err(err) => {
tracing::warn!(
"Could not push next scheduled job for {}: {err}",
schedule.path,
);
push_err = Some(err);
}
}
}
Err(e) => {
tracing::error!(
"Could not create savepoint for schedule push: {e:#}",
);
push_err = Some(Error::internal_err(format!(
"Could not create savepoint: {e:#}"
)));
}
}
} else {
if script_path != schedule.script_path {
tracing::warn!(
"Schedule {} in {w_id} has a different script path than the job. Not scheduling again", schedule.path
);
} else {
tracing::info!(
"Schedule {} in {w_id} is disabled. Not scheduling again.",
}
if let Some(ref err) = push_err {
if matches!(err, Error::QuotaExceeded(_) | Error::NotFound(_)) {
tracing::error!(
"Could not push next scheduled job for {}: {err}. Disabling schedule.",
schedule.path
);
let disable_result = sqlx::query!(
"UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
err.to_string(),
&schedule.workspace_id,
&schedule.path
)
.execute(&mut *tx)
.await;
#[cfg(feature = "failpoints")]
let disable_result = if schedule_failpoints::is_active(schedule_failpoints::ScheduleFailPoint::ScheduleDisable) {
Err(sqlx::Error::Protocol("failpoint: schedule disable".to_string()))
} else {
disable_result
};
if let Err(disable_err) = disable_result {
report_error_to_workspace_handler_or_critical_side_channel(
job,
db,
format!(
"Could not push next scheduled job for {} and could not disable schedule: {disable_err}",
schedule.path,
),
)
.await;
} else {
push_err = None;
}
}
Ok(())
}
(tx, push_err)
}
pub const ERROR_HANDLER_PATH_TEAMS: &str = "/workspace-or-schedule-error-handler-teams";

19
backend/windmill-worker/src/worker.rs
View File

@@ -138,7 +138,7 @@ use crate::{
pwsh_executor::handle_powershell_job,
result_processor::{process_result, start_background_processor},
schema::schema_validator_from_main_arg_sig,
worker_flow::handle_flow,
worker_flow::{handle_flow, SchedulePushZombieError},
worker_lockfiles::{
handle_app_dependency_job, handle_dependency_job, handle_flow_dependency_job,
},
@@ -2964,7 +2964,7 @@ pub async fn handle_queued_job(
// Not a preview: fetch from the cache or the database.
_ => cache::job::fetch_flow(db, &job.kind, job.runnable_id).await?,
};
Box::pin(handle_flow(
match Box::pin(handle_flow(
job,
&flow_data,
db,
@@ -2978,8 +2978,19 @@ pub async fn handle_queued_job(
&killpill_rx,
))
.warn_after_seconds(10)
.await?;
Ok(true)
.await
{
Err(err) if err.downcast_ref::<SchedulePushZombieError>().is_some() => {
tracing::error!(
"Schedule push zombie: {err}. Leaving flow job in queue for zombie detection to restart."
);
Ok(true)
}
other => {
other?;
Ok(true)
}
}
} else {
return Err(Error::internal_err(
"Could not handle flow job with agent worker".to_string(),

99
backend/windmill-worker/src/worker_flow.rs
View File

@@ -30,6 +30,7 @@ use sqlx::types::Json;
use sqlx::{FromRow, Postgres, Transaction};
use tracing::instrument;
use uuid::Uuid;
use backon::{BackoffBuilder, ConstantBuilder, Retryable};
use windmill_common::auth::get_job_perms;
#[cfg(feature = "benchmark")]
use windmill_common::bench::BenchmarkIter;
@@ -67,7 +68,8 @@ use windmill_common::{
use windmill_queue::schedule::get_schedule_opt;
use windmill_queue::{
add_completed_job, add_completed_job_error, append_logs, get_mini_pulled_job,
handle_maybe_scheduled_job, insert_concurrency_key, interpolate_args, CanceledBy, FlowRunners,
try_schedule_next_job, insert_concurrency_key, interpolate_args,
report_error_to_workspace_handler_or_critical_side_channel, CanceledBy, FlowRunners,
MiniCompletedJob, MiniPulledJob, PushArgs, PushIsolationLevel, SameWorkerPayload, WrappedError,
};
@@ -76,6 +78,17 @@ use windmill_audit::ActionKind;
use windmill_common::audit::AuditAuthor;
use windmill_queue::{canceled_job_to_result, push};
#[derive(Debug)]
pub struct SchedulePushZombieError(pub String);
impl std::fmt::Display for SchedulePushZombieError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.0)
}
}
impl std::error::Error for SchedulePushZombieError {}
/// Helper function to write itered data to separate table
/// Returns None if data was written to separate table, Some(itered) if it should be stored in JSONB
async fn write_itered_to_db(
@@ -2177,22 +2190,76 @@ pub async fn handle_flow(
.await?;
if let Some(schedule) = schedule {
if let Err(err) = handle_maybe_scheduled_job(
db,
&MiniCompletedJob::from(flow_job.clone()),
&schedule,
flow_job.runnable_path.as_ref().unwrap(),
&flow_job.workspace_id,
)
.warn_after_seconds(5)
.await
{
match err {
Error::QuotaExceeded(_) => return Err(err.into()),
// scheduling next job failed and could not disable schedule => make zombie job to retry
_ => return Ok(()),
let mini_job = MiniCompletedJob::from(flow_job.clone());
let runnable_path = flow_job.runnable_path.as_ref().unwrap().clone();
let schedule_push_result = (|| async {
let tx = db.begin().warn_after_seconds(5).await
.map_err(|e| Error::internal_err(format!("begin tx for schedule push: {e:#}")))?;
let (tx, schedule_push_err) = try_schedule_next_job(
db,
tx,
&mini_job,
&schedule,
&runnable_path,
)
.await;
if let Some(err) = schedule_push_err {
return Err(err);
}
};
tx.commit().warn_after_seconds(5).await
.map_err(|e| Error::internal_err(format!("commit schedule push: {e:#}")))?;
Ok::<(), Error>(())
})
.retry(
ConstantBuilder::default()
.with_delay(std::time::Duration::from_secs(3))
.with_max_times(10)
.build(),
)
.when(|err: &Error| !matches!(err, Error::QuotaExceeded(_) | Error::NotFound(_)))
.notify(|err: &Error, dur: std::time::Duration| {
tracing::error!(
"Could not push next scheduled job for flow schedule {}, retrying in {dur:#?}: {err:#?}",
schedule.path
);
})
.sleep(tokio::time::sleep)
.await;
// Non-retryable errors (QuotaExceeded, NotFound) are handled inside
// try_schedule_next_job (schedule disabled, returns None), so they never
// reach here. This handles only transient errors after retry exhaustion.
if let Err(err) = schedule_push_result {
tracing::error!(
"Could not push next scheduled job for {} after retries: {err}. Disabling schedule.",
schedule.path
);
if let Err(disable_err) = sqlx::query!(
"UPDATE schedule SET enabled = false, error = $1 WHERE workspace_id = $2 AND path = $3",
err.to_string(),
&flow_job.workspace_id,
&schedule.path
)
.execute(db)
.await
{
report_error_to_workspace_handler_or_critical_side_channel(
&mini_job,
db,
format!(
"Could not push next scheduled job for {} and could not disable schedule: {disable_err}",
schedule.path,
),
)
.await;
return Err(SchedulePushZombieError(
format!(
"Could not push or disable schedule {} after retries",
schedule.path
),
).into());
}
}
} else {
tracing::error!(
"Schedule {schedule_path} in {} not found. Impossible to schedule again",