diff --git a/backend/.sqlx/query-0be8c6f698a715d015126a7871a39fd7c25ad5269d3313ee1a32533b6d346446.json b/backend/.sqlx/query-0be8c6f698a715d015126a7871a39fd7c25ad5269d3313ee1a32533b6d346446.json new file mode 100644 index 0000000000..27ee0f7d23 --- /dev/null +++ b/backend/.sqlx/query-0be8c6f698a715d015126a7871a39fd7c25ad5269d3313ee1a32533b6d346446.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "0be8c6f698a715d015126a7871a39fd7c25ad5269d3313ee1a32533b6d346446" +} diff --git a/backend/.sqlx/query-38a3fbc28e827d08a928d441274c5eb28780abc8adffcc7175f6c8d4ff8849ca.json b/backend/.sqlx/query-38a3fbc28e827d08a928d441274c5eb28780abc8adffcc7175f6c8d4ff8849ca.json new file mode 100644 index 0000000000..a3929e5e4b --- /dev/null +++ b/backend/.sqlx/query-38a3fbc28e827d08a928d441274c5eb28780abc8adffcc7175f6c8d4ff8849ca.json @@ -0,0 +1,23 @@ +{ + "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-3e3d12a51cb524fbd3d6949e150cb608acfbe8c8eade1939e813086380c205e0.json b/backend/.sqlx/query-3e3d12a51cb524fbd3d6949e150cb608acfbe8c8eade1939e813086380c205e0.json new file mode 100644 index 0000000000..d6fd72a8b5 --- /dev/null +++ b/backend/.sqlx/query-3e3d12a51cb524fbd3d6949e150cb608acfbe8c8eade1939e813086380c205e0.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO concurrency_counter(concurrency_id, job_uuids) 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": "3e3d12a51cb524fbd3d6949e150cb608acfbe8c8eade1939e813086380c205e0" +} diff --git a/backend/.sqlx/query-467247beca4dae321ab3acc3a99f655412761157cda54bbd3e58cada3de3396c.json b/backend/.sqlx/query-467247beca4dae321ab3acc3a99f655412761157cda54bbd3e58cada3de3396c.json deleted file mode 100644 index cdd02265a1..0000000000 --- a/backend/.sqlx/query-467247beca4dae321ab3acc3a99f655412761157cda54bbd3e58cada3de3396c.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE concurrency_counter SET counter = counter - 1 WHERE concurrency_id = $1", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [] - }, - "hash": "467247beca4dae321ab3acc3a99f655412761157cda54bbd3e58cada3de3396c" -} diff --git a/backend/.sqlx/query-71879f3644dde27f2d93e5511e420579dfd2ffab72bb8480a1622b68703d4ea7.json b/backend/.sqlx/query-71879f3644dde27f2d93e5511e420579dfd2ffab72bb8480a1622b68703d4ea7.json deleted file mode 100644 index 2c9b770df5..0000000000 --- a/backend/.sqlx/query-71879f3644dde27f2d93e5511e420579dfd2ffab72bb8480a1622b68703d4ea7.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE concurrency_counter SET counter = counter - 1 WHERE concurrency_id = $1 RETURNING counter", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "counter", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "71879f3644dde27f2d93e5511e420579dfd2ffab72bb8480a1622b68703d4ea7" -} diff --git a/backend/.sqlx/query-86652246b20f9bd023d3419ec382176780a87f32673b53de6f899564c1494147.json b/backend/.sqlx/query-86652246b20f9bd023d3419ec382176780a87f32673b53de6f899564c1494147.json deleted file mode 100644 index cee6f85329..0000000000 --- a/backend/.sqlx/query-86652246b20f9bd023d3419ec382176780a87f32673b53de6f899564c1494147.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO concurrency_counter VALUES ($1, 1)\n ON CONFLICT (concurrency_id) \n DO UPDATE SET counter = concurrency_counter.counter + 1\n RETURNING concurrency_counter.counter", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "counter", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Varchar" - ] - }, - "nullable": [ - false - ] - }, - "hash": "86652246b20f9bd023d3419ec382176780a87f32673b53de6f899564c1494147" -} diff --git a/backend/migrations/20231023135159_concurrency_job_uuids.down.sql b/backend/migrations/20231023135159_concurrency_job_uuids.down.sql new file mode 100644 index 0000000000..e886804a29 --- /dev/null +++ b/backend/migrations/20231023135159_concurrency_job_uuids.down.sql @@ -0,0 +1,3 @@ +-- Add down migration script here +ALTER TABLE concurrency_counter ADD COLUMN counter INTEGER NOT NULL DEFAULT 0; +ALTER TABLE concurrency_counter DROP COLUMN job_uuids; diff --git a/backend/migrations/20231023135159_concurrency_job_uuids.up.sql b/backend/migrations/20231023135159_concurrency_job_uuids.up.sql new file mode 100644 index 0000000000..34b18cea39 --- /dev/null +++ b/backend/migrations/20231023135159_concurrency_job_uuids.up.sql @@ -0,0 +1,3 @@ +-- Add up migration script here +ALTER TABLE concurrency_counter ADD COLUMN job_uuids jsonb NOT NULL DEFAULT '{}'::jsonb; +ALTER TABLE concurrency_counter DROP COLUMN counter; diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 2beec58939..e84bc175d7 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -401,8 +401,9 @@ pub async fn add_completed_job< } if queued_job.concurrent_limit.is_some() { if let Err(e) = sqlx::query_scalar!( - "UPDATE concurrency_counter SET counter = counter - 1 WHERE concurrency_id = $1", - queued_job.full_path() + "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1", + queued_job.full_path(), + queued_job.id.hyphenated().to_string(), ) .execute(&mut tx) .await @@ -1096,12 +1097,18 @@ pub async fn pull( job_custom_concurrency_time_window_s ); + 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 running_job = sqlx::query_scalar!( - "INSERT INTO concurrency_counter VALUES ($1, 1) + "INSERT INTO concurrency_counter(concurrency_id, job_uuids) VALUES ($1, $2) ON CONFLICT (concurrency_id) - DO UPDATE SET counter = concurrency_counter.counter + 1 - RETURNING concurrency_counter.counter", + DO UPDATE SET job_uuids = jsonb_set(concurrency_counter.job_uuids, array[$3], '{}') + RETURNING (SELECT COUNT(*) FROM jsonb_object_keys(job_uuids))", pulled_job.full_path(), + jobs_uuids_init_json_value, + pulled_job.id.hyphenated().to_string(), ) .fetch_one(&mut tx) .await @@ -1110,7 +1117,7 @@ pub async fn pull( "Error getting concurrency count for script path {job_script_path}: {e}" )) })?; - tracing::debug!("running_job: {}", running_job); + tracing::debug!("running_job: {}", running_job.unwrap_or(0)); let script_path_live_stats = sqlx::query!( "SELECT COALESCE(j.min_started_at, q.min_started_at) AS min_started_at, COALESCE(completed_count, 0) AS completed_count @@ -1138,7 +1145,8 @@ pub async fn pull( })?; let concurrent_jobs_for_this_script = - script_path_live_stats.completed_count.unwrap_or_default() as i32 + running_job; + script_path_live_stats.completed_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 @@ -1151,8 +1159,10 @@ pub async fn pull( return Ok(Option::Some(pulled_job)); } let x = sqlx::query_scalar!( - "UPDATE concurrency_counter SET counter = counter - 1 WHERE concurrency_id = $1 RETURNING counter", - pulled_job.full_path() + "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1 RETURNING (SELECT COUNT(*) FROM jsonb_object_keys(job_uuids))", + pulled_job.full_path(), + pulled_job.id.hyphenated().to_string(), + ) .fetch_one(&mut tx) .await @@ -1161,7 +1171,7 @@ pub async fn pull( "Error decreasing concurrency count for script path {job_script_path}: {e}" )) })?; - tracing::debug!("running_job after decrease: {}", x); + tracing::debug!("running_job after decrease: {}", x.unwrap_or(0)); let job_uuid: Uuid = pulled_job.id; let min_started_at: Option> = script_path_live_stats.min_started_at;