From 6ba2ba2717746b20783682625e2fac4376b2d316 Mon Sep 17 00:00:00 2001 From: YellowSnnowmann <167776381+YellowSnnowmann@users.noreply.github.com> Date: Tue, 2 Jun 2026 09:36:42 +0530 Subject: [PATCH] fix(cron): make cron_run non-blocking, enqueue async execution (#3127) (#3142) Co-authored-by: Claude Sonnet 4.6 Co-authored-by: Steven Enamakel --- .../intelligence/memoryGraphLayout.test.ts | 2 +- app/src/lib/i18n/ar.ts | 1 + app/src/lib/i18n/bn.ts | 1 + app/src/lib/i18n/de.ts | 1 + app/src/lib/i18n/en.ts | 1 + app/src/lib/i18n/es.ts | 1 + app/src/lib/i18n/fr.ts | 1 + app/src/lib/i18n/hi.ts | 1 + app/src/lib/i18n/id.ts | 1 + app/src/lib/i18n/it.ts | 1 + app/src/lib/i18n/ko.ts | 1 + app/src/lib/i18n/pl.ts | 1 + app/src/lib/i18n/pt.ts | 1 + app/src/lib/i18n/ru.ts | 1 + app/src/lib/i18n/zh-CN.ts | 1 + app/src/pages/Routines.tsx | 57 +++++++- app/src/utils/tauriCommands/cron.ts | 12 +- src/openhuman/cron/mod.rs | 6 +- src/openhuman/cron/ops.rs | 127 ++++++++++++++---- src/openhuman/cron/ops_tests.rs | 56 ++++++++ src/openhuman/cron/store.rs | 12 ++ 21 files changed, 245 insertions(+), 41 deletions(-) diff --git a/app/src/components/intelligence/memoryGraphLayout.test.ts b/app/src/components/intelligence/memoryGraphLayout.test.ts index 7ccef7f44..628fdaacf 100644 --- a/app/src/components/intelligence/memoryGraphLayout.test.ts +++ b/app/src/components/intelligence/memoryGraphLayout.test.ts @@ -43,7 +43,7 @@ describe('memoryGraphLayout', () => { it('nodeRadius grows with summary level and is fixed for chunk/contact', () => { expect(nodeRadius(summary({ level: 0 }))).toBe(5); - expect(nodeRadius(summary({ level: 3 }))).toBeCloseTo(12.5); + expect(nodeRadius(summary({ level: 3 }))).toBe(12.5); expect(nodeRadius(summary({ level: 99 }))).toBe(252.5); expect(nodeRadius(contact())).toBe(9); expect(nodeRadius(chunk())).toBe(3); diff --git a/app/src/lib/i18n/ar.ts b/app/src/lib/i18n/ar.ts index 91985d2b2..5d36b2247 100644 --- a/app/src/lib/i18n/ar.ts +++ b/app/src/lib/i18n/ar.ts @@ -214,6 +214,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'لم يهرب بعد', 'routines.runNow': 'اركض الآن', 'routines.running': 'تشغيل...', + 'routines.runNowTimedOut': 'انتهت مهلة التشغيل — يرجى التحديث والمحاولة مرة أخرى.', 'routines.viewHistory': 'تاريخ الرؤية', 'routines.loadingHistory': 'التعبئة...', 'routines.noHistory': 'لا يوجد تاريخ بعد', diff --git a/app/src/lib/i18n/bn.ts b/app/src/lib/i18n/bn.ts index 8e6bb1d3e..dcfd944dd 100644 --- a/app/src/lib/i18n/bn.ts +++ b/app/src/lib/i18n/bn.ts @@ -216,6 +216,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'এখনো চালাও নি', 'routines.runNow': 'এখন সঞ্চালন করুন', 'routines.running': 'চলমান...', + 'routines.runNowTimedOut': 'রান টাইমআউট হয়েছে — অনুগ্রহ করে রিফ্রেশ করুন এবং আবার চেষ্টা করুন।', 'routines.viewHistory': 'পূর্ববর্তী তথ্য প্রদর্শন', 'routines.loadingHistory': 'লোড করা হচ্ছে...', 'routines.noHistory': 'এখনো কোন ইতিহাস নেই।', diff --git a/app/src/lib/i18n/de.ts b/app/src/lib/i18n/de.ts index 44d4918ec..583c0c89c 100644 --- a/app/src/lib/i18n/de.ts +++ b/app/src/lib/i18n/de.ts @@ -225,6 +225,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'Noch nicht ausgeführt', 'routines.runNow': 'Ausführung starten', 'routines.running': 'Läuft…', + 'routines.runNowTimedOut': 'Ausführung abgebrochen – bitte Seite neu laden und erneut versuchen.', 'routines.viewHistory': 'Verlauf ansehen', 'routines.loadingHistory': 'Laden…', 'routines.noHistory': 'Noch kein Verlauf', diff --git a/app/src/lib/i18n/en.ts b/app/src/lib/i18n/en.ts index 016360ff0..109f5d81a 100644 --- a/app/src/lib/i18n/en.ts +++ b/app/src/lib/i18n/en.ts @@ -237,6 +237,7 @@ const en: TranslationMap = { 'routines.notRunYet': 'Not run yet', 'routines.runNow': 'Run Now', 'routines.running': 'Running…', + 'routines.runNowTimedOut': 'Run timed out — please refresh and try again.', 'routines.viewHistory': 'View history', 'routines.loadingHistory': 'Loading…', 'routines.noHistory': 'No run history yet.', diff --git a/app/src/lib/i18n/es.ts b/app/src/lib/i18n/es.ts index 7cb1f0e91..1365d0980 100644 --- a/app/src/lib/i18n/es.ts +++ b/app/src/lib/i18n/es.ts @@ -226,6 +226,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'Aún no se ha ejecutado', 'routines.runNow': 'Corre ahora', 'routines.running': 'Corriendo…', + 'routines.runNowTimedOut': 'Tiempo de espera agotado — actualiza e intenta de nuevo.', 'routines.viewHistory': 'Ver historial', 'routines.loadingHistory': 'Cargando…', 'routines.noHistory': 'Aún no hay historial de ejecuciones.', diff --git a/app/src/lib/i18n/fr.ts b/app/src/lib/i18n/fr.ts index 93abd8fd2..9e09062ec 100644 --- a/app/src/lib/i18n/fr.ts +++ b/app/src/lib/i18n/fr.ts @@ -224,6 +224,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'Pas encore exécuté', 'routines.runNow': 'Cours maintenant', 'routines.running': 'Course…', + 'routines.runNowTimedOut': 'Délai dépassé — veuillez rafraîchir et réessayer.', 'routines.viewHistory': "Afficher l'historique", 'routines.loadingHistory': 'Chargement…', 'routines.noHistory': "Pas encore d'historique d'exécution.", diff --git a/app/src/lib/i18n/hi.ts b/app/src/lib/i18n/hi.ts index 4d32f43c4..ed3e83566 100644 --- a/app/src/lib/i18n/hi.ts +++ b/app/src/lib/i18n/hi.ts @@ -216,6 +216,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'अभी तक नहीं चला', 'routines.runNow': 'अब चलायें', 'routines.running': 'दौड़ना...', + 'routines.runNowTimedOut': 'चलाने में समय समाप्त — कृपया रिफ्रेश करें और पुनः प्रयास करें।', 'routines.viewHistory': 'इतिहास देखें', 'routines.loadingHistory': 'लोड...', 'routines.noHistory': 'अभी तक कोई रन इतिहास नहीं है।', diff --git a/app/src/lib/i18n/id.ts b/app/src/lib/i18n/id.ts index 96ae2414b..3e5853b47 100644 --- a/app/src/lib/i18n/id.ts +++ b/app/src/lib/i18n/id.ts @@ -218,6 +218,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'Belum pernah dijalankan', 'routines.runNow': 'Jalankan Sekarang', 'routines.running': 'Berjalan...', + 'routines.runNowTimedOut': 'Waktu habis — silakan muat ulang dan coba lagi.', 'routines.viewHistory': 'Lihat riwayat', 'routines.loadingHistory': 'Memuat…', 'routines.noHistory': 'Belum ada riwayat eksekusi.', diff --git a/app/src/lib/i18n/it.ts b/app/src/lib/i18n/it.ts index c91b1fa54..bded398db 100644 --- a/app/src/lib/i18n/it.ts +++ b/app/src/lib/i18n/it.ts @@ -222,6 +222,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'Non ancora eseguito', 'routines.runNow': 'Corri adesso', 'routines.running': 'Correndo…', + 'routines.runNowTimedOut': "Timeout durante l'esecuzione — aggiorna e riprova.", 'routines.viewHistory': 'Visualizza cronologia', 'routines.loadingHistory': 'Caricamento…', 'routines.noHistory': 'Ancora nessuna cronologia delle esecuzioni.', diff --git a/app/src/lib/i18n/ko.ts b/app/src/lib/i18n/ko.ts index 9413695c3..2aa9bc609 100644 --- a/app/src/lib/i18n/ko.ts +++ b/app/src/lib/i18n/ko.ts @@ -216,6 +216,7 @@ const messages: TranslationMap = { 'routines.notRunYet': '아직 실행되지 않음', 'routines.runNow': '지금 실행', 'routines.running': '실행 중…', + 'routines.runNowTimedOut': '실행 시간 초과 — 새로고침 후 다시 시도하세요.', 'routines.viewHistory': '기록 보기', 'routines.loadingHistory': '로딩 중…', 'routines.noHistory': '아직 실행 기록이 없습니다.', diff --git a/app/src/lib/i18n/pl.ts b/app/src/lib/i18n/pl.ts index 2ca862c31..0c5730800 100644 --- a/app/src/lib/i18n/pl.ts +++ b/app/src/lib/i18n/pl.ts @@ -220,6 +220,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'Jeszcze nie uruchomiono', 'routines.runNow': 'Uruchom teraz', 'routines.running': 'Uruchamianie…', + 'routines.runNowTimedOut': 'Przekroczono limit czasu — odśwież stronę i spróbuj ponownie.', 'routines.viewHistory': 'Wyświetl historię', 'routines.loadingHistory': 'Ładowanie…', 'routines.noHistory': 'Brak historii uruchomień.', diff --git a/app/src/lib/i18n/pt.ts b/app/src/lib/i18n/pt.ts index afbc75798..3bd37b218 100644 --- a/app/src/lib/i18n/pt.ts +++ b/app/src/lib/i18n/pt.ts @@ -224,6 +224,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'Ainda não executado', 'routines.runNow': 'Corra Agora', 'routines.running': 'Executando…', + 'routines.runNowTimedOut': 'Execução expirou — atualize e tente novamente.', 'routines.viewHistory': 'Ver histórico', 'routines.loadingHistory': 'Carregando…', 'routines.noHistory': 'Ainda não há histórico de execuções.', diff --git a/app/src/lib/i18n/ru.ts b/app/src/lib/i18n/ru.ts index 6a1753780..f0761749c 100644 --- a/app/src/lib/i18n/ru.ts +++ b/app/src/lib/i18n/ru.ts @@ -218,6 +218,7 @@ const messages: TranslationMap = { 'routines.notRunYet': 'Еще не запущен', 'routines.runNow': 'Беги сейчас', 'routines.running': 'Бег…', + 'routines.runNowTimedOut': 'Время выполнения истекло — обновите страницу и попробуйте снова.', 'routines.viewHistory': 'Посмотреть историю', 'routines.loadingHistory': 'Загрузка…', 'routines.noHistory': 'Истории запусков пока нет.', diff --git a/app/src/lib/i18n/zh-CN.ts b/app/src/lib/i18n/zh-CN.ts index fd6147eae..83e6ce0c5 100644 --- a/app/src/lib/i18n/zh-CN.ts +++ b/app/src/lib/i18n/zh-CN.ts @@ -206,6 +206,7 @@ const messages: TranslationMap = { 'routines.notRunYet': '尚未运行', 'routines.runNow': '立即运行', 'routines.running': '运行中…', + 'routines.runNowTimedOut': '运行超时，请刷新后重试。', 'routines.viewHistory': '查看历史', 'routines.loadingHistory': '加载中…', 'routines.noHistory': '暂无运行历史。', diff --git a/app/src/pages/Routines.tsx b/app/src/pages/Routines.tsx index f4779625c..01e1ace7e 100644 --- a/app/src/pages/Routines.tsx +++ b/app/src/pages/Routines.tsx @@ -1,5 +1,5 @@ import createDebug from 'debug'; -import { useCallback, useEffect, useState } from 'react'; +import { useCallback, useEffect, useRef, useState } from 'react'; import { useNavigate } from 'react-router-dom'; import RoutineCard from '../components/routines/RoutineCard'; @@ -25,6 +25,14 @@ const Routines = () => { const [runsByJob, setRunsByJob] = useState>({}); const [busyKeys, setBusyKeys] = useState>(new Set()); + const isMountedRef = useRef(true); + + useEffect(() => { + return () => { + isMountedRef.current = false; + }; + }, []); + const addBusy = (key: string) => setBusyKeys(prev => new Set(prev).add(key)); const removeBusy = (key: string) => setBusyKeys(prev => { @@ -80,21 +88,58 @@ const Routines = () => { addBusy(key); setError(null); try { - await openhumanCronRun(jobId); - const runs = await openhumanCronRuns(jobId, 10); - setRunsByJob(prev => ({ ...prev, [jobId]: runs.result })); - // Refresh job list to update last_status + const runResponse = await openhumanCronRun(jobId); + + if (runResponse.result.status === 'queued') { + // Job was enqueued asynchronously — poll until a new run record appears. + // Compare by the latest run's id (not list length) so this works correctly + // when the job already has >= 10 runs and the list stays at the fetch limit. + const previousLatestId = (runsByJob[jobId] ?? [])[0]?.id; + const POLL_INTERVAL_MS = 2000; + const MAX_WAIT_MS = 120_000; + let elapsed = 0; + + while (elapsed < MAX_WAIT_MS) { + await new Promise(resolve => setTimeout(resolve, POLL_INTERVAL_MS)); + if (!isMountedRef.current) return; + elapsed += POLL_INTERVAL_MS; + const runs = await openhumanCronRuns(jobId, 10); + if (!isMountedRef.current) return; + setRunsByJob(prev => ({ ...prev, [jobId]: runs.result })); + const latest = runs.result[0]; + if ( + latest?.id !== undefined && + latest.id !== previousLatestId && + latest.status !== 'queued' + ) { + break; + } + } + if (elapsed >= MAX_WAIT_MS && isMountedRef.current) { + setError(t('routines.runNowTimedOut')); + } + } else { + // Synchronous response (legacy path — kept for backward compatibility). + const runs = await openhumanCronRuns(jobId, 10); + setRunsByJob(prev => ({ ...prev, [jobId]: runs.result })); + } + + if (!isMountedRef.current) return; + + // Refresh job list to update last_status regardless of path. const response = await openhumanCronList(); + if (!isMountedRef.current) return; setJobs( [...response.result].sort( (a, b) => new Date(a.next_run).getTime() - new Date(b.next_run).getTime() ) ); } catch (err) { + if (!isMountedRef.current) return; const message = err instanceof Error ? err.message : String(err); setError(message); } finally { - removeBusy(key); + if (isMountedRef.current) removeBusy(key); } }; diff --git a/app/src/utils/tauriCommands/cron.ts b/app/src/utils/tauriCommands/cron.ts index 2c1bb7a1a..6379c1093 100644 --- a/app/src/utils/tauriCommands/cron.ts +++ b/app/src/utils/tauriCommands/cron.ts @@ -111,9 +111,9 @@ export async function openhumanCronRun( ): Promise< CommandResponse<{ job_id: string; - status: 'ok' | 'error' | string; - duration_ms: number; - output: string; + status: 'queued' | 'ok' | 'error' | string; + duration_ms?: number; + output?: string; }> > { if (!isTauri()) { @@ -122,9 +122,9 @@ export async function openhumanCronRun( return await callCoreRpc< CommandResponse<{ job_id: string; - status: 'ok' | 'error' | string; - duration_ms: number; - output: string; + status: 'queued' | 'ok' | 'error' | string; + duration_ms?: number; + output?: string; }> >({ method: 'openhuman.cron_run', params: { job_id: jobId } }); } diff --git a/src/openhuman/cron/mod.rs b/src/openhuman/cron/mod.rs index 3bfedfeaa..c3728ab63 100644 --- a/src/openhuman/cron/mod.rs +++ b/src/openhuman/cron/mod.rs @@ -21,9 +21,9 @@ pub use schemas::{ }; #[allow(unused_imports)] pub use store::{ - add_agent_job, add_agent_job_with_definition, add_job, add_shell_job, clear_all_jobs, due_jobs, - get_job, list_jobs, list_runs, record_last_run, record_run, remove_job, reschedule_after_run, - update_job, + add_agent_job, add_agent_job_with_definition, add_job, add_shell_job, clear_all_jobs, + delete_queued_runs, due_jobs, get_job, list_jobs, list_runs, record_last_run, record_run, + remove_job, reschedule_after_run, update_job, }; pub use types::{ ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, JobType, Schedule, SessionTarget, diff --git a/src/openhuman/cron/ops.rs b/src/openhuman/cron/ops.rs index ff036b563..06f4994eb 100644 --- a/src/openhuman/cron/ops.rs +++ b/src/openhuman/cron/ops.rs @@ -5,7 +5,30 @@ use crate::openhuman::cron::{ use crate::openhuman::security::SecurityPolicy; use crate::rpc::RpcOutcome; use anyhow::Result; +use once_cell::sync::Lazy; use serde_json::json; +use std::collections::HashSet; +use std::sync::Mutex; + +/// Guard against duplicate concurrent "Run Now" executions per job. +/// Entries are inserted before spawning and removed when the task completes. +static ACTIVE_RUNS: Lazy>> = Lazy::new(|| Mutex::new(HashSet::new())); + +/// RAII guard that removes a job ID from `ACTIVE_RUNS` when dropped. +/// +/// Ensures cleanup runs on normal completion, panic, or future cancellation — +/// so a hung or aborted background task can never permanently lock a job_id. +struct ActiveRunGuard { + job_id: String, +} + +impl Drop for ActiveRunGuard { + fn drop(&mut self) { + if let Ok(mut active) = ACTIVE_RUNS.lock() { + active.remove(&self.job_id); + } + } +} pub fn add_once(config: &Config, delay: &str, command: &str) -> Result { let duration = parse_human_delay(delay)?; @@ -186,43 +209,97 @@ pub async fn cron_run( config: &Config, job_id: &str, ) -> Result, String> { - if job_id.trim().is_empty() { + let job_id = job_id.trim(); + if job_id.is_empty() { return Err("Missing 'job_id' parameter".to_string()); } if !config.cron.enabled { return Err("cron is disabled by config (cron.enabled=false)".to_string()); } - let job = cron::get_job(config, job_id.trim()).map_err(|e| e.to_string())?; - let started_at = chrono::Utc::now(); - let (success, output) = cron::scheduler::execute_job_now(config, &job).await; - let finished_at = chrono::Utc::now(); - let duration_ms = (finished_at - started_at).num_milliseconds(); - let status = if success { "ok" } else { "error" }; + // Guard against duplicate concurrent executions for the same job. + { + let mut active = ACTIVE_RUNS + .lock() + .map_err(|_| "ACTIVE_RUNS lock poisoned".to_string())?; + if active.contains(job_id) { + tracing::debug!(job_id, "[cron_run] rejected duplicate concurrent execution"); + return Err(format!("cron job '{job_id}' is already running")); + } + active.insert(job_id.to_string()); + } - let _ = cron::record_run( - config, - &job.id, - started_at, - finished_at, - status, - Some(&output), - duration_ms, - ); - let _ = cron::record_last_run(config, &job.id, finished_at, success, &output); + let job = match cron::get_job(config, job_id) { + Ok(j) => j, + Err(e) => { + if let Ok(mut active) = ACTIVE_RUNS.lock() { + active.remove(job_id); + } + return Err(e.to_string()); + } + }; - // Deliver via the same path as the scheduler loop so proactive - // messages and alerts are sent on "Run Now" too. - cron::scheduler::deliver_job(config, &job, &output).await; + // Insert a "queued" placeholder run record immediately so the frontend + // poller can observe the run as soon as the RPC returns — otherwise the + // run list stays unchanged until execute_job_now finishes. + let queued_at = chrono::Utc::now(); + let _ = cron::record_run(config, &job.id, queued_at, queued_at, "queued", None, 0); + + let config_owned = config.clone(); + let job_id_owned = job_id.to_string(); + + tracing::debug!(job_id, "[cron_run] enqueuing background execution"); + + // Spawn the execution as a background task so the RPC handler returns + // immediately instead of blocking for the full job duration. + tokio::spawn(async move { + // Drop guard ensures job_id is removed from ACTIVE_RUNS even on panic + // or future cancellation, not just on the normal completion path. + let _guard = ActiveRunGuard { + job_id: job_id_owned.clone(), + }; + + tracing::debug!(job_id = %job_id_owned, "[cron_run] background task started"); + + let started_at = chrono::Utc::now(); + let (success, output) = cron::scheduler::execute_job_now(&config_owned, &job).await; + let finished_at = chrono::Utc::now(); + let duration_ms = (finished_at - started_at).num_milliseconds(); + let status = if success { "ok" } else { "error" }; + + tracing::debug!( + job_id = %job_id_owned, + status, + duration_ms, + "[cron_run] background task finished" + ); + + // Remove the "queued" placeholder before inserting the real result + // so we don't leave orphaned rows in the run history. + let _ = cron::delete_queued_runs(&config_owned, &job.id); + + let _ = cron::record_run( + &config_owned, + &job.id, + started_at, + finished_at, + status, + Some(&output), + duration_ms, + ); + let _ = cron::record_last_run(&config_owned, &job.id, finished_at, success, &output); + + // Deliver via the same path as the scheduler loop so proactive + // messages and alerts are sent on "Run Now" too. + cron::scheduler::deliver_job(&config_owned, &job, &output).await; + }); Ok(RpcOutcome::new( json!({ - "job_id": job.id, - "status": status, - "duration_ms": duration_ms, - "output": output + "job_id": job_id, + "status": "queued", }), - vec![format!("cron job run: {}", job_id.trim())], + vec![format!("cron job enqueued: {}", job_id)], )) } diff --git a/src/openhuman/cron/ops_tests.rs b/src/openhuman/cron/ops_tests.rs index 3b78ab9c0..63b1e3327 100644 --- a/src/openhuman/cron/ops_tests.rs +++ b/src/openhuman/cron/ops_tests.rs @@ -418,3 +418,59 @@ async fn cron_runs_returns_empty_history_for_new_job() { assert!(out.value.is_empty()); assert!(out.logs.iter().any(|l| l.contains("cron run history"))); } + +// ── cron_run ──────────────────────────────────────────────────── + +#[tokio::test] +async fn cron_run_rejects_empty_job_id() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let err = cron_run(&config, "").await.unwrap_err(); + assert!(err.contains("Missing 'job_id'")); +} + +#[tokio::test] +async fn cron_run_rejects_whitespace_job_id() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let err = cron_run(&config, " ").await.unwrap_err(); + assert!(err.contains("Missing 'job_id'")); +} + +#[tokio::test] +async fn cron_run_errors_when_cron_disabled() { + let tmp = TempDir::new().unwrap(); + let config = disabled_cron_config(&tmp); + let err = cron_run(&config, "some-id").await.unwrap_err(); + assert!(err.contains("cron is disabled")); +} + +#[tokio::test] +async fn cron_run_returns_queued_immediately_for_valid_job() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let job = make_job(&config, "*/5 * * * *", None, "echo hello"); + + let out = cron_run(&config, &job.id).await.unwrap(); + assert_eq!(out.value["job_id"], json!(job.id)); + assert_eq!(out.value["status"], json!("queued")); + assert!(out.logs.iter().any(|l| l.contains("enqueued"))); +} + +#[tokio::test] +async fn cron_run_rejects_duplicate_concurrent_execution() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let job = make_job(&config, "*/5 * * * *", None, "echo hello"); + + // Seed ACTIVE_RUNS directly to simulate an in-flight execution without + // spawning a real long-running subprocess (deterministic, no timing dependency). + ACTIVE_RUNS.lock().unwrap().insert(job.id.clone()); + let err = cron_run(&config, &job.id).await.unwrap_err(); + ACTIVE_RUNS.lock().unwrap().remove(&job.id); + + assert!( + err.contains("already running"), + "expected 'already running', got: {err}" + ); +} diff --git a/src/openhuman/cron/store.rs b/src/openhuman/cron/store.rs index 36a2e9114..d4f0adde9 100644 --- a/src/openhuman/cron/store.rs +++ b/src/openhuman/cron/store.rs @@ -394,6 +394,18 @@ pub fn record_run( }) } +/// Remove all "queued" placeholder records for a given job so that only the +/// real (ok/error) result row remains in the run history. +pub fn delete_queued_runs(config: &Config, job_id: &str) -> Result { + with_connection(config, |conn| { + let deleted = conn.execute( + "DELETE FROM cron_runs WHERE job_id = ?1 AND status = 'queued'", + params![job_id], + )?; + Ok(deleted) + }) +} + fn truncate_cron_output(output: &str) -> String { if output.len() <= MAX_CRON_OUTPUT_BYTES { return output.to_string();