From bbb397b6ad954052f0bd33cc4ff8897eed66e4db Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 7 Feb 2026 14:16:32 +0100 Subject: [PATCH] fix: improve scheduling reliability in extreme pool contention conditions (#7825) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * 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 * test: add failpoint markers to try_schedule_next_job Co-Authored-By: Claude Opus 4.6 * chore: remove plan.md Co-Authored-By: Claude Opus 4.6 * 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 * 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 * 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 * 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 --------- Co-authored-by: Claude Opus 4.6 --- backend/Cargo.toml | 2 +- backend/tests/schedule_push.rs | 1072 +++++++++++++++++++- backend/windmill-queue/Cargo.toml | 1 + backend/windmill-queue/src/jobs.rs | 260 +++-- backend/windmill-worker/src/worker.rs | 19 +- backend/windmill-worker/src/worker_flow.rs | 99 +- 6 files changed, 1318 insertions(+), 135 deletions(-) diff --git a/backend/Cargo.toml b/backend/Cargo.toml index e03ac49e1d..40063f2cda 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -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 diff --git a/backend/tests/schedule_push.rs b/backend/tests/schedule_push.rs index 53c8711e82..b26336b637 100644 --- a/backend/tests/schedule_push.rs +++ b/backend/tests/schedule_push.rs @@ -8,7 +8,7 @@ mod schedule_push { use windmill_common::schedule::Schedule; use windmill_common::scripts::ScriptHash; use windmill_common::users::username_to_permissioned_as; - use windmill_queue::jobs::{handle_maybe_scheduled_job, MiniCompletedJob}; + use windmill_queue::jobs::{try_schedule_next_job, MiniCompletedJob}; use windmill_queue::schedule::push_scheduled_job; fn make_schedule(overrides: impl FnOnce(&mut Schedule)) -> Schedule { @@ -442,7 +442,7 @@ mod schedule_push { } // ----------------------------------------------------------------------- - // handle_maybe_scheduled_job: disabled schedule does not push + // try_schedule_next_job: disabled schedule does not push // ----------------------------------------------------------------------- #[sqlx::test(fixtures("base", "schedule_push"))] @@ -452,21 +452,23 @@ mod schedule_push { }); let job = make_completed_job(&schedule); - let result = handle_maybe_scheduled_job( + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( &db, + tx, &job, &schedule, &schedule.script_path, - "test-workspace", ) .await; - assert!(result.is_ok()); + tx.commit().await?; + assert!(err.is_none()); assert_eq!(count_queued_jobs(&db).await, 0); Ok(()) } // ----------------------------------------------------------------------- - // handle_maybe_scheduled_job: script path mismatch does not push + // try_schedule_next_job: script path mismatch does not push // ----------------------------------------------------------------------- #[sqlx::test(fixtures("base", "schedule_push"))] @@ -474,21 +476,23 @@ mod schedule_push { let schedule = make_schedule(|_| {}); let job = make_completed_job(&schedule); - let result = handle_maybe_scheduled_job( + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( &db, + tx, &job, &schedule, "f/system/different_script", - "test-workspace", ) .await; - assert!(result.is_ok()); + tx.commit().await?; + assert!(err.is_none()); assert_eq!(count_queued_jobs(&db).await, 0); Ok(()) } // ----------------------------------------------------------------------- - // handle_maybe_scheduled_job: enabled + matching path pushes next job + // try_schedule_next_job: enabled + matching path pushes next job // ----------------------------------------------------------------------- #[sqlx::test(fixtures("base", "schedule_push"))] @@ -496,15 +500,17 @@ mod schedule_push { let schedule = make_schedule(|_| {}); let job = make_completed_job(&schedule); - let result = handle_maybe_scheduled_job( + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( &db, + tx, &job, &schedule, &schedule.script_path, - "test-workspace", ) .await; - assert!(result.is_ok()); + tx.commit().await?; + assert!(err.is_none()); assert_eq!(count_queued_jobs(&db).await, 1); let (_, path, trigger, trigger_kind) = get_queued_job(&db).await.unwrap(); @@ -515,7 +521,7 @@ mod schedule_push { } // ----------------------------------------------------------------------- - // handle_maybe_scheduled_job: on_behalf_of_email via handle path + // try_schedule_next_job: on_behalf_of_email via handle path // ----------------------------------------------------------------------- #[sqlx::test(fixtures("base", "schedule_push"))] @@ -526,15 +532,17 @@ mod schedule_push { }); let job = make_completed_job(&schedule); - let result = handle_maybe_scheduled_job( + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( &db, + tx, &job, &schedule, &schedule.script_path, - "test-workspace", ) .await; - assert!(result.is_ok()); + tx.commit().await?; + assert!(err.is_none()); assert_eq!(count_queued_jobs(&db).await, 1); let email = sqlx::query_scalar::<_, String>( @@ -547,12 +555,12 @@ mod schedule_push { } // ----------------------------------------------------------------------- - // handle_maybe_scheduled_job: push failure disables schedule + // try_schedule_next_job: push failure returns error to caller + // (caller is responsible for retry + eventual disable) // ----------------------------------------------------------------------- #[sqlx::test(fixtures("base", "schedule_push"))] async fn test_handle_push_failure_disables_schedule(db: Pool) -> anyhow::Result<()> { - // Insert a schedule row so handle_maybe_scheduled_job can disable it sqlx::query( "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) VALUES ('test-workspace', 'f/system/bad_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/nonexistent', false, 'test@windmill.dev', '{}', false, false)" @@ -566,19 +574,21 @@ mod schedule_push { }); let job = make_completed_job(&schedule); - let result = handle_maybe_scheduled_job( + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( &db, + tx, &job, &schedule, &schedule.script_path, - "test-workspace", ) .await; - // Should succeed (error is handled internally by disabling schedule) - assert!(result.is_ok()); + // NotFound: schedule disabled internally, no error returned (caller commits) + assert!(err.is_none()); + tx.commit().await?; assert_eq!(count_queued_jobs(&db).await, 0); - // Schedule should be disabled with an error + // Schedule should be disabled (NotFound is non-retryable) let (enabled, error): (bool, Option) = sqlx::query_as( "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'", ) @@ -588,4 +598,1018 @@ mod schedule_push { assert!(error.is_some()); Ok(()) } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: successful push is atomic with tx commit + // If the caller commits, both the next tick and any prior writes persist. + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_success_atomic_with_commit(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/test_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/test_script', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|_| {}); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + assert!(err.is_none()); + tx.commit().await?; + + // Next tick was pushed + assert_eq!(count_queued_jobs(&db).await, 1); + + // Schedule stayed enabled + let enabled: bool = sqlx::query_scalar( + "SELECT enabled FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/test_schedule'", + ) + .fetch_one(&db) + .await?; + assert!(enabled); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: successful push rolls back if tx is dropped + // Ensures no next tick leaks when the outer tx is not committed. + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_success_rolls_back_on_drop(db: Pool) -> anyhow::Result<()> { + let schedule = make_schedule(|_| {}); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + assert!(err.is_none()); + + // Intentionally drop tx without committing (simulates caller failure) + drop(tx); + + // Nothing should be visible — the next tick must NOT leak + assert_eq!(count_queued_jobs(&db).await, 0); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: schedule disable rolls back if tx is dropped + // The schedule must stay enabled when the caller doesn't commit. + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_failure_disable_rolls_back_on_drop(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/bad_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/nonexistent', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|s| { + s.path = "f/system/bad_schedule".to_string(); + s.script_path = "f/system/nonexistent".to_string(); + }); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + // NotFound: schedule disabled in tx, no error returned + assert!(err.is_none()); + + // Drop without commit — simulates zombie retry path + drop(tx); + + // Schedule should STILL be enabled (disable was rolled back with tx) + let enabled: bool = sqlx::query_scalar( + "SELECT enabled FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'", + ) + .fetch_one(&db) + .await?; + assert!(enabled, "schedule must stay enabled when tx is rolled back"); + assert_eq!(count_queued_jobs(&db).await, 0); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: tx remains usable after successful push + // The caller can perform additional writes on the returned tx. + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_tx_usable_after_success(db: Pool) -> anyhow::Result<()> { + let schedule = make_schedule(|_| {}); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (mut tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + assert!(err.is_none()); + + // Write something else on the same tx + sqlx::query("INSERT INTO global_settings (name, value) VALUES ('_test_after_push', '42'::jsonb)") + .execute(&mut *tx) + .await?; + tx.commit().await?; + + // Both the pushed job and the extra write should be visible + assert_eq!(count_queued_jobs(&db).await, 1); + let val: serde_json::Value = sqlx::query_scalar( + "SELECT value FROM global_settings WHERE name = '_test_after_push'", + ) + .fetch_one(&db) + .await?; + assert_eq!(val, serde_json::json!(42)); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: tx remains usable after push failure + disable + // The caller can still write on the returned tx after a schedule disable. + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_tx_usable_after_failure(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/bad_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/nonexistent', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|s| { + s.path = "f/system/bad_schedule".to_string(); + s.script_path = "f/system/nonexistent".to_string(); + }); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (mut tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + // NotFound: schedule disabled in tx, no error returned + assert!(err.is_none()); + + // Write something else on the returned tx — tx is still usable + sqlx::query("INSERT INTO global_settings (name, value) VALUES ('_test_after_fail', '99'::jsonb)") + .execute(&mut *tx) + .await?; + tx.commit().await?; + + // Schedule should be disabled (NotFound is non-retryable) + let (enabled, error): (bool, Option) = sqlx::query_as( + "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'", + ) + .fetch_one(&db) + .await?; + assert!(!enabled); + assert!(error.is_some()); + + // Extra write should still be committed (tx is usable after push failure) + let val: serde_json::Value = sqlx::query_scalar( + "SELECT value FROM global_settings WHERE name = '_test_after_fail'", + ) + .fetch_one(&db) + .await?; + assert_eq!(val, serde_json::json!(99)); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: flow schedule pushes next job + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_try_schedule_flow(db: Pool) -> anyhow::Result<()> { + let schedule = make_schedule(|s| { + s.is_flow = true; + s.script_path = "f/system/test_flow".to_string(); + s.path = "f/system/flow_schedule".to_string(); + }); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + tx.commit().await?; + assert!(err.is_none()); + assert_eq!(count_queued_jobs(&db).await, 1); + + let (_, path, trigger, _) = get_queued_job(&db).await.unwrap(); + assert_eq!(path.as_deref(), Some("f/system/test_flow")); + assert_eq!(trigger.as_deref(), Some("f/system/flow_schedule")); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: script with retry wraps in SingleStepFlow + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_try_schedule_with_retry(db: Pool) -> anyhow::Result<()> { + let schedule = make_schedule(|s| { + s.retry = Some(serde_json::json!({ + "constant": { "attempts": 3, "seconds": 10 } + })); + }); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + tx.commit().await?; + assert!(err.is_none()); + assert_eq!(count_queued_jobs(&db).await, 1); + + let kind = sqlx::query_scalar::<_, String>( + "SELECT kind::text FROM v2_job j JOIN v2_job_queue q ON j.id = q.id LIMIT 1", + ) + .fetch_one(&db) + .await?; + assert_eq!(kind, "singlestepflow"); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: push failure error message is stored on schedule + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_push_failure_stores_error_message(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/bad_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/nonexistent', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|s| { + s.path = "f/system/bad_schedule".to_string(); + s.script_path = "f/system/nonexistent".to_string(); + }); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + // NotFound: schedule disabled in tx, no error returned (caller commits) + assert!(err.is_none()); + tx.commit().await?; + + let error: String = sqlx::query_scalar( + "SELECT error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'", + ) + .fetch_one(&db) + .await?; + // Error should mention the script that couldn't be found + assert!( + error.contains("nonexistent"), + "error message should describe the failure, got: {error}" + ); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: disabled schedule leaves no side effects + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_disabled_schedule_no_side_effects(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/disabled_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', false, 'f/system/test_script', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|s| { + s.path = "f/system/disabled_schedule".to_string(); + s.enabled = false; + }); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + tx.commit().await?; + assert!(err.is_none()); + + // Schedule should remain disabled (not re-enabled) and no error set + let (enabled, error): (bool, Option) = sqlx::query_as( + "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/disabled_schedule'", + ) + .fetch_one(&db) + .await?; + assert!(!enabled); + assert!(error.is_none(), "disabled schedule should not get an error set"); + assert_eq!(count_queued_jobs(&db).await, 0); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: path mismatch leaves schedule unchanged + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_path_mismatch_no_side_effects(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/test_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/test_script', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|_| {}); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + "f/system/different_script", + ) + .await; + tx.commit().await?; + assert!(err.is_none()); + + // Schedule should remain enabled, no error, no jobs pushed + let (enabled, error): (bool, Option) = sqlx::query_as( + "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/test_schedule'", + ) + .fetch_one(&db) + .await?; + assert!(enabled, "schedule must stay enabled on path mismatch"); + assert!(error.is_none(), "no error should be set on path mismatch"); + assert_eq!(count_queued_jobs(&db).await, 0); + Ok(()) + } + + // ----------------------------------------------------------------------- + // try_schedule_next_job: push failure with schedule not in DB + // When the schedule row doesn't exist, the UPDATE affects 0 rows but + // doesn't error. The function should return (tx, None). + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_push_failure_schedule_not_in_db(db: Pool) -> anyhow::Result<()> { + // Do NOT insert a schedule row — the disable UPDATE will match 0 rows + let schedule = make_schedule(|s| { + s.path = "f/system/ghost_schedule".to_string(); + s.script_path = "f/system/nonexistent".to_string(); + }); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + drop(tx); + // NotFound: disable succeeds (UPDATE 0 rows is not an error), no error returned + assert!(err.is_none()); + assert_eq!(count_queued_jobs(&db).await, 0); + Ok(()) + } + + // ----------------------------------------------------------------------- + // Critical invariant: after commit, it's impossible to have schedule + // enabled + no next tick + function returned success. + // We verify: if push succeeds, both the job AND the schedule's + // unmodified enabled state are committed together. + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_invariant_success_means_tick_committed(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/test_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/test_script', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|_| {}); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + assert!(err.is_none()); + tx.commit().await?; + + // After commit: schedule enabled AND next tick exists — invariant holds + let enabled: bool = sqlx::query_scalar( + "SELECT enabled FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/test_schedule'", + ) + .fetch_one(&db) + .await?; + assert!(enabled, "schedule must be enabled after successful push"); + assert_eq!( + count_queued_jobs(&db).await, + 1, + "next tick must exist after successful push + commit" + ); + Ok(()) + } + + // ----------------------------------------------------------------------- + // Critical invariant: after commit with push failure, the schedule is + // disabled. It's never the case that we commit with the schedule still + // enabled and no next tick. + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_invariant_failure_means_disabled_after_commit( + db: Pool, + ) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/bad_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/nonexistent', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|s| { + s.path = "f/system/bad_schedule".to_string(); + s.script_path = "f/system/nonexistent".to_string(); + }); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + // NotFound: schedule disabled in tx, no error returned (caller commits) + assert!(err.is_none()); + tx.commit().await?; + + // After commit: no next tick, but schedule is disabled — invariant holds + assert_eq!(count_queued_jobs(&db).await, 0); + let enabled: bool = sqlx::query_scalar( + "SELECT enabled FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'", + ) + .fetch_one(&db) + .await?; + assert!( + !enabled, + "schedule must be disabled when push fails with NotFound and tx commits" + ); + Ok(()) + } + + // ----------------------------------------------------------------------- + // Critical invariant: if tx is NOT committed (zombie path), neither + // the next tick nor the schedule disable persists. The schedule stays + // enabled so that zombie retry can re-attempt. + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_invariant_rollback_preserves_schedule_for_retry( + db: Pool, + ) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/bad_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/nonexistent', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|s| { + s.path = "f/system/bad_schedule".to_string(); + s.script_path = "f/system/nonexistent".to_string(); + }); + let job = make_completed_job(&schedule); + + let tx = db.begin().await?; + let (tx, err) = try_schedule_next_job( + &db, + tx, + &job, + &schedule, + &schedule.script_path, + ) + .await; + // NotFound: schedule disabled in tx, no error returned + assert!(err.is_none()); + + // Simulate zombie path: drop tx without commit + drop(tx); + + // Schedule still enabled (disable was rolled back with tx) — ready for retry + let (enabled, error): (bool, Option) = sqlx::query_as( + "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'", + ) + .fetch_one(&db) + .await?; + assert!(enabled, "schedule must remain enabled after rollback"); + assert!(error.is_none(), "error must not persist after rollback"); + assert_eq!(count_queued_jobs(&db).await, 0); + Ok(()) + } + + // =================================================================== + // Failpoint tests — feature-gated, only compiled under `failpoints` + // =================================================================== + + mod failpoint_tests { + use super::*; + use windmill_queue::jobs::schedule_failpoints::{ScheduleFailPoint, ACTIVE}; + + // --------------------------------------------------------------- + // SavepointCreate failpoint → schedule disabled, 0 jobs + // --------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_failpoint_savepoint_create_disables(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/test_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/test_script', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|_| {}); + let job = make_completed_job(&schedule); + + ACTIVE.scope(ScheduleFailPoint::SavepointCreate, async { + let tx = db.begin().await.unwrap(); + let (tx, err) = try_schedule_next_job( + &db, tx, &job, &schedule, &schedule.script_path, + ).await; + // Transient error returned to caller for retry + assert!(err.is_some()); + drop(tx); + + assert_eq!(count_queued_jobs(&db).await, 0); + let enabled: bool = sqlx::query_scalar( + "SELECT enabled FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/test_schedule'", + ) + .fetch_one(&db) + .await + .unwrap(); + assert!(enabled, "schedule must stay enabled for caller retry"); + }).await; + Ok(()) + } + + // --------------------------------------------------------------- + // Push failpoint → schedule disabled, 0 jobs + // --------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_failpoint_push_disables(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/test_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/test_script', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|_| {}); + let job = make_completed_job(&schedule); + + ACTIVE.scope(ScheduleFailPoint::Push, async { + let tx = db.begin().await.unwrap(); + let (tx, err) = try_schedule_next_job( + &db, tx, &job, &schedule, &schedule.script_path, + ).await; + // Transient error returned to caller for retry + assert!(err.is_some()); + drop(tx); + + assert_eq!(count_queued_jobs(&db).await, 0); + let enabled: bool = sqlx::query_scalar( + "SELECT enabled FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/test_schedule'", + ) + .fetch_one(&db) + .await + .unwrap(); + assert!(enabled, "schedule must stay enabled for caller retry"); + }).await; + Ok(()) + } + + // --------------------------------------------------------------- + // SavepointCommit failpoint → schedule disabled, 0 jobs + // --------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_failpoint_savepoint_commit_disables(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/test_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/test_script', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|_| {}); + let job = make_completed_job(&schedule); + + ACTIVE.scope(ScheduleFailPoint::SavepointCommit, async { + let tx = db.begin().await.unwrap(); + let (tx, err) = try_schedule_next_job( + &db, tx, &job, &schedule, &schedule.script_path, + ).await; + // Transient error returned to caller for retry + assert!(err.is_some()); + drop(tx); + + assert_eq!(count_queued_jobs(&db).await, 0, "pushed job must be rolled back"); + let enabled: bool = sqlx::query_scalar( + "SELECT enabled FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/test_schedule'", + ) + .fetch_one(&db) + .await + .unwrap(); + assert!(enabled, "schedule must stay enabled for caller retry"); + }).await; + Ok(()) + } + + // --------------------------------------------------------------- + // ScheduleDisable failpoint → returns Some(err), caller doesn't commit + // --------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_failpoint_schedule_disable_returns_err(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/bad_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/nonexistent', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|s| { + s.path = "f/system/bad_schedule".to_string(); + s.script_path = "f/system/nonexistent".to_string(); + }); + let job = make_completed_job(&schedule); + + ACTIVE.scope(ScheduleFailPoint::ScheduleDisable, async { + let tx = db.begin().await.unwrap(); + let (_tx, err) = try_schedule_next_job( + &db, tx, &job, &schedule, &schedule.script_path, + ).await; + assert!(err.is_some(), "must return Some(err) when disable fails"); + }).await; + Ok(()) + } + + // --------------------------------------------------------------- + // ScheduleDisable failpoint + tx drop → schedule stays enabled + // --------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_failpoint_disable_failure_rollback(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/bad_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/nonexistent', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|s| { + s.path = "f/system/bad_schedule".to_string(); + s.script_path = "f/system/nonexistent".to_string(); + }); + let job = make_completed_job(&schedule); + + ACTIVE.scope(ScheduleFailPoint::ScheduleDisable, async { + let tx = db.begin().await.unwrap(); + let (tx, err) = try_schedule_next_job( + &db, tx, &job, &schedule, &schedule.script_path, + ).await; + assert!(err.is_some(), "must return Some(err) when disable fails"); + drop(tx); + + let (enabled, error): (bool, Option) = sqlx::query_as( + "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/bad_schedule'", + ) + .fetch_one(&db) + .await + .unwrap(); + assert!(enabled, "schedule must stay enabled when tx is dropped after disable failure"); + assert!(error.is_none(), "error must not persist after rollback"); + }).await; + Ok(()) + } + + // --------------------------------------------------------------- + // PushQuotaExceeded failpoint (script) → schedule disabled, 0 jobs, + // no error handler notification (QuotaExceeded is silenced) + // --------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_failpoint_push_quota_exceeded_script(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/test_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/test_script', false, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|_| {}); + let job = make_completed_job(&schedule); + + ACTIVE.scope(ScheduleFailPoint::PushQuotaExceeded, async { + let tx = db.begin().await.unwrap(); + let (tx, err) = try_schedule_next_job( + &db, tx, &job, &schedule, &schedule.script_path, + ).await; + // QuotaExceeded: schedule disabled internally, no error returned + assert!(err.is_none(), "QuotaExceeded should be handled internally (returns None)"); + tx.commit().await.unwrap(); + + assert_eq!(count_queued_jobs(&db).await, 0); + let (enabled, error): (bool, Option) = sqlx::query_as( + "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/test_schedule'", + ) + .fetch_one(&db) + .await + .unwrap(); + assert!(!enabled, "schedule must be disabled after QuotaExceeded"); + assert!(error.is_some(), "error must be set on schedule"); + assert!(error.unwrap().contains("quota"), "error message should mention quota"); + }).await; + Ok(()) + } + + // --------------------------------------------------------------- + // PushQuotaExceeded failpoint (flow) → schedule disabled, 0 jobs + // --------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_failpoint_push_quota_exceeded_flow(db: Pool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/flow_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/test_flow', true, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + let schedule = make_schedule(|s| { + s.is_flow = true; + s.script_path = "f/system/test_flow".to_string(); + s.path = "f/system/flow_schedule".to_string(); + }); + let job = make_completed_job(&schedule); + + ACTIVE.scope(ScheduleFailPoint::PushQuotaExceeded, async { + let tx = db.begin().await.unwrap(); + let (tx, err) = try_schedule_next_job( + &db, tx, &job, &schedule, &schedule.script_path, + ).await; + // QuotaExceeded: schedule disabled internally, no error returned + assert!(err.is_none(), "QuotaExceeded should be handled internally (returns None)"); + tx.commit().await.unwrap(); + + assert_eq!(count_queued_jobs(&db).await, 0); + let (enabled, error): (bool, Option) = sqlx::query_as( + "SELECT enabled, error FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/flow_schedule'", + ) + .fetch_one(&db) + .await + .unwrap(); + assert!(!enabled, "flow schedule must be disabled after QuotaExceeded"); + assert!(error.is_some(), "error must be set on flow schedule"); + assert!(error.unwrap().contains("quota"), "error message should mention quota"); + }).await; + Ok(()) + } + } + + // =================================================================== + // Zombie detection tests — verify that a flow left in queue after + // SchedulePushZombieError meets the restart criteria in monitor.rs + // =================================================================== + + // ----------------------------------------------------------------------- + // When both schedule push AND post-retry disable fail, the flow job stays + // in v2_job_queue as a zombie. This test simulates that state and verifies: + // + // 1. The zombie detection query (from handle_zombie_flows) finds the flow + // 2. The flow meets restart criteria: first module is WaitingForPriorSteps + // and same_worker is false + // 3. After the restart UPDATE, the flow is re-queued (running=false) + // 4. The schedule remains enabled for retry on next flow execution + // ----------------------------------------------------------------------- + + #[sqlx::test(fixtures("base", "schedule_push"))] + async fn test_zombie_flow_after_schedule_push_failure_meets_restart_criteria( + db: Pool, + ) -> anyhow::Result<()> { + use windmill_common::flow_status::{FlowStatus, FlowStatusModule}; + + let flow_job_id = uuid::Uuid::new_v4(); + let now = Utc::now(); + let stale_ping = now - chrono::Duration::minutes(5); + + // Schedule still enabled — simulates both push and disable failing + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, edited_at, schedule, timezone, enabled, script_path, is_flow, email, extra_perms, ws_error_handler_muted, no_flow_overlap) + VALUES ('test-workspace', 'f/system/test_schedule', 'test-user', now(), '0 0 */5 * * *', 'UTC', true, 'f/system/test_flow', true, 'test@windmill.dev', '{}', false, false)" + ) + .execute(&db) + .await?; + + // Flow job in v2_job (kind=flow, triggered by schedule, same_worker=false) + sqlx::query( + "INSERT INTO v2_job (id, workspace_id, created_at, created_by, permissioned_as, permissioned_as_email, kind, runnable_path, trigger, trigger_kind, same_worker, visible_to_owner, tag) + VALUES ($1, 'test-workspace', $2, 'test-user', 'u/test-user', 'test@windmill.dev', 'flow', 'f/system/test_flow', 'f/system/test_schedule', 'schedule', false, false, 'flow')" + ) + .bind(flow_job_id) + .bind(now - chrono::Duration::minutes(5)) + .execute(&db) + .await?; + + // Queue entry: running=true (worker caught SchedulePushZombieError and returned Ok) + sqlx::query( + "INSERT INTO v2_job_queue (id, workspace_id, created_at, scheduled_for, running, started_at, tag, suspend) + VALUES ($1, 'test-workspace', $2, $2, true, $3, 'flow', 0)" + ) + .bind(flow_job_id) + .bind(now - chrono::Duration::minutes(5)) + .bind(now - chrono::Duration::minutes(4)) + .execute(&db) + .await?; + + // Stale ping — older than the 60s zombie transition timeout + sqlx::query("INSERT INTO v2_job_runtime (id, ping) VALUES ($1, $2)") + .bind(flow_job_id) + .bind(stale_ping) + .execute(&db) + .await?; + + // Flow status at step 0, first module = WaitingForPriorSteps + // This is the initial state of a flow that hasn't started any steps yet + let flow_status = serde_json::json!({ + "step": 0, + "modules": [{"type": "WaitingForPriorSteps", "id": "a"}], + "failure_module": {"type": "WaitingForPriorSteps", "id": "failure"}, + "retry": {"fail_count": 0, "failed_jobs": []}, + "cleanup_module": {"flow_jobs_to_clean": []} + }); + + sqlx::query("INSERT INTO v2_job_status (id, flow_status) VALUES ($1, $2::jsonb)") + .bind(flow_job_id) + .bind(&flow_status) + .execute(&db) + .await?; + + // Run the same zombie detection query from handle_zombie_flows (60s timeout) + let zombie_flows = sqlx::query_as::<_, (uuid::Uuid, String, Option, Option)>( + r#" + SELECT + j.id, j.workspace_id, j.same_worker, + COALESCE(s.flow_status, s.workflow_as_code_status)::text AS flow_status + FROM v2_job_queue q + JOIN v2_job j USING (id) + LEFT JOIN v2_job_runtime r USING (id) + LEFT JOIN v2_job_status s USING (id) + WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null + AND q.scheduled_for <= now() + AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode') + AND r.ping IS NOT NULL + AND r.ping < NOW() - ('60' || ' seconds')::interval + AND q.canceled_by IS NULL + "#, + ) + .fetch_all(&db) + .await?; + + assert_eq!(zombie_flows.len(), 1, "zombie flow must be detected"); + let (id, _ws, same_worker, flow_status_json) = &zombie_flows[0]; + assert_eq!(*id, flow_job_id); + + // Replicate the exact branching logic from handle_zombie_flows (monitor.rs:2711-2754). + // Only flows matching the restart condition get restarted; others are cancelled. + let status = flow_status_json + .as_deref() + .and_then(|x| serde_json::from_str::(x).ok()); + let should_restart = !same_worker.unwrap_or(false) + && status.is_some_and(|s| { + s.modules + .get(0) + .is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. })) + }); + + assert!( + should_restart, + "flow must match the restart branch (not the cancel branch) in handle_zombie_flows" + ); + + // Apply the restart action — same UPDATE as handle_zombie_flows + sqlx::query( + "UPDATE v2_job_queue SET running = false, started_at = null + WHERE id = $1 AND canceled_by IS NULL", + ) + .bind(flow_job_id) + .execute(&db) + .await?; + + // Flow is re-queued for processing + let (running, started_at): (bool, Option>) = sqlx::query_as( + "SELECT running, started_at FROM v2_job_queue WHERE id = $1", + ) + .bind(flow_job_id) + .fetch_one(&db) + .await?; + assert!(!running, "flow must not be running after zombie restart"); + assert!(started_at.is_none(), "started_at must be null after zombie restart"); + + // Schedule still enabled — will be retried when flow re-executes + let enabled: bool = sqlx::query_scalar( + "SELECT enabled FROM schedule WHERE workspace_id = 'test-workspace' AND path = 'f/system/test_schedule'", + ) + .fetch_one(&db) + .await?; + assert!(enabled, "schedule must remain enabled for retry after zombie restart"); + + // Flow job is NOT in v2_job_completed (it was never completed with error) + let completed_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM v2_job_completed WHERE id = $1", + ) + .bind(flow_job_id) + .fetch_one(&db) + .await?; + assert_eq!(completed_count, 0, "flow must not be in completed_job — it's a zombie, not an error"); + + Ok(()) + } } diff --git a/backend/windmill-queue/Cargo.toml b/backend/windmill-queue/Cargo.toml index 16178502fe..ad485d33d8 100644 --- a/backend/windmill-queue/Cargo.toml +++ b/backend/windmill-queue/Cargo.toml @@ -14,6 +14,7 @@ private = [] enterprise = ["windmill-common/enterprise"] cloud = [] benchmark = ["windmill-common/benchmark"] +failpoints = [] prometheus = ["dep:prometheus"] smtp = [] diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 417b545f10..281d93a435 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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( .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( .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, + mut tx: Transaction<'c, Postgres>, job: &MiniCompletedJob, schedule: &Schedule, script_path: &str, - w_id: &str, -) -> windmill_common::error::Result<()> { +) -> (Transaction<'c, Postgres>, Option) { + 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"; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index f356a79b0b..982e74c25d 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -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::().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(), diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 42306fbfe5..d3f2a39eae 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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",