From 87dd52296f4bb8a585fffd08a6d13e9b1e4856b8 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 1 Aug 2025 11:26:23 +0000 Subject: [PATCH] fix: cleanup concurrency_counter automatically + remove orphans keys automatically --- ...d758dc393f797890d597e5996146247f512ac.json | 38 ++ ...16fb41fece9b3d46fde2d546d70b46f9008e3.json | 15 - ...0b4f28ed5507797165e8c577ce6a95d11694c.json | 15 + ...2b5cd13eb3364a8a8546007dec7ab528b1f9d.json | 20 + ...20cfeee64fd47ded72fce55cc75e0bbb291a8.json | 15 - ...8b0310b14d75e85a558c8cb70f2761923d175.json | 15 + backend/src/main.rs | 28 +- backend/src/monitor.rs | 135 +++++++ backend/windmill-queue/src/jobs.rs | 360 ++++++++++-------- 9 files changed, 442 insertions(+), 199 deletions(-) create mode 100644 backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json delete mode 100644 backend/.sqlx/query-2bf5f7f2cf9d85a5d23e5db2f7616fb41fece9b3d46fde2d546d70b46f9008e3.json create mode 100644 backend/.sqlx/query-7b48820a08fbb3bee0fb2dc802a0b4f28ed5507797165e8c577ce6a95d11694c.json create mode 100644 backend/.sqlx/query-9f4811fe735d401b62f4b7bf3db2b5cd13eb3364a8a8546007dec7ab528b1f9d.json delete mode 100644 backend/.sqlx/query-ca3ba808e020c8c7a35eaef842b20cfeee64fd47ded72fce55cc75e0bbb291a8.json create mode 100644 backend/.sqlx/query-f9e1334ecb17b313924587d96fc8b0310b14d75e85a558c8cb70f2761923d175.json diff --git a/backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json b/backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json new file mode 100644 index 0000000000..469bd6f42c --- /dev/null +++ b/backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json @@ -0,0 +1,38 @@ +{ + "db_name": "PostgreSQL", + "query": "\nWITH lockable_counters AS (\n SELECT concurrency_id, job_uuids\n FROM concurrency_counter\n WHERE job_uuids != '{}'::jsonb\n FOR UPDATE SKIP LOCKED\n),\nall_job_uuids AS (\n SELECT DISTINCT jsonb_object_keys(job_uuids) AS job_uuid\n FROM lockable_counters\n),\norphaned_job_uuids AS (\n SELECT job_uuid\n FROM all_job_uuids\n WHERE job_uuid NOT IN (\n SELECT id::text \n FROM v2_job_queue \n FOR SHARE SKIP LOCKED\n )\n),\norphaned_array AS (\n SELECT ARRAY(SELECT job_uuid FROM orphaned_job_uuids) AS orphaned_keys\n),\nbefore_update AS (\n SELECT lc.concurrency_id, lc.job_uuids, oa.orphaned_keys\n FROM lockable_counters lc, orphaned_array oa\n WHERE lc.job_uuids ?| oa.orphaned_keys\n),\naffected_rows AS (\n UPDATE concurrency_counter \n SET job_uuids = job_uuids - orphaned_array.orphaned_keys\n FROM orphaned_array\n WHERE concurrency_counter.concurrency_id IN (\n SELECT concurrency_id FROM before_update\n )\n RETURNING concurrency_id, job_uuids AS updated_job_uuids\n),\nexpanded_orphaned AS (\n SELECT bu.concurrency_id, \n bu.job_uuids AS original_job_uuids,\n unnest(bu.orphaned_keys) AS orphaned_key\n FROM before_update bu\n)\nSELECT \n eo.concurrency_id,\n eo.orphaned_key,\n eo.original_job_uuids,\n ar.updated_job_uuids\nFROM expanded_orphaned eo\nJOIN affected_rows ar ON eo.concurrency_id = ar.concurrency_id\nWHERE eo.original_job_uuids ? eo.orphaned_key\nORDER BY eo.concurrency_id, eo.orphaned_key\n", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "concurrency_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "orphaned_key", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "original_job_uuids", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "updated_job_uuids", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + null, + false, + false + ] + }, + "hash": "0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac" +} diff --git a/backend/.sqlx/query-2bf5f7f2cf9d85a5d23e5db2f7616fb41fece9b3d46fde2d546d70b46f9008e3.json b/backend/.sqlx/query-2bf5f7f2cf9d85a5d23e5db2f7616fb41fece9b3d46fde2d546d70b46f9008e3.json deleted file mode 100644 index d60b424505..0000000000 --- a/backend/.sqlx/query-2bf5f7f2cf9d85a5d23e5db2f7616fb41fece9b3d46fde2d546d70b46f9008e3.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Int4" - ] - }, - "nullable": [] - }, - "hash": "2bf5f7f2cf9d85a5d23e5db2f7616fb41fece9b3d46fde2d546d70b46f9008e3" -} diff --git a/backend/.sqlx/query-7b48820a08fbb3bee0fb2dc802a0b4f28ed5507797165e8c577ce6a95d11694c.json b/backend/.sqlx/query-7b48820a08fbb3bee0fb2dc802a0b4f28ed5507797165e8c577ce6a95d11694c.json new file mode 100644 index 0000000000..56481d8eb0 --- /dev/null +++ b/backend/.sqlx/query-7b48820a08fbb3bee0fb2dc802a0b4f28ed5507797165e8c577ce6a95d11694c.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Int4" + ] + }, + "nullable": [] + }, + "hash": "7b48820a08fbb3bee0fb2dc802a0b4f28ed5507797165e8c577ce6a95d11694c" +} diff --git a/backend/.sqlx/query-9f4811fe735d401b62f4b7bf3db2b5cd13eb3364a8a8546007dec7ab528b1f9d.json b/backend/.sqlx/query-9f4811fe735d401b62f4b7bf3db2b5cd13eb3364a8a8546007dec7ab528b1f9d.json new file mode 100644 index 0000000000..4c9693ce27 --- /dev/null +++ b/backend/.sqlx/query-9f4811fe735d401b62f4b7bf3db2b5cd13eb3364a8a8546007dec7ab528b1f9d.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "\nWITH rows_to_delete AS (\n SELECT concurrency_id\n FROM concurrency_counter\n WHERE job_uuids = '{}'::jsonb\n FOR UPDATE SKIP LOCKED\n)\nDELETE FROM concurrency_counter\nWHERE concurrency_id IN (SELECT concurrency_id FROM rows_to_delete) RETURNING concurrency_id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "concurrency_id", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false + ] + }, + "hash": "9f4811fe735d401b62f4b7bf3db2b5cd13eb3364a8a8546007dec7ab528b1f9d" +} diff --git a/backend/.sqlx/query-ca3ba808e020c8c7a35eaef842b20cfeee64fd47ded72fce55cc75e0bbb291a8.json b/backend/.sqlx/query-ca3ba808e020c8c7a35eaef842b20cfeee64fd47ded72fce55cc75e0bbb291a8.json deleted file mode 100644 index 6d44fb2840..0000000000 --- a/backend/.sqlx/query-ca3ba808e020c8c7a35eaef842b20cfeee64fd47ded72fce55cc75e0bbb291a8.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Int4" - ] - }, - "nullable": [] - }, - "hash": "ca3ba808e020c8c7a35eaef842b20cfeee64fd47ded72fce55cc75e0bbb291a8" -} diff --git a/backend/.sqlx/query-f9e1334ecb17b313924587d96fc8b0310b14d75e85a558c8cb70f2761923d175.json b/backend/.sqlx/query-f9e1334ecb17b313924587d96fc8b0310b14d75e85a558c8cb70f2761923d175.json new file mode 100644 index 0000000000..f21a486c83 --- /dev/null +++ b/backend/.sqlx/query-f9e1334ecb17b313924587d96fc8b0310b14d75e85a558c8cb70f2761923d175.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Int4" + ] + }, + "nullable": [] + }, + "hash": "f9e1334ecb17b313924587d96fc8b0310b14d75e85a558c8cb70f2761923d175" +} diff --git a/backend/src/main.rs b/backend/src/main.rs index 6cb13ffb83..c0c3123b97 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -95,7 +95,7 @@ use crate::monitor::{ reload_job_default_timeout_setting, reload_jwt_secret_setting, reload_license_key, reload_npm_config_registry_setting, reload_pip_index_url_setting, reload_retention_period_setting, reload_scim_token_setting, reload_smtp_config, - reload_worker_config, + reload_worker_config, MonitorIteration, }; #[cfg(feature = "parquet")] @@ -144,10 +144,12 @@ fn update_ca_certificates_if_requested() { let ca_cert_path = std::env::var("RUN_UPDATE_CA_CERTIFICATE_PATH") .unwrap_or_else(|_| "/usr/sbin/update-ca-certificates".to_string()); - println!("RUN_UPDATE_CA_CERTIFICATE_AT_START=true, running: {}", ca_cert_path); + println!( + "RUN_UPDATE_CA_CERTIFICATE_AT_START=true, running: {}", + ca_cert_path + ); - let output = std::process::Command::new(&ca_cert_path) - .output(); + let output = std::process::Command::new(&ca_cert_path).output(); match output { Ok(result) => { @@ -155,11 +157,17 @@ fn update_ca_certificates_if_requested() { println!("Successfully updated CA certificates"); } else { let stderr = String::from_utf8_lossy(&result.stderr); - println!("Failed to update CA certificates, but continuing startup: {}", stderr.trim()); + println!( + "Failed to update CA certificates, but continuing startup: {}", + stderr.trim() + ); } } Err(e) => { - println!("Could not run update-ca-certificates command, but continuing startup: {}", e); + println!( + "Could not run update-ca-certificates command, but continuing startup: {}", + e + ); } } } @@ -565,6 +573,7 @@ Windmill Community Edition {GIT_VERSION} worker_mode, true, killpill_tx.clone(), + None, ) .await; @@ -792,6 +801,8 @@ Windmill Community Edition {GIT_VERSION} let h = tokio::spawn(async move { let mut listener = retry_listen_pg(&db_url).await; let mut last_listener_refresh = Instant::now(); + let mut monitor_iteration: u64 = 0; + let rd_shift: u8 = rand::rng().random_range(0..200); loop { let db = db.clone(); tokio::select! { @@ -1127,8 +1138,13 @@ Windmill Community Edition {GIT_VERSION} worker_mode, false, tx.clone(), + Some(MonitorIteration { + rd_shift, + iter: monitor_iteration, + }), ) .await; + monitor_iteration += 1; if server_mode { if !*windmill_common::QUIET_LOGS { tracing::info!("monitor task finished"); diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 0deddb75e5..4492afe692 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1337,6 +1337,17 @@ pub async fn monitor_pool(db: &DB) { } } +pub struct MonitorIteration { + pub rd_shift: u8, + pub iter: u64, +} + +impl MonitorIteration { + pub fn should_run(&self, period: u8) -> bool { + self.iter % (period as u64) == self.rd_shift as u64 + } +} + pub async fn monitor_db( conn: &Connection, base_internal_url: &str, @@ -1344,6 +1355,7 @@ pub async fn monitor_db( _worker_mode: bool, initial_load: bool, _killpill_tx: KillpillSender, + iteration: Option, ) { let zombie_jobs_f = async { if server_mode && !initial_load && !*DISABLE_ZOMBIE_JOBS_MONITORING { @@ -1358,6 +1370,29 @@ pub async fn monitor_db( } } }; + + // run every 5 minutes + let cleanup_concurrency_counters_f = async { + if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(10) { + if let Some(db) = conn.as_sql() { + if let Err(e) = cleanup_concurrency_counters_orphaned_keys(&db).await { + tracing::error!("Error cleaning up concurrency counters: {:?}", e); + } + } + } + }; + + // run every 10 minutes + let cleanup_concurrency_counters_empty_keys_f = async { + if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(20) { + if let Some(db) = conn.as_sql() { + if let Err(e) = cleanup_concurrency_counters_empty_keys(&db).await { + tracing::error!("Error cleaning up concurrency counters: {:?}", e); + } + } + } + }; + let expired_items_f = async { if server_mode && !initial_load { if let Some(db) = conn.as_sql() { @@ -1441,6 +1476,8 @@ pub async fn monitor_db( low_disk_alerts_f, apply_autoscaling_f, update_min_worker_version_f, + cleanup_concurrency_counters_f, + cleanup_concurrency_counters_empty_keys_f, ); } @@ -1983,6 +2020,104 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker } } +async fn cleanup_concurrency_counters_orphaned_keys(db: &DB) -> error::Result<()> { + let result = sqlx::query!( + " +WITH lockable_counters AS ( + SELECT concurrency_id, job_uuids + FROM concurrency_counter + WHERE job_uuids != '{}'::jsonb + FOR UPDATE SKIP LOCKED +), +all_job_uuids AS ( + SELECT DISTINCT jsonb_object_keys(job_uuids) AS job_uuid + FROM lockable_counters +), +orphaned_job_uuids AS ( + SELECT job_uuid + FROM all_job_uuids + WHERE job_uuid NOT IN ( + SELECT id::text + FROM v2_job_queue + FOR SHARE SKIP LOCKED + ) +), +orphaned_array AS ( + SELECT ARRAY(SELECT job_uuid FROM orphaned_job_uuids) AS orphaned_keys +), +before_update AS ( + SELECT lc.concurrency_id, lc.job_uuids, oa.orphaned_keys + FROM lockable_counters lc, orphaned_array oa + WHERE lc.job_uuids ?| oa.orphaned_keys +), +affected_rows AS ( + UPDATE concurrency_counter + SET job_uuids = job_uuids - orphaned_array.orphaned_keys + FROM orphaned_array + WHERE concurrency_counter.concurrency_id IN ( + SELECT concurrency_id FROM before_update + ) + RETURNING concurrency_id, job_uuids AS updated_job_uuids +), +expanded_orphaned AS ( + SELECT bu.concurrency_id, + bu.job_uuids AS original_job_uuids, + unnest(bu.orphaned_keys) AS orphaned_key + FROM before_update bu +) +SELECT + eo.concurrency_id, + eo.orphaned_key, + eo.original_job_uuids, + ar.updated_job_uuids +FROM expanded_orphaned eo +JOIN affected_rows ar ON eo.concurrency_id = ar.concurrency_id +WHERE eo.original_job_uuids ? eo.orphaned_key +ORDER BY eo.concurrency_id, eo.orphaned_key +", + ) + .fetch_all(db) + .await?; + + if result.len() > 0 { + tracing::info!("Cleaned up {} concurrency counters", result.len()); + for row in result { + tracing::info!("Concurrency counter cleaned up: concurrency_id: {}, orphaned_key: {:?}, original_job_uuids: {:?}, updated_job_uuids: {:?}", row.concurrency_id, row.orphaned_key, row.original_job_uuids, row.updated_job_uuids); + } + } + Ok(()) +} + +async fn cleanup_concurrency_counters_empty_keys(db: &DB) -> error::Result<()> { + let result = sqlx::query!( + " +WITH rows_to_delete AS ( + SELECT concurrency_id + FROM concurrency_counter + WHERE job_uuids = '{}'::jsonb + FOR UPDATE SKIP LOCKED +) +DELETE FROM concurrency_counter +WHERE concurrency_id IN (SELECT concurrency_id FROM rows_to_delete) RETURNING concurrency_id", + ) + .fetch_all(db) + .await?; + + if result.len() > 0 { + tracing::info!( + "Cleaned up {} empty concurrency counters: {:?}", + result.len(), + result + .iter() + .map(|x| x.concurrency_id.to_string()) + .collect::>() + .join(",") + ); + } + + Ok(()) +} + async fn handle_zombie_flows(db: &DB) -> error::Result<()> { let flows = sqlx::query!( r#" diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 7fc3c90b60..4ddb28a492 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -810,7 +810,8 @@ pub async fn add_completed_job( ) .await; - restart_job_if_perpetual(db, queued_job, canceled_by).await?; + restart_job_if_perpetual(db, queued_job, &canceled_by).await?; + // tracing::error!("4 {:?}", start.elapsed()); Ok(queued_job.id) @@ -1070,29 +1071,24 @@ async fn commit_completed_job( } } if queued_job.concurrent_limit.is_some() { - let concurrency_key = match concurrency_key(db, &queued_job.id).await { - Ok(c) => c, - Err(e) => { - tracing::error!( - "Could not get concurrency key for job {} defaulting to default key: {e:?}", - queued_job.id - ); - "".to_string() - } - }; - if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_empty() { + let concurrency_key = concurrency_key(db, &queued_job.id).await?; + if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_none() { tracing::warn!("Concurrency limit is disabled, skipping"); } else { - if let Err(e) = sqlx::query_scalar!( - "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1", - concurrency_key, - queued_job.id.hyphenated().to_string(), - ) - .execute(&mut *tx) - .await - { - tracing::error!("Could not decrement concurrency counter: {}", e); - } + let concurrency_key = concurrency_key.unwrap(); + sqlx::query_scalar!( + "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1", + concurrency_key, + queued_job.id.hyphenated().to_string(), + ) + .execute(&mut *tx) + .await + .map_err(|e| { + Error::internal_err(format!( + "Could not decrement concurrency counter for job_id={}: {e:#}", + queued_job.id + )) + })?; } if let Err(e) = sqlx::query_scalar!( @@ -1101,12 +1097,11 @@ async fn commit_completed_job( ) .execute(&mut *tx) .await - .map_err(|e| { - Error::internal_err(format!( - "Error updating to add ended_at timestamp concurrency_key={concurrency_key}: {e:#}" - )) - }) { - tracing::error!("Could not update concurrency_key: {}", e); + { + tracing::error!( + "Could not update concurrency_key ended_at for job_id={}: {e:#}", + queued_job.id, + ); } tracing::debug!("decremented concurrency counter"); } @@ -1178,93 +1173,124 @@ async fn check_result_size( async fn restart_job_if_perpetual( db: &Pool, queued_job: &MiniPulledJob, - canceled_by: Option, + canceled_by: &Option, ) -> Result<(), Error> { - Ok( - if !queued_job.is_flow_step() && queued_job.kind == JobKind::Script && canceled_by.is_none() - { - if let Some(hash) = queued_job.runnable_id { - let p = sqlx::query_scalar!( - "SELECT restart_unless_cancelled FROM script WHERE hash = $1 AND workspace_id = $2", - hash.0, - &queued_job.workspace_id - ) - .fetch_optional(db) - .await? - .flatten() - .unwrap_or(false); + if !queued_job.is_flow_step() && queued_job.kind == JobKind::Script && canceled_by.is_none() { + if let Some(hash) = queued_job.runnable_id { + (|| restart_job_if_perpetual_inner(db, queued_job, hash)) + .retry( + ConstantBuilder::default() + .with_delay(std::time::Duration::from_secs(3)) + .with_max_times(5) + .build(), + ) + .notify(|err, dur| { + tracing::error!( + "Could not apply perpetual job restart, retrying in {dur:#?}, err: {err:#?}" + ); + }) + .sleep(tokio::time::sleep) + .await?; + } + } + Ok(()) +} - if p { - let tx = PushIsolationLevel::IsolatedRoot(db.clone()); - - // perpetual jobs can run one job per 10s max. If the job was faster than 10s, schedule the next one with the appropriate delay - let now = chrono::Utc::now(); - let scheduled_for = if now - .signed_duration_since(queued_job.started_at.unwrap_or(now)) - .num_seconds() - < 10 - { - let next_run = queued_job.started_at.unwrap_or(now) - + chrono::Duration::try_seconds(10).unwrap(); - tracing::warn!("Perpetual script {:?} is running too fast, only 1 job per 10s it supported. Scheduling next run for {:?}", queued_job.runnable_path, next_run); - Some(next_run) - } else { - None - }; - - let ehm = HashMap::new(); - let (_uuid, tx) = push( - db, - tx, - &queued_job.workspace_id, - JobPayload::ScriptHash { - hash, - path: queued_job.runnable_path().to_string(), - custom_concurrency_key: custom_concurrency_key(db, &queued_job.id) - .await?, - concurrent_limit: queued_job.concurrent_limit, - concurrency_time_window_s: queued_job.concurrency_time_window_s, - cache_ttl: queued_job.cache_ttl, - dedicated_worker: None, - language: queued_job - .script_lang - .clone() - .unwrap_or_else(|| ScriptLang::Deno), - priority: queued_job.priority, - apply_preprocessor: false, - }, - queued_job - .args - .as_ref() - .map(|x| PushArgs::from(&x.0)) - .unwrap_or_else(|| PushArgs::from(&ehm)), - &queued_job.created_by, - &queued_job.permissioned_as_email, - queued_job.permissioned_as.clone(), - Some(&format!("add.completed.job{}", queued_job.id)), - scheduled_for, - queued_job.schedule_path(), - None, - None, - None, - false, - false, - None, - queued_job.visible_to_owner, - Some(queued_job.tag.clone()), - queued_job.timeout, - None, - queued_job.priority, - None, - ) - .await?; - if let Err(e) = tx.commit().await { - tracing::error!("Could not restart job {}: {}", queued_job.id, e); - } - } - } - }, +async fn restart_job_if_perpetual_inner( + db: &Pool, + queued_job: &MiniPulledJob, + hash: ScriptHash, +) -> Result<(), Error> { + let restart = sqlx::query_scalar!( + "SELECT restart_unless_cancelled FROM script WHERE hash = $1 AND workspace_id = $2", + hash.0, + &queued_job.workspace_id ) + .fetch_optional(db) + .await? + .flatten() + .unwrap_or(false); + + if restart { + let tx = PushIsolationLevel::IsolatedRoot(db.clone()); + + // perpetual jobs can run one job per 10s max. If the job was faster than 10s, schedule the next one with the appropriate delay + let now = now_from_db(db).await?; + let scheduled_for = if now + .signed_duration_since(queued_job.started_at.unwrap_or(now)) + .num_seconds() + < 10 + { + let next_run = + queued_job.started_at.unwrap_or(now) + chrono::Duration::try_seconds(10).unwrap(); + tracing::warn!("Perpetual script {:?} is running too fast, only 1 job per 10s it supported. Scheduling next run for {:?}", queued_job.runnable_path, next_run); + Some(next_run) + } else { + None + }; + + let ehm = HashMap::new(); + let (_uuid, tx) = push( + db, + tx, + &queued_job.workspace_id, + JobPayload::ScriptHash { + hash, + path: queued_job.runnable_path().to_string(), + custom_concurrency_key: custom_concurrency_key(db, &queued_job.id).await?, + concurrent_limit: queued_job.concurrent_limit, + concurrency_time_window_s: queued_job.concurrency_time_window_s, + cache_ttl: queued_job.cache_ttl, + dedicated_worker: None, + language: queued_job + .script_lang + .clone() + .unwrap_or_else(|| ScriptLang::Deno), + priority: queued_job.priority, + apply_preprocessor: false, + }, + queued_job + .args + .as_ref() + .map(|x| PushArgs::from(&x.0)) + .unwrap_or_else(|| PushArgs::from(&ehm)), + &queued_job.created_by, + &queued_job.permissioned_as_email, + queued_job.permissioned_as.clone(), + Some(&format!("add.completed.job{}", queued_job.id)), + scheduled_for, + queued_job.schedule_path(), + None, + None, + None, + false, + false, + None, + queued_job.visible_to_owner, + Some(queued_job.tag.clone()), + queued_job.timeout, + None, + queued_job.priority, + None, + ) + .await?; + tx.commit().await?; + } + Ok(()) +} + +async fn has_failure_module(db: &Pool, job: &MiniPulledJob) -> bool { + if let Ok(flow) = cache::job::fetch_flow(db, job.kind, job.runnable_id).await { + return flow.value().failure_module.is_some(); + } + sqlx::query_scalar!( + "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM v2_job WHERE id = $1", + job.id + ) + .fetch_one(db) + .await + .unwrap_or(Some(false)) + .unwrap_or(false) } #[cfg(feature = "enterprise")] @@ -1277,20 +1303,6 @@ async fn apply_completed_job_error_handlers, job: &MiniPulledJob) -> bool { - if let Ok(flow) = cache::job::fetch_flow(db, job.kind, job.runnable_id).await { - return flow.value().failure_module.is_some(); - } - sqlx::query_scalar!( - "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM v2_job WHERE id = $1", - job.id - ) - .fetch_one(db) - .await - .unwrap_or(Some(false)) - .unwrap_or(false) - } - if queued_job.permissioned_as_email == ERROR_HANDLER_USER_EMAIL { let base_url = BASE_URL.read().await; let w_id = &queued_job.workspace_id; @@ -1398,31 +1410,34 @@ fn apply_completed_job_cloud_usage( let premium_workspace = windmill_common::workspaces::is_premium_workspace(&db, &w_id).await; tokio::time::timeout(std::time::Duration::from_secs(10), async move { - let _ = sqlx::query!( - "INSERT INTO usage (id, is_workspace, month_, usage) - VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) - ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2", - w_id, - additional_usage as i32 - ) - .execute(&db) - .await - .map_err(|e| Error::internal_err(format!("updating usage: {e:#}"))); - - if !premium_workspace { let _ = sqlx::query!( "INSERT INTO usage (id, is_workspace, month_, usage) - VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) + VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2", - email, + w_id, additional_usage as i32 ) .execute(&db) .await - .map_err(|e| Error::internal_err(format!("updating usage: {e:#}"))); - }}).await.unwrap_or_else(|_| { - tracing::error!("Could not update usage for workspace {w_id2} and permissioned as {email2}, stopped after 10s"); - }); + .map_err(|e| { + Error::internal_err(format!("updating usage: {e:#}")) + }); + + if !premium_workspace { + let _ = sqlx::query!( + "INSERT INTO usage (id, is_workspace, month_, usage) + VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) + ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2", + email, + additional_usage as i32 + ) + .execute(&db) + .await + .map_err(|e| Error::internal_err(format!("updating usage: {e:#}"))); + } + }).await.unwrap_or_else(|_| { + tracing::error!("Could not update usage for workspace {w_id2} and permissioned as {email2}, stopped after 10s"); + }); }); } } @@ -2565,16 +2580,12 @@ pub async fn pull( return Ok(PulledJobResult { job: Some(pulled_job), suspended }); } - let job_concurrency_key = match concurrency_key(db, &pulled_job.id).await { - Ok(key) => key, - Err(e) => { - tracing::error!( - "Could not get concurrency key for job {} defaulting to default key: {e:?}", - pulled_job.id - ); - "".to_string() - } - }; + let job_concurrency_key = concurrency_key(db, &pulled_job.id).await?; + if job_concurrency_key.is_none() { + tracing::warn!("No concurrency key found for job {}", pulled_job.id); + return Ok(PulledJobResult { job: None, suspended }); + } + let job_concurrency_key = job_concurrency_key.unwrap(); tracing::debug!("Concurrency key is '{}'", job_concurrency_key); let job_custom_concurrent_limit = pulled_job.concurrent_limit.unwrap(); // setting concurrency_time_window to 0 will count only the currently running jobs @@ -2821,17 +2832,40 @@ pub async fn custom_concurrency_key( db: &Pool, job_id: &Uuid, ) -> Result, sqlx::Error> { - sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", job_id) - .fetch_optional(db) // this should no longer be fetch optional - .await + let fut = async || { + sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", job_id) + .fetch_optional(db) // this should no longer be fetch optional + .await + }; + fut.retry( + ConstantBuilder::default() + .with_delay(std::time::Duration::from_secs(3)) + .with_max_times(5) + .build(), + ) + .notify(|err, dur| { + tracing::error!( + "Could not get concurrency key for job {job_id}, retrying in {dur:#?}, err: {err:#?}" + ); + }) + .await } -async fn concurrency_key(db: &Pool, id: &Uuid) -> windmill_common::error::Result { - not_found_if_none( - custom_concurrency_key(db, id).await?, - "ConcurrencyKey", - id.to_string(), - ) +async fn concurrency_key( + db: &Pool, + id: &Uuid, +) -> windmill_common::error::Result> { + custom_concurrency_key(db, id) + .await + .map(|x| { + if x.is_none() { + tracing::info!("No concurrency key found for job {id}, defaulting to empty string"); + } + return x; + }) + .map_err(|e| { + Error::internal_err(format!("Could not get concurrency key for job {id}: {e:#}")) + }) } pub fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String {