diff --git a/backend/.sqlx/query-7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24.json b/backend/.sqlx/query-8c15ea27be82665fe74a4be3182a5049fa937e30b4115caf0ba8d61d34f2089a.json similarity index 50% rename from backend/.sqlx/query-7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24.json rename to backend/.sqlx/query-8c15ea27be82665fe74a4be3182a5049fa937e30b4115caf0ba8d61d34f2089a.json index f20c2c1562..9b20861537 100644 --- a/backend/.sqlx/query-7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24.json +++ b/backend/.sqlx/query-8c15ea27be82665fe74a4be3182a5049fa937e30b4115caf0ba8d61d34f2089a.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' \n WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", + "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', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", "describe": { "columns": [ { @@ -30,5 +30,5 @@ true ] }, - "hash": "7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24" + "hash": "8c15ea27be82665fe74a4be3182a5049fa937e30b4115caf0ba8d61d34f2089a" } diff --git a/backend/.sqlx/query-ae25fae1aca2cffc43a6054cd54639ba9deaaef419ad4cb2655e7e95f602a688.json b/backend/.sqlx/query-ae25fae1aca2cffc43a6054cd54639ba9deaaef419ad4cb2655e7e95f602a688.json new file mode 100644 index 0000000000..5634e60233 --- /dev/null +++ b/backend/.sqlx/query-ae25fae1aca2cffc43a6054cd54639ba9deaaef419ad4cb2655e7e95f602a688.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO job_logs (job_id, logs) VALUES ($1,'Restarted job after not receiving job''s ping for too long the ' || now() || '\n\n') \n ON CONFLICT (job_id) DO UPDATE SET logs = job_logs.logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE job_logs.job_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "ae25fae1aca2cffc43a6054cd54639ba9deaaef419ad4cb2655e7e95f602a688" +} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 2c204b38a9..105076b3f5 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1248,7 +1248,7 @@ async fn handle_zombie_jobs ) { if *RESTART_ZOMBIE_JOBS { let restarted = sqlx::query!( - "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' + "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', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping", *ZOMBIE_JOB_TIMEOUT, @@ -1261,11 +1261,23 @@ async fn handle_zombie_jobs if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _); } + let base_url = BASE_URL.read().await.clone(); for r in restarted { + let last_ping = if let Some(x) = r.last_ping { + format!("last ping at {x}") + } else { + "no last ping".to_string() + }; + let url = format!("{}/run/{}?workspace={}", base_url, r.id, r.workspace_id,); let error_message = format!( - "Zombie job detected, restarting it: {} {} {:?}", - r.id, r.workspace_id, r.last_ping + "Zombie job {} on {} ({}) detected, restarting it, {}", + r.id, r.workspace_id, url, last_ping ); + + let _ = sqlx::query!(" + INSERT INTO job_logs (job_id, logs) VALUES ($1,'Restarted job after not receiving job''s ping for too long the ' || now() || '\n\n') + ON CONFLICT (job_id) DO UPDATE SET logs = job_logs.logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE job_logs.job_id = $1", r.id) + .execute(db).await; tracing::error!(error_message); report_critical_error(error_message, db.clone()).await; } diff --git a/frontend/src/routes/(root)/(logged)/workers/+page.svelte b/frontend/src/routes/(root)/(logged)/workers/+page.svelte index 91b4aea911..f6becd6ce9 100644 --- a/frontend/src/routes/(root)/(logged)/workers/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/workers/+page.svelte @@ -18,7 +18,7 @@ import { enterpriseLicense, superadmin } from '$lib/stores' import { sendUserToast } from '$lib/toast' import { displayDate, groupBy, pluralize, truncate } from '$lib/utils' - import { AlertTriangle, FileJson, LineChart, Plus, Search } from 'lucide-svelte' + import { AlertTriangle, FileJson, LineChart, List, Plus, Search } from 'lucide-svelte' import { onDestroy, onMount } from 'svelte' import AutoComplete from 'simple-svelte-autocomplete' @@ -237,6 +237,9 @@ return [worker_group[0], filteredWorkerGroup] } + const openSearchWithPrefilledText: (t?: string) => void = getContext( + 'openSearchWithPrefilledText' + ) {#if $superadmin} @@ -292,6 +295,20 @@ Queue metrics +
+ +
{/if}