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",