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 <noreply@anthropic.com>
This commit is contained in:
Cameron CordesandClaude Opus 5.5 committed 2026-10-30 19:08:48 -04:00
1 parent c7d810b812
commit 47ab6b3c00
9 files changed
+106 -4

No files matched your search

+3 -1
View File
@@ -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": <ms ComfyUI ran it, or null until finished / when ComfyUI gave no timings>,
"pending": [{"n","preview": "/enhance/jobs/{id}/preview/{n}","approved"}],
"saved": [{"n","path","library_id"}]}
```
@@ -0,0 +1 @@
ALTER TABLE enhance_jobs DROP COLUMN run_ms;
@@ -0,0 +1 @@
ALTER TABLE enhance_jobs ADD COLUMN run_ms BIGINT;
+21 -3
View File
@@ -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<BigInt>)]
pub run_ms: Option<i64>,
}
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<String>,
/// `Some` replaces the `saved` document.
pub saved: Option<String>,
/// `Some` sets the run time; `None` keeps any existing value.
pub run_ms: Option<i64>,
}
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::<Text, _>(&job.id)
.bind::<Text, _>(&job.original_hash)
@@ -156,6 +162,7 @@ impl EnhanceJobDao for SqliteEnhanceJobDao {
.bind::<Integer, _>(job.discarded)
.bind::<BigInt, _>(job.created_at)
.bind::<BigInt, _>(job.updated_at)
.bind::<Nullable<BigInt>, _>(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::<Text, _>(&u.status)
@@ -220,6 +228,7 @@ impl EnhanceJobDao for SqliteEnhanceJobDao {
.bind::<Nullable<Integer>, _>(u.queue_position)
.bind::<Nullable<Text>, _>(&u.pending)
.bind::<Nullable<Text>, _>(&u.saved)
.bind::<Nullable<BigInt>, _>(u.run_ms)
.bind::<BigInt, _>(chrono::Utc::now().timestamp())
.bind::<Text, _>(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");
+1
View File
@@ -60,6 +60,7 @@ diesel::table! {
discarded -> Integer,
created_at -> BigInt,
updated_at -> BigInt,
run_ms -> Nullable<BigInt>,
}
}
+65
View File
@@ -112,6 +112,8 @@ pub struct HistoryEntry {
pub staged: Vec<Staged>,
/// Last event time, unix millis.
pub timestamp_ms: Option<i64>,
/// How long ComfyUI ran it: end event minus `execution_start`, in ms.
pub run_ms: Option<i64>,
}
#[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);
}
}
+3
View File
@@ -85,6 +85,8 @@ pub struct JobView {
reference_path: Option<String>,
created_at: i64,
updated_at: i64,
/// Milliseconds ComfyUI ran the job; null until it has finished.
run_ms: Option<i64>,
pending: Vec<PendingView>,
saved: Vec<SavedView>,
}
@@ -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()
+8
View File
@@ -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)
+3
View File
@@ -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))",