From d685332771dcdfc718d7a2077fe1cdb3b323909b Mon Sep 17 00:00:00 2001 From: hugocasa Date: Thu, 8 Oct 2026 15:26:51 +0200 Subject: [PATCH 01/38] fix: send hub script completions to the workspace handlers (#11613) * fix: send hub script completions to the workspace handlers Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref to a37413ccdd52e65bc240bad3e87084bf4100e119 This commit updates the EE repository reference after PR #858 was merged in windmill-ee-private. Previous ee-repo-ref: 6e48f2012d73a6ef22130f4e28b987526b9c1057 New ee-repo-ref: a37413ccdd52e65bc240bad3e87084bf4100e119 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: windmill-internal-app[bot] --- backend/ee-repo-ref.txt | 2 +- backend/tests/error_handler.rs | 67 ++++++++++++++++++++++++++++++++ backend/tests/success_handler.rs | 67 ++++++++++++++++++++++++++++++++ 3 files changed, 135 insertions(+), 1 deletion(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 14bd926cf5..1acd2de3f0 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -127575148c84bb6132d2946c0a95c6dea0f58bb2 +a37413ccdd52e65bc240bad3e87084bf4100e119 diff --git a/backend/tests/error_handler.rs b/backend/tests/error_handler.rs index 4f58685221..b1f740566f 100644 --- a/backend/tests/error_handler.rs +++ b/backend/tests/error_handler.rs @@ -411,3 +411,70 @@ async fn test_error_handler_not_triggered_on_success(db: Pool) -> anyh Ok(()) } + +/// A hub script run at the top level (a schedule, a trigger, the API) is a `script_hub` job, +/// and its failure must reach the workspace error handler like a workspace script's. +#[cfg(all(feature = "enterprise", feature = "private"))] +#[sqlx::test(fixtures("base"))] +async fn test_error_handler_triggered_on_hub_script_failure( + db: Pool, +) -> anyhow::Result<()> { + use windmill_common::jobs::JobPayload; + + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + + let version = "990000002"; + let hub_dir = &*windmill_common::worker::HUB_CACHE_DIR; + tokio::fs::create_dir_all(hub_dir).await?; + tokio::fs::write( + format!("{hub_dir}/{version}"), + r#"{"content":"exit 1","lockfile":null,"language":"bash","schema":{},"summary":null}"#, + ) + .await?; + + // The workspace handler settings are cached per workspace id across the tests of this + // binary, so this reuses the handler path every other test here configures. + sqlx::query( + "INSERT INTO script (workspace_id, hash, path, content, language, kind, created_by, schema, summary, description, lock) + VALUES ('test-workspace', 1111111112, 'f/test/error_handler', 'echo handled', 'bash', 'script', 'test-user', '{}', '', '', '')", + ) + .execute(&db) + .await?; + sqlx::query( + r#"UPDATE workspace_settings SET error_handler = '{"path": "script/f/test/error_handler"}'::jsonb + WHERE workspace_id = 'test-workspace'"#, + ) + .execute(&db) + .await?; + + let completed_job = RunJob::from(JobPayload::ScriptHub { + path: format!("hub/{version}/test/fail"), + apply_preprocessor: false, + }) + .run_until_complete(&db, false, server.addr.port()) + .await; + assert!(!completed_job.success, "Hub script should have failed"); + + let mut handler_parent = None; + for _ in 0..50 { + handler_parent = sqlx::query_scalar::<_, Option>( + "SELECT parent_job FROM v2_job + WHERE workspace_id = 'test-workspace' AND permissioned_as_email = 'error_handler@windmill.dev'", + ) + .fetch_optional(&db) + .await?; + if handler_parent.is_some() { + break; + } + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; + } + assert_eq!( + handler_parent, + Some(Some(completed_job.id)), + "Workspace error handler should handle the failed hub script" + ); + + Ok(()) +} diff --git a/backend/tests/success_handler.rs b/backend/tests/success_handler.rs index 8fc4182d9e..f876c1984b 100644 --- a/backend/tests/success_handler.rs +++ b/backend/tests/success_handler.rs @@ -239,3 +239,70 @@ export async function main(path: string, email: string, job_id: string, is_flow: Ok(()) } + +/// A hub script run at the top level (a schedule, a trigger, the API) is a `script_hub` job, +/// and its success must reach the workspace success handler like a workspace script's. +#[cfg(all(feature = "enterprise", feature = "private"))] +#[sqlx::test(fixtures("base"))] +async fn test_success_handler_triggered_on_hub_script_success( + db: Pool, +) -> anyhow::Result<()> { + use windmill_common::jobs::JobPayload; + + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + + let version = "990000003"; + let hub_dir = &*windmill_common::worker::HUB_CACHE_DIR; + tokio::fs::create_dir_all(hub_dir).await?; + tokio::fs::write( + format!("{hub_dir}/{version}"), + r#"{"content":"echo ok","lockfile":null,"language":"bash","schema":{},"summary":null}"#, + ) + .await?; + + // The workspace handler settings are cached per workspace id across the tests of this + // binary, so this reuses the handler path every other test here configures. + sqlx::query( + "INSERT INTO script (workspace_id, hash, path, content, language, kind, created_by, schema, summary, description, lock) + VALUES ('test-workspace', 1234567891, 'f/test/success_handler', 'echo handled', 'bash', 'script', 'test-user', '{}', '', '', '')", + ) + .execute(&db) + .await?; + sqlx::query( + r#"UPDATE workspace_settings SET success_handler = '{"path": "script/f/test/success_handler"}'::jsonb + WHERE workspace_id = 'test-workspace'"#, + ) + .execute(&db) + .await?; + + let completed_job = RunJob::from(JobPayload::ScriptHub { + path: format!("hub/{version}/test/ok"), + apply_preprocessor: false, + }) + .run_until_complete(&db, false, server.addr.port()) + .await; + assert!(completed_job.success, "Hub script should have succeeded"); + + let mut handler_parent = None; + for _ in 0..50 { + handler_parent = sqlx::query_scalar::<_, Option>( + "SELECT parent_job FROM v2_job + WHERE workspace_id = 'test-workspace' AND permissioned_as_email = 'success_handler@windmill.dev'", + ) + .fetch_optional(&db) + .await?; + if handler_parent.is_some() { + break; + } + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; + } + assert_eq!( + handler_parent, + Some(Some(completed_job.id)), + "Workspace success handler should handle the successful hub script" + ); + + Ok(()) +} From e233908c1e99a246dc7c94fa86ea82c5cbe9d6b3 Mon Sep 17 00:00:00 2001 From: Guilhem Date: Thu, 8 Oct 2026 17:31:35 +0200 Subject: [PATCH 02/38] fix: name the open workspace settings tab in the page header breadcrumb (#11617) * feat: name the open workspace settings tab in the page header breadcrumb Co-Authored-By: Claude Opus 5.5 (1M context) * fix: let the page header breadcrumb shrink when nothing else in the bar can Co-Authored-By: Claude Opus 5.5 (1M context) * fix: keep the breadcrumb's slower shrink when a summary shares the bar Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) --- .../src/lib/components/NavBreadcrumb.svelte | 9 +++++++++ .../src/lib/components/PageHeaderBar.svelte | 17 ++++++++++++++--- .../lib/components/pageHeaderRegistry.svelte.ts | 3 +++ .../(logged)/workspace_settings/+page.svelte | 12 +++++++++++- 4 files changed, 37 insertions(+), 4 deletions(-) diff --git a/frontend/src/lib/components/NavBreadcrumb.svelte b/frontend/src/lib/components/NavBreadcrumb.svelte index 889fe17377..3081bcca4c 100644 --- a/frontend/src/lib/components/NavBreadcrumb.svelte +++ b/frontend/src/lib/components/NavBreadcrumb.svelte @@ -396,6 +396,15 @@ not this component) — so the lead can run the whole line. onPick={() => {}} /> {:else if section} + {#if section.parent} + + + {section.parent} + + {@render slash()} + {/if} -
+ page the controls beside it belong to. + The 0.3 only holds against a summary or an actions box that shrinks too: alone, a factor + sum below 1 makes flex hand out only that share of the overflow, and the bar overflows + instead. --> +
@@ -161,7 +172,7 @@ The row's height matches the sidebar's own header row, so the two read as one ba
- {#if item && (item.summaryContent || item.summary)} + {#if item && hasSummary}
diff --git a/frontend/src/lib/components/pageHeaderRegistry.svelte.ts b/frontend/src/lib/components/pageHeaderRegistry.svelte.ts index 3167d636e9..bd30855e15 100644 --- a/frontend/src/lib/components/pageHeaderRegistry.svelte.ts +++ b/frontend/src/lib/components/pageHeaderRegistry.svelte.ts @@ -24,6 +24,9 @@ export type PageHeaderItem = { /** A page with no item of its own, such as a list: the breadcrumb is the section alone. */ export type PageHeaderSection = { label: string + /** The page the section belongs to, when the section is one pane of it — a settings page's + * active tab. Drawn as a lead segment before `label`. */ + parent?: string /** Rendered in place of `label` when the name is itself a control, such as an editable title. */ content?: Snippet /** Scopes the segment's picker, when the section maps to a kind of item. */ diff --git a/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte b/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte index 022048bbf7..1eda558a4a 100644 --- a/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte @@ -1457,6 +1457,12 @@ } ] ) + + // The label of the entry the sidebar highlights, so the breadcrumb and the sidebar name the + // same pane: both read `tab`, which resolves the URL's aliases (`teams`, `deploy_to`, …). + const tabLabel = $derived( + navigationGroups.flatMap((g) => g.items).find((i) => i.id === tab)?.label + ) @@ -1464,7 +1470,11 @@ - +
From 8f28a5860595dbe558e6777a06bb97b6ec6a3e67 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 8 Oct 2026 18:05:37 +0200 Subject: [PATCH 03/38] fix: store a WAC parent's first start and wall duration (#11625) * fix: store a workflow-as-code parent's first start and wall duration Co-Authored-By: Claude Opus 5.5 * fix: keep a parked child's first start in its parent's timeline Co-Authored-By: Claude Opus 5.5 * fix: record a parked step with its whole span in the flow, cover inline checkpoints Co-Authored-By: Claude Opus 5.5 * 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 Co-authored-by: windmill-internal-app[bot] --- ...adad56b50619036fab3f2c1bdbbf280a80217.json | 22 +++ ...2f2a64dc559e3228d3b763f68b1ad3491ae6e.json | 45 ++++++ ...a85c53c49e4ef0f42c146324d5c71fbd1e21d.json | 31 ---- ...23bf1c925feebe746299cd1f0eda3f8b949d5.json | 33 ----- ...cd9182a4812b14735e502958eaafeb4c6b33d.json | 17 +++ ...571de793b377dd6058065f51b12c974e5b665.json | 28 ---- ...40c7e5d2d8a51aea2d36f2bc8c6f86d53a9b8.json | 43 ++++++ ...05e6f5c25e23b7c41ce1c0e5a83bf7b118bfd.json | 22 --- ...db8a21a2eaacf971bf046cef1ee14cf04e323.json | 28 ++++ ...49bf8b3523d8dde1ef6b8f1950423830cc6ed.json | 17 --- backend/ee-repo-ref.txt | 2 +- backend/tests/wac_suspend_started_at.rs | 88 ++++++++++++ backend/windmill-api-users/src/users.rs | 2 +- backend/windmill-queue/src/jobs.rs | 132 +++++++++++++----- backend/windmill-worker/src/bun_executor.rs | 5 +- .../windmill-worker/src/result_processor.rs | 40 ++++-- backend/windmill-worker/src/wac_executor.rs | 10 +- backend/windmill-worker/src/worker_flow.rs | 8 +- 18 files changed, 389 insertions(+), 184 deletions(-) create mode 100644 backend/.sqlx/query-1333436e3d0466eccdaf99fb70aadad56b50619036fab3f2c1bdbbf280a80217.json create mode 100644 backend/.sqlx/query-71c9e0c0f1d9eba0f24c8e410ca2f2a64dc559e3228d3b763f68b1ad3491ae6e.json delete mode 100644 backend/.sqlx/query-9212902d4eb2834ab3b429968a7a85c53c49e4ef0f42c146324d5c71fbd1e21d.json delete mode 100644 backend/.sqlx/query-975158ae72ec12f3f933fba1d9623bf1c925feebe746299cd1f0eda3f8b949d5.json create mode 100644 backend/.sqlx/query-9e7a5e1ebba42217092971dd340cd9182a4812b14735e502958eaafeb4c6b33d.json delete mode 100644 backend/.sqlx/query-a2f6ca89faaa8d8739f67cd1aae571de793b377dd6058065f51b12c974e5b665.json create mode 100644 backend/.sqlx/query-ce39fd9650500b16f7247c8173a40c7e5d2d8a51aea2d36f2bc8c6f86d53a9b8.json delete mode 100644 backend/.sqlx/query-e12f852fa196d16862151ab490b05e6f5c25e23b7c41ce1c0e5a83bf7b118bfd.json create mode 100644 backend/.sqlx/query-e8779f1cddc475554dbbd2e3122db8a21a2eaacf971bf046cef1ee14cf04e323.json delete mode 100644 backend/.sqlx/query-f0498c9bddc0d4d8948167f3ed849bf8b3523d8dde1ef6b8f1950423830cc6ed.json 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 From 5d90f96863b64e9bc88ed35429f8068a6a788d1e Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 8 Oct 2026 22:17:44 +0200 Subject: [PATCH 04/38] feat: export worker cpu, memory and occupancy metrics (#11629) * feat: export worker cpu, memory and occupancy metrics Co-Authored-By: Claude Opus 5.5 * fix: refresh the exported memory limit with the cgroup readings Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref to e128a9b9bedd0b6a46d355b69db13bf8f650c4e6 This commit updates the EE repository reference after PR #864 was merged in windmill-ee-private. Previous ee-repo-ref: b286a486c69e74ec0742a2aac9d4cb1161eb0c6b New ee-repo-ref: e128a9b9bedd0b6a46d355b69db13bf8f650c4e6 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: windmill-internal-app[bot] --- backend/ee-repo-ref.txt | 2 +- backend/tests/otel.rs | 27 +- backend/windmill-common/src/otel_oss.rs | 33 +++ backend/windmill-common/src/worker.rs | 93 +++++++ backend/windmill-worker/src/handle_child.rs | 1 + backend/windmill-worker/src/lib.rs | 1 + .../windmill-worker/src/resource_metrics.rs | 235 ++++++++++++++++++ backend/windmill-worker/src/worker.rs | 1 + backend/windmill-worker/src/worker_utils.rs | 13 +- 9 files changed, 402 insertions(+), 4 deletions(-) create mode 100644 backend/windmill-worker/src/resource_metrics.rs diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 2de431e127..d292c22c2e 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -ab8476e5feedc98b727f7d83acffa1807b3b9c5b +e128a9b9bedd0b6a46d355b69db13bf8f650c4e6 diff --git a/backend/tests/otel.rs b/backend/tests/otel.rs index f1e042037f..30585c08a1 100644 --- a/backend/tests/otel.rs +++ b/backend/tests/otel.rs @@ -176,6 +176,10 @@ async fn test_all_metrics_e2e() { otel_set_worker_uptime("w-uptime", 3600.0); otel_set_health_status_phase("healthy"); otel_set_health_db_unresponsive(true); + otel_set_worker_memory("w-res", "gpu", Some(512), Some(128), Some(1024)); + otel_add_worker_cpu("w-res", "gpu", 1.5, 40, 7, 0.35); + otel_set_worker_occupancy_rate("w-res", "gpu", "5m", 0.25); + otel_record_job_memory_peak("python3", 100 * 1024 * 1024); // ── Histograms ────────────────────────────────────────────────── @@ -189,7 +193,7 @@ async fn test_all_metrics_e2e() { let metrics = flush_and_get_metrics(&state); let names = metric_names(&metrics); - // ── Verify all 21 metric names are present ────────────────────── + // ── Verify all metric names are present ───────────────────────── let expected = [ "windmill.queue.push_count", @@ -213,6 +217,15 @@ async fn test_all_metrics_e2e() { "windmill.worker.uptime", "windmill.health.status", "windmill.health.db_unresponsive", + "windmill.worker.memory_usage", + "windmill.worker.windmill_memory_usage", + "windmill.worker.memory_limit", + "windmill.worker.cpu_usage", + "windmill.worker.cpu_periods", + "windmill.worker.cpu_throttled_periods", + "windmill.worker.cpu_throttled_time", + "windmill.worker.occupancy_rate", + "windmill.worker.job_memory_peak", ]; for name in expected { assert!( @@ -290,6 +303,18 @@ async fn test_all_metrics_e2e() { .expect("worker.busy data point with worker=worker-test-1 not found"); assert_eq!(dp.1, 1); + let m = find_metric(&metrics, "windmill.worker.memory_limit").unwrap(); + let values = gauge_i64_values(m); + let dp = values + .iter() + .find(|(attrs, _)| { + attrs + .iter() + .any(|kv| kv.key.as_str() == "worker_group" && kv.value.as_str() == "gpu") + }) + .expect("worker.memory_limit data point with worker_group=gpu not found"); + assert_eq!(dp.1, 1024); + let m = find_metric(&metrics, "windmill.db.pool.active").unwrap(); assert_eq!(gauge_i64_values(m)[0].1, 5); let m = find_metric(&metrics, "windmill.db.pool.idle").unwrap(); diff --git a/backend/windmill-common/src/otel_oss.rs b/backend/windmill-common/src/otel_oss.rs index bd0a5202df..95e635dd8f 100644 --- a/backend/windmill-common/src/otel_oss.rs +++ b/backend/windmill-common/src/otel_oss.rs @@ -132,3 +132,36 @@ pub fn otel_set_health_status_phase(_phase: &str) {} #[cfg(not(feature = "private"))] pub fn otel_set_health_db_unresponsive(_unresponsive: bool) {} + +#[cfg(not(feature = "private"))] +pub fn otel_set_worker_memory( + _worker: &str, + _worker_group: &str, + _usage: Option, + _windmill_usage: Option, + _limit: Option, +) { +} + +#[cfg(not(feature = "private"))] +pub fn otel_add_worker_cpu( + _worker: &str, + _worker_group: &str, + _usage_secs: f64, + _periods: u64, + _throttled_periods: u64, + _throttled_secs: f64, +) { +} + +#[cfg(not(feature = "private"))] +pub fn otel_set_worker_occupancy_rate( + _worker: &str, + _worker_group: &str, + _window: &'static str, + _rate: f64, +) { +} + +#[cfg(not(feature = "private"))] +pub fn otel_record_job_memory_peak(_tag: &str, _bytes: u64) {} diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 52fa2da905..017c244670 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -1783,6 +1783,73 @@ pub fn get_cpu_period() -> Option { Some(100000) } +/// Bucket bounds in bytes, 16 MiB to 32 GiB, shared by the Prometheus and OTel job +/// peak-memory histograms so the two exports stay comparable. +pub const JOB_MEMORY_PEAK_BUCKETS: [f64; 12] = { + let mut buckets = [0.0; 12]; + let mut i = 0; + while i < 12 { + buckets[i] = (16u64 << (20 + i)) as f64; + i += 1; + } + buckets +}; + +/// Cumulative CPU accounting of the worker's cgroup. The throttling fields stay at 0 +/// when the cgroup has no CPU controller, which is also what an unthrottled one reports. +#[derive(Debug, Clone, Copy, Default, PartialEq)] +pub struct CpuStat { + pub usage_usec: u64, + pub nr_periods: u64, + pub nr_throttled: u64, + pub throttled_usec: u64, +} + +fn cpu_stat_field(stat: &str, key: &str) -> Option { + stat.lines().find_map(|line| { + let (k, v) = line.split_once(' ')?; + (k == key).then(|| v.trim().parse().ok()).flatten() + }) +} + +fn parse_cgroup_v2_cpu_stat(stat: &str) -> Option { + Some(CpuStat { + usage_usec: cpu_stat_field(stat, "usage_usec")?, + nr_periods: cpu_stat_field(stat, "nr_periods").unwrap_or(0), + nr_throttled: cpu_stat_field(stat, "nr_throttled").unwrap_or(0), + throttled_usec: cpu_stat_field(stat, "throttled_usec").unwrap_or(0), + }) +} + +/// cgroup v1 splits the same figures over two controllers and counts time in nanoseconds. +fn parse_cgroup_v1_cpu_stat(cpuacct_usage: &str, stat: &str) -> Option { + Some(CpuStat { + usage_usec: cpuacct_usage.trim().parse::().ok()? / 1000, + nr_periods: cpu_stat_field(stat, "nr_periods").unwrap_or(0), + nr_throttled: cpu_stat_field(stat, "nr_throttled").unwrap_or(0), + throttled_usec: cpu_stat_field(stat, "throttled_time").unwrap_or(0) / 1000, + }) +} + +#[cfg(not(windows))] +pub fn get_cpu_stat() -> Option { + const V1_USAGE: &str = "/sys/fs/cgroup/cpuacct/cpuacct.usage"; + if Path::new(V1_USAGE).exists() { + parse_cgroup_v1_cpu_stat( + &std::fs::read_to_string(V1_USAGE).ok()?, + &std::fs::read_to_string("/sys/fs/cgroup/cpu/cpu.stat").unwrap_or_default(), + ) + } else { + let cgroup_path = get_cgroupv2_path()?; + parse_cgroup_v2_cpu_stat(&std::fs::read_to_string(format!("{cgroup_path}/cpu.stat")).ok()?) + } +} + +#[cfg(windows)] +pub fn get_cpu_stat() -> Option { + None +} + /// CPUs the process is allowed to run on, ignoring any bandwidth quota — the count /// Go's `NumCPU` reports. `available_parallelism` cannot stand in for it: that folds /// the quota in, so a fraction of a CPU makes it report a single-core machine. @@ -2919,6 +2986,32 @@ mod tests { use super::*; use std::collections::HashMap; + #[test] + fn cgroup_cpu_stat_is_normalized_to_microseconds() { + let v2 = "usage_usec 2500000\nuser_usec 2000000\nsystem_usec 500000\n\ + nr_periods 40\nnr_throttled 7\nthrottled_usec 350000\nnr_bursts 0\n"; + let expected = CpuStat { + usage_usec: 2_500_000, + nr_periods: 40, + nr_throttled: 7, + throttled_usec: 350_000, + }; + assert_eq!(parse_cgroup_v2_cpu_stat(v2), Some(expected)); + + let v1_stat = "nr_periods 40\nnr_throttled 7\nthrottled_time 350000000\n"; + assert_eq!( + parse_cgroup_v1_cpu_stat("2500000000\n", v1_stat), + Some(expected) + ); + + // No CPU controller: usage is still reported, throttling reads as zero. + assert_eq!( + parse_cgroup_v2_cpu_stat("usage_usec 12\nuser_usec 8\nsystem_usec 4\n"), + Some(CpuStat { usage_usec: 12, ..Default::default() }) + ); + assert_eq!(parse_cgroup_v2_cpu_stat("nr_periods 40\n"), None); + } + #[test] fn datatable_role_is_read_from_the_leading_comment_block() { let role = |code| SqlAnnotations::datatable_role(code); diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index c25ad82a83..84fd547641 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -1049,6 +1049,7 @@ where let wm_memory_usage = get_windmill_memory_usage(); tracing::info!("job {job_id} on {worker_name} in {w_id} worker memory snapshot {}kB/{}kB", memory_usage.unwrap_or_default()/1024, wm_memory_usage.unwrap_or_default()/1024); let occupancy = occupancy_metrics.as_mut().map(|x| x.update_occupancy_metrics()); + crate::resource_metrics::record_worker_resources(worker_name, memory_usage, wm_memory_usage, occupancy.as_ref()); if job_id != Uuid::nil() { if let Err(err) = update_worker_ping_from_job(&conn, &job_id, w_id, worker_name, memory_usage, wm_memory_usage, occupancy).await { tracing::error!("Unable to update worker ping for job {} in workspace {}. Error was: {:?}", job_id, w_id, err); diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 6059936af6..aa1d9ab209 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -81,6 +81,7 @@ mod prepare_deps; mod python_executor; #[cfg(feature = "python")] mod python_versions; +mod resource_metrics; pub mod result_processor; #[cfg(feature = "rust")] mod rust_executor; diff --git a/backend/windmill-worker/src/resource_metrics.rs b/backend/windmill-worker/src/resource_metrics.rs new file mode 100644 index 0000000000..53c7dc5b4d --- /dev/null +++ b/backend/windmill-worker/src/resource_metrics.rs @@ -0,0 +1,235 @@ +use std::sync::{atomic::Ordering, Mutex, OnceLock}; + +use windmill_common::{ + otel_oss::{ + otel_add_worker_cpu, otel_record_job_memory_peak, otel_set_worker_memory, + otel_set_worker_occupancy_rate, + }, + worker::{get_cpu_stat, get_memory, CpuStat, WORKER_GROUP}, + OTEL_METRICS_ENABLED, +}; + +use crate::common::OccupancyResult; + +#[cfg(feature = "prometheus")] +struct PromMetrics { + memory_usage: prometheus::IntGaugeVec, + windmill_memory_usage: prometheus::IntGaugeVec, + memory_limit: prometheus::IntGaugeVec, + cpu_usage: prometheus::CounterVec, + cpu_periods: prometheus::IntCounterVec, + cpu_throttled_periods: prometheus::IntCounterVec, + cpu_throttled_seconds: prometheus::CounterVec, + occupancy_rate: prometheus::GaugeVec, + job_memory_peak: prometheus::HistogramVec, +} + +#[cfg(feature = "prometheus")] +lazy_static::lazy_static! { + static ref PROM_METRICS: Option = + if windmill_common::METRICS_ENABLED.load(Ordering::Relaxed) { + const WORKER: &[&str] = &["name", "worker_group"]; + Some(PromMetrics { + memory_usage: prometheus::register_int_gauge_vec!( + "worker_memory_usage_bytes", + "Memory used by the worker's container, page cache excluded", + WORKER + ) + .unwrap(), + windmill_memory_usage: prometheus::register_int_gauge_vec!( + "worker_windmill_memory_usage_bytes", + "Resident memory of the worker process itself", + WORKER + ) + .unwrap(), + memory_limit: prometheus::register_int_gauge_vec!( + "worker_memory_limit_bytes", + "Memory limit of the worker's container", + WORKER + ) + .unwrap(), + cpu_usage: prometheus::register_counter_vec!( + "worker_cpu_usage_seconds_total", + "CPU time consumed by the worker's container", + WORKER + ) + .unwrap(), + cpu_periods: prometheus::register_int_counter_vec!( + "worker_cpu_periods_total", + "CPU enforcement periods elapsed for the worker's container", + WORKER + ) + .unwrap(), + cpu_throttled_periods: prometheus::register_int_counter_vec!( + "worker_cpu_throttled_periods_total", + "CPU enforcement periods in which the worker's container was throttled", + WORKER + ) + .unwrap(), + cpu_throttled_seconds: prometheus::register_counter_vec!( + "worker_cpu_throttled_seconds_total", + "Time the worker's container spent throttled by its CPU limit", + WORKER + ) + .unwrap(), + occupancy_rate: prometheus::register_gauge_vec!( + "worker_occupancy_rate", + "Share of time the worker spent executing jobs, over the window", + &["name", "worker_group", "window"] + ) + .unwrap(), + job_memory_peak: prometheus::register_histogram_vec!( + "worker_job_memory_peak_bytes", + "Peak memory of a job's main process", + &["tag"], + windmill_common::worker::JOB_MEMORY_PEAK_BUCKETS.to_vec() + ) + .unwrap(), + }) + } else { + None + }; +} + +fn metrics_enabled() -> bool { + #[cfg(feature = "prometheus")] + if PROM_METRICS.is_some() { + return true; + } + OTEL_METRICS_ENABLED.load(Ordering::Relaxed) +} + +// Memory and CPU are read from the cgroup, which every worker of a process shares +// (native mode runs several). Only the first worker to report exports them, so that +// summing the series over workers does not count one container several times. +static CONTAINER_SERIES_OWNER: OnceLock = OnceLock::new(); +static MEMORY_LIMIT: Mutex>> = Mutex::new(None); + +/// Replaces the exported memory limit, for the ping that re-reads the cgroup limits. +pub(crate) fn set_memory_limit(limit: Option) { + *MEMORY_LIMIT.lock().unwrap() = Some(limit); +} +static LAST_CPU_STAT: Mutex = + Mutex::new(CpuStat { usage_usec: 0, nr_periods: 0, nr_throttled: 0, throttled_usec: 0 }); + +/// Exports the worker's resource readings. Meant for the ping ticks, which have already +/// read memory usage and occupancy; the only extra work is the cgroup `cpu.stat` read. +pub(crate) fn record_worker_resources( + worker_name: &str, + memory_usage: Option, + wm_memory_usage: Option, + occupancy: Option<&OccupancyResult>, +) { + if !metrics_enabled() { + return; + } + let worker_group = WORKER_GROUP.as_str(); + + if let Some(occupancy) = occupancy { + let windows = [ + ("total", Some(occupancy.occupancy_rate)), + ("15s", occupancy.occupancy_rate_15s), + ("5m", occupancy.occupancy_rate_5m), + ("30m", occupancy.occupancy_rate_30m), + ]; + for (window, rate) in windows { + let Some(rate) = rate.filter(|r| r.is_finite()) else { + continue; + }; + #[cfg(feature = "prometheus")] + if let Some(m) = PROM_METRICS.as_ref() { + m.occupancy_rate + .with_label_values(&[worker_name, worker_group, window]) + .set(rate as f64); + } + otel_set_worker_occupancy_rate(worker_name, worker_group, window, rate as f64); + } + } + + if CONTAINER_SERIES_OWNER.get_or_init(|| worker_name.to_string()) != worker_name { + return; + } + + let memory_limit = *MEMORY_LIMIT.lock().unwrap().get_or_insert_with(get_memory); + #[cfg(feature = "prometheus")] + if let Some(m) = PROM_METRICS.as_ref() { + let labels = &[worker_name, worker_group]; + if let Some(v) = memory_usage { + m.memory_usage.with_label_values(labels).set(v); + } + if let Some(v) = wm_memory_usage { + m.windmill_memory_usage.with_label_values(labels).set(v); + } + if let Some(v) = memory_limit { + m.memory_limit.with_label_values(labels).set(v); + } + } + otel_set_worker_memory( + worker_name, + worker_group, + memory_usage, + wm_memory_usage, + memory_limit, + ); + + let Some(stat) = get_cpu_stat() else { + return; + }; + let delta = { + let mut last = LAST_CPU_STAT.lock().unwrap(); + // A reading below the previous one means the cgroup changed under the process: + // restart from it rather than exporting a negative increase. + let delta = if stat.usage_usec < last.usage_usec { + CpuStat::default() + } else { + CpuStat { + usage_usec: stat.usage_usec - last.usage_usec, + nr_periods: stat.nr_periods.saturating_sub(last.nr_periods), + nr_throttled: stat.nr_throttled.saturating_sub(last.nr_throttled), + throttled_usec: stat.throttled_usec.saturating_sub(last.throttled_usec), + } + }; + *last = stat; + delta + }; + let usage_secs = delta.usage_usec as f64 / 1e6; + let throttled_secs = delta.throttled_usec as f64 / 1e6; + #[cfg(feature = "prometheus")] + if let Some(m) = PROM_METRICS.as_ref() { + let labels = &[worker_name, worker_group]; + m.cpu_usage.with_label_values(labels).inc_by(usage_secs); + m.cpu_periods + .with_label_values(labels) + .inc_by(delta.nr_periods); + m.cpu_throttled_periods + .with_label_values(labels) + .inc_by(delta.nr_throttled); + m.cpu_throttled_seconds + .with_label_values(labels) + .inc_by(throttled_secs); + } + otel_add_worker_cpu( + worker_name, + worker_group, + usage_secs, + delta.nr_periods, + delta.nr_throttled, + throttled_secs, + ); +} + +/// `mem_peak_kb` is the job's `mem_peak`: non-positive when the executor had no process +/// to sample or the reading failed, and those are not observations. +pub(crate) fn record_job_memory_peak(tag: &str, mem_peak_kb: i32) { + if mem_peak_kb <= 0 { + return; + } + let bytes = mem_peak_kb as u64 * 1024; + #[cfg(feature = "prometheus")] + if let Some(m) = PROM_METRICS.as_ref() { + m.job_memory_peak + .with_label_values(&[tag]) + .observe(bytes as f64); + } + otel_record_job_memory_peak(tag, bytes); +} diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index d0d5d8b4e2..5b5df51b30 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -5130,6 +5130,7 @@ pub async fn handle_queued_job( // WAC v2 job suspended while waiting for child jobs — don't complete it return Ok(JobOutcome::Completed); } + crate::resource_metrics::record_job_memory_peak(&cjob.tag, mem_peak); process_result( cjob, result.map(|x| Arc::new(x)), diff --git a/backend/windmill-worker/src/worker_utils.rs b/backend/windmill-worker/src/worker_utils.rs index cabf004f4d..c634c13b6a 100644 --- a/backend/windmill-worker/src/worker_utils.rs +++ b/backend/windmill-worker/src/worker_utils.rs @@ -39,17 +39,26 @@ pub(crate) async fn update_worker_ping_full( let wm_memory_usage = get_windmill_memory_usage(); let (vcpus, memory) = if read_cgroups { - (get_vcpus(), get_memory()) + let memory = get_memory(); + crate::resource_metrics::set_memory_limit(memory); + (get_vcpus(), memory) } else { (None, None) }; + let occupancy = occupancy_metrics.update_occupancy_metrics(); + crate::resource_metrics::record_worker_resources( + worker_name, + memory_usage, + wm_memory_usage, + Some(&occupancy), + ); let OccupancyResult { occupancy_rate, occupancy_rate_15s, occupancy_rate_5m, occupancy_rate_30m, - } = occupancy_metrics.update_occupancy_metrics(); + } = occupancy; let ping_start = std::time::Instant::now(); if let Err(e) = (|| { From 987c321ebcd305df79fa945e941abc2b2e1d3e95 Mon Sep 17 00:00:00 2001 From: Alexander Petric Date: Thu, 8 Oct 2026 16:32:55 -0400 Subject: [PATCH 05/38] fix: uniform decrypt error and clear stored credentials on logout (#11632) * fix: uniform decrypt error and clear stored credentials on logout Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01Kv72vdCDggCZEjJSmNCwnX * fix: also clear mcp-oauth-callback on logout, cover the new keys in the logout test Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01Kv72vdCDggCZEjJSmNCwnX --------- Co-authored-by: Claude Opus 5.5 (1M context) --- backend/windmill-common/src/variables.rs | 42 +++++++++++++++++++----- frontend/src/lib/logout.test.ts | 16 +++++++++ frontend/src/lib/logout.ts | 16 +++++---- 3 files changed, 60 insertions(+), 14 deletions(-) diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index 1feb4182ab..b3774cdda4 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -14,7 +14,7 @@ use crate::utils::WarnAfterExt; use crate::worker::Connection; use crate::{worker::WORKER_GROUP, BASE_URL, DB}; use chrono::{SecondsFormat, Utc}; -use magic_crypt::{MagicCrypt256, MagicCryptError, MagicCryptTrait}; +use magic_crypt::{MagicCrypt256, MagicCryptTrait}; use quick_cache::sync::Cache; use serde::{Deserialize, Serialize}; @@ -375,13 +375,15 @@ pub fn encrypt(mc: &MagicCrypt256, value: &str) -> String { mc.encrypt_str_to_base64(value) } +const DECRYPT_FAILED: &str = + "Could not decrypt value. The value may have been encrypted with a different key."; + +/// Every failure returns the same message so responses carry no signal about why decryption +/// failed (padding, base64, UTF-8); the cause is only logged. pub fn decrypt(mc: &MagicCrypt256, value: String) -> error::Result { - mc.decrypt_base64_to_string(value).map_err(|e| match e { - MagicCryptError::DecryptError(_) => error::Error::internal_err( - "Could not decrypt value. The value may have been encrypted with a different key." - .to_string(), - ), - _ => error::Error::internal_err(e.to_string()), + mc.decrypt_base64_to_string(value).map_err(|e| { + tracing::warn!("decryption failed: {e}"); + error::Error::internal_err(DECRYPT_FAILED.to_string()) }) } @@ -746,7 +748,31 @@ pub async fn get_variable_or_self_as( #[cfg(test)] mod tests { - use super::{can_bind_as_prologue_const, escape_js_single_quoted, is_valid_js_identifier}; + use super::{ + can_bind_as_prologue_const, decrypt, escape_js_single_quoted, is_valid_js_identifier, + DECRYPT_FAILED, + }; + use crate::error::Error; + use magic_crypt::MagicCryptTrait; + + #[test] + fn decrypt_failures_all_return_the_same_message() { + let mc = magic_crypt::new_magic_crypt!("workspace-key", 256); + let other = magic_crypt::new_magic_crypt!("another-key", 256); + let failures = [ + "not base64 at all!".to_string(), + mc.encrypt_bytes_to_base64(&[0xff, 0xfe, 0xfd]), + other.encrypt_str_to_base64("secret"), + ]; + for value in failures { + match decrypt(&mc, value.clone()) { + Err(Error::InternalErrLoc { message, .. }) => { + assert_eq!(message, DECRYPT_FAILED, "for {value:?}") + } + other => panic!("expected the uniform decrypt error for {value:?}, got {other:?}"), + } + } + } #[test] fn is_valid_js_identifier_gates_prologue_injection() { diff --git a/frontend/src/lib/logout.test.ts b/frontend/src/lib/logout.test.ts index 160bce9934..5bdc15bac3 100644 --- a/frontend/src/lib/logout.test.ts +++ b/frontend/src/lib/logout.test.ts @@ -12,9 +12,25 @@ import { clearUser } from './logout' describe('clearUser', () => { beforeEach(() => { sessionStorage.clear() + localStorage.clear() vi.clearAllMocks() }) + it('drops credential-bearing localStorage entries and keeps preferences', async () => { + const credentialKeys = ['oauth-callback', 'mcp-oauth-callback', 'test_dev_token'] + for (const key of credentialKeys) { + localStorage.setItem(key, 'a-token') + } + localStorage.setItem('dark-mode', 'true') + + await clearUser() + + for (const key of credentialKeys) { + expect(localStorage.getItem(key)).toBeNull() + } + expect(localStorage.getItem('dark-mode')).toBe('true') + }) + // Logging out while impersonating must not leave the impersonator's own token behind. it('drops the impersonation token backup', async () => { sessionStorage.setItem('pre_impersonation_token', 'a-token') diff --git a/frontend/src/lib/logout.ts b/frontend/src/lib/logout.ts index 58fce16fe3..b4782badd3 100644 --- a/frontend/src/lib/logout.ts +++ b/frontend/src/lib/logout.ts @@ -6,15 +6,19 @@ import { clearStores } from './storeUtils' // Note: logout and logoutWithRedirect have been moved to logoutKit.ts // as they depend on SvelteKit navigation -// The impersonation backup holds the impersonator's own auth token, so it must go on -// logout too, not only when impersonation is ended from the banner — otherwise a usable -// token stays readable for the lifetime of the tab. -function clearImpersonationFromStorage(): void { +// Every browser-storage entry that can hold a credential must go on logout. The impersonation +// backup is the impersonator's own auth token; `oauth-callback` and `mcp-oauth-callback` carry a +// connected account's token response when the OAuth popup had no opener to post it to; +// `test_dev_token` is the session token of the /test_dev pages. +function clearCredentialsFromStorage(): void { try { sessionStorage.removeItem('pre_impersonation_token') sessionStorage.removeItem('pre_impersonation_email') + localStorage.removeItem('oauth-callback') + localStorage.removeItem('mcp-oauth-callback') + localStorage.removeItem('test_dev_token') } catch (e) { - console.error('error interacting with session storage', e) + console.error('error interacting with browser storage', e) } } @@ -23,7 +27,7 @@ export async function clearUser() { noteSessionEmail(undefined) accountSetup.reset() clearStores() - clearImpersonationFromStorage() + clearCredentialsFromStorage() await UserService.logout() } catch (error) {} } From 6acf1858540e83b65b5218c50c98af692092d22e Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 8 Oct 2026 22:35:07 +0200 Subject: [PATCH 06/38] feat: add an api base url instance setting for displayed urls (#11628) * feat: add an api base url instance setting for displayed urls * feat: show the api base url as cli remote, keep mcp url on origin * fix: keep push captures on origin and load the api base url on svix page * fix: reject shell metacharacters in webhook and api base urls * test: drop the trivial api base url resolver test * feat: hide the api base url setting behind a toggle * fix: let the quick setup toggle reveal the api base url field * chore: update ee-repo-ref to 017710bdd0a10de857f0f40800dde238cdb1ad24 This commit updates the EE repository reference after PR #863 was merged in windmill-ee-private. Previous ee-repo-ref: f67cd621bd45840e6389e0469a67378864f269a0 New ee-repo-ref: 017710bdd0a10de857f0f40800dde238cdb1ad24 Automated by sync-ee-ref workflow. --------- Co-authored-by: windmill-internal-app[bot] --- backend/ee-repo-ref.txt | 2 +- backend/windmill-api-settings/src/lib.rs | 34 +++++------ backend/windmill-api/openapi.yaml | 5 +- .../windmill-common/src/global_settings.rs | 41 ++++++++++--- .../windmill-common/src/instance_config.rs | 58 +++++++++++-------- frontend/src/lib/apiBaseUrl.svelte.ts | 22 +++++++ frontend/src/lib/components/CliHelpBox.svelte | 3 +- frontend/src/lib/components/JobArgs.svelte | 3 +- .../src/lib/components/WorkerGroup.svelte | 3 +- .../components/git_sync/GitPushPreview.svelte | 3 +- .../components/home/HomeConnectDrawer.svelte | 15 +++-- .../src/lib/components/instanceSettings.ts | 31 ++++++++-- .../instanceSettingsWebhookBaseUrl.test.ts | 16 +++-- .../components/settings/CreateToken.svelte | 2 + .../azure/AzureTriggerEditorInner.svelte | 7 ++- .../gcp/GcpTriggerEditorConfigSection.svelte | 22 +------ .../triggers/gcp/GcpTriggerEditorInner.svelte | 7 ++- .../triggers/http/OpenAPISpecGenerator.svelte | 8 +-- .../src/lib/components/triggers/http/utils.ts | 10 ++-- .../webhook/WebhooksConfigSection.svelte | 10 ++-- frontend/src/lib/instanceUi.ts | 3 +- frontend/src/lib/utils_deployable.ts | 8 +-- frontend/src/lib/utils_workspace_deploy.ts | 2 +- .../svix/create-webhook/+page@(root).svelte | 8 ++- 24 files changed, 205 insertions(+), 118 deletions(-) create mode 100644 frontend/src/lib/apiBaseUrl.svelte.ts diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index d292c22c2e..cc3c1a826c 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -e128a9b9bedd0b6a46d355b69db13bf8f650c4e6 +017710bdd0a10de857f0f40800dde238cdb1ad24 diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index c79b596617..f9f0a993fa 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -58,10 +58,10 @@ use windmill_common::{ error::{self, pg_error_message, JsonResult, Result}, get_database_url, global_settings::{ - ACCENT_COLOR_SETTING, AI_CONFIG_SETTING, APP_WORKSPACED_ROUTE_SETTING, - AUTOMATE_USERNAME_CREATION_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, CUSTOM_TAGS_SETTING, - DEFAULT_TAGS_WORKSPACES_SETTING, DISABLE_HUB_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS, - EXTERNAL_INSTANCE_PG_SETTING, + ACCENT_COLOR_SETTING, AI_CONFIG_SETTING, API_BASE_URL_SETTING, + APP_WORKSPACED_ROUTE_SETTING, AUTOMATE_USERNAME_CREATION_SETTING, + CRITICAL_ALERT_MUTE_UI_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING, + DISABLE_HUB_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS, EXTERNAL_INSTANCE_PG_SETTING, GITHUB_APP_WEBHOOK_BASE_URL_SETTING, HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS_SETTING, HTTP_ROUTE_WORKSPACED_ROUTE_SETTING, HUB_ACCESSIBLE_URL_SETTING, HUB_BASE_URL_SETTING, INSTANCE_BANNER_SETTING, MAX_RETENTION_OVERRIDE_WORKSPACES, @@ -1238,26 +1238,22 @@ async fn run_setting_pre_write_hook( } } } - GITHUB_APP_WEBHOOK_BASE_URL_SETTING => { - // A bad value here yields a webhook GitHub can never deliver to, and the + GITHUB_APP_WEBHOOK_BASE_URL_SETTING | API_BASE_URL_SETTING => { + // A bad webhook base yields a webhook GitHub can never deliver to, and the // failure only shows up much later as "falling back to polling" on a - // repository — so reject it at the boundary instead. + // repository; a bad API base puts an unreachable URL in front of every user. + // So reject both at the boundary instead. match value { // Clearing (delete row) is handled by the caller; allow it through. serde_json::Value::Null => {} serde_json::Value::String(s) if s.trim().is_empty() => {} serde_json::Value::String(s) => { - windmill_common::global_settings::validate_webhook_base_url(s).map_err( - |e| { - error::Error::BadRequest(format!( - "{GITHUB_APP_WEBHOOK_BASE_URL_SETTING}: {e}" - )) - }, - )?; + windmill_common::global_settings::validate_webhook_base_url(s) + .map_err(|e| error::Error::BadRequest(format!("{key}: {e}")))?; } _ => { return Err(error::Error::BadRequest(format!( - "{GITHUB_APP_WEBHOOK_BASE_URL_SETTING} must be a URL string" + "{key} must be a URL string" ))); } } @@ -1394,8 +1390,12 @@ async fn set_instance_config( // Applied and audited one setting at a time: the writes are not transactional, so a // batch that fails midway must not leave the settings it did write unrecorded. - let instance_config::SettingsDiff { upserts, deletes, mut previous_values, unchanged_count } = - settings_diff; + let instance_config::SettingsDiff { + upserts, + deletes, + mut previous_values, + unchanged_count, + } = settings_diff; let steps = upserts .into_iter() .map(|(k, v)| (k, Some(v))) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 5e635e39bd..0d21402a78 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -1919,7 +1919,7 @@ paths: /settings/instance_ui: get: - summary: get the instance-wide announcement banner and accent color + summary: get the instance-wide announcement banner, accent color and API base url description: | Readable by any authenticated user; every page load reads it. Each value is the raw stored setting, or null when unset. @@ -1928,7 +1928,7 @@ paths: - setting responses: "200": - description: the instance_banner and accent_color settings + description: the instance_banner, accent_color and api_base_url settings content: application/json: schema: @@ -1936,6 +1936,7 @@ paths: properties: instance_banner: {} accent_color: {} + api_base_url: {} /settings_u/ruff_config: get: diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index e1a731f3e6..1cd3609820 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -181,6 +181,12 @@ pub const GITHUB_ENTERPRISE_APP_SETTING: &str = "github_enterprise_app"; /// `base_url` when unset; set it when the browser-facing URL is not reachable /// from GitHub and a separate ingress fronts the API for inbound webhooks. pub const GITHUB_APP_WEBHOOK_BASE_URL_SETTING: &str = "github_app_webhook_base_url"; +/// Base URL the UI shows for the endpoints external clients call (webhooks, HTTP +/// routes, push trigger endpoints, the CLI), instead of the origin the UI is browsed on. +/// For deployments that serve the API on a separate domain or behind a gateway. +/// EE only. Readable by any authenticated user, like [`INSTANCE_BANNER_SETTING`]: +/// every user is shown URLs built from it. +pub const API_BASE_URL_SETTING: &str = "api_base_url"; /// Instance-wide announcement rendered above every page of the app (maintenance /// windows, incidents). Readable by any authenticated user, unlike most settings: /// the banner exists to be shown to everyone, so it must never hold anything the @@ -313,12 +319,17 @@ pub fn validate_accent_color(value: &serde_json::Value) -> Result<(), String> { pub struct InstanceUi { pub instance_banner: Option, pub accent_color: Option, + pub api_base_url: Option, } pub async fn get_instance_ui(db: &Pool) -> error::Result { let rows = sqlx::query!( "SELECT name, value FROM global_settings WHERE name = ANY($1)", - &[INSTANCE_BANNER_SETTING, ACCENT_COLOR_SETTING] as &[&str] + &[ + INSTANCE_BANNER_SETTING, + ACCENT_COLOR_SETTING, + API_BASE_URL_SETTING + ] as &[&str] ) .fetch_all(db) .await?; @@ -327,17 +338,22 @@ pub async fn get_instance_ui(db: &Pool) -> error::Result { match row.name.as_str() { INSTANCE_BANNER_SETTING => ui.instance_banner = Some(row.value), ACCENT_COLOR_SETTING => ui.accent_color = Some(row.value), + API_BASE_URL_SETTING => ui.api_base_url = Some(row.value), _ => {} } } Ok(ui) } -/// Validate a [`GITHUB_APP_WEBHOOK_BASE_URL_SETTING`] value. +const SHELL_ACTIVE_CHARS: [char; 16] = [ + '$', '`', '\'', '"', '\\', ';', '&', '|', '<', '>', '(', ')', '{', '}', '*', '!', +]; + +/// Validate a [`GITHUB_APP_WEBHOOK_BASE_URL_SETTING`] or [`API_BASE_URL_SETTING`] value. /// -/// The receiver path is appended to it verbatim, so anything that doesn't -/// concatenate into a URL GitHub can POST to must be rejected at write time -/// rather than silently producing an unreachable hook: a wrong scheme +/// A path is appended to it verbatim, so anything that doesn't concatenate into a +/// URL an external caller can reach must be rejected at write time rather than +/// silently producing an unreachable hook or endpoint: a wrong scheme /// (`httpss://`), a missing host, embedded whitespace, or a query/fragment /// (appending a path after `?`/`#` keeps it inside the query/fragment). /// @@ -353,7 +369,7 @@ pub fn validate_webhook_base_url(value: &str) -> Result<(), String> { url::Url::parse(value).map_err(|e| format!("must be an absolute http(s) URL: {e}"))?; if !url.username().is_empty() || url.password().is_some() { return Err( - "must not embed a username or password: the receiver URL is stored in workspace settings, where it is readable by workspace admins".to_string(), + "must not embed a username or password: the URL is readable by users who are not instance admins".to_string(), ); } if !matches!(url.scheme(), "http" | "https") { @@ -366,13 +382,21 @@ pub fn validate_webhook_base_url(value: &str) -> Result<(), String> { } if url.query().is_some() || url.fragment().is_some() { return Err( - "must not include a query string or fragment, since the webhook path is appended to it" + "must not include a query string or fragment, since a path is appended to it" .to_string(), ); } if value.chars().any(char::is_whitespace) { return Err("must not contain whitespace".to_string()); } + // The UI pastes the URL into shell commands users copy, where double quotes + // do not stop `$(...)` or backticks from running. + if value.contains(SHELL_ACTIVE_CHARS) { + return Err( + "must not contain shell metacharacters, since it is shown inside commands users copy" + .to_string(), + ); + } // Matches the sibling `base_url` / `hub_base_url` convention. The receiver // builder trims it anyway, so this is about keeping the stored value canonical // rather than about reachability. @@ -1101,6 +1125,9 @@ mod tests { "https://hooks.example.com#", "https://user:password@hooks.example.com", "https://user@hooks.example.com", + "https://hooks.example.com/$(uname)", + "https://hooks.example.com/`uname`", + "https://hooks.example.com/a;b", ]; for v in accept { assert!( diff --git a/backend/windmill-common/src/instance_config.rs b/backend/windmill-common/src/instance_config.rs index 700ba71df5..bd7c28628c 100644 --- a/backend/windmill-common/src/instance_config.rs +++ b/backend/windmill-common/src/instance_config.rs @@ -281,6 +281,8 @@ pub struct GlobalSettings { #[serde(skip_serializing_if = "Option::is_none")] pub github_app_webhook_base_url: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub api_base_url: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub email_domain: Option, #[serde(skip_serializing_if = "Option::is_none")] pub hub_base_url: Option, @@ -1375,8 +1377,8 @@ pub fn diff_worker_configs( } /// Declaratively replace the global settings, rejecting a -/// `github_app_webhook_base_url` or `http_route_default_allowed_origins` the -/// API would reject. +/// `github_app_webhook_base_url`, `api_base_url` or +/// `http_route_default_allowed_origins` the API would reject. /// /// Every declarative writer (the `sync-config` CLI, the Kubernetes operator's /// ConfigMap sync) MUST go through this rather than calling @@ -1399,29 +1401,35 @@ pub async fn sync_global_settings_declarative( current: &BTreeMap, desired: &BTreeMap, ) -> anyhow::Result<()> { - let webhook_key = crate::global_settings::GITHUB_APP_WEBHOOK_BASE_URL_SETTING; - // Non-string shapes are rejected rather than ignored: `as_str()` alone would let a - // bool/number/object through as if the key were absent, and the diff below would - // then persist it — where the HTTP path answers "must be a URL string". - match desired.get(webhook_key) { - None | Some(serde_json::Value::Null) => {} - Some(serde_json::Value::String(s)) if s.trim().is_empty() => {} - Some(serde_json::Value::String(s)) => crate::global_settings::validate_webhook_base_url(s) - .map_err(|e| anyhow::anyhow!("{webhook_key}: {e}"))?, - // Names the JSON kind rather than printing it: this is the last message on - // this path that could report submitted content, and an object or array could - // carry a secret into `sync-config` output and operator logs. - Some(other) => { - let kind = match other { - serde_json::Value::Bool(_) => "a boolean", - serde_json::Value::Number(_) => "a number", - serde_json::Value::Array(_) => "an array", - serde_json::Value::Object(_) => "an object", - _ => "a non-string value", - }; - return Err(anyhow::anyhow!( - "{webhook_key} must be a URL string, got {kind}" - )); + for url_key in [ + crate::global_settings::GITHUB_APP_WEBHOOK_BASE_URL_SETTING, + crate::global_settings::API_BASE_URL_SETTING, + ] { + // Non-string shapes are rejected rather than ignored: `as_str()` alone would let a + // bool/number/object through as if the key were absent, and the diff below would + // then persist it — where the HTTP path answers "must be a URL string". + match desired.get(url_key) { + None | Some(serde_json::Value::Null) => {} + Some(serde_json::Value::String(s)) if s.trim().is_empty() => {} + Some(serde_json::Value::String(s)) => { + crate::global_settings::validate_webhook_base_url(s) + .map_err(|e| anyhow::anyhow!("{url_key}: {e}"))? + } + // Names the JSON kind rather than printing it: this is the last message on + // this path that could report submitted content, and an object or array could + // carry a secret into `sync-config` output and operator logs. + Some(other) => { + let kind = match other { + serde_json::Value::Bool(_) => "a boolean", + serde_json::Value::Number(_) => "a number", + serde_json::Value::Array(_) => "an array", + serde_json::Value::Object(_) => "an object", + _ => "a non-string value", + }; + return Err(anyhow::anyhow!( + "{url_key} must be a URL string, got {kind}" + )); + } } } diff --git a/frontend/src/lib/apiBaseUrl.svelte.ts b/frontend/src/lib/apiBaseUrl.svelte.ts new file mode 100644 index 0000000000..f87a33f0d8 --- /dev/null +++ b/frontend/src/lib/apiBaseUrl.svelte.ts @@ -0,0 +1,22 @@ +import { fromStore } from 'svelte/store' +import { base } from '$lib/base' +import { instanceUi } from '$lib/instanceUi' + +const ui = fromStore(instanceUi) + +/** + * Base of every URL shown for an external client to call (webhooks, HTTP routes, push + * trigger endpoints, the CLI remote): the `api_base_url` instance setting, else the + * origin being browsed. Append `/api/...` to it. + * + * Not for URLs the editor itself calls or opens (captures, previews, OAuth popups, links + * to pages): those must stay on the browsing origin, where the session cookie lives. + * + * Reactive: the setting lands after page load, so read it inside a `$derived` or the + * template rather than caching the result. + */ +export function apiBaseUrl(): string { + const setting = ui.current?.api_base_url + const configured = typeof setting === 'string' ? setting.trim() : '' + return configured || `${window.location.origin}${base}` +} diff --git a/frontend/src/lib/components/CliHelpBox.svelte b/frontend/src/lib/components/CliHelpBox.svelte index 0bee33825c..e91b039328 100644 --- a/frontend/src/lib/components/CliHelpBox.svelte +++ b/frontend/src/lib/components/CliHelpBox.svelte @@ -2,10 +2,11 @@ import ClipboardPanel from './details/ClipboardPanel.svelte' import Section from './Section.svelte' import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte' + import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' const operatingWorkspace = useOperatingWorkspace() - let url = $derived(`${window.location.protocol}//${window.location.hostname}/`) + let url = $derived(`${apiBaseUrl()}/`)
diff --git a/frontend/src/lib/components/JobArgs.svelte b/frontend/src/lib/components/JobArgs.svelte index 64ca912608..e0712d66c7 100644 --- a/frontend/src/lib/components/JobArgs.svelte +++ b/frontend/src/lib/components/JobArgs.svelte @@ -15,6 +15,7 @@ import { isWindmillTooBigObject } from './job_args' import { downloadViaClient, shouldDownloadViaClient } from '$lib/utils/downloadFile' import { appendViewToken } from '$lib/viewToken' + import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' interface Props { id?: string | undefined @@ -246,7 +247,7 @@ ${Object.entries(displayArgs) If using the wmill client in your code, set the following env variables: 
BASE_URL="{window.location.origin}"
+				>BASE_URL="{apiBaseUrl()}"
 WM_TOKEN="{''}"
diff --git a/frontend/src/lib/components/WorkerGroup.svelte b/frontend/src/lib/components/WorkerGroup.svelte index 3b399a5a1e..5e8957c31a 100644 --- a/frontend/src/lib/components/WorkerGroup.svelte +++ b/frontend/src/lib/components/WorkerGroup.svelte @@ -55,6 +55,7 @@ import TagList from './TagList.svelte' import DedicatedWorkersSelector from './DedicatedWorkersSelector.svelte' import { computeHashedTag } from './dedicated_worker' + import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' import ObjectStoreConfigSettings, { type ObjectStoreConfig } from './ObjectStoreConfigSettings.svelte' @@ -1238,7 +1239,7 @@

Requires a superadmin token.

curl -X POST '{window.location.origin}/api/configs/update/worker__{name}' \
+					>curl -X POST '{apiBaseUrl()}/api/configs/update/worker__{name}' \
   -H 'Content-Type: application/json' \
   -H 'Authorization: Bearer <token>' \
   -d '{JSON.stringify(cleanWorkerGroupConfig(nconfig ?? {}), null, 2)}'
') const cliCommands = $derived(`npm install -g windmill-cli wmill workspace add ${workspaceId} ${workspaceId} ${origin} @@ -55,7 +56,8 @@ wmill sync pull`)

Local setup

Run this in your local repo to bind the current workspace, create - wmill.yaml, and pull the latest files.

@@ -80,15 +82,18 @@ wmill sync pull`) />

- wmill workspace add will handle authentication, - wmill init bootstraps the local config, and - wmill sync pull fetches the workspace content. diff --git a/frontend/src/lib/components/instanceSettings.ts b/frontend/src/lib/components/instanceSettings.ts index 9701b74e30..75dada92ab 100644 --- a/frontend/src/lib/components/instanceSettings.ts +++ b/frontend/src/lib/components/instanceSettings.ts @@ -175,8 +175,7 @@ export function isValidWebhookBaseUrl(value: unknown): boolean { return ( (url.protocol === 'http:' || url.protocol === 'https:') && url.host !== '' && - // Userinfo would end up in the per-repository receiver stored in workspace - // settings, which workspace admins can read. + // Userinfo would end up in URLs that users who are not instance admins can read. url.username === '' && url.password === '' && // Tested on the raw string, not `url.search`/`url.hash`: those are `''` for a @@ -187,6 +186,8 @@ export function isValidWebhookBaseUrl(value: unknown): boolean { // `new URL` silently percent-encodes a space in the path, where the server // rejects it outright. !/\s/.test(trimmed) && + // Shell-active characters: the URL is pasted into commands users copy. + !/[$`'"\\;&|<>(){}*!]/.test(trimmed) && !trimmed.endsWith('/') ) } @@ -207,7 +208,29 @@ export const settings: Record = { (value?.startsWith('http') && value.includes('://') && !value?.endsWith('/') && - !value?.endsWith(' ')) + !value?.endsWith(' ')), + advancedToggle: { + label: 'The API is served on a different url than the UI', + onChange(values) { + values['api_base_url'] = values['api_base_url'] == null ? '' : null + return values + }, + checked: (values) => values['api_base_url'] != null + } + }, + { + label: 'API base url', + description: + 'Base url shown to users for the endpoints external clients call (webhooks, HTTP routes, push trigger endpoints, the CLI remote), without trailing slash. Leave empty to show the url the UI is browsed on. Set it when the API is served on a separate domain or behind an API gateway. If that url does not also serve the UI, the CLI must authenticate with a token rather than the browser login.', + key: 'api_base_url', + fieldType: 'text', + placeholder: 'https://api.windmill.company.com', + storage: 'setting', + ee_only: '', + hiddenIfNull: true, + error: + 'API base url must be an http:// or https:// url with a host, no embedded username or password, no query string or fragment, no shell metacharacters, and no trailing slash', + isValid: isValidWebhookBaseUrl }, { label: 'Email domain', @@ -1223,7 +1246,7 @@ export const settings: Record = { storage: 'setting', ee_only: '', error: - 'Webhook base url must be an http:// or https:// url with a host, no embedded username or password, no query string or fragment, and no trailing slash', + 'Webhook base url must be an http:// or https:// url with a host, no embedded username or password, no query string or fragment, no shell metacharacters, and no trailing slash', isValid: isValidWebhookBaseUrl } ], diff --git a/frontend/src/lib/components/instanceSettingsWebhookBaseUrl.test.ts b/frontend/src/lib/components/instanceSettingsWebhookBaseUrl.test.ts index c8545b073e..175a251315 100644 --- a/frontend/src/lib/components/instanceSettingsWebhookBaseUrl.test.ts +++ b/frontend/src/lib/components/instanceSettingsWebhookBaseUrl.test.ts @@ -29,7 +29,10 @@ const REJECTED = [ 'https://hooks.example.com?', 'https://hooks.example.com#', 'https://user:password@hooks.example.com', - 'https://user@hooks.example.com' + 'https://user@hooks.example.com', + 'https://hooks.example.com/$(uname)', + 'https://hooks.example.com/`uname`', + 'https://hooks.example.com/a;b' ] describe('isValidWebhookBaseUrl', () => { @@ -41,10 +44,13 @@ describe('isValidWebhookBaseUrl', () => { expect(isValidWebhookBaseUrl(value)).toBe(false) }) - it.each([true, 42, {}, [], null] as unknown[])('rejects the non-string %j without throwing', (value) => { - // `Setting.isValid` is typed `any` and YAML mode can supply any JSON shape. - expect(isValidWebhookBaseUrl(value as never)).toBe(value === null) - }) + it.each([true, 42, {}, [], null] as unknown[])( + 'rejects the non-string %j without throwing', + (value) => { + // `Setting.isValid` is typed `any` and YAML mode can supply any JSON shape. + expect(isValidWebhookBaseUrl(value as never)).toBe(value === null) + } + ) it('treats unset and blank as valid, since the setting is optional', () => { expect(isValidWebhookBaseUrl(undefined)).toBe(true) diff --git a/frontend/src/lib/components/settings/CreateToken.svelte b/frontend/src/lib/components/settings/CreateToken.svelte index 8ee5391443..56952eb584 100644 --- a/frontend/src/lib/components/settings/CreateToken.svelte +++ b/frontend/src/lib/components/settings/CreateToken.svelte @@ -245,6 +245,8 @@ // Undefined wherever the workspace is: `/api/mcp/w/undefined/mcp` reads like a real URL // and is copyable, so the OAuth panel withholds it rather than showing a broken one. The // token branch guards the same case by disabling its generate button. + // On the browsing origin, not the API base url: the OAuth metadata the server + // publishes for this URL is built from `base_url`, and a client rejects a mismatch. const mcpUrl = $derived( isAllWorkspaces ? `${window.location.origin}/api/mcp/gateway` diff --git a/frontend/src/lib/components/triggers/azure/AzureTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/azure/AzureTriggerEditorInner.svelte index 4394a5878f..ed0049168e 100644 --- a/frontend/src/lib/components/triggers/azure/AzureTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/azure/AzureTriggerEditorInner.svelte @@ -35,6 +35,7 @@ import LocalDraftBanner from '$lib/components/LocalDraftBanner.svelte' import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte' import TriggerSuspendedJobsModal from '../TriggerSuspendedJobsModal.svelte' + import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' import { base } from '$lib/base' import Tabs from '$lib/components/common/tabs/Tabs.svelte' import Tab from '$lib/components/common/tabs/Tab.svelte' @@ -83,7 +84,7 @@ let permissionedAs = $state(undefined) let selectedPermissionedAs = $state(undefined) let preservePermissionedAs = $state(false) - let base_endpoint = $derived(`${window.location.origin}${base}`) + let base_endpoint = $derived(apiBaseUrl()) let optionTabSelected: 'error_handler' | 'retries' = $state('error_handler') let errorHandlerSelected: ErrorHandler = $state('slack') @@ -320,7 +321,9 @@ topic_name, subscription_name, event_type_filters, - base_endpoint, + // The capture URL shown to the user stays on the browsing origin, so the + // endpoint registered for it must too. + base_endpoint: `${window.location.origin}${base}`, path } } diff --git a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorConfigSection.svelte b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorConfigSection.svelte index d7868a650a..a33573a1d1 100644 --- a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorConfigSection.svelte +++ b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorConfigSection.svelte @@ -13,7 +13,7 @@ } from '$lib/gen' import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte' import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte' - import { base } from '$lib/base' + import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' import Toggle from '$lib/components/Toggle.svelte' import { Button, Url } from '$lib/components/common' @@ -28,23 +28,17 @@ useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte' - // Declared before `DEFAULT_PUSH_CONFIG` / the `base_endpoint` prop default, - // which call `getBaseUrl()` (a `wsId` reader) during component init. const operatingWorkspace = useOperatingWorkspace() const operatingUser = useOperatingUser() const actingUser = $derived(operatingUser.current) const wsId = $derived($operatingWorkspace) + const base_endpoint = $derived(`${apiBaseUrl()}/api/gcp/w/${wsId!}`) let topic_items: string[] = $state([]) let subscription_items: string[] = $state([]) let loadingTopic = $state(false) let loadingSubscription = $state(false) - const DEFAULT_PUSH_CONFIG: PushConfig = { - audience: getBaseUrl(), - authenticate: false - } - async function loadAllPubSubTopicsFromProject() { // Listing is admin-only under application default credentials, and a non-admin viewing an // inherited trigger cannot change the topic anyway, so asking would only raise a 403 toast. @@ -117,7 +111,6 @@ delivery_type?: DeliveryType | undefined delivery_config: PushConfig | undefined subscription_mode?: SubscriptionMode - base_endpoint?: string path?: string showTestingBadge?: boolean cloud_subscription_id?: string @@ -139,7 +132,6 @@ delivery_type = $bindable('pull'), delivery_config = $bindable(), subscription_mode = $bindable('create_update'), - base_endpoint = $bindable(getBaseUrl()), auto_acknowledge_msg = $bindable(true), ack_deadline = $bindable(), path = '', @@ -199,17 +191,9 @@ if (!delivery_type) { delivery_type = 'pull' } else if (delivery_type === 'push' && !delivery_config) { - delivery_config = DEFAULT_PUSH_CONFIG + delivery_config = { audience: base_endpoint, authenticate: false } } }) - function getBaseUrl() { - return `${window.location.origin}${base}/api/gcp/w/${wsId!}` - } - - $effect(() => { - !base_endpoint && (base_endpoint = getBaseUrl()) - }) - $effect(() => { if (emptyStringTrimmed(subscription_id) && !emptyStringTrimmed(path)) { subscription_id = `windmill-${wsId!}-${path.replaceAll('/', '_')}` diff --git a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte index cfee255553..49ba66448c 100644 --- a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte @@ -36,6 +36,7 @@ import LocalDraftBanner from '$lib/components/LocalDraftBanner.svelte' import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte' import TriggerSuspendedJobsModal from '../TriggerSuspendedJobsModal.svelte' + import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' import { base } from '$lib/base' import Tabs from '$lib/components/common/tabs/Tabs.svelte' import Tab from '$lib/components/common/tabs/Tab.svelte' @@ -88,7 +89,7 @@ let permissionedAs = $state(undefined) let selectedPermissionedAs = $state(undefined) let preservePermissionedAs = $state(false) - let base_endpoint = $derived(`${window.location.origin}${base}`) + let base_endpoint = $derived(apiBaseUrl()) let auto_acknowledge_msg = $state(true) let ack_deadline: number | undefined = $state() let optionTabSelected: 'settings' | 'error_handler' | 'retries' = $state('error_handler') @@ -376,7 +377,9 @@ subscription_id, delivery_type, delivery_config, - base_endpoint, + // The capture URL shown to the user stays on the browsing origin, so the + // endpoint registered for it must too. + base_endpoint: `${window.location.origin}${base}`, auto_acknowledge_msg, ack_deadline, topic_id, diff --git a/frontend/src/lib/components/triggers/http/OpenAPISpecGenerator.svelte b/frontend/src/lib/components/triggers/http/OpenAPISpecGenerator.svelte index b0cc491393..024086552c 100644 --- a/frontend/src/lib/components/triggers/http/OpenAPISpecGenerator.svelte +++ b/frontend/src/lib/components/triggers/http/OpenAPISpecGenerator.svelte @@ -15,7 +15,7 @@ import SimpleEditor from '$lib/components/SimpleEditor.svelte' import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte' import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte' - import { base } from '$lib/base' + import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' import Label from '$lib/components/Label.svelte' import Tooltip from '$lib/components/Tooltip.svelte' import { ClipboardCopy, Download, Trash } from 'lucide-svelte' @@ -136,7 +136,7 @@ obj = { openapi_spec_format: lang, info, - url: `${window.location.origin}${base}`, + url: apiBaseUrl(), http_route_filters: getHttpRouteFilters(), webhook_filters: getWebhookFilters() } @@ -155,7 +155,7 @@ requestBody: { openapi_spec_format, info, - url: `${window.location.origin}${base}`, + url: apiBaseUrl(), http_route_filters: getHttpRouteFilters(), webhook_filters: getWebhookFilters() } @@ -225,7 +225,7 @@ {/snippet} MAX_ALLOWED_ORIGIN_LEN) return `'${origin.slice(0, 40)}…' is longer than any origin a browser sends` - if (origin.includes(',')) - return `'${origin}' must not contain a comma, which separates entries` + if (origin.includes(',')) return `'${origin}' must not contain a comma, which separates entries` if (origin.toLowerCase() === 'null') return `'null' is what a sandboxed iframe sends, so it would allow any page that can open one` if (!/^[\x21-\x7e]+$/.test(origin)) @@ -85,8 +84,7 @@ export function allowedOriginWarning(origin: string): string | undefined { if (rest.startsWith(':')) return `'${origin}' has no host` // An unclosed bracket would otherwise leave `portStart` at zero, which reads // as "no port" and lets the entry through unremarked. - if (rest.startsWith('[') && !rest.includes(']')) - return `'${origin}' has an unclosed IPv6 host` + if (rest.startsWith('[') && !rest.includes(']')) return `'${origin}' has an unclosed IPv6 host` const portStart = rest.startsWith('[') ? rest.indexOf(']') + 1 : rest.indexOf(':') // A trailing colon is a port, an empty one — distinct from having none. const port = portStart > 0 && rest[portStart] === ':' ? rest.slice(portStart + 1) : undefined @@ -160,7 +158,7 @@ export function getHttpRoute( workspaced_route: boolean, workspace_id: string ) { - return `${location.origin}${base}/api/${route_prefix}/${ + return `${apiBaseUrl()}/api/${route_prefix}/${ isCloudHosted() || workspaced_route ? workspace_id + '/' : '' }${route_path ?? ''}` } diff --git a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte index 9033aab3cf..218430316a 100644 --- a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte +++ b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte @@ -16,7 +16,7 @@ import ClipboardPanel from '../../details/ClipboardPanel.svelte' import { copyToClipboard, isObject, readFieldsRecursively } from '$lib/utils' // import { page } from '$app/state' - import { base } from '$lib/base' + import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' import TriggerTokens from '../TriggerTokens.svelte' import { userStore } from '$lib/stores' import UserSettings from '../../UserSettings.svelte' @@ -46,7 +46,7 @@ const operatingWorkspace = useOperatingWorkspace() const wsId = $derived($operatingWorkspace) - const WEBHOOK_BASE_URL = $derived(`${location.origin}${base}/api/w/${wsId}/jobs`) + const WEBHOOK_BASE_URL = $derived(`${apiBaseUrl()}/api/w/${wsId}/jobs`) let baseWebhookUrl = $derived.by(() => { let webhookUrlPath: string @@ -223,9 +223,7 @@ async function triggerJob() { function waitForJobCompletion(UUID) { return new Promise(async (resolve, reject) => { try { - const endpoint = \`${ - location.origin - }/api/w/${wsId}/jobs_u/completed/get_result_maybe/\${UUID}\`; + const endpoint = \`${apiBaseUrl()}/api/w/${wsId}/jobs_u/completed/get_result_maybe/\${UUID}\`; const checkResponse = await fetch(endpoint, { method: 'GET', headers: ${JSON.stringify(headers(), null, 2).replaceAll('\n', '\n\t\t\t\t')} @@ -268,7 +266,7 @@ ${ ? 'echo -E $RESULT | jq' : requestType === 'async' ? ` -URL="${location.origin}/api/w/${wsId}/jobs_u/completed/get_result_maybe/$UUID" +URL="${apiBaseUrl()}/api/w/${wsId}/jobs_u/completed/get_result_maybe/$UUID" while true; do curl -s -H "Authorization: Bearer $TOKEN" $URL -o res.json COMPLETED=$(cat res.json | jq .completed) diff --git a/frontend/src/lib/instanceUi.ts b/frontend/src/lib/instanceUi.ts index b9b87d139c..3a7bb1b99c 100644 --- a/frontend/src/lib/instanceUi.ts +++ b/frontend/src/lib/instanceUi.ts @@ -2,7 +2,8 @@ import { writable } from 'svelte/store' import type { GetInstanceUiResponse } from '$lib/gen' /** - * The last `/settings/instance_ui` response: the announcement banner and accent color. + * The last `/settings/instance_ui` response: the announcement banner, accent color + * and API base url. * `undefined` until one lands, and left as is when a fetch fails, so a transient error * never retracts an announcement still in force. Written only by `InstanceUiSync`. */ diff --git a/frontend/src/lib/utils_deployable.ts b/frontend/src/lib/utils_deployable.ts index 63ea88adcb..e3f7fa83ed 100644 --- a/frontend/src/lib/utils_deployable.ts +++ b/frontend/src/lib/utils_deployable.ts @@ -16,7 +16,7 @@ import { type WorkspaceDeployUISettings } from './gen' import type { TriggerKind } from './components/triggers' -import { base } from './base' +import { apiBaseUrl } from './apiBaseUrl.svelte' type DeployUIType = 'script' | 'flow' | 'app' | 'resource' | 'variable' | 'secret' | 'trigger' @@ -250,9 +250,7 @@ export async function getTriggersDeployData( ...gcpTrigger, delivery_config: gcpTrigger.delivery_config ?? undefined, base_endpoint: - gcpTrigger.delivery_type === 'push' - ? (targetBaseUrl ?? `${window.location.origin}${base}`) - : undefined, + gcpTrigger.delivery_type === 'push' ? (targetBaseUrl ?? apiBaseUrl()) : undefined, permissioned_as: onBehalfOf, preserve_permissioned_as: preservePermissionedAs } @@ -320,7 +318,7 @@ export async function getTriggersDeployData( base_endpoint: azureTrigger.azure_mode === 'namespace_pull' ? undefined - : (targetBaseUrl ?? `${window.location.origin}${base}`), + : (targetBaseUrl ?? apiBaseUrl()), permissioned_as: onBehalfOf, preserve_permissioned_as: preservePermissionedAs }, diff --git a/frontend/src/lib/utils_workspace_deploy.ts b/frontend/src/lib/utils_workspace_deploy.ts index 280d0daa00..fa50805e3a 100644 --- a/frontend/src/lib/utils_workspace_deploy.ts +++ b/frontend/src/lib/utils_workspace_deploy.ts @@ -314,7 +314,7 @@ function makeProvider( existsTriggerByKind: (kind, p) => triggerServiceFor(kind).exists(p), getTriggerForDeploy: async (kind, p) => { // Reuses the existing per-kind transform map (e.g. GCP wipes - // subscription_id and computes base_endpoint from window.location). + // subscription_id and computes base_endpoint for this instance). // Operational-state strip is applied by the shared `deployItem` // after this returns. const { data } = await getTriggersDeployData( diff --git a/frontend/src/routes/(root)/(logged)/svix/create-webhook/+page@(root).svelte b/frontend/src/routes/(root)/(logged)/svix/create-webhook/+page@(root).svelte index d4ac558cb8..212696782b 100644 --- a/frontend/src/routes/(root)/(logged)/svix/create-webhook/+page@(root).svelte +++ b/frontend/src/routes/(root)/(logged)/svix/create-webhook/+page@(root).svelte @@ -4,6 +4,8 @@ import ScriptPicker from '$lib/components/ScriptPicker.svelte' import { Button } from '$lib/components/common' import { page } from '$app/stores' + import { apiBaseUrl } from '$lib/apiBaseUrl.svelte' + import InstanceUiSync from '$lib/components/InstanceUiSync.svelte' import { userStore, workspaceStore } from '$lib/stores' import { sendUserToast } from '$lib/toast' import UserSettings from '$lib/components/UserSettings.svelte' @@ -26,11 +28,11 @@ let webhook: string switch (kind) { case 'script': - webhook = `${$page.url.origin}/api/w/${$workspaceStore}/jobs/run/p/${path}` + webhook = `${apiBaseUrl()}/api/w/${$workspaceStore}/jobs/run/p/${path}` redirectSuccess(webhook, token) break case 'flow': - webhook = `${$page.url.origin}/api/w/${$workspaceStore}/jobs/run/f/${path}` + webhook = `${apiBaseUrl()}/api/w/${$workspaceStore}/jobs/run/f/${path}` redirectSuccess(webhook, token) break default: @@ -72,6 +74,8 @@ }) + + { From 0a95ea47facaede2545cc4741d4a70c1c847e11a Mon Sep 17 00:00:00 2001 From: AlexRV12 <71396855+AlexRV12@users.noreply.github.com> Date: Thu, 8 Oct 2026 23:23:31 +0200 Subject: [PATCH 07/38] fix: keep raw app deploy panel settings shared with the editor (#11616) * fix: keep raw app deploy panel settings shared with the editor Co-Authored-By: Claude Opus 5.5 * fix: take the first deploy's policy copy after bundling Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- .../raw_apps/RawAppEditorHeader.svelte | 29 ++++++++++++------- 1 file changed, 18 insertions(+), 11 deletions(-) diff --git a/frontend/src/lib/components/raw_apps/RawAppEditorHeader.svelte b/frontend/src/lib/components/raw_apps/RawAppEditorHeader.svelte index c8e9106943..bea1d1ea33 100644 --- a/frontend/src/lib/components/raw_apps/RawAppEditorHeader.svelte +++ b/frontend/src/lib/components/raw_apps/RawAppEditorHeader.svelte @@ -207,7 +207,7 @@ let { summary = $bindable(), - policy = $bindable(), + policy, diffDrawer = undefined, savedApp = $bindable(undefined), version = $bindable(undefined), @@ -352,8 +352,12 @@ saveDrawerOpen = false } - async function computeTriggerables() { - policy = await updateRawAppPolicy(runnables, policy) + // Returns a copy: `policy` is the editor's own object, which the Deploy panel's + // instant toggles mutate in place. Reassigning it would detach the panel from the + // editor, so later toggles would never reach the preview. After a successful save, + // copy only `triggerables_v2` back so the editor's value matches what was deployed. + function policyWithTriggerables() { + return updateRawAppPolicy(runnables, policy) } async function createApp(path: string) { @@ -364,9 +368,9 @@ if (!policy.execution_mode) { policy.execution_mode = 'publisher' } - await computeTriggerables() try { const { js, css } = await getBundle() + const deployedPolicy = await policyWithTriggerables() await AppService.createAppRaw({ workspace: opWorkspace!, formData: { @@ -374,7 +378,7 @@ value: app, path, summary: summary, - policy, + policy: deployedPolicy, deployment_message: deploymentMsg, custom_path: customPath, preserve_on_behalf_of: preserveOnBehalfOf || undefined, @@ -384,6 +388,7 @@ css } }) + policy.triggerables_v2 = deployedPolicy.triggerables_v2 // New path now exists server-side — drop the autocomplete cache so // it shows up immediately instead of after the 60s TTL. invalidateWorkspacePaths(opWorkspace!) @@ -391,7 +396,7 @@ summary: summary, value: structuredClone(stateSnapshot(app)), path: path, - policy: policy, + policy: deployedPolicy, custom_path: customPath, labels: $state.snapshot(labels) } @@ -589,10 +594,10 @@ return } const { js, css } = await getBundle() - await computeTriggerables() if (!policy.execution_mode) { policy.execution_mode = 'publisher' } + const deployedPolicy = await policyWithTriggerables() const deployed = await AppService.updateAppRaw({ workspace: opWorkspace!, path: appPath!, @@ -600,7 +605,7 @@ app: { value: app!, summary: summary, - policy, + policy: deployedPolicy, path: npath, deployment_message: deploymentMsg, preserve_on_behalf_of: preserveOnBehalfOf || undefined, @@ -614,12 +619,13 @@ css } }) + policy.triggerables_v2 = deployedPolicy.triggerables_v2 invalidateWorkspacePaths(opWorkspace!) savedApp = { summary: summary, value: structuredClone(stateSnapshot(app)), path: npath, - policy, + policy: deployedPolicy, custom_path: customPath, labels: $state.snapshot(labels) } @@ -655,12 +661,13 @@ } async function setPublishState(message?: string) { - await computeTriggerables() + const nextPolicy = await policyWithTriggerables() await AppService.updateApp({ workspace: opWorkspace!, path: appPath, - requestBody: { policy } + requestBody: { policy: nextPolicy } }) + policy.triggerables_v2 = nextPolicy.triggerables_v2 if (message) { sendUserToast(message) } else if (policy.execution_mode == 'anonymous') { From c52ba2fd30e19986522420ff02fae224a075b4f6 Mon Sep 17 00:00:00 2001 From: hugocasa Date: Thu, 8 Oct 2026 23:24:00 +0200 Subject: [PATCH 08/38] feat: agent nits: multi-select, memory-off notice, agent step label (#11622) * feat: agent nits: multi-select, memory-off notice, agent step label Co-Authored-By: Claude Opus 5.5 * fix: say saved, not renamed, on a new agent's first deploy Co-Authored-By: Claude Opus 5.5 * fix: run the agent page form on cmd+enter, select agents by keyboard Co-Authored-By: Claude Opus 5.5 * test: pin the agent step label Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- .../components/FlowStatusViewerInner.svelte | 6 +++- .../src/lib/components/FlowTimeline.svelte | 4 +-- .../components/common/table/AgentRow.svelte | 14 ++++++--- .../lib/components/flows/agentDraft.svelte.ts | 7 ++++- .../flows/content/AgentEditorHost.svelte | 5 ++-- .../conversations/agentEditorChat.test.ts | 10 ++++++- .../flows/conversations/agentEditorChat.ts | 9 ++++++ .../map/FlowModuleSchemaItemViewer.svelte | 4 ++- frontend/src/lib/components/home/Item.svelte | 6 ++-- .../src/lib/components/home/ItemsList.svelte | 3 +- .../lib/components/home/bulkActions.test.ts | 9 ++++++ .../src/lib/components/home/bulkActions.ts | 20 +++++++++---- .../components/home/homeSelection.svelte.ts | 2 +- .../agents/get/[...path]/+page.svelte | 29 +++++++++++++++++++ .../(root)/(logged)/run/[...run]/+page.svelte | 4 ++- 15 files changed, 107 insertions(+), 25 deletions(-) diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index 1d56c20b38..638956add2 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/frontend/src/lib/components/FlowStatusViewerInner.svelte @@ -69,6 +69,7 @@ retainLinkedToolsScope } from './flows/linkedAgentToolsStore.svelte' import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte' + import { AGENT_STEP_ID, agentStepLabel } from './flows/conversations/agentEditorChat' const operatingWorkspace = useOperatingWorkspace() @@ -1238,6 +1239,8 @@ id: string type: FlowModuleValue['type'] suspend?: boolean + /** Shown in place of the id. */ + label?: string } function allModulesForTimeline( @@ -1250,7 +1253,8 @@ ({ id: x.id, type: x.value.type, - suspend: x.suspend != undefined + suspend: x.suspend != undefined, + label: x.id === AGENT_STEP_ID ? agentStepLabel(job?.script_path) : undefined }) as FlowModuleForTimeline, { skipToolNodes: true diff --git a/frontend/src/lib/components/FlowTimeline.svelte b/frontend/src/lib/components/FlowTimeline.svelte index 4ed13c1df1..640c8728ce 100644 --- a/frontend/src/lib/components/FlowTimeline.svelte +++ b/frontend/src/lib/components/FlowTimeline.svelte @@ -179,12 +179,12 @@ />

{/if} - {#each flowModules as { id: k, type: typ, suspend: isSuspend } (k)} + {#each flowModules as { id: k, type: typ, suspend: isSuspend, label } (k)} {@const subItems = items?.[k]?.filter((x) => x.created_at && x.started_at)}
{k.startsWith('subflow:') ? k.substring(8) : k} + >{label ?? (k.startsWith('subflow:') ? k.substring(8) : k)} {#if localModuleStates[k]?.selectedForloop && (typ == 'forloopflow' || typ == 'whileloopflow')}
{/if} {/if} {:else} -
+
{#snippet header()}{/snippet} {#snippet footer()}{/snippet} @@ -1154,199 +1268,249 @@ {@const load_more_prefix = is_load_more ? item_key.slice(0, -1) : ''} {@const file_info = allFilesByKey[item_key]} -
+
{#if is_load_more} {@const loadMoreNesting = nestingLevelOf(load_more_prefix + 'x') - 2 * rootPathNestingLevel} {@const loadingMore = folderState[load_more_prefix]?.loading === true} - - -
!loadingMore && loadMore(load_more_prefix)} - class={twMerge( - 'flex flex-row h-full text-xs items-center justify-start text-secondary', - loadingMore ? 'cursor-default' : 'cursor-pointer' - )} + disabled={loadingMore} + class="w-full text-xs font-normal flex gap-2 items-center h-8 pr-1 text-secondary hover:bg-surface-hover cursor-pointer disabled:cursor-default disabled:hover:bg-transparent" + style={treeRowIndent(loadMoreNesting)} > -
- -
- {#if loadingMore} - - {:else} - - {/if} -
-
Load more
-
-
+ + {#if loadingMore} + + {:else} + + {/if} + Load more + {:else if file_info} {@const nestingLevel = file_info.nestingLevel - 2 * rootPathNestingLevel} - - -
selectItem(index)} + title={file_info.full_key.slice(rootPath.length)} class={twMerge( - 'flex flex-row h-full font-semibold text-xs items-center justify-start', - selectedFileKey !== undefined && selectedFileKey.s3 === file_info.full_key - ? 'bg-surface-hover' - : '' + 'w-full text-xs font-normal text-primary flex gap-2 items-center h-8 pr-1 cursor-pointer', + isSelected ? 'bg-surface-secondary' : 'hover:bg-surface-hover' )} + style={treeRowIndent(nestingLevel)} > -
- {#if file_info.type === 'folder'} - {#if folderState[file_info.full_key]?.loading} - - {:else if file_info.collapsed}{:else}{/if} -
- - {#if file_info.display_name === ''} - (empty name) - {:else}{file_info.display_name}{/if} - {#if !lazyMode} - ({file_info.count}{count % 1000 === 0 && - lastKeyFolders[file_info.nestingLevel / 2] === file_info.display_name - ? '+' - : ''} item{file_info.count === 1 ? '' : 's'}) - {/if} -
+ {#if isFolder} + {#if folderState[file_info.full_key]?.loading} + {:else} - -
- {file_info.display_name} -
+ {/if} -
-
+ {:else} + + {/if} + + + + {#if file_info.display_name === ''} + (empty name) + {:else}{file_info.display_name}{/if} + + + {#if isFolder && !lazyMode} + + {file_info.count}{count % 1000 === 0 && + lastKeyFolders[file_info.nestingLevel / 2] === file_info.display_name + ? '+' + : ''} + + {/if} + {/if}
{/snippet}
{#if fileListLoading === true} -
- Loading content +
+ Loading...
{:else if lazyMode}
{displayedCount} item{displayedCount === 1 ? '' : 's'} shown
{:else} -
+
{displayedCount}{flatHasMore ? '+' : ''} {displayedCount !== count ? 'filtered ' : ''}items (including inside folders)
{#if flatHasMore} - {/if} {/if}
{/if} -
+ {/if} -
- {#if fileMetadata === undefined} -
- {#if fileInfoLoading} -
- {:else if fileListUnavailable} -
- {:else} -
- {/if} -
- {:else} -
-
-

- {((p) => (p.startsWith(rootPath) ? p.slice(rootPath.length) : p))( - fileMetadata.fileKey - )} -

- {#if filePreview !== undefined && (!hideS3SpecificDetails || !readOnlyMode || allowDelete)} -
- {#if !hideS3SpecificDetails} - {@const downloadApiPath = `/w/${ws}/job_helpers/download_s3_file?file_key=${encodeURIComponent(fileMetadata?.fileKey ?? '')}${storage ? `&storage=${storage}` : ''}${s3ResourcePath ? `&s3_resource_path=${encodeURIComponent(s3ResourcePath)}` : ''}`} - {@const downloadName = - fileMetadata?.fileKey.split('/').pop() ?? 'unnamed_download.file'} - {#if shouldDownloadViaClient()} -
+ {:else if !csvOptions} +
+ {#if filePreview.contentType === 'Csv'} + Previewing a {filePreview.contentType?.toLowerCase()} file. Separator character: +
+ +
+ Header row: +
+ +
+ {:else} + Previewing a {filePreview.contentType?.toLowerCase()} file. + {/if} +
+ {/if} 
{#if !emptyString(filePreview.contentPreview)}{filePreview.contentPreview}{:else if filePreview.contentType !== undefined}Preview impossible.{/if}
diff --git a/frontend/src/lib/components/apps/components/display/dbtable/utils.ts b/frontend/src/lib/components/apps/components/display/dbtable/utils.ts index 21371aa08b..ae3a82e97c 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/utils.ts +++ b/frontend/src/lib/components/apps/components/display/dbtable/utils.ts @@ -41,6 +41,8 @@ export type ColumnDef = { children?: ColumnDef[] // DBExplorer ignored?: boolean + /** False when the column cannot be filtered on, e.g. a nested value. */ + filterable?: boolean hideInsert?: boolean editable?: boolean overrideDefaultValue?: boolean diff --git a/frontend/src/lib/components/dbTableFilters.ts b/frontend/src/lib/components/dbTableFilters.ts index 9d9c83e6b0..c3266b3ac7 100644 --- a/frontend/src/lib/components/dbTableFilters.ts +++ b/frontend/src/lib/components/dbTableFilters.ts @@ -52,7 +52,7 @@ export function buildDbTableFilterSchema(columns: ColumnDef[]): DbTableFilterSch const columnOfKey: Record = {} const keyOfColumn: Record = {} for (const col of columns) { - if (!col.field) continue + if (!col.field || col.filterable === false) continue let base = col.field.replace(/\W+/g, '_') || 'column' if (base === FREE_TEXT_KEY) base = `${base}col` let key = base diff --git a/frontend/src/lib/components/s3TableOps.test.ts b/frontend/src/lib/components/s3TableOps.test.ts new file mode 100644 index 0000000000..f32ce47c56 --- /dev/null +++ b/frontend/src/lib/components/s3TableOps.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from 'vitest' +import { datatypeOf, jsonTableRows, serverTableFormat } from './s3TableOps' +import { dbColumnKind } from './dbTableFilters' + +describe('serverTableFormat', () => { + it('pages delimited text and newline-delimited JSON through the csv reader', () => { + expect(serverTableFormat('a/b.parquet')).toBe('parquet') + for (const key of ['a.csv', 'a.TSV', 'a.jsonl', 'a.ndjson']) { + expect(serverTableFormat(key)).toBe('csv') + } + }) + + it('leaves plain JSON to the browser', () => { + expect(serverTableFormat('a.json')).toBeUndefined() + expect(serverTableFormat('a.txt')).toBeUndefined() + }) +}) + +describe('datatypeOf', () => { + it('maps Arrow types to the kinds the grid filters by', () => { + expect(dbColumnKind(datatypeOf('Int64'))).toBe('number') + expect(dbColumnKind(datatypeOf('UInt8'))).toBe('number') + expect(dbColumnKind(datatypeOf('Float64'))).toBe('number') + expect(dbColumnKind(datatypeOf('Decimal128(21, 1)'))).toBe('number') + expect(dbColumnKind(datatypeOf('Boolean'))).toBe('boolean') + expect(dbColumnKind(datatypeOf('Utf8View'))).toBe('text') + }) + + it('names a nested type without its field list', () => { + expect(datatypeOf('List(Field { name: "item", data_type: Utf8 })')).toBe('list') + expect(datatypeOf('Timestamp(Microsecond, None)')).toBe('timestamp') + }) +}) + +describe('jsonTableRows', () => { + it('keeps objects and wraps other elements in a value column', () => { + expect(jsonTableRows([{ a: 1 }, 2, null])).toEqual([{ a: 1 }, { value: 2 }, { value: null }]) + }) + + it('has no table for an empty array or a non-array', () => { + expect(jsonTableRows([])).toBeUndefined() + expect(jsonTableRows({ a: 1 })).toBeUndefined() + }) +}) diff --git a/frontend/src/lib/components/s3TableOps.ts b/frontend/src/lib/components/s3TableOps.ts new file mode 100644 index 0000000000..f5d7bd758b --- /dev/null +++ b/frontend/src/lib/components/s3TableOps.ts @@ -0,0 +1,170 @@ +import { HelpersService, OpenAPI } from '$lib/gen' +import { getHeaders } from '$lib/gen/core/request' +import { ColumnIdentity, type ColumnDef } from './apps/components/display/dbtable/utils' +import type { IDbTableOps } from './dbOps' + +/** Files the server can page through as a table, without reading them whole. */ +export function serverTableFormat(fileKey: string): 'parquet' | 'csv' | undefined { + const key = fileKey.toLowerCase() + if (key.endsWith('.parquet')) return 'parquet' + if (/\.(csv|tsv|jsonl|ndjson)$/.test(key)) return 'csv' + return undefined +} + +export type S3TableSource = { + workspace: string + fileKey: string + storage?: string + s3ResourcePath?: string + csvSeparator?: string + csvHasHeader?: boolean +} + +type Chunk = { + columns: string[] + column_types?: string[] + rows: Record[] + csv_separator?: string +} + +/** An Arrow type as the SQL type name the grid filters and formats by. */ +export function datatypeOf(arrowType: string | undefined): string { + const t = arrowType ?? '' + if (/^U?Int\d+$/.test(t)) return 'bigint' + if (/^Float\d+$/.test(t)) return 'double' + if (/^Decimal/.test(t)) return 'decimal' + if (t === 'Boolean') return 'boolean' + if (/^(Large)?Utf8/.test(t)) return 'text' + // A nested type spells out its fields, e.g. `List(Field { name: "item", … })`. + return t.split('(')[0].toLowerCase() || 'text' +} + +/** Read-only table ops over an object storage file, paged, searched, filtered and sorted + * by the server, with the CSV separator it read the file with (guessed when none is given). + */ +export async function s3TableOps( + source: S3TableSource +): Promise<{ ops: IDbTableOps; csvSeparator?: string }> { + const format = serverTableFormat(source.fileKey) + if (!format) throw new Error(`${source.fileKey} cannot be read as a table`) + const common = { + workspace: source.workspace, + path: source.fileKey, + storage: source.storage, + s3ResourcePath: source.s3ResourcePath + } + const csvOptions: { csvSeparator?: string; csvHasHeader?: boolean } = + format === 'csv' ? { csvSeparator: source.csvSeparator, csvHasHeader: source.csvHasHeader } : {} + const filtersParam = (columnFilters: Record | undefined) => + columnFilters && Object.keys(columnFilters).length ? JSON.stringify(columnFilters) : undefined + const searchParam = (quicksearch: string) => quicksearch.trim() || undefined + + async function loadChunk(params: { + offset: number + limit: number + quicksearch?: string + columnFilters?: Record + orderBy?: string + isDesc?: boolean + }): Promise { + const data = { + ...common, + ...csvOptions, + offset: params.offset, + limit: params.limit, + searchTerm: searchParam(params.quicksearch ?? ''), + filters: filtersParam(params.columnFilters), + sortCol: params.orderBy, + sortDesc: params.isDesc + } + return (await (format === 'parquet' + ? HelpersService.loadParquetPreview(data) + : HelpersService.loadCsvPreview(data))) as Chunk + } + + const head = await loadChunk({ offset: 0, limit: 0 }) + // Later reads reuse the separator the server guessed, rather than each guessing again. + if (format === 'csv') csvOptions.csvSeparator ??= head.csv_separator + const colDefs: ColumnDef[] = head.columns.map((field, i) => ({ + field, + datatype: datatypeOf(head.column_types?.[i]), + // The server cannot read a struct, map or union as text to filter it. + filterable: !/^(Struct|Map|Union)/.test(head.column_types?.[i] ?? ""), + defaultvalue: '', + isprimarykey: false, + isidentity: ColumnIdentity.No, + isnullable: 'YES', + isenum: false + })) + + const ops: IDbTableOps = { + dbType: 'duckdb', + tableKey: source.fileKey.split('/').pop() ?? source.fileKey, + colDefs, + getCount: async ({ quicksearch, columnFilters }) => { + const res = await HelpersService.loadTableRowCount({ + ...common, + ...csvOptions, + searchTerm: searchParam(quicksearch), + filters: filtersParam(columnFilters) + }) + return res.count ?? 0 + }, + getRows: async ({ + offset, + limit, + quicksearch, + columnFilters, + order_by, + is_desc, + explicitSort + }) => + ( + await loadChunk({ + offset, + limit, + quicksearch, + columnFilters, + // Unsorted, a file keeps its own row order. + orderBy: explicitSort ? order_by : undefined, + isDesc: is_desc + }) + ).rows + } + return { ops, csvSeparator: head.csv_separator } +} + +/** Plain JSON cannot be paged by the server, so only a file this small is read whole into + * the browser to be shown as a table. */ +export const JSON_TABLE_MAX_BYTES = 10 * 1024 * 1024 + +export function isJsonFile(fileKey: string): boolean { + return fileKey.toLowerCase().endsWith('.json') +} + +/** Rows of a JSON array, an element that is not an object becoming a one-column row; + * undefined for anything else. */ +export function jsonTableRows(value: unknown): Record[] | undefined { + if (!Array.isArray(value) || value.length === 0) return undefined + return value.map((v) => + v !== null && typeof v === 'object' && !Array.isArray(v) + ? (v as Record) + : { value: v } + ) +} + +export async function fetchJsonTableRows( + source: S3TableSource +): Promise[] | undefined> { + const query = new URLSearchParams({ file_key: source.fileKey }) + if (source.storage) query.set('storage', source.storage) + if (source.s3ResourcePath) query.set('s3_resource_path', source.s3ResourcePath) + const apiPath = `/w/${source.workspace}/job_helpers/download_s3_file?${query}` + const headers = await getHeaders(OpenAPI, { method: 'GET', url: apiPath }) + const response = await fetch(`${OpenAPI.BASE}${apiPath}`, { + headers, + credentials: OpenAPI.CREDENTIALS + }) + if (!response.ok) throw new Error(`${response.status} ${response.statusText}`) + return jsonTableRows(await response.json()) +}