starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-25 08:00:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(backend): report wall-clock duration for workflow-as-code roots

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
2026-06-03 20:11:13 +02:00
parent c19441bc8c
commit 83cd26ee2e
4 changed files with 92 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-36c4e57afcab22f4b6825ccebe47767b8a8fe0a638250f7c7777e5a9f7530e5c.json
Generated
-31
View File
@@ -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"
}
backend/.sqlx/query-4e62d7054b7bdb4be07035847b1d2aa9ad5225ebbd0b77b32afc24473cb3f631.json
Generated
+31
View File
@@ -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"
}
backend/tests/python_jobs.rs
+51 -4
View File
@@ -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<Postgres>,
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<sqlx::types::Uuid> = None;
let mut result_out: Option<serde_json::Value> = 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,
backend/windmill-queue/src/jobs.rs
+10 -1
View File
@@ -1068,7 +1068,16 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
, 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