From 47ab6b3c0004ea0d2c129247435f188fdda84d58 Mon Sep 17 00:00:00 2001 From: Cameron Cordes Date: Fri, 30 Oct 2026 19:08:48 -0400 Subject: [PATCH] Expose ComfyUI run time on Enhance jobs as run_ms Parse execution_start vs the end event from /history, store it when the poller moves a job to ready/failed (never overwritten with null), and return it as "run_ms" on every job endpoint. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 4 +- .../down.sql | 1 + .../up.sql | 1 + src/database/enhance_dao.rs | 24 ++++++- src/database/schema.rs | 1 + src/enhance/comfy.rs | 65 +++++++++++++++++++ src/enhance/handlers.rs | 3 + src/enhance/integration_tests.rs | 8 +++ src/enhance/service.rs | 3 + 9 files changed, 106 insertions(+), 4 deletions(-) create mode 100644 migrations/2026-10-30-200000_add_enhance_run_ms/down.sql create mode 100644 migrations/2026-10-30-200000_add_enhance_run_ms/up.sql diff --git a/CLAUDE.md b/CLAUDE.md index b059cd0..13bc31e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -467,7 +467,8 @@ auth and stays LAN-only, so ImageApi relays previews too. `/history/{id}`: in queue -> queued/running with `queue_position`; finished -> `ready` (staged results in `pending`) or `failed` (ComfyUI's message); known to neither -> **`expired`** (ComfyUI restarted and cleared its temp folder). An - unreachable ComfyUI for `unreachable_grace_secs` -> `unavailable`. A preview or + unreachable ComfyUI for `unreachable_grace_secs` -> `unavailable`. `run_ms` is the `execution_start` -> end-event gap from `/history`, stored when the poller + moves a job to `ready`/`failed` and never overwritten with null. A preview or approve that finds the staged result gone also expires the job (HTTP 410). - **Idle `/free`**: every 30 s, if the queue is empty, none of our jobs is active and ComfyUI's newest `/history` entry has not changed for `idle_free_secs` (it includes @@ -502,6 +503,7 @@ POST /enhance/jobs/{id}/discard -> job (hidden from the list; ComfyUI untou job = {"id","status","queue_position"?,"error"?,"path","library_id","template", "prompt"?,"reference_path"?,"created_at","updated_at", + "run_ms": , "pending": [{"n","preview": "/enhance/jobs/{id}/preview/{n}","approved"}], "saved": [{"n","path","library_id"}]} ``` diff --git a/migrations/2026-10-30-200000_add_enhance_run_ms/down.sql b/migrations/2026-10-30-200000_add_enhance_run_ms/down.sql new file mode 100644 index 0000000..99f21d9 --- /dev/null +++ b/migrations/2026-10-30-200000_add_enhance_run_ms/down.sql @@ -0,0 +1 @@ +ALTER TABLE enhance_jobs DROP COLUMN run_ms; diff --git a/migrations/2026-10-30-200000_add_enhance_run_ms/up.sql b/migrations/2026-10-30-200000_add_enhance_run_ms/up.sql new file mode 100644 index 0000000..d2f79e0 --- /dev/null +++ b/migrations/2026-10-30-200000_add_enhance_run_ms/up.sql @@ -0,0 +1 @@ +ALTER TABLE enhance_jobs ADD COLUMN run_ms BIGINT; diff --git a/src/database/enhance_dao.rs b/src/database/enhance_dao.rs index 67633c3..1ddbf57 100644 --- a/src/database/enhance_dao.rs +++ b/src/database/enhance_dao.rs @@ -59,11 +59,14 @@ pub struct EnhanceJobRow { pub created_at: i64, #[diesel(sql_type = BigInt)] pub updated_at: i64, + /// Milliseconds ComfyUI actually ran the job; set once it finishes. + #[diesel(sql_type = Nullable)] + pub run_ms: Option, } const COLUMNS: &str = "id, original_hash, library_id, rel_path, template, prompt, reference_hash, \ reference_path, comfy_prompt_id, status, error, queue_position, pending, saved, \ - discarded, created_at, updated_at"; + discarded, created_at, updated_at, run_ms"; /// What the poller / approve changes on an existing job. #[derive(Debug, Clone, PartialEq, Eq)] @@ -75,6 +78,8 @@ pub struct JobUpdate { pub pending: Option, /// `Some` replaces the `saved` document. pub saved: Option, + /// `Some` sets the run time; `None` keeps any existing value. + pub run_ms: Option, } impl JobUpdate { @@ -85,6 +90,7 @@ impl JobUpdate { queue_position: None, pending: None, saved: None, + run_ms: None, } } } @@ -137,7 +143,7 @@ impl EnhanceJobDao for SqliteEnhanceJobDao { trace_db_call("insert", "enhance_job_insert", |_span| { let mut conn = self.connection.lock().expect("Unable to get EnhanceJobDao"); sql_query(format!( - "INSERT INTO enhance_jobs ({COLUMNS}) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)" + "INSERT INTO enhance_jobs ({COLUMNS}) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)" )) .bind::(&job.id) .bind::(&job.original_hash) @@ -156,6 +162,7 @@ impl EnhanceJobDao for SqliteEnhanceJobDao { .bind::(job.discarded) .bind::(job.created_at) .bind::(job.updated_at) + .bind::, _>(job.run_ms) .execute(&mut *conn) .map(|_| ()) .map_err(|e| anyhow::anyhow!("Insert error: {}", e)) @@ -212,7 +219,8 @@ impl EnhanceJobDao for SqliteEnhanceJobDao { let mut conn = self.connection.lock().expect("Unable to get EnhanceJobDao"); sql_query( "UPDATE enhance_jobs SET status = ?, error = ?, queue_position = ?, \ - pending = COALESCE(?, pending), saved = COALESCE(?, saved), updated_at = ? \ + pending = COALESCE(?, pending), saved = COALESCE(?, saved), run_ms = COALESCE(?, run_ms), \ + updated_at = ? \ WHERE id = ?", ) .bind::(&u.status) @@ -220,6 +228,7 @@ impl EnhanceJobDao for SqliteEnhanceJobDao { .bind::, _>(u.queue_position) .bind::, _>(&u.pending) .bind::, _>(&u.saved) + .bind::, _>(u.run_ms) .bind::(chrono::Utc::now().timestamp()) .bind::(id) .execute(&mut *conn) @@ -285,6 +294,7 @@ mod tests { discarded: 0, created_at: created, updated_at: created, + run_ms: None, } } @@ -318,10 +328,18 @@ mod tests { queue_position: None, pending: Some("[{\"id\":\"x\"}]".into()), saved: None, + run_ms: Some(4200), }, ) .unwrap(); let a = d.get("a").unwrap().unwrap(); + assert_eq!(a.run_ms, Some(4200)); + d.update("a", &JobUpdate::status("ready")).unwrap(); + assert_eq!( + d.get("a").unwrap().unwrap().run_ms, + Some(4200), + "a null update never overwrites run_ms" + ); assert_eq!(a.status, "ready"); assert_eq!(a.pending, "[{\"id\":\"x\"}]"); assert_eq!(a.saved, "[]", "unset columns are left alone"); diff --git a/src/database/schema.rs b/src/database/schema.rs index e59accf..9e40375 100644 --- a/src/database/schema.rs +++ b/src/database/schema.rs @@ -60,6 +60,7 @@ diesel::table! { discarded -> Integer, created_at -> BigInt, updated_at -> BigInt, + run_ms -> Nullable, } } diff --git a/src/enhance/comfy.rs b/src/enhance/comfy.rs index 3189f65..0445265 100644 --- a/src/enhance/comfy.rs +++ b/src/enhance/comfy.rs @@ -112,6 +112,8 @@ pub struct HistoryEntry { pub staged: Vec, /// Last event time, unix millis. pub timestamp_ms: Option, + /// How long ComfyUI ran it: end event minus `execution_start`, in ms. + pub run_ms: Option, } #[derive(Debug, Clone, PartialEq, Eq, Default)] @@ -398,6 +400,22 @@ pub(crate) fn parse_history_entry(entry: &Value) -> HistoryEntry { .filter_map(|m| m.get(1)?.get("timestamp")?.as_i64()) .max(); + let event_ts = |pred: &dyn Fn(&str) -> bool| { + messages + .iter() + .filter(|m| m.get(0).and_then(Value::as_str).is_some_and(pred)) + .find_map(|m| m.get(1)?.get("timestamp")?.as_i64()) + }; + let run_ms = event_ts(&|k| k == "execution_start") + .zip(event_ts(&|k| { + matches!( + k, + "execution_success" | "execution_error" | "execution_interrupted" + ) + })) + .map(|(start, end)| end - start) + .filter(|ms| *ms >= 0); + let mut staged = Vec::new(); if let Some(outputs) = entry["outputs"].as_object() { for out in outputs.values() { @@ -424,5 +442,52 @@ pub(crate) fn parse_history_entry(entry: &Value) -> HistoryEntry { error, staged, timestamp_ms, + run_ms, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn entry(messages: Value) -> Value { + json!({"status": {"status_str": "success", "completed": true, "messages": messages}}) + } + + #[test] + fn run_ms_is_end_minus_start_on_success() { + let e = parse_history_entry(&entry(json!([ + ["execution_start", {"timestamp": 1000}], + ["execution_success", {"timestamp": 8500}] + ]))); + assert_eq!(e.run_ms, Some(7500)); + } + + #[test] + fn run_ms_is_measured_on_error_too() { + let e = parse_history_entry(&entry(json!([ + ["execution_start", {"timestamp": 1000}], + ["execution_error", {"timestamp": 1900, "exception_message": "boom"}] + ]))); + assert_eq!(e.run_ms, Some(900)); + assert_eq!(e.error.as_deref(), Some("boom")); + } + + #[test] + fn run_ms_is_none_without_a_start_or_end_or_when_negative() { + let no_start = parse_history_entry(&entry(json!([ + ["execution_success", {"timestamp": 8500}] + ]))); + assert_eq!(no_start.run_ms, None); + let no_end = parse_history_entry(&entry(json!([ + ["execution_start", {"timestamp": 1000}] + ]))); + assert_eq!(no_end.run_ms, None); + let negative = parse_history_entry(&entry(json!([ + ["execution_start", {"timestamp": 9000}], + ["execution_success", {"timestamp": 8500}] + ]))); + assert_eq!(negative.run_ms, None); } } diff --git a/src/enhance/handlers.rs b/src/enhance/handlers.rs index 1757b68..4714a36 100644 --- a/src/enhance/handlers.rs +++ b/src/enhance/handlers.rs @@ -85,6 +85,8 @@ pub struct JobView { reference_path: Option, created_at: i64, updated_at: i64, + /// Milliseconds ComfyUI ran the job; null until it has finished. + run_ms: Option, pending: Vec, saved: Vec, } @@ -118,6 +120,7 @@ pub fn job_view(job: &EnhanceJobRow) -> JobView { reference_path: job.reference_path.clone(), created_at: job.created_at, updated_at: job.updated_at, + run_ms: job.run_ms, pending, saved: saved .into_iter() diff --git a/src/enhance/integration_tests.rs b/src/enhance/integration_tests.rs index 6abe693..66008b9 100644 --- a/src/enhance/integration_tests.rs +++ b/src/enhance/integration_tests.rs @@ -569,6 +569,8 @@ async fn job_moves_queued_running_ready_and_the_preview_is_relayed() { fx.svc.poll_once().await; let (_, j, _) = call(&fx, "GET", &format!("/enhance/jobs/{id}"), None).await; assert_eq!(j["status"], "ready"); + // finished_entry runs 3000 ms before its end timestamp. + assert_eq!(j["run_ms"], 3000); assert!(j.get("queue_position").is_none()); assert_eq!(j["pending"].as_array().unwrap().len(), 1); assert_eq!(j["pending"][0]["n"], 0); @@ -580,6 +582,8 @@ async fn job_moves_queued_running_ready_and_the_preview_is_relayed() { let (st, _, bytes) = call(&fx, "GET", &uri, None).await; assert_eq!(st, 200); assert_eq!(bytes, b"PNG:imageapi_pending_abc.png"); + let (_, list0, _) = call(&fx, "GET", &format!("/enhance/jobs?path={PHOTO}"), None).await; + assert_eq!(list0["jobs"][0]["run_ms"], 3000, "list endpoint carries it"); let (st, _, _) = call(&fx, "GET", &format!("/enhance/jobs/{id}/preview/5"), None).await; assert_eq!(st, 404); @@ -616,6 +620,9 @@ async fn failed_run_reports_comfys_message() { .await; assert_eq!(j["status"], "failed"); assert_eq!(j["error"], "CUDA out of memory"); + // No execution_start in that history entry: the field is present but null. + assert!(j.as_object().unwrap().contains_key("run_ms")); + assert!(j["run_ms"].is_null()); } #[actix_rt::test] @@ -933,6 +940,7 @@ async fn unreachable_active_jobs_become_unavailable_after_the_grace_period() { discarded: 0, created_at: now, updated_at: now, + run_ms: None, }; SqliteEnhanceJobDao::from_shared(fx.conn.clone()) .insert(&dao_job) diff --git a/src/enhance/service.rs b/src/enhance/service.rs index 8a3ba53..253836a 100644 --- a/src/enhance/service.rs +++ b/src/enhance/service.rs @@ -361,6 +361,7 @@ impl EnhanceService { discarded: 0, created_at: now, updated_at: now, + run_ms: None, }; self.dao .lock() @@ -520,6 +521,7 @@ impl EnhanceService { .error .unwrap_or_else(|| "ComfyUI reported an error".to_string()), ); + u.run_ms = entry.run_ms; self.update_job(&job.id, &u); return; } @@ -531,6 +533,7 @@ impl EnhanceService { } let mut u = JobUpdate::status(STATUS_READY); u.pending = serde_json::to_string(&entry.staged).ok(); + u.run_ms = entry.run_ms; self.update_job(&job.id, &u); info!( "enhance: job {} ready ({} result(s))",