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

fix: Track job UUIDs in concurrency table instead of a simple counter (#2498)

Browse Source 
* fix: Track job UUIDs in concurrency table instead of a simple counter

* sqlx prepare
This commit is contained in:
Guillaume Bouvignies
2023-10-24 09:28:46 +02:00
committed by GitHub
parent affaf57250
commit eacef8cd5b
9 changed files with 88 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0be8c6f698a715d015126a7871a39fd7c25ad5269d3313ee1a32533b6d346446.json
Generated
+15
View File
@@ -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"
}
backend/.sqlx/query-38a3fbc28e827d08a928d441274c5eb28780abc8adffcc7175f6c8d4ff8849ca.json
Generated
+23
View File
@@ -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"
}
backend/.sqlx/query-3e3d12a51cb524fbd3d6949e150cb608acfbe8c8eade1939e813086380c205e0.json
Generated
+24
View File
@@ -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"
}
backend/.sqlx/query-467247beca4dae321ab3acc3a99f655412761157cda54bbd3e58cada3de3396c.json
Generated
-14
View File
@@ -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"
}
backend/.sqlx/query-71879f3644dde27f2d93e5511e420579dfd2ffab72bb8480a1622b68703d4ea7.json
Generated
-22
View File
@@ -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"
}
backend/.sqlx/query-86652246b20f9bd023d3419ec382176780a87f32673b53de6f899564c1494147.json
Generated
-22
View File
@@ -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"
}
backend/migrations/20231023135159_concurrency_job_uuids.down.sql
+3
View File
@@ -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;
backend/migrations/20231023135159_concurrency_job_uuids.up.sql
+3
View File
@@ -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;
backend/windmill-queue/src/jobs.rs
+20 -10
View File
@@ -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<R: rsmq_async::RsmqConnection + Send + Clone>(
job_custom_concurrency_time_window_s
);
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 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<R: rsmq_async::RsmqConnection + Send + Clone>(
"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<R: rsmq_async::RsmqConnection + Send + Clone>(
})?;
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<R: rsmq_async::RsmqConnection + Send + Clone>(
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<R: rsmq_async::RsmqConnection + Send + Clone>(
"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<DateTime<Utc>> = script_path_live_stats.min_started_at;