diff --git a/backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json b/backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json index 2be39fce26..4bcf3c6ce3 100644 --- a/backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json +++ b/backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json @@ -130,28 +130,28 @@ }, { "ordinal": 25, - "name": "teams_command_script", - "type_info": "Text" - }, - { - "ordinal": 26, - "name": "teams_team_id", - "type_info": "Text" - }, - { - "ordinal": 27, - "name": "teams_team_name", - "type_info": "Text" - }, - { - "ordinal": 28, "name": "ai_models", "type_info": "VarcharArray" }, { - "ordinal": 29, + "ordinal": 26, "name": "code_completion_model", "type_info": "Varchar" + }, + { + "ordinal": 27, + "name": "teams_command_script", + "type_info": "Text" + }, + { + "ordinal": 28, + "name": "teams_team_id", + "type_info": "Text" + }, + { + "ordinal": 29, + "name": "teams_team_name", + "type_info": "Text" } ], "parameters": { @@ -185,10 +185,10 @@ true, true, true, - true, - true, - true, false, + true, + true, + true, true ] }, diff --git a/backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json b/backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json index 9243288c9d..14685a8bfa 100644 --- a/backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json +++ b/backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json @@ -130,28 +130,28 @@ }, { "ordinal": 25, - "name": "teams_command_script", - "type_info": "Text" - }, - { - "ordinal": 26, - "name": "teams_team_id", - "type_info": "Text" - }, - { - "ordinal": 27, - "name": "teams_team_name", - "type_info": "Text" - }, - { - "ordinal": 28, "name": "ai_models", "type_info": "VarcharArray" }, { - "ordinal": 29, + "ordinal": 26, "name": "code_completion_model", "type_info": "Varchar" + }, + { + "ordinal": 27, + "name": "teams_command_script", + "type_info": "Text" + }, + { + "ordinal": 28, + "name": "teams_team_id", + "type_info": "Text" + }, + { + "ordinal": 29, + "name": "teams_team_name", + "type_info": "Text" } ], "parameters": { @@ -185,10 +185,10 @@ true, true, true, - true, - true, - true, false, + true, + true, + true, true ] }, diff --git a/backend/.sqlx/query-3e3d12a51cb524fbd3d6949e150cb608acfbe8c8eade1939e813086380c205e0.json b/backend/.sqlx/query-b6d657d98de40a25b2bf66471c376d8a2424dce6c4ab2a43cf863911c833fd30.json similarity index 53% rename from backend/.sqlx/query-3e3d12a51cb524fbd3d6949e150cb608acfbe8c8eade1939e813086380c205e0.json rename to backend/.sqlx/query-b6d657d98de40a25b2bf66471c376d8a2424dce6c4ab2a43cf863911c833fd30.json index d6fd72a8b5..c775d232e1 100644 --- a/backend/.sqlx/query-3e3d12a51cb524fbd3d6949e150cb608acfbe8c8eade1939e813086380c205e0.json +++ b/backend/.sqlx/query-b6d657d98de40a25b2bf66471c376d8a2424dce6c4ab2a43cf863911c833fd30.json @@ -1,6 +1,6 @@ { "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))", + "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": [ { @@ -20,5 +20,5 @@ null ] }, - "hash": "3e3d12a51cb524fbd3d6949e150cb608acfbe8c8eade1939e813086380c205e0" + "hash": "b6d657d98de40a25b2bf66471c376d8a2424dce6c4ab2a43cf863911c833fd30" } diff --git a/backend/.sqlx/query-c013880b8567bf5545b0c93ffb4b69a1e6798103df08344fbc9719e53f336c29.json b/backend/.sqlx/query-c013880b8567bf5545b0c93ffb4b69a1e6798103df08344fbc9719e53f336c29.json new file mode 100644 index 0000000000..904c4eba5f --- /dev/null +++ b/backend/.sqlx/query-c013880b8567bf5545b0c93ffb4b69a1e6798103df08344fbc9719e53f336c29.json @@ -0,0 +1,22 @@ +{ + "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-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json b/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json index 5bfff47576..c2dfed73a2 100644 --- a/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json +++ b/backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - true + null ] }, "hash": "ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927" diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index c676d1e6f7..ac9fa133dc 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -1919,22 +1919,13 @@ pub async fn pull( 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(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, + let (mut tx, running_job) = update_concurrency_counter( + tx, + job_concurrency_key.clone(), jobs_uuids_init_json_value, pulled_job.id.hyphenated().to_string(), ) - .fetch_one(&mut *tx) - .await - .map_err(|e| { - Error::internal_err(format!( - "Error getting concurrency count for script path {job_script_path}: {e:#}" - )) - })?; + .await?; tracing::debug!("running_job: {}", running_job.unwrap_or(0)); let completed_count = sqlx::query!( @@ -2069,6 +2060,39 @@ pub async fn pull( } } +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)> { + // 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,