From ed806bf9d07de9f22c8a00260984e94eafd6ddf8 Mon Sep 17 00:00:00 2001 From: Pyra <92104930+pyranota@users.noreply.github.com> Date: Wed, 24 Sep 2025 15:40:39 +0200 Subject: [PATCH] fix(backend): rework `dependency_map` handling (#6598) * v0 Signed-off-by: pyranota * optimize relocks * make it work with relative relative imports Signed-off-by: pyranota * use fallback Signed-off-by: pyranota * remove dbg and todos Signed-off-by: pyranota * future proof a bit Signed-off-by: pyranota * cleanup Signed-off-by: pyranota * more cleanup Signed-off-by: pyranota * remove final TODO Signed-off-by: pyranota * do not use bytemuck Signed-off-by: pyranota * optimize hashing Signed-off-by: pyranota * implementation 1 Signed-off-by: pyranota * almost v0 Signed-off-by: pyranota * v0 Signed-off-by: pyranota * add comments and use fallback Signed-off-by: pyranota * call dissolve for apps Signed-off-by: pyranota * add comms Signed-off-by: pyranota * refactor v0 (partially tested + dirty) Signed-off-by: pyranota * finishing Signed-off-by: pyranota * remove TODO Signed-off-by: pyranota * Update SQLx metadata * silence unused argument Signed-off-by: pyranota * cleanup Signed-off-by: pyranota * implement rebuild_map endpoint Signed-off-by: pyranota * update windmill api client Signed-off-by: pyranota * almost finish with tests Signed-off-by: pyranota * add proper testing Signed-off-by: pyranota * remove unused fixtures Signed-off-by: pyranota * Update SQLx metadata * partial cleanup Signed-off-by: pyranota * Update backend/windmill-worker/src/scoped_dependency_map.rs Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com> * Update backend/windmill-common/src/scripts.rs Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com> * evil doings Signed-off-by: pyranota * more cleanup * Update SQLx metadata * more cleanup Signed-off-by: pyranota * fixing CI Signed-off-by: pyranota * remove python from default features Signed-off-by: pyranota --------- Signed-off-by: pyranota Co-authored-by: Pyra <92104930+pyranye@users.noreply.github.com> Co-authored-by: GitHub Action Co-authored-by: windmill-internal-app[bot] Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com> --- ...ca1805e3a1530fc64bf95eaee5b645e885251.json | 28 + ...e121305d63cfcd88f69700125d17fb2c56a1f.json | 46 ++ ...e6bb2782e1f8ef5f9fe752b356927a8605100.json | 22 + ...7418c16f96f307bafa926a18316055b7c90c4.json | 12 + ...46bee53a07c8d1264eeb44bc94233bc06bbfd.json | 16 - ...45bd1c24f93db32970d9c06623b1e335138ba.json | 28 + ...d67f36d1fd87fe4cf4f002e18367088c497c.json} | 6 +- ...e3670e50344f39b0d61eef8e60c2cdb57ae1f.json | 18 + ...0a4665a518da0560fd9ed7add05c66da3898f.json | 20 + ...f8067c1b43a97ea96845d92e2cbf85fbd6311.json | 28 + backend/Cargo.lock | 12 + backend/Cargo.toml | 2 +- backend/tests/fixtures/dependency_map.sql | 113 ++++ backend/tests/fixtures/relative_python.sql | 3 +- backend/tests/python_jobs.rs | 8 +- backend/tests/relative_imports.rs | 549 ++++++++++++++++++ backend/tests/worker.rs | 21 +- backend/windmill-api-client/src/codegen.rs | 222 ++++++- backend/windmill-api/openapi.yaml | 53 ++ backend/windmill-api/src/apps.rs | 2 +- backend/windmill-api/src/lib.rs | 3 +- backend/windmill-api/src/scripts.rs | 3 + backend/windmill-api/src/workspaces.rs | 39 ++ backend/windmill-common/src/apps.rs | 70 +++ backend/windmill-common/src/flows.rs | 53 +- backend/windmill-common/src/scripts.rs | 13 +- backend/windmill-worker/src/lib.rs | 1 + .../src/scoped_dependency_map.rs | 445 ++++++++++++++ .../windmill-worker/src/worker_lockfiles.rs | 421 +++++++------- 29 files changed, 1973 insertions(+), 284 deletions(-) create mode 100644 backend/.sqlx/query-094587579285fc5656f6104716dca1805e3a1530fc64bf95eaee5b645e885251.json create mode 100644 backend/.sqlx/query-13297889361ac6839d6c4bd0b8ae121305d63cfcd88f69700125d17fb2c56a1f.json create mode 100644 backend/.sqlx/query-1492b88c75722465b1a5c138729e6bb2782e1f8ef5f9fe752b356927a8605100.json create mode 100644 backend/.sqlx/query-3c5b6001aac7fb58ec9bfad1bfd7418c16f96f307bafa926a18316055b7c90c4.json delete mode 100644 backend/.sqlx/query-6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd.json create mode 100644 backend/.sqlx/query-79624ae15f909bd6ab4f015e32345bd1c24f93db32970d9c06623b1e335138ba.json rename backend/.sqlx/{query-958ed17dafffdd37e636ccd244dc4ca60cbf562e6f6a371d5f9a9943fb30254c.json => query-e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c.json} (68%) create mode 100644 backend/.sqlx/query-ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f.json create mode 100644 backend/.sqlx/query-f03d52c091d10d27a274cebaf370a4665a518da0560fd9ed7add05c66da3898f.json create mode 100644 backend/.sqlx/query-fd403acc343182fdab100263f8ef8067c1b43a97ea96845d92e2cbf85fbd6311.json create mode 100644 backend/tests/fixtures/dependency_map.sql create mode 100644 backend/tests/relative_imports.rs create mode 100644 backend/windmill-worker/src/scoped_dependency_map.rs diff --git a/backend/.sqlx/query-094587579285fc5656f6104716dca1805e3a1530fc64bf95eaee5b645e885251.json b/backend/.sqlx/query-094587579285fc5656f6104716dca1805e3a1530fc64bf95eaee5b645e885251.json new file mode 100644 index 0000000000..2c26de7440 --- /dev/null +++ b/backend/.sqlx/query-094587579285fc5656f6104716dca1805e3a1530fc64bf95eaee5b645e885251.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path, versions[array_upper(versions, 1)] as version FROM flow WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "version", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "094587579285fc5656f6104716dca1805e3a1530fc64bf95eaee5b645e885251" +} diff --git a/backend/.sqlx/query-13297889361ac6839d6c4bd0b8ae121305d63cfcd88f69700125d17fb2c56a1f.json b/backend/.sqlx/query-13297889361ac6839d6c4bd0b8ae121305d63cfcd88f69700125d17fb2c56a1f.json new file mode 100644 index 0000000000..3a17f37f37 --- /dev/null +++ b/backend/.sqlx/query-13297889361ac6839d6c4bd0b8ae121305d63cfcd88f69700125d17fb2c56a1f.json @@ -0,0 +1,46 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT workspace_id, importer_path, importer_kind::text, imported_path, importer_node_id\n FROM dependency_map WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "importer_path", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "importer_kind", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "imported_path", + "type_info": "Varchar" + }, + { + "ordinal": 4, + "name": "importer_node_id", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + null, + false, + false + ] + }, + "hash": "13297889361ac6839d6c4bd0b8ae121305d63cfcd88f69700125d17fb2c56a1f" +} diff --git a/backend/.sqlx/query-1492b88c75722465b1a5c138729e6bb2782e1f8ef5f9fe752b356927a8605100.json b/backend/.sqlx/query-1492b88c75722465b1a5c138729e6bb2782e1f8ef5f9fe752b356927a8605100.json new file mode 100644 index 0000000000..d5b29d83dd --- /dev/null +++ b/backend/.sqlx/query-1492b88c75722465b1a5c138729e6bb2782e1f8ef5f9fe752b356927a8605100.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value FROM app_version WHERE id = $1 LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "1492b88c75722465b1a5c138729e6bb2782e1f8ef5f9fe752b356927a8605100" +} diff --git a/backend/.sqlx/query-3c5b6001aac7fb58ec9bfad1bfd7418c16f96f307bafa926a18316055b7c90c4.json b/backend/.sqlx/query-3c5b6001aac7fb58ec9bfad1bfd7418c16f96f307bafa926a18316055b7c90c4.json new file mode 100644 index 0000000000..c6f9924347 --- /dev/null +++ b/backend/.sqlx/query-3c5b6001aac7fb58ec9bfad1bfd7418c16f96f307bafa926a18316055b7c90c4.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dependency_map WHERE workspace_id = 'test-workspace'", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "3c5b6001aac7fb58ec9bfad1bfd7418c16f96f307bafa926a18316055b7c90c4" +} diff --git a/backend/.sqlx/query-6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd.json b/backend/.sqlx/query-6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd.json deleted file mode 100644 index 511fec6586..0000000000 --- a/backend/.sqlx/query-6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM dependency_map\n WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND\n AND workspace_id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd" -} diff --git a/backend/.sqlx/query-79624ae15f909bd6ab4f015e32345bd1c24f93db32970d9c06623b1e335138ba.json b/backend/.sqlx/query-79624ae15f909bd6ab4f015e32345bd1c24f93db32970d9c06623b1e335138ba.json new file mode 100644 index 0000000000..459394a6cf --- /dev/null +++ b/backend/.sqlx/query-79624ae15f909bd6ab4f015e32345bd1c24f93db32970d9c06623b1e335138ba.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path, hash FROM script WHERE workspace_id = $1 AND archived = false", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "hash", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "79624ae15f909bd6ab4f015e32345bd1c24f93db32970d9c06623b1e335138ba" +} diff --git a/backend/.sqlx/query-958ed17dafffdd37e636ccd244dc4ca60cbf562e6f6a371d5f9a9943fb30254c.json b/backend/.sqlx/query-e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c.json similarity index 68% rename from backend/.sqlx/query-958ed17dafffdd37e636ccd244dc4ca60cbf562e6f6a371d5f9a9943fb30254c.json rename to backend/.sqlx/query-e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c.json index 31501b4094..915cf7416b 100644 --- a/backend/.sqlx/query-958ed17dafffdd37e636ccd244dc4ca60cbf562e6f6a371d5f9a9943fb30254c.json +++ b/backend/.sqlx/query-e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c.json @@ -1,18 +1,18 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id)\n VALUES ($1, $2, $4::text::IMPORTER_KIND, $3, $5) ON CONFLICT DO NOTHING", + "query": "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id)\n VALUES ($1, $2, $3::text::IMPORTER_KIND, $4, $5) ON CONFLICT DO NOTHING", "describe": { "columns": [], "parameters": { "Left": [ - "Varchar", "Varchar", "Varchar", "Text", + "Varchar", "Varchar" ] }, "nullable": [] }, - "hash": "958ed17dafffdd37e636ccd244dc4ca60cbf562e6f6a371d5f9a9943fb30254c" + "hash": "e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c" } diff --git a/backend/.sqlx/query-ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f.json b/backend/.sqlx/query-ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f.json new file mode 100644 index 0000000000..499e2e6730 --- /dev/null +++ b/backend/.sqlx/query-ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f.json @@ -0,0 +1,18 @@ +{ + "db_name": "PostgreSQL", + "query": "\n DELETE FROM dependency_map\n WHERE workspace_id = $1\n AND importer_path = $2\n AND importer_kind = $3::text::IMPORTER_KIND\n AND importer_node_id = $4\n AND imported_path = $5\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f" +} diff --git a/backend/.sqlx/query-f03d52c091d10d27a274cebaf370a4665a518da0560fd9ed7add05c66da3898f.json b/backend/.sqlx/query-f03d52c091d10d27a274cebaf370a4665a518da0560fd9ed7add05c66da3898f.json new file mode 100644 index 0000000000..c5297f1499 --- /dev/null +++ b/backend/.sqlx/query-f03d52c091d10d27a274cebaf370a4665a518da0560fd9ed7add05c66da3898f.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value::text FROM app_version WHERE id = 0 AND app_id = 2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value", + "type_info": "Text" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "f03d52c091d10d27a274cebaf370a4665a518da0560fd9ed7add05c66da3898f" +} diff --git a/backend/.sqlx/query-fd403acc343182fdab100263f8ef8067c1b43a97ea96845d92e2cbf85fbd6311.json b/backend/.sqlx/query-fd403acc343182fdab100263f8ef8067c1b43a97ea96845d92e2cbf85fbd6311.json new file mode 100644 index 0000000000..fbd0a96f98 --- /dev/null +++ b/backend/.sqlx/query-fd403acc343182fdab100263f8ef8067c1b43a97ea96845d92e2cbf85fbd6311.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path, versions[array_upper(versions, 1)] as version FROM app WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "version", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "fd403acc343182fdab100263f8ef8067c1b43a97ea96845d92e2cbf85fbd6311" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 9c3b85cf4f..aae82ac328 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1166,6 +1166,7 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", "axum-core", + "axum-macros", "bytes", "futures-util", "http 1.3.1", @@ -1214,6 +1215,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "axum-macros" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57d123550fa8d071b7255cb0cc04dc302baa6c8c4a79f55701552684d8399bce" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.106", +] + [[package]] name = "az" version = "1.2.1" diff --git a/backend/Cargo.toml b/backend/Cargo.toml index e852412f7c..9aa99ab062 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -207,7 +207,7 @@ reqwest-middleware = { version = "^0", features = ["json"] } bitflags = "2.9.4" memchr = "2.7.4" -axum = { version = "^0.7", features = ["multipart"] } +axum = { version = "^0.7", features = ["multipart", "macros"] } headers = "^0" hyper = { version = "^1", features = ["full"] } tokio = { version = "=1.46.1", features = ["full", "tracing", "time"] } diff --git a/backend/tests/fixtures/dependency_map.sql b/backend/tests/fixtures/dependency_map.sql new file mode 100644 index 0000000000..2c244b08df --- /dev/null +++ b/backend/tests/fixtures/dependency_map.sql @@ -0,0 +1,113 @@ +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'test-user', +' +def main(): + return "f/rel/leaf_1" +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +'', +'', +'f/rel/leaf_1', 333400, 'python3', ''); +-- Padded Hex: 0000000000051658 + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'test-user', +' +def main(): + return "f/rel/leaf_2" +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +'', +'', +'f/rel/leaf_2', 333401, 'python3', ''); +-- Padded Hex: 0000000000051659 + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'test-user', +' +from f.rel.leaf_1 import main as lf_1; + +def main(): + return lf_1(); +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +'', +'', +'f/rel/branch', 333402, 'python3', ''); +-- Padded Hex: 000000000005165A + +INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( +'test-workspace', +'test-user', +' +from f.rel.branch import main as br; +from f.rel.leaf_1 import main as lf_1; +from f.rel.leaf_2 import main as lf_2; + +def main(): + return [br(), lf_1(), lf_2]; +', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', +'', +'', +'f/rel/root_script', 333403, 'python3', ''); +-- Padded Hex: 000000000005165B + +INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES ( +'test-workspace', +'', +'', +'f/rel/root_flow', +'{1443253234253454}', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object","order":[]}', +$tag${"modules":[{"id":"nstep1","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"from f.rel.branch import main as br;\nfrom f.rel.leaf_1 import main as lf_1;\nfrom ..leaf_1 import main as lf_12;\nfrom ...rel.leaf_2 import main as lf_2;\n\ndef check():\n return [br(), lf_1(), lf_2(), lf_12()];\n\ndef main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}},{"id":"nstep2","value":{"type":"branchall","branches":[{"expr":"","modules":[{"id":"nstep2_1","value":{"lock":"{\n \"dependencies\": {}\n}\n//bun.lock\n","type":"rawscript","assets":[],"content":"// import * as wmill from \"windmill-client\"\n\nexport async function main(x: string) {\n return x\n}\n","language":"bun","input_transforms":{"x":{"type":"static","value":""}}}}],"parallel":true,"skip_failure":false},{"expr":"false","modules":[{"id":"nstep2_2","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"from f.rel.leaf_2 import main as lf_2;\n\ndef check():\n return [lf_2()];\n\ndef main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}}],"summary":"","parallel":true,"skip_failure":false}],"parallel":true},"summary":""},{"id":"nstep3","value":{"type":"branchone","default":[{"id":"nstep3_2","value":{"lock":"{\n \"dependencies\": {}\n}\n//bun.lock\n","type":"rawscript","assets":[],"content":"// import * as wmill from \"windmill-client\"\n\nexport async function main(x: string) {\n return x\n}\n","language":"bun","input_transforms":{"x":{"type":"static","value":""}}}}],"branches":[{"expr":"false","modules":[{"id":"nstep3_1","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"def main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}}],"summary":"","parallel":true,"skip_failure":true}]},"summary":""},{"id":"nstep4","value":{"type":"whileloopflow","modules":[{"id":"nstep4_1","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"from f.rel.branch import main as br;\n\ndef check():\n return [br()];\n\ndef main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}}],"skip_failures":false}},{"id":"nstep5","value":{"type":"forloopflow","modules":[{"id":"nstep5_1","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"from f.rel.branch import main as br;\nfrom f.rel.leaf_1 import main as lf_1;\nfrom ..leaf_1 import main as lf_12;\nfrom ...rel.leaf_2 import main as lf_2;\n\ndef check():\n return [br(), lf_1(), lf_2(), lf_12()];\n\ndef main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}}],"iterator":{"expr":"['dynamic or static array']","type":"javascript"},"parallel":false,"skip_failures":true}},{"id":"nstep_ai","value":{"type":"aiagent","tools":[{"id":"m","value":{"type":"rawscript","content":"import wmill;\n\ndef main(x):\n return x\n","language":"python3","input_transforms":{"x":{"type":"static"}}},"summary":"tool"}],"input_transforms":{"image":{"type":"static"},"provider":{"type":"static","value":{"kind":"openai"}},"output_type":{"type":"static","value":"text"},"temperature":{"type":"static"},"user_message":{"type":"static","value":""},"output_schema":{"type":"static"},"system_prompt":{"type":"static","value":""},"max_completion_tokens":{"type":"static"}}},"continue_on_error":false}],"failure_module":{"id":"failure","value":{"type":"rawscript","content":"from f.rel.leaf_2 import main as lf_2;\n\ndef check():\n return [lf_2()];\n\nimport os\n\ndef main(message: str, name: str, step_id: str):\n flow_id = os.environ.get(\"WM_ROOT_FLOW_JOB_ID\")\n print(\"message\", message)\n print(\"name\", name)\n print(\"step_id\", step_id)\n return { \"message\": message, \"flow_id\": flow_id, \"step_id\": step_id, \"recover\": False }","language":"python3","input_transforms":{"name":{"expr":"error.name","type":"javascript"},"message":{"expr":"error.message","type":"javascript"},"step_id":{"expr":"error.step_id","type":"javascript"}}}},"preprocessor_module":{"id":"preprocessor","value":{"type":"rawscript","content":"from f.rel.branch import main as br;\nfrom f.rel.leaf_1 import main as lf_1;\nfrom ..leaf_1 import main as lf_12;\nfrom ...rel.leaf_2 import main as lf_2;\n\ndef check():\n return [br(), lf_1(), lf_2(), lf_12()];\n\ndef preprocessor(event):\n return {\n # return the args to be passed to the runnable\n }\n","language":"python3","input_transforms":{"event":{"type":"static"}}}}}$tag$, +'system' +); + +INSERT INTO public.flow_version(id, workspace_id, path, schema, value, created_by) VALUES ( +1443253234253454, +'test-workspace', +'f/rel/root_flow', +'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object","order":[]}', +$tag${"modules":[{"id":"nstep1","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"from f.rel.branch import main as br;\nfrom f.rel.leaf_1 import main as lf_1;\nfrom ..leaf_1 import main as lf_12;\nfrom ...rel.leaf_2 import main as lf_2;\n\ndef check():\n return [br(), lf_1(), lf_2(), lf_12()];\n\ndef main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}},{"id":"nstep2","value":{"type":"branchall","branches":[{"expr":"","modules":[{"id":"nstep2_1","value":{"lock":"{\n \"dependencies\": {}\n}\n//bun.lock\n","type":"rawscript","assets":[],"content":"// import * as wmill from \"windmill-client\"\n\nexport async function main(x: string) {\n return x\n}\n","language":"bun","input_transforms":{"x":{"type":"static","value":""}}}}],"parallel":true,"skip_failure":false},{"expr":"false","modules":[{"id":"nstep2_2","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"from f.rel.leaf_2 import main as lf_2;\n\ndef check():\n return [lf_2()];\n\ndef main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}}],"summary":"","parallel":true,"skip_failure":false}],"parallel":true},"summary":""},{"id":"nstep3","value":{"type":"branchone","default":[{"id":"nstep3_2","value":{"lock":"{\n \"dependencies\": {}\n}\n//bun.lock\n","type":"rawscript","assets":[],"content":"// import * as wmill from \"windmill-client\"\n\nexport async function main(x: string) {\n return x\n}\n","language":"bun","input_transforms":{"x":{"type":"static","value":""}}}}],"branches":[{"expr":"false","modules":[{"id":"nstep3_1","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"def main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}}],"summary":"","parallel":true,"skip_failure":true}]},"summary":""},{"id":"nstep4","value":{"type":"whileloopflow","modules":[{"id":"nstep4_1","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"from f.rel.branch import main as br;\n\ndef check():\n return [br()];\n\ndef main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}}],"skip_failures":false}},{"id":"nstep5","value":{"type":"forloopflow","modules":[{"id":"nstep5_1","value":{"lock":"# py: 3.11\n","type":"rawscript","assets":[],"content":"from f.rel.branch import main as br;\nfrom f.rel.leaf_1 import main as lf_1;\nfrom ..leaf_1 import main as lf_12;\nfrom ...rel.leaf_2 import main as lf_2;\n\ndef check():\n return [br(), lf_1(), lf_2(), lf_12()];\n\ndef main(x: str):\n return x","language":"python3","input_transforms":{"x":{"type":"static","value":""}}}}],"iterator":{"expr":"['dynamic or static array']","type":"javascript"},"parallel":false,"skip_failures":true}},{"id":"nstep_ai","value":{"type":"aiagent","tools":[{"id":"m","value":{"type":"rawscript","content":"import wmill;\n\ndef main(x):\n return x\n","language":"python3","input_transforms":{"x":{"type":"static"}}},"summary":"tool"}],"input_transforms":{"image":{"type":"static"},"provider":{"type":"static","value":{"kind":"openai"}},"output_type":{"type":"static","value":"text"},"temperature":{"type":"static"},"user_message":{"type":"static","value":""},"output_schema":{"type":"static"},"system_prompt":{"type":"static","value":""},"max_completion_tokens":{"type":"static"}}},"continue_on_error":false}],"failure_module":{"id":"failure","value":{"type":"rawscript","content":"from f.rel.leaf_2 import main as lf_2;\n\ndef check():\n return [lf_2()];\n\nimport os\n\ndef main(message: str, name: str, step_id: str):\n flow_id = os.environ.get(\"WM_ROOT_FLOW_JOB_ID\")\n print(\"message\", message)\n print(\"name\", name)\n print(\"step_id\", step_id)\n return { \"message\": message, \"flow_id\": flow_id, \"step_id\": step_id, \"recover\": False }","language":"python3","input_transforms":{"name":{"expr":"error.name","type":"javascript"},"message":{"expr":"error.message","type":"javascript"},"step_id":{"expr":"error.step_id","type":"javascript"}}}},"preprocessor_module":{"id":"preprocessor","value":{"type":"rawscript","content":"from f.rel.branch import main as br;\nfrom f.rel.leaf_1 import main as lf_1;\nfrom ..leaf_1 import main as lf_12;\nfrom ...rel.leaf_2 import main as lf_2;\n\ndef check():\n return [br(), lf_1(), lf_2(), lf_12()];\n\ndef preprocessor(event):\n return {\n # return the args to be passed to the runnable\n }\n","language":"python3","input_transforms":{"event":{"type":"static"}}}}}$tag$, +'system' +); + +INSERT INTO public.app(id, workspace_id, path, versions, policy) VALUES ( +2, +'test-workspace', +'f/rel/root_app', +'{0}', +'{}' +); + +INSERT INTO public.app_version(id, app_id, value, created_by) VALUES ( +0, +2, +$tag${"grid":[{"3":{"h":2,"w":6,"x":0,"y":0,"fixed":true,"fullHeight":false},"12":{"h":2,"w":12,"x":0,"y":0,"fixed":true,"fullHeight":false},"id":"topbar","data":{"id":"topbar","type":"containercomponent","customCss":{"container":{"class":"!p-0","style":""}},"configuration":{},"numberOfSubgrids":1}},{"3":{"h":8,"w":2,"x":0,"y":2,"fixed":false,"fullHeight":false},"12":{"h":2,"w":6,"x":0,"y":2,"fixed":false,"fullHeight":false},"id":"a","data":{"id":"a","type":"containercomponent","customCss":{"container":{"class":"","style":""}},"configuration":{},"numberOfSubgrids":1}},{"3":{"h":1,"w":1,"x":2,"y":2,"fixed":false,"fullHeight":false},"12":{"h":1,"w":2,"x":6,"y":2,"fixed":false,"fullHeight":false},"id":"dontpressmeplz","data":{"id":"dontpressmeplz","type":"buttoncomponent","customCss":{"button":{"class":"","style":""},"container":{"class":"","style":""}},"recomputeIds":[],"configuration":{"size":{"type":"static","value":"xs"},"color":{"type":"static","value":"blue"},"label":{"type":"static","value":"Press me"},"onError":{"type":"oneOf","selected":"errorOverlay","configuration":{"open":{"id":{"type":"static","value":""}},"close":{"id":{"type":"static","value":""}},"setTab":{"setTab":{"type":"static","value":[]}},"gotoUrl":{"url":{"type":"static","value":""},"newTab":{"type":"static","value":true}},"errorOverlay":{},"sendErrorToast":{"message":{"type":"static","value":"An error occured"},"appendError":{"type":"static","value":true}}}},"disabled":{"type":"static","value":false},"afterIcon":{"type":"static"},"onSuccess":{"type":"oneOf","selected":"none","configuration":{"none":{},"open":{"id":{"type":"static","value":""}},"close":{"id":{"type":"static","value":""}},"setTab":{"setTab":{"type":"static","value":[]}},"gotoUrl":{"url":{"type":"static","value":""},"newTab":{"type":"static","value":true}},"openModal":{"modalId":{"type":"static","value":""}},"sendToast":{"message":{"type":"static","value":""}},"clearFiles":{"id":{"type":"static","value":""}},"closeModal":{"modalId":{"type":"static","value":""}}}},"beforeIcon":{"type":"static"},"fillContainer":{"type":"static","value":false},"triggerOnAppLoad":{"type":"static","value":false},"confirmationModal":{"type":"oneOf","selected":"none","configuration":{"none":{},"confirmationModal":{"title":{"type":"static","value":"Title"},"description":{"type":"static","value":"Are you sure?"},"confirmationText":{"type":"static","value":"Confirm"}}}}},"componentInput":{"type":"runnable","fields":{"x":{"type":"static","value":null,"fieldType":"string"}},"runnable":{"name":"Inline Script","type":"runnableByName","inlineScript":{"path":"u/admin@windmill.dev/newapp/Inline_Script","schema":{"type":"object","$schema":"https://json-schema.org/draft/2020-12/schema","required":["x"],"properties":{"x":{"default":null,"description":"","originalType":"string","type":"string"}}},"content":"from f.rel.leaf_2 import main as lf_2;\n\ndef check():\n return [lf_2()];\n \ndef main(x: str):\n return x","language":"python3"}},"fieldType":"any","autoRefresh":false,"recomputeOnInputChanged":false},"verticalAlignment":"center","horizontalAlignment":"center"}},{"3":{"h":1,"w":1,"x":2,"y":3,"fixed":false,"fullHeight":false},"12":{"h":1,"w":2,"x":8,"y":2,"fixed":false,"fullHeight":false},"id":"d","data":{"id":"d","type":"checkboxcomponent","customCss":{"text":{"class":"","style":""},"container":{"class":"","style":""}},"recomputeIds":[],"configuration":{"label":{"type":"static","value":"Label"},"disabled":{"type":"static","value":false},"defaultValue":{"type":"static","value":false}},"verticalAlignment":"center","horizontalAlignment":"center"}},{"3":{"h":1,"w":1,"x":2,"y":4,"fixed":false,"fullHeight":false},"12":{"h":1,"w":2,"x":6,"y":3,"fixed":false,"fullHeight":false},"id":"youcanpressme","data":{"id":"youcanpressme","type":"buttoncomponent","customCss":{"button":{"class":"","style":""},"container":{"class":"","style":""}},"recomputeIds":[],"configuration":{"size":{"type":"static","value":"xs"},"color":{"type":"static","value":"blue"},"label":{"type":"static","value":"Press me"},"onError":{"type":"oneOf","selected":"errorOverlay","configuration":{"open":{"id":{"type":"static","value":""}},"close":{"id":{"type":"static","value":""}},"setTab":{"setTab":{"type":"static","value":[]}},"gotoUrl":{"url":{"type":"static","value":""},"newTab":{"type":"static","value":true}},"errorOverlay":{},"sendErrorToast":{"message":{"type":"static","value":"An error occured"},"appendError":{"type":"static","value":true}}}},"disabled":{"type":"static","value":false},"afterIcon":{"type":"static"},"onSuccess":{"type":"oneOf","selected":"none","configuration":{"none":{},"open":{"id":{"type":"static","value":""}},"close":{"id":{"type":"static","value":""}},"setTab":{"setTab":{"type":"static","value":[]}},"gotoUrl":{"url":{"type":"static","value":""},"newTab":{"type":"static","value":true}},"openModal":{"modalId":{"type":"static","value":""}},"sendToast":{"message":{"type":"static","value":""}},"clearFiles":{"id":{"type":"static","value":""}},"closeModal":{"modalId":{"type":"static","value":""}}}},"beforeIcon":{"type":"static"},"fillContainer":{"type":"static","value":false},"triggerOnAppLoad":{"type":"static","value":false},"confirmationModal":{"type":"oneOf","selected":"none","configuration":{"none":{},"confirmationModal":{"title":{"type":"static","value":"Title"},"description":{"type":"static","value":"Are you sure?"},"confirmationText":{"type":"static","value":"Confirm"}}}}},"componentInput":{"type":"runnable","fields":{"x":{"type":"static","value":null,"fieldType":"string"}},"runnable":{"name":"Inline Script","type":"runnableByName","inlineScript":{"path":"u/admin/easy_to_use_app/Inline_Script","schema":{"type":"object","$schema":"https://json-schema.org/draft/2020-12/schema","required":["x"],"properties":{"x":{"default":null,"description":"","originalType":"string","type":"string"}}},"content":"from f.rel.branch import main as br;\n\ndef check():\n return [br()];\n\ndef main(x: str):\n return x","language":"python3"}},"fieldType":"any","autoRefresh":false,"recomputeOnInputChanged":false},"verticalAlignment":"center","horizontalAlignment":"center"}}],"theme":{"path":"f/app_themes/theme_0","type":"path"},"subgrids":{"a-0":[{"3":{"h":1,"w":1,"x":0,"y":0,"fixed":false,"fullHeight":false},"12":{"h":2,"w":5,"x":0,"y":0,"fixed":false,"fullHeight":false},"id":"pressmeplz","data":{"id":"pressmeplz","type":"buttoncomponent","customCss":{"button":{"class":"","style":""},"container":{"class":"","style":""}},"recomputeIds":[],"configuration":{"size":{"type":"static","value":"xs"},"color":{"type":"static","value":"blue"},"label":{"type":"static","value":"Press me"},"onError":{"type":"oneOf","selected":"errorOverlay","configuration":{"open":{"id":{"type":"static","value":""}},"close":{"id":{"type":"static","value":""}},"setTab":{"setTab":{"type":"static","value":[]}},"gotoUrl":{"url":{"type":"static","value":""},"newTab":{"type":"static","value":true}},"errorOverlay":{},"sendErrorToast":{"message":{"type":"static","value":"An error occured"},"appendError":{"type":"static","value":true}}}},"disabled":{"type":"static","value":false},"afterIcon":{"type":"static"},"onSuccess":{"type":"oneOf","selected":"none","configuration":{"none":{},"open":{"id":{"type":"static","value":""}},"close":{"id":{"type":"static","value":""}},"setTab":{"setTab":{"type":"static","value":[]}},"gotoUrl":{"url":{"type":"static","value":""},"newTab":{"type":"static","value":true}},"openModal":{"modalId":{"type":"static","value":""}},"sendToast":{"message":{"type":"static","value":""}},"clearFiles":{"id":{"type":"static","value":""}},"closeModal":{"modalId":{"type":"static","value":""}}}},"beforeIcon":{"type":"static"},"fillContainer":{"type":"static","value":false},"triggerOnAppLoad":{"type":"static","value":false},"confirmationModal":{"type":"oneOf","selected":"none","configuration":{"none":{},"confirmationModal":{"title":{"type":"static","value":"Title"},"description":{"type":"static","value":"Are you sure?"},"confirmationText":{"type":"static","value":"Confirm"}}}}},"componentInput":{"type":"runnable","fields":{"x":{"type":"static","value":null,"fieldType":"string"}},"runnable":{"name":"Inline Script","type":"runnableByName","inlineScript":{"path":"u/admin@windmill.dev/newapp/Inline_Script","schema":{"type":"object","$schema":"https://json-schema.org/draft/2020-12/schema","required":["x"],"properties":{"x":{"default":null,"description":"","originalType":"string","type":"string"}}},"content":"from f.rel.branch import main as br;\nfrom f.rel.leaf_1 import main as lf_1;\nfrom ..leaf_1 import main as lf_12;\nfrom ...rel.leaf_2 import main as lf_2;\n\ndef check():\n return [br(), lf_1(), lf_2(), lf_12()];\n\n\ndef main(x: str):\n return x","language":"python3"}},"fieldType":"any","autoRefresh":false,"recomputeOnInputChanged":false},"verticalAlignment":"center","horizontalAlignment":"center"}}],"topbar-0":[{"3":{"h":1,"w":6,"x":0,"y":0,"fixed":false,"fullHeight":false},"12":{"h":1,"w":6,"x":0,"y":0,"fixed":false,"fullHeight":false},"id":"title","data":{"id":"title","type":"textcomponent","customCss":{"text":{"class":"text-xl font-semibold whitespace-nowrap truncate","style":""},"container":{"class":"","style":""}},"configuration":{"style":{"type":"static","value":"Body"},"tooltip":{"expr":"`Author: ${ctx.author}`","type":"evalv2","value":"","fieldType":"text","connections":[{"id":"author","componentId":"ctx"}]},"copyButton":{"type":"static","value":false},"disableNoText":{"type":"static","value":true,"fieldType":"boolean"}},"componentInput":{"eval":"${ctx.summary}","type":"templatev2","fieldType":"template","connections":[{"id":"summary","componentId":"ctx"}]},"verticalAlignment":"center","horizontalAlignment":"left"}},{"3":{"h":1,"w":3,"x":0,"y":1,"fixed":false,"fullHeight":false},"12":{"h":1,"w":6,"x":6,"y":0,"fixed":false,"fullHeight":false},"id":"recomputeall","data":{"id":"recomputeall","type":"recomputeallcomponent","customCss":{"container":{"class":"","style":""}},"menuItems":[],"configuration":{"defaultRefreshInterval":{"type":"static","value":"0"}},"verticalAlignment":"center","horizontalAlignment":"right"}}]},"fullscreen":false,"norefreshbar":false,"hideLegacyTopBar":true,"hiddenInlineScripts":[],"unusedInlineScripts":[],"mobileViewOnSmallerScreens":false}$tag$, +'system' +); + +-- Prebuild dependency_map +-- It would be done by Windmill, but this one is static. +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/branch', 'script', 'f/rel/leaf_1', ''); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_script', 'script', 'f/rel/branch', ''); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_script', 'script', 'f/rel/leaf_1', ''); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_script', 'script', 'f/rel/leaf_2', ''); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_app', 'app', 'f/rel/leaf_2', 'dontpressmeplz'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_flow', 'flow', 'f/rel/branch', 'nstep1'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_flow', 'flow', 'f/rel/leaf_1', 'nstep1'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_flow', 'flow', 'f/rel/leaf_2', 'nstep1'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_flow', 'flow', 'f/rel/leaf_2', 'nstep2_2'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_flow', 'flow', 'f/rel/branch', 'nstep4_1'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_flow', 'flow', 'f/rel/branch', 'nstep5_1'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_flow', 'flow', 'f/rel/leaf_1', 'nstep5_1'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_flow', 'flow', 'f/rel/leaf_2', 'nstep5_1'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_app', 'app', 'f/rel/branch', 'pressmeplz'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_app', 'app', 'f/rel/leaf_1', 'pressmeplz'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_app', 'app', 'f/rel/leaf_2', 'pressmeplz'); +INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/rel/root_app', 'app', 'f/rel/branch', 'youcanpressme'); + diff --git a/backend/tests/fixtures/relative_python.sql b/backend/tests/fixtures/relative_python.sql index 05e8453ccc..7a70e06d20 100644 --- a/backend/tests/fixtures/relative_python.sql +++ b/backend/tests/fixtures/relative_python.sql @@ -22,7 +22,6 @@ def main(): '', 'f/system_relative/different_folder_script', 12347, 'python3', ''); - INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( 'test-workspace', 'test-user', @@ -38,4 +37,4 @@ def main(): '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', '', '', -'f/system_relative/nested_script', 12348, 'python3', ''); \ No newline at end of file +'f/system_relative/nested_script', 12348, 'python3', ''); diff --git a/backend/tests/python_jobs.rs b/backend/tests/python_jobs.rs index c13c5c7876..cc7d1cb5b9 100644 --- a/backend/tests/python_jobs.rs +++ b/backend/tests/python_jobs.rs @@ -1,8 +1,8 @@ mod common; use crate::common::*; -use sqlx::Pool; use sqlx::postgres::Postgres; -use windmill_common::scripts::{ ScriptLang}; +use sqlx::Pool; +use windmill_common::scripts::ScriptLang; #[cfg(feature = "python")] #[sqlx::test(fixtures("base", "lockfile_python"))] @@ -162,7 +162,6 @@ use windmill_common::jobs::RawCode; #[cfg(feature = "python")] #[sqlx::test(fixtures("base"))] async fn test_python_job(db: Pool) -> anyhow::Result<()> { - initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); @@ -211,7 +210,6 @@ async fn test_python_global_site_packages(db: Pool) -> anyhow::Result< // 3.12 { - let content = r#"# py: ==3.12 #requirements: # @@ -355,7 +353,6 @@ def main(): Ok(()) } - #[cfg(feature = "python")] #[sqlx::test(fixtures("base", "relative_python"))] async fn test_relative_imports_python(db: Pool) -> anyhow::Result<()> { @@ -391,4 +388,3 @@ def main(): run_preview_relative_imports(&db, content, ScriptLang::Python3).await?; Ok(()) } - diff --git a/backend/tests/relative_imports.rs b/backend/tests/relative_imports.rs new file mode 100644 index 0000000000..91af80918b --- /dev/null +++ b/backend/tests/relative_imports.rs @@ -0,0 +1,549 @@ +// TODO: move all related logic here (if anything left anywhere in codebase) +mod common; +mod dependency_map { + use sqlx::{Pool, Postgres}; + use tokio_stream::StreamExt; + use windmill_api_client::types::NewScript; + + use crate::common::{in_test_worker, listen_for_completed_jobs, ApiServer}; + + pub async fn initialize_tracing() { + use std::sync::Once; + + static ONCE: Once = Once::new(); + ONCE.call_once(|| { + let _ = windmill_common::tracing_init::initialize_tracing( + "test", + &windmill_common::utils::Mode::Standalone, + "test", + ); + }); + } + + async fn rebuild_dmap(client: &windmill_api_client::Client) -> bool { + client + .client() + .post(format!( + "{}/w/test-workspace/workspaces/rebuild_dependency_map", + client.baseurl() + )) + .send() + .await + .unwrap() + .status() + .is_success() + } + + async fn init(db: Pool) -> (windmill_api_client::Client, u16, ApiServer) { + initialize_tracing().await; + let server = ApiServer::start(db).await.unwrap(); + let port = server.addr.port(); + let client = windmill_api_client::create_client( + &format!("http://localhost:{port}"), + "SECRET_TOKEN".to_string(), + ); + (client, port, server) + } + + async fn _clear_dmap(db: &Pool) { + sqlx::query!("DELETE FROM dependency_map WHERE workspace_id = 'test-workspace'") + .execute(db) + .await + .unwrap(); + } + + /// Corrects map according to provided replacements. + /// Only changes importer_path and/or id + /// Does not affect imported_path nor kind! + fn corrected_dmap(replacements: Vec<(&str, &str)>) -> Vec<(String, String, String, String)> { + CORRECT_DMAP + .clone() + .into_iter() + .map(|e| { + let mut r = ( + e.0.to_owned(), + e.1.to_owned(), + e.2.to_owned(), + e.3.to_owned(), + ); + for (from, to) in &replacements { + r = ( + r.0.replace(from, to), + r.1, // Kind should be immutable + r.2, // Imported path should be immutable + // We do not modify script contents in test, so we can assume scripts always import the same path + // Modification of kind or imported path considered to be incorrect. + r.3.replace(from, to), + ); + } + r + }) + .collect() + } + + async fn assert_dmap( + db: &Pool, + importer: Option, + expected: Vec<( + impl Into, + impl Into, + impl Into, + impl Into, + )>, + ) { + let dmap = sqlx::query_as::<_, (String, String, String, String)>( + "SELECT importer_path, importer_kind::text, imported_path, importer_node_id FROM dependency_map WHERE workspace_id = 'test-workspace' AND ($1::text IS NULL OR importer_path = $1::text)", + ) + .bind(importer) + .fetch_all(db) + .await + .unwrap(); + + assert_eq!( + dmap, + expected + .into_iter() + .map(|(f, s, t, fo)| (f.into(), s.into(), t.into(), fo.into())) + .collect::>() + ); + } + + fn quick_ns( + content: &str, + language: windmill_api_client::types::ScriptLang, + path: &str, + lock: Option, + parent_hash: Option, + ) -> NewScript { + NewScript { + content: content.into(), + language, + lock, + parent_hash, + path: path.into(), + concurrent_limit: None, + concurrency_time_window_s: None, + cache_ttl: None, + dedicated_worker: None, + description: "".to_string(), + draft_only: None, + envs: vec![], + is_template: None, + kind: None, + summary: "".to_string(), + tag: None, + schema: std::collections::HashMap::new(), + ws_error_handler_muted: Some(false), + priority: None, + delete_after_use: None, + timeout: None, + restart_unless_cancelled: None, + deployment_message: None, + concurrency_key: None, + visible_to_runner_only: None, + no_main_func: None, + codebase: None, + has_preprocessor: None, + on_behalf_of_email: None, + assets: vec![], + } + } + + lazy_static::lazy_static! { + pub static ref CORRECT_DMAP: Vec<(&'static str, &'static str, &'static str, &'static str)> = vec![ + ("f/rel/branch", "script", "f/rel/leaf_1", ""), + ("f/rel/root_script", "script", "f/rel/branch", ""), + ("f/rel/root_script", "script", "f/rel/leaf_1", ""), + ("f/rel/root_script", "script", "f/rel/leaf_2", ""), + ("f/rel/root_app", "app", "f/rel/leaf_2", "dontpressmeplz"), + ("f/rel/root_flow", "flow", "f/rel/branch", "nstep1"), + ("f/rel/root_flow", "flow", "f/rel/leaf_1", "nstep1"), + ("f/rel/root_flow", "flow", "f/rel/leaf_2", "nstep1"), + ("f/rel/root_flow", "flow", "f/rel/leaf_2", "nstep2_2"), + ("f/rel/root_flow", "flow", "f/rel/branch", "nstep4_1"), + ("f/rel/root_flow", "flow", "f/rel/branch", "nstep5_1"), + ("f/rel/root_flow", "flow", "f/rel/leaf_1", "nstep5_1"), + ("f/rel/root_flow", "flow", "f/rel/leaf_2", "nstep5_1"), + ("f/rel/root_app", "app", "f/rel/branch", "pressmeplz"), + ("f/rel/root_app", "app", "f/rel/leaf_1", "pressmeplz"), + ("f/rel/root_app", "app", "f/rel/leaf_2", "pressmeplz"), + ("f/rel/root_app", "app", "f/rel/branch", "youcanpressme")]; + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_rebuild_correctness(db: Pool) -> anyhow::Result<()> { + let (client, _port, _s) = init(db.clone()).await; + assert_dmap(&db, None, CORRECT_DMAP.clone()).await; + // rebuild map + assert!(rebuild_dmap(&client).await); + assert_dmap(&db, None, CORRECT_DMAP.clone()).await; + Ok(()) + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_rebuild_lock(db: Pool) -> anyhow::Result<()> { + let (client, _port, _s) = init(db.clone()).await; + + // Spawn first rebuild + let handle = { + let client = client.clone(); + tokio::spawn(async move { rebuild_dmap(&client).await }) + }; + + // Immidiately spawn another + let res = client + .client() + .post(format!( + "{}/w/test-workspace/workspaces/rebuild_dependency_map", + client.baseurl() + )) + .send() + .await + .unwrap() + .text() + .await + .unwrap(); + + // Should tell us there is already rebuilt in progress + // Or if it is too fast we will be able to trigger it second time + assert!(&res == "There is already one task pending, try again later." || &res == "Success"); + + assert!(handle.await.unwrap()); + Ok(()) + } + + // If you deploy from cli and you use raw requirements you don't want the script be included in dmap + // Otherwise script will be overwritten once any relative import is updated + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_with_requirements_txt(db: Pool) -> anyhow::Result<()> { + let (client, _port, _s) = init(db.clone()).await; + + client + .create_script( + "test-workspace", + &quick_ns( + " +from f.rel.branch import main as br; +from f.rel.leaf_1 import main as lf_1; +from f.rel.leaf_2 import main as lf_2; + +def main(): + return [br(), lf_1(), lf_2]; + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/root_script", + Some(format!("# from requirements.txt")), + Some("000000000005165B".into()), + ), + ) + .await + .unwrap(); + + assert_dmap( + &db, + Some("f/rel/root_script".into()), + vec![ + ("f/rel/root_script", "script", "f/rel/branch", ""), + ("f/rel/root_script", "script", "f/rel/leaf_1", ""), + ("f/rel/root_script", "script", "f/rel/leaf_2", ""), + ], + ) + .await; + + tokio::time::sleep(std::time::Duration::from_secs(13)).await; + + assert_dmap( + &db, + Some("f/rel/root_script".into()), + Vec::<(String, String, String, String)>::new(), + ) + .await; + + Ok(()) + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_without_requirements_txt( + db: Pool, + ) -> anyhow::Result<()> { + let (client, _port, _s) = init(db.clone()).await; + + client + .create_script( + "test-workspace", + &quick_ns( + " +from f.rel.branch import main as br; +from f.rel.leaf_1 import main as lf_1; +from f.rel.leaf_2 import main as lf_2; + +def main(): + return [br(), lf_1(), lf_2]; + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/root_script", + // We still want to pass lock to it. + Some(format!("# py311")), + Some("000000000005165B".into()), + ), + ) + .await + .unwrap(); + assert_dmap(&db, None, CORRECT_DMAP.clone()).await; + tokio::time::sleep(std::time::Duration::from_secs(13)).await; + assert_dmap(&db, None, CORRECT_DMAP.clone()).await; + Ok(()) + } + // Consider simple one. Only referenced directly. No deep connections + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_rename_leaf_2(db: Pool) -> anyhow::Result<()> { + let (client, port, _s) = init(db.clone()).await; + client + .create_script( + "test-workspace", + &quick_ns( + " +def main(): + return 'leaf3'; + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/leaf_2_renamed", + None, + Some("0000000000051659".into()), + ), + ) + .await + .unwrap(); + + let mut completed = listen_for_completed_jobs(&db).await; + in_test_worker(&db, completed.next(), port).await; + + // Changing leafs should not change dependency map + assert_dmap(&db, None, CORRECT_DMAP.clone()).await; + Ok(()) + } + + // Consider hard one. Referenced deeply and exists in double references. + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_rename_leaf_1(db: Pool) -> anyhow::Result<()> { + let (client, port, _s) = init(db.clone()).await; + client + .create_script( + "test-workspace", + &quick_ns( + " +def main(): + return 'leaf1'; + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/leaf_1_renamed", + None, + Some("0000000000051658".into()), + ), + ) + .await + .unwrap(); + + let mut completed = listen_for_completed_jobs(&db).await; + in_test_worker(&db, completed.next(), port).await; + + // Changing leafs should not change dependency map + assert_dmap(&db, None, CORRECT_DMAP.clone()).await; + Ok(()) + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_rename_branch(db: Pool) -> anyhow::Result<()> { + let (client, port, _s) = init(db.clone()).await; + client + .create_script( + "test-workspace", + &quick_ns( + " +from f.rel.leaf_1 import main as lf_1; + +def main(): + return lf_1(); + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/branch_renamed", + None, + Some("000000000005165A".into()), + ), + ) + .await + .unwrap(); + + let mut completed = listen_for_completed_jobs(&db).await; + in_test_worker(&db, completed.next(), port).await; + + // Changing branches SHOULD change dependency map + // Though it should only change branch item in dmap when it is importer. + // All entries when branch is imported should not change. + let mut corrected_dmap = CORRECT_DMAP.clone(); + // Corresponds to importer path of branch entry + corrected_dmap[0].0 = "f/rel/branch_renamed"; + assert_dmap(&db, None, corrected_dmap).await; + Ok(()) + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_rename_primary_script(db: Pool) -> anyhow::Result<()> { + let (client, port, _s) = init(db.clone()).await; + + client + .create_script( + "test-workspace", + &quick_ns( + " +from f.rel.branch import main as br; +from f.rel.leaf_1 import main as lf_1; +from f.rel.leaf_2 import main as lf_2; + +def main(): + return [br(), lf_1(), lf_2]; + ", + windmill_api_client::types::ScriptLang::Python3, + "f/rel/root_script_renamed", + None, + Some("000000000005165B".into()), + ), + ) + .await + .unwrap(); + + let corrected_dmap = corrected_dmap(vec![("root_script", "root_script_renamed")]); + let mut completed = listen_for_completed_jobs(&db).await; + in_test_worker(&db, completed.next(), port).await; + assert_dmap(&db, None, corrected_dmap.clone()).await; + Ok(()) + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_rename_primary_flow(db: Pool) -> anyhow::Result<()> { + use windmill_common::{cache::flow::fetch_version, flows::NewFlow}; + + let (client, port, _s) = init(db.clone()).await; + let flow = fetch_version(&db, 1443253234253454).await.unwrap(); + let res = client + .client() + .post(format!( + "{}/w/test-workspace/flows/update/{}", + client.baseurl(), + "f/rel/root_flow" // encode_path() + )) + .json(&NewFlow { + path: "f/rel/root_flow_renamed".into(), + summary: "".into(), + description: None, + value: serde_json::from_str( + &serde_json::to_string(flow.value()) + .unwrap() + .replace("nstep1", "Foxes") + .replace("nstep2_2", "like") + .replace("nstep_4_1", "Emeralds"), + ) + .unwrap(), + schema: None, + draft_only: None, + tag: None, + dedicated_worker: None, + timeout: None, + deployment_message: None, + visible_to_runner_only: None, + on_behalf_of_email: None, + }) + .send() + .await + .unwrap(); + + assert_eq!(res.text().await.unwrap(), "f/rel/root_flow_renamed"); + + let mut completed = listen_for_completed_jobs(&db).await; + in_test_worker(&db, completed.next(), port).await; + + assert_dmap( + &db, + None, + corrected_dmap(vec![ + ("f/rel/root_flow", "f/rel/root_flow_renamed"), + ("nstep1", "Foxes"), + ("nstep2_2", "like"), + ("nstep_4_1", "Emeralds"), + ]), + ) + .await; + Ok(()) + } + + #[cfg(feature = "python")] + #[sqlx::test(fixtures("base", "dependency_map"))] + async fn relative_imports_test_rename_primary_app(db: Pool) -> anyhow::Result<()> { + let (client, port, _s) = init(db.clone()).await; + + let app_value: String = + sqlx::query_scalar!("SELECT value::text FROM app_version WHERE id = 0 AND app_id = 2") + .fetch_one(&db) + .await + .unwrap() + .unwrap(); + + // TODO: There is: + // 1. update app + // 2. create app + // 3. update app raw + // Ideally all of them should be handled + let res = client + .client() + .post(format!( + "{}/w/test-workspace/apps/update/{}", + client.baseurl(), + "f/rel/root_app" // encode_path() + )) + .json(&windmill_api::EditApp { + path: Some("f/rel/root_app_renamed".into()), + summary: None, + value: serde_json::from_str( + &app_value + .replace("dontpressmeplz", "Apps") + .replace("youcanpressme", "Work"), + ) + .unwrap(), + policy: None, + deployment_message: None, + custom_path: None, + }) + .send() + .await + .unwrap(); + + assert_eq!( + res.text().await.unwrap(), + "app f/rel/root_app updated (npath: \"f/rel/root_app_renamed\")" + ); + + let mut completed = listen_for_completed_jobs(&db).await; + in_test_worker(&db, completed.next(), port).await; + + assert_dmap( + &db, + None, + corrected_dmap(vec![ + ("f/rel/root_app", "f/rel/root_app_renamed"), + ("dontpressmeplz", "Apps"), + ("youcanpressme", "Work"), + ]), + ) + .await; + Ok(()) + } +} diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 379c92606b..979ec4ce0e 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1,6 +1,5 @@ use serde::de::DeserializeOwned; - #[cfg(feature = "enterprise")] use chrono::Timelike; @@ -20,13 +19,12 @@ use windmill_api_client::types::{EditSchedule, NewSchedule, ScriptArgs}; use windmill_common::flows::InputTransform; #[cfg(any(feature = "python", feature = "deno_core"))] -use windmill_common::flow_status::{RestartedFrom}; +use windmill_common::flow_status::RestartedFrom; use windmill_common::{ - flows::{ FlowValue}, - jobs::{ JobPayload, RawCode}, - scripts::{ScriptLang}, - + flows::FlowValue, + jobs::{JobPayload, RawCode}, + scripts::ScriptLang, }; mod common; use common::*; @@ -34,7 +32,6 @@ use common::*; #[cfg(feature = "enterprise")] use futures::StreamExt; - // async fn _print_job(id: Uuid, db: &Pool) -> Result<(), anyhow::Error> { // tracing::info!( // "{:#?}", @@ -45,7 +42,6 @@ use futures::StreamExt; // Ok(()) // } - #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_iteration(db: Pool) -> anyhow::Result<()> { @@ -167,8 +163,6 @@ async fn test_iteration_parallel(db: Pool) -> anyhow::Result<()> { Ok(()) } - - #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_deno_flow(db: Pool) -> anyhow::Result<()> { @@ -341,7 +335,6 @@ use windmill_common::flows::FlowModuleValue; #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_deno_flow_same_worker(db: Pool) -> anyhow::Result<()> { - initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; @@ -1138,8 +1131,6 @@ public class Main { Ok(()) } - - #[sqlx::test(fixtures("base"))] async fn test_bun_job_datetime(db: Pool) -> anyhow::Result<()> { initialize_tracing().await; @@ -2627,7 +2618,6 @@ async fn test_flow_schedule_handlers(db: Pool) -> anyhow::Result<()> { Ok(()) } - #[sqlx::test(fixtures("base", "relative_bun"))] async fn test_relative_imports_bun(db: Pool) -> anyhow::Result<()> { let content = r#" @@ -2699,8 +2689,6 @@ export async function main() { Ok(()) } - - #[sqlx::test(fixtures("base", "result_format"))] async fn test_result_format(db: Pool) -> anyhow::Result<()> { let ordered_result_job_id = "1eecb96a-c8b0-4a3d-b1b6-087878c55e41"; @@ -2929,4 +2917,3 @@ async fn test_workflow_as_code(db: Pool) -> anyhow::Result<()> { .await; Ok(()) } - diff --git a/backend/windmill-api-client/src/codegen.rs b/backend/windmill-api-client/src/codegen.rs index 2ef8fe8cac..a295d9df8f 100644 --- a/backend/windmill-api-client/src/codegen.rs +++ b/backend/windmill-api-client/src/codegen.rs @@ -8,9 +8,80 @@ pub mod types { #[allow(unused_imports)] use std::convert::TryFrom; #[derive(Clone, Debug, Deserialize, Serialize)] + pub struct AiAgent { + pub input_transforms: std::collections::HashMap, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parallel: Option, + pub tools: Vec, + #[serde(rename = "type")] + pub type_: AiAgentType, + } + impl From<&AiAgent> for AiAgent { + fn from(value: &AiAgent) -> Self { + value.clone() + } + } + #[derive( + Clone, + Copy, + Debug, + Deserialize, + Eq, + Hash, + Ord, + PartialEq, + PartialOrd, + Serialize + )] + pub enum AiAgentType { + #[serde(rename = "aiagent")] + Aiagent, + } + impl From<&AiAgentType> for AiAgentType { + fn from(value: &AiAgentType) -> Self { + value.clone() + } + } + impl ToString for AiAgentType { + fn to_string(&self) -> String { + match *self { + Self::Aiagent => "aiagent".to_string(), + } + } + } + impl std::str::FromStr for AiAgentType { + type Err = &'static str; + fn from_str(value: &str) -> Result { + match value { + "aiagent" => Ok(Self::Aiagent), + _ => Err("invalid value"), + } + } + } + impl std::convert::TryFrom<&str> for AiAgentType { + type Error = &'static str; + fn try_from(value: &str) -> Result { + value.parse() + } + } + impl std::convert::TryFrom<&String> for AiAgentType { + type Error = &'static str; + fn try_from(value: &String) -> Result { + value.parse() + } + } + impl std::convert::TryFrom for AiAgentType { + type Error = &'static str; + fn try_from(value: String) -> Result { + value.parse() + } + } + #[derive(Clone, Debug, Deserialize, Serialize)] pub struct AiConfig { #[serde(default, skip_serializing_if = "Option::is_none")] pub code_completion_model: Option, + #[serde(default, skip_serializing_if = "std::collections::HashMap::is_empty")] + pub custom_prompts: std::collections::HashMap, #[serde(default, skip_serializing_if = "Option::is_none")] pub default_model: Option, #[serde(default, skip_serializing_if = "std::collections::HashMap::is_empty")] @@ -1295,8 +1366,8 @@ pub mod types { Websocket, #[serde(rename = "kafka")] Kafka, - #[serde(rename = "email")] - Email, + #[serde(rename = "default_email")] + DefaultEmail, #[serde(rename = "nats")] Nats, #[serde(rename = "postgres")] @@ -1307,6 +1378,8 @@ pub mod types { Mqtt, #[serde(rename = "gcp")] Gcp, + #[serde(rename = "email")] + Email, } impl From<&CaptureTriggerKind> for CaptureTriggerKind { fn from(value: &CaptureTriggerKind) -> Self { @@ -1320,12 +1393,13 @@ pub mod types { Self::Http => "http".to_string(), Self::Websocket => "websocket".to_string(), Self::Kafka => "kafka".to_string(), - Self::Email => "email".to_string(), + Self::DefaultEmail => "default_email".to_string(), Self::Nats => "nats".to_string(), Self::Postgres => "postgres".to_string(), Self::Sqs => "sqs".to_string(), Self::Mqtt => "mqtt".to_string(), Self::Gcp => "gcp".to_string(), + Self::Email => "email".to_string(), } } } @@ -1337,12 +1411,13 @@ pub mod types { "http" => Ok(Self::Http), "websocket" => Ok(Self::Websocket), "kafka" => Ok(Self::Kafka), - "email" => Ok(Self::Email), + "default_email" => Ok(Self::DefaultEmail), "nats" => Ok(Self::Nats), "postgres" => Ok(Self::Postgres), "sqs" => Ok(Self::Sqs), "mqtt" => Ok(Self::Mqtt), "gcp" => Ok(Self::Gcp), + "email" => Ok(Self::Email), _ => Err("invalid value"), } } @@ -1493,6 +1568,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK Flownode, #[serde(rename = "appscript")] Appscript, + #[serde(rename = "aiagent")] + Aiagent, } impl From<&CompletedJobJobKind> for CompletedJobJobKind { fn from(value: &CompletedJobJobKind) -> Self { @@ -1516,6 +1593,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK Self::Flowscript => "flowscript".to_string(), Self::Flownode => "flownode".to_string(), Self::Appscript => "appscript".to_string(), + Self::Aiagent => "aiagent".to_string(), } } } @@ -1537,6 +1615,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK "flowscript" => Ok(Self::Flowscript), "flownode" => Ok(Self::Flownode), "appscript" => Ok(Self::Appscript), + "aiagent" => Ok(Self::Aiagent), _ => Err("invalid value"), } } @@ -1705,6 +1784,21 @@ the execution of this script will be permissioned_as and by extension its DT_TOK } } #[derive(Clone, Debug, Deserialize, Serialize)] + pub struct CreateWorkspaceFork { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub color: Option, + pub id: String, + pub name: String, + pub parent_workspace_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub username: Option, + } + impl From<&CreateWorkspaceFork> for CreateWorkspaceFork { + fn from(value: &CreateWorkspaceFork) -> Self { + value.clone() + } + } + #[derive(Clone, Debug, Deserialize, Serialize)] pub struct CriticalAlert { ///Acknowledgment status of the alert, can be true, false, or null if not set #[serde(default, skip_serializing_if = "Option::is_none")] @@ -1799,6 +1893,24 @@ the execution of this script will be permissioned_as and by extension its DT_TOK } } #[derive(Clone, Debug, Deserialize, Serialize)] + pub struct DependencyMap { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub imported_path: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub importer_kind: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub importer_node_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub importer_path: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspace_id: Option, + } + impl From<&DependencyMap> for DependencyMap { + fn from(value: &DependencyMap) -> Self { + value.clone() + } + } + #[derive(Clone, Debug, Deserialize, Serialize)] pub struct DucklakeSettings { pub ducklakes: std::collections::HashMap, } @@ -1909,6 +2021,27 @@ the execution of this script will be permissioned_as and by extension its DT_TOK } } #[derive(Clone, Debug, Deserialize, Serialize)] + pub struct EditEmailTrigger { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error_handler_args: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error_handler_path: Option, + pub is_flow: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub local_part: Option, + pub path: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retry: Option, + pub script_path: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspaced_local_part: Option, + } + impl From<&EditEmailTrigger> for EditEmailTrigger { + fn from(value: &EditEmailTrigger) -> Self { + value.clone() + } + } + #[derive(Clone, Debug, Deserialize, Serialize)] pub struct EditHttpTrigger { pub authentication_method: AuthenticationMethod, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -2234,6 +2367,23 @@ the execution of this script will be permissioned_as and by extension its DT_TOK } } #[derive(Clone, Debug, Deserialize, Serialize)] + pub struct EmailTrigger { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error_handler_args: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error_handler_path: Option, + pub local_part: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retry: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspaced_local_part: Option, + } + impl From<&EmailTrigger> for EmailTrigger { + fn from(value: &EmailTrigger) -> Self { + value.clone() + } + } + #[derive(Clone, Debug, Deserialize, Serialize)] pub struct EndpointTool { ///JSON schema for request body #[serde(default, skip_serializing_if = "Option::is_none")] @@ -2481,7 +2631,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK #[serde(default, skip_serializing_if = "Option::is_none")] pub suspend: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub timeout: Option, + pub timeout: Option, pub value: FlowModuleValue, } impl From<&FlowModule> for FlowModule { @@ -2555,6 +2705,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK BranchOne(BranchOne), BranchAll(BranchAll), Identity(Identity), + AiAgent(AiAgent), } impl From<&FlowModuleValue> for FlowModuleValue { fn from(value: &FlowModuleValue) -> Self { @@ -2601,6 +2752,11 @@ the execution of this script will be permissioned_as and by extension its DT_TOK Self::Identity(value) } } + impl From for FlowModuleValue { + fn from(value: AiAgent) -> Self { + Self::AiAgent(value) + } + } #[derive(Clone, Debug, Deserialize, Serialize)] pub struct FlowPreview { pub args: ScriptArgs, @@ -2648,6 +2804,10 @@ the execution of this script will be permissioned_as and by extension its DT_TOK } #[derive(Clone, Debug, Deserialize, Serialize)] pub struct FlowStatusModule { + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub agent_actions: Vec>, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub agent_actions_success: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub approvers: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -4810,6 +4970,26 @@ the execution of this script will be permissioned_as and by extension its DT_TOK } } #[derive(Clone, Debug, Deserialize, Serialize)] + pub struct NewEmailTrigger { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error_handler_args: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error_handler_path: Option, + pub is_flow: bool, + pub local_part: String, + pub path: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retry: Option, + pub script_path: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workspaced_local_part: Option, + } + impl From<&NewEmailTrigger> for NewEmailTrigger { + fn from(value: &NewEmailTrigger) -> Self { + value.clone() + } + } + #[derive(Clone, Debug, Deserialize, Serialize)] pub struct NewHttpTrigger { pub authentication_method: AuthenticationMethod, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -5907,6 +6087,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK #[derive(Clone, Debug, Deserialize, Serialize)] pub struct Preview { pub args: ScriptArgs, + ///The code to run #[serde(default, skip_serializing_if = "Option::is_none")] pub content: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -5917,8 +6098,10 @@ the execution of this script will be permissioned_as and by extension its DT_TOK pub language: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub lock: Option, + ///The path to the script #[serde(default, skip_serializing_if = "Option::is_none")] pub path: Option, + ///The hash of the script #[serde(default, skip_serializing_if = "Option::is_none")] pub script_hash: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -6127,6 +6310,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK Flownode, #[serde(rename = "appscript")] Appscript, + #[serde(rename = "aiagent")] + Aiagent, } impl From<&QueuedJobJobKind> for QueuedJobJobKind { fn from(value: &QueuedJobJobKind) -> Self { @@ -6150,6 +6335,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK Self::Flowscript => "flowscript".to_string(), Self::Flownode => "flownode".to_string(), Self::Appscript => "appscript".to_string(), + Self::Aiagent => "aiagent".to_string(), } } } @@ -6171,6 +6357,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK "flowscript" => Ok(Self::Flowscript), "flownode" => Ok(Self::Flownode), "appscript" => Ok(Self::Appscript), + "aiagent" => Ok(Self::Aiagent), _ => Err("invalid value"), } } @@ -6676,6 +6863,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK pub constant: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub exponential: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub retry_if: Option, } impl From<&Retry> for Retry { fn from(value: &Retry) -> Self { @@ -6710,6 +6899,15 @@ the execution of this script will be permissioned_as and by extension its DT_TOK value.clone() } } + #[derive(Clone, Debug, Deserialize, Serialize)] + pub struct RetryRetryIf { + pub expr: String, + } + impl From<&RetryRetryIf> for RetryRetryIf { + fn from(value: &RetryRetryIf) -> Self { + value.clone() + } + } #[derive( Clone, Copy, @@ -7861,6 +8059,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK } #[derive(Clone, Debug, Deserialize, Serialize)] pub struct TriggersCount { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub default_email_count: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub email_count: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -8069,10 +8269,14 @@ the execution of this script will be permissioned_as and by extension its DT_TOK #[derive(Clone, Debug, Deserialize, Serialize)] pub struct UserWorkspaceListWorkspacesItem { pub color: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub created_by: Option, pub id: String, pub name: String, #[serde(default, skip_serializing_if = "Option::is_none")] pub operator_settings: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_workspace_id: Option, pub username: String, } impl From<&UserWorkspaceListWorkspacesItem> for UserWorkspaceListWorkspacesItem { @@ -8479,6 +8683,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK pub id: String, pub name: String, pub owner: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_workspace_id: Option, } impl From<&Workspace> for Workspace { fn from(value: &Workspace) -> Self { @@ -8550,6 +8756,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK pub email: String, pub is_admin: bool, pub operator: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_workspace_id: Option, pub workspace_id: String, } impl From<&WorkspaceInvite> for WorkspaceInvite { @@ -8561,7 +8769,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK #[derive(Clone, Debug)] /**Client for Windmill API -Version: 1.526.1*/ +Version: 1.543.0*/ pub struct Client { pub(crate) baseurl: String, pub(crate) client: reqwest::Client, @@ -8607,7 +8815,7 @@ impl Client { /// This string is pulled directly from the source OpenAPI /// document and may be in any format the API selects. pub fn api_version(&self) -> &'static str { - "1.526.1" + "1.543.0" } } impl Client { diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index ba53fb2c37..724cfb5f23 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -2134,6 +2134,40 @@ paths: schema: type: string + /w/{workspace}/workspaces/rebuild_dependency_map: + post: + summary: rebuild dependency map + operationId: rebuildDependencyMap + tags: + - workspace + parameters: + - $ref: "#/components/parameters/WorkspaceId" + responses: + "200": + description: status + content: + text/plain: + schema: + type: string + + /w/{workspace}/workspaces/get_dependency_map: + get: + summary: get dependency map + operationId: getDependencyMap + tags: + - workspace + parameters: + - $ref: "#/components/parameters/WorkspaceId" + responses: + "200": + description: dmap + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/DependencyMap" + /w/{workspace}/workspaces/edit_slack_command: post: summary: edit slack command @@ -17645,6 +17679,25 @@ components: - owner - created_at + DependencyMap: + type: object + properties: + workspace_id: + type: string + nullable: true + importer_path: + type: string + nullable: true + importer_kind: + type: string + nullable: true + imported_path: + type: string + nullable: true + importer_node_id: + type: string + nullable: true + WorkspaceInvite: type: object properties: diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 8254e544e7..ff14dc1ab8 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -268,7 +268,7 @@ pub struct CreateApp { pub custom_path: Option, } -#[derive(Deserialize)] +#[derive(Serialize, Deserialize)] pub struct EditApp { pub path: Option, pub summary: Option, diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 3f8f3a993a..3ec3c1b176 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -87,7 +87,7 @@ pub mod ee; pub mod ee_oss; pub mod embeddings; mod favorite; -mod flows; +pub mod flows; mod folders; mod granular_acls; mod groups; @@ -182,6 +182,7 @@ mod workspaces_oss; #[cfg(feature = "mcp")] mod mcp; +pub use apps::EditApp; pub const DEFAULT_BODY_LIMIT: usize = 2097152 * 100; // 200MB lazy_static::lazy_static! { diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 5e378a0fb3..c2a03f6bbb 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -1036,6 +1036,9 @@ async fn create_script_internal<'c>( let content = ns.content.clone(); let language = ns.language.clone(); tokio::spawn(async move { + // TODO: I don't think we want this. We might want to send dependency job. But skip any calculations if lock is already present. + // It will allow us to make code more consistent and predictable. + // wait for 10 seconds to make sure the script is deployed and that the CLI sync that pushed it (f one) is complete tokio::time::sleep(std::time::Duration::from_secs(10)).await; if let Err(e) = process_relative_imports( diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 823240a801..537cff8002 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -51,6 +51,7 @@ use windmill_common::{ utils::{paginate, rd_string, require_admin, Pagination}, }; use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; +use windmill_worker::scoped_dependency_map::{DependencyMap, ScopedDependencyMap}; #[cfg(feature = "enterprise")] use windmill_common::utils::require_admin_or_devops; @@ -78,6 +79,8 @@ pub fn workspaced_service() -> Router { .route("/invite_user", post(invite_user)) .route("/add_user", post(add_user)) .route("/delete_invite", post(delete_invite)) + .route("/rebuild_dependency_map", post(rebuild_dependency_map)) + .route("/get_dependency_map", get(get_dependency_map)) .route("/get_settings", get(get_settings)) .route("/get_deploy_to", get(get_deploy_to)) .route("/edit_slack_command", post(edit_slack_command)) @@ -3351,6 +3354,42 @@ async fn get_workspace_name( Ok(workspace) } +async fn get_dependency_map( + authed: ApiAuthed, + Path(w_id): Path, + Extension(user_db): Extension, +) -> JsonResult> { + require_admin(authed.is_admin, &authed.username)?; + + let mut tx = user_db.begin(&authed).await?; + let dmap = sqlx::query_as!( + DependencyMap, + " + SELECT workspace_id, importer_path, importer_kind::text, imported_path, importer_node_id + FROM dependency_map WHERE workspace_id = $1", + &w_id + ) + .fetch_all(&mut *tx) + .await?; + + tx.commit().await?; + + Ok(Json(dmap)) +} + +#[axum::debug_handler] +async fn rebuild_dependency_map( + Extension(db): Extension, + Path(w_id): Path, + authed: ApiAuthed, +) -> Result { + require_admin(authed.is_admin, &authed.username)?; + if *CLOUD_HOSTED { + return Err(Error::BadRequest("Disabled on Cloud".into())); + } + ScopedDependencyMap::rebuild_map(&w_id, &db).await +} + #[derive(Deserialize)] struct ChangeWorkspaceName { new_name: String, diff --git a/backend/windmill-common/src/apps.rs b/backend/windmill-common/src/apps.rs index b4552111e0..e27504a9f6 100644 --- a/backend/windmill-common/src/apps.rs +++ b/backend/windmill-common/src/apps.rs @@ -9,8 +9,11 @@ use std::{collections::HashMap, sync::Arc}; use serde::{Deserialize, Serialize}; +use serde_json::{from_value, Value}; use tokio::sync::RwLock; +use crate::{error, scripts::ScriptLang}; + lazy_static::lazy_static! { pub static ref APP_WORKSPACED_ROUTE: Arc> = Arc::new(RwLock::new(false)); } @@ -33,3 +36,70 @@ pub struct ListAppQuery { pub struct RawAppValue { pub files: HashMap, } + +pub struct AppInlineScript { + pub language: Option, + pub content: String, + pub lock: Option, +} + +/// Traverse FlowValue while invoking provided by caller callback on leafs +// #[async_recursion::async_recursion(?Send)] +pub fn traverse_app_inline_scripts< + C: FnMut(AppInlineScript, Option) -> error::Result<()>, +>( + value: &Value, + // Set to None. + container_id: Option, + cb: &mut C, +) -> error::Result<()> { + match value { + Value::Object(object) => { + if let Some(Value::Object(script)) = object.get("inlineScript") { + let (language, lock, code) = ( + script + .get("language") + .cloned() + .map(|v| from_value::(v).ok()) + .flatten(), + script + .get("lock") + .and_then(Value::as_str) + .map(str::to_owned), + script + .get("content") + .and_then(Value::as_str) + .map(str::to_owned) + .ok_or(error::Error::internal_err( + "Missing `content` in inlineScript".to_string(), + ))?, + ); + if language.is_some() { + cb( + AppInlineScript { language, content: code.to_owned(), lock }, + container_id.clone(), + )?; + } + } else { + for (_, value) in object { + traverse_app_inline_scripts( + value, + object + .get("id") + .and_then(Value::as_str) + .map(str::to_owned) + .or(container_id.clone()), + cb, + )?; + } + } + } + Value::Array(array) => { + for value in array { + traverse_app_inline_scripts(value, container_id.clone(), cb)?; + } + } + _ => {} + } + Ok(()) +} diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 4f93812f6a..d4f8308880 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -93,7 +93,7 @@ pub struct ListableFlow { pub deployment_msg: Option, } -#[derive(Debug, Deserialize, sqlx::FromRow)] +#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)] pub struct NewFlow { pub path: String, pub summary: String, @@ -158,6 +158,34 @@ impl FlowValue { flow_module } + + /// Traverse FlowValue while invoking provided by caller callback on leafs + // #[async_recursion::async_recursion(?Send)] + // TODO: We may be want this async. + pub fn traverse_leafs crate::error::Result<()>>( + modules: &Vec, + cb: &mut C, + ) -> crate::error::Result<()> { + use FlowModuleValue::*; + for module in modules { + match serde_json::from_str::(module.value.get())? { + s @ (Script { .. } + | RawScript { .. } + | Flow { .. } + | FlowScript { .. } + | Identity) => cb(&s, &module.id)?, + ForloopFlow { modules, .. } + | WhileloopFlow { modules, .. } + | AIAgent { tools: modules, .. } => Self::traverse_leafs(&modules, cb)?, + BranchOne { branches, .. } | BranchAll { branches, .. } => { + for branch in branches { + Self::traverse_leafs(&branch.modules, cb)?; + } + } + } + } + Ok(()) + } } #[derive(Debug, Copy, Clone)] @@ -565,6 +593,7 @@ pub struct Branch { rename_all(serialize = "lowercase", deserialize = "lowercase") )] pub enum FlowModuleValue { + /// Reference to another script on the workspace Script { #[serde(default)] #[serde(alias = "input_transform")] @@ -577,12 +606,16 @@ pub enum FlowModuleValue { #[serde(skip_serializing_if = "Option::is_none")] is_trigger: Option, }, + + /// Reference to another flow on the workspace Flow { #[serde(default)] #[serde(alias = "input_transform")] input_transforms: HashMap, path: String, }, + + /// For loop node ForloopFlow { iterator: InputTransform, modules: Vec, @@ -594,6 +627,8 @@ pub enum FlowModuleValue { #[serde(skip_serializing_if = "Option::is_none")] parallelism: Option, }, + + /// While loop node WhileloopFlow { modules: Vec, #[serde(skip_serializing_if = "Option::is_none")] @@ -601,17 +636,24 @@ pub enum FlowModuleValue { #[serde(default = "default_false")] skip_failures: bool, }, + + /// Branch-one node BranchOne { branches: Vec, default: Vec, #[serde(skip_serializing_if = "Option::is_none")] default_node: Option, }, + + /// Branch-all node BranchAll { branches: Vec, #[serde(default = "default_true")] parallel: bool, }, + + /// Inline script node + /// Only exists if parsed from value from `flow_version` | `flow` table. RawScript { #[serde(default)] #[serde(alias = "input_transform", serialize_with = "ordered_map")] @@ -635,8 +677,13 @@ pub enum FlowModuleValue { #[serde(skip_serializing_if = "Option::is_none")] assets: Option>, }, + + /// Just a placeholder Identity, - // Internal only, never exposed to the frontend. + + /// Also Inline script node, but instead of being baked into flow, it references `flow_node` + /// Internal only, never exposed to the frontend. + /// Only exists if parsed from value from `flow_version_lite` table. FlowScript { #[serde(default)] #[serde(alias = "input_transform", serialize_with = "ordered_map")] @@ -656,6 +703,8 @@ pub enum FlowModuleValue { #[serde(skip_serializing_if = "Option::is_none")] assets: Option>, }, + + // AI agent node AIAgent { input_transforms: HashMap, tools: Vec, diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 2a9981d01f..fcfd099c67 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -137,6 +137,12 @@ impl Into for ScriptHash { } } +impl From for ScriptHash { + fn from(value: i64) -> Self { + Self(value) + } +} + #[derive(PartialEq, sqlx::Type)] #[sqlx(transparent, no_pg_array)] pub struct ScriptHashes(pub Vec); @@ -160,7 +166,10 @@ impl<'de> Deserialize<'de> for ScriptHash { D: Deserializer<'de>, { let s = String::deserialize(deserializer)?; - let i = to_i64(&s).map_err(|e| D::Error::custom(format!("{}", e)))?; + let i = to_i64(&s).map_err(|e| { + tracing::error!("Could not deserialize ScriptHash. Note, input should be in Hex and digit amount should be divisible by 16 (can be padded). err: {}", &e); + D::Error::custom(format!("{}", e)) + })?; Ok(ScriptHash(i)) } } @@ -329,7 +338,7 @@ impl Hash for Schema { } } -#[derive(Serialize, Deserialize, Hash)] +#[derive(Serialize, Deserialize, Hash, Debug)] pub struct NewScript { pub path: String, pub parent_hash: Option, diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 87f8282e9e..613c9237ba 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -57,6 +57,7 @@ pub mod result_processor; mod rust_executor; mod sanitized_sql_params; mod schema; +pub mod scoped_dependency_map; mod universal_pkg_installer; mod worker; mod worker_flow; diff --git a/backend/windmill-worker/src/scoped_dependency_map.rs b/backend/windmill-worker/src/scoped_dependency_map.rs new file mode 100644 index 0000000000..4f031a4a42 --- /dev/null +++ b/backend/windmill-worker/src/scoped_dependency_map.rs @@ -0,0 +1,445 @@ +use serde::Serialize; +use tokio::sync::RwLock; +use windmill_common::{ + apps::traverse_app_inline_scripts, + cache, + error::{Error, Result}, + flows::{FlowModuleValue, FlowValue}, + scripts::ScriptLang, +}; + +use std::collections::HashSet; + +use crate::worker_lockfiles::{extract_relative_imports, LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT}; + +// TODO: To be removed in future versions +lazy_static::lazy_static! { + pub static ref WMDEBUG_NO_DMAP_DISSOLVE: bool = std::env::var("WMDEBUG_NO_DMAP_DISSOLVE").is_ok(); +} + +#[derive(Serialize)] +pub struct DependencyMap { + pub workspace_id: Option, + pub importer_path: Option, + pub importer_kind: Option, + pub imported_path: Option, + pub importer_node_id: Option, +} + +#[derive(Debug)] +pub struct ScopedDependencyMap { + dmap: HashSet<(String, String)>, + w_id: String, + importer_path: String, + importer_kind: String, +} + +impl ScopedDependencyMap { + /// Calls DB, however is assumed to be called once per dependency job + /// AND is scoped to smaller subset of data + /// So it is not too expensive + pub(crate) async fn fetch_maybe_rearranged<'a>( + w_id: &str, + importer_path: &str, + importer_kind: &str, + parent_path: &Option, + executor: impl sqlx::Executor<'a, Database = sqlx::Postgres>, + ) -> Result { + if parent_path + .as_ref() + .is_some_and(|x| !x.is_empty() && x != importer_path) + { + tracing::info!( + workspace_id = %w_id, + "detected top level rename from: {} to: {importer_path} on object of kind: {importer_kind}. reflecting in dependency_map.", + parent_path.clone().unwrap_or_default(), + ); + + let dmap = sqlx::query_as::<_, (String, String)>( + " +UPDATE dependency_map + SET importer_path = $1 + WHERE importer_path = $2 + AND importer_kind = $3::text::IMPORTER_KIND + AND workspace_id = $4 +RETURNING importer_node_id, imported_path + ", + ) + .bind(importer_path) + .bind(parent_path.clone().unwrap()) + .bind(importer_kind) + .bind(w_id) + .fetch_all(executor) + .await?; + Ok(Self { + dmap: HashSet::from_iter(dmap.into_iter()), + w_id: w_id.to_owned(), + importer_path: importer_path.to_owned(), + importer_kind: importer_kind.to_owned(), + }) + } else { + Self::fetch(w_id, importer_path, importer_kind, executor).await + } + } + + /// Almost same as [[Self::fetch_maybe_rearranged]], however only reads values, thus a bit faster. + pub async fn fetch<'a>( + w_id: &str, + importer_path: &str, + importer_kind: &str, + executor: impl sqlx::Executor<'a, Database = sqlx::Postgres>, + ) -> Result { + let dmap = sqlx::query_as::<_, (String, String)>( + " +SELECT importer_node_id, imported_path + FROM dependency_map + WHERE workspace_id = $1 + AND importer_path = $2 + AND importer_kind = $3::text::IMPORTER_KIND", + ) + .bind(w_id) + .bind(importer_path) + .bind(importer_kind) + .fetch_all(executor) + .await?; + + Ok(Self { + dmap: HashSet::from_iter(dmap.into_iter()), + w_id: w_id.to_owned(), + importer_path: importer_path.to_owned(), + importer_kind: importer_kind.to_owned(), + }) + } + + /// Add missing entries to `dependency_map` + /// Remove matching entries + pub(crate) async fn patch<'c>( + &mut self, + relative_imports: Option>, + node_id: String, // Flow Step/Node ID + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, + ) -> Result> { + self.patch_tx_ref(relative_imports, &node_id, &mut tx) + .await?; + Ok(tx) + } + + pub(crate) async fn patch_tx_ref<'c>( + &mut self, + relative_imports: Option>, + node_id: &str, // Flow Step/Node ID + tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, + ) -> Result<()> { + let Some(mut relative_imports) = relative_imports else { + tracing::info!("relative imports are not found for: importer - {}, importer_node_id - {}, importer_kind - {}", + &self.importer_path, + &node_id, + &self.importer_kind, + ); + return Ok(()); + }; + + // This does: + // 1. remove all relative imports from relative_imports that ARE tracked in dependency_map + // 2. remove corresponding trackers from dependency_map + // + // After this operation `relative_imports` variable has only untracked imports. + // We will handle those in the next expression. + // + // After all `reduce`'s called ScopedDependencyMap has only extra/orphan imports + // these are going to be clean up by calling [dissolve] + // NOTE: `retain` iterates over vec and remove the ones whose closures returned false. + relative_imports.retain(|imported_path| { + !self + .dmap + // As dmap is HashSet, removing is O(1) operation + // thus making entire process very efficient + // NOTE: `remove` returns true if item was removed and false if wasn't. + .remove(&(node_id.to_owned(), imported_path.to_owned())) + }); + + // As mentioned above, usually this will always be empty. + if !relative_imports.is_empty() { + tracing::info!("adding missing entries to dependency_map: importer_node_id - {}, importer_kind - {}, new_imported_paths - {:?}", + &node_id, + &self.importer_kind, + &relative_imports, + ); + } + + for import in relative_imports { + sqlx::query!( + "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) + VALUES ($1, $2, $3::text::IMPORTER_KIND, $4, $5) ON CONFLICT DO NOTHING", + &self.w_id, + &self.importer_path, + &self.importer_kind, + import, + node_id + ) + .execute(&mut **tx) + .await?; + + tracing::info!("added entry to dependency_map: {import:?}"); + } + Ok(()) + } + + /// clean orphan entries from `dependency_map` + pub(crate) async fn dissolve<'a>( + self, + mut tx: sqlx::Transaction<'a, sqlx::Postgres>, + ) -> sqlx::Transaction<'a, sqlx::Postgres> { + if *WMDEBUG_NO_DMAP_DISSOLVE { + tracing::warn!( + "WMDEBUG_NO_DMAP_DISSOLVE usually should not be used. Behavior might be unstable." + ); + return tx; + } + + tracing::info!("dissolving dependency_map: {:?}", &self); + + // We _could_ shove it into single query, but this query is rarely called AND let's keep it simple for redability. + for (importer_node_id, imported_path) in self.dmap.into_iter() { + tracing::info!("cleaning orphan entry from dependency_map: importer_kind - {}, imported_path - {}, importer_node_id - {}", + &self.importer_kind, + &imported_path, + &importer_node_id, + ); + + // Dissolve MUST succeed. Error in dissolve MUST not block the execution. + if let Err(err) = sqlx::query!( + " + DELETE FROM dependency_map + WHERE workspace_id = $1 + AND importer_path = $2 + AND importer_kind = $3::text::IMPORTER_KIND + AND importer_node_id = $4 + AND imported_path = $5 + ", + &self.w_id, + &self.importer_path, + &self.importer_kind, + &importer_node_id, + &imported_path, + ) + .execute(&mut *tx) + .await + { + tracing::error!( + "error while cleaning dependency_map for: importer_node_id - {}, imported_path - {}, importer_path - {}: {err}", + importer_node_id, + imported_path, + self.importer_path, + ); + } + } + tx + } + + /// Selectively clean dependency_map for object + /// If `importer_node_id` is None will clear all nodes. + pub(crate) async fn clear_map_for_item<'c>( + item_path: &str, + w_id: &str, + importer_kind: &str, + mut tx: sqlx::Transaction<'c, sqlx::Postgres>, + importer_node_id: &Option, + ) -> sqlx::Transaction<'c, sqlx::Postgres> { + tracing::warn!( + importer = item_path, + kind = importer_kind, + node_id = importer_node_id, + workspace_id = w_id, + "discovered orphan entry in `dependency_map`. It will be healed automatically, however please report this issue to Windmill Team. It is also advised to rebuild maps in workspace settings in troubleshooting.", + ); + + // MUST succeed. Error MUST not block the execution. + if let Err(err) = sqlx::query!( + "DELETE FROM dependency_map + WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND + AND workspace_id = $2 AND ($4::text IS NULL OR importer_node_id = $4::text)", + item_path, + w_id, + importer_kind, + importer_node_id.clone(), + ) + .execute(&mut *tx) + .await + { + tracing::error!( + workspace_id = w_id, + "error while clearing discovered orphan: {err}" + ); + } + tx + } + + /// Run if you want to rebuild maps on specific workspace. + /// Potentially takes much time + pub async fn rebuild_map(w_id: &str, db: &sqlx::Pool) -> Result { + async fn inner<'c>(w_id: &str, db: &sqlx::Pool) -> Result { + // Scripts + tracing::info!(workspace_id = w_id, "Rebuilding dependency map for scripts"); + for r in sqlx::query!( + "SELECT path, hash FROM script WHERE workspace_id = $1 AND archived = false", + w_id + ) + .fetch_all(db) + .await? + { + let (sd, smd) = cache::script::fetch(&db.clone().into(), r.hash.into()).await?; + let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "script", db).await?; + let mut tx = db.begin().await?; + + if (smd.language.is_some_and(|v| v == ScriptLang::Bun) + && sd + .lock + .as_ref() + .is_some_and(|v| v.contains("generatedFromPackageJson"))) + || (smd.language.is_some_and(|v| v == ScriptLang::Python3) + && sd.lock.as_ref().is_some_and(|v| { + v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT) + })) + { + // if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map + // because we do not want to have dependencies be recomputed automatically. Empty relative imports passed + // to update_script_dependency_map will clear the dependency map. + } else { + tx = dmap + .patch( + extract_relative_imports(&sd.code, &r.path, &smd.language), + "".into(), + tx, + ) + .await?; + } + if !*WMDEBUG_NO_DMAP_DISSOLVE { + dmap.dissolve(tx).await.commit().await?; + } + tracing::info!(workspace_id = w_id, "Rebuilt for script {}", &r.path); + } + + // Fetch only top level versions and paths + // It is not fetching value + tracing::info!(workspace_id = w_id, "Rebuilding dependency map for flows"); + for r in sqlx::query!("SELECT path, versions[array_upper(versions, 1)] as version FROM flow WHERE workspace_id = $1", w_id).fetch_all(db).await? { + if let Some(version) = r.version { + // To reduce stress on db try to fetch from cache + // Since our flow versions are immutable it is safe to assume if we have cache for specific version/id it is up to date. + let flow_data = cache::flow::fetch_version(&db.clone().into(), version).await?; + + // Create map for specific flow + let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "flow", db).await?; + + // Traverse retrieved flow modules + let mut tx = db.begin().await?; + let mut to_process = vec![]; + FlowValue::traverse_leafs(&flow_data.flow.modules, &mut |fmv, id| { + match fmv { + // Since we fetched from flow_version it is safe to assume all inline scripts are in form of RawScript. + FlowModuleValue::RawScript { content, language, .. } => { + to_process.push(( + extract_relative_imports( + content, + &(r.path.clone() + "/flow"), + &Some(language.clone()), + ), + id.clone(), + )); + } + // But just in case we will also handle other cases. + FlowModuleValue::FlowScript { .. } => { + // Abort will cancel transaction. + return Err(Error::internal_err("FlowScript is not supposed to be in flow.")); + } + _ => {} + } + Ok(()) + })?; + + for (ri, id) in to_process { + tx = dmap.patch(ri, id, tx).await?; + } + + if !*WMDEBUG_NO_DMAP_DISSOLVE { + dmap.dissolve(tx).await.commit().await?; + } + + tracing::info!(workspace_id = w_id, "Rebuilt for flow {}", &r.path); + } else { + tracing::error!(workspace_id = w_id, "version is never supposed to be none. skipping flow."); + return Err(Error::internal_err("version was none")); + } + } + + // Apps + tracing::info!(workspace_id = w_id, "Rebuilding dependency map for apps"); + for r in sqlx::query!("SELECT path, versions[array_upper(versions, 1)] as version FROM app WHERE workspace_id = $1", w_id).fetch_all(db).await? { + if let Some(version) = r.version { + // TODO: Use cache when implemented. + let value = sqlx::query_scalar!( + "SELECT value FROM app_version WHERE id = $1 LIMIT 1", + version + ) + .fetch_one(db) + .await?; + + let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "app", db).await?; + let mut tx = db.begin().await?; + let mut to_process = vec![]; + traverse_app_inline_scripts(&value, None, &mut |ais, id| { + to_process.push(( + extract_relative_imports( + &ais.content, + &(r.path.clone() + "/app"), + &ais.language, + ), + id, + )); + + Ok(()) + })?; + for (ri, id) in to_process { + tx = dmap.patch(ri, id.unwrap_or_default(), tx).await?; + } + if !*WMDEBUG_NO_DMAP_DISSOLVE { + dmap.dissolve(tx).await.commit().await?; + } + tracing::info!(workspace_id = w_id, "Rebuilt for app {}", &r.path); + } else { + tracing::error!( + workspace_id = w_id, + "version is never supposed to be none. skipping app." + ); + return Err(Error::internal_err("version was none")); + } + } + + Ok("Success".into()) + } + + lazy_static::lazy_static! { + pub static ref LOCKED: RwLock = RwLock::new(false); + } + + if *LOCKED.read().await { + tracing::warn!( + workspace_id = w_id, + "Tried to rebuild dependency map. However rebuild is already in progress." + ); + Ok("There is already one task pending, try again later.".into()) + } else { + tracing::info!(workspace_id = w_id, "Rebuilding dependency map"); + + if *WMDEBUG_NO_DMAP_DISSOLVE { + tracing::warn!("WMDEBUG_NO_DMAP_DISSOLVE usually should not be used. Behavior might be unstable. Please contact Windmill Team for support.") + } + + *LOCKED.write().await = true; + let r = inner(w_id, db).await; + *LOCKED.write().await = false; + r + } + } +} diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 35e89eb7eb..43a5125d79 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -5,6 +5,7 @@ use std::path::{Component, Path, PathBuf}; #[cfg(feature = "python")] use crate::ansible_executor::{get_git_repos_lock, AnsibleDependencyLocks}; +use crate::scoped_dependency_map::{ScopedDependencyMap, WMDEBUG_NO_DMAP_DISSOLVE}; use async_recursion::async_recursion; use itertools::Itertools; use serde_json::value::RawValue; @@ -69,115 +70,6 @@ use crate::{ go_executor::install_go_dependencies, }; -pub async fn update_script_dependency_map( - job_id: &Uuid, - db: &DB, - w_id: &str, - parent_path: &Option, - script_path: &str, - relative_imports: Vec, -) -> error::Result<()> { - let importer_kind = "script"; - - let mut tx = db.begin().await?; - tx = clear_dependency_parent_path(parent_path, script_path, w_id, importer_kind, tx).await?; - - tx = clear_dependency_map_for_item(script_path, w_id, importer_kind, tx, &None).await?; - - if !relative_imports.is_empty() { - let mut logs = "".to_string(); - logs.push_str("\n--- RELATIVE IMPORTS ---\n\n"); - logs.push_str(&relative_imports.join("\n")); - - tx = add_relative_imports_to_dependency_map( - script_path, - w_id, - relative_imports, - importer_kind, - tx, - &mut logs, - None, - ) - .await?; - append_logs(job_id, w_id, logs, &db.into()).await; - } - tx.commit().await?; - - Ok(()) -} - -async fn add_relative_imports_to_dependency_map<'c>( - script_path: &str, - w_id: &str, - relative_imports: Vec, - importer_kind: &str, - mut tx: sqlx::Transaction<'c, sqlx::Postgres>, - logs: &mut String, - node_id: Option, -) -> error::Result> { - for import in relative_imports { - sqlx::query!( - "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) - VALUES ($1, $2, $4::text::IMPORTER_KIND, $3, $5) ON CONFLICT DO NOTHING", - w_id, - script_path, - import, - importer_kind, - node_id.clone().unwrap_or_default() - ) - .execute(&mut *tx) - .await?; - logs.push_str(&format!("{}\n", import)); - } - Ok(tx) -} - -async fn clear_dependency_map_for_item<'c>( - item_path: &str, - w_id: &str, - importer_kind: &str, - mut tx: sqlx::Transaction<'c, sqlx::Postgres>, - importer_node_id: &Option, -) -> Result> { - sqlx::query!( - "DELETE FROM dependency_map - WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND - AND workspace_id = $2 AND ($4::text IS NULL OR importer_node_id = $4::text)", - item_path, - w_id, - importer_kind, - importer_node_id.clone() - ) - .execute(&mut *tx) - .await?; - Ok(tx) -} - -async fn clear_dependency_parent_path<'c>( - parent_path: &Option, - item_path: &str, - w_id: &str, - importer_kind: &str, - mut tx: sqlx::Transaction<'c, sqlx::Postgres>, -) -> Result> { - if parent_path - .as_ref() - .is_some_and(|x| !x.is_empty() && x != item_path) - { - sqlx::query!( - "DELETE FROM dependency_map - WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND - AND workspace_id = $2", - parent_path.clone().unwrap(), - w_id, - importer_kind - ) - .execute(&mut *tx) - .await?; - } - Ok(tx) -} - fn try_normalize(path: &Path) -> Option { let mut ret = PathBuf::new(); @@ -237,6 +129,7 @@ pub fn extract_relative_imports( _ => None, } } + #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_dependency_job( job: &MiniPulledJob, @@ -369,6 +262,7 @@ pub async fn handle_dependency_job( // where Second will **always** do with lock being not null let deployed_hash = if script_info.lock.is_some() && !*WMDEBUG_NO_HASH_CHANGE_ON_DJ { let path = script_info.path.clone(); + let mut tx = db.begin().await?; // This entire section exists to solve following problem: // @@ -439,7 +333,12 @@ pub async fn handle_dependency_job( FROM script WHERE hash = $2 AND workspace_id = $3; ", new_hash, current_hash.0, w_id, &content).execute(db).await?; - tracing::info!("Updated script at path {} with hash {} to new hash {}", path, current_hash.0, new_hash); + tracing::info!( + "Updated script at path {} with hash {} to new hash {}", + path, + current_hash.0, + new_hash + ); // Archive current sqlx::query!( "UPDATE script SET archived = true WHERE hash = $1 AND workspace_id = $2", @@ -448,7 +347,11 @@ pub async fn handle_dependency_job( ) .execute(&mut *tx) .await?; - tracing::info!("Archived script at path {} from dependency job {}", path, current_hash.0); + tracing::info!( + "Archived script at path {} from dependency job {}", + path, + current_hash.0 + ); tx.commit().await?; ScriptHash(new_hash) @@ -549,7 +452,7 @@ fn remove_ansi_codes(s: &str) -> String { pub async fn process_relative_imports( db: &sqlx::Pool, - job_id: Option, + _job_id: Option, args: Option<&Json>>>, w_id: &str, script_path: &str, @@ -562,40 +465,50 @@ pub async fn process_relative_imports( permissioned_as: &str, lock: Option, ) -> error::Result<()> { - let relative_imports = extract_relative_imports(&code, script_path, script_lang); - if let Some(relative_imports) = relative_imports { - if (script_lang.is_some_and(|v| v == ScriptLang::Bun) - && lock - .as_ref() - .is_some_and(|v| v.contains("generatedFromPackageJson"))) - || (script_lang.is_some_and(|v| v == ScriptLang::Python3) + // TODO: Should be moved into handle_dependency_job body to be more consistent with how flows and apps are handled + { + let relative_imports = extract_relative_imports(&code, script_path, script_lang); + if let Some(relative_imports) = relative_imports { + let mut tx = db.begin().await?; + let mut dependency_map = ScopedDependencyMap::fetch_maybe_rearranged( + &w_id, + script_path, + "script", + &parent_path, + db, + ) + .await?; + if (script_lang.is_some_and(|v| v == ScriptLang::Bun) && lock .as_ref() - .is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT))) - { - // if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map - // because we do not want to have dependencies be recomputed automatically. Empty relative imports passed - // to update_script_dependency_map will clear the dependency map. - update_script_dependency_map( - &job_id.unwrap_or_else(|| Uuid::nil()), - db, - w_id, - &parent_path, - script_path, - vec![], - ) - .await?; - } else { - update_script_dependency_map( - &job_id.unwrap_or_else(|| Uuid::nil()), - db, - w_id, - &parent_path, - script_path, - relative_imports, - ) - .await?; + .is_some_and(|v| v.contains("generatedFromPackageJson"))) + || (script_lang.is_some_and(|v| v == ScriptLang::Python3) + && lock + .as_ref() + .is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT))) + { + // if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map + // because we do not want to have dependencies be recomputed automatically. Empty relative imports passed + // to update_script_dependency_map will clear the dependency map. + + // TODO: Rework the logic for synchronized raw requirements PR. + // For now we will just do nothing and let dissolve clear every item related to this script. + } else { + tx = dependency_map + .patch( + Some(relative_imports), + // Ideally should be None, but due to current implementation will use empty string to represent None. + "".into(), + tx, + ) + .await?; + } + // If felt into first branch which did not call .patch(, this operation will clean dependency_map for this script. + dependency_map.dissolve(tx).await.commit().await?; } + } + + { let already_visited = args .map(|x| { x.get("already_visited") @@ -604,6 +517,9 @@ pub async fn process_relative_imports( }) .flatten() .unwrap_or_default(); + + // But currently we will do this extra db call for every script regardless of whether they have relative imports or not + // Script might have no relative imports but still be referenced by someone else. if let Err(e) = trigger_dependents_to_recompute_dependencies( w_id, script_path, @@ -620,6 +536,7 @@ pub async fn process_relative_imports( tracing::error!(%e, "error triggering dependents to recompute dependencies"); } } + Ok(()) } @@ -710,9 +627,9 @@ pub async fn trigger_dependents_to_recompute_dependencies( "SELECT versions[array_upper(versions, 1)] FROM flow WHERE path = $1 AND workspace_id = $2", s.importer_path, w_id, - ).fetch_one(&mut *flow_tx) + ).fetch_optional(&mut *flow_tx) .await - .map_err(to_anyhow); + .map_err(to_anyhow).map(Option::flatten); match r { // TODO: Fallback - remove eventually. @@ -770,11 +687,22 @@ pub async fn trigger_dependents_to_recompute_dependencies( } } Ok(None) => { - tracing::error!( - "no flow version found for path {path}", - path = s.importer_path - ); - // Do not commit the transaction. It will be dropped and rollbacked + if *WMDEBUG_NO_DMAP_DISSOLVE { + tracing::warn!("WMDEBUG_NO_DMAP_DISSOLVE usually should not be used. Behavior might be unstable."); + } else { + // Remember the path we used to query the flow was fetched just now from dependency_map + // if dependency_map advertise unexistent path, as part of self-healing it should be removed + ScopedDependencyMap::clear_map_for_item( + &s.importer_path, + w_id, + "flow", + flow_tx, + &None, + ) + .await + .commit() + .await?; + } continue; } Err(err) => { @@ -792,6 +720,7 @@ pub async fn trigger_dependents_to_recompute_dependencies( args.insert( "components_to_relock".to_string(), + // TODO: unsafe. Importer Node Ids are not checked. They can simply be array of empty strings! to_raw_value(&s.importer_node_ids), ); @@ -799,9 +728,9 @@ pub async fn trigger_dependents_to_recompute_dependencies( "SELECT versions[array_upper(versions, 1)] FROM app WHERE path = $1 AND workspace_id = $2", s.importer_path, w_id, - ).fetch_one(&mut *tx) + ).fetch_optional(&mut *tx) .await - .map_err(to_anyhow); + .map_err(to_anyhow).map(Option::flatten); match r { // Get current version of current flow. @@ -842,11 +771,22 @@ pub async fn trigger_dependents_to_recompute_dependencies( } } Ok(None) => { - tracing::error!( - "no app version found for path {path}", - path = s.importer_path - ); - // Do not commit the transaction. It will be dropped and rollbacked + if *WMDEBUG_NO_DMAP_DISSOLVE { + tracing::warn!("WMDEBUG_NO_DMAP_DISSOLVE usually should not be used. Behavior might be unstable."); + } else { + // Remember the path we used to query the flow was fetched just now from dependency_map + // if dependency_map advertise unexistent path, as part of self-healing it should be removed + ScopedDependencyMap::clear_map_for_item( + &s.importer_path, + w_id, + "app", + tx, + &None, + ) + .await + .commit() + .await?; + } continue; } Err(err) => { @@ -992,8 +932,15 @@ pub async fn handle_flow_dependency_job( .clone(); let mut tx = db.begin().await?; - tx = clear_dependency_parent_path(&parent_path, &job_path, &job.workspace_id, "flow", tx) - .await?; + + let mut dependency_map = ScopedDependencyMap::fetch_maybe_rearranged( + &job.workspace_id, + &job_path, + "flow", + &parent_path, + &mut *tx, + ) + .await?; if !skip_flow_update { sqlx::query!( @@ -1026,6 +973,7 @@ pub async fn handle_flow_dependency_job( occupancy_metrics, skip_flow_update, raw_deps, + &mut dependency_map, ) .await?; @@ -1083,6 +1031,7 @@ pub async fn handle_flow_dependency_job( tracing::error!(%job.id, %err, "error checking cancellation for job {0}: {err}", job.id); false }) { + // Drop tx and thus cancel any changes return Ok(to_raw_value_owned(json!({ "status": "Flow lock generation was canceled", }))); @@ -1093,6 +1042,8 @@ pub async fn handle_flow_dependency_job( Error::internal_err("Flow Dependency requires script hash (flow version)".to_owned()) })?; + tx = dependency_map.dissolve(tx).await; + sqlx::query!( "UPDATE flow SET value = $1 WHERE path = $2 AND workspace_id = $3", &new_flow_value as &Json>, @@ -1111,6 +1062,7 @@ pub async fn handle_flow_dependency_job( // Compute a lite version of the flow value (`RawScript` => `FlowScript`). let mut value_lite = flow.clone(); + tx = reduce_flow( tx, &mut value_lite.modules, @@ -1197,6 +1149,8 @@ struct LockModuleError { error: Error, } +// TODO: Maybe use [FlowValue::traverse_leafs] +// IMPORTANT: If updating this function, make sure you also update [FlowValue::traverse_leafs] async fn lock_modules<'c>( modules: Vec, job: &MiniPulledJob, @@ -1214,7 +1168,7 @@ async fn lock_modules<'c>( occupancy_metrics: &mut OccupancyMetrics, skip_flow_update: bool, raw_deps: Option>, - // (modules to replace old seq (even unmmodified ones), new transaction, modified ids) ) + dependency_map: &mut ScopedDependencyMap, // (modules to replace old seq (even unmmodified ones), new transaction, modified ids) ) ) -> Result<( Vec, sqlx::Transaction<'c, sqlx::Postgres>, @@ -1268,6 +1222,7 @@ async fn lock_modules<'c>( occupancy_metrics, skip_flow_update, raw_deps.clone(), + dependency_map, )) .await?; e.value = FlowModuleValue::ForloopFlow { @@ -1304,6 +1259,7 @@ async fn lock_modules<'c>( occupancy_metrics, skip_flow_update, raw_deps.clone(), + dependency_map, )) .await?; nmodified_ids.extend(inner_modified_ids); @@ -1332,6 +1288,7 @@ async fn lock_modules<'c>( occupancy_metrics, skip_flow_update, raw_deps.clone(), + dependency_map, )) .await?; e.value = FlowModuleValue::WhileloopFlow { @@ -1365,6 +1322,7 @@ async fn lock_modules<'c>( occupancy_metrics, skip_flow_update, raw_deps.clone(), + dependency_map, )) .await?; nmodified_ids.extend(inner_modified_ids); @@ -1391,6 +1349,7 @@ async fn lock_modules<'c>( occupancy_metrics, skip_flow_update, raw_deps.clone(), + dependency_map, )) .await?; errors.extend(ninner_errors); @@ -1442,8 +1401,19 @@ async fn lock_modules<'c>( .await?; } + let dep_path = path.clone().unwrap_or_else(|| job_path.to_string()); + let relative_imports = extract_relative_imports( + &content, + &format!("{dep_path}/flow"), + &Some(language.clone()), + ); + if let Some(locks_to_reload) = locks_to_reload { if !locks_to_reload.contains(&e.id) { + tx = dependency_map + .patch(relative_imports.clone(), e.id.clone(), tx) + .await?; + new_flow_modules.push(e); continue; } @@ -1451,6 +1421,10 @@ async fn lock_modules<'c>( if lock.as_ref().is_some_and(|x| !x.trim().is_empty()) { let skip_creating_new_lock = skip_creating_new_lock(&language, &content); if skip_creating_new_lock { + tx = dependency_map + .patch(relative_imports.clone(), e.id.clone(), tx) + .await?; + new_flow_modules.push(e); continue; } @@ -1498,36 +1472,9 @@ async fn lock_modules<'c>( // 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( - &job_path, - &job.workspace_id, - "flow", - tx, - &Some(e.id.clone()), - ) - .await?; - let relative_imports = extract_relative_imports( - &content, - &format!("{dep_path}/flow"), - &Some(language.clone()), - ); - if let Some(relative_imports) = relative_imports { - let mut logs = "".to_string(); - logs.push_str(format!("\n\n--- RELATIVE IMPORTS of {} ---\n\n", e.id).as_str()); - - tx = add_relative_imports_to_dependency_map( - &dep_path, - &job.workspace_id, - relative_imports, - "flow", - tx, - &mut logs, - Some(e.id.clone()), - ) + tx = dependency_map + .patch(relative_imports.clone(), e.id.clone(), tx) .await?; - append_logs(&job.id, &job.workspace_id, logs, &db.into()).await; - } if language == ScriptLang::Bun || language == ScriptLang::Bunnative { let anns = windmill_common::worker::TypeScriptAnnotations::parse(&content); @@ -1638,7 +1585,6 @@ async fn insert_flow_node<'c>( Ok((tx, FlowNodeId(id))) } -// TODO: Clean up dependency map when moved/renamed? async fn insert_app_script( db: &sqlx::Pool, path: &str, @@ -1771,6 +1717,7 @@ async fn reduce_flow<'c>( Some(language), ) .await?; + val = FlowScript { input_transforms, id, @@ -1906,6 +1853,10 @@ fn skip_creating_new_lock(language: &ScriptLang, content: &str) -> bool { true } +// TODO: Use transaction? +// TODO: Use abstracted traverse function. +// +// IMPORTANT: If updating this function, make sure you also update [traverse_app_inline_scripts] #[async_recursion] async fn lock_modules_app( value: Value, @@ -1923,6 +1874,7 @@ async fn lock_modules_app( locks_to_reload: &Option>, // Represents the closest container id container_id: Option, + dependency_map: &mut ScopedDependencyMap, ) -> Result { match value { Value::Object(mut m) => { @@ -1957,6 +1909,12 @@ async fn lock_modules_app( .to_string(); let mut logs = "".to_string(); + let relative_imports = extract_relative_imports( + &content, + &format!("{job_path}/app"), + &Some(language.clone()), + ); + if let Some((l, id)) = locks_to_reload .as_ref() .zip(container_id.as_ref()) @@ -1970,6 +1928,15 @@ async fn lock_modules_app( }) { if !l.contains(id) { + dependency_map + .patch( + relative_imports.clone(), + container_id.unwrap_or_default(), + db.begin().await?, + ) + .await? + .commit() + .await?; return Ok(Value::Object(m.clone())); } } else if v @@ -1977,6 +1944,16 @@ async fn lock_modules_app( .is_some_and(|x| !x.as_str().unwrap().trim().is_empty()) { if skip_creating_new_lock(&language, &content) { + dependency_map + .patch( + relative_imports.clone(), + container_id.unwrap_or_default(), + db.begin().await?, + ) + .await? + .commit() + .await?; + logs.push_str( "Found already locked inline script. Skipping lock...\n", ); @@ -2007,44 +1984,15 @@ async fn lock_modules_app( Ok(new_lock) => { append_logs(&job.id, &job.workspace_id, logs, &db.into()).await; - let mut tx = db.begin().await?; - - tx = clear_dependency_map_for_item( - &job_path, - &job.workspace_id, - "app", - tx, - &container_id, - ) - .await?; - - let relative_imports = extract_relative_imports( - &content, - &format!("{job_path}/app"), - &Some(language.clone()), - ); - - if let Some(relative_imports) = relative_imports { - let mut logs = "".to_string(); - logs.push_str( - format!("\n\n--- RELATIVE IMPORTS ---\n\n").as_str(), - ); - - tx = add_relative_imports_to_dependency_map( - &job_path, - &job.workspace_id, - relative_imports, - "app", - tx, - &mut logs, - container_id, + dependency_map + .patch( + relative_imports.clone(), + container_id.unwrap_or_default(), + db.begin().await?, ) + .await? + .commit() .await?; - append_logs(&job.id, &job.workspace_id, logs, &db.into()) - .await; - } - - tx.commit().await?; let anns = windmill_common::worker::TypeScriptAnnotations::parse( @@ -2104,6 +2052,7 @@ async fn lock_modules_app( .and_then(Value::as_str) .map(str::to_owned) .or(container_id.clone()), + dependency_map, ) .await?, ); @@ -2129,6 +2078,7 @@ async fn lock_modules_app( occupancy_metrics, locks_to_reload, container_id.clone(), + dependency_map, ) .await?, ); @@ -2192,6 +2142,17 @@ pub async fn handle_app_dependency_job( .await? .map(|record| (record.app_id, record.value)); + let (_, parent_path) = get_deployment_msg_and_parent_path_from_args(job.args.clone()); + + let mut dependency_map = ScopedDependencyMap::fetch_maybe_rearranged( + &job.workspace_id, + &job_path, + "app", + &parent_path, + db, + ) + .await?; + // TODO: Use transaction for entire segment? if let Some((app_id, value)) = record { let value = lock_modules_app( @@ -2209,9 +2170,17 @@ pub async fn handle_app_dependency_job( occupancy_metrics, &components_to_relock, None, + &mut dependency_map, ) .await?; + // TODO: Dissolve in the end? + dependency_map + .dissolve(db.begin().await?) + .await + .commit() + .await?; + // Compute a lite version of the app value (w/ `inlineScript.{lock,code}`). let mut value_lite = value.clone(); reduce_app(db, &job_path, &mut value_lite, app_id).await?;