diff --git a/backend/.sqlx/query-05cb171b610bfb45f6228128a385cde8a5b86d7ca377a028004cc382e12faf41.json b/backend/.sqlx/query-05cb171b610bfb45f6228128a385cde8a5b86d7ca377a028004cc382e12faf41.json new file mode 100644 index 0000000000..5290fab4c4 --- /dev/null +++ b/backend/.sqlx/query-05cb171b610bfb45f6228128a385cde8a5b86d7ca377a028004cc382e12faf41.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO concurrency_counter(concurrency_id, job_uuids) \n VALUES ($1, '{}'::jsonb)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "05cb171b610bfb45f6228128a385cde8a5b86d7ca377a028004cc382e12faf41" +} diff --git a/backend/.sqlx/query-0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c.json b/backend/.sqlx/query-0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c.json deleted file mode 100644 index b8f224dbc2..0000000000 --- a/backend/.sqlx/query-0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c" -} diff --git a/backend/.sqlx/query-1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514.json b/backend/.sqlx/query-1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514.json new file mode 100644 index 0000000000..1be74d47f2 --- /dev/null +++ b/backend/.sqlx/query-1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH inserted_concurrency_counter AS (\n INSERT INTO concurrency_counter (concurrency_id, job_uuids) \n VALUES ($1, '{}'::jsonb)\n ON CONFLICT DO NOTHING\n )\n INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514" +} diff --git a/backend/.sqlx/query-38a3fbc28e827d08a928d441274c5eb28780abc8adffcc7175f6c8d4ff8849ca.json b/backend/.sqlx/query-38a3fbc28e827d08a928d441274c5eb28780abc8adffcc7175f6c8d4ff8849ca.json deleted file mode 100644 index a3929e5e4b..0000000000 --- a/backend/.sqlx/query-38a3fbc28e827d08a928d441274c5eb28780abc8adffcc7175f6c8d4ff8849ca.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1 RETURNING (SELECT COUNT(*) FROM jsonb_object_keys(job_uuids))", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "count", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "38a3fbc28e827d08a928d441274c5eb28780abc8adffcc7175f6c8d4ff8849ca" -} diff --git a/backend/.sqlx/query-3fa3d1fa1add8e187fcbaf7351b721ad0f3e2888af207e8830ccf5e921c5fd60.json b/backend/.sqlx/query-3fa3d1fa1add8e187fcbaf7351b721ad0f3e2888af207e8830ccf5e921c5fd60.json new file mode 100644 index 0000000000..3bb4151982 --- /dev/null +++ b/backend/.sqlx/query-3fa3d1fa1add8e187fcbaf7351b721ad0f3e2888af207e8830ccf5e921c5fd60.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO concurrency_counter(concurrency_id, job_uuids) \n VALUES ($1, $2)\n ON CONFLICT (concurrency_id)\n DO UPDATE SET job_uuids = jsonb_set(concurrency_counter.job_uuids, array[$3], '{}')", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Jsonb", + "Text" + ] + }, + "nullable": [] + }, + "hash": "3fa3d1fa1add8e187fcbaf7351b721ad0f3e2888af207e8830ccf5e921c5fd60" +} diff --git a/backend/.sqlx/query-5930b2fa72fd15d692bcf3e14d95f85dd67ab1514d7c48668ab2f10aa6b201ba.json b/backend/.sqlx/query-5930b2fa72fd15d692bcf3e14d95f85dd67ab1514d7c48668ab2f10aa6b201ba.json new file mode 100644 index 0000000000..d616940b6b --- /dev/null +++ b/backend/.sqlx/query-5930b2fa72fd15d692bcf3e14d95f85dd67ab1514d7c48668ab2f10aa6b201ba.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_runtime SET ping = now() WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "5930b2fa72fd15d692bcf3e14d95f85dd67ab1514d7c48668ab2f10aa6b201ba" +} diff --git a/backend/.sqlx/query-ab9e47e5b510e7df5a41db12896675393a6bb27f8e14245410751961218a7df5.json b/backend/.sqlx/query-6b6f8f7b4a6b6e7e41a9da8b6dfdbcae842ff252cc355bd91aeeb5e26dcc74f3.json similarity index 54% rename from backend/.sqlx/query-ab9e47e5b510e7df5a41db12896675393a6bb27f8e14245410751961218a7df5.json rename to backend/.sqlx/query-6b6f8f7b4a6b6e7e41a9da8b6dfdbcae842ff252cc355bd91aeeb5e26dcc74f3.json index 868541a175..80619c8b2a 100644 --- a/backend/.sqlx/query-ab9e47e5b510e7df5a41db12896675393a6bb27f8e14245410751961218a7df5.json +++ b/backend/.sqlx/query-6b6f8f7b4a6b6e7e41a9da8b6dfdbcae842ff252cc355bd91aeeb5e26dcc74f3.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT COALESCE((SELECT MIN(started_at) as min_started_at\n FROM v2_as_queue\n WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $2 AND canceled = false AND concurrent_limit > 0), $3) as min_started_at, now() AS now", + "query": "SELECT COALESCE((SELECT MIN(started_at) as min_started_at\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id\n WHERE v2_job.runnable_path = $1 AND v2_job.kind != 'dependencies' AND v2_job_queue.running = true AND v2_job_queue.workspace_id = $2 AND v2_job_queue.canceled_by IS NULL AND v2_job.concurrent_limit > 0), $3) as min_started_at, now() AS now", "describe": { "columns": [ { @@ -26,5 +26,5 @@ null ] }, - "hash": "ab9e47e5b510e7df5a41db12896675393a6bb27f8e14245410751961218a7df5" + "hash": "6b6f8f7b4a6b6e7e41a9da8b6dfdbcae842ff252cc355bd91aeeb5e26dcc74f3" } diff --git a/backend/.sqlx/query-7145a1a6cf0ccfae5874c882e63ff0d370cf1ab4d41f68494a940951eaa52bc3.json b/backend/.sqlx/query-7145a1a6cf0ccfae5874c882e63ff0d370cf1ab4d41f68494a940951eaa52bc3.json deleted file mode 100644 index edec8c5053..0000000000 --- a/backend/.sqlx/query-7145a1a6cf0ccfae5874c882e63ff0d370cf1ab4d41f68494a940951eaa52bc3.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "WITH ping AS (UPDATE v2_job_runtime SET ping = NULL WHERE id = $2 RETURNING id)\n UPDATE v2_job_queue SET\n running = false,\n started_at = null,\n scheduled_for = $1\n WHERE id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Timestamptz", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "7145a1a6cf0ccfae5874c882e63ff0d370cf1ab4d41f68494a940951eaa52bc3" -} diff --git a/backend/.sqlx/query-7f6d6952abee71fb6bb5604766c91e9605494a797b9c01c54fbdbe8949ab625d.json b/backend/.sqlx/query-7f6d6952abee71fb6bb5604766c91e9605494a797b9c01c54fbdbe8949ab625d.json new file mode 100644 index 0000000000..0667ac7d65 --- /dev/null +++ b/backend/.sqlx/query-7f6d6952abee71fb6bb5604766c91e9605494a797b9c01c54fbdbe8949ab625d.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH ping AS (\n UPDATE v2_job_runtime SET ping = null WHERE id = $2\n )\n UPDATE v2_job_queue SET\n running = false,\n started_at = null,\n scheduled_for = $1\n WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Timestamptz", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "7f6d6952abee71fb6bb5604766c91e9605494a797b9c01c54fbdbe8949ab625d" +} diff --git a/backend/.sqlx/query-7b1e6b67a20ae1128118d5f5cc0db4007fb9dc6fd20582a46ebb951fca3a7abd.json b/backend/.sqlx/query-8c2541cdfb84bfdbdc28285641166fe4c284dd6ed5245fbb90650d99afbf3812.json similarity index 58% rename from backend/.sqlx/query-7b1e6b67a20ae1128118d5f5cc0db4007fb9dc6fd20582a46ebb951fca3a7abd.json rename to backend/.sqlx/query-8c2541cdfb84bfdbdc28285641166fe4c284dd6ed5245fbb90650d99afbf3812.json index a094b8580f..88dcd47312 100644 --- a/backend/.sqlx/query-7b1e6b67a20ae1128118d5f5cc0db4007fb9dc6fd20582a46ebb951fca3a7abd.json +++ b/backend/.sqlx/query-8c2541cdfb84bfdbdc28285641166fe4c284dd6ed5245fbb90650d99afbf3812.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT CAST(ROUND(AVG(duration_ms), 0) AS BIGINT) AS avg_duration_s FROM\n (SELECT duration_ms FROM concurrency_key LEFT JOIN v2_as_completed_job ON v2_as_completed_job.id = concurrency_key.job_id WHERE key = $1 AND ended_at IS NOT NULL\n ORDER BY ended_at\n DESC LIMIT 10) AS t", + "query": "SELECT CAST(ROUND(AVG(duration_ms), 0) AS BIGINT) AS avg_duration_s FROM\n (SELECT duration_ms FROM concurrency_key LEFT JOIN v2_job_completed ON v2_job_completed.id = concurrency_key.job_id WHERE key = $1 AND ended_at IS NOT NULL\n ORDER BY ended_at\n DESC LIMIT 10) AS t", "describe": { "columns": [ { @@ -18,5 +18,5 @@ null ] }, - "hash": "7b1e6b67a20ae1128118d5f5cc0db4007fb9dc6fd20582a46ebb951fca3a7abd" + "hash": "8c2541cdfb84bfdbdc28285641166fe4c284dd6ed5245fbb90650d99afbf3812" } diff --git a/backend/.sqlx/query-b6d657d98de40a25b2bf66471c376d8a2424dce6c4ab2a43cf863911c833fd30.json b/backend/.sqlx/query-b6d657d98de40a25b2bf66471c376d8a2424dce6c4ab2a43cf863911c833fd30.json deleted file mode 100644 index c775d232e1..0000000000 --- a/backend/.sqlx/query-b6d657d98de40a25b2bf66471c376d8a2424dce6c4ab2a43cf863911c833fd30.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO concurrency_counter(concurrency_id, job_uuids) \n VALUES ($1, $2)\n ON CONFLICT (concurrency_id) \n DO UPDATE SET job_uuids = jsonb_set(concurrency_counter.job_uuids, array[$3], '{}')\n RETURNING (SELECT COUNT(*) FROM jsonb_object_keys(job_uuids))", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "count", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Varchar", - "Jsonb", - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "b6d657d98de40a25b2bf66471c376d8a2424dce6c4ab2a43cf863911c833fd30" -} diff --git a/backend/.sqlx/query-c013880b8567bf5545b0c93ffb4b69a1e6798103df08344fbc9719e53f336c29.json b/backend/.sqlx/query-c013880b8567bf5545b0c93ffb4b69a1e6798103df08344fbc9719e53f336c29.json deleted file mode 100644 index 904c4eba5f..0000000000 --- a/backend/.sqlx/query-c013880b8567bf5545b0c93ffb4b69a1e6798103df08344fbc9719e53f336c29.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT job_uuids FROM concurrency_counter \n WHERE concurrency_id = $1 \n FOR UPDATE", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "job_uuids", - "type_info": "Jsonb" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "c013880b8567bf5545b0c93ffb4b69a1e6798103df08344fbc9719e53f336c29" -} diff --git a/backend/.sqlx/query-cbb93da0b7719a27d2ae1ec0f653322cae965dc5d8ebc98f69d5922fa1192561.json b/backend/.sqlx/query-cbb93da0b7719a27d2ae1ec0f653322cae965dc5d8ebc98f69d5922fa1192561.json new file mode 100644 index 0000000000..684e857290 --- /dev/null +++ b/backend/.sqlx/query-cbb93da0b7719a27d2ae1ec0f653322cae965dc5d8ebc98f69d5922fa1192561.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT COALESCE(\n (SELECT COUNT(*) FROM jsonb_object_keys(job_uuids)),\n 0\n )\n FROM concurrency_counter \n WHERE concurrency_id = $1\n FOR UPDATE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "coalesce", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "cbb93da0b7719a27d2ae1ec0f653322cae965dc5d8ebc98f69d5922fa1192561" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 5d5699c639..1cd709f740 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -9a3f58425136b0f9e9c8106151537b1e271ec838 \ No newline at end of file +280b866cd0b179c16c894197a891f24495a26549 \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index afec50c6a9..a2762405bc 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -116,9 +116,10 @@ where } lazy_static::lazy_static! { - static ref PG_LISTENER_REFRESH_PERIOD_SECS: Option = std::env::var("PG_LISTENER_REFRESH_PERIOD_SECS") + static ref PG_LISTENER_REFRESH_PERIOD_SECS: u64 = std::env::var("PG_LISTENER_REFRESH_PERIOD_SECS") .ok() - .and_then(|x| x.parse::().ok()); + .and_then(|x| x.parse::().ok()) + .unwrap_or(3600 * 12); } pub fn main() -> anyhow::Result<()> { @@ -866,12 +867,24 @@ Windmill Community Edition {GIT_VERSION} }; }, _ = tokio::time::sleep(Duration::from_secs(30)) => { - if PG_LISTENER_REFRESH_PERIOD_SECS.is_some_and(|x| last_listener_refresh.elapsed() > Duration::from_secs(x)) { - tracing::info!("Refreshing pg listener"); + if last_listener_refresh.elapsed() > Duration::from_secs(*PG_LISTENER_REFRESH_PERIOD_SECS) { + tracing::info!("Refreshing pg listeners, settings and license key after {}s", Duration::from_secs(*PG_LISTENER_REFRESH_PERIOD_SECS).as_secs()); if let Err(e) = listener.unlisten_all().await { tracing::error!(error = %e, "Could not unlisten to database"); } listener = retry_listen_pg(&db_url).await; + initial_load( + &db, + tx.clone(), + worker_mode, + server_mode, + #[cfg(feature = "parquet")] + disable_s3_store, + ) + .await; + if let Err(err) = reload_license_key(&db).await { + tracing::error!("Failed to reload license key: {err:#}"); + } last_listener_refresh = Instant::now(); } diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index ad0376fe51..5148359981 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -4968,6 +4968,13 @@ async fn add_batch_jobs( } if let Some(custom_concurrency_key) = custom_concurrency_key { + sqlx::query!( + "INSERT INTO concurrency_counter(concurrency_id, job_uuids) + VALUES ($1, '{}'::jsonb)", + &custom_concurrency_key + ) + .execute(&mut *tx) + .await?; sqlx::query!( "INSERT INTO concurrency_key (job_id, key) SELECT id, $1 FROM unnest($2::uuid[]) as id", custom_concurrency_key, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 2a183e4631..f0eda8328a 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -62,6 +62,7 @@ use windmill_common::BASE_URL; #[cfg(feature = "cloud")] use windmill_common::users::SUPERADMIN_SYNC_EMAIL; +use crate::jobs_ee::update_concurrency_counter; use crate::schedule::{get_schedule_opt, push_scheduled_job}; #[cfg(feature = "prometheus")] @@ -1887,11 +1888,6 @@ pub async fn pull( return Ok((Option::Some(pulled_job), suspended)); } - let mut tx = db.begin().await?; - - // Else the job is subject to concurrency limits - let job_script_path = pulled_job.script_path.clone().unwrap(); - let job_concurrency_key = match concurrency_key(db, &pulled_job).await { Ok(key) => key, Err(e) => { @@ -1915,104 +1911,67 @@ pub async fn pull( job_custom_concurrency_time_window_s ); - sqlx::query_scalar!( - "SELECT null FROM v2_job_queue WHERE id = $1 FOR UPDATE", - pulled_job.id - ) - .fetch_one(&mut *tx) - .await - .context("lock job in queue")?; - let jobs_uuids_init_json_value = serde_json::from_str::( format!("{{\"{}\": {{}}}}", pulled_job.id.hyphenated().to_string()).as_str(), ) .expect("Unable to serialize job_uuids column to proper JSON"); - let (mut tx, running_job) = update_concurrency_counter( - tx, - job_concurrency_key.clone(), - jobs_uuids_init_json_value, - pulled_job.id.hyphenated().to_string(), - ) - .await?; - tracing::debug!("running_job: {}", running_job.unwrap_or(0)); - let completed_count = sqlx::query!( - "SELECT COUNT(*) as count, COALESCE(MAX(ended_at), now() - INTERVAL '1 second' * $2) as max_ended_at FROM concurrency_key WHERE key = $1 AND ended_at >= (now() - INTERVAL '1 second' * $2)", - job_concurrency_key, - f64::from(job_custom_concurrency_time_window_s), - ).fetch_one(&mut *tx).await.map_err(|e| { - Error::internal_err(format!( - "Error getting completed count for key {job_concurrency_key}: {e:#}" - )) - })?; - - let min_started_at = sqlx::query!( - "SELECT COALESCE((SELECT MIN(started_at) as min_started_at - FROM v2_as_queue - WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $2 AND canceled = false AND concurrent_limit > 0), $3) as min_started_at, now() AS now", - job_script_path, - &pulled_job.workspace_id, - completed_count.max_ended_at - ) - .fetch_one(&mut *tx) - .await - .map_err(|e| { - Error::internal_err(format!( - "Error getting concurrency count for script path {job_script_path}: {e:#}" - )) - })?; - - let concurrent_jobs_for_this_script = - completed_count.count.unwrap_or_default() as i32 + running_job.unwrap_or(0) as i32; - tracing::debug!( - "Current concurrent jobs for this script: {}", - concurrent_jobs_for_this_script - ); - if concurrent_jobs_for_this_script <= job_custom_concurrent_limit { + let (within_limit, max_ended_at) = if *DISABLE_CONCURRENCY_LIMIT { + tracing::warn!("Concurrency limit is disabled, skipping"); + (true, None) + } else { + update_concurrency_counter( + db, + &pulled_job.id, + job_concurrency_key.clone(), + jobs_uuids_init_json_value, + pulled_job.id.hyphenated().to_string(), + job_custom_concurrency_time_window_s, + job_custom_concurrent_limit, + ) + .await? + }; + if within_limit { #[cfg(feature = "prometheus")] if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { QUEUE_PULL_COUNT.inc(); } - tx.commit().await?; return Ok((Option::Some(pulled_job), suspended)); } - if *DISABLE_CONCURRENCY_LIMIT { - tracing::warn!("Concurrency limit is disabled, skipping"); - } else { - let x = sqlx::query_scalar!( - "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1 RETURNING (SELECT COUNT(*) FROM jsonb_object_keys(job_uuids))", - job_concurrency_key, - pulled_job.id.hyphenated().to_string(), + let job_script_path = pulled_job.script_path.clone().unwrap_or_default(); + + let min_started_at = sqlx::query!( + "SELECT COALESCE((SELECT MIN(started_at) as min_started_at + FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id + WHERE v2_job.runnable_path = $1 AND v2_job.kind != 'dependencies' AND v2_job_queue.running = true AND v2_job_queue.workspace_id = $2 AND v2_job_queue.canceled_by IS NULL AND v2_job.concurrent_limit > 0), $3) as min_started_at, now() AS now", + job_script_path, + &pulled_job.workspace_id, + max_ended_at ) - .fetch_one(&mut *tx) + .fetch_one(db) .await .map_err(|e| { Error::internal_err(format!( - "Error decreasing concurrency count for script path {job_script_path}: {e:#}" + "Error getting min started at for script path {job_script_path}: {e:#}" )) })?; - tracing::debug!("running_job after decrease: {}", x.unwrap_or(0)); - } let job_uuid: Uuid = pulled_job.id; let avg_script_duration: Option = sqlx::query_scalar!( "SELECT CAST(ROUND(AVG(duration_ms), 0) AS BIGINT) AS avg_duration_s FROM - (SELECT duration_ms FROM concurrency_key LEFT JOIN v2_as_completed_job ON v2_as_completed_job.id = concurrency_key.job_id WHERE key = $1 AND ended_at IS NOT NULL + (SELECT duration_ms FROM concurrency_key LEFT JOIN v2_job_completed ON v2_job_completed.id = concurrency_key.job_id WHERE key = $1 AND ended_at IS NOT NULL ORDER BY ended_at DESC LIMIT 10) AS t", job_concurrency_key ) - .fetch_one(&mut *tx) + .fetch_one(db) .await?; - tracing::debug!("avg script duration computed: {:?}", avg_script_duration); + tracing::debug!( + "avg script duration computed: {}", + avg_script_duration.unwrap_or(0) + ); - // let before_me = sqlx::query!( - // "SELECT schedu FROM queue WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $2 AND canceled = false AND started_at < $3 ORDER BY started_at DESC LIMIT 1", - // job_script_path, - // &pulled_job.workspace_id, - // min_started_at.now.unwrap() - // ) // optimal scheduling is: 'older_job_in_concurrency_time_window_started_timestamp + script_avg_duration + concurrency_time_window_s' let inc = Duration::try_milliseconds( avg_script_duration.map(|x| i64::from(x + 100)).unwrap_or(0), @@ -2023,8 +1982,9 @@ pub async fn pull( .unwrap_or_default(); let now = min_started_at.now.unwrap(); - let min_started_p_inc = (min_started_at.min_started_at.unwrap_or(now) + inc) - .max(now + Duration::try_seconds(3).unwrap_or_default()); + let min_started_at_or_now = min_started_at.min_started_at.unwrap_or(now); + let min_started_p_inc = + (min_started_at_or_now + inc).max(now + Duration::try_seconds(3).unwrap_or_default()); let mut estimated_next_schedule_timestamp = min_started_p_inc; let all_jobs = sqlx::query_scalar!( @@ -2032,17 +1992,21 @@ pub async fn pull( WHERE key = $1 AND running = false AND canceled_by IS NULL AND scheduled_for >= $2", job_concurrency_key, estimated_next_schedule_timestamp - inc - ).fetch_all(&mut *tx).await?; + ).fetch_all(db).await?; + tracing::debug!( + "all_jobs: {:?}, estimated_next_schedule_timestamp: {:?}, inc: {:?}", + all_jobs, + estimated_next_schedule_timestamp, + inc + ); let mut i = 0; loop { let jobs_in_window = all_jobs .iter() - .filter(|&scheduled_for| { - scheduled_for >= &(estimated_next_schedule_timestamp - inc) - && scheduled_for < &estimated_next_schedule_timestamp - }) - .count() as i32; + .filter(|&scheduled_for| scheduled_for <= &estimated_next_schedule_timestamp) + .count() as i32 + - (job_custom_concurrent_limit * i); tracing::debug!("estimated_next_schedule_timestamp: {:?}, jobs_in_window: {jobs_in_window}, inc: {inc}", estimated_next_schedule_timestamp); @@ -2054,18 +2018,20 @@ pub async fn pull( } } - tracing::info!("Job '{}' from path '{}' with concurrency key '{}' has reached its concurrency limit of {} jobs run in the last {} seconds. This job will be re-queued for next execution at {} (avg script duration: {:?}, number of time windows full: {})", + tracing::info!("Job '{}' from path '{}' with concurrency key '{}' has reached its concurrency limit of {} jobs run in the last {} seconds. This job will be re-queued for next execution at {} (min_started_at: {min_started_at_or_now}, avg script duration: {:?}, number of time windows full: {})", job_uuid, job_script_path, job_concurrency_key, job_custom_concurrent_limit, job_custom_concurrency_time_window_s, estimated_next_schedule_timestamp, avg_script_duration, i); let job_log_event = format!( - "\nRe-scheduled job to {estimated_next_schedule_timestamp} due to concurrency limits with key {job_concurrency_key} and limit {job_custom_concurrent_limit} in the last {job_custom_concurrency_time_window_s} seconds (avg script duration: {:?}, number of time windows full: {})", + "\nRe-scheduled job to {estimated_next_schedule_timestamp} due to concurrency limits with key {job_concurrency_key} and limit {job_custom_concurrent_limit} in the last {job_custom_concurrency_time_window_s} seconds (min_started_at: {min_started_at_or_now}, avg script duration: {:?}, number of time windows full: {})\n", avg_script_duration, i ); let _ = append_logs(&job_uuid, &pulled_job.workspace_id, job_log_event, db).await; - // if using posgtres, then we're able to re-queue the entire batch of scheduled job for this script_path, so we do it sqlx::query!( - "WITH ping AS (UPDATE v2_job_runtime SET ping = NULL WHERE id = $2 RETURNING id) + " + WITH ping AS ( + UPDATE v2_job_runtime SET ping = null WHERE id = $2 + ) UPDATE v2_job_queue SET running = false, started_at = null, @@ -2074,51 +2040,12 @@ pub async fn pull( estimated_next_schedule_timestamp, job_uuid, ) - .execute(&mut *tx) + .execute(db) .await .map_err(|e| Error::internal_err(format!("Could not update and re-queue job {job_uuid}. The job will be marked as running but it is not running: {e:#}")))?; - - tx.commit().await? } } -async fn update_concurrency_counter<'c>( - mut tx: Transaction<'c, sqlx::Postgres>, - job_concurrency_key: String, - jobs_uuids_init_json_value: serde_json::Value, - pulled_job_id: String, -) -> anyhow::Result<(Transaction<'c, sqlx::Postgres>, Option)> { - if *DISABLE_CONCURRENCY_LIMIT { - tracing::warn!("Concurrency limit is disabled, skipping"); - return Ok((tx, None)); - } - // 1. Try to lock the row first - let _ = sqlx::query!( - "SELECT job_uuids FROM concurrency_counter - WHERE concurrency_id = $1 - FOR UPDATE", - job_concurrency_key - ) - .fetch_optional(&mut *tx) - .await?; - - // 2. Insert if missing, otherwise update - let running_job = sqlx::query_scalar!( - "INSERT INTO concurrency_counter(concurrency_id, job_uuids) - VALUES ($1, $2) - ON CONFLICT (concurrency_id) - DO UPDATE SET job_uuids = jsonb_set(concurrency_counter.job_uuids, array[$3], '{}') - RETURNING (SELECT COUNT(*) FROM jsonb_object_keys(job_uuids))", - job_concurrency_key, - jobs_uuids_init_json_value, - pulled_job_id - ) - .fetch_one(&mut *tx) - .await?; - - Ok((tx, running_job)) -} - async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>( db: &Pool, suspend_first: bool, @@ -3767,7 +3694,12 @@ pub async fn push<'c, 'd>( &job_kind, )); sqlx::query!( - "INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)", + "WITH inserted_concurrency_counter AS ( + INSERT INTO concurrency_counter (concurrency_id, job_uuids) + VALUES ($1, '{}'::jsonb) + ON CONFLICT DO NOTHING + ) + INSERT INTO concurrency_key(key, job_id) VALUES ($1, $2)", concurrency_key, job_id, ) diff --git a/backend/windmill-queue/src/jobs_ee.rs b/backend/windmill-queue/src/jobs_ee.rs new file mode 100644 index 0000000000..ea1f86b529 --- /dev/null +++ b/backend/windmill-queue/src/jobs_ee.rs @@ -0,0 +1,16 @@ +use chrono::{DateTime, Utc}; +use uuid::Uuid; +use windmill_common::DB; + +#[allow(dead_code)] +pub(crate) async fn update_concurrency_counter( + _db: &DB, + _job_id: &Uuid, + _job_concurrency_key: String, + _jobs_uuids_init_json_value: serde_json::Value, + _pulled_job_id: String, + _job_custom_concurrency_time_window_s: i32, + _limit: i32, +) -> anyhow::Result<(bool, Option>)> { + Ok((true, None)) +} diff --git a/backend/windmill-queue/src/lib.rs b/backend/windmill-queue/src/lib.rs index b47a3c3238..c2d022d16a 100644 --- a/backend/windmill-queue/src/lib.rs +++ b/backend/windmill-queue/src/lib.rs @@ -7,6 +7,6 @@ */ mod jobs; +pub mod jobs_ee; pub mod schedule; - pub use jobs::*; diff --git a/frontend/src/lib/components/runs/RunRow.svelte b/frontend/src/lib/components/runs/RunRow.svelte index 1ea568034e..4430708091 100644 --- a/frontend/src/lib/components/runs/RunRow.svelte +++ b/frontend/src/lib/components/runs/RunRow.svelte @@ -134,6 +134,11 @@ Scheduled for {displayDate(job.scheduled_for)} {:else if job.canceled} Cancelling job... (created ) + {:else if `scheduled_for` in job && job.scheduled_for && forLater(job.scheduled_for)} + Waiting for executor (scheduled for ) {:else} Waiting for executor (created ) {/if}