From e4784e89dac9fbfe492fa2d25fcaf373afc3b340 Mon Sep 17 00:00:00 2001 From: Lucas Abel <22837557+uael@users.noreply.github.com> Date: Mon, 25 Nov 2024 23:20:51 +0100 Subject: [PATCH] feat(backend): implement flow scripts (#4748) * feat(backend): implement flow scripts * conditionally resolve reduced flow value * conditionally use flow scripts based on the min worker version * grant all to `flow_{version_lite,node}` * do not silently discard errors * env: add flow script feature flag * flow script on cloud automatically --------- Co-authored-by: Ruben Fiszel --- ...a3d89ca7f1a88a48fcf0b1ac07144924c1707.json | 22 ++ ...9177a87e1accd192402e21db5ae09c3498ab0.json | 3 +- ...b32cbde4240e143df1c980843dcdfeef3499e.json | 28 ++ ...a3254556c0307bb6bb596613b1ce2ee7ec4cb.json | 15 + ...98a03f751b246c40daf056fced0fd91f6dd73.json | 3 +- ...b2be02f0f945e587c90d7e0f23706bf6e9a31.json | 23 ++ ...95fbe9c6b59e27c368eaf78b33ecdfa95140f.json | 23 ++ ...d9ee7969f98975406fa13118d2b7e808958cf.json | 15 + ...11480ff040bb987a48b6f66522bc9b7e2c1d8.json | 28 ++ ...7163476791448201bdd22a9e463a4116b9582.json | 23 ++ ...120444af65b1dc43a234821cde5bf6bf8b74f.json | 27 ++ ...86e43bbf89add915bfb09d94e25e8517d65ff.json | 23 ++ ...adc18b72323e06a1f32e7a1a7589649888927.json | 16 + ...b6b493e53583017c18e2ab44f44125c52d548.json | 3 +- backend/Cargo.lock | 1 + .../20241121133158_flow_node.down.sql | 4 + .../20241121133158_flow_node.up.sql | 29 ++ backend/windmill-api/Cargo.toml | 1 + backend/windmill-api/src/jobs.rs | 305 ++++++++++-------- backend/windmill-api/src/users.rs | 9 + backend/windmill-api/src/workspaces.rs | 8 + backend/windmill-common/src/flows.rs | 126 +++++++- backend/windmill-common/src/jobs.rs | 14 +- backend/windmill-common/src/worker.rs | 5 + backend/windmill-queue/src/jobs.rs | 94 ++++-- .../windmill-worker/src/dedicated_worker.rs | 44 ++- backend/windmill-worker/src/worker.rs | 22 +- backend/windmill-worker/src/worker_flow.rs | 70 ++-- .../windmill-worker/src/worker_lockfiles.rs | 153 +++++++-- 29 files changed, 911 insertions(+), 226 deletions(-) create mode 100644 backend/.sqlx/query-00a16dbe1729d530f136279471fa3d89ca7f1a88a48fcf0b1ac07144924c1707.json create mode 100644 backend/.sqlx/query-15a5db0202c61ed187d9594311cb32cbde4240e143df1c980843dcdfeef3499e.json create mode 100644 backend/.sqlx/query-2de4ac337d25cc483829818393fa3254556c0307bb6bb596613b1ce2ee7ec4cb.json create mode 100644 backend/.sqlx/query-625c400bd81369a8142b8c7fa67b2be02f0f945e587c90d7e0f23706bf6e9a31.json create mode 100644 backend/.sqlx/query-721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f.json create mode 100644 backend/.sqlx/query-79992fd5650a99a4391e7c9e590d9ee7969f98975406fa13118d2b7e808958cf.json create mode 100644 backend/.sqlx/query-8653373c2c8301c902e21e8bf5a11480ff040bb987a48b6f66522bc9b7e2c1d8.json create mode 100644 backend/.sqlx/query-94a38b83d59f8c010bd96852bf57163476791448201bdd22a9e463a4116b9582.json create mode 100644 backend/.sqlx/query-aa98ab0e4b9a0eb41a804ab047a120444af65b1dc43a234821cde5bf6bf8b74f.json create mode 100644 backend/.sqlx/query-dc44783a555d41c9df4cd303af686e43bbf89add915bfb09d94e25e8517d65ff.json create mode 100644 backend/.sqlx/query-ed12d54e9cdc18672bbb34d2500adc18b72323e06a1f32e7a1a7589649888927.json create mode 100644 backend/migrations/20241121133158_flow_node.down.sql create mode 100644 backend/migrations/20241121133158_flow_node.up.sql diff --git a/backend/.sqlx/query-00a16dbe1729d530f136279471fa3d89ca7f1a88a48fcf0b1ac07144924c1707.json b/backend/.sqlx/query-00a16dbe1729d530f136279471fa3d89ca7f1a88a48fcf0b1ac07144924c1707.json new file mode 100644 index 0000000000..9c581eae99 --- /dev/null +++ b/backend/.sqlx/query-00a16dbe1729d530f136279471fa3d89ca7f1a88a48fcf0b1ac07144924c1707.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value as \"value: sqlx::types::Json>\" FROM flow_version WHERE id = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value: sqlx::types::Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "00a16dbe1729d530f136279471fa3d89ca7f1a88a48fcf0b1ac07144924c1707" +} diff --git a/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json b/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json index fc94db0309..71680b0922 100644 --- a/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json +++ b/backend/.sqlx/query-0ad36c1598ff4ece0c325eaeb9a9177a87e1accd192402e21db5ae09c3498ab0.json @@ -42,7 +42,8 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow" + "singlescriptflow", + "flowscript" ] } } diff --git a/backend/.sqlx/query-15a5db0202c61ed187d9594311cb32cbde4240e143df1c980843dcdfeef3499e.json b/backend/.sqlx/query-15a5db0202c61ed187d9594311cb32cbde4240e143df1c980843dcdfeef3499e.json new file mode 100644 index 0000000000..d06eadbccf --- /dev/null +++ b/backend/.sqlx/query-15a5db0202c61ed187d9594311cb32cbde4240e143df1c980843dcdfeef3499e.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "lock", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "code!: String", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + true, + true + ] + }, + "hash": "15a5db0202c61ed187d9594311cb32cbde4240e143df1c980843dcdfeef3499e" +} diff --git a/backend/.sqlx/query-2de4ac337d25cc483829818393fa3254556c0307bb6bb596613b1ce2ee7ec4cb.json b/backend/.sqlx/query-2de4ac337d25cc483829818393fa3254556c0307bb6bb596613b1ce2ee7ec4cb.json new file mode 100644 index 0000000000..87e3975ddf --- /dev/null +++ b/backend/.sqlx/query-2de4ac337d25cc483829818393fa3254556c0307bb6bb596613b1ce2ee7ec4cb.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO flow_version_lite (id, value) VALUES ($1, $2)\n ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "2de4ac337d25cc483829818393fa3254556c0307bb6bb596613b1ce2ee7ec4cb" +} diff --git a/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json b/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json index a23bb3827f..695607059e 100644 --- a/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json +++ b/backend/.sqlx/query-337f31c2172194cd594042c561998a03f751b246c40daf056fced0fd91f6dd73.json @@ -32,7 +32,8 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow" + "singlescriptflow", + "flowscript" ] } } diff --git a/backend/.sqlx/query-625c400bd81369a8142b8c7fa67b2be02f0f945e587c90d7e0f23706bf6e9a31.json b/backend/.sqlx/query-625c400bd81369a8142b8c7fa67b2be02f0f945e587c90d7e0f23706bf6e9a31.json new file mode 100644 index 0000000000..248f44b23a --- /dev/null +++ b/backend/.sqlx/query-625c400bd81369a8142b8c7fa67b2be02f0f945e587c90d7e0f23706bf6e9a31.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT coalesce(flow_version_lite.value, flow_version.value) as \"value!: sqlx::types::Json>\" FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n LEFT JOIN flow_version_lite \n ON flow_version_lite.id = flow_version.id\n WHERE flow.path = $1 AND flow.workspace_id = $2 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value!: sqlx::types::Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "625c400bd81369a8142b8c7fa67b2be02f0f945e587c90d7e0f23706bf6e9a31" +} diff --git a/backend/.sqlx/query-721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f.json b/backend/.sqlx/query-721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f.json new file mode 100644 index 0000000000..9dcfbc697b --- /dev/null +++ b/backend/.sqlx/query-721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT coalesce(flow_version_lite.value, flow_version.value) AS \"value!: sqlx::types::Json>\" \n FROM flow \n LEFT JOIN flow_version \n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n LEFT JOIN flow_version_lite \n ON flow_version_lite.id = flow_version.id\n WHERE flow.path = $1 AND flow.workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value!: sqlx::types::Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "721ab4b4ce441a86d43e1b89ab295fbe9c6b59e27c368eaf78b33ecdfa95140f" +} diff --git a/backend/.sqlx/query-79992fd5650a99a4391e7c9e590d9ee7969f98975406fa13118d2b7e808958cf.json b/backend/.sqlx/query-79992fd5650a99a4391e7c9e590d9ee7969f98975406fa13118d2b7e808958cf.json new file mode 100644 index 0000000000..8447584680 --- /dev/null +++ b/backend/.sqlx/query-79992fd5650a99a4391e7c9e590d9ee7969f98975406fa13118d2b7e808958cf.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE flow_node SET workspace_id = $1 WHERE workspace_id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Text" + ] + }, + "nullable": [] + }, + "hash": "79992fd5650a99a4391e7c9e590d9ee7969f98975406fa13118d2b7e808958cf" +} diff --git a/backend/.sqlx/query-8653373c2c8301c902e21e8bf5a11480ff040bb987a48b6f66522bc9b7e2c1d8.json b/backend/.sqlx/query-8653373c2c8301c902e21e8bf5a11480ff040bb987a48b6f66522bc9b7e2c1d8.json new file mode 100644 index 0000000000..f57226bfa9 --- /dev/null +++ b/backend/.sqlx/query-8653373c2c8301c902e21e8bf5a11480ff040bb987a48b6f66522bc9b7e2c1d8.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "lock", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "code!: String", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + true, + true + ] + }, + "hash": "8653373c2c8301c902e21e8bf5a11480ff040bb987a48b6f66522bc9b7e2c1d8" +} diff --git a/backend/.sqlx/query-94a38b83d59f8c010bd96852bf57163476791448201bdd22a9e463a4116b9582.json b/backend/.sqlx/query-94a38b83d59f8c010bd96852bf57163476791448201bdd22a9e463a4116b9582.json new file mode 100644 index 0000000000..f3667e8f1a --- /dev/null +++ b/backend/.sqlx/query-94a38b83d59f8c010bd96852bf57163476791448201bdd22a9e463a4116b9582.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT coalesce(flow_version_lite.value, flow_version.value) as \"value!: sqlx::types::Json>\" FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n LEFT JOIN flow_version_lite \n ON flow_version_lite.id = flow_version.id\n WHERE flow.path = $1 AND flow.workspace_id = $2 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value!: sqlx::types::Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "94a38b83d59f8c010bd96852bf57163476791448201bdd22a9e463a4116b9582" +} diff --git a/backend/.sqlx/query-aa98ab0e4b9a0eb41a804ab047a120444af65b1dc43a234821cde5bf6bf8b74f.json b/backend/.sqlx/query-aa98ab0e4b9a0eb41a804ab047a120444af65b1dc43a234821cde5bf6bf8b74f.json new file mode 100644 index 0000000000..d70e81e786 --- /dev/null +++ b/backend/.sqlx/query-aa98ab0e4b9a0eb41a804ab047a120444af65b1dc43a234821cde5bf6bf8b74f.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH existing AS (\n SELECT id FROM flow_node\n WHERE hash = $1 AND path = $2 AND workspace_id = $3 AND code = $4 AND lock = $5 AND flow = $6\n LIMIT 1\n ),\n inserted AS (\n INSERT INTO flow_node (hash, path, workspace_id, code, lock, flow)\n VALUES ($1, $2, $3, $4, $5, $6)\n ON CONFLICT DO NOTHING\n RETURNING id\n )\n SELECT id FROM existing\n UNION ALL\n SELECT id FROM inserted\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text", + "Text", + "Text", + "Text", + "Jsonb" + ] + }, + "nullable": [ + null + ] + }, + "hash": "aa98ab0e4b9a0eb41a804ab047a120444af65b1dc43a234821cde5bf6bf8b74f" +} diff --git a/backend/.sqlx/query-dc44783a555d41c9df4cd303af686e43bbf89add915bfb09d94e25e8517d65ff.json b/backend/.sqlx/query-dc44783a555d41c9df4cd303af686e43bbf89add915bfb09d94e25e8517d65ff.json new file mode 100644 index 0000000000..5f72d6aa21 --- /dev/null +++ b/backend/.sqlx/query-dc44783a555d41c9df4cd303af686e43bbf89add915bfb09d94e25e8517d65ff.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT flow_version.value as \"value!: sqlx::types::Json>\" FROM flow \n LEFT JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 AND flow.workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value!: sqlx::types::Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "dc44783a555d41c9df4cd303af686e43bbf89add915bfb09d94e25e8517d65ff" +} diff --git a/backend/.sqlx/query-ed12d54e9cdc18672bbb34d2500adc18b72323e06a1f32e7a1a7589649888927.json b/backend/.sqlx/query-ed12d54e9cdc18672bbb34d2500adc18b72323e06a1f32e7a1a7589649888927.json new file mode 100644 index 0000000000..0088122f66 --- /dev/null +++ b/backend/.sqlx/query-ed12d54e9cdc18672bbb34d2500adc18b72323e06a1f32e7a1a7589649888927.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE flow_node SET path = REGEXP_REPLACE(path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1') WHERE path LIKE ('u/' || $2 || '/%') AND workspace_id = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "ed12d54e9cdc18672bbb34d2500adc18b72323e06a1f32e7a1a7589649888927" +} diff --git a/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json b/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json index e5a7c10282..a3465b1c1f 100644 --- a/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json +++ b/backend/.sqlx/query-f9fc0084fe086ef80005bb64a8bb6b493e53583017c18e2ab44f44125c52d548.json @@ -46,7 +46,8 @@ "noop", "appdependencies", "deploymentcallback", - "singlescriptflow" + "singlescriptflow", + "flowscript" ] } } diff --git a/backend/Cargo.lock b/backend/Cargo.lock index d37e6c3032..58e08286fb 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -10485,6 +10485,7 @@ dependencies = [ "candle-transformers", "chrono", "chrono-tz 0.10.0", + "const_format", "cookie 0.17.0", "cron", "datafusion", diff --git a/backend/migrations/20241121133158_flow_node.down.sql b/backend/migrations/20241121133158_flow_node.down.sql new file mode 100644 index 0000000000..89db08e503 --- /dev/null +++ b/backend/migrations/20241121133158_flow_node.down.sql @@ -0,0 +1,4 @@ +-- Add down migration script here +DROP TABLE IF EXISTS flow_version_lite; +DROP TABLE IF EXISTS flow_node; +DROP INDEX IF EXISTS flow_node_hash; diff --git a/backend/migrations/20241121133158_flow_node.up.sql b/backend/migrations/20241121133158_flow_node.up.sql new file mode 100644 index 0000000000..152128767b --- /dev/null +++ b/backend/migrations/20241121133158_flow_node.up.sql @@ -0,0 +1,29 @@ +-- Add up migration script here +ALTER TYPE JOB_KIND ADD VALUE IF NOT EXISTS 'flowscript'; + +-- Same as `flow_version` but with a "lite" value (e.g. `RawScript` replaced by `FlowScript`). +CREATE TABLE flow_version_lite ( + id BIGSERIAL PRIMARY KEY, + value JSONB, + FOREIGN KEY (id) REFERENCES flow_version (id) ON DELETE CASCADE +); + +GRANT ALL ON flow_version_lite TO windmill_user; +GRANT ALL ON flow_version_lite TO windmill_admin; + +-- Either a script or a flow value. +CREATE TABLE flow_node ( + id BIGSERIAL PRIMARY KEY, + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id), + hash BIGINT NOT NULL, + path VARCHAR(255) NOT NULL, -- flow path. + lock TEXT, + code TEXT, + flow JSONB, + FOREIGN KEY (path, workspace_id) REFERENCES flow (path, workspace_id) ON DELETE CASCADE +); + +GRANT ALL ON flow_node TO windmill_user; +GRANT ALL ON flow_node TO windmill_admin; + +CREATE INDEX flow_node_hash ON flow_node (hash); diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index f05c2983c8..b9968a75c8 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -97,6 +97,7 @@ jsonwebtoken = { workspace = true } matchit.workspace = true tokio-tungstenite.workspace = true rdkafka = { workspace = true, optional = true } +const_format.workspace = true pin-project.workspace = true http.workspace = true diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 67e64e8d9a..2ba8139f70 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -63,7 +63,7 @@ use windmill_common::{ db::UserDB, error::{self, to_anyhow, Error}, flow_status::{Approval, FlowStatus, FlowStatusModule}, - flows::{add_virtual_items_if_necessary, FlowValue}, + flows::{add_virtual_items_if_necessary, resolve_maybe_value, FlowValue}, jobs::{script_path_to_payload, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode}, oauth2::HmacSha256, scripts::{ScriptHash, ScriptLang}, @@ -593,7 +593,7 @@ async fn get_flow_job_debug_info( Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, ) -> error::Result { - let job = get_queued_job_ex(&db, &w_id, id, false, None).await?; + let job = GetQuery::new().fetch_queued(&db, id, &w_id).await?; if let Some(job) = job { let is_flow = job.is_flow(); if job.is_flow_step || !is_flow { @@ -630,7 +630,7 @@ async fn get_flow_job_debug_info( } } for job_id in job_ids { - let job = get_job_internal(&db, w_id.as_str(), job_id, false, Some(&opt_authed)).await; + let job = GetQuery::new().with_auth(&opt_authed).fetch(&db, job_id, &w_id).await; if let Ok(job) = job { jobs.insert(job.id().to_string(), job); } @@ -658,14 +658,11 @@ async fn get_job( Path((w_id, id)): Path<(String, Uuid)>, Query(GetJobQuery { no_logs }): Query, ) -> error::Result { - let mut job = get_job_internal( - &db, - w_id.as_str(), - id, - no_logs.unwrap_or(false), - Some(&opt_authed), - ) - .await?; + let mut get = GetQuery::new().with_auth(&opt_authed); + if no_logs.unwrap_or(false) { + get = get.without_logs(); + } + let mut job = get.fetch(&db, id, &w_id).await?; job.fetch_outstanding_wait_time(&db).await?; log_job_view(&db, opt_authed.as_ref(), &w_id, &id).await?; @@ -673,136 +670,154 @@ async fn get_job( Ok(Json(job).into_response()) } -lazy_static::lazy_static! { - static ref GET_COMPLETED_JOB_QUERY_NO_LOGS: String = generate_get_job_query(true, "completed_job_view"); - static ref GET_COMPLETED_JOB_QUERY: String = generate_get_job_query(false, "completed_job_view"); - static ref GET_QUEUED_JOB_QUERY_NO_LOGS: String = generate_get_job_query(true, "queue_view"); - static ref GET_QUEUED_JOB_QUERY: String = generate_get_job_query(false, "queue_view"); -} -fn generate_get_job_query(no_logs: bool, table: &str) -> String { - let log_expr = if no_logs { - "null".to_string() - } else { - format!("right({table}.logs, 20000)") +macro_rules! get_job_query { + ("completed_job_view", $($opts:tt)*) => { + get_job_query!( + @impl "completed_job_view", ($($opts)*), + "duration_ms, success, result, deleted, is_skipped, result->'wm_labels' as labels, \ + CASE WHEN result is null or pg_column_size(result) < 90000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result", + ) }; - let additional_fields = if table == "completed_job_view" { - "duration_ms, - success, - result, - deleted, - is_skipped, - result->'wm_labels' as labels, - CASE WHEN result is null or pg_column_size(result) < 90000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result" - } else { - "scheduled_for, - running, - last_ping, - suspend, - suspend_until, - same_worker, - pre_run_error, - visible_to_owner, - root_job, - leaf_jobs, - tag, - concurrent_limit, - concurrency_time_window_s, - timeout, - flow_step_id, - cache_ttl - " + ("queue_view", $($opts:tt)*) => { + get_job_query!( + @impl "queue_view", ($($opts)*), + "scheduled_for, running, last_ping, suspend, suspend_until, same_worker, pre_run_error, visible_to_owner, \ + root_job, leaf_jobs, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl", + ) }; - return format!("SELECT - id, {table}.workspace_id, parent_job, created_by, {table}.created_at, started_at, script_hash, script_path, - CASE WHEN args is null or pg_column_size(args) < 90000 THEN args ELSE '{{\"reason\": \"WINDMILL_TOO_BIG\"}}'::jsonb END as args, - {log_expr} as logs, raw_code, canceled, canceled_by, canceled_reason, job_kind, - schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, - raw_lock, email, visible_to_owner, mem_peak, tag, priority, {additional_fields} - FROM {table} - WHERE id = $1 AND {table}.workspace_id = $2"); -} -pub async fn get_queued_job_ex( - db: &DB, - workspace_id: &str, - job_id: Uuid, - no_logs: bool, - // first optional is if authed need to be checked, second is the opt_authed itself - opt_authed: Option<&Option>, -) -> error::Result>> { - let query = if no_logs { - &*GET_QUEUED_JOB_QUERY_NO_LOGS - } else { - &*GET_QUEUED_JOB_QUERY - }; - let job = sqlx::query_as::<_, JobExtended>(query) - .bind(job_id) - .bind(workspace_id) - .fetch_optional(db) - .await?; - - if let Some(job) = job.as_ref() { - if opt_authed.is_some_and(|x| x.is_none()) && job.created_by != "anonymous" { - return Err(Error::BadRequest( - "As a non logged in user, you can only see jobs ran by anonymous users".to_string(), - )); + (@impl $table:literal, (with_logs: $with_logs:expr, $($rest:tt)*), $additional_fields:literal, $($args:tt)*) => { + if $with_logs { + get_job_query!(@impl $table, ($($rest)*), $additional_fields, logs = const_format::formatcp!("right({}.logs, 20000)", $table), $($args)*) + } else { + get_job_query!(@impl $table, ($($rest)*), $additional_fields, logs = "null", $($args)*) } - } - - Ok(job) -} -pub async fn get_completed_job_ex( - db: &DB, - workspace_id: &str, - job_id: Uuid, - no_logs: bool, - // first optional is if authed need to be checked, second is the opt_authed itself - opt_authed: Option<&Option>, -) -> error::Result>> { - let query = if no_logs { - &*GET_COMPLETED_JOB_QUERY_NO_LOGS - } else { - &*GET_COMPLETED_JOB_QUERY }; - let cjob = sqlx::query_as::<_, JobExtended>(query) - .bind(job_id) - .bind(workspace_id) - .fetch_optional(db) - .await?; - - if let Some(job) = cjob.as_ref() { - if opt_authed.is_some_and(|x| x.is_none()) && job.created_by != "anonymous" { - return Err(Error::BadRequest( - "As a non logged in user, you can only see jobs ran by anonymous users".to_string(), - )); + (@impl $table:literal, (with_code: $with_code:expr, $($rest:tt)*), $additional_fields:literal, $($args:tt)*) => { + if $with_code { + get_job_query!(@impl $table, ($($rest)*), $additional_fields, lock = "raw_lock", code = "raw_code", $($args)*) + } else { + get_job_query!(@impl $table, ($($rest)*), $additional_fields, lock = "null", code = "null", $($args)*) } + }; + (@impl $table:literal, (with_flow: $with_flow:expr, $($rest:tt)*), $additional_fields:literal, $($args:tt)*) => { + if $with_flow { + get_job_query!(@impl $table, ($($rest)*), $additional_fields, flow = "raw_flow", $($args)*) + } else { + get_job_query!(@impl $table, ($($rest)*), $additional_fields, flow = "null", $($args)*) + } + }; + (@impl $table:literal, (), $additional_fields:literal, $($args:tt)*) => { + const_format::formatcp!( + "SELECT \ + id, {table}.workspace_id, parent_job, created_by, {table}.created_at, started_at, script_hash, script_path, \ + CASE WHEN args is null or pg_column_size(args) < 90000 THEN args ELSE '{{\"reason\": \"WINDMILL_TOO_BIG\"}}'::jsonb END as args, \ + {logs} as logs, {code} as raw_code, canceled, canceled_by, canceled_reason, job_kind, \ + schedule_path, permissioned_as, flow_status, {flow} as raw_flow, is_flow_step, language, \ + {lock} as raw_lock, email, visible_to_owner, mem_peak, tag, priority, {additional_fields} \ + FROM {table} \ + WHERE id = $1 AND {table}.workspace_id = $2 LIMIT 1", + table = $table, + additional_fields = $additional_fields, + $($args)* + ) } - - if let Some(mut cjob) = cjob { - cjob.inner = format_completed_job_result(cjob.inner); - return Ok(Some(cjob)); - } - - Ok(cjob) } -pub async fn get_job_internal( - db: &DB, - workspace_id: &str, - job_id: Uuid, - no_logs: bool, - // first optional is if authed need to be checked, second is the opt_authed itself - opt_authed: Option<&Option>, -) -> error::Result { - let cjob = get_completed_job_ex(db, workspace_id, job_id, no_logs, opt_authed.clone()) - .await? - .map(Job::CompletedJob); - match cjob { - Some(cjob) => Ok(cjob), - None => { - let job_maybe = get_queued_job_ex(db, workspace_id, job_id, no_logs, opt_authed) - .await? - .map(Job::QueuedJob); - not_found_if_none(job_maybe, "Job", job_id.to_string()) +#[derive(Copy, Clone)] +struct GetQuery<'a> { + with_logs: bool, + with_code: bool, + with_flow: bool, + with_auth: Option<&'a Option>, +} + +impl<'a> GetQuery<'a> { + fn new() -> Self { + Self { with_logs: true, with_code: true, with_flow: true, with_auth: None } + } + + fn without_logs(self) -> Self { + Self { with_logs: false, ..self } + } + + fn without_code(self) -> Self { + Self { with_code: false, ..self } + } + + fn without_flow(self) -> Self { + Self { with_flow: false, ..self } + } + + fn with_auth(self, auth: &'a Option) -> Self { + Self { with_auth: Some(auth), ..self } + } + + fn check_auth(self, email: Option<&str>) -> error::Result<()> { + if let Some(email) = email { + if self.with_auth.is_some_and(|x| x.is_none()) && email != "anonymous" { + return Err(Error::BadRequest( + "As a non logged in user, you can only see jobs ran by anonymous users".to_string(), + )); + } + } + Ok(()) + } + + async fn fetch_queued(self, db: &DB, job_id: Uuid, workspace_id: &str) -> error::Result>> { + let query = get_job_query!("queue_view", + with_logs: self.with_logs, + with_code: self.with_code, + with_flow: self.with_flow, + ); + let mut job = sqlx::query_as::<_, JobExtended>(query) + .bind(job_id) + .bind(workspace_id) + .fetch_optional(db) + .await?; + + self.check_auth(job.as_ref().map(|job| job.created_by.as_str()))?; + if self.with_flow { + job = resolve_maybe_value(db, workspace_id, self.with_code, job, |job| job.raw_flow.as_mut()).await?; + } + Ok(job) + } + + async fn fetch_completed(self, db: &DB, job_id: Uuid, workspace_id: &str) -> error::Result>> { + let query = get_job_query!("completed_job_view", + with_logs: self.with_logs, + with_code: self.with_code, + with_flow: self.with_flow, + ); + let mut cjob = sqlx::query_as::<_, JobExtended>(query) + .bind(job_id) + .bind(workspace_id) + .fetch_optional(db) + .await?; + + self.check_auth(cjob.as_ref().map(|job| job.created_by.as_str()))?; + if self.with_flow { + cjob = resolve_maybe_value(db, workspace_id, self.with_code, cjob, |job| job.raw_flow.as_mut()).await?; + } + if let Some(mut cjob) = cjob { + cjob.inner = format_completed_job_result(cjob.inner); + return Ok(Some(cjob)); + } + Ok(cjob) + } + + async fn fetch(self, db: &DB, job_id: Uuid, workspace_id: &str) -> error::Result { + let cjob = self.fetch_completed(db, job_id, workspace_id) + .await? + .map(Job::CompletedJob); + + match cjob { + Some(cjob) => Ok(cjob), + None => { + let job_maybe = self.fetch_queued(db, job_id, workspace_id) + .await? + .map(Job::QueuedJob); + not_found_if_none(job_maybe, "Job", job_id.to_string()) + } } } } @@ -1787,8 +1802,12 @@ async fn resume_suspended_job_internal( verify_suspended_secret(&w_id, &db, job_id, resume_id, &approver, secret).await?; let parent_flow_info = get_suspended_parent_flow_info(job_id, &db).await?; - let parent_flow = - get_job_internal(&db, w_id.as_str(), parent_flow_info.id, false, None).await?; + let parent_flow = GetQuery::new() + .without_logs() + .without_code() + .without_flow() + .fetch(&db, parent_flow_info.id, &w_id) + .await?; let flow_status = parent_flow .flow_status() .ok_or_else(|| anyhow::anyhow!("unable to find the flow status in the flow job"))?; @@ -2056,7 +2075,11 @@ pub async fn get_suspended_job_flow( .flatten() .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?; - let flow = get_job_internal(&db, w_id.as_str(), flow_id, true, None).await?; + let flow = GetQuery::new() + .without_logs() + .without_code() + .fetch(&db, flow_id, &w_id) + .await?; let flow_status = flow .flow_status() @@ -3130,7 +3153,7 @@ pub async fn run_workflow_as_code( i += 1; } - let job = get_queued_job_ex(&db, &w_id, job_id, true, None).await?; + let job = GetQuery::new().without_logs().fetch_queued(&db, job_id, &w_id).await?; if *CLOUD_HOSTED { tracing::info!("workflow_as_code_tracing id {i} "); @@ -4415,10 +4438,12 @@ async fn add_batch_jobs( (value, JobKind::FlowPreview, None) } else if let Some(path) = batch_info.path { let value_json = sqlx::query!( - "SELECT flow_version.value AS \"value: sqlx::types::Json>\" FROM flow + "SELECT coalesce(flow_version_lite.value, flow_version.value) as \"value!: sqlx::types::Json>\" FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] - WHERE flow.path = $1 AND flow.workspace_id = $2", + LEFT JOIN flow_version_lite + ON flow_version_lite.id = flow_version.id + WHERE flow.path = $1 AND flow.workspace_id = $2 LIMIT 1", &path, &w_id ) .fetch_optional(&db) @@ -5103,7 +5128,7 @@ async fn get_completed_job<'a>( Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, ) -> error::Result { - let job_o = get_completed_job_ex(&db, &w_id, id, false, Some(&opt_authed)).await?; + let job_o = GetQuery::new().with_auth(&opt_authed).fetch_completed(&db, id, &w_id).await?; let cj = not_found_if_none(job_o, "Completed Job", id.to_string())?; let response = Json(cj).into_response(); diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index 3afe303cb9..0b12ddef27 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -3015,6 +3015,15 @@ async fn update_username_in_workpsace<'c>( .execute(&mut **tx) .await?; + sqlx::query!( + r#"UPDATE flow_node SET path = REGEXP_REPLACE(path,'u/' || $2 || '/(.*)','u/' || $1 || '/\1') WHERE path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#, + new_username, + old_username, + w_id + ) + .execute(&mut **tx) + .await?; + sqlx::query!( "DELETE FROM flow WHERE path LIKE ('u/' || $1 || '/%') AND workspace_id = $2", old_username, diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 5942c838fc..c6d6e5301c 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -2882,6 +2882,14 @@ async fn change_workspace_id( .execute(&mut *tx) .await?; + sqlx::query!( + "UPDATE flow_node SET workspace_id = $1 WHERE workspace_id = $2", + &rw.new_id, + &old_id + ) + .execute(&mut *tx) + .await?; + sqlx::query!("DELETE FROM flow WHERE workspace_id = $1", &old_id) .execute(&mut *tx) .await?; diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index d357a5ec72..c5f1b9dede 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -14,10 +14,14 @@ use std::{ use rand::Rng; use serde::{Deserialize, Serialize, Serializer}; +use sqlx::types::Json; +use sqlx::types::JsonRawValue; use crate::{ + error::Error, more_serde::{default_empty_string, default_id, default_null, default_true, is_default}, scripts::{Schema, ScriptHash, ScriptLang}, + worker::to_raw_value, }; #[derive(Serialize, Deserialize, sqlx::FromRow)] @@ -26,7 +30,7 @@ pub struct Flow { pub path: String, pub summary: String, pub description: String, - pub value: sqlx::types::Json>, + pub value: Json>, pub edited_by: String, pub edited_at: chrono::DateTime, pub archived: bool, @@ -398,6 +402,11 @@ pub enum InputTransform { }, } +/// Id in the `flow_node` table. +#[derive(Serialize, Deserialize, Debug, Copy, Clone, Hash)] +#[serde(transparent)] +pub struct FlowNodeId(pub i64); + #[derive(Serialize, Deserialize, Debug, Clone)] pub struct Branch { #[serde(skip_serializing_if = "Option::is_none")] @@ -480,6 +489,24 @@ pub enum FlowModuleValue { is_trigger: Option, }, Identity, + // Internal only, never exposed to the frontend. + FlowScript { + #[serde(default)] + #[serde(alias = "input_transform", serialize_with = "ordered_map")] + input_transforms: HashMap, + id: FlowNodeId, + #[serde(skip_serializing_if = "is_none_or_empty")] + tag: Option, + language: ScriptLang, + #[serde(skip_serializing_if = "Option::is_none")] + custom_concurrency_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + concurrent_limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + concurrency_time_window_s: Option, + #[serde(skip_serializing_if = "Option::is_none")] + is_trigger: Option, + }, } fn is_none_or_empty(expr: &Option) -> bool { @@ -510,6 +537,7 @@ struct UntaggedFlowModuleValue { concurrent_limit: Option, concurrency_time_window_s: Option, is_trigger: Option, + id: Option, } impl<'de> Deserialize<'de> for FlowModuleValue { @@ -582,6 +610,20 @@ impl<'de> Deserialize<'de> for FlowModuleValue { concurrency_time_window_s: untagged.concurrency_time_window_s, is_trigger: untagged.is_trigger, }), + "flowscript" => Ok(FlowModuleValue::FlowScript { + input_transforms: untagged.input_transforms.unwrap_or_default(), + id: untagged + .id + .ok_or_else(|| serde::de::Error::missing_field("id"))?, + tag: untagged.tag, + language: untagged + .language + .ok_or_else(|| serde::de::Error::missing_field("language"))?, + custom_concurrency_key: untagged.custom_concurrency_key, + concurrent_limit: untagged.concurrent_limit, + concurrency_time_window_s: untagged.concurrency_time_window_s, + is_trigger: untagged.is_trigger, + }), "identity" => Ok(FlowModuleValue::Identity), other => Err(serde::de::Error::unknown_variant( other, @@ -651,3 +693,85 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec) { }); } } + +/// Resolve the value of a flow if any. +pub async fn resolve_maybe_value( + e: &sqlx::PgPool, + workspace_id: &str, + with_code: bool, + maybe: Option, + value_mut: impl FnOnce(&mut T) -> Option<&mut Json>> +) -> Result, Error> { + let Some(mut container) = maybe else { return Ok(None); }; + let Some(value) = value_mut(&mut container) else { return Ok(Some(container)); }; + resolve_value(e, workspace_id, &mut value.0, with_code).await?; + Ok(Some(container)) +} + +/// Resolve modules recursively. +pub async fn resolve_value( + e: &sqlx::PgPool, + workspace_id: &str, + value: &mut Box, + with_code: bool, +) -> Result<(), Error> { + let mut val = serde_json::from_str::(value.get()) + .map_err(|err| Error::InternalErr(format!("resolve: Failed to parse flow value: {}", err)))?; + for module in &mut val.modules { + resolve_module(e, workspace_id, &mut module.value, with_code).await?; + } + *value = to_raw_value(&val); + Ok(()) +} + +/// Resolve module value recursively. +pub async fn resolve_module( + e: &sqlx::PgPool, + workspace_id: &str, + value: &mut Box, + with_code: bool, +) -> Result<(), Error> { + use FlowModuleValue::*; + + let mut val = serde_json::from_str::(value.get()) + .map_err(|err| Error::InternalErr(format!("resolve: Failed to parse flow module value: {}", err)))?; + match &mut val { + FlowScript { .. } => { + // In order to avoid an unnecessary `.clone()` of `val`, take ownership of it's content + // using `std::mem::replace`. + let FlowScript { + input_transforms, id, tag, language, + custom_concurrency_key, concurrent_limit, concurrency_time_window_s, is_trigger + } = std::mem::replace(&mut val, Identity) else { unreachable!() }; + // Load script lock file and code content. + let (lock, content) = if !with_code { + (Some("...".to_string()), "...".to_string()) + } else { + sqlx::query!("SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1", id.0) + .fetch_one(e) + .await + .map_err(Error::SqlErr) + .map(|record| (record.lock, record.code))? + }; + val = RawScript { + input_transforms, content, lock, path: None, tag, language, custom_concurrency_key, + concurrent_limit, concurrency_time_window_s, is_trigger + }; + }, + ForloopFlow { modules, .. } | WhileloopFlow { modules, .. } => { + for module in modules { + Box::pin(resolve_module(e, workspace_id, &mut module.value, with_code)).await?; + } + }, + BranchOne { branches, .. } | BranchAll { branches, .. } => { + for branch in branches { + for module in &mut branch.modules { + Box::pin(resolve_module(e, workspace_id, &mut module.value, with_code)).await?; + } + } + } + _ => {} + } + *value = to_raw_value(&val); + Ok(()) +} diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index ed9b456690..5a8fd24fb1 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -16,13 +16,13 @@ pub const PREPROCESSOR_FAKE_ENTRYPOINT: &str = "__WM_PREPROCESSOR"; use crate::{ error::{self, to_anyhow, Error}, flow_status::{FlowStatus, RestartedFrom}, - flows::{FlowValue, Retry}, + flows::{FlowNodeId, FlowValue, Retry}, get_latest_deployed_hash_for_path, scripts::{ScriptHash, ScriptLang}, worker::{to_raw_value, TMP_DIR}, }; -#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Clone)] +#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Copy, Clone)] #[sqlx(type_name = "JOB_KIND", rename_all = "lowercase")] #[serde(rename_all(serialize = "lowercase"))] pub enum JobKind { @@ -39,6 +39,7 @@ pub enum JobKind { AppDependencies, Noop, DeploymentCallback, + FlowScript, } #[derive(sqlx::FromRow, Debug, Serialize, Clone)] @@ -263,6 +264,15 @@ pub enum JobPayload { priority: Option, apply_preprocessor: bool, }, + FlowScript { + id: FlowNodeId, // flow_node(id). + language: ScriptLang, + custom_concurrency_key: Option, + concurrent_limit: Option, + concurrency_time_window_s: Option, + cache_ttl: Option, + dedicated_worker: Option, + }, Code(RawCode), Dependencies { path: String, diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 5b5b93f425..1f3dc68411 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -93,6 +93,10 @@ lazy_static::lazy_static! { pub static ref MIN_VERSION: Arc> = Arc::new(RwLock::new(Version::new(0, 0, 0))); pub static ref MIN_VERSION_IS_AT_LEAST_1_427: Arc> = Arc::new(RwLock::new(false)); + pub static ref MIN_VERSION_IS_AT_LEAST_1_432: Arc> = Arc::new(RwLock::new(false)); + + // Features flags: + pub static ref DISABLE_FLOW_SCRIPT: bool = std::env::var("DISABLE_FLOW_SCRIPT").ok().is_some_and(|x| x == "1" || x == "true"); } pub async fn make_suspended_pull_query(wc: &WorkerConfig) { @@ -593,6 +597,7 @@ pub async fn update_min_version<'c, E: sqlx::Executor<'c, Database = sqlx::Postg } *MIN_VERSION_IS_AT_LEAST_1_427.write().await = min_version >= Version::new(1, 427, 0); + *MIN_VERSION_IS_AT_LEAST_1_432.write().await = min_version >= Version::new(1, 432, 0); *MIN_VERSION.write().await = min_version.clone(); min_version >= cur_version diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 7ee95c373a..58bc738419 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -58,8 +58,9 @@ use windmill_common::{ users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL}, utils::{not_found_if_none, report_critical_error, StripPath}, worker::{ - to_raw_value, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, - MIN_VERSION_IS_AT_LEAST_1_427, NO_LOGS, WORKER_PULL_QUERIES, WORKER_SUSPENDED_PULL_QUERY, + to_raw_value, CLOUD_HOSTED, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, + DISABLE_FLOW_SCRIPT, MIN_VERSION_IS_AT_LEAST_1_427, MIN_VERSION_IS_AT_LEAST_1_432, NO_LOGS, + WORKER_PULL_QUERIES, WORKER_SUSPENDED_PULL_QUERY, }, DB, METRICS_ENABLED, }; @@ -2930,11 +2931,6 @@ lazy_static::lazy_static! { pub static ref RE_ARG_TAG: Regex = Regex::new(r#"\$args\[(\w+)\]"#).unwrap(); } -#[derive(sqlx::FromRow)] -struct FlowRawValue { - pub value: sqlx::types::Json>, -} - // #[instrument(level = "trace", skip_all)] pub async fn push<'c, 'd>( _db: &Pool, @@ -3192,6 +3188,29 @@ pub async fn push<'c, 'd>( priority, ) } + JobPayload::FlowScript { + id, // flow_node(id). + language, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + dedicated_worker, + } => ( + Some(id.0), + None, + None, + JobKind::FlowScript, + None, + None, + Some(language), + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + cache_ttl, + dedicated_worker, + None, + ), JobPayload::ScriptHub { path } => { if path == "hub/7771/slack" || path == "hub/7836/slack" { permissioned_as = SUPERADMIN_NOTIFICATION_EMAIL.to_string(); @@ -3292,17 +3311,15 @@ pub async fn push<'c, 'd>( ), JobPayload::FlowDependencies { path, dedicated_worker, version } => { let value_json = fetch_scalar_isolated!( - sqlx::query_as::<_, FlowRawValue>("SELECT value FROM flow_version WHERE id = $1",) - .bind(&version), + sqlx::query_scalar!("SELECT value as \"value: sqlx::types::Json>\" FROM flow_version WHERE id = $1 LIMIT 1", &version), tx )? .ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", path)))?; - let value = - serde_json::from_str::(value_json.value.get()).map_err(|err| { - Error::InternalErr(format!( - "could not convert json to flow for {path}: {err:?}" - )) - })?; + let value = serde_json::from_str::(value_json.get()).map_err(|err| { + Error::InternalErr(format!( + "could not convert json to flow for {path}: {err:?}" + )) + })?; ( Some(version), Some(path), @@ -3464,24 +3481,38 @@ pub async fn push<'c, 'd>( ) } JobPayload::Flow { path, dedicated_worker, apply_preprocessor } => { - let value_json = fetch_scalar_isolated!( - sqlx::query_as::<_, FlowRawValue>( - "SELECT flow_version.value FROM flow - LEFT JOIN flow_version - ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] - WHERE flow.path = $1 AND flow.workspace_id = $2", + // Do not use the lite version unless all workers are updated. + let value_json = if *DISABLE_FLOW_SCRIPT || (!*MIN_VERSION_IS_AT_LEAST_1_432.read().await && !*CLOUD_HOSTED) { + fetch_scalar_isolated!( + sqlx::query_scalar!( + "SELECT flow_version.value as \"value!: sqlx::types::Json>\" FROM flow + LEFT JOIN flow_version + ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] + WHERE flow.path = $1 AND flow.workspace_id = $2", + &path, &workspace_id + ), + tx ) - .bind(&path) - .bind(&workspace_id), - tx - )? + } else { + fetch_scalar_isolated!( + sqlx::query_scalar!( + "SELECT coalesce(flow_version_lite.value, flow_version.value) as \"value!: sqlx::types::Json>\" FROM flow + LEFT JOIN flow_version + ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] + LEFT JOIN flow_version_lite + ON flow_version_lite.id = flow_version.id + WHERE flow.path = $1 AND flow.workspace_id = $2 LIMIT 1", + &path, &workspace_id + ), + tx + ) + }? .ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", path)))?; - let mut value = - serde_json::from_str::(value_json.value.get()).map_err(|err| { - Error::InternalErr(format!( - "could not convert json to flow for {path}: {err:?}" - )) - })?; + let mut value = serde_json::from_str::(value_json.get()).map_err(|err| { + Error::InternalErr(format!( + "could not convert json to flow for {path}: {err:?}" + )) + })?; let priority = value.priority; add_virtual_items_if_necessary(&mut value.modules); if same_worker { @@ -3936,6 +3967,7 @@ pub async fn push<'c, 'd>( JobKind::FlowDependencies => "jobs.run.flow_dependencies", JobKind::AppDependencies => "jobs.run.app_dependencies", JobKind::DeploymentCallback => "jobs.run.deployment_callback", + JobKind::FlowScript => "jobs.run.flow_script", }; let audit_author = if format!("u/{user}") != permissioned_as && user != permissioned_as { diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index 0a39a33610..5a7c9023c9 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -392,6 +392,44 @@ async fn spawn_dedicated_workers_for_flow( workers.push(dedi_w); } } + FlowModuleValue::FlowScript { id, language, .. } => { + let spawn = sqlx::query!( + "SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1 LIMIT 1", + id.0 + ) + .fetch_one(db) + .await + .map(|record| SpawnWorker::RawScript { + path: "".to_string(), + content: record.code, + lock: record.lock, + lang: language.clone(), + }); + match spawn { + Ok(spawn) => { + if let Some(dedi_w) = spawn_dedicated_worker( + spawn, + w_id, + killpill_tx.clone(), + killpill_rx, + db, + worker_dir, + base_internal_url, + worker_name, + job_completed_tx, + Some(module.id.clone()), + ) + .await + { + workers.push(dedi_w); + } + }, + Err(err) => tracing::error!( + "failed to get script for module: {:?}, err: {:?}", + module, err + ) + } + }, FlowModuleValue::Flow { .. } => (), FlowModuleValue::Identity => (), } @@ -422,10 +460,12 @@ pub async fn create_dedicated_worker_map( if let Some(flow_path) = _wp.path.strip_prefix("flow/") { is_flow_worker = true; let value = sqlx::query_scalar!( - "SELECT flow_version.value + "SELECT coalesce(flow_version_lite.value, flow_version.value) AS \"value!: sqlx::types::Json>\" FROM flow LEFT JOIN flow_version ON flow_version.id = flow.versions[array_upper(flow.versions, 1)] + LEFT JOIN flow_version_lite + ON flow_version_lite.id = flow_version.id WHERE flow.path = $1 AND flow.workspace_id = $2", flow_path, _wp.workspace_id @@ -434,7 +474,7 @@ pub async fn create_dedicated_worker_map( .await; if let Ok(v) = value { if let Some(v) = v { - let value = serde_json::from_value::(v).map_err(|err| { + let value = serde_json::from_str::(v.get()).map_err(|err| { Error::InternalErr(format!( "could not convert json to flow for {flow_path}: {err:?}" )) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index d453e56960..0ea0024969 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1862,7 +1862,11 @@ async fn handle_queued_job( let cached_res_path = if job.cache_ttl.is_some() { let version_hash = if let Some(h) = job.script_hash { - format!("script_{}", h.to_string()) + if matches!(job.job_kind, JobKind::FlowScript) { + format!("flowscript_{}", h.to_string()) + } else { + format!("script_{}", h.to_string()) + } } else if let Some(rc) = raw_code.as_ref() { use std::hash::Hasher; let mut s = DefaultHasher::new(); @@ -2273,6 +2277,22 @@ async fn handle_code_execution_job( ) .await? } + JobKind::FlowScript => { + let (lockfile, content) = sqlx::query!( + "SELECT lock, code AS \"code!: String\" FROM flow_node WHERE id = $1 LIMIT 1", + job.script_hash.unwrap_or(ScriptHash(0)).0 + ) + .fetch_one(db) + .await + .map(|record| (record.lock, record.code))?; + ContentReqLangEnvs { + content, + lockfile, + language: job.language.to_owned(), + envs: None, + codebase: None, + } + }, JobKind::DeploymentCallback => { get_script_content_by_path(job.script_path.clone(), &job.workspace_id, db).await? } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 9a5e734dd2..e87030eb91 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -2253,6 +2253,7 @@ async fn push_next_flow_job( Ok( FlowModuleValue::Script { input_transforms, .. } | FlowModuleValue::RawScript { input_transforms, .. } + | FlowModuleValue::FlowScript { input_transforms, .. } | FlowModuleValue::Flow { input_transforms, .. }, ) => { let ctx = get_transform_context(&flow_job, &previous_id, &status).await?; @@ -2994,10 +2995,10 @@ async fn compute_next_flow_transform( if is_skipped { return trivial_next_job(JobPayload::Identity); } - match &module.get_value()? { + match module.get_value()? { FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity), FlowModuleValue::Flow { path, .. } => { - let payload = flow_to_payload(path, &delete_after_use); + let payload = flow_to_payload(&path, &delete_after_use); Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(payload), NextStatus::NextStep, @@ -3005,7 +3006,7 @@ async fn compute_next_flow_transform( } FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => { let payload = - script_to_payload(script_hash, script_path, db, flow_job, module, tag_override) + script_to_payload(&script_hash, &script_path, db, flow_job, module, &tag_override) .await?; Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(payload), @@ -3036,14 +3037,14 @@ async fn compute_next_flow_transform( }); let payload = raw_script_to_payload( path, - content, - language, - lock, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + &content, + &language, + &lock, + &custom_concurrency_key, + &concurrent_limit, + &concurrency_time_window_s, module, - tag, + &tag, &delete_after_use, ); Ok(NextFlowTransform::Continue( @@ -3051,9 +3052,37 @@ async fn compute_next_flow_transform( NextStatus::NextStep, )) } + FlowModuleValue::FlowScript { + id, // flow_node(id). + tag, + language, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + .. + } => { + let payload = JobPayloadWithTag { + payload: JobPayload::FlowScript { + id, + language, + custom_concurrency_key: custom_concurrency_key.clone(), + concurrent_limit, + concurrency_time_window_s, + cache_ttl: module.cache_ttl.map(|x| x as i32), + dedicated_worker: None, + }, + tag: tag.clone(), + delete_after_use, + timeout: module.timeout, + }; + Ok(NextFlowTransform::Continue( + ContinuePayload::SingleJob(payload), + NextStatus::NextStep, + )) + }, FlowModuleValue::WhileloopFlow { modules, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg - let is_simple = is_simple_modules(modules, flow); + let is_simple = is_simple_modules(&modules, flow); let (flow_jobs, flow_jobs_success) = match status_module { FlowStatusModule::InProgress { flow_jobs: Some(flow_jobs), @@ -3077,7 +3106,7 @@ async fn compute_next_flow_transform( }, while_loop: true, }, - modules, + &modules, flow_job, is_simple, db, @@ -3089,12 +3118,13 @@ async fn compute_next_flow_transform( /* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */ FlowModuleValue::ForloopFlow { modules, iterator, parallel, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg - let is_simple = !parallel && is_simple_modules(modules, flow); + let is_simple = !parallel && is_simple_modules(&modules, flow); // if is_simple { // match value { // FlowModuleValue::Script { input_transforms, .. } // | FlowModuleValue::RawScript { input_transforms, .. } + // | FlowModuleValue::FlowScript { input_transforms, .. } // | FlowModuleValue::Flow { input_transforms, .. } => { // Some(input_transforms.clone()) // } @@ -3106,14 +3136,14 @@ async fn compute_next_flow_transform( flow_job, previous_id, status, - iterator, + &iterator, arc_last_job_result, resumes, resume, approvers, arc_flow_job_args, client, - parallel, + ¶llel, ) .await?; @@ -3124,7 +3154,7 @@ async fn compute_next_flow_transform( flow, status, ns, - modules, + &modules, flow_job, is_simple, db, @@ -3150,7 +3180,7 @@ async fn compute_next_flow_transform( let continue_payload = { let flow_value = FlowValue { - modules: (*modules).clone(), + modules, failure_module: flow.failure_module.clone(), same_worker: flow.same_worker, concurrent_limit: None, @@ -3181,6 +3211,7 @@ async fn compute_next_flow_transform( // match value { // FlowModuleValue::Script { input_transforms, .. } // | FlowModuleValue::RawScript { input_transforms, .. } + // | FlowModuleValue::FlowScript { input_transforms, .. } // | FlowModuleValue::Flow { input_transforms, .. } => { // Some(input_transforms.clone()) // } @@ -3281,7 +3312,7 @@ async fn compute_next_flow_transform( | FlowStatusModule::WaitingForExecutor { .. } => { if branches.is_empty() { return Ok(NextFlowTransform::EmptyInnerFlows); - } else if *parallel { + } else if parallel { return Ok(NextFlowTransform::Continue( ContinuePayload::BranchAllJobs( branches @@ -3346,7 +3377,7 @@ async fn compute_next_flow_transform( flow_jobs: Some(flow_jobs), flow_jobs_success, .. - } if !*parallel => ( + } if !parallel => ( BranchAllStatus { branch: branch + 1, len: len.clone() }, flow_jobs.clone(), flow_jobs_success.clone(), @@ -3439,6 +3470,7 @@ async fn next_loop_iteration( match value { FlowModuleValue::Script { input_transforms, .. } | FlowModuleValue::RawScript { input_transforms, .. } + | FlowModuleValue::FlowScript { input_transforms, .. } | FlowModuleValue::Flow { input_transforms, .. } => { Some(input_transforms.clone()) } diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 0de308337a..998488537b 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -8,7 +8,7 @@ use sqlx::types::Json; use uuid::Uuid; use windmill_common::error::Error; use windmill_common::error::Result; -use windmill_common::flows::{FlowModule, FlowModuleValue}; +use windmill_common::flows::{FlowModule, FlowModuleValue, FlowNodeId}; use windmill_common::get_latest_deployed_hash_for_path; use windmill_common::jobs::JobPayload; use windmill_common::scripts::ScriptHash; @@ -606,7 +606,9 @@ pub async fn handle_flow_dependency_job( occupancy_metrics, ) .await?; - let new_flow_value = serde_json::to_value(flow).map_err(to_anyhow)?; + let new_flow_value = sqlx::types::Json( + serde_json::value::to_raw_value(&flow).map_err(to_anyhow)? + ); // Re-check cancelation to ensure we don't accidentially override a flow. if sqlx::query_scalar!("SELECT canceled FROM queue WHERE id = $1", job.id) @@ -630,7 +632,7 @@ pub async fn handle_flow_dependency_job( sqlx::query!( "UPDATE flow SET value = $1 WHERE path = $2 AND workspace_id = $3", - new_flow_value, + &new_flow_value as &sqlx::types::Json>, job_path, job.workspace_id ) @@ -638,12 +640,23 @@ pub async fn handle_flow_dependency_job( .await?; sqlx::query!( "UPDATE flow_version SET value = $1 WHERE id = $2", - new_flow_value, + &new_flow_value as &sqlx::types::Json>, version ) .execute(db) .await?; + // Compute a lite version of the flow value (`RawScript` => `FlowScript`). + let mut value_lite = flow.clone(); + tx = reduce(tx, &mut value_lite.modules, &job_path, &job.workspace_id).await?; + sqlx::query!( + "INSERT INTO flow_version_lite (id, value) VALUES ($1, $2) + ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value", + version, sqlx::types::Json(to_raw_value(&value_lite)) as sqlx::types::Json>, + ) + .execute(db) + .await?; + tx.commit().await?; if let Err(e) = handle_deployment_metadata( @@ -913,7 +926,7 @@ async fn lock_modules<'c>( ) .await; // - match new_lock { + let lock = match new_lock { Ok(new_lock) => { let dep_path = path.clone().unwrap_or_else(|| job_path.to_string()); tx = clear_dependency_map_for_item( @@ -954,20 +967,7 @@ async fn lock_modules<'c>( language = ScriptLang::Bun; }; } - e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript { - lock: Some(new_lock), - path, - input_transforms, - content, - language, - tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, - is_trigger, - }); - new_flow_modules.push(e); - continue; + Some(new_lock) } Err(error) => { // TODO: Record flow raw script error lock logs @@ -977,24 +977,123 @@ async fn lock_modules<'c>( error = ?error, "Failed to generate flow lock for raw script" ); - e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript { - lock: None, - path, - input_transforms, + None + } + }; + e.value = windmill_common::worker::to_raw_value(&FlowModuleValue::RawScript { + lock, + path, + input_transforms, + content, + language, + tag, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + is_trigger, + }); + new_flow_modules.push(e); + continue; + } + Ok((new_flow_modules, tx, modified_ids)) +} + +async fn insert_flow_node<'c>( + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, + path: &str, + workspace_id: &str, + code: Option<&String>, + lock: Option<&String>, + flow: Option<&Json>>, +) -> Result<(sqlx::Transaction<'c, sqlx::Postgres>, FlowNodeId)> { + let hash = { + use std::hash::{DefaultHasher, Hasher, Hash}; + + let mut hasher = DefaultHasher::new(); + code.hash(&mut hasher); + lock.hash(&mut hasher); + flow.inspect(|flow| flow.get().hash(&mut hasher)); + hasher.finish() as i64 + }; + + // Insert the flow node if it doesn't exist. + let id = sqlx::query_scalar!( + r#" + WITH existing AS ( + SELECT id FROM flow_node + WHERE hash = $1 AND path = $2 AND workspace_id = $3 AND code = $4 AND lock = $5 AND flow = $6 + LIMIT 1 + ), + inserted AS ( + INSERT INTO flow_node (hash, path, workspace_id, code, lock, flow) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT DO NOTHING + RETURNING id + ) + SELECT id FROM existing + UNION ALL + SELECT id FROM inserted + "#, + hash, path, workspace_id, code, lock, flow as Option<&Json>> + ) + .fetch_one(&mut *tx) + .await? + .ok_or(error::Error::InternalErr("Failed to cache".to_string()))?; + Ok((tx, FlowNodeId(id))) +} + +async fn reduce<'c>( + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, + modules: &mut Vec, + path: &str, + workspace_id: &str, +) -> Result> { + use FlowModuleValue::*; + for module in &mut *modules { + let mut val = serde_json::from_str::(module.value.get()) + .map_err(|err| Error::InternalErr(format!("reduce: Failed to parse flow module value: {}", err)))?; + match &mut val { + RawScript { .. } => { + // In order to avoid an unnecessary `.clone()` of `val`, take ownership of it's content + // using `std::mem::replace`. + let RawScript { + lock, content, language, + input_transforms, tag, custom_concurrency_key, concurrent_limit, concurrency_time_window_s, is_trigger, - }); - new_flow_modules.push(e); - continue; + .. + } = std::mem::replace(&mut val, Identity) else { unreachable!() }; + let id; + (tx, id) = insert_flow_node(tx, path, workspace_id, Some(&content), lock.as_ref(), None).await?; + val = FlowScript { + input_transforms, + id, + tag, + language, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + is_trigger, + }; + }, + ForloopFlow { modules, .. } | WhileloopFlow { modules, .. } => { + tx = Box::pin(reduce(tx, &mut *modules, path, workspace_id)).await?; } + BranchOne { branches, .. } | BranchAll { branches, .. } => { + for branch in &mut *branches { + tx = Box::pin(reduce(tx, &mut branch.modules, path, workspace_id)).await?; + } + } + _ => {} } + module.value = to_raw_value(&val); } - Ok((new_flow_modules, tx, modified_ids)) + Ok(tx) } fn skip_creating_new_lock(language: &ScriptLang, content: &str) -> bool {