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

fix: store a WAC parent's first start and wall duration (#11625)

Browse Source 
* fix: store a workflow-as-code parent's first start and wall duration

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: keep a parked child's first start in its parent's timeline

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: record a parked step with its whole span in the flow, cover inline checkpoints

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* chore: update ee-repo-ref to ab8476e5feedc98b727f7d83acffa1807b3b9c5b

This commit updates the EE repository reference after PR #861 was merged in windmill-ee-private.

Previous ee-repo-ref: d5d95f51bb175944874e585e5931acb1eddd637a

New ee-repo-ref: ab8476e5feedc98b727f7d83acffa1807b3b9c5b

Automated by sync-ee-ref workflow.

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
2026-10-08 18:05:37 +02:00
committed by GitHub
co-authored by Claude Opus 5.5 windmill-internal-app[bot]
parent e233908c1e
commit 8f28a58605
18 changed files with 389 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-1333436e3d0466eccdaf99fb70aadad56b50619036fab3f2c1bdbbf280a80217.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "WITH prev AS (SELECT started_at FROM v2_job_queue WHERE id = $1)\n UPDATE v2_job_queue q SET running = false, started_at = null,\n extras = CASE WHEN q.started_at IS NULL OR q.extras ? 'wac_first_started_at' THEN q.extras\n ELSE jsonb_set(COALESCE(q.extras, '{}'::jsonb), '{wac_first_started_at}', to_jsonb(q.started_at))\n END\n FROM prev WHERE q.id = $1\n RETURNING (extract(epoch FROM now() - prev.started_at) * 1000)::bigint",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "int8",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "1333436e3d0466eccdaf99fb70aadad56b50619036fab3f2c1bdbbf280a80217"
}
backend/.sqlx/query-71c9e0c0f1d9eba0f24c8e410ca2f2a64dc559e3228d3b763f68b1ad3491ae6e.json
Generated
+45
View File
@@ -0,0 +1,45 @@
{
"db_name": "PostgreSQL",
"query": "WITH deleted AS (\n DELETE FROM v2_job_queue WHERE id = $1\n RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason,\n (extras->>'wac_first_started_at')::timestamptz AS first_started_at,\n COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000)::bigint AS run_ms\n ), completed AS (\n INSERT INTO v2_job_completed AS cj\n ( workspace_id\n , id\n , started_at\n , duration_ms\n , extras\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 d.workspace_id, d.id, COALESCE(d.first_started_at, d.started_at),\n CASE WHEN d.first_started_at IS NULL THEN d.run_ms\n ELSE EXTRACT('epoch' FROM (now() - d.first_started_at))*1000 END,\n CASE WHEN d.first_started_at IS NOT NULL\n THEN jsonb_build_object('wac_last_segment_ms', d.run_ms) END,\n $3::text::jsonb, $10,\n CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,\n CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,\n s.flow_status, s.workflow_as_code_status, $8,\n CASE WHEN $4::BOOL OR d.canceled_by IS NOT NULL 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,\n d.worker\n FROM deleted d LEFT JOIN v2_job_status s ON s.id = d.id\n ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb,\n canceled_by = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,\n canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END\n RETURNING cj.started_at, cj.duration_ms,\n COALESCE((cj.extras->>'wac_last_segment_ms')::bigint, cj.duration_ms) AS last_run_ms\n ), parent_ping AS (\n UPDATE v2_job_runtime r SET ping = now()\n FROM v2_job_queue q\n WHERE r.id = $11 AND q.id = r.id AND q.workspace_id = $12\n AND EXISTS (SELECT 1 FROM completed)\n )\n SELECT started_at AS \"started_at?\", duration_ms AS \"duration_ms!\",\n last_run_ms AS \"last_run_ms!\" FROM completed",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "started_at?",
"type_info": "Timestamptz"
},
{
"ordinal": 1,
"name": "duration_ms!",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "last_run_ms!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid",
"Bool",
"Text",
"Bool",
"Varchar",
"Text",
"Bool",
"Int4",
"Int8",
"TextArray",
"Uuid",
"Text"
]
},
"nullable": [
true,
false,
null
]
},
"hash": "71c9e0c0f1d9eba0f24c8e410ca2f2a64dc559e3228d3b763f68b1ad3491ae6e"
}
backend/.sqlx/query-9212902d4eb2834ab3b429968a7a85c53c49e4ef0f42c146324d5c71fbd1e21d.json
Generated
-31
View File
@@ -1,31 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH deleted AS (\n DELETE FROM v2_job_queue WHERE id = $1\n RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason\n ), completed AS (\n 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 d.workspace_id, d.id, d.started_at,\n COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(d.started_at, now()))))*1000),\n $3::text::jsonb, $10,\n CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,\n CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,\n s.flow_status, s.workflow_as_code_status, $8,\n CASE WHEN $4::BOOL OR d.canceled_by IS NOT NULL 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,\n d.worker\n FROM deleted d LEFT JOIN v2_job_status s ON s.id = d.id\n ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb,\n canceled_by = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,\n canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END\n RETURNING duration_ms\n )\n SELECT duration_ms AS \"duration_ms!\" FROM completed",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "duration_ms!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid",
"Bool",
"Text",
"Bool",
"Varchar",
"Text",
"Bool",
"Int4",
"Int8",
"TextArray"
]
},
"nullable": [
false
]
},
"hash": "9212902d4eb2834ab3b429968a7a85c53c49e4ef0f42c146324d5c71fbd1e21d"
}
backend/.sqlx/query-975158ae72ec12f3f933fba1d9623bf1c925feebe746299cd1f0eda3f8b949d5.json
Generated
-33
View File
@@ -1,33 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH deleted AS (\n DELETE FROM v2_job_queue WHERE id = $1\n RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason\n ), completed AS (\n 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 d.workspace_id, d.id, d.started_at,\n COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(d.started_at, now()))))*1000),\n $3::text::jsonb, $10,\n CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,\n CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,\n s.flow_status, s.workflow_as_code_status, $8,\n CASE WHEN $4::BOOL OR d.canceled_by IS NOT NULL 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,\n d.worker\n FROM deleted d LEFT JOIN v2_job_status s ON s.id = d.id\n ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb,\n canceled_by = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,\n canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END\n RETURNING duration_ms\n ), parent_ping AS (\n UPDATE v2_job_runtime r SET ping = now()\n FROM v2_job_queue q\n WHERE r.id = $11 AND q.id = r.id AND q.workspace_id = $12\n AND EXISTS (SELECT 1 FROM completed)\n )\n SELECT duration_ms AS \"duration_ms!\" FROM completed",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "duration_ms!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid",
"Bool",
"Text",
"Bool",
"Varchar",
"Text",
"Bool",
"Int4",
"Int8",
"TextArray",
"Uuid",
"Text"
]
},
"nullable": [
false
]
},
"hash": "975158ae72ec12f3f933fba1d9623bf1c925feebe746299cd1f0eda3f8b949d5"
}
backend/.sqlx/query-9e7a5e1ebba42217092971dd340cd9182a4812b14735e502958eaafeb4c6b33d.json
Generated
+17
View File
@@ -0,0 +1,17 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE v2_job_queue\n SET suspend = $3, suspend_until = now() + make_interval(secs => $4), started_at = null,\n extras = CASE WHEN started_at IS NULL OR extras ? 'wac_first_started_at' THEN extras\n ELSE jsonb_set(COALESCE(extras, '{}'::jsonb), '{wac_first_started_at}', to_jsonb(started_at))\n END\n WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Text",
"Int4",
"Float8"
]
},
"nullable": []
},
"hash": "9e7a5e1ebba42217092971dd340cd9182a4812b14735e502958eaafeb4c6b33d"
}
backend/.sqlx/query-a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665.json
Generated
-28
View File
@@ -1,28 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT usr.email, usage.executions\n FROM usr, LATERAL (\n SELECT COALESCE(SUM(c.duration_ms + 1000)/1000 , 0)::BIGINT executions\n FROM v2_job_completed c JOIN v2_job j USING (id)\n WHERE j.workspace_id = $1\n AND j.kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow')\n AND j.permissioned_as_email = usr.email\n AND now() - '1 week'::interval < j.created_at\n ) usage\n WHERE workspace_id = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "executions",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
null
]
},
"hash": "a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665"
}
backend/.sqlx/query-ce39fd9650500b16f7247c8173a40c7e5d2d8a51aea2d36f2bc8c6f86d53a9b8.json
Generated
+43
View File
@@ -0,0 +1,43 @@
{
"db_name": "PostgreSQL",
"query": "WITH deleted AS (\n DELETE FROM v2_job_queue WHERE id = $1\n RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason,\n (extras->>'wac_first_started_at')::timestamptz AS first_started_at,\n COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000)::bigint AS run_ms\n ), completed AS (\n INSERT INTO v2_job_completed AS cj\n ( workspace_id\n , id\n , started_at\n , duration_ms\n , extras\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 d.workspace_id, d.id, COALESCE(d.first_started_at, d.started_at),\n CASE WHEN d.first_started_at IS NULL THEN d.run_ms\n ELSE EXTRACT('epoch' FROM (now() - d.first_started_at))*1000 END,\n CASE WHEN d.first_started_at IS NOT NULL\n THEN jsonb_build_object('wac_last_segment_ms', d.run_ms) END,\n $3::text::jsonb, $10,\n CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,\n CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,\n s.flow_status, s.workflow_as_code_status, $8,\n CASE WHEN $4::BOOL OR d.canceled_by IS NOT NULL 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,\n d.worker\n FROM deleted d LEFT JOIN v2_job_status s ON s.id = d.id\n ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb,\n canceled_by = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,\n canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END\n RETURNING cj.started_at, cj.duration_ms,\n COALESCE((cj.extras->>'wac_last_segment_ms')::bigint, cj.duration_ms) AS last_run_ms\n )\n SELECT started_at AS \"started_at?\", duration_ms AS \"duration_ms!\",\n last_run_ms AS \"last_run_ms!\" FROM completed",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "started_at?",
"type_info": "Timestamptz"
},
{
"ordinal": 1,
"name": "duration_ms!",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "last_run_ms!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid",
"Bool",
"Text",
"Bool",
"Varchar",
"Text",
"Bool",
"Int4",
"Int8",
"TextArray"
]
},
"nullable": [
true,
false,
null
]
},
"hash": "ce39fd9650500b16f7247c8173a40c7e5d2d8a51aea2d36f2bc8c6f86d53a9b8"
}
backend/.sqlx/query-e12f852fa196d16862151ab490b05e6f5c25e23b7c41ce1c0e5a83bf7b118bfd.json
Generated
-22
View File
@@ -1,22 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH prev AS (SELECT started_at FROM v2_job_queue WHERE id = $1)\n UPDATE v2_job_queue q SET running = false, started_at = null\n FROM prev WHERE q.id = $1\n RETURNING (extract(epoch FROM now() - prev.started_at) * 1000)::bigint",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "int8",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "e12f852fa196d16862151ab490b05e6f5c25e23b7c41ce1c0e5a83bf7b118bfd"
}
backend/.sqlx/query-e8779f1cddc475554dbbd2e3122db8a21a2eaacf971bf046cef1ee14cf04e323.json
Generated
+28
View File
@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT usr.email, usage.executions\n FROM usr, LATERAL (\n SELECT COALESCE(SUM(COALESCE((c.extras->>'wac_last_segment_ms')::bigint, c.duration_ms) + 1000)/1000 , 0)::BIGINT executions\n FROM v2_job_completed c JOIN v2_job j USING (id)\n WHERE j.workspace_id = $1\n AND j.kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow')\n AND j.permissioned_as_email = usr.email\n AND now() - '1 week'::interval < j.created_at\n ) usage\n WHERE workspace_id = $1\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "executions",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
null
]
},
"hash": "e8779f1cddc475554dbbd2e3122db8a21a2eaacf971bf046cef1ee14cf04e323"
}
backend/.sqlx/query-f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed.json
Generated
-17
View File
@@ -1,17 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE v2_job_queue\n SET suspend = $3, suspend_until = now() + make_interval(secs => $4), started_at = null\n WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Text",
"Int4",
"Float8"
]
},
"nullable": []
},
"hash": "f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
a37413ccdd52e65bc240bad3e87084bf4100e119
ab8476e5feedc98b727f7d83acffa1807b3b9c5b
backend/tests/wac_suspend_started_at.rs
+88
View File
@@ -109,3 +109,91 @@ async fn wac_suspend_stands_down_for_a_cancel(db: Pool<Postgres>) -> anyhow::Res
Ok(())
}
/// Each park clears `started_at`, so the completion would otherwise store only the last
/// segment: a workflow that waited an hour on its tasks would read as a sub-second job that
/// started an hour late. The completed row spans the workflow from its first start, and
/// keeps the last segment aside: that is what is metered and what worker-time readers sum.
#[sqlx::test]
async fn wac_completion_spans_the_workflow_and_keeps_the_last_segment(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
let job_id = Uuid::new_v4();
sqlx::query(
"INSERT INTO v2_job (id, workspace_id, created_by, created_at, permissioned_as, \
permissioned_as_email, kind, script_lang, runnable_path, tag, visible_to_owner) \
VALUES ($1, 'test-workspace', 'test-user', now() - interval '2 hours', 'u/test-user', \
'test@windmill.dev', 'script', 'bun', 'u/test-user/wac', 'bun', true)",
)
.bind(job_id)
.execute(&db)
.await?;
sqlx::query(
"INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running, tag, started_at) \
VALUES ($1, 'test-workspace', now(), true, 'bun', now() - interval '1 hour')",
)
.bind(job_id)
.execute(&db)
.await?;
// Two parks, each followed by the pull re-stamping `started_at`: the second park must
// not replace the first start with its own segment's.
for resumed_ago in ["10 minutes", "3 seconds"] {
let mut tx = db.begin().await?;
suspend_wac_parent(&mut tx, &job_id, "test-workspace", 1, 3600.0).await?;
tx.commit().await?;
sqlx::query(
"UPDATE v2_job_queue SET suspend = 0, started_at = now() - $2::text::interval \
WHERE id = $1",
)
.bind(job_id)
.bind(resumed_ago)
.execute(&db)
.await?;
}
let job = windmill_queue::get_mini_completed_job(&job_id, "test-workspace", &db)
.await?
.unwrap();
let result = serde_json::value::to_raw_value(&serde_json::json!("done"))?;
let (_, span) = windmill_queue::add_completed_job(
&db,
&job,
true,
false,
sqlx::types::Json(&result),
None,
0,
None,
false,
Some(3000),
false,
)
.await?;
let (started_ago_s, duration_ms, last_segment_ms): (f64, i64, Option<i64>) = sqlx::query_as(
"SELECT extract(epoch FROM now() - started_at)::float8, duration_ms, \
(extras->>'wac_last_segment_ms')::bigint \
FROM v2_job_completed WHERE id = $1",
)
.bind(job_id)
.fetch_one(&db)
.await?;
assert!(
(started_ago_s - 3600.0).abs() < 60.0,
"started_at must be the first start, got {started_ago_s}s ago"
);
assert!(
(duration_ms - 3_600_000).abs() < 60_000,
"duration_ms must be the wall time since the first start, got {duration_ms}"
);
assert_eq!(last_segment_ms, Some(3000));
assert_eq!(
span.duration_ms, duration_ms,
"a flow records the step with the span the row holds"
);
assert!(span.outlasts_run(job.started_at));
assert_eq!(span.last_run_ms, 3000, "only the last segment is left to meter");
Ok(())
}
backend/windmill-api-users/src/users.rs
+1 -1
View File
@@ -490,7 +490,7 @@ async fn list_user_usage(
"
SELECT usr.email, usage.executions
FROM usr, LATERAL (
SELECT COALESCE(SUM(c.duration_ms + 1000)/1000 , 0)::BIGINT executions
SELECT COALESCE(SUM(COALESCE((c.extras->>'wac_last_segment_ms')::bigint, c.duration_ms) + 1000)/1000 , 0)::BIGINT executions
FROM v2_job_completed c JOIN v2_job j USING (id)
WHERE j.workspace_id = $1
AND j.kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlestepflow')
backend/windmill-queue/src/jobs.rs
+100 -32
View File
@@ -1221,7 +1221,7 @@ pub async fn add_completed_job_pre_shaped_failure<T: Serialize + Send + Sync + V
worker_name: &str,
flow_is_done: bool,
duration: Option<i64>,
) -> Result<(), Error> {
) -> Result<CompletedSpan, Error> {
record_failure_metrics(completed_job, worker_name).await;
tracing::error!(
@@ -1229,7 +1229,7 @@ pub async fn add_completed_job_pre_shaped_failure<T: Serialize + Send + Sync + V
completed_job.id,
completed_job.workspace_id,
);
let _ = add_completed_job(
let (_, span) = add_completed_job(
db,
completed_job,
false,
@@ -1244,7 +1244,7 @@ pub async fn add_completed_job_pre_shaped_failure<T: Serialize + Send + Sync + V
)
.warn_after_seconds(10)
.await?;
Ok(())
Ok(span)
}
pub async fn add_completed_job_error(
@@ -1257,6 +1257,31 @@ pub async fn add_completed_job_error(
flow_is_done: bool,
duration: Option<i64>,
) -> Result<WrappedError, Error> {
add_completed_job_error_with_span(
db,
completed_job,
mem_peak,
canceled_by,
e,
worker_name,
flow_is_done,
duration,
)
.await
.map(|(wrapped, _)| wrapped)
}
/// `add_completed_job_error`, also handing back what the completed row holds.
pub async fn add_completed_job_error_with_span(
db: &Pool<Postgres>,
completed_job: &MiniCompletedJob,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
e: serde_json::Value,
worker_name: &str,
flow_is_done: bool,
duration: Option<i64>,
) -> Result<(WrappedError, CompletedSpan), Error> {
record_failure_metrics(completed_job, worker_name).await;
let result = WrappedError { error: e };
@@ -1266,7 +1291,7 @@ pub async fn add_completed_job_error(
completed_job.workspace_id,
serde_json::to_string(&result).unwrap_or_else(|_| "".to_string())
);
let _ = add_completed_job(
let (_, span) = add_completed_job(
db,
&completed_job,
false,
@@ -1281,7 +1306,7 @@ pub async fn add_completed_job_error(
)
.warn_after_seconds(10)
.await?;
Ok(result)
Ok((result, span))
}
lazy_static::lazy_static! {
@@ -1320,7 +1345,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
flow_is_done: bool,
duration: Option<i64>,
from_cache: bool,
) -> Result<(Uuid, i64), Error> {
) -> Result<(Uuid, CompletedSpan), Error> {
// tracing::error!("Start");
// let start = tokio::time::Instant::now();
@@ -1354,7 +1379,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
};
let result_columns = result_columns.as_ref();
let (opt_uuid, duration, _skip_downstream_error_handlers, wac_parent_ready) = (|| {
let (opt_uuid, span, _skip_downstream_error_handlers, wac_parent_ready) = (|| {
commit_completed_job(
db,
completed_job,
@@ -1395,7 +1420,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
// if scheduling next job failed, return the job_id early to ensure the job get retried after a timeout
if let Some(job_id) = opt_uuid {
return Ok((job_id, duration));
return Ok((job_id, span));
}
// Auto-resolve a retry chain that ultimately worked, from whichever of the two
@@ -1436,7 +1461,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
}
#[cfg(feature = "cloud")]
apply_completed_job_cloud_usage(db, completed_job, duration);
apply_completed_job_cloud_usage(db, completed_job, span.last_run_ms);
#[cfg(all(feature = "enterprise", feature = "private"))]
crate::jobs_ee::apply_completed_job_error_handlers(
@@ -1453,7 +1478,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
// tracing::error!("4 {:?}", start.elapsed());
Ok((completed_job.id, duration))
Ok((completed_job.id, span))
}
async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
@@ -1471,7 +1496,7 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
// True when a native script retry was enqueued for this failed attempt, i.e.
// this is not the terminal attempt — schedule completion handlers must wait.
retry_pending: bool,
) -> windmill_common::error::Result<(Option<Uuid>, i64, bool, bool)> {
) -> windmill_common::error::Result<(Option<Uuid>, CompletedSpan, bool, bool)> {
// let start = std::time::Instant::now();
let job_id = completed_job.id;
@@ -1538,11 +1563,11 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
&& monitor_parent.is_none()
&& !is_scheduled
{
let Some(duration) = completion.execute(&mut *db.acquire().await?).await? else {
let Some(completed) = completion.execute(&mut *db.acquire().await?).await? else {
return Err(not_in_queue_error(db, job_id).await);
};
log_completed_job(completed_job, duration, success);
return Ok((None, duration, false, false));
log_completed_job(completed_job, completed.duration_ms, success);
return Ok((None, completed, false, false));
}
let mut tx = db.begin().warn_after_seconds(10).await?;
@@ -1550,7 +1575,8 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
// The parent's rows are locked ahead of the child's own queue row (see
// `record_child_completion` for the order this must keep), so the duration it stamps is read
// before the completion: the one the completed row will hold. `now()` is fixed for the
// transaction, and a completed row already there keeps its own duration.
// transaction, and a completed row already there keeps its own duration. For a child that
// itself parked it is the last run's, matching the last start its timeline entry carries.
let mut wac_parent_ready = false;
if let Some(parent_job) = wac_parent {
let Some(duration) = sqlx::query_scalar!(
@@ -1577,7 +1603,7 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
.await?;
}
let Some(duration) = completion.execute(&mut *tx).await? else {
let Some(completed) = completion.execute(&mut *tx).await? else {
return Err(not_in_queue_error(&mut *tx, job_id).await);
};
@@ -1787,16 +1813,35 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
tx.commit().warn_after_seconds(10).await?;
log_completed_job(completed_job, duration, success);
log_completed_job(completed_job, completed.duration_ms, success);
// tracing::info!("completed job: {:?}", start.elapsed().as_micros());
Ok((
None,
duration,
completed,
_skip_downstream_error_handlers,
wac_parent_ready,
))
}
/// What a job's completion wrote. `started_at` and `duration_ms` are the completed row's.
/// `last_run_ms` is the time the job's last run held a worker, which is what gets metered;
/// it differs from `duration_ms` only for a job that parked.
#[derive(Debug, Clone, Copy)]
pub struct CompletedSpan {
pub started_at: Option<chrono::DateTime<chrono::Utc>>,
pub duration_ms: i64,
pub last_run_ms: i64,
}
impl CompletedSpan {
/// Whether the row spans more than the run that started at `last_started_at`, i.e. the
/// job parked and resumed. A flow records such a step with the row's span, so the step
/// reads like the job does.
pub fn outlasts_run(&self, last_started_at: Option<chrono::DateTime<chrono::Utc>>) -> bool {
matches!((self.started_at, last_started_at), (Some(first), Some(last)) if first < last)
}
}
/// What a job's completion writes, for `Completion::execute`.
struct Completion<'a> {
completed_job: &'a MiniCompletedJob,
@@ -1813,6 +1858,11 @@ impl Completion<'_> {
/// Moves the job from the queue to the completed jobs and refreshes a flow step's parent
/// ping, as one statement. Returns `None` when the job was no longer in the queue.
///
/// For a job that parked (a Workflow-as-Code parent, see `suspend_wac_parent`) the row's
/// `started_at` and `duration_ms` span the whole workflow from its first start, and the
/// last run goes to `extras.wac_last_segment_ms` for readers that sum worker time from
/// completed rows.
///
/// The completion takes its cancellation from the queue row it deletes, not only from
/// `canceled_by`: that is what the worker last read, and the delete waits for a cancel still
/// being written, so the deleted row is the final word on whether the job was canceled.
@@ -1820,7 +1870,7 @@ impl Completion<'_> {
/// It locks the queue row before the completed row's key. Any other writer completing a job
/// (the monitor's zombie fallback, debounce) must take them in the same order, or the two
/// deadlock.
async fn execute(&self, conn: &mut sqlx::PgConnection) -> error::Result<Option<i64>> {
async fn execute(&self, conn: &mut sqlx::PgConnection) -> error::Result<Option<CompletedSpan>> {
let Completion {
completed_job,
success,
@@ -1851,16 +1901,20 @@ impl Completion<'_> {
.then_some(completed_job.parent_job)
.flatten()
else {
return sqlx::query_scalar!(
return sqlx::query_as!(
CompletedSpan,
"WITH deleted AS (
DELETE FROM v2_job_queue WHERE id = $1
RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason
RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason,
(extras->>'wac_first_started_at')::timestamptz AS first_started_at,
COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000)::bigint AS run_ms
), completed AS (
INSERT INTO v2_job_completed AS cj
( workspace_id
, id
, started_at
, duration_ms
, extras
, result
, result_columns
, canceled_by
@@ -1871,8 +1925,11 @@ impl Completion<'_> {
, status
, worker
)
SELECT d.workspace_id, d.id, d.started_at,
COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(d.started_at, now()))))*1000),
SELECT d.workspace_id, d.id, COALESCE(d.first_started_at, d.started_at),
CASE WHEN d.first_started_at IS NULL THEN d.run_ms
ELSE EXTRACT('epoch' FROM (now() - d.first_started_at))*1000 END,
CASE WHEN d.first_started_at IS NOT NULL
THEN jsonb_build_object('wac_last_segment_ms', d.run_ms) END,
$3::text::jsonb, $10,
CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,
CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,
@@ -1888,9 +1945,11 @@ impl Completion<'_> {
THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,
canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL
THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END
RETURNING duration_ms
RETURNING cj.started_at, cj.duration_ms,
COALESCE((cj.extras->>'wac_last_segment_ms')::bigint, cj.duration_ms) AS last_run_ms
)
SELECT duration_ms AS \"duration_ms!\" FROM completed",
SELECT started_at AS \"started_at?\", duration_ms AS \"duration_ms!\",
last_run_ms AS \"last_run_ms!\" FROM completed",
/* $1 */ completed_job.id,
/* $2 */ success,
/* $3 */ result,
@@ -1910,16 +1969,20 @@ impl Completion<'_> {
// A canceled flow is pinged too: it is completed by its next transition like any other
// flow, and the zombie flow monitor needs the ping to finish the cancel if that
// transition is lost.
sqlx::query_scalar!(
sqlx::query_as!(
CompletedSpan,
"WITH deleted AS (
DELETE FROM v2_job_queue WHERE id = $1
RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason
RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason,
(extras->>'wac_first_started_at')::timestamptz AS first_started_at,
COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000)::bigint AS run_ms
), completed AS (
INSERT INTO v2_job_completed AS cj
( workspace_id
, id
, started_at
, duration_ms
, extras
, result
, result_columns
, canceled_by
@@ -1930,8 +1993,11 @@ impl Completion<'_> {
, status
, worker
)
SELECT d.workspace_id, d.id, d.started_at,
COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(d.started_at, now()))))*1000),
SELECT d.workspace_id, d.id, COALESCE(d.first_started_at, d.started_at),
CASE WHEN d.first_started_at IS NULL THEN d.run_ms
ELSE EXTRACT('epoch' FROM (now() - d.first_started_at))*1000 END,
CASE WHEN d.first_started_at IS NOT NULL
THEN jsonb_build_object('wac_last_segment_ms', d.run_ms) END,
$3::text::jsonb, $10,
CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,
CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,
@@ -1947,14 +2013,16 @@ impl Completion<'_> {
THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,
canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL
THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END
RETURNING duration_ms
RETURNING cj.started_at, cj.duration_ms,
COALESCE((cj.extras->>'wac_last_segment_ms')::bigint, cj.duration_ms) AS last_run_ms
), parent_ping AS (
UPDATE v2_job_runtime r SET ping = now()
FROM v2_job_queue q
WHERE r.id = $11 AND q.id = r.id AND q.workspace_id = $12
AND EXISTS (SELECT 1 FROM completed)
)
SELECT duration_ms AS \"duration_ms!\" FROM completed",
SELECT started_at AS \"started_at?\", duration_ms AS \"duration_ms!\",
last_run_ms AS \"last_run_ms!\" FROM completed",
/* $1 */ completed_job.id,
/* $2 */ success,
/* $3 */ result,
@@ -2021,7 +2089,7 @@ async fn check_result_size<T: ValidableJson>(
db: &Pool<Postgres>,
queued_job: &MiniCompletedJob,
result: Json<&T>,
) -> Option<Result<(Option<Uuid>, i64, bool, bool), Error>> {
) -> Option<Result<(Option<Uuid>, CompletedSpan, bool, bool), Error>> {
let result_size = result.size() / 1024 / 1024;
if result_size > 2 {
if result_size > *MAX_RESULT_SIZE_MB {
backend/windmill-worker/src/bun_executor.rs
+4 -1
View File
@@ -3731,7 +3731,10 @@ pub async fn handle_wac_v2_output(
// status row, the order every child completion takes.
let segment_ms = sqlx::query_scalar!(
"WITH prev AS (SELECT started_at FROM v2_job_queue WHERE id = $1)
UPDATE v2_job_queue q SET running = false, started_at = null
UPDATE v2_job_queue q SET running = false, started_at = null,
extras = CASE WHEN q.started_at IS NULL OR q.extras ? 'wac_first_started_at' THEN q.extras
ELSE jsonb_set(COALESCE(q.extras, '{}'::jsonb), '{wac_first_started_at}', to_jsonb(q.started_at))
END
FROM prev WHERE q.id = $1
RETURNING (extract(epoch FROM now() - prev.started_at) * 1000)::bigint",
job.id,
backend/windmill-worker/src/result_processor.rs
+27 -13
View File
@@ -1776,7 +1776,7 @@ pub async fn process_completed_job(
add_time!(bench, "pre add_completed_job");
let (_, duration) = add_completed_job(
let (_, span) = add_completed_job(
db,
&job,
true,
@@ -1835,7 +1835,10 @@ pub async fn process_completed_job(
true,
canceled_by,
result,
started_at.map(|x| FlowJobDuration { started_at: x, duration_ms: duration }),
span.started_at.or(started_at).map(|x| FlowJobDuration {
started_at: x,
duration_ms: span.duration_ms,
}),
StepFailureKind::Normal,
&same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(),
&worker_dir,
@@ -1867,9 +1870,10 @@ pub async fn process_completed_job(
// `wm_failure` field) so a real runtime failure whose raw
// result happens to contain a `wm_failure` field still goes
// through the standard `WrappedError { error: ... }` wrap path.
let downstream_result: Arc<Box<RawValue>> = if is_pre_shaped_wm_failure_result(result.get())
{
windmill_queue::add_completed_job_pre_shaped_failure(
let (downstream_result, span): (Arc<Box<RawValue>>, _) = if is_pre_shaped_wm_failure_result(
result.get(),
) {
let span = windmill_queue::add_completed_job_pre_shaped_failure(
db,
&job,
mem_peak.to_owned(),
@@ -1880,9 +1884,9 @@ pub async fn process_completed_job(
None,
)
.await?;
result.clone()
(result.clone(), span)
} else {
let wrapped = add_completed_job_error(
let (wrapped, span) = windmill_queue::add_completed_job_error_with_span(
db,
&job,
mem_peak.to_owned(),
@@ -1895,7 +1899,10 @@ pub async fn process_completed_job(
None,
)
.await?;
Arc::new(serde_json::value::to_raw_value(&wrapped).unwrap())
(
Arc::new(serde_json::value::to_raw_value(&wrapped).unwrap()),
span,
)
};
#[cfg(all(feature = "enterprise", feature = "private"))]
if job.kind == JobKind::DeploymentCallback {
@@ -1927,12 +1934,19 @@ pub async fn process_completed_job(
false,
canceled_by,
downstream_result,
duration.and_then(|d| {
job.started_at.map(|started_at| FlowJobDuration {
started_at: started_at,
duration_ms: d,
if span.outlasts_run(job.started_at) {
span.started_at.map(|started_at| FlowJobDuration {
started_at,
duration_ms: span.duration_ms,
})
}),
} else {
duration.and_then(|d| {
job.started_at.map(|started_at| FlowJobDuration {
started_at: started_at,
duration_ms: d,
})
})
},
StepFailureKind::Normal,
&same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(),
&worker_dir,
backend/windmill-worker/src/wac_executor.rs
+9 -1
View File
@@ -113,6 +113,11 @@ pub enum WacPark {
/// handler — falls back to `now() - started_at`. Left pointing at the first segment,
/// that fallback reports the whole sleep or approval wait as execution time.
///
/// The first park keeps that first start in `extras.wac_first_started_at`, so the
/// completed row can still say when the workflow began and how long it took overall
/// (`Completion::execute`). The queue row's `started_at` stays per segment. The inline
/// checkpoint requeue clears `started_at` on its own and keeps the first start the same way.
///
/// Call it before any write to the parent's `v2_job_status` row in the same
/// transaction: a child's completion locks the queue row and then the status row
/// (`record_child_completion`), and taking them the other way round here can
@@ -156,7 +161,10 @@ pub async fn suspend_wac_parent(
sqlx::query!(
"UPDATE v2_job_queue
SET suspend = $3, suspend_until = now() + make_interval(secs => $4), started_at = null
SET suspend = $3, suspend_until = now() + make_interval(secs => $4), started_at = null,
extras = CASE WHEN started_at IS NULL OR extras ? 'wac_first_started_at' THEN extras
ELSE jsonb_set(COALESCE(extras, '{}'::jsonb), '{wac_first_started_at}', to_jsonb(started_at))
END
WHERE id = $1 AND workspace_id = $2",
job_id,
w_id,
backend/windmill-worker/src/worker_flow.rs
+4 -4
View File
@@ -2097,7 +2097,7 @@ pub async fn update_flow_status_after_job_completion_internal(
)
.await?;
let duration = if success {
let (_, duration) = add_completed_job(
let (_, span) = add_completed_job(
db,
&cflow_job,
true,
@@ -2111,9 +2111,9 @@ pub async fn update_flow_status_after_job_completion_internal(
false,
)
.await?;
duration
span.duration_ms
} else {
let (_, duration) = add_completed_job(
let (_, span) = add_completed_job(
db,
&cflow_job,
false,
@@ -2131,7 +2131,7 @@ pub async fn update_flow_status_after_job_completion_internal(
false,
)
.await?;
duration
span.duration_ms
};
flow_job_duration = flow_job
.started_at