kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: improve job deletion performance and batching

Browse Source
This commit is contained in:
Ruben Fiszel
2026-01-17 07:33:34 +00:00
parent 4f06b0f22f
commit 30abc3ff22
10 changed files with 243 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json generated
View File

@@ -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"
}

14
backend/.sqlx/query-1ccc9db6c6cb51bebeb90d003e12c8f7922450515d1eaee8b3abee5b31f72527.json generated Normal file
View File

@@ -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"
}

23
backend/.sqlx/query-306e0156ee1541710c1c6512ecb4f61baeb3ae6f31ba3fd57a3ec485108a7f49.json generated Normal file
View File

@@ -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"
}

4
backend/.sqlx/query-c6ef0acdf20bd71dd26de981fb49f178ba8a1b8c1e01e0fec1dfd6a54ea7a894.json → backend/.sqlx/query-45d0552a169b63241a0ecc0e7ec6a533d7eec520376b99b2fd8000f0f0acba7b.json generated
View File

@@ -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"
}

4
backend/.sqlx/query-70a6880960d17218bc5bf05287e2a6d9a6393c6bb1783ab8903d87dd099e236b.json → backend/.sqlx/query-4d80985dd0794a01a2af18ae7abf4a3ab8ba3d162ed5d04735caea7295da0b20.json generated
View File

@@ -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<str>\", 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<str>\", 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"
}

22
backend/.sqlx/query-5820d34be1a7f7b72e656c692f53146f45ad4a6e584e917a0a86280d8f473c10.json generated
View File

@@ -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"
}

38
backend/.sqlx/query-598805056a3b102c9c2fd69d8330fa78b06106236522e48ab5fe009cac059468.json generated Normal file
View File

@@ -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"
}

14
backend/.sqlx/query-9da0cea2a5d0464ca78cfeccf6cedf2b1c0e6e6cb3c9183a937a68465debdb06.json generated
View File

@@ -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"
}

256
backend/src/monitor.rs
View File

@@ -155,6 +155,20 @@ lazy_static::lazy_static! {
.ok()
.and_then(|x| x.parse::<u64>().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::<i64>().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::<i32>().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<usize> {
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<Uuid> = 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<String>,
tmp_dir: &str,
@@ -2262,7 +2330,7 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, 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()
)

2
backend/windmill-common/src/global_settings.rs
View File

@@ -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;