diff --git a/backend/.sqlx/query-1333436e3d0466eccdaf99fb70aadad56b50619036fab3f2c1bdbbf280a80217.json b/backend/.sqlx/query-1333436e3d0466eccdaf99fb70aadad56b50619036fab3f2c1bdbbf280a80217.json new file mode 100644 index 0000000000..1946f55366 --- /dev/null +++ b/backend/.sqlx/query-1333436e3d0466eccdaf99fb70aadad56b50619036fab3f2c1bdbbf280a80217.json @@ -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" +} diff --git a/backend/.sqlx/query-71c9e0c0f1d9eba0f24c8e410ca2f2a64dc559e3228d3b763f68b1ad3491ae6e.json b/backend/.sqlx/query-71c9e0c0f1d9eba0f24c8e410ca2f2a64dc559e3228d3b763f68b1ad3491ae6e.json new file mode 100644 index 0000000000..ba1cc13fac --- /dev/null +++ b/backend/.sqlx/query-71c9e0c0f1d9eba0f24c8e410ca2f2a64dc559e3228d3b763f68b1ad3491ae6e.json @@ -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" +} diff --git a/backend/.sqlx/query-9212902d4eb2834ab3b429968a7a85c53c49e4ef0f42c146324d5c71fbd1e21d.json b/backend/.sqlx/query-9212902d4eb2834ab3b429968a7a85c53c49e4ef0f42c146324d5c71fbd1e21d.json deleted file mode 100644 index f105dd7825..0000000000 --- a/backend/.sqlx/query-9212902d4eb2834ab3b429968a7a85c53c49e4ef0f42c146324d5c71fbd1e21d.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-975158ae72ec12f3f933fba1d9623bf1c925feebe746299cd1f0eda3f8b949d5.json b/backend/.sqlx/query-975158ae72ec12f3f933fba1d9623bf1c925feebe746299cd1f0eda3f8b949d5.json deleted file mode 100644 index 0b166b32a4..0000000000 --- a/backend/.sqlx/query-975158ae72ec12f3f933fba1d9623bf1c925feebe746299cd1f0eda3f8b949d5.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-9e7a5e1ebba42217092971dd340cd9182a4812b14735e502958eaafeb4c6b33d.json b/backend/.sqlx/query-9e7a5e1ebba42217092971dd340cd9182a4812b14735e502958eaafeb4c6b33d.json new file mode 100644 index 0000000000..59ed0e1767 --- /dev/null +++ b/backend/.sqlx/query-9e7a5e1ebba42217092971dd340cd9182a4812b14735e502958eaafeb4c6b33d.json @@ -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" +} diff --git a/backend/.sqlx/query-a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665.json b/backend/.sqlx/query-a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665.json deleted file mode 100644 index 2358af807f..0000000000 --- a/backend/.sqlx/query-a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-ce39fd9650500b16f7247c8173a40c7e5d2d8a51aea2d36f2bc8c6f86d53a9b8.json b/backend/.sqlx/query-ce39fd9650500b16f7247c8173a40c7e5d2d8a51aea2d36f2bc8c6f86d53a9b8.json new file mode 100644 index 0000000000..ba7a89ef73 --- /dev/null +++ b/backend/.sqlx/query-ce39fd9650500b16f7247c8173a40c7e5d2d8a51aea2d36f2bc8c6f86d53a9b8.json @@ -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" +} diff --git a/backend/.sqlx/query-e12f852fa196d16862151ab490b05e6f5c25e23b7c41ce1c0e5a83bf7b118bfd.json b/backend/.sqlx/query-e12f852fa196d16862151ab490b05e6f5c25e23b7c41ce1c0e5a83bf7b118bfd.json deleted file mode 100644 index ae893b6791..0000000000 --- a/backend/.sqlx/query-e12f852fa196d16862151ab490b05e6f5c25e23b7c41ce1c0e5a83bf7b118bfd.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-e8779f1cddc475554dbbd2e3122db8a21a2eaacf971bf046cef1ee14cf04e323.json b/backend/.sqlx/query-e8779f1cddc475554dbbd2e3122db8a21a2eaacf971bf046cef1ee14cf04e323.json new file mode 100644 index 0000000000..e22953610d --- /dev/null +++ b/backend/.sqlx/query-e8779f1cddc475554dbbd2e3122db8a21a2eaacf971bf046cef1ee14cf04e323.json @@ -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" +} diff --git a/backend/.sqlx/query-f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed.json b/backend/.sqlx/query-f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed.json deleted file mode 100644 index 5f86fb97d8..0000000000 --- a/backend/.sqlx/query-f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed.json +++ /dev/null @@ -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" -} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 1acd2de3f0..2de431e127 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -a37413ccdd52e65bc240bad3e87084bf4100e119 +ab8476e5feedc98b727f7d83acffa1807b3b9c5b diff --git a/backend/tests/wac_suspend_started_at.rs b/backend/tests/wac_suspend_started_at.rs index 5a83730d24..ca364e5993 100644 --- a/backend/tests/wac_suspend_started_at.rs +++ b/backend/tests/wac_suspend_started_at.rs @@ -109,3 +109,91 @@ async fn wac_suspend_stands_down_for_a_cancel(db: Pool) -> 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, +) -> 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) = 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(()) +} diff --git a/backend/windmill-api-users/src/users.rs b/backend/windmill-api-users/src/users.rs index ac9defcd53..46bd6825e8 100644 --- a/backend/windmill-api-users/src/users.rs +++ b/backend/windmill-api-users/src/users.rs @@ -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') diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index e8e3716d2b..389ae1d112 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -1221,7 +1221,7 @@ pub async fn add_completed_job_pre_shaped_failure, -) -> Result<(), Error> { +) -> Result { record_failure_metrics(completed_job, worker_name).await; tracing::error!( @@ -1229,7 +1229,7 @@ pub async fn add_completed_job_pre_shaped_failure, ) -> Result { + 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, + completed_job: &MiniCompletedJob, + mem_peak: i32, + canceled_by: Option, + e: serde_json::Value, + worker_name: &str, + flow_is_done: bool, + duration: Option, +) -> 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( flow_is_done: bool, duration: Option, 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( }; 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( // 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( } #[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( // tracing::error!("4 {:?}", start.elapsed()); - Ok((completed_job.id, duration)) + Ok((completed_job.id, span)) } async fn commit_completed_job( @@ -1471,7 +1496,7 @@ async fn commit_completed_job( // 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, i64, bool, bool)> { +) -> windmill_common::error::Result<(Option, CompletedSpan, bool, bool)> { // let start = std::time::Instant::now(); let job_id = completed_job.id; @@ -1538,11 +1563,11 @@ async fn commit_completed_job( && 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( // 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( .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( 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>, + 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>) -> 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> { + async fn execute(&self, conn: &mut sqlx::PgConnection) -> error::Result> { 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( db: &Pool, queued_job: &MiniCompletedJob, result: Json<&T>, -) -> Option, i64, bool, bool), Error>> { +) -> Option, CompletedSpan, bool, bool), Error>> { let result_size = result.size() / 1024 / 1024; if result_size > 2 { if result_size > *MAX_RESULT_SIZE_MB { diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index bb47d3e4be..1ae588ef11 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -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, diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 0490ed1754..12edd7823e 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -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> = if is_pre_shaped_wm_failure_result(result.get()) - { - windmill_queue::add_completed_job_pre_shaped_failure( + let (downstream_result, span): (Arc>, _) = 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, diff --git a/backend/windmill-worker/src/wac_executor.rs b/backend/windmill-worker/src/wac_executor.rs index 45d76c74e8..3cc9c1e23a 100644 --- a/backend/windmill-worker/src/wac_executor.rs +++ b/backend/windmill-worker/src/wac_executor.rs @@ -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, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 6d76d0d771..4f67780f18 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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