From 67c8aef9faea5afe8ce330b477f16b4aed1779a5 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 8 Feb 2026 15:51:44 +0000 Subject: [PATCH] fix: box push() future to prevent stack overflow in nested async chains The push() function generates a ~13KB async state machine that gets inlined into every caller's future. In deeply nested async chains (e.g. flow execution), this causes stack overflows. Boxing the future at the definition site via a thin wrapper reduces each caller's stack footprint to a single pointer. This also reverts the RUST_MIN_STACK workaround from CI. Co-Authored-By: Claude Opus 4.6 --- .github/workflows/backend-test.yml | 1 - backend/windmill-queue/src/jobs.rs | 273 ++++++++++++++++++++--------- 2 files changed, 194 insertions(+), 80 deletions(-) diff --git a/.github/workflows/backend-test.yml b/.github/workflows/backend-test.yml index 20a709a5c9..2d137b787e 100644 --- a/.github/workflows/backend-test.yml +++ b/.github/workflows/backend-test.yml @@ -214,7 +214,6 @@ jobs: RUST_LOG_STYLE: never CARGO_NET_GIT_FETCH_WITH_CLI: true CARGO_BUILD_JOBS: 4 - RUST_MIN_STACK: 8388608 WMDEBUG_FORCE_V0_WORKSPACE_DEPENDENCIES: 1 WMDEBUG_FORCE_RUNNABLE_SETTINGS_V0: 1 WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT: 1 diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 763622a9c9..62f843ae72 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -41,15 +41,15 @@ use windmill_common::auth::JobPerms; #[cfg(feature = "benchmark")] use windmill_common::bench::BenchmarkIter; use windmill_common::jobs::{JobTriggerKind, EMAIL_ERROR_HANDLER_USER_EMAIL}; +use windmill_common::min_version::{ + MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2, +}; use windmill_common::runnable_settings::{ ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, RunnableSettings, RunnableSettingsTrait, }; use windmill_common::triggers::TriggerMetadata; use windmill_common::utils::{calculate_hash, configure_client, now_from_db}; -use windmill_common::min_version::{ - MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2, -}; use windmill_common::worker::{Connection, SCRIPT_TOKEN_EXPIRY}; use windmill_common::{ @@ -67,11 +67,11 @@ use windmill_common::{ StopAfterIf, }, jobs::{get_payload_tag_from_prefixed_path, JobKind, JobPayload, QueuedJob, RawCode}, + min_version::{MIN_VERSION_IS_AT_LEAST_1_432, MIN_VERSION_IS_AT_LEAST_1_440}, schedule::Schedule, scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang}, users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL}, utils::{not_found_if_none, report_critical_error, StripPath, WarnAfterExt}, - min_version::{MIN_VERSION_IS_AT_LEAST_1_432, MIN_VERSION_IS_AT_LEAST_1_440}, worker::{ to_raw_value, CLOUD_HOSTED, DISABLE_FLOW_SCRIPT, NO_LOGS, WORKER_PULL_QUERIES, WORKER_SUSPENDED_PULL_QUERY, @@ -1175,11 +1175,14 @@ async fn commit_completed_job( } if completed_job.concurrent_limit.is_some() - || windmill_common::runnable_settings::prefetch_cached_from_handle(completed_job.runnable_settings_handle, db) - .await? - .1 - .concurrent_limit - .is_some() + || windmill_common::runnable_settings::prefetch_cached_from_handle( + completed_job.runnable_settings_handle, + db, + ) + .await? + .1 + .concurrent_limit + .is_some() { let concurrency_key = concurrency_key(db, &completed_job.id).await?; if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_none() { @@ -1590,7 +1593,15 @@ async fn fetch_error_handler_from_db( db: &Pool, w_id: &str, ) -> Result<(Option, Option>>, bool, bool), Error> { - sqlx::query_as::<_, (Option, Option>>, Option, Option)>( + sqlx::query_as::< + _, + ( + Option, + Option>>, + Option, + Option, + ), + >( r#" SELECT error_handler->>'path', @@ -1606,7 +1617,12 @@ async fn fetch_error_handler_from_db( .await .context("fetching error handler info from workspace_settings")? .map(|(path, extra_args, muted_on_cancel, muted_on_user_path)| { - (path, extra_args, muted_on_cancel.unwrap_or(false), muted_on_user_path.unwrap_or(false)) + ( + path, + extra_args, + muted_on_cancel.unwrap_or(false), + muted_on_user_path.unwrap_or(false), + ) }) .ok_or_else(|| Error::internal_err(format!("no workspace settings for id {w_id}"))) } @@ -1620,19 +1636,14 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> let w_id = &queued_job.workspace_id; let now = chrono::Utc::now().timestamp(); - let (error_handler, error_handler_extra_args, error_handler_muted_on_cancel, error_handler_muted_on_user_path) = - if let Some(cached) = WORKSPACE_ERROR_HANDLER_CACHE.get(w_id) { - if cached.4 > now { - (cached.0.clone(), cached.1.clone(), cached.2, cached.3) - } else { - let row = fetch_error_handler_from_db(db, w_id).await?; - let expiry = now + WORKSPACE_HANDLER_CACHE_TTL_SECONDS; - WORKSPACE_ERROR_HANDLER_CACHE.insert( - w_id.clone(), - (row.0.clone(), row.1.clone(), row.2, row.3, expiry), - ); - row - } + let ( + error_handler, + error_handler_extra_args, + error_handler_muted_on_cancel, + error_handler_muted_on_user_path, + ) = if let Some(cached) = WORKSPACE_ERROR_HANDLER_CACHE.get(w_id) { + if cached.4 > now { + (cached.0.clone(), cached.1.clone(), cached.2, cached.3) } else { let row = fetch_error_handler_from_db(db, w_id).await?; let expiry = now + WORKSPACE_HANDLER_CACHE_TTL_SECONDS; @@ -1641,7 +1652,16 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> (row.0.clone(), row.1.clone(), row.2, row.3, expiry), ); row - }; + } + } else { + let row = fetch_error_handler_from_db(db, w_id).await?; + let expiry = now + WORKSPACE_HANDLER_CACHE_TTL_SECONDS; + WORKSPACE_ERROR_HANDLER_CACHE.insert( + w_id.clone(), + (row.0.clone(), row.1.clone(), row.2, row.3, expiry), + ); + row + }; if is_canceled && error_handler_muted_on_cancel { return Ok(()); @@ -1741,19 +1761,15 @@ pub async fn send_success_to_workspace_handler<'a, 'c, T: Serialize + Send + Syn } else { let row = fetch_success_handler_from_db(db, w_id).await?; let expiry = now + WORKSPACE_HANDLER_CACHE_TTL_SECONDS; - WORKSPACE_SUCCESS_HANDLER_CACHE.insert( - w_id.clone(), - (row.0.clone(), row.1.clone(), expiry), - ); + WORKSPACE_SUCCESS_HANDLER_CACHE + .insert(w_id.clone(), (row.0.clone(), row.1.clone(), expiry)); row } } else { let row = fetch_success_handler_from_db(db, w_id).await?; let expiry = now + WORKSPACE_HANDLER_CACHE_TTL_SECONDS; - WORKSPACE_SUCCESS_HANDLER_CACHE.insert( - w_id.clone(), - (row.0.clone(), row.1.clone(), expiry), - ); + WORKSPACE_SUCCESS_HANDLER_CACHE + .insert(w_id.clone(), (row.0.clone(), row.1.clone(), expiry)); row }; @@ -1811,23 +1827,22 @@ pub async fn try_schedule_next_job<'c>( &job.workspace_id ); - 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 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 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())); + push_err = Some(Error::internal_err( + "failpoint: savepoint create".to_string(), + )); } if push_err.is_none() { @@ -1852,25 +1867,39 @@ pub async fn try_schedule_next_job<'c>( )), }; #[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 - }; + 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); + 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())); + push_err = Some(Error::internal_err( + "failpoint: savepoint commit".to_string(), + )); } else { match savepoint.commit().await { Ok(()) => {} @@ -1895,9 +1924,7 @@ pub async fn try_schedule_next_job<'c>( } } Err(e) => { - tracing::error!( - "Could not create savepoint for schedule push: {e:#}", - ); + tracing::error!("Could not create savepoint for schedule push: {e:#}",); push_err = Some(Error::internal_err(format!( "Could not create savepoint: {e:#}" ))); @@ -1920,8 +1947,12 @@ pub async fn try_schedule_next_job<'c>( .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())) + let disable_result = if schedule_failpoints::is_active( + schedule_failpoints::ScheduleFailPoint::ScheduleDisable, + ) { + Err(sqlx::Error::Protocol( + "failpoint: schedule disable".to_string(), + )) } else { disable_result }; @@ -2162,7 +2193,10 @@ pub async fn push_success_handler<'a, 'c, T: Serialize + Send + Sync>( on_behalf_of.permissioned_as.clone(), ) } else { - (SUCCESS_HANDLER_USER_EMAIL, SUCCESS_HANDLER_USER_GROUP.to_string()) + ( + SUCCESS_HANDLER_USER_EMAIL, + SUCCESS_HANDLER_USER_GROUP.to_string(), + ) }; let tx = PushIsolationLevel::IsolatedRoot(db.clone()); @@ -2845,9 +2879,12 @@ impl PulledJobResult { }; let DebouncingSettings { debounce_delay_s, debounce_args_to_accumulate, .. } = - windmill_common::runnable_settings::prefetch_cached_from_handle(j.runnable_settings_handle, db) - .await? - .0; + windmill_common::runnable_settings::prefetch_cached_from_handle( + j.runnable_settings_handle, + db, + ) + .await? + .0; let (kind, j_id) = (j.kind, j.id); let is_djob_to_debounce = kind.is_dependency() @@ -3136,10 +3173,17 @@ pub async fn pull( #[cfg(feature = "private")] let concurrency_settings = if let Some(ref j) = job { - windmill_common::runnable_settings::prefetch_cached_from_handle(j.runnable_settings_handle, db) - .await? - .1 - .maybe_fallback(None, j.concurrent_limit, j.concurrency_time_window_s) + windmill_common::runnable_settings::prefetch_cached_from_handle( + j.runnable_settings_handle, + db, + ) + .await? + .1 + .maybe_fallback( + None, + j.concurrent_limit, + j.concurrency_time_window_s, + ) } else { Default::default() }; @@ -3203,11 +3247,13 @@ pub async fn pull( }); }; - let concurrency_settings = - windmill_common::runnable_settings::prefetch_cached_from_handle(job.runnable_settings_handle, db) - .await? - .1 - .maybe_fallback(None, job.concurrent_limit, job.concurrency_time_window_s); + let concurrency_settings = windmill_common::runnable_settings::prefetch_cached_from_handle( + job.runnable_settings_handle, + db, + ) + .await? + .1 + .maybe_fallback(None, job.concurrent_limit, job.concurrency_time_window_s); let has_concurent_limit = concurrency_settings.concurrent_limit.is_some(); @@ -4241,8 +4287,74 @@ async fn check_usage_limits( #[cfg(feature = "cloud")] use crate::cloud_usage::increment_usage_async; -// #[instrument(level = "trace", skip_all)] +// Thin wrapper that boxes the future to reduce async state machine sizes in callers. +// Without this, the ~13KB future of push_inner is inlined into every caller's state machine, +// causing stack overflows in deeply nested async call chains (e.g. flow execution). pub async fn push<'c, 'd>( + _db: &Pool, + tx: PushIsolationLevel<'c>, + workspace_id: &str, + job_payload: JobPayload, + args: PushArgs<'d>, + user: &str, + email: &str, + permissioned_as: String, + token_prefix: Option<&str>, + scheduled_for_o: Option>, + schedule_path: Option, + parent_job: Option, + root_job: Option, + flow_innermost_root_job: Option, + job_id: Option, + _is_flow_step: bool, + same_worker: bool, + pre_run_error: Option<&windmill_common::error::Error>, + visible_to_owner: bool, + tag: Option, + custom_timeout: Option, + flow_step_id: Option, + _priority_override: Option, + authed: Option<&Authed>, + running: bool, + end_user_email: Option, + trigger: Option, + suspended_mode: Option, +) -> Result<(Uuid, Transaction<'c, Postgres>), Error> { + Box::pin(push_inner( + _db, + tx, + workspace_id, + job_payload, + args, + user, + email, + permissioned_as, + token_prefix, + scheduled_for_o, + schedule_path, + parent_job, + root_job, + flow_innermost_root_job, + job_id, + _is_flow_step, + same_worker, + pre_run_error, + visible_to_owner, + tag, + custom_timeout, + flow_step_id, + _priority_override, + authed, + running, + end_user_email, + trigger, + suspended_mode, + )) + .await +} + +// #[instrument(level = "trace", skip_all)] +async fn push_inner<'c, 'd>( _db: &Pool, mut tx: PushIsolationLevel<'c>, workspace_id: &str, @@ -5367,10 +5479,13 @@ pub async fn push<'c, 'd>( (job_kind, scheduled_for_o) }; - let runnable_settings_handle = windmill_common::runnable_settings::insert_rs(RunnableSettings { - debouncing_settings: debouncing_settings.insert_cached(_db).await?, - concurrency_settings: concurrency_settings.insert_cached(_db).await?, - }, _db) + let runnable_settings_handle = windmill_common::runnable_settings::insert_rs( + RunnableSettings { + debouncing_settings: debouncing_settings.insert_cached(_db).await?, + concurrency_settings: concurrency_settings.insert_cached(_db).await?, + }, + _db, + ) .await?; let (guarded_concurrent_limit, guarded_concurrency_time_window_s) =