From 6fb6249cf9b7848ead0b341125ab3f7b87a1a2c4 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 9 Jun 2026 19:01:52 +0000 Subject: [PATCH] refactor: ungate cascade producers, squash pipeline migrations Co-Authored-By: Claude Fable 5 --- ...9e89cc4a702158899f7fdc66d0922e8fb9b29.json | 23 ++ ...71052c112e6bbb3cf834f16176cbb7e1ac319.json | 4 +- ...9659c1e6a24d5454a691e31e0179f58c24366.json | 41 +++ ...72ece70ec92c5b04baad2622639850445900.json} | 7 +- ...0781c37b753bd3b4a4c7658f8d6ce54c104f9.json | 20 -- ...6b571ae0f3355cfd5eb95a227a2a98136e02b.json | 22 ++ ...839f316fe5412f99f3d3bbbdc0c65f55ab794.json | 20 ++ ...153c43903f929ae5d62fbba12610f89c36d55.json | 2 +- ...988159130fc3a8cde91d419a1787b6bb8a51.json} | 4 +- ...1267af8e42a3b6aa382f1dd483bec7219c67c.json | 12 + ...6e78437a5e4d3b5e2341ec5f7725a28870270.json | 2 +- ...34db7b001d740d524075f19b91bae6fdb41b9.json | 23 ++ ...15edc8e637d44031b83c01ef09a96ade472ae.json | 24 ++ ...0c3a44de6e32651a9a71c1aef49da2696a04f.json | 12 + ...d62eef3ab3b920665ca5d6f9ce0ef89e53fb3.json | 22 ++ ...064b8d536de4f4091548c8c70ce11199c9df3.json | 41 --- ...0421205023_add_asset_materializer.down.sql | 2 - ...260421205023_add_asset_materializer.up.sql | 33 --- ...0421212900_pipeline_folders_index.down.sql | 1 - ...260421212900_pipeline_folders_index.up.sql | 8 - ...041559_pipeline_refactor_triggers.down.sql | 30 -- ...22041559_pipeline_refactor_triggers.up.sql | 31 --- ...616_pipeline_trigger_kinds_expand.down.sql | 12 - ...44616_pipeline_trigger_kinds_expand.up.sql | 11 - .../20260423050000_script_trigger.down.sql | 3 + .../20260423050000_script_trigger.up.sql | 45 +++ ...2_rename_materializer_to_pipeline.down.sql | 7 - ...102_rename_materializer_to_pipeline.up.sql | 14 - ...7_backfill_asset_from_script_assets.up.sql | 2 +- .../windmill-parser/src/asset_parser.rs | 8 +- backend/tests/asset_trigger_dispatch.rs | 9 +- backend/windmill-api/openapi.yaml | 143 ++++++++++ backend/windmill-api/src/jobs.rs | 73 +++-- backend/windmill-queue/src/asset_dispatch.rs | 135 +++++---- .../src/lib/components/ScriptEditor.svelte | 263 +++++++----------- .../AssetGraph/PipelinePickerModal.svelte | 19 +- .../src/lib/components/home/ItemsList.svelte | 21 +- 37 files changed, 641 insertions(+), 508 deletions(-) create mode 100644 backend/.sqlx/query-0a568f630e069118fe302099a709e89cc4a702158899f7fdc66d0922e8fb9b29.json create mode 100644 backend/.sqlx/query-2484323d94f249be30f4472ece89659c1e6a24d5454a691e31e0179f58c24366.json rename backend/.sqlx/{query-1c9cc934ae0784c849e140dabd8cfbc43b7822a4a693220eaaf1317ec5f0c6bd.json => query-26e63135fcd8e7d48e25de190a2f72ece70ec92c5b04baad2622639850445900.json} (67%) delete mode 100644 backend/.sqlx/query-3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9.json create mode 100644 backend/.sqlx/query-45b6c748090a0a6bf71a995413b6b571ae0f3355cfd5eb95a227a2a98136e02b.json create mode 100644 backend/.sqlx/query-4c81384b579bad74b64c72ca053839f316fe5412f99f3d3bbbdc0c65f55ab794.json rename backend/.sqlx/{query-ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce.json => query-77424d40104cf271e5ee5118100a988159130fc3a8cde91d419a1787b6bb8a51.json} (86%) create mode 100644 backend/.sqlx/query-a60306f2bae0702363787c4cf7c1267af8e42a3b6aa382f1dd483bec7219c67c.json create mode 100644 backend/.sqlx/query-c4966cf071a8504f578eed5518134db7b001d740d524075f19b91bae6fdb41b9.json create mode 100644 backend/.sqlx/query-c608cdc9cd2e41992fb86431faa15edc8e637d44031b83c01ef09a96ade472ae.json create mode 100644 backend/.sqlx/query-e9c2e8c50fc45576453885340800c3a44de6e32651a9a71c1aef49da2696a04f.json create mode 100644 backend/.sqlx/query-ecce519d0cf0c31df4612e0ccd8d62eef3ab3b920665ca5d6f9ce0ef89e53fb3.json delete mode 100644 backend/.sqlx/query-f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3.json delete mode 100644 backend/migrations/20260421205023_add_asset_materializer.down.sql delete mode 100644 backend/migrations/20260421205023_add_asset_materializer.up.sql delete mode 100644 backend/migrations/20260421212900_pipeline_folders_index.down.sql delete mode 100644 backend/migrations/20260421212900_pipeline_folders_index.up.sql delete mode 100644 backend/migrations/20260422041559_pipeline_refactor_triggers.down.sql delete mode 100644 backend/migrations/20260422041559_pipeline_refactor_triggers.up.sql delete mode 100644 backend/migrations/20260423044616_pipeline_trigger_kinds_expand.down.sql delete mode 100644 backend/migrations/20260423044616_pipeline_trigger_kinds_expand.up.sql create mode 100644 backend/migrations/20260423050000_script_trigger.down.sql create mode 100644 backend/migrations/20260423050000_script_trigger.up.sql delete mode 100644 backend/migrations/20260430174102_rename_materializer_to_pipeline.down.sql delete mode 100644 backend/migrations/20260430174102_rename_materializer_to_pipeline.up.sql diff --git a/backend/.sqlx/query-0a568f630e069118fe302099a709e89cc4a702158899f7fdc66d0922e8fb9b29.json b/backend/.sqlx/query-0a568f630e069118fe302099a709e89cc4a702158899f7fdc66d0922e8fb9b29.json new file mode 100644 index 0000000000..c407574e15 --- /dev/null +++ b/backend/.sqlx/query-0a568f630e069118fe302099a709e89cc4a702158899f7fdc66d0922e8fb9b29.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path FROM flow WHERE workspace_id = $1 AND path = ANY($2) AND archived = false", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "TextArray" + ] + }, + "nullable": [ + false + ] + }, + "hash": "0a568f630e069118fe302099a709e89cc4a702158899f7fdc66d0922e8fb9b29" +} diff --git a/backend/.sqlx/query-19513c4158267cc7fe10d999ad571052c112e6bbb3cf834f16176cbb7e1ac319.json b/backend/.sqlx/query-19513c4158267cc7fe10d999ad571052c112e6bbb3cf834f16176cbb7e1ac319.json index 8944a6001f..dfc8540468 100644 --- a/backend/.sqlx/query-19513c4158267cc7fe10d999ad571052c112e6bbb3cf834f16176cbb7e1ac319.json +++ b/backend/.sqlx/query-19513c4158267cc7fe10d999ad571052c112e6bbb3cf834f16176cbb7e1ac319.json @@ -5,7 +5,7 @@ "columns": [ { "ordinal": 0, - "name": "id", + "name": "id!", "type_info": "Uuid" } ], @@ -16,7 +16,7 @@ ] }, "nullable": [ - false + null ] }, "hash": "19513c4158267cc7fe10d999ad571052c112e6bbb3cf834f16176cbb7e1ac319" diff --git a/backend/.sqlx/query-2484323d94f249be30f4472ece89659c1e6a24d5454a691e31e0179f58c24366.json b/backend/.sqlx/query-2484323d94f249be30f4472ece89659c1e6a24d5454a691e31e0179f58c24366.json new file mode 100644 index 0000000000..e350cc8233 --- /dev/null +++ b/backend/.sqlx/query-2484323d94f249be30f4472ece89659c1e6a24d5454a691e31e0179f58c24366.json @@ -0,0 +1,41 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT kind, path, script_path, is_flow FROM (\n SELECT 'schedule' AS kind, path, script_path, is_flow FROM schedule\n WHERE workspace_id = $1\n AND script_path IS NOT NULL\n UNION ALL\n SELECT 'email', path, script_path, is_flow FROM email_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'kafka', path, script_path, is_flow FROM kafka_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'mqtt', path, script_path, is_flow FROM mqtt_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'nats', path, script_path, is_flow FROM nats_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'postgres', path, script_path, is_flow FROM postgres_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'sqs', path, script_path, is_flow FROM sqs_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'gcp', path, script_path, is_flow FROM gcp_trigger\n WHERE workspace_id = $1\n ) t\n WHERE ($2::text IS NULL OR script_path LIKE $2)\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "kind", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "script_path", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "is_flow", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null, + null, + null, + null + ] + }, + "hash": "2484323d94f249be30f4472ece89659c1e6a24d5454a691e31e0179f58c24366" +} diff --git a/backend/.sqlx/query-1c9cc934ae0784c849e140dabd8cfbc43b7822a4a693220eaaf1317ec5f0c6bd.json b/backend/.sqlx/query-26e63135fcd8e7d48e25de190a2f72ece70ec92c5b04baad2622639850445900.json similarity index 67% rename from backend/.sqlx/query-1c9cc934ae0784c849e140dabd8cfbc43b7822a4a693220eaaf1317ec5f0c6bd.json rename to backend/.sqlx/query-26e63135fcd8e7d48e25de190a2f72ece70ec92c5b04baad2622639850445900.json index e2591e8636..02a419de06 100644 --- a/backend/.sqlx/query-1c9cc934ae0784c849e140dabd8cfbc43b7822a4a693220eaaf1317ec5f0c6bd.json +++ b/backend/.sqlx/query-26e63135fcd8e7d48e25de190a2f72ece70ec92c5b04baad2622639850445900.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT\n e.subscriber_path AS \"subscriber_path!\",\n e.asset_kind AS \"asset_kind!: windmill_common::assets::AssetKind\",\n e.asset_path AS \"asset_path!\",\n e.outcome::text AS \"outcome!\",\n e.child_job_id,\n e.partition,\n e.received_inputs,\n e.required_inputs,\n e.debounce_s,\n e.reason,\n e.created_at AS \"created_at!\"\n FROM dispatch_event e\n JOIN v2_job j ON j.id = e.producer_job_id\n WHERE e.producer_job_id = $1\n AND e.workspace_id = $2\n AND ($3::text[] IS NULL OR j.tag = ANY($3))\n ORDER BY e.id", + "query": "SELECT\n subscriber_path AS \"subscriber_path!\",\n asset_kind AS \"asset_kind!: windmill_common::assets::AssetKind\",\n asset_path AS \"asset_path!\",\n outcome::text AS \"outcome!\",\n child_job_id,\n partition,\n received_inputs,\n required_inputs,\n debounce_s,\n reason,\n created_at AS \"created_at!\"\n FROM dispatch_event\n WHERE producer_job_id = $1 AND workspace_id = $2\n ORDER BY id", "describe": { "columns": [ { @@ -76,8 +76,7 @@ "parameters": { "Left": [ "Uuid", - "Text", - "TextArray" + "Text" ] }, "nullable": [ @@ -94,5 +93,5 @@ false ] }, - "hash": "1c9cc934ae0784c849e140dabd8cfbc43b7822a4a693220eaaf1317ec5f0c6bd" + "hash": "26e63135fcd8e7d48e25de190a2f72ece70ec92c5b04baad2622639850445900" } diff --git a/backend/.sqlx/query-3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9.json b/backend/.sqlx/query-3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9.json deleted file mode 100644 index 843efefd4f..0000000000 --- a/backend/.sqlx/query-3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO schedule (\n workspace_id, path, schedule, timezone, edited_by, script_path,\n is_flow, enabled, email, permissioned_as,\n ws_error_handler_muted, no_flow_overlap, cron_version,\n managed\n )\n VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', true)\n ON CONFLICT (workspace_id, path) DO UPDATE\n SET schedule = EXCLUDED.schedule,\n edited_at = now(),\n edited_by = EXCLUDED.edited_by,\n managed = true\n WHERE schedule.managed\n OR schedule.script_path = EXCLUDED.script_path\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Varchar", - "Varchar", - "Bool", - "Varchar", - "Varchar" - ] - }, - "nullable": [] - }, - "hash": "3fdb89d2eaccb78991c6741bb1f0781c37b753bd3b4a4c7658f8d6ce54c104f9" -} diff --git a/backend/.sqlx/query-45b6c748090a0a6bf71a995413b6b571ae0f3355cfd5eb95a227a2a98136e02b.json b/backend/.sqlx/query-45b6c748090a0a6bf71a995413b6b571ae0f3355cfd5eb95a227a2a98136e02b.json new file mode 100644 index 0000000000..0d9e4e859d --- /dev/null +++ b/backend/.sqlx/query-45b6c748090a0a6bf71a995413b6b571ae0f3355cfd5eb95a227a2a98136e02b.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path AS \"path!\" FROM flow WHERE workspace_id = $1 AND archived = false", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path!", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "45b6c748090a0a6bf71a995413b6b571ae0f3355cfd5eb95a227a2a98136e02b" +} diff --git a/backend/.sqlx/query-4c81384b579bad74b64c72ca053839f316fe5412f99f3d3bbbdc0c65f55ab794.json b/backend/.sqlx/query-4c81384b579bad74b64c72ca053839f316fe5412f99f3d3bbbdc0c65f55ab794.json new file mode 100644 index 0000000000..9d8e1e15cc --- /dev/null +++ b/backend/.sqlx/query-4c81384b579bad74b64c72ca053839f316fe5412f99f3d3bbbdc0c65f55ab794.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT has_changes FROM workspace_diff\n WHERE path = 'f/shared/renamed_away' AND kind = 'script' AND source_workspace_id = 'test-workspace'", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "has_changes", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + true + ] + }, + "hash": "4c81384b579bad74b64c72ca053839f316fe5412f99f3d3bbbdc0c65f55ab794" +} diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce.json b/backend/.sqlx/query-77424d40104cf271e5ee5118100a988159130fc3a8cde91d419a1787b6bb8a51.json similarity index 86% rename from backend/.sqlx/query-ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce.json rename to backend/.sqlx/query-77424d40104cf271e5ee5118100a988159130fc3a8cde91d419a1787b6bb8a51.json index 153532085e..0768563002 100644 --- a/backend/.sqlx/query-ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce.json +++ b/backend/.sqlx/query-77424d40104cf271e5ee5118100a988159130fc3a8cde91d419a1787b6bb8a51.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT\n runnable_kind AS \"runnable_kind!: AssetUsageKind\",\n runnable_path AS \"runnable_path!\",\n trigger_kind::text AS \"trigger_kind!\",\n trigger_ref AS \"trigger_ref!\"\n FROM script_trigger\n WHERE workspace_id = $1\n AND trigger_kind IN ('asset', 'schedule')\n AND ($2::text IS NULL OR runnable_path LIKE $2)\n ", + "query": "\n SELECT\n runnable_kind AS \"runnable_kind!: AssetUsageKind\",\n runnable_path AS \"runnable_path!\",\n trigger_kind::text AS \"trigger_kind!\",\n trigger_ref AS \"trigger_ref!\"\n FROM script_trigger\n WHERE workspace_id = $1\n AND trigger_kind = 'asset'\n AND ($2::text IS NULL OR runnable_path LIKE $2)\n ", "describe": { "columns": [ { @@ -48,5 +48,5 @@ false ] }, - "hash": "ad75a63f5b757de1f4bb9789f8aa7ea25e1dd949ca957a8d5bbfd1166d4b80ce" + "hash": "77424d40104cf271e5ee5118100a988159130fc3a8cde91d419a1787b6bb8a51" } diff --git a/backend/.sqlx/query-a60306f2bae0702363787c4cf7c1267af8e42a3b6aa382f1dd483bec7219c67c.json b/backend/.sqlx/query-a60306f2bae0702363787c4cf7c1267af8e42a3b6aa382f1dd483bec7219c67c.json new file mode 100644 index 0000000000..c726a6df9c --- /dev/null +++ b/backend/.sqlx/query-a60306f2bae0702363787c4cf7c1267af8e42a3b6aa382f1dd483bec7219c67c.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO script (workspace_id, path, hash, content, summary, description, language, created_by, created_at, archived, schema_validation, ws_error_handler_muted, deleted)\n VALUES ('wm-fork-test-workspace', 'f/shared/renamed_away', 67890, 'def main(): return 1', '', '', 'python3', 'test@windmill.dev', NOW(), true, false, false, false)", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "a60306f2bae0702363787c4cf7c1267af8e42a3b6aa382f1dd483bec7219c67c" +} diff --git a/backend/.sqlx/query-afb0762c88d9232b79090f2e5966e78437a5e4d3b5e2341ec5f7725a28870270.json b/backend/.sqlx/query-afb0762c88d9232b79090f2e5966e78437a5e4d3b5e2341ec5f7725a28870270.json index f0c91daf16..b492b38f49 100644 --- a/backend/.sqlx/query-afb0762c88d9232b79090f2e5966e78437a5e4d3b5e2341ec5f7725a28870270.json +++ b/backend/.sqlx/query-afb0762c88d9232b79090f2e5966e78437a5e4d3b5e2341ec5f7725a28870270.json @@ -13,4 +13,4 @@ "nullable": [] }, "hash": "afb0762c88d9232b79090f2e5966e78437a5e4d3b5e2341ec5f7725a28870270" -} \ No newline at end of file +} diff --git a/backend/.sqlx/query-c4966cf071a8504f578eed5518134db7b001d740d524075f19b91bae6fdb41b9.json b/backend/.sqlx/query-c4966cf071a8504f578eed5518134db7b001d740d524075f19b91bae6fdb41b9.json new file mode 100644 index 0000000000..2a720785aa --- /dev/null +++ b/backend/.sqlx/query-c4966cf071a8504f578eed5518134db7b001d740d524075f19b91bae6fdb41b9.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT path FROM script WHERE workspace_id = $1 AND path = ANY($2) AND archived = false", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "TextArray" + ] + }, + "nullable": [ + false + ] + }, + "hash": "c4966cf071a8504f578eed5518134db7b001d740d524075f19b91bae6fdb41b9" +} diff --git a/backend/.sqlx/query-c608cdc9cd2e41992fb86431faa15edc8e637d44031b83c01ef09a96ade472ae.json b/backend/.sqlx/query-c608cdc9cd2e41992fb86431faa15edc8e637d44031b83c01ef09a96ade472ae.json new file mode 100644 index 0000000000..247d4de14a --- /dev/null +++ b/backend/.sqlx/query-c608cdc9cd2e41992fb86431faa15edc8e637d44031b83c01ef09a96ade472ae.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT created_by AS \"created_by!\"\n FROM v2_job\n WHERE id = $1 AND workspace_id = $2 AND ($3::text[] IS NULL OR tag = ANY($3))", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "created_by!", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text", + "TextArray" + ] + }, + "nullable": [ + false + ] + }, + "hash": "c608cdc9cd2e41992fb86431faa15edc8e637d44031b83c01ef09a96ade472ae" +} diff --git a/backend/.sqlx/query-e9c2e8c50fc45576453885340800c3a44de6e32651a9a71c1aef49da2696a04f.json b/backend/.sqlx/query-e9c2e8c50fc45576453885340800c3a44de6e32651a9a71c1aef49da2696a04f.json new file mode 100644 index 0000000000..738c6a473f --- /dev/null +++ b/backend/.sqlx/query-e9c2e8c50fc45576453885340800c3a44de6e32651a9a71c1aef49da2696a04f.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO workspace_diff\n (source_workspace_id, fork_workspace_id, path, kind, ahead, behind, has_changes, exists_in_source, exists_in_fork)\n VALUES ('test-workspace', 'wm-fork-test-workspace', 'f/shared/renamed_away', 'script', 1, 0, true, false, true)", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "e9c2e8c50fc45576453885340800c3a44de6e32651a9a71c1aef49da2696a04f" +} diff --git a/backend/.sqlx/query-ecce519d0cf0c31df4612e0ccd8d62eef3ab3b920665ca5d6f9ce0ef89e53fb3.json b/backend/.sqlx/query-ecce519d0cf0c31df4612e0ccd8d62eef3ab3b920665ca5d6f9ce0ef89e53fb3.json new file mode 100644 index 0000000000..d40882073d --- /dev/null +++ b/backend/.sqlx/query-ecce519d0cf0c31df4612e0ccd8d62eef3ab3b920665ca5d6f9ce0ef89e53fb3.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path AS \"path!\" FROM script\n WHERE workspace_id = $1\n AND archived = false\n AND deleted = false", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path!", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "ecce519d0cf0c31df4612e0ccd8d62eef3ab3b920665ca5d6f9ce0ef89e53fb3" +} diff --git a/backend/.sqlx/query-f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3.json b/backend/.sqlx/query-f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3.json deleted file mode 100644 index 34b3be608b..0000000000 --- a/backend/.sqlx/query-f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3.json +++ /dev/null @@ -1,41 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT kind, path, script_path, is_flow FROM (\n SELECT 'email' AS kind, path, script_path, is_flow FROM email_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'kafka', path, script_path, is_flow FROM kafka_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'mqtt', path, script_path, is_flow FROM mqtt_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'nats', path, script_path, is_flow FROM nats_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'postgres', path, script_path, is_flow FROM postgres_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'sqs', path, script_path, is_flow FROM sqs_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'gcp', path, script_path, is_flow FROM gcp_trigger\n WHERE workspace_id = $1\n ) t\n WHERE ($2::text IS NULL OR script_path LIKE $2)\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "kind", - "type_info": "Text" - }, - { - "ordinal": 1, - "name": "path", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "script_path", - "type_info": "Varchar" - }, - { - "ordinal": 3, - "name": "is_flow", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - null, - null, - null, - null - ] - }, - "hash": "f360806ee1b8e53f16800219b00064b8d536de4f4091548c8c70ce11199c9df3" -} diff --git a/backend/migrations/20260421205023_add_asset_materializer.down.sql b/backend/migrations/20260421205023_add_asset_materializer.down.sql deleted file mode 100644 index 318351178e..0000000000 --- a/backend/migrations/20260421205023_add_asset_materializer.down.sql +++ /dev/null @@ -1,2 +0,0 @@ -DROP TABLE IF EXISTS asset_materializer_history; -DROP TABLE IF EXISTS asset_materializer; diff --git a/backend/migrations/20260421205023_add_asset_materializer.up.sql b/backend/migrations/20260421205023_add_asset_materializer.up.sql deleted file mode 100644 index 91f661c770..0000000000 --- a/backend/migrations/20260421205023_add_asset_materializer.up.sql +++ /dev/null @@ -1,33 +0,0 @@ --- Current materializer: exactly one runnable owns each (workspace, kind, path) asset. --- Updated via last-deploy-wins on `// materialize ` annotations. -CREATE TABLE asset_materializer ( - workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, - asset_kind ASSET_KIND NOT NULL, - asset_path VARCHAR(255) NOT NULL, - runnable_kind ASSET_USAGE_KIND NOT NULL, - runnable_path VARCHAR(255) NOT NULL, - deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - deployed_by VARCHAR(50) NOT NULL, - PRIMARY KEY (workspace_id, asset_kind, asset_path) -); - -CREATE INDEX idx_asset_materializer_runnable - ON asset_materializer (workspace_id, runnable_kind, runnable_path); - --- Append-only history. Every claim (including takeovers) gets a row. --- Revocations (redeploy without annotation) are represented by absence in --- `asset_materializer` rather than a row here, so "who last materialized X" --- is the most recent history row for X regardless of current state. -CREATE TABLE asset_materializer_history ( - id BIGSERIAL PRIMARY KEY, - workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, - asset_kind ASSET_KIND NOT NULL, - asset_path VARCHAR(255) NOT NULL, - runnable_kind ASSET_USAGE_KIND NOT NULL, - runnable_path VARCHAR(255) NOT NULL, - deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - deployed_by VARCHAR(50) NOT NULL -); - -CREATE INDEX idx_asset_materializer_history_asset - ON asset_materializer_history (workspace_id, asset_kind, asset_path, deployed_at DESC); diff --git a/backend/migrations/20260421212900_pipeline_folders_index.down.sql b/backend/migrations/20260421212900_pipeline_folders_index.down.sql deleted file mode 100644 index 6c5f740b1d..0000000000 --- a/backend/migrations/20260421212900_pipeline_folders_index.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP INDEX IF EXISTS idx_script_materializer_path; diff --git a/backend/migrations/20260421212900_pipeline_folders_index.up.sql b/backend/migrations/20260421212900_pipeline_folders_index.up.sql deleted file mode 100644 index 053a635b45..0000000000 --- a/backend/migrations/20260421212900_pipeline_folders_index.up.sql +++ /dev/null @@ -1,8 +0,0 @@ --- Fast lookups for: --- 1. "does folder F have a pipeline?" (exists check on prefix) --- 2. "list all folders with a pipeline" (distinct folder from path) --- The partial predicate keeps the index tiny on workspaces with few --- materializer scripts, and text_pattern_ops lets 'f/foo/%' LIKE scans use it. -CREATE INDEX IF NOT EXISTS idx_script_materializer_path - ON script (workspace_id, path text_pattern_ops) - WHERE auto_kind = 'materializer' AND archived = false AND deleted = false; diff --git a/backend/migrations/20260422041559_pipeline_refactor_triggers.down.sql b/backend/migrations/20260422041559_pipeline_refactor_triggers.down.sql deleted file mode 100644 index b96eaf8041..0000000000 --- a/backend/migrations/20260422041559_pipeline_refactor_triggers.down.sql +++ /dev/null @@ -1,30 +0,0 @@ -DROP TABLE IF EXISTS script_trigger; -DROP TYPE IF EXISTS SCRIPT_TRIGGER_KIND; - --- Recreate the materializer tables so rolling back to the pre-refactor --- backend code boots. Data is gone either way (irrecoverable). -CREATE TABLE asset_materializer ( - workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, - asset_kind ASSET_KIND NOT NULL, - asset_path VARCHAR(255) NOT NULL, - runnable_kind ASSET_USAGE_KIND NOT NULL, - runnable_path VARCHAR(255) NOT NULL, - deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - deployed_by VARCHAR(50) NOT NULL, - PRIMARY KEY (workspace_id, asset_kind, asset_path) -); -CREATE INDEX idx_asset_materializer_runnable - ON asset_materializer (workspace_id, runnable_kind, runnable_path); - -CREATE TABLE asset_materializer_history ( - id BIGSERIAL PRIMARY KEY, - workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, - asset_kind ASSET_KIND NOT NULL, - asset_path VARCHAR(255) NOT NULL, - runnable_kind ASSET_USAGE_KIND NOT NULL, - runnable_path VARCHAR(255) NOT NULL, - deployed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - deployed_by VARCHAR(50) NOT NULL -); -CREATE INDEX idx_asset_materializer_history_asset - ON asset_materializer_history (workspace_id, asset_kind, asset_path, deployed_at DESC); diff --git a/backend/migrations/20260422041559_pipeline_refactor_triggers.up.sql b/backend/migrations/20260422041559_pipeline_refactor_triggers.up.sql deleted file mode 100644 index aea9c7ebbd..0000000000 --- a/backend/migrations/20260422041559_pipeline_refactor_triggers.up.sql +++ /dev/null @@ -1,31 +0,0 @@ --- Simplified pipeline model: `// materialize` is a bare opt-in marker (sets --- auto_kind='materializer') and the per-asset ownership tracking goes away. --- Writes are already tracked in the `asset` table via the parser, so the --- `asset_materializer*` tables no longer earn their keep. -DROP TABLE IF EXISTS asset_materializer_history; -DROP TABLE IF EXISTS asset_materializer; - --- Execution DAG edges declared via `// on ` annotations. --- For `trigger_kind='asset'`: trigger_ref is `://` (kind from --- parse_asset_syntax, so downstream lookups match the `asset` table). --- For `trigger_kind='schedule'`: trigger_ref is the raw cron expression. -CREATE TYPE SCRIPT_TRIGGER_KIND AS ENUM ('asset', 'schedule'); - -CREATE TABLE script_trigger ( - id BIGSERIAL PRIMARY KEY, - workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, - runnable_kind ASSET_USAGE_KIND NOT NULL, - runnable_path VARCHAR(255) NOT NULL, - trigger_kind SCRIPT_TRIGGER_KIND NOT NULL, - trigger_ref TEXT NOT NULL -); - --- Per-runnable lookup (wipe-on-deploy, list-triggers-for-script). -CREATE INDEX idx_script_trigger_runnable - ON script_trigger (workspace_id, runnable_kind, runnable_path); - --- Reverse lookup: "which scripts are triggered by asset X?" (the asset → script --- edges in the graph). trigger_ref is unbounded text so can't share the --- asset_kind btree, but this covers the common prefix-scan use case. -CREATE INDEX idx_script_trigger_ref - ON script_trigger (workspace_id, trigger_kind, trigger_ref); diff --git a/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.down.sql b/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.down.sql deleted file mode 100644 index c2b2966e7d..0000000000 --- a/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.down.sql +++ /dev/null @@ -1,12 +0,0 @@ --- Postgres doesn't support removing enum values in-place. The only safe --- rollback is to recreate the type with the original set and rewrite the --- column, deleting any rows using values introduced in the up migration. -DELETE FROM script_trigger - WHERE trigger_kind NOT IN ('asset', 'schedule'); - -CREATE TYPE SCRIPT_TRIGGER_KIND_OLD AS ENUM ('asset', 'schedule'); -ALTER TABLE script_trigger - ALTER COLUMN trigger_kind TYPE SCRIPT_TRIGGER_KIND_OLD - USING trigger_kind::text::SCRIPT_TRIGGER_KIND_OLD; -DROP TYPE SCRIPT_TRIGGER_KIND; -ALTER TYPE SCRIPT_TRIGGER_KIND_OLD RENAME TO SCRIPT_TRIGGER_KIND; diff --git a/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.up.sql b/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.up.sql deleted file mode 100644 index 6d440b782c..0000000000 --- a/backend/migrations/20260423044616_pipeline_trigger_kinds_expand.up.sql +++ /dev/null @@ -1,11 +0,0 @@ --- Expand the pipeline trigger enum to cover every non-integration --- (i.e. non-native) trigger kind Windmill supports. Each value mirrors a --- keyword the annotation parser recognises in `// on ` lines. -ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'webhook'; -ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'email'; -ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'kafka'; -ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'mqtt'; -ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'nats'; -ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'postgres'; -ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'sqs'; -ALTER TYPE SCRIPT_TRIGGER_KIND ADD VALUE IF NOT EXISTS 'gcp'; diff --git a/backend/migrations/20260423050000_script_trigger.down.sql b/backend/migrations/20260423050000_script_trigger.down.sql new file mode 100644 index 0000000000..5171e8f2d0 --- /dev/null +++ b/backend/migrations/20260423050000_script_trigger.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idx_script_pipeline_path; +DROP TABLE IF EXISTS script_trigger; +DROP TYPE IF EXISTS SCRIPT_TRIGGER_KIND; diff --git a/backend/migrations/20260423050000_script_trigger.up.sql b/backend/migrations/20260423050000_script_trigger.up.sql new file mode 100644 index 0000000000..9abc42b1f1 --- /dev/null +++ b/backend/migrations/20260423050000_script_trigger.up.sql @@ -0,0 +1,45 @@ +-- Execution DAG edges declared via `// on ` +-- annotations. +-- For `trigger_kind='asset'`: trigger_ref is `://` (kind from +-- parse_asset_syntax, so downstream lookups match the `asset` table). +-- The other kinds mirror the keywords the annotation parser recognises in +-- `// on ` lines (every non-integration trigger kind; their +-- trigger_ref is the trigger row path, or empty for marker-only forms). +-- +-- The idempotency guards (IF NOT EXISTS / duplicate_object) are load-bearing: +-- this migration squashes several pre-release ones, so databases migrated +-- from the unsquashed history already contain the final objects. +DO $$ BEGIN + CREATE TYPE SCRIPT_TRIGGER_KIND AS ENUM ( + 'asset', 'schedule', 'webhook', 'email', 'kafka', 'mqtt', 'nats', + 'postgres', 'sqs', 'gcp'); +EXCEPTION WHEN duplicate_object THEN NULL; +END $$; + +CREATE TABLE IF NOT EXISTS script_trigger ( + id BIGSERIAL PRIMARY KEY, + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + runnable_kind ASSET_USAGE_KIND NOT NULL, + runnable_path VARCHAR(255) NOT NULL, + trigger_kind SCRIPT_TRIGGER_KIND NOT NULL, + trigger_ref TEXT NOT NULL +); + +-- Per-runnable lookup (wipe-on-deploy, list-triggers-for-script). +CREATE INDEX IF NOT EXISTS idx_script_trigger_runnable + ON script_trigger (workspace_id, runnable_kind, runnable_path); + +-- Reverse lookup: "which scripts are triggered by asset X?" (the asset → script +-- edges in the graph). trigger_ref is unbounded text so can't share the +-- asset_kind btree, but this covers the common prefix-scan use case. +CREATE INDEX IF NOT EXISTS idx_script_trigger_ref + ON script_trigger (workspace_id, trigger_kind, trigger_ref); + +-- Fast lookups for: +-- 1. "does folder F have a pipeline?" (exists check on prefix) +-- 2. "list all folders with a pipeline" (distinct folder from path) +-- The partial predicate keeps the index tiny on workspaces with few +-- pipeline scripts, and text_pattern_ops lets 'f/foo/%' LIKE scans use it. +CREATE INDEX IF NOT EXISTS idx_script_pipeline_path + ON script (workspace_id, path text_pattern_ops) + WHERE auto_kind = 'pipeline' AND archived = false AND deleted = false; diff --git a/backend/migrations/20260430174102_rename_materializer_to_pipeline.down.sql b/backend/migrations/20260430174102_rename_materializer_to_pipeline.down.sql deleted file mode 100644 index f81f9cb2bd..0000000000 --- a/backend/migrations/20260430174102_rename_materializer_to_pipeline.down.sql +++ /dev/null @@ -1,7 +0,0 @@ --- Revert: rename pipeline-membership marker back to materializer. -DROP INDEX IF EXISTS idx_script_pipeline_path; -CREATE INDEX IF NOT EXISTS idx_script_materializer_path - ON script (workspace_id, path text_pattern_ops) - WHERE auto_kind = 'materializer' AND archived = false AND deleted = false; - -UPDATE script SET auto_kind = 'materializer' WHERE auto_kind = 'pipeline'; diff --git a/backend/migrations/20260430174102_rename_materializer_to_pipeline.up.sql b/backend/migrations/20260430174102_rename_materializer_to_pipeline.up.sql deleted file mode 100644 index 3f1968b7a8..0000000000 --- a/backend/migrations/20260430174102_rename_materializer_to_pipeline.up.sql +++ /dev/null @@ -1,14 +0,0 @@ --- Rename the pipeline-membership marker from `materializer` to `pipeline` --- to better reflect that membership is broader than materialization --- (test/notify/cleanup scripts are members too without producing assets). --- The change is text-only — `auto_kind` is a varchar column, not an enum, --- so this is just a value swap plus an index rebuild. - -UPDATE script SET auto_kind = 'pipeline' WHERE auto_kind = 'materializer'; - --- Rebuild the partial index that backs the folder picker and graph-scope --- query. The index name and predicate both move to the new keyword. -DROP INDEX IF EXISTS idx_script_materializer_path; -CREATE INDEX IF NOT EXISTS idx_script_pipeline_path - ON script (workspace_id, path text_pattern_ops) - WHERE auto_kind = 'pipeline' AND archived = false AND deleted = false; diff --git a/backend/migrations/20260523091727_backfill_asset_from_script_assets.up.sql b/backend/migrations/20260523091727_backfill_asset_from_script_assets.up.sql index bf6c537a2a..ecc77fcf0b 100644 --- a/backend/migrations/20260523091727_backfill_asset_from_script_assets.up.sql +++ b/backend/migrations/20260523091727_backfill_asset_from_script_assets.up.sql @@ -5,7 +5,7 @@ -- (asset-trigger cascade) and the asset-graph lineage view actually query. -- Without this, a pre-feature script "succeeds" but the dispatcher sees -- no writes, no subscribers are matched, and the dispatch_event panel --- stays empty — exactly the gap we hit in https://… (no public link yet). +-- stays empty. -- -- Idempotent: scoped to (workspace, path) pairs that have ZERO asset rows -- under `usage_kind = 'script'`, so re-running can't duplicate. The diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 5525bb8475..1ac1638959 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -472,8 +472,8 @@ fn parse_kv_opts(s: &str) -> BTreeMap { // whose first non-whitespace tokens are a comment prefix (`//`, `#`, or // `--`) followed by one of the recognized keywords: // - `pipeline` → opt-in marker (must be alone on the line) -// - `schedule ""` → top-level inline cron schedule -// - `on ` → asset / native trigger edge +// - `on ` → asset / native trigger edge (including +// the marker-only `on schedule` form) // - `partitioned [opts]` → partition declaration // - `freshness ` → SLA / active backstop // - `tag ` → worker-tag override (annotation wins @@ -837,7 +837,7 @@ mod pipeline_annotation_tests { let out = parse_pipeline_annotations( "// on s3://lake/raw/{partition}/events.parquet\n\ // on s3://lake/dim/customers.parquet\n\ - // schedule \"@daily\"", + // on schedule", ); assert_eq!(out.triggers.len(), 3); assert!(out.triggers[0].is_partition_bearing()); @@ -1064,7 +1064,7 @@ mod pipeline_annotation_tests { fn combined() { let code = concat!( "// pipeline\n", - "// schedule \"0 0 * * *\"\n", + "// on schedule\n", "// on s3://in.csv\n", "// partitioned daily tz=\"UTC\"\n", "// freshness 2h\n", diff --git a/backend/tests/asset_trigger_dispatch.rs b/backend/tests/asset_trigger_dispatch.rs index 80f7fe351a..3143d59465 100644 --- a/backend/tests/asset_trigger_dispatch.rs +++ b/backend/tests/asset_trigger_dispatch.rs @@ -32,10 +32,13 @@ async fn seed_script( content: &str, language: &str, ) -> anyhow::Result { - // Hash needs to be unique per (workspace, hash). Derive from path so - // sibling scripts in the same test don't collide. + // Hash needs to be unique per (workspace, hash). Derive from path AND + // content: the worker's script cache (`cache::script::fetch`) is keyed + // by hash alone and is process-global, so tests running in the same + // process that seed the same path with different content would poison + // each other's cache if the hash came from the path only. let mut h = 0i64; - for b in path.bytes() { + for b in path.bytes().chain(content.bytes()) { h = h.wrapping_mul(31).wrapping_add(b as i64); } sqlx::query( diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index f651692005..2263d05d7d 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -20426,6 +20426,149 @@ paths: type: string description: The asset path + /w/{workspace}/assets/graph: + get: + summary: Get the workspace-wide asset <-> runnable graph + operationId: getAssetsGraph + tags: + - asset + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: asset_kinds + in: query + description: Filter by asset kinds (comma-separated list) + schema: + type: string + - name: folder + in: query + description: Scope the graph to runnables in a single folder + schema: + type: string + responses: + "200": + description: asset graph nodes, lineage edges and trigger edges + content: + application/json: + schema: + type: object + required: [assets, runnables, edges, triggers] + properties: + assets: + type: array + items: + type: object + required: [kind, path] + properties: + kind: + $ref: "#/components/schemas/AssetKind" + path: + type: string + runnables: + type: array + items: + type: object + required: [path, usage_kind] + properties: + path: + type: string + usage_kind: + $ref: "#/components/schemas/AssetUsageKind" + in_pipeline: + type: boolean + description: True iff the script is a pipeline member (deployed with `// pipeline`). Omitted when false. + edges: + type: array + items: + type: object + required: + [runnable_path, runnable_kind, asset_kind, asset_path] + properties: + runnable_path: + type: string + runnable_kind: + $ref: "#/components/schemas/AssetUsageKind" + asset_kind: + $ref: "#/components/schemas/AssetKind" + asset_path: + type: string + access_type: + $ref: "#/components/schemas/AssetUsageAccessType" + triggers: + type: array + items: + oneOf: + - type: object + description: Asset trigger edge (`// on `) + required: + [ + trigger_kind, + asset_kind, + asset_path, + runnable_kind, + runnable_path, + ] + properties: + trigger_kind: + type: string + enum: [asset] + asset_kind: + $ref: "#/components/schemas/AssetKind" + asset_path: + type: string + runnable_kind: + $ref: "#/components/schemas/AssetUsageKind" + runnable_path: + type: string + - type: object + description: Native trigger edge (schedule, email, kafka, ...). `path` is the trigger row's path. + required: + [trigger_kind, path, runnable_kind, runnable_path] + properties: + trigger_kind: + type: string + enum: + - schedule + - email + - kafka + - mqtt + - nats + - postgres + - sqs + - gcp + path: + type: string + runnable_kind: + $ref: "#/components/schemas/AssetUsageKind" + runnable_path: + type: string + + /w/{workspace}/assets/pipelines: + get: + summary: List folders that contain at least one pipeline-member script + operationId: listPipelineFolders + tags: + - asset + parameters: + - $ref: "#/components/parameters/WorkspaceId" + responses: + "200": + description: folders containing pipeline scripts, with their script counts + content: + application/json: + schema: + type: array + items: + type: object + required: [folder, script_count] + properties: + folder: + type: string + description: The folder name (without the `f/` prefix) + script_count: + type: integer + format: int64 + description: Number of pipeline-member scripts in the folder + /w/{workspace}/volumes/list: get: summary: List all volumes in the workspace diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 891fc3c9bf..b327476c86 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -8943,8 +8943,10 @@ struct DispatchEvent { } async fn get_dispatch_events( + OptViewToken(view_token): OptViewToken, OptAuthed(opt_authed): OptAuthed, Extension(db): Extension, + Extension(user_db): Extension, Path((w_id, id)): Path<(String, Uuid)>, ) -> error::JsonResult> { let tags = opt_authed @@ -8952,32 +8954,59 @@ async fn get_dispatch_events( .map(|authed| get_scope_tags(authed)) .flatten(); - // Tag-gate via the producer's v2_job row so token scopes that restrict - // job visibility apply here too. The dispatch_event FK to v2_job(id) - // guarantees the producer row exists for any extant event. - let rows = sqlx::query!( - r#"SELECT - e.subscriber_path AS "subscriber_path!", - e.asset_kind AS "asset_kind!: windmill_common::assets::AssetKind", - e.asset_path AS "asset_path!", - e.outcome::text AS "outcome!", - e.child_job_id, - e.partition, - e.received_inputs, - e.required_inputs, - e.debounce_s, - e.reason, - e.created_at AS "created_at!" - FROM dispatch_event e - JOIN v2_job j ON j.id = e.producer_job_id - WHERE e.producer_job_id = $1 - AND e.workspace_id = $2 - AND ($3::text[] IS NULL OR j.tag = ANY($3)) - ORDER BY e.id"#, + // Gate on the producer job's visibility, exactly like + // get_completed_job_timing on the same unauthed router: scope tags + // first, then per-job read access for authed users, anonymous-only + // jobs otherwise. The dispatch_event FK to v2_job(id) guarantees the + // producer row exists for any extant event. + let producer = sqlx::query!( + r#"SELECT created_by AS "created_by!" + FROM v2_job + WHERE id = $1 AND workspace_id = $2 AND ($3::text[] IS NULL OR tag = ANY($3))"#, id, &w_id, tags.as_ref().map(|v| v.as_slice()) as Option<&[&str]>, ) + .fetch_optional(&db) + .await?; + let producer = not_found_if_none(producer, "Job", id.to_string())?; + + if let Some(authed) = opt_authed.as_ref() { + require_job_read_access( + &db, + &user_db, + authed, + &w_id, + &id, + &producer.created_by, + view_token.as_deref(), + ) + .await?; + } else if producer.created_by != "anonymous" { + return Err(Error::BadRequest( + "As a non logged in user, you can only see jobs ran by anonymous users".to_string(), + )); + } + + let rows = sqlx::query!( + r#"SELECT + subscriber_path AS "subscriber_path!", + asset_kind AS "asset_kind!: windmill_common::assets::AssetKind", + asset_path AS "asset_path!", + outcome::text AS "outcome!", + child_job_id, + partition, + received_inputs, + required_inputs, + debounce_s, + reason, + created_at AS "created_at!" + FROM dispatch_event + WHERE producer_job_id = $1 AND workspace_id = $2 + ORDER BY id"#, + id, + &w_id, + ) .fetch_all(&db) .await?; diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 226c060df9..6bb58373ea 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -8,10 +8,11 @@ //! Runtime fan-out for asset-triggered scripts. //! -//! When a producer pipeline script (`// pipeline`) writes an asset and a -//! downstream script subscribes to that asset via `// on s3://...`, this -//! module pushes a job for each subscriber after the producer's job -//! completes successfully. +//! When a script writes an asset and a downstream script subscribes to +//! that asset via `// on s3://...`, this module pushes a job for each +//! subscriber after the producer's job completes successfully. Any +//! asset-writing top-level script cascades — there is no `// pipeline` +//! gate on the producer side; subscriptions alone define the graph. //! //! Eligibility (V1, narrow on purpose): //! - Producer kind is `Script` or `Preview`. Flows defer. @@ -49,14 +50,15 @@ use sqlx::types::Json; use sqlx::{Pool, Postgres}; use std::collections::HashMap; use uuid::Uuid; -use windmill_common::assets::{parse_asset_trigger_ref, AssetKind, PARTITION_TOKEN}; +use windmill_common::assets::{AssetKind, PARTITION_TOKEN}; use windmill_common::error::{self, Result}; -use windmill_common::get_latest_hash_for_path; +use windmill_common::get_latest_deployed_hash_for_path; use windmill_common::jobs::{JobKind, JobPayload, JobTriggerKind}; use windmill_common::partition::PARTITION_ARG; -use windmill_common::runnable_settings::{ConcurrencySettings, DebouncingSettings}; +use windmill_common::runnable_settings::DebouncingSettings; +use windmill_common::scripts::ScriptHash; use windmill_common::triggers::TriggerMetadata; -use windmill_common::users::username_to_permissioned_as; +use windmill_common::users::{get_email_from_permissioned_as, username_to_permissioned_as}; use windmill_common::worker::to_raw_value; use windmill_common::DB; @@ -185,8 +187,15 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result _ => return Ok(DispatchResult::default()), }; - // Args were moved from v2_job to v2_job_completed by add_completed_job - // before dispatch runs. Fetch from v2_job_completed. + // Cheapest filter first: this hook runs on every top-level script + // completion, and the overwhelmingly common case is a script that + // writes no asset — one indexed query and out, before touching the + // job's args JSONB. + let writes = fetch_producer_writes(db, &job.workspace_id, runnable_path).await?; + if writes.is_empty() { + return Ok(DispatchResult::default()); + } + let args = fetch_args(db, &job.workspace_id, job.id).await?; if read_skip_arg(args.as_ref()) { return Ok(DispatchResult::default()); @@ -210,11 +219,6 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result return Ok(DispatchResult::default()); } - let writes = fetch_producer_writes(db, &job.workspace_id, runnable_path).await?; - if writes.is_empty() { - return Ok(DispatchResult::default()); - } - let mut dispatched = Vec::new(); for (asset_kind, asset_path) in writes { let Some(prefix) = prefix_for(asset_kind) else { @@ -507,14 +511,6 @@ async fn fetch_subscribers( ) .fetch_all(db) .await?; - // Sanity check: parsing the stored trigger_ref must succeed. If it - // doesn't, the row is corrupt; skip it loudly rather than silently. - if parse_asset_trigger_ref(trigger_ref).is_none() { - tracing::warn!( - "asset-trigger dispatch: trigger_ref {} did not round-trip through parse_asset_trigger_ref", - trigger_ref - ); - } Ok(rows .into_iter() .map(|r| Subscriber { @@ -662,31 +658,29 @@ async fn push_subscriber( retry_count: Option, retry_delay_s: Option, ) -> Result { - let ( - hash, - tag, - _concurrency_key, - _concurrent_limit, - _concurrency_time_window_s, - _debounce_key, - _debounce_delay_s, - cache_ttl, - cache_ignore_s3_path, - language, - dedicated_worker, - priority, - _timeout, - on_behalf_of_email, - created_by, - _runnable_settings_handle, - labels, - ) = get_latest_hash_for_path(db, &producer.workspace_id, subscriber_path, false).await?; + // Same resolution as every other trigger path (`script_path_to_payload`): + // latest deployed hash plus the script's own runnable settings + // (concurrency, debounce, timeout), resolved through the + // runnable-settings handle. The cascade must not bypass a subscriber's + // concurrency limit just because it was triggered by an asset write. + let script = get_latest_deployed_hash_for_path( + None, + db.clone(), + &producer.workspace_id, + subscriber_path, + ) + .await? + .prefetch_cached(db) + .await?; + let hash = ScriptHash(script.hash); + let tag = script.tag; + let concurrency_settings = script.runnable_settings.concurrency_settings; // Debounce is opt-in per subscriber edge (`// debounce` / - // `// on … debounce=`). Default = none (fan-out — the user's intent - // unless they ask otherwise). When set, the window is keyed by + // `// on … debounce=`). When set, the window is keyed by // (subscriber, partition) so distinct partitions never collapse and - // "latest within the window" falls out for free. + // "latest within the window" falls out for free. When not set, the + // subscriber's own script-level debounce settings (if any) apply. let debouncing_settings = match debounce_s { Some(s) if s > 0 => DebouncingSettings { debounce_key: Some(format!( @@ -697,7 +691,7 @@ async fn push_subscriber( debounce_delay_s: Some(s), ..DebouncingSettings::default() }, - _ => DebouncingSettings::default(), + _ => script.runnable_settings.debouncing_settings, }; // Retry is only available via the flow runtime — wrap the script in a @@ -725,42 +719,45 @@ async fn push_subscriber( error_handler_path: None, error_handler_args: None, skip_handler: None, - cache_ttl, - cache_ignore_s3_path, - priority, + cache_ttl: script.cache_ttl, + cache_ignore_s3_path: script.cache_ignore_s3_path, + priority: script.priority, tag_override: tag.clone(), trigger_path: None, apply_preprocessor: false, - concurrency_settings: ConcurrencySettings::default(), + concurrency_settings, debouncing_settings, } } else { JobPayload::ScriptHash { hash, path: subscriber_path.to_string(), - cache_ttl, - cache_ignore_s3_path, - dedicated_worker, - language, - priority, + cache_ttl: script.cache_ttl, + cache_ignore_s3_path: script.cache_ignore_s3_path, + dedicated_worker: script.dedicated_worker, + language: script.language, + priority: script.priority, apply_preprocessor: false, debouncing_settings, - concurrency_settings: ConcurrencySettings::default(), - labels, + concurrency_settings, + labels: script.labels, } }; - // Subscriber's own on_behalf_of_email controls identity when set; - // otherwise we run as the producer. This keeps the asset cascade - // attributable to whoever originally wrote the asset, while still - // honoring scripts that explicitly opted into a service-account email. - let (permissioned_as, email) = if let Some(obo) = on_behalf_of_email { - (username_to_permissioned_as(&created_by), obo) - } else { - ( - producer.permissioned_as.clone(), - producer.permissioned_as_email.clone(), - ) + // Run the subscriber under its deployer's identity — never the + // producer's. Subscriptions are workspace-wide, so attributing the run + // to the producer would let anyone who can deploy a `// on` script + // execute code with the permissions of whoever happens to write the + // asset (e.g. an admin's scheduled job). `on_behalf_of_email` (an + // explicit service-account opt-in at deploy) takes precedence for the + // email; otherwise the deployer's email is resolved from their + // username. + let permissioned_as = username_to_permissioned_as(&script.created_by); + let email = match script.on_behalf_of_email { + Some(obo) => obo, + None => { + get_email_from_permissioned_as(&permissioned_as, &producer.workspace_id, db).await? + } }; let mut args: HashMap> = HashMap::new(); @@ -808,7 +805,7 @@ async fn push_subscriber( None, true, tag, - None, + script.timeout, None, None, None, diff --git a/frontend/src/lib/components/ScriptEditor.svelte b/frontend/src/lib/components/ScriptEditor.svelte index 7f95460151..324f6829ff 100644 --- a/frontend/src/lib/components/ScriptEditor.svelte +++ b/frontend/src/lib/components/ScriptEditor.svelte @@ -1754,33 +1754,14 @@ class="flex flex-row divide-x divide-gray-800 dark:divide-gray-300 items-stretch" > {#if testIsLoading} - + {@render cancelTestButton('md', 'w-full')} {:else} {@const disableTriggerButton = customUi?.previewPanel?.disableTriggerButton === true} - + {@render runTestButton( + 'md', + `w-full ${!disableTriggerButton ? 'rounded-r-none' : ''}` + )} {#if !disableTriggerButton} {/if} @@ -1853,16 +1834,7 @@ {/if}
{#if testIsLoading} - + {@render cancelTestButton('sm', 'shadow-md')} {:else if (customUi?.previewPanel?.downstreamSubscribers ?? 0) > 0} - - {/if} - {#snippet capturesTab()} -
- -
- {/snippet} - {#snippet customResultPanel()} - - {/snippet} - + {@render testLogPanel()} {/key} @@ -2220,6 +2068,95 @@
+{#snippet cancelTestButton(size: 'sm' | 'md', btnClasses: string)} + +{/snippet} + +{#snippet runTestButton(size: 'sm' | 'md', btnClasses: string)} + +{/snippet} + + +{#snippet testLogPanel()} + { + historyTabActive = tab === 'history' + if (historyTabActive) { + loadPastTests() + } + }} + previewIsLoading={debugMode ? $debugState.running && !$debugState.stopped : testIsLoading} + {editor} + {diffEditor} + args={activeModuleTab !== null ? testPanelArgs : args} + {showCaptures} + customUi={customUi?.previewPanel} + showCustomResultPanel={showDebugPanel} + > + {#if scriptProgress && !debugMode} + + + {/if} + {#snippet capturesTab()} +
+ +
+ {/snippet} + {#snippet customResultPanel()} + + {/snippet} + +{/snippet} + {#snippet addModuleForm(close: () => void)}
diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelinePickerModal.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelinePickerModal.svelte index bf85c8b455..20a5c42155 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelinePickerModal.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelinePickerModal.svelte @@ -5,15 +5,10 @@ import Button from '$lib/components/common/button/Button.svelte' import FolderPicker from '$lib/components/FolderPicker.svelte' import Modal from '$lib/components/common/modal/Modal.svelte' - import { OpenAPI } from '$lib/gen' + import { AssetService, type ListPipelineFoldersResponse } from '$lib/gen' import { resource } from 'runed' import { ArrowRight, Loader2 } from 'lucide-svelte' - interface PipelineFolder { - folder: string - script_count: number - } - interface Props { open: boolean // When provided, the current folder is dropped from the existing list @@ -24,15 +19,9 @@ let pipelines = resource( () => $workspaceStore, - async (ws, _prev, { signal }) => { - if (!ws) return [] as PipelineFolder[] - const base_url = OpenAPI.BASE ?? '' - const res = await fetch(`${base_url}/w/${ws}/assets/pipelines`, { - credentials: 'include', - signal - }) - if (!res.ok) throw new Error(`GET /assets/pipelines → ${res.status}`) - return (await res.json()) as PipelineFolder[] + async (ws) => { + if (!ws) return [] as ListPipelineFoldersResponse + return await AssetService.listPipelineFolders({ workspace: ws }) } ) diff --git a/frontend/src/lib/components/home/ItemsList.svelte b/frontend/src/lib/components/home/ItemsList.svelte index 1adf010bb9..6391d63933 100644 --- a/frontend/src/lib/components/home/ItemsList.svelte +++ b/frontend/src/lib/components/home/ItemsList.svelte @@ -4,13 +4,13 @@ import Toggle from '$lib/components/Toggle.svelte' import { AppService, + AssetService, FlowService, type ListableApp, type Script, ScriptService, type Flow, - type ListableRawApp, - OpenAPI + type ListableRawApp } from '$lib/gen' import { resource } from 'runed' import { userStore, workspaceStore } from '$lib/stores' @@ -79,16 +79,15 @@ // thanks to the partial index on script.auto_kind. let pipelineFoldersRes = resource( () => $workspaceStore, - async (ws, _prev, { signal }) => { + async (ws) => { if (!ws) return new Set() - const base_url = OpenAPI.BASE ?? '' - const res = await fetch(`${base_url}/w/${ws}/assets/pipelines`, { - credentials: 'include', - signal - }) - if (!res.ok) return new Set() - const rows = (await res.json()) as Array<{ folder: string }> - return new Set(rows.map((r) => r.folder)) + try { + const rows = await AssetService.listPipelineFolders({ workspace: ws }) + return new Set(rows.map((r) => r.folder)) + } catch { + // Decorative tree entry — degrade to "no pipelines" on failure. + return new Set() + } } ) let pipelineFolders = $derived(pipelineFoldersRes.current ?? new Set())