starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-09 16:05:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

improve concurrency limits II (#5485)

Browse Source
This commit is contained in:
Ruben Fiszel
2025-03-17 17:24:50 +01:00
committed by GitHub
parent eee7d33bd8
commit 8a7b119d4d
20 changed files with 207 additions and 237 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-05cb171b610bfb45f6228128a385cde8a5b86d7ca377a028004cc382e12faf41.json
Generated
+14
View File
@@ -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"
}
backend/.sqlx/query-0a1c95c4376b944661bab13271091cf3ea0afe68fb8e08e7aea239dc735c625c.json
Generated
-15
View File
@@ -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"
}
backend/.sqlx/query-1bceaf6e9f25745b7f70128054ca81d68f3d56d4782e99e05b4f1cb362683514.json
Generated
+15
View File
@@ -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"
}
backend/.sqlx/query-38a3fbc28e827d08a928d441274c5eb28780abc8adffcc7175f6c8d4ff8849ca.json
Generated
-23
View File
@@ -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"
}
backend/.sqlx/query-3fa3d1fa1add8e187fcbaf7351b721ad0f3e2888af207e8830ccf5e921c5fd60.json
Generated
+16
View File
@@ -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"
}
backend/.sqlx/query-5930b2fa72fd15d692bcf3e14d95f85dd67ab1514d7c48668ab2f10aa6b201ba.json
Generated
+14
View File
@@ -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"
}
backend/.sqlx/query-ab9e47e5b510e7df5a41db12896675393a6bb27f8e14245410751961218a7df5.json → backend/.sqlx/query-6b6f8f7b4a6b6e7e41a9da8b6dfdbcae842ff252cc355bd91aeeb5e26dcc74f3.json
Generated
+2 -2
View File
@@ -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"
}
backend/.sqlx/query-7145a1a6cf0ccfae5874c882e63ff0d370cf1ab4d41f68494a940951eaa52bc3.json
Generated
-15
View File
@@ -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"
}
backend/.sqlx/query-7f6d6952abee71fb6bb5604766c91e9605494a797b9c01c54fbdbe8949ab625d.json
Generated
+15
View File
@@ -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"
}
backend/.sqlx/query-7b1e6b67a20ae1128118d5f5cc0db4007fb9dc6fd20582a46ebb951fca3a7abd.json → backend/.sqlx/query-8c2541cdfb84bfdbdc28285641166fe4c284dd6ed5245fbb90650d99afbf3812.json
Generated
+2 -2
View File
@@ -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"
}
backend/.sqlx/query-b6d657d98de40a25b2bf66471c376d8a2424dce6c4ab2a43cf863911c833fd30.json
Generated
-24
View File
@@ -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"
}
backend/.sqlx/query-c013880b8567bf5545b0c93ffb4b69a1e6798103df08344fbc9719e53f336c29.json
Generated
-22
View File
@@ -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"
}
backend/.sqlx/query-cbb93da0b7719a27d2ae1ec0f653322cae965dc5d8ebc98f69d5922fa1192561.json
Generated
+22
View File
@@ -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"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
9a3f58425136b0f9e9c8106151537b1e271ec838
280b866cd0b179c16c894197a891f24495a26549
backend/src/main.rs
+17 -4
View File
@@ -116,9 +116,10 @@ where
}
lazy_static::lazy_static! {
static ref PG_LISTENER_REFRESH_PERIOD_SECS: Option<u64> = 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::<u64>().ok());
.and_then(|x| x.parse::<u64>().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();
}
backend/windmill-api/src/jobs.rs
+7
View File
@@ -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,
backend/windmill-queue/src/jobs.rs
+60 -128
View File
@@ -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::<serde_json::Value>(
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<i64> = 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<i64>)> {
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<Postgres>,
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,
)
backend/windmill-queue/src/jobs_ee.rs
+16
View File
@@ -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<DateTime<Utc>>)> {
Ok((true, None))
}
backend/windmill-queue/src/lib.rs
+1 -1
View File
@@ -7,6 +7,6 @@
*/
mod jobs;
pub mod jobs_ee;
pub mod schedule;
pub use jobs::*;
frontend/src/lib/components/runs/RunRow.svelte
+5
View File
@@ -134,6 +134,11 @@
Scheduled for {displayDate(job.scheduled_for)}
{:else if job.canceled}
Cancelling job... (created <TimeAgo agoOnlyIfRecent date={job.created_at || ''} />)
{:else if `scheduled_for` in job && job.scheduled_for && forLater(job.scheduled_for)}
Waiting for executor (scheduled for <TimeAgo
agoOnlyIfRecent
date={job.scheduled_for || ''}
/>)
{:else}
Waiting for executor (created <TimeAgo agoOnlyIfRecent date={job.created_at || ''} />)
{/if}