From 30abc3ff22df36fe6c6a0b7f8875a163bba86a2d Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 17 Jan 2026 07:33:34 +0000 Subject: [PATCH] fix: improve job deletion performance and batching --- ...d758dc393f797890d597e5996146247f512ac.json | 38 --- ...2c8f7922450515d1eaee8b3abee5b31f72527.json | 14 + ...4f61baeb3ae6f31ba3fd57a3ec485108a7f49.json | 23 ++ ...a533d7eec520376b99b2fd8000f0f0acba7b.json} | 4 +- ...4a3ab8ba3d162ed5d04735caea7295da0b20.json} | 4 +- ...3146f45ad4a6e584e917a0a86280d8f473c10.json | 22 -- ...0fa78b06106236522e48ab5fe009cac059468.json | 38 +++ ...edf2b1c0e6e6cb3c9183a937a68465debdb06.json | 14 - backend/src/monitor.rs | 256 +++++++++++------- .../windmill-common/src/global_settings.rs | 2 + 10 files changed, 243 insertions(+), 172 deletions(-) delete mode 100644 backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json create mode 100644 backend/.sqlx/query-1ccc9db6c6cb51bebeb90d003e12c8f7922450515d1eaee8b3abee5b31f72527.json create mode 100644 backend/.sqlx/query-306e0156ee1541710c1c6512ecb4f61baeb3ae6f31ba3fd57a3ec485108a7f49.json rename backend/.sqlx/{query-c6ef0acdf20bd71dd26de981fb49f178ba8a1b8c1e01e0fec1dfd6a54ea7a894.json => query-45d0552a169b63241a0ecc0e7ec6a533d7eec520376b99b2fd8000f0f0acba7b.json} (50%) rename backend/.sqlx/{query-70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b.json => query-4d80985dd0794a01a2af18ae7abf4a3ab8ba3d162ed5d04735caea7295da0b20.json} (91%) delete mode 100644 backend/.sqlx/query-5820d34be1a7f7b72e656c692f53146f45ad4a6e584e917a0a86280d8f473c10.json create mode 100644 backend/.sqlx/query-598805056a3b102c9c2fd69d8330fa78b06106236522e48ab5fe009cac059468.json delete mode 100644 backend/.sqlx/query-9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06.json diff --git a/backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json b/backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json deleted file mode 100644 index 469bd6f42c..0000000000 --- a/backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json +++ /dev/null @@ -1,38 +0,0 @@ -{ - "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-1ccc9db6c6cb51bebeb90d003e12c8f7922450515d1eaee8b3abee5b31f72527.json b/backend/.sqlx/query-1ccc9db6c6cb51bebeb90d003e12c8f7922450515d1eaee8b3abee5b31f72527.json new file mode 100644 index 0000000000..4dc0e74577 --- /dev/null +++ b/backend/.sqlx/query-1ccc9db6c6cb51bebeb90d003e12c8f7922450515d1eaee8b3abee5b31f72527.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM concurrency_key WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [] + }, + "hash": "1ccc9db6c6cb51bebeb90d003e12c8f7922450515d1eaee8b3abee5b31f72527" +} diff --git a/backend/.sqlx/query-306e0156ee1541710c1c6512ecb4f61baeb3ae6f31ba3fd57a3ec485108a7f49.json b/backend/.sqlx/query-306e0156ee1541710c1c6512ecb4f61baeb3ae6f31ba3fd57a3ec485108a7f49.json new file mode 100644 index 0000000000..1e05f032f0 --- /dev/null +++ b/backend/.sqlx/query-306e0156ee1541710c1c6512ecb4f61baeb3ae6f31ba3fd57a3ec485108a7f49.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM v2_job_completed\n WHERE id IN (\n SELECT id FROM v2_job_completed\n WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval\n ORDER BY completed_at ASC\n LIMIT $2\n FOR UPDATE SKIP LOCKED\n )\n RETURNING id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Int8", + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "306e0156ee1541710c1c6512ecb4f61baeb3ae6f31ba3fd57a3ec485108a7f49" +} diff --git a/backend/.sqlx/query-c6ef0acdf20bd71dd26de981fb49f178ba8a1b8c1e01e0fec1dfd6a54ea7a894.json b/backend/.sqlx/query-45d0552a169b63241a0ecc0e7ec6a533d7eec520376b99b2fd8000f0f0acba7b.json similarity index 50% rename from backend/.sqlx/query-c6ef0acdf20bd71dd26de981fb49f178ba8a1b8c1e01e0fec1dfd6a54ea7a894.json rename to backend/.sqlx/query-45d0552a169b63241a0ecc0e7ec6a533d7eec520376b99b2fd8000f0f0acba7b.json index 0bd82635da..acad0a4a1d 100644 --- a/backend/.sqlx/query-c6ef0acdf20bd71dd26de981fb49f178ba8a1b8c1e01e0fec1dfd6a54ea7a894.json +++ b/backend/.sqlx/query-45d0552a169b63241a0ecc0e7ec6a533d7eec520376b99b2fd8000f0f0acba7b.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\nWITH rows_to_delete AS (\n SELECT concurrency_id\n FROM concurrency_counter\n \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", + "query": "\nWITH rows_to_delete AS (\n SELECT concurrency_id\n FROM concurrency_counter\n\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": [ { @@ -16,5 +16,5 @@ false ] }, - "hash": "c6ef0acdf20bd71dd26de981fb49f178ba8a1b8c1e01e0fec1dfd6a54ea7a894" + "hash": "45d0552a169b63241a0ecc0e7ec6a533d7eec520376b99b2fd8000f0f0acba7b" } diff --git a/backend/.sqlx/query-70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b.json b/backend/.sqlx/query-4d80985dd0794a01a2af18ae7abf4a3ab8ba3d162ed5d04735caea7295da0b20.json similarity index 91% rename from backend/.sqlx/query-70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b.json rename to backend/.sqlx/query-4d80985dd0794a01a2af18ae7abf4a3ab8ba3d162ed5d04735caea7295da0b20.json index 7cfd1e5b64..4847a9a2b1 100644 --- a/backend/.sqlx/query-70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b.json +++ b/backend/.sqlx/query-4d80985dd0794a01a2af18ae7abf4a3ab8ba3d162ed5d04735caea7295da0b20.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT\n j.id AS \"id!\", j.workspace_id AS \"workspace_id!\", j.parent_job, j.flow_step_id IS NOT NULL AS \"is_flow_step?\",\n COALESCE(s.flow_status, s.workflow_as_code_status) AS \"flow_status: Box\", r.ping AS last_ping, j.same_worker AS \"same_worker?\"\n 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)\n WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()\n AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode')\n AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval\n AND q.canceled_by IS NULL\n \n ", + "query": "\n SELECT\n j.id AS \"id!\", j.workspace_id AS \"workspace_id!\", j.parent_job, j.flow_step_id IS NOT NULL AS \"is_flow_step?\",\n COALESCE(s.flow_status, s.workflow_as_code_status) AS \"flow_status: Box\", r.ping AS last_ping, j.same_worker AS \"same_worker?\"\n 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)\n WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()\n AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode')\n AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval\n AND q.canceled_by IS NULL\n\n ", "describe": { "columns": [ { @@ -54,5 +54,5 @@ false ] }, - "hash": "70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b" + "hash": "4d80985dd0794a01a2af18ae7abf4a3ab8ba3d162ed5d04735caea7295da0b20" } diff --git a/backend/.sqlx/query-5820d34be1a7f7b72e656c692f53146f45ad4a6e584e917a0a86280d8f473c10.json b/backend/.sqlx/query-5820d34be1a7f7b72e656c692f53146f45ad4a6e584e917a0a86280d8f473c10.json deleted file mode 100644 index b83ec1db19..0000000000 --- a/backend/.sqlx/query-5820d34be1a7f7b72e656c692f53146f45ad4a6e584e917a0a86280d8f473c10.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM v2_job_completed c\n WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval \n RETURNING c.id", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [ - false - ] - }, - "hash": "5820d34be1a7f7b72e656c692f53146f45ad4a6e584e917a0a86280d8f473c10" -} diff --git a/backend/.sqlx/query-598805056a3b102c9c2fd69d8330fa78b06106236522e48ab5fe009cac059468.json b/backend/.sqlx/query-598805056a3b102c9c2fd69d8330fa78b06106236522e48ab5fe009cac059468.json new file mode 100644 index 0000000000..3c0b930a43 --- /dev/null +++ b/backend/.sqlx/query-598805056a3b102c9c2fd69d8330fa78b06106236522e48ab5fe009cac059468.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": "598805056a3b102c9c2fd69d8330fa78b06106236522e48ab5fe009cac059468" +} diff --git a/backend/.sqlx/query-9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06.json b/backend/.sqlx/query-9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06.json deleted file mode 100644 index 9f42f0f54d..0000000000 --- a/backend/.sqlx/query-9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM concurrency_key WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [] - }, - "hash": "9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06" -} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 51ab25fee9..115dcf6d27 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -155,6 +155,20 @@ lazy_static::lazy_static! { .ok() .and_then(|x| x.parse::().ok()); + + /// Batch size for job cleanup deletion queries. Default: 10000. + /// Larger values delete more jobs per batch but hold locks longer. + static ref JOB_CLEANUP_BATCH_SIZE: i64 = std::env::var("JOB_CLEANUP_BATCH_SIZE") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(20000); + + /// Maximum number of batches to process per cleanup iteration. Default: 10. + /// Set to 0 for unlimited (process until no expired jobs remain). + static ref JOB_CLEANUP_MAX_BATCHES: i32 = std::env::var("JOB_CLEANUP_MAX_BATCHES") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(20); } pub async fn initial_load( @@ -910,99 +924,153 @@ pub async fn delete_expired_items(db: &DB) -> () { let job_retention_secs = *JOB_RETENTION_SECS.read().await; if job_retention_secs > 0 { - match db.begin().await { - Ok(mut tx) => { - let deleted_jobs = sqlx::query_scalar!( - "DELETE FROM v2_job_completed c - WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval - RETURNING c.id", - job_retention_secs - ) - .fetch_all(&mut *tx) - .await; + let batch_size = *JOB_CLEANUP_BATCH_SIZE; + let max_batches = *JOB_CLEANUP_MAX_BATCHES; + let cleanup_start = Instant::now(); + let mut total_deleted = 0u64; + let mut batch_num = 0i32; - match deleted_jobs { - Ok(deleted_jobs) => { - if deleted_jobs.len() > 0 { - tracing::info!( - "deleted {} jobs completed JOB_RETENTION_SECS {} ago: {:?}", - deleted_jobs.len(), - job_retention_secs, - deleted_jobs, - ); - if let Err(e) = sqlx::query!( - "DELETE FROM job_stats WHERE job_id = ANY($1)", - &deleted_jobs - ) - .execute(&mut *tx) - .await - { - tracing::error!("Error deleting job stats: {:?}", e); - } - match sqlx::query_scalar!( - "DELETE FROM job_logs WHERE job_id = ANY($1) RETURNING log_file_index", - &deleted_jobs - ) - .fetch_all(&mut *tx) - .await - { - Ok(log_file_index) => { - let paths = log_file_index - .into_iter() - .filter_map(|opt| opt) - .flat_map(|inner_vec| inner_vec.into_iter()) - .collect(); - delete_log_files_from_disk_and_store(paths, TMP_DIR, "").await; - } - Err(e) => tracing::error!("Error deleting job stats: {:?}", e), - } - if let Err(e) = sqlx::query!( - "DELETE FROM concurrency_key WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval ", - job_retention_secs - ) - .execute(&mut *tx) - .await - { - tracing::error!("Error deleting custom concurrency key: {:?}", e); - } - - if let Err(e) = - sqlx::query!("DELETE FROM v2_job WHERE id = ANY($1)", &deleted_jobs) - .execute(&mut *tx) - .await - { - tracing::error!("Error deleting job: {:?}", e); - } - - // should already be deleted but just in case - if let Err(e) = sqlx::query!( - "DELETE FROM job_result_stream_v2 WHERE job_id = ANY($1)", - &deleted_jobs - ) - .execute(&mut *tx) - .await - { - tracing::error!("Error deleting job result stream: {:?}", e); - } - } - } - Err(e) => { - tracing::error!("Error deleting expired jobs: {:?}", e) - } - } - - match tx.commit().await { - Ok(_) => (), - Err(err) => tracing::error!("Error deleting expired jobs: {:?}", err), - } + // Process batches until no more expired jobs or max batches reached + loop { + if max_batches > 0 && batch_num >= max_batches { + tracing::debug!( + "Job cleanup: reached max batches limit ({}), will continue next iteration", + max_batches + ); + break; } - Err(err) => { - tracing::error!("Error deleting expired jobs: {:?}", err) + + // Each batch runs in its own transaction to avoid long-running locks + let batch_result = delete_expired_jobs_batch(db, job_retention_secs, batch_size).await; + + match batch_result { + Ok(deleted_count) => { + if deleted_count == 0 { + // No more expired jobs to delete + break; + } + total_deleted += deleted_count as u64; + batch_num += 1; + } + Err(e) => { + tracing::error!("Error in job cleanup batch {}: {:?}", batch_num, e); + break; + } } } + + if total_deleted > 0 { + tracing::info!( + "Job cleanup completed: deleted {} jobs in {} batches, took {:?}", + total_deleted, + batch_num, + cleanup_start.elapsed() + ); + } + + // Clean up concurrency keys separately (not tied to specific job IDs) + if let Err(e) = sqlx::query!( + "DELETE FROM concurrency_key WHERE ended_at <= now() - ($1::bigint::text || ' s')::interval", + job_retention_secs + ) + .execute(db) + .await + { + tracing::error!("Error deleting custom concurrency key: {:?}", e); + } } } +/// Delete a batch of expired jobs with LIMIT and SKIP LOCKED for high-scale environments. +/// Uses a single transaction per batch to minimize lock duration. +/// Returns the number of jobs deleted in this batch. +async fn delete_expired_jobs_batch( + db: &DB, + job_retention_secs: i64, + batch_size: i64, +) -> error::Result { + let mut tx = db.begin().await?; + + // Use FOR UPDATE SKIP LOCKED to avoid contention between replicas + // ORDER BY completed_at ensures we delete oldest jobs first + let deleted_jobs: Vec = sqlx::query_scalar!( + "DELETE FROM v2_job_completed + WHERE id IN ( + SELECT id FROM v2_job_completed + WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval + ORDER BY completed_at ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + RETURNING id", + job_retention_secs, + batch_size + ) + .fetch_all(&mut *tx) + .await?; + + let deleted_count = deleted_jobs.len(); + + if deleted_count > 0 { + tracing::debug!( + "Deleting batch of {} expired jobs (retention: {}s)", + deleted_count, + job_retention_secs + ); + + // Delete related records for this batch + if let Err(e) = sqlx::query!( + "DELETE FROM job_stats WHERE job_id = ANY($1)", + &deleted_jobs + ) + .execute(&mut *tx) + .await + { + tracing::error!("Error deleting job stats: {:?}", e); + } + + match sqlx::query_scalar!( + "DELETE FROM job_logs WHERE job_id = ANY($1) RETURNING log_file_index", + &deleted_jobs + ) + .fetch_all(&mut *tx) + .await + { + Ok(log_file_index) => { + let paths = log_file_index + .into_iter() + .filter_map(|opt| opt) + .flat_map(|inner_vec| inner_vec.into_iter()) + .collect(); + delete_log_files_from_disk_and_store(paths, TMP_DIR, "").await; + } + Err(e) => tracing::error!("Error deleting job logs: {:?}", e), + } + + if let Err(e) = sqlx::query!("DELETE FROM v2_job WHERE id = ANY($1)", &deleted_jobs) + .execute(&mut *tx) + .await + { + tracing::error!("Error deleting job: {:?}", e); + } + + // Should already be deleted but just in case + if let Err(e) = sqlx::query!( + "DELETE FROM job_result_stream_v2 WHERE job_id = ANY($1)", + &deleted_jobs + ) + .execute(&mut *tx) + .await + { + tracing::error!("Error deleting job result stream: {:?}", e); + } + } + + tx.commit().await?; + + Ok(deleted_count) +} + async fn delete_log_files_from_disk_and_store( paths_to_delete: Vec, tmp_dir: &str, @@ -2262,7 +2330,7 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, node_n sqlx::query_scalar!("SELECT j.id 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 r.ping < now() - ($1 || ' seconds')::interval - AND q.running = true AND j.kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow') AND j.same_worker = false", + AND q.running = true AND j.kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow') AND j.same_worker = false", ZOMBIE_JOB_TIMEOUT.as_str()) .fetch_all(db) .await @@ -2396,8 +2464,8 @@ orphaned_job_uuids AS ( SELECT job_uuid FROM all_job_uuids WHERE job_uuid NOT IN ( - SELECT id::text - FROM v2_job_queue + SELECT id::text + FROM v2_job_queue FOR SHARE SKIP LOCKED ) ), @@ -2410,7 +2478,7 @@ before_update AS ( WHERE lc.job_uuids ?| oa.orphaned_keys ), affected_rows AS ( - UPDATE concurrency_counter + UPDATE concurrency_counter SET job_uuids = job_uuids - orphaned_array.orphaned_keys FROM orphaned_array WHERE concurrency_counter.concurrency_id IN ( @@ -2419,12 +2487,12 @@ affected_rows AS ( RETURNING concurrency_id, job_uuids AS updated_job_uuids ), expanded_orphaned AS ( - SELECT bu.concurrency_id, + SELECT bu.concurrency_id, bu.job_uuids AS original_job_uuids, unnest(bu.orphaned_keys) AS orphaned_key FROM before_update bu ) -SELECT +SELECT eo.concurrency_id, eo.orphaned_key, eo.original_job_uuids, @@ -2453,7 +2521,7 @@ async fn cleanup_concurrency_counters_empty_keys(db: &DB) -> error::Result<()> { WITH rows_to_delete AS ( SELECT concurrency_id FROM concurrency_counter - + WHERE job_uuids = '{}'::jsonb FOR UPDATE SKIP LOCKED ) @@ -2489,7 +2557,7 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> { AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode') AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval AND q.canceled_by IS NULL - + "#, FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str() ) diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index a73f488aae..6dbc8c8ba6 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -115,6 +115,8 @@ pub const ENV_SETTINGS: &[&str] = &[ "PG_SCHEMA", "PG_LISTENER_REFRESH_PERIOD_SECS", "AI_REQUEST_TIMEOUT_SECONDS", + "JOB_CLEANUP_BATCH_SIZE", + "JOB_CLEANUP_MAX_BATCHES", ]; use crate::error;