Merge pull request 'Enhance: expose ComfyUI run time as run_ms' (#197) from feat/enhance-run-ms into master

Reviewed-on: #197

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": <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"}]}
 ```

migrations/2026-10-30-200000_add_enhance_run_ms/down.sql

@@ -0,0 +1 @@
+ALTER TABLE enhance_jobs DROP COLUMN run_ms;

migrations/2026-10-30-200000_add_enhance_run_ms/up.sql

@@ -0,0 +1 @@
+ALTER TABLE enhance_jobs ADD COLUMN run_ms BIGINT;

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<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");

src/database/schema.rs

@@ -60,6 +60,7 @@ diesel::table! {
         discarded -> Integer,
         created_at -> BigInt,
         updated_at -> BigInt,
+        run_ms -> Nullable<BigInt>,
     }
 }
 

src/enhance/comfy.rs

@@ -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);
     }
 }

src/enhance/handlers.rs

@@ -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()

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)

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))",