From efbba24581663c44cea62027707b6ad74f36af18 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 25 Feb 2024 18:25:40 +0100 Subject: [PATCH] feat: more resilient flows in case of crash during transitions --- backend/.cargo/config.toml | 4 - ...4d6c44132758a6ff14f76485b7b089db25d47.json | 14 ++ ...eb6da8ceb45258ee5817ec71e73bebbe415bd.json | 15 ++ ...9c009214eeaee8fd27a0d3050998b354c45ff.json | 15 ++ ...369afe534519bf48f61f88f60786db870c24.json} | 4 +- ...25495e55df7f2ad839aea22d0286525eb823e.json | 14 ++ backend/Cargo.toml | 2 + ...225142058_make_last_ping_nullable.down.sql | 1 + ...40225142058_make_last_ping_nullable.up.sql | 3 + backend/src/main.rs | 22 ++- backend/src/monitor.rs | 107 +++++++++-- backend/windmill-queue/src/jobs.rs | 16 ++ backend/windmill-worker/Cargo.toml | 1 + backend/windmill-worker/src/worker.rs | 8 +- backend/windmill-worker/src/worker_flow.rs | 170 +++++++----------- .../src/lib/components/runs/JobLoader.svelte | 2 +- 16 files changed, 262 insertions(+), 136 deletions(-) create mode 100644 backend/.sqlx/query-019edf515dbc2c04960b29b84ee4d6c44132758a6ff14f76485b7b089db25d47.json create mode 100644 backend/.sqlx/query-45c9ecf8b1f8cbca7c75dab24a1eb6da8ceb45258ee5817ec71e73bebbe415bd.json create mode 100644 backend/.sqlx/query-631d4637e4137a0680ffa56e4639c009214eeaee8fd27a0d3050998b354c45ff.json rename backend/.sqlx/{query-003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91.json => query-7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24.json} (63%) create mode 100644 backend/.sqlx/query-c05be905e46c5b0a2186ba859a725495e55df7f2ad839aea22d0286525eb823e.json create mode 100644 backend/migrations/20240225142058_make_last_ping_nullable.down.sql create mode 100644 backend/migrations/20240225142058_make_last_ping_nullable.up.sql diff --git a/backend/.cargo/config.toml b/backend/.cargo/config.toml index 28362c5b25..7d27c4a1e1 100644 --- a/backend/.cargo/config.toml +++ b/backend/.cargo/config.toml @@ -1,8 +1,4 @@ [build] -rustflags = [ - "--cfg", - "tokio_unstable" -] incremental = true [target.x86_64-apple-darwin] diff --git a/backend/.sqlx/query-019edf515dbc2c04960b29b84ee4d6c44132758a6ff14f76485b7b089db25d47.json b/backend/.sqlx/query-019edf515dbc2c04960b29b84ee4d6c44132758a6ff14f76485b7b089db25d47.json new file mode 100644 index 0000000000..4b7036c985 --- /dev/null +++ b/backend/.sqlx/query-019edf515dbc2c04960b29b84ee4d6c44132758a6ff14f76485b7b089db25d47.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET last_ping = null\n WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "019edf515dbc2c04960b29b84ee4d6c44132758a6ff14f76485b7b089db25d47" +} diff --git a/backend/.sqlx/query-45c9ecf8b1f8cbca7c75dab24a1eb6da8ceb45258ee5817ec71e73bebbe415bd.json b/backend/.sqlx/query-45c9ecf8b1f8cbca7c75dab24a1eb6da8ceb45258ee5817ec71e73bebbe415bd.json new file mode 100644 index 0000000000..02677837d1 --- /dev/null +++ b/backend/.sqlx/query-45c9ecf8b1f8cbca7c75dab24a1eb6da8ceb45258ee5817ec71e73bebbe415bd.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue SET last_ping = now() WHERE id = $1 AND workspace_id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "45c9ecf8b1f8cbca7c75dab24a1eb6da8ceb45258ee5817ec71e73bebbe415bd" +} diff --git a/backend/.sqlx/query-631d4637e4137a0680ffa56e4639c009214eeaee8fd27a0d3050998b354c45ff.json b/backend/.sqlx/query-631d4637e4137a0680ffa56e4639c009214eeaee8fd27a0d3050998b354c45ff.json new file mode 100644 index 0000000000..d38bf22973 --- /dev/null +++ b/backend/.sqlx/query-631d4637e4137a0680ffa56e4639c009214eeaee8fd27a0d3050998b354c45ff.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue\n SET last_ping = null\n WHERE id = $1 AND last_ping = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Timestamptz" + ] + }, + "nullable": [] + }, + "hash": "631d4637e4137a0680ffa56e4639c009214eeaee8fd27a0d3050998b354c45ff" +} diff --git a/backend/.sqlx/query-003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91.json b/backend/.sqlx/query-7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24.json similarity index 63% rename from backend/.sqlx/query-003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91.json rename to backend/.sqlx/query-7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24.json index 00797f7524..45ddd29b01 100644 --- a/backend/.sqlx/query-003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91.json +++ b/backend/.sqlx/query-7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", + "query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' \n WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", "describe": { "columns": [ { @@ -30,5 +30,5 @@ false ] }, - "hash": "003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91" + "hash": "7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24" } diff --git a/backend/.sqlx/query-c05be905e46c5b0a2186ba859a725495e55df7f2ad839aea22d0286525eb823e.json b/backend/.sqlx/query-c05be905e46c5b0a2186ba859a725495e55df7f2ad839aea22d0286525eb823e.json new file mode 100644 index 0000000000..6679b7849f --- /dev/null +++ b/backend/.sqlx/query-c05be905e46c5b0a2186ba859a725495e55df7f2ad839aea22d0286525eb823e.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue SET running = false, started_at = null WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "c05be905e46c5b0a2186ba859a725495e55df7f2ad839aea22d0286525eb823e" +} diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 3cd992e4ef..32d0590b71 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -46,11 +46,13 @@ pg_embed = ["dep:pg-embed"] embedding = ["windmill-api/embedding"] parquet = ["windmill-api/parquet", "windmill-common/parquet", "windmill-worker/parquet"] prometheus = ["windmill-common/prometheus", "windmill-api/prometheus", "windmill-worker/prometheus", "windmill-queue/prometheus"] +flow_testing = ["windmill-worker/flow_testing"] [dependencies] anyhow.workspace = true tokio.workspace = true dotenv.workspace = true +windmill-queue.workspace = true windmill-common = { workspace = true, default-features = false } windmill-git-sync.workspace = true windmill-api = { workspace = true, default-features = false } diff --git a/backend/migrations/20240225142058_make_last_ping_nullable.down.sql b/backend/migrations/20240225142058_make_last_ping_nullable.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20240225142058_make_last_ping_nullable.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20240225142058_make_last_ping_nullable.up.sql b/backend/migrations/20240225142058_make_last_ping_nullable.up.sql new file mode 100644 index 0000000000..1a64a62d09 --- /dev/null +++ b/backend/migrations/20240225142058_make_last_ping_nullable.up.sql @@ -0,0 +1,3 @@ +-- Add up migration script here +ALTER TABLE queue +ALTER COLUMN last_ping DROP NOT NULL; \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index 6e86f6f2d0..dd3ae1202b 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -624,6 +624,7 @@ pub async fn run_workers().ok()) .unwrap_or_else(|| "30".to_string()); + static ref FLOW_ZOMBIE_TRANSITION_TIMEOUT: String = std::env::var("FLOW_ZOMBIE_TRANSITION_TIMEOUT") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or_else(|| "10".to_string()); + pub static ref RESTART_ZOMBIE_JOBS: bool = std::env::var("RESTART_ZOMBIE_JOBS") .ok() @@ -566,15 +566,21 @@ pub async fn monitor_pool(db: &DB) { } } -pub async fn monitor_db( +pub async fn monitor_db( db: &Pool, base_internal_url: &str, - rsmq: Option, + rsmq: Option, server_mode: bool, ) { let zombie_jobs_f = async { if server_mode { handle_zombie_jobs(db, base_internal_url, rsmq.clone(), "server").await; + match handle_zombie_flows(db, rsmq.clone()).await { + Err(err) => { + tracing::error!("Error handling zombie flows: {:?}", err); + } + _ => {} + } } }; let expired_items_f = async { @@ -766,7 +772,8 @@ async fn handle_zombie_jobs ) { if *RESTART_ZOMBIE_JOBS { let restarted = sqlx::query!( - "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval + "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' + WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", *ZOMBIE_JOB_TIMEOUT, ) @@ -779,8 +786,8 @@ async fn handle_zombie_jobs QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _); } for r in restarted { - tracing::info!( - "restarted zombie job {} {} {}", + tracing::error!( + "Zombie job detected, restarting it: {} {} {:?}", r.id, r.workspace_id, r.last_ping @@ -788,7 +795,8 @@ async fn handle_zombie_jobs } } - let mut timeout_query = "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow')".to_string(); + let mut timeout_query = "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval + AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow')".to_string(); if *RESTART_ZOMBIE_JOBS { timeout_query.push_str(" AND same_worker = true"); }; @@ -853,3 +861,70 @@ async fn handle_zombie_jobs .await; } } + + +async fn handle_zombie_flows( + db: &DB, + rsmq: Option, +) -> error::Result<()> { + + let flows = sqlx::query_as::<_, QueuedJob>( + r#" + SELECT * + FROM queue + WHERE running = true AND suspend = 0 AND scheduled_for <= now() AND (job_kind = 'flow' OR job_kind = 'flowpreview') + AND last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval + "#, + ).bind(FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str()) + .fetch_all(db) + .await?; + + // TODO: for now only log zombie flows + let mut tx = db.begin().await.unwrap(); + for flow in flows { + let status = flow.parse_flow_status(); + if status.is_some_and(|s| s.modules.get(0).is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. }))) + { + tracing::error!( + "Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.", + flow.id, + flow.workspace_id + ); + // if the flow hasn't started and is a zombie, we can simply restart it + sqlx::query!( + "UPDATE queue SET running = false, started_at = null WHERE id = $1", + flow.id + ) + .execute(db) + .await?; + } else { + // if it was started, we can't restart it, so we cancel it + tracing::error!( + "Zombie flow detected: {} in workspace {}. Cancelling it.", + flow.id, + flow.workspace_id + ); + let (mut ntx, _) = cancel_job( + "monitor", + Some("Flow cancelled as it was hanging in between 2 steps".to_string()), + flow.id, + flow.workspace_id.as_str(), + tx, + db, + rsmq.clone(), + false, + ) + .await?; + // if the flow hasn't started and is a zombie, we can simply restart it + sqlx::query!( + "UPDATE queue SET running = false, started_at = null WHERE id = $1", + flow.id + ) + .execute(&mut *ntx) + .await?; + tx = ntx; + } + } + tx.commit().await?; + Ok(()) +} \ No newline at end of file diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 2504cf032a..7decf303dd 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -564,6 +564,22 @@ pub async fn add_completed_job< tx = delete_job(tx, &queued_job.workspace_id, job_id).await?; // tracing::error!("3 {:?}", start.elapsed()); + if queued_job.is_flow_step + { + if let Some(parent_job) = queued_job.parent_job { + // persist the flow last progress timestamp to avoid zombie flow jobs + tracing::debug!("Persisting flow last progress timestamp to flow job: {:?}", parent_job); + sqlx::query!( + "UPDATE queue SET last_ping = now() WHERE id = $1 AND workspace_id = $2", + parent_job, + &queued_job.workspace_id + ) + .execute(&mut tx) + .await?; + } + + } + if !queued_job.is_flow_step && queued_job.schedule_path.is_some() && queued_job.script_path.is_some() diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index d0047a9922..cf6b25399d 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -15,6 +15,7 @@ enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "dep: benchmark = ["windmill-queue/benchmark"] flamegraph = [] parquet = ["windmill-common/parquet"] +flow_testing = [] [dependencies] windmill-queue.workspace = true diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index d25cf8608e..1e53d61a0b 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -2373,12 +2373,12 @@ pub async fn handle_job_error error::Result<()> { // this is manual tailrecursion because async_recursion blows up the stack // todo!(); + potentially_crash_for_testing(); + let mut rec = update_flow_status_after_job_completion_internal( db, client, @@ -92,6 +94,7 @@ pub async fn update_flow_status_after_job_completion< ) .await?; while let Some(nrec) = rec { + potentially_crash_for_testing(); rec = match update_flow_status_after_job_completion_internal( db, client, @@ -578,13 +581,13 @@ pub async fn update_flow_status_after_job_completion_internal< let done = if !should_continue_flow { let logs = if flow_job.canceled { - "Flow job canceled".to_string() + "Flow job canceled\n".to_string() } else if stop_early { - format!("Flow job stopped early because of a stop early predicate returning true") + format!("Flow job stopped early because of a stop early predicate returning true\n") } else if success { - "Flow job completed with success".to_string() + "Flow job completed with success\n".to_string() } else { - "Flow job completed with error".to_string() + "Flow job completed with error\n".to_string() }; #[cfg(feature = "enterprise")] @@ -1038,30 +1041,6 @@ pub async fn handle_flow( .parse_flow_status() .with_context(|| "Unable to parse flow status")?; - if !flow_job.is_flow_step - && flow_job.schedule_path.is_some() - && flow_job.script_path.is_some() - && status.step == 0 - { - let tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into(); - - match handle_maybe_scheduled_job( - tx, - db, - flow_job.schedule_path.as_ref().unwrap(), - flow_job.script_path.as_ref().unwrap(), - &flow_job.workspace_id, - ) - .await - { - Ok(tx) => { - tx.commit().await?; - } - Err(e) => { - tracing::error!("Error during handle_maybe_scheduled_job: {e}"); - } - } - } push_next_flow_job( flow_job, @@ -1106,6 +1085,25 @@ pub struct RawArgs { pub args: Option>>>, } +lazy_static::lazy_static! { + static ref CRASH_FORCEFULLY_AT_STEP: Option = std::env::var("CRASH_FORCEFULLY_AT_STEP") + .ok() + .and_then(|x| x.parse::().ok()); + + static ref CRASH_STEP_COUNTER: AtomicUsize = std::sync::atomic::AtomicUsize::new(0); +} + +#[inline(always)] +fn potentially_crash_for_testing() { + #[cfg(feature = "flow_testing")] + if let Some(crash_at) = CRASH_FORCEFULLY_AT_STEP.as_ref() { + let counter = CRASH_STEP_COUNTER.fetch_add(1, Ordering::SeqCst); + if &counter == crash_at { + panic!("CRASH#1 - expected crash for testing at step {}", crash_at); + } + } +} + // #[async_recursion] // #[instrument(level = "trace", skip_all)] async fn push_next_flow_job( @@ -1160,29 +1158,7 @@ async fn push_next_flow_job "error sending update flow message to job completed channel: {e}" )) })?; - // let r; - // return update_flow_status_after_job_completion( - // db, - // client, - // flow_job.id, - // &Uuid::nil(), - // flow_job.workspace_id.as_str(), - // true, - // if flow.modules.is_empty() { - // r = to_raw_value(&flow_job_args); - // &r - // } else { - // // it has to be an empty for loop event - // serde_json::from_str("[]").unwrap() - // }, - // true, - // same_worker_tx, - // worker_dir, - // None, - // rsmq, - // worker_name, - // ) - // .await; + return Ok(()); } @@ -1229,25 +1205,7 @@ async fn push_next_flow_job "error sending update flow message to job completed channel: {e}" )) })?; - // return update_flow_status_after_job_completion( - // db, - // client, - // flow_job.id, - // &Uuid::nil(), - // flow_job.workspace_id.as_str(), - // true, - // serde_json::from_str( - // "\"not allowed to overlap, scheduling next iteration\"", - // ) - // .unwrap(), - // true, - // same_worker_tx, - // worker_dir, - // Some(true), - // rsmq, - // worker_name, - // ) - // .await; + return Ok(()); } } @@ -1280,22 +1238,7 @@ async fn push_next_flow_job )) })?; - // return update_flow_status_after_job_completion( - // db, - // client, - // flow_job.id, - // &Uuid::nil(), - // flow_job.workspace_id.as_str(), - // true, - // serde_json::from_str("\"stopped early\"").unwrap(), - // true, - // same_worker_tx, - // worker_dir, - // Some(true), - // rsmq, - // worker_name, - // ) - // .await; + return Ok(()); } } @@ -1487,6 +1430,14 @@ async fn push_next_flow_job .execute(&mut *tx) .await?; + sqlx::query!( + "UPDATE queue + SET last_ping = null + WHERE id = $1 AND last_ping = $2", + flow_job.id, + flow_job.last_ping + ).execute(&mut *tx).await?; + tx.commit().await?; return Ok(()); @@ -1536,23 +1487,6 @@ async fn push_next_flow_job "error sending update flow message to job completed channel: {e}" )) })?; - // update_flow_status_after_job_completion( - // db, - // client, - // parent_job, - // &flow_job.id, - // &flow_job.workspace_id, - // true, - // &to_raw_value(&result), - // false, - // same_worker_tx.clone(), - // &worker_dir, - // None, - // rsmq, - // worker_name, - // ) - // .await?; - return Ok(()); } } return Ok(()); @@ -2123,6 +2057,34 @@ async fn push_next_flow_job .await?; }; + + potentially_crash_for_testing(); + + if !flow_job.is_flow_step + && status.step == 0 + { + + if flow_job.schedule_path.is_some() + && flow_job.script_path.is_some() { + tx = handle_maybe_scheduled_job( + tx, + db, + flow_job.schedule_path.as_ref().unwrap(), + flow_job.script_path.as_ref().unwrap(), + &flow_job.workspace_id, + ) + .await?; + } + sqlx::query!( + "UPDATE queue + SET last_ping = null + WHERE id = $1", + flow_job.id + ).execute(&mut tx).await?; + } + + + tx.commit().await?; tracing::info!(id = %flow_job.id, root_id = %job_root, "all next flow jobs pushed: {uuids:?}"); diff --git a/frontend/src/lib/components/runs/JobLoader.svelte b/frontend/src/lib/components/runs/JobLoader.svelte index a4df49c44f..3136d7f04b 100644 --- a/frontend/src/lib/components/runs/JobLoader.svelte +++ b/frontend/src/lib/components/runs/JobLoader.svelte @@ -59,7 +59,7 @@ function computeJobKinds(jobKindsCat: string | undefined): string { if (jobKindsCat == 'all') { - return `${CompletedJob.job_kind.SCRIPT},${CompletedJob.job_kind.FLOW},${CompletedJob.job_kind.DEPENDENCIES},${CompletedJob.job_kind.FLOWDEPENDENCIES},${CompletedJob.job_kind.APPDEPENDENCIES}` + return `${CompletedJob.job_kind.SCRIPT},${CompletedJob.job_kind.FLOW},${CompletedJob.job_kind.DEPENDENCIES},${CompletedJob.job_kind.FLOWDEPENDENCIES},${CompletedJob.job_kind.APPDEPENDENCIES},${CompletedJob.job_kind.PREVIEW},${CompletedJob.job_kind.FLOWPREVIEW}` } else if (jobKindsCat == 'dependencies') { return `${CompletedJob.job_kind.DEPENDENCIES},${CompletedJob.job_kind.FLOWDEPENDENCIES},${CompletedJob.job_kind.APPDEPENDENCIES}` } else if (jobKindsCat == 'previews') {