From 83cd26ee2e3a5c97659bfe8ddad9aade241669de Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Wed, 3 Jun 2026 20:11:13 +0200 Subject: [PATCH] fix(backend): report wall-clock duration for workflow-as-code roots Co-Authored-By: Claude Opus 4.8 --- ...7767b8a8fe0a638250f7c7777e5a9f7530e5c.json | 31 ----------- ...d2aa9ad5225ebbd0b77b32afc24473cb3f631.json | 31 +++++++++++ backend/tests/python_jobs.rs | 55 +++++++++++++++++-- backend/windmill-queue/src/jobs.rs | 11 +++- 4 files changed, 92 insertions(+), 36 deletions(-) delete mode 100644 backend/.sqlx/query-36c4e57afcab22f4b6825ccebe47767b8a8fe0a638250f7c7777e5a9f7530e5c.json create mode 100644 backend/.sqlx/query-4e62d7054b7bdb4be07035847b1d2aa9ad5225ebbd0b77b32afc24473cb3f631.json diff --git a/backend/.sqlx/query-36c4e57afcab22f4b6825ccebe47767b8a8fe0a638250f7c7777e5a9f7530e5c.json b/backend/.sqlx/query-36c4e57afcab22f4b6825ccebe47767b8a8fe0a638250f7c7777e5a9f7530e5c.json deleted file mode 100644 index e810fc4754..0000000000 --- a/backend/.sqlx/query-36c4e57afcab22f4b6825ccebe47767b8a8fe0a638250f7c7777e5a9f7530e5c.json +++ /dev/null @@ -1,31 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO v2_job_completed AS cj\n ( workspace_id\n , id\n , started_at\n , duration_ms\n , result\n , result_columns\n , canceled_by\n , canceled_reason\n , flow_status\n , workflow_as_code_status\n , memory_peak\n , status\n , worker\n )\n SELECT q.workspace_id, q.id, started_at, COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000), $3, $10, $5, $6,\n flow_status, workflow_as_code_status,\n $8, CASE WHEN $4::BOOL THEN 'canceled'::job_status\n WHEN $7::BOOL THEN 'skipped'::job_status\n WHEN $2::BOOL THEN 'success'::job_status\n ELSE 'failure'::job_status END AS status,\n q.worker\n FROM v2_job_queue q LEFT JOIN v2_job_status USING (id) WHERE q.id = $1\n ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3 RETURNING duration_ms AS \"duration_ms!\"", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "duration_ms!", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Bool", - "Jsonb", - "Bool", - "Varchar", - "Text", - "Bool", - "Int4", - "Int8", - "TextArray" - ] - }, - "nullable": [ - false - ] - }, - "hash": "36c4e57afcab22f4b6825ccebe47767b8a8fe0a638250f7c7777e5a9f7530e5c" -} diff --git a/backend/.sqlx/query-4e62d7054b7bdb4be07035847b1d2aa9ad5225ebbd0b77b32afc24473cb3f631.json b/backend/.sqlx/query-4e62d7054b7bdb4be07035847b1d2aa9ad5225ebbd0b77b32afc24473cb3f631.json new file mode 100644 index 0000000000..bbbecb8725 --- /dev/null +++ b/backend/.sqlx/query-4e62d7054b7bdb4be07035847b1d2aa9ad5225ebbd0b77b32afc24473cb3f631.json @@ -0,0 +1,31 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO v2_job_completed AS cj\n ( workspace_id\n , id\n , started_at\n , duration_ms\n , result\n , result_columns\n , canceled_by\n , canceled_reason\n , flow_status\n , workflow_as_code_status\n , memory_peak\n , status\n , worker\n )\n SELECT q.workspace_id, q.id, started_at,\n -- Workflow-as-code roots (identified by the `_checkpoint` written by the WAC\n -- executor) suspend while their task jobs run, so the worker-measured `$9`\n -- duration only covers the orchestration script's own compute, not the tasks.\n -- `started_at` is preserved across resumes (pull uses `coalesce(started_at, now())`),\n -- so fall back to the wall-clock elapsed time here, exactly like flows do.\n CASE WHEN workflow_as_code_status -> '_checkpoint' IS NOT NULL\n THEN (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000\n ELSE COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000)\n END, $3, $10, $5, $6,\n flow_status, workflow_as_code_status,\n $8, CASE WHEN $4::BOOL THEN 'canceled'::job_status\n WHEN $7::BOOL THEN 'skipped'::job_status\n WHEN $2::BOOL THEN 'success'::job_status\n ELSE 'failure'::job_status END AS status,\n q.worker\n FROM v2_job_queue q LEFT JOIN v2_job_status USING (id) WHERE q.id = $1\n ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3 RETURNING duration_ms AS \"duration_ms!\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "duration_ms!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Bool", + "Jsonb", + "Bool", + "Varchar", + "Text", + "Bool", + "Int4", + "Int8", + "TextArray" + ] + }, + "nullable": [ + false + ] + }, + "hash": "4e62d7054b7bdb4be07035847b1d2aa9ad5225ebbd0b77b32afc24473cb3f631" +} diff --git a/backend/tests/python_jobs.rs b/backend/tests/python_jobs.rs index 3a1550d395..9aa4cdcf4f 100644 --- a/backend/tests/python_jobs.rs +++ b/backend/tests/python_jobs.rs @@ -1283,11 +1283,19 @@ async def main(n: int): let db_ref = &db; + // Returns (final result, completed_steps checkpoint, wall-clock elapsed, + // stored parent `duration_ms`, DB wall-clock `completed_at - started_at` in ms). async fn run_once( db: &Pool, port: u16, content: String, - ) -> (serde_json::Value, serde_json::Value, std::time::Duration) { + ) -> ( + serde_json::Value, + serde_json::Value, + std::time::Duration, + i64, + f64, + ) { let mut job_id_out: Option = None; let mut result_out: Option = None; let t0 = std::time::Instant::now(); @@ -1340,17 +1348,56 @@ async def main(n: int): ) }); - (result_out.unwrap(), ckpt, elapsed) + // A WAC v2 root suspends while its steps run, so its stored `duration_ms` + // must be the end-to-end wall-clock (`now() - started_at`, like flows), + // NOT just the final replay's compute time. Fetch both the stored value + // and the DB-computed wall-clock to assert they agree. (Non-macro query + // so it needs no offline sqlx cache entry.) + let (duration_ms, wallclock_ms): (i64, f64) = sqlx::query_as( + "SELECT duration_ms, + (EXTRACT('epoch' FROM (completed_at - started_at)) * 1000)::float8 + FROM v2_job_completed WHERE id = $1", + ) + .bind(job_id) + .fetch_one(db) + .await + .expect("v2_job_completed timing fetch"); + + ( + result_out.unwrap(), + ckpt, + elapsed, + duration_ms, + wallclock_ms, + ) } // --- Legacy path: worker-side suspend & replay --- - let (legacy_result, legacy_ckpt, legacy_elapsed) = + let (legacy_result, legacy_ckpt, legacy_elapsed, legacy_duration_ms, legacy_wallclock_ms) = run_once(db_ref, port, workflow_content(false)).await; // --- Fast path: SDK persists the delta via the new API endpoint --- - let (fast_result, fast_ckpt, fast_elapsed) = + let (fast_result, fast_ckpt, fast_elapsed, fast_duration_ms, fast_wallclock_ms) = run_once(db_ref, port, workflow_content(true)).await; + // Regression check (workflow-as-code execution time): the WAC root's stored + // `duration_ms` must reflect the full end-to-end wall-clock — the same + // semantics flows use — and not just the orchestration script's final-replay + // compute time. Previously it excluded the suspend/replay round-trips spent + // running the steps, so it read far below the actual wall-clock. We allow a + // small tolerance for the sub-second gap between the worker's completion and + // the `now()` evaluated when the row is committed. + for (label, duration_ms, wallclock_ms) in [ + ("legacy", legacy_duration_ms, legacy_wallclock_ms), + ("fast", fast_duration_ms, fast_wallclock_ms), + ] { + assert!( + (duration_ms as f64 - wallclock_ms).abs() <= 250.0, + "{label} WAC v2 root duration_ms ({duration_ms}ms) should match the \ + end-to-end wall-clock ({wallclock_ms:.0}ms) like flows do" + ); + } + // Behavioral equivalence: same final result and same completed_steps. assert_eq!( legacy_result, fast_result, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 964fd558f1..1838558f1e 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -1068,7 +1068,16 @@ async fn commit_completed_job( , status , worker ) - SELECT q.workspace_id, q.id, started_at, COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000), $3, $10, $5, $6, + SELECT q.workspace_id, q.id, started_at, + -- Workflow-as-code roots (identified by the `_checkpoint` written by the WAC + -- executor) suspend while their task jobs run, so the worker-measured `$9` + -- duration only covers the orchestration script's own compute, not the tasks. + -- `started_at` is preserved across resumes (pull uses `coalesce(started_at, now())`), + -- so fall back to the wall-clock elapsed time here, exactly like flows do. + CASE WHEN workflow_as_code_status -> '_checkpoint' IS NOT NULL + THEN (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000 + ELSE COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000) + END, $3, $10, $5, $6, flow_status, workflow_as_code_status, $8, CASE WHEN $4::BOOL THEN 'canceled'::job_status WHEN $7::BOOL THEN 'skipped'::job_status