From b5703c8a37ff079e4857ffc39a13f2f5276741ba Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 31 Jan 2025 00:39:40 +0100 Subject: [PATCH 1/3] restart concurrency limits --- backend/src/monitor.rs | 38 ++++++++++++++++++++++++++++++++++---- 1 file changed, 34 insertions(+), 4 deletions(-) diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 587a25d64a..33488c6356 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1506,9 +1506,20 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> { async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str, worker_name: &str) { if *RESTART_ZOMBIE_JOBS { let restarted = sqlx::query!( - "UPDATE queue SET running = false, started_at = null - WHERE last_ping < now() - ($1 || ' seconds')::interval - AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", + "WITH zombie_jobs AS ( + UPDATE queue SET running = false, started_at = null + WHERE last_ping < now() - ($1 || ' seconds')::interval + AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false + RETURNING id, workspace_id, last_ping + ), + update_concurrency AS ( + UPDATE concurrency_counter cc + SET job_uuids = job_uuids - zj.id::text + FROM zombie_jobs zj + INNER JOIN concurrency_key ck ON ck.job_id = zj.id + WHERE cc.concurrency_id = ck.key + ) + SELECT id, workspace_id, last_ping FROM zombie_jobs", *ZOMBIE_JOB_TIMEOUT, ) .fetch_all(db) @@ -1652,12 +1663,31 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> { tracing::error!(error_message); report_critical_error(error_message, db.clone(), Some(&flow.workspace_id), None).await; // if the flow hasn't started and is a zombie, we can simply restart it + let mut tx = db.begin().await?; + + let concurrency_key = + sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", flow.id) + .fetch_optional(&mut *tx) + .await?; + + if let Some(key) = concurrency_key { + sqlx::query!( + "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1", + key, + flow.id.hyphenated().to_string() + ) + .execute(&mut *tx) + .await?; + } + sqlx::query!( "UPDATE queue SET running = false, started_at = null WHERE id = $1 AND canceled = false", flow.id ) - .execute(db) + .execute(&mut *tx) .await?; + + tx.commit().await?; } else { let id = flow.id.clone(); let last_ping = flow.last_ping.clone(); From 587f16dcedcbeb9001b41aeb7bec287c75f02f1c Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 31 Jan 2025 00:53:12 +0100 Subject: [PATCH 2/3] sqlx --- ...5d67db17904859e5f01f1d578af4e1d062a85.json | 34 ------------------- ...9c6da2c3d1e04d49306adaaaaf06e54ee8357.json | 34 +++++++++++++++++++ 2 files changed, 34 insertions(+), 34 deletions(-) delete mode 100644 backend/.sqlx/query-84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85.json create mode 100644 backend/.sqlx/query-9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357.json diff --git a/backend/.sqlx/query-84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85.json b/backend/.sqlx/query-84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85.json deleted file mode 100644 index cdcf0228ca..0000000000 --- a/backend/.sqlx/query-84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85.json +++ /dev/null @@ -1,34 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE queue SET running = false, started_at = null\n WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "workspace_id", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "last_ping", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false, - false, - true - ] - }, - "hash": "84576f6add15108ade2ed3a88185d67db17904859e5f01f1d578af4e1d062a85" -} diff --git a/backend/.sqlx/query-9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357.json b/backend/.sqlx/query-9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357.json new file mode 100644 index 0000000000..8ae3d80ca6 --- /dev/null +++ b/backend/.sqlx/query-9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH zombie_jobs AS (\n UPDATE queue SET running = false, started_at = null\n WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false \n RETURNING id, workspace_id, last_ping\n ),\n update_concurrency AS (\n UPDATE concurrency_counter cc\n SET job_uuids = job_uuids - zj.id::text\n FROM zombie_jobs zj\n INNER JOIN concurrency_key ck ON ck.job_id = zj.id\n WHERE cc.concurrency_id = ck.key\n )\n SELECT id, workspace_id, last_ping FROM zombie_jobs", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "last_ping", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + true + ] + }, + "hash": "9a9b639611459659ae355a43f219c6da2c3d1e04d49306adaaaaf06e54ee8357" +} From 7956d3a7eddcd67aa1cae8239330fdf49104ada4 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 31 Jan 2025 09:26:29 +0100 Subject: [PATCH 3/3] only mask passwords in more places --- frontend/src/lib/components/FlowStatusWaitingForEvents.svelte | 2 +- frontend/src/lib/components/RunPageSchedules.svelte | 2 +- .../components/apps/components/display/dbtable/InsertRow.svelte | 2 +- .../components/apps/components/helpers/RunnableComponent.svelte | 1 + .../src/lib/components/apps/editor/AppReportsDrawerInner.svelte | 1 + 5 files changed, 5 insertions(+), 3 deletions(-) diff --git a/frontend/src/lib/components/FlowStatusWaitingForEvents.svelte b/frontend/src/lib/components/FlowStatusWaitingForEvents.svelte index 49de5fc21d..2e9bf984b4 100644 --- a/frontend/src/lib/components/FlowStatusWaitingForEvents.svelte +++ b/frontend/src/lib/components/FlowStatusWaitingForEvents.svelte @@ -145,7 +145,7 @@ {#if job?.raw_flow?.modules?.[approvalStep]?.suspend?.resume_form?.schema}
- +
{/if} - + {#if emptyString($primarySchedule.cron)}

Define a schedule frequency first

{/if} diff --git a/frontend/src/lib/components/apps/components/display/dbtable/InsertRow.svelte b/frontend/src/lib/components/apps/components/display/dbtable/InsertRow.svelte index 7ccbfb9373..70102962df 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/InsertRow.svelte +++ b/frontend/src/lib/components/apps/components/display/dbtable/InsertRow.svelte @@ -156,5 +156,5 @@ {#if schema} - + {/if} diff --git a/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte b/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte index a5d4e42ee5..5303f90267 100644 --- a/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte +++ b/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte @@ -769,6 +769,7 @@ {#if render && (autoRefresh || forceSchemaDisplay) && schemaStripped && Object.keys(schemaStripped?.properties ?? {}).length > 0}