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

fix: cleanup concurrency_counter automatically + remove orphans keys automatically

Browse Source
This commit is contained in:
Ruben Fiszel
2025-08-01 11:26:23 +00:00
parent 493707668b
commit 87dd52296f
9 changed files with 442 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac.json
Generated
+38
View File
@@ -0,0 +1,38 @@
{
"db_name": "PostgreSQL",
"query": "\nWITH lockable_counters AS (\n SELECT concurrency_id, job_uuids\n FROM concurrency_counter\n WHERE job_uuids != '{}'::jsonb\n FOR UPDATE SKIP LOCKED\n),\nall_job_uuids AS (\n SELECT DISTINCT jsonb_object_keys(job_uuids) AS job_uuid\n FROM lockable_counters\n),\norphaned_job_uuids AS (\n SELECT job_uuid\n FROM all_job_uuids\n WHERE job_uuid NOT IN (\n SELECT id::text \n FROM v2_job_queue \n FOR SHARE SKIP LOCKED\n )\n),\norphaned_array AS (\n SELECT ARRAY(SELECT job_uuid FROM orphaned_job_uuids) AS orphaned_keys\n),\nbefore_update AS (\n SELECT lc.concurrency_id, lc.job_uuids, oa.orphaned_keys\n FROM lockable_counters lc, orphaned_array oa\n WHERE lc.job_uuids ?| oa.orphaned_keys\n),\naffected_rows AS (\n UPDATE concurrency_counter \n SET job_uuids = job_uuids - orphaned_array.orphaned_keys\n FROM orphaned_array\n WHERE concurrency_counter.concurrency_id IN (\n SELECT concurrency_id FROM before_update\n )\n RETURNING concurrency_id, job_uuids AS updated_job_uuids\n),\nexpanded_orphaned AS (\n SELECT bu.concurrency_id, \n bu.job_uuids AS original_job_uuids,\n unnest(bu.orphaned_keys) AS orphaned_key\n FROM before_update bu\n)\nSELECT \n eo.concurrency_id,\n eo.orphaned_key,\n eo.original_job_uuids,\n ar.updated_job_uuids\nFROM expanded_orphaned eo\nJOIN affected_rows ar ON eo.concurrency_id = ar.concurrency_id\nWHERE eo.original_job_uuids ? eo.orphaned_key\nORDER BY eo.concurrency_id, eo.orphaned_key\n",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "concurrency_id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "orphaned_key",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "original_job_uuids",
"type_info": "Jsonb"
},
{
"ordinal": 3,
"name": "updated_job_uuids",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
null,
false,
false
]
},
"hash": "0a1c10bd2232b0770a7816e1bd8d758dc393f797890d597e5996146247f512ac"
}
backend/.sqlx/query-2bf5f7f2cf9d85a5d23e5db2f7616fb41fece9b3d46fde2d546d70b46f9008e3.json
Generated
-15
View File
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Int4"
]
},
"nullable": []
},
"hash": "2bf5f7f2cf9d85a5d23e5db2f7616fb41fece9b3d46fde2d546d70b46f9008e3"
}
backend/.sqlx/query-7b48820a08fbb3bee0fb2dc802a0b4f28ed5507797165e8c577ce6a95d11694c.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Int4"
]
},
"nullable": []
},
"hash": "7b48820a08fbb3bee0fb2dc802a0b4f28ed5507797165e8c577ce6a95d11694c"
}
backend/.sqlx/query-9f4811fe735d401b62f4b7bf3db2b5cd13eb3364a8a8546007dec7ab528b1f9d.json
Generated
+20
View File
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "\nWITH rows_to_delete AS (\n SELECT concurrency_id\n FROM concurrency_counter\n WHERE job_uuids = '{}'::jsonb\n FOR UPDATE SKIP LOCKED\n)\nDELETE FROM concurrency_counter\nWHERE concurrency_id IN (SELECT concurrency_id FROM rows_to_delete) RETURNING concurrency_id",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "concurrency_id",
"type_info": "Varchar"
}
],
"parameters": {
"Left": []
},
"nullable": [
false
]
},
"hash": "9f4811fe735d401b62f4b7bf3db2b5cd13eb3364a8a8546007dec7ab528b1f9d"
}
backend/.sqlx/query-ca3ba808e020c8c7a35eaef842b20cfeee64fd47ded72fce55cc75e0bbb291a8.json
Generated
-15
View File
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Int4"
]
},
"nullable": []
},
"hash": "ca3ba808e020c8c7a35eaef842b20cfeee64fd47ded72fce55cc75e0bbb291a8"
}
backend/.sqlx/query-f9e1334ecb17b313924587d96fc8b0310b14d75e85a558c8cb70f2761923d175.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO usage (id, is_workspace, month_, usage) \n VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2) \n ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Int4"
]
},
"nullable": []
},
"hash": "f9e1334ecb17b313924587d96fc8b0310b14d75e85a558c8cb70f2761923d175"
}
backend/src/main.rs
+22 -6
View File
@@ -95,7 +95,7 @@ use crate::monitor::{
reload_job_default_timeout_setting, reload_jwt_secret_setting, reload_license_key,
reload_npm_config_registry_setting, reload_pip_index_url_setting,
reload_retention_period_setting, reload_scim_token_setting, reload_smtp_config,
reload_worker_config,
reload_worker_config, MonitorIteration,
};
#[cfg(feature = "parquet")]
@@ -144,10 +144,12 @@ fn update_ca_certificates_if_requested() {
let ca_cert_path = std::env::var("RUN_UPDATE_CA_CERTIFICATE_PATH")
.unwrap_or_else(|_| "/usr/sbin/update-ca-certificates".to_string());
println!("RUN_UPDATE_CA_CERTIFICATE_AT_START=true, running: {}", ca_cert_path);
println!(
"RUN_UPDATE_CA_CERTIFICATE_AT_START=true, running: {}",
ca_cert_path
);
let output = std::process::Command::new(&ca_cert_path)
.output();
let output = std::process::Command::new(&ca_cert_path).output();
match output {
Ok(result) => {
@@ -155,11 +157,17 @@ fn update_ca_certificates_if_requested() {
println!("Successfully updated CA certificates");
} else {
let stderr = String::from_utf8_lossy(&result.stderr);
println!("Failed to update CA certificates, but continuing startup: {}", stderr.trim());
println!(
"Failed to update CA certificates, but continuing startup: {}",
stderr.trim()
);
}
}
Err(e) => {
println!("Could not run update-ca-certificates command, but continuing startup: {}", e);
println!(
"Could not run update-ca-certificates command, but continuing startup: {}",
e
);
}
}
}
@@ -565,6 +573,7 @@ Windmill Community Edition {GIT_VERSION}
worker_mode,
true,
killpill_tx.clone(),
None,
)
.await;
@@ -792,6 +801,8 @@ Windmill Community Edition {GIT_VERSION}
let h = tokio::spawn(async move {
let mut listener = retry_listen_pg(&db_url).await;
let mut last_listener_refresh = Instant::now();
let mut monitor_iteration: u64 = 0;
let rd_shift: u8 = rand::rng().random_range(0..200);
loop {
let db = db.clone();
tokio::select! {
@@ -1127,8 +1138,13 @@ Windmill Community Edition {GIT_VERSION}
worker_mode,
false,
tx.clone(),
Some(MonitorIteration {
rd_shift,
iter: monitor_iteration,
}),
)
.await;
monitor_iteration += 1;
if server_mode {
if !*windmill_common::QUIET_LOGS {
tracing::info!("monitor task finished");
backend/src/monitor.rs
+135
View File
@@ -1337,6 +1337,17 @@ pub async fn monitor_pool(db: &DB) {
}
}
pub struct MonitorIteration {
pub rd_shift: u8,
pub iter: u64,
}
impl MonitorIteration {
pub fn should_run(&self, period: u8) -> bool {
self.iter % (period as u64) == self.rd_shift as u64
}
}
pub async fn monitor_db(
conn: &Connection,
base_internal_url: &str,
@@ -1344,6 +1355,7 @@ pub async fn monitor_db(
_worker_mode: bool,
initial_load: bool,
_killpill_tx: KillpillSender,
iteration: Option<MonitorIteration>,
) {
let zombie_jobs_f = async {
if server_mode && !initial_load && !*DISABLE_ZOMBIE_JOBS_MONITORING {
@@ -1358,6 +1370,29 @@ pub async fn monitor_db(
}
}
};
// run every 5 minutes
let cleanup_concurrency_counters_f = async {
if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(10) {
if let Some(db) = conn.as_sql() {
if let Err(e) = cleanup_concurrency_counters_orphaned_keys(&db).await {
tracing::error!("Error cleaning up concurrency counters: {:?}", e);
}
}
}
};
// run every 10 minutes
let cleanup_concurrency_counters_empty_keys_f = async {
if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(20) {
if let Some(db) = conn.as_sql() {
if let Err(e) = cleanup_concurrency_counters_empty_keys(&db).await {
tracing::error!("Error cleaning up concurrency counters: {:?}", e);
}
}
}
};
let expired_items_f = async {
if server_mode && !initial_load {
if let Some(db) = conn.as_sql() {
@@ -1441,6 +1476,8 @@ pub async fn monitor_db(
low_disk_alerts_f,
apply_autoscaling_f,
update_min_worker_version_f,
cleanup_concurrency_counters_f,
cleanup_concurrency_counters_empty_keys_f,
);
}
@@ -1983,6 +2020,104 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker
}
}
async fn cleanup_concurrency_counters_orphaned_keys(db: &DB) -> error::Result<()> {
let result = sqlx::query!(
"
WITH lockable_counters AS (
SELECT concurrency_id, job_uuids
FROM concurrency_counter
WHERE job_uuids != '{}'::jsonb
FOR UPDATE SKIP LOCKED
),
all_job_uuids AS (
SELECT DISTINCT jsonb_object_keys(job_uuids) AS job_uuid
FROM lockable_counters
),
orphaned_job_uuids AS (
SELECT job_uuid
FROM all_job_uuids
WHERE job_uuid NOT IN (
SELECT id::text
FROM v2_job_queue
FOR SHARE SKIP LOCKED
)
),
orphaned_array AS (
SELECT ARRAY(SELECT job_uuid FROM orphaned_job_uuids) AS orphaned_keys
),
before_update AS (
SELECT lc.concurrency_id, lc.job_uuids, oa.orphaned_keys
FROM lockable_counters lc, orphaned_array oa
WHERE lc.job_uuids ?| oa.orphaned_keys
),
affected_rows AS (
UPDATE concurrency_counter
SET job_uuids = job_uuids - orphaned_array.orphaned_keys
FROM orphaned_array
WHERE concurrency_counter.concurrency_id IN (
SELECT concurrency_id FROM before_update
)
RETURNING concurrency_id, job_uuids AS updated_job_uuids
),
expanded_orphaned AS (
SELECT bu.concurrency_id,
bu.job_uuids AS original_job_uuids,
unnest(bu.orphaned_keys) AS orphaned_key
FROM before_update bu
)
SELECT
eo.concurrency_id,
eo.orphaned_key,
eo.original_job_uuids,
ar.updated_job_uuids
FROM expanded_orphaned eo
JOIN affected_rows ar ON eo.concurrency_id = ar.concurrency_id
WHERE eo.original_job_uuids ? eo.orphaned_key
ORDER BY eo.concurrency_id, eo.orphaned_key
",
)
.fetch_all(db)
.await?;
if result.len() > 0 {
tracing::info!("Cleaned up {} concurrency counters", result.len());
for row in result {
tracing::info!("Concurrency counter cleaned up: concurrency_id: {}, orphaned_key: {:?}, original_job_uuids: {:?}, updated_job_uuids: {:?}", row.concurrency_id, row.orphaned_key, row.original_job_uuids, row.updated_job_uuids);
}
}
Ok(())
}
async fn cleanup_concurrency_counters_empty_keys(db: &DB) -> error::Result<()> {
let result = sqlx::query!(
"
WITH rows_to_delete AS (
SELECT concurrency_id
FROM concurrency_counter
WHERE job_uuids = '{}'::jsonb
FOR UPDATE SKIP LOCKED
)
DELETE FROM concurrency_counter
WHERE concurrency_id IN (SELECT concurrency_id FROM rows_to_delete) RETURNING concurrency_id",
)
.fetch_all(db)
.await?;
if result.len() > 0 {
tracing::info!(
"Cleaned up {} empty concurrency counters: {:?}",
result.len(),
result
.iter()
.map(|x| x.concurrency_id.to_string())
.collect::<Vec<_>>()
.join(",")
);
}
Ok(())
}
async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
let flows = sqlx::query!(
r#"
backend/windmill-queue/src/jobs.rs
+197 -163
View File
@@ -810,7 +810,8 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
)
.await;
restart_job_if_perpetual(db, queued_job, canceled_by).await?;
restart_job_if_perpetual(db, queued_job, &canceled_by).await?;
// tracing::error!("4 {:?}", start.elapsed());
Ok(queued_job.id)
@@ -1070,29 +1071,24 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
}
}
if queued_job.concurrent_limit.is_some() {
let concurrency_key = match concurrency_key(db, &queued_job.id).await {
Ok(c) => c,
Err(e) => {
tracing::error!(
"Could not get concurrency key for job {} defaulting to default key: {e:?}",
queued_job.id
);
"".to_string()
}
};
if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_empty() {
let concurrency_key = concurrency_key(db, &queued_job.id).await?;
if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_none() {
tracing::warn!("Concurrency limit is disabled, skipping");
} else {
if let Err(e) = sqlx::query_scalar!(
"UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1",
concurrency_key,
queued_job.id.hyphenated().to_string(),
)
.execute(&mut *tx)
.await
{
tracing::error!("Could not decrement concurrency counter: {}", e);
}
let concurrency_key = concurrency_key.unwrap();
sqlx::query_scalar!(
"UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1",
concurrency_key,
queued_job.id.hyphenated().to_string(),
)
.execute(&mut *tx)
.await
.map_err(|e| {
Error::internal_err(format!(
"Could not decrement concurrency counter for job_id={}: {e:#}",
queued_job.id
))
})?;
}
if let Err(e) = sqlx::query_scalar!(
@@ -1101,12 +1097,11 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
)
.execute(&mut *tx)
.await
.map_err(|e| {
Error::internal_err(format!(
"Error updating to add ended_at timestamp concurrency_key={concurrency_key}: {e:#}"
))
}) {
tracing::error!("Could not update concurrency_key: {}", e);
{
tracing::error!(
"Could not update concurrency_key ended_at for job_id={}: {e:#}",
queued_job.id,
);
}
tracing::debug!("decremented concurrency counter");
}
@@ -1178,93 +1173,124 @@ async fn check_result_size<T: ValidableJson>(
async fn restart_job_if_perpetual(
db: &Pool<Postgres>,
queued_job: &MiniPulledJob,
canceled_by: Option<CanceledBy>,
canceled_by: &Option<CanceledBy>,
) -> Result<(), Error> {
Ok(
if !queued_job.is_flow_step() && queued_job.kind == JobKind::Script && canceled_by.is_none()
{
if let Some(hash) = queued_job.runnable_id {
let p = sqlx::query_scalar!(
"SELECT restart_unless_cancelled FROM script WHERE hash = $1 AND workspace_id = $2",
hash.0,
&queued_job.workspace_id
)
.fetch_optional(db)
.await?
.flatten()
.unwrap_or(false);
if !queued_job.is_flow_step() && queued_job.kind == JobKind::Script && canceled_by.is_none() {
if let Some(hash) = queued_job.runnable_id {
(|| restart_job_if_perpetual_inner(db, queued_job, hash))
.retry(
ConstantBuilder::default()
.with_delay(std::time::Duration::from_secs(3))
.with_max_times(5)
.build(),
)
.notify(|err, dur| {
tracing::error!(
"Could not apply perpetual job restart, retrying in {dur:#?}, err: {err:#?}"
);
})
.sleep(tokio::time::sleep)
.await?;
}
}
Ok(())
}
if p {
let tx = PushIsolationLevel::IsolatedRoot(db.clone());
// perpetual jobs can run one job per 10s max. If the job was faster than 10s, schedule the next one with the appropriate delay
let now = chrono::Utc::now();
let scheduled_for = if now
.signed_duration_since(queued_job.started_at.unwrap_or(now))
.num_seconds()
< 10
{
let next_run = queued_job.started_at.unwrap_or(now)
+ chrono::Duration::try_seconds(10).unwrap();
tracing::warn!("Perpetual script {:?} is running too fast, only 1 job per 10s it supported. Scheduling next run for {:?}", queued_job.runnable_path, next_run);
Some(next_run)
} else {
None
};
let ehm = HashMap::new();
let (_uuid, tx) = push(
db,
tx,
&queued_job.workspace_id,
JobPayload::ScriptHash {
hash,
path: queued_job.runnable_path().to_string(),
custom_concurrency_key: custom_concurrency_key(db, &queued_job.id)
.await?,
concurrent_limit: queued_job.concurrent_limit,
concurrency_time_window_s: queued_job.concurrency_time_window_s,
cache_ttl: queued_job.cache_ttl,
dedicated_worker: None,
language: queued_job
.script_lang
.clone()
.unwrap_or_else(|| ScriptLang::Deno),
priority: queued_job.priority,
apply_preprocessor: false,
},
queued_job
.args
.as_ref()
.map(|x| PushArgs::from(&x.0))
.unwrap_or_else(|| PushArgs::from(&ehm)),
&queued_job.created_by,
&queued_job.permissioned_as_email,
queued_job.permissioned_as.clone(),
Some(&format!("add.completed.job{}", queued_job.id)),
scheduled_for,
queued_job.schedule_path(),
None,
None,
None,
false,
false,
None,
queued_job.visible_to_owner,
Some(queued_job.tag.clone()),
queued_job.timeout,
None,
queued_job.priority,
None,
)
.await?;
if let Err(e) = tx.commit().await {
tracing::error!("Could not restart job {}: {}", queued_job.id, e);
}
}
}
},
async fn restart_job_if_perpetual_inner(
db: &Pool<Postgres>,
queued_job: &MiniPulledJob,
hash: ScriptHash,
) -> Result<(), Error> {
let restart = sqlx::query_scalar!(
"SELECT restart_unless_cancelled FROM script WHERE hash = $1 AND workspace_id = $2",
hash.0,
&queued_job.workspace_id
)
.fetch_optional(db)
.await?
.flatten()
.unwrap_or(false);
if restart {
let tx = PushIsolationLevel::IsolatedRoot(db.clone());
// perpetual jobs can run one job per 10s max. If the job was faster than 10s, schedule the next one with the appropriate delay
let now = now_from_db(db).await?;
let scheduled_for = if now
.signed_duration_since(queued_job.started_at.unwrap_or(now))
.num_seconds()
< 10
{
let next_run =
queued_job.started_at.unwrap_or(now) + chrono::Duration::try_seconds(10).unwrap();
tracing::warn!("Perpetual script {:?} is running too fast, only 1 job per 10s it supported. Scheduling next run for {:?}", queued_job.runnable_path, next_run);
Some(next_run)
} else {
None
};
let ehm = HashMap::new();
let (_uuid, tx) = push(
db,
tx,
&queued_job.workspace_id,
JobPayload::ScriptHash {
hash,
path: queued_job.runnable_path().to_string(),
custom_concurrency_key: custom_concurrency_key(db, &queued_job.id).await?,
concurrent_limit: queued_job.concurrent_limit,
concurrency_time_window_s: queued_job.concurrency_time_window_s,
cache_ttl: queued_job.cache_ttl,
dedicated_worker: None,
language: queued_job
.script_lang
.clone()
.unwrap_or_else(|| ScriptLang::Deno),
priority: queued_job.priority,
apply_preprocessor: false,
},
queued_job
.args
.as_ref()
.map(|x| PushArgs::from(&x.0))
.unwrap_or_else(|| PushArgs::from(&ehm)),
&queued_job.created_by,
&queued_job.permissioned_as_email,
queued_job.permissioned_as.clone(),
Some(&format!("add.completed.job{}", queued_job.id)),
scheduled_for,
queued_job.schedule_path(),
None,
None,
None,
false,
false,
None,
queued_job.visible_to_owner,
Some(queued_job.tag.clone()),
queued_job.timeout,
None,
queued_job.priority,
None,
)
.await?;
tx.commit().await?;
}
Ok(())
}
async fn has_failure_module(db: &Pool<Postgres>, job: &MiniPulledJob) -> bool {
if let Ok(flow) = cache::job::fetch_flow(db, job.kind, job.runnable_id).await {
return flow.value().failure_module.is_some();
}
sqlx::query_scalar!(
"SELECT raw_flow->'failure_module' != 'null'::jsonb FROM v2_job WHERE id = $1",
job.id
)
.fetch_one(db)
.await
.unwrap_or(Some(false))
.unwrap_or(false)
}
#[cfg(feature = "enterprise")]
@@ -1277,20 +1303,6 @@ async fn apply_completed_job_error_handlers<T: Serialize + Send + Sync + Validab
_skip_downstream_error_handlers: bool,
) {
if !success {
async fn has_failure_module(db: &Pool<Postgres>, job: &MiniPulledJob) -> bool {
if let Ok(flow) = cache::job::fetch_flow(db, job.kind, job.runnable_id).await {
return flow.value().failure_module.is_some();
}
sqlx::query_scalar!(
"SELECT raw_flow->'failure_module' != 'null'::jsonb FROM v2_job WHERE id = $1",
job.id
)
.fetch_one(db)
.await
.unwrap_or(Some(false))
.unwrap_or(false)
}
if queued_job.permissioned_as_email == ERROR_HANDLER_USER_EMAIL {
let base_url = BASE_URL.read().await;
let w_id = &queued_job.workspace_id;
@@ -1398,31 +1410,34 @@ fn apply_completed_job_cloud_usage(
let premium_workspace =
windmill_common::workspaces::is_premium_workspace(&db, &w_id).await;
tokio::time::timeout(std::time::Duration::from_secs(10), async move {
let _ = sqlx::query!(
"INSERT INTO usage (id, is_workspace, month_, usage)
VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2)
ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2",
w_id,
additional_usage as i32
)
.execute(&db)
.await
.map_err(|e| Error::internal_err(format!("updating usage: {e:#}")));
if !premium_workspace {
let _ = sqlx::query!(
"INSERT INTO usage (id, is_workspace, month_, usage)
VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2)
VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2)
ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2",
email,
w_id,
additional_usage as i32
)
.execute(&db)
.await
.map_err(|e| Error::internal_err(format!("updating usage: {e:#}")));
}}).await.unwrap_or_else(|_| {
tracing::error!("Could not update usage for workspace {w_id2} and permissioned as {email2}, stopped after 10s");
});
.map_err(|e| {
Error::internal_err(format!("updating usage: {e:#}"))
});
if !premium_workspace {
let _ = sqlx::query!(
"INSERT INTO usage (id, is_workspace, month_, usage)
VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2)
ON CONFLICT (id, is_workspace, month_) DO UPDATE SET usage = usage.usage + $2",
email,
additional_usage as i32
)
.execute(&db)
.await
.map_err(|e| Error::internal_err(format!("updating usage: {e:#}")));
}
}).await.unwrap_or_else(|_| {
tracing::error!("Could not update usage for workspace {w_id2} and permissioned as {email2}, stopped after 10s");
});
});
}
}
@@ -2565,16 +2580,12 @@ pub async fn pull(
return Ok(PulledJobResult { job: Some(pulled_job), suspended });
}
let job_concurrency_key = match concurrency_key(db, &pulled_job.id).await {
Ok(key) => key,
Err(e) => {
tracing::error!(
"Could not get concurrency key for job {} defaulting to default key: {e:?}",
pulled_job.id
);
"".to_string()
}
};
let job_concurrency_key = concurrency_key(db, &pulled_job.id).await?;
if job_concurrency_key.is_none() {
tracing::warn!("No concurrency key found for job {}", pulled_job.id);
return Ok(PulledJobResult { job: None, suspended });
}
let job_concurrency_key = job_concurrency_key.unwrap();
tracing::debug!("Concurrency key is '{}'", job_concurrency_key);
let job_custom_concurrent_limit = pulled_job.concurrent_limit.unwrap();
// setting concurrency_time_window to 0 will count only the currently running jobs
@@ -2821,17 +2832,40 @@ pub async fn custom_concurrency_key(
db: &Pool<Postgres>,
job_id: &Uuid,
) -> Result<Option<String>, sqlx::Error> {
sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", job_id)
.fetch_optional(db) // this should no longer be fetch optional
.await
let fut = async || {
sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", job_id)
.fetch_optional(db) // this should no longer be fetch optional
.await
};
fut.retry(
ConstantBuilder::default()
.with_delay(std::time::Duration::from_secs(3))
.with_max_times(5)
.build(),
)
.notify(|err, dur| {
tracing::error!(
"Could not get concurrency key for job {job_id}, retrying in {dur:#?}, err: {err:#?}"
);
})
.await
}
async fn concurrency_key(db: &Pool<Postgres>, id: &Uuid) -> windmill_common::error::Result<String> {
not_found_if_none(
custom_concurrency_key(db, id).await?,
"ConcurrencyKey",
id.to_string(),
)
async fn concurrency_key(
db: &Pool<Postgres>,
id: &Uuid,
) -> windmill_common::error::Result<Option<String>> {
custom_concurrency_key(db, id)
.await
.map(|x| {
if x.is_none() {
tracing::info!("No concurrency key found for job {id}, defaulting to empty string");
}
return x;
})
.map_err(|e| {
Error::internal_err(format!("Could not get concurrency key for job {id}: {e:#}"))
})
}
pub fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String {