diff --git a/backend/.sqlx/query-6aaddd80f8c07cfafea2021c1879c3d7b2fb156a43299e9a1209d05293c4f50f.json b/backend/.sqlx/query-6aaddd80f8c07cfafea2021c1879c3d7b2fb156a43299e9a1209d05293c4f50f.json new file mode 100644 index 0000000000..144bcb13f2 --- /dev/null +++ b/backend/.sqlx/query-6aaddd80f8c07cfafea2021c1879c3d7b2fb156a43299e9a1209d05293c4f50f.json @@ -0,0 +1,64 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dispatch_event (\n workspace_id, producer_job_id, subscriber_path,\n asset_kind, asset_path, outcome,\n child_job_id, partition,\n received_inputs, required_inputs,\n debounce_s, reason\n )\n SELECT $1, $2, sp, ak, ap, oc, cj, pt, ri, rq, db, rs\n FROM unnest(\n $3::text[], $4::ASSET_KIND[], $5::text[], $6::DISPATCH_OUTCOME[],\n $7::uuid[], $8::text[], $9::int[], $10::int[], $11::int[], $12::text[]\n ) AS t(sp, ak, ap, oc, cj, pt, ri, rq, db, rs)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Uuid", + "TextArray", + { + "Custom": { + "name": "asset_kind[]", + "kind": { + "Array": { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + } + } + } + }, + "TextArray", + { + "Custom": { + "name": "dispatch_outcome[]", + "kind": { + "Array": { + "Custom": { + "name": "dispatch_outcome", + "kind": { + "Enum": [ + "dispatched", + "join_pending", + "skipped" + ] + } + } + } + } + } + }, + "UuidArray", + "TextArray", + "Int4Array", + "Int4Array", + "Int4Array", + "TextArray" + ] + }, + "nullable": [] + }, + "hash": "6aaddd80f8c07cfafea2021c1879c3d7b2fb156a43299e9a1209d05293c4f50f" +} diff --git a/backend/.sqlx/query-d3d87b9a4d62977dea5af95dd457bb759229805cff771cfa4275a46bfc80e1ab.json b/backend/.sqlx/query-d3d87b9a4d62977dea5af95dd457bb759229805cff771cfa4275a46bfc80e1ab.json new file mode 100644 index 0000000000..b5d905b9b6 --- /dev/null +++ b/backend/.sqlx/query-d3d87b9a4d62977dea5af95dd457bb759229805cff771cfa4275a46bfc80e1ab.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job\n SET args = CASE\n WHEN args ? 'partition'\n THEN $1 || jsonb_build_object('partition', args -> 'partition')\n ELSE $1\n END,\n preprocessed = TRUE\n WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Jsonb", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "d3d87b9a4d62977dea5af95dd457bb759229805cff771cfa4275a46bfc80e1ab" +} diff --git a/backend/.sqlx/query-f7790aaa332e81046ececc71b728364a3489a3c9151ac153c419ddce51d0d891.json b/backend/.sqlx/query-f7790aaa332e81046ececc71b728364a3489a3c9151ac153c419ddce51d0d891.json deleted file mode 100644 index c28d5f488a..0000000000 --- a/backend/.sqlx/query-f7790aaa332e81046ececc71b728364a3489a3c9151ac153c419ddce51d0d891.json +++ /dev/null @@ -1,50 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO dispatch_event (\n workspace_id, producer_job_id, subscriber_path,\n asset_kind, asset_path, outcome,\n child_job_id, partition,\n received_inputs, required_inputs,\n debounce_s, reason\n ) VALUES ($1, $2, $3, $4::ASSET_KIND, $5, $6::DISPATCH_OUTCOME,\n $7, $8, $9, $10, $11, $12)", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Uuid", - "Varchar", - { - "Custom": { - "name": "asset_kind", - "kind": { - "Enum": [ - "s3object", - "resource", - "variable", - "ducklake", - "datatable", - "volume" - ] - } - } - }, - "Text", - { - "Custom": { - "name": "dispatch_outcome", - "kind": { - "Enum": [ - "dispatched", - "join_pending", - "skipped" - ] - } - } - }, - "Uuid", - "Text", - "Int4", - "Int4", - "Int4", - "Text" - ] - }, - "nullable": [] - }, - "hash": "f7790aaa332e81046ececc71b728364a3489a3c9151ac153c419ddce51d0d891" -} diff --git a/backend/colors-after.png b/backend/colors-after.png deleted file mode 100644 index 0264d011ea..0000000000 Binary files a/backend/colors-after.png and /dev/null differ diff --git a/backend/colors-v2-edit.png b/backend/colors-v2-edit.png deleted file mode 100644 index fbce0310fa..0000000000 Binary files a/backend/colors-v2-edit.png and /dev/null differ diff --git a/backend/colors-v2.png b/backend/colors-v2.png deleted file mode 100644 index d6c56764ab..0000000000 Binary files a/backend/colors-v2.png and /dev/null differ diff --git a/backend/migrations/20260423050000_script_trigger.up.sql b/backend/migrations/20260423050000_script_trigger.up.sql index 9abc42b1f1..cb105ba3f3 100644 --- a/backend/migrations/20260423050000_script_trigger.up.sql +++ b/backend/migrations/20260423050000_script_trigger.up.sql @@ -6,9 +6,18 @@ -- `// on ` lines (every non-integration trigger kind; their -- trigger_ref is the trigger row path, or empty for marker-only forms). -- +-- Per-edge columns that are in fact script-level properties (every row for +-- a given runnable carries the same value, set once at deploy) but live on +-- the edge so the dispatcher reads everything from a single query: +-- join_all `// trigger all` AND-join barrier (else OR, the default). +-- debounce_s `// on debounce=` (else script-level `// debounce`); only +-- asset-cascade edges carry one. NULL = no debounce. +-- retry_* `// retry []` cascade retry. NULL = none. +-- -- 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. +-- from the unsquashed history already contain the final objects and +-- re-applying must be a no-op. DO $$ BEGIN CREATE TYPE SCRIPT_TRIGGER_KIND AS ENUM ( 'asset', 'schedule', 'webhook', 'email', 'kafka', 'mqtt', 'nats', @@ -22,7 +31,11 @@ CREATE TABLE IF NOT EXISTS script_trigger ( 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 + trigger_ref TEXT NOT NULL, + join_all BOOLEAN NOT NULL DEFAULT FALSE, + debounce_s INTEGER, + retry_count SMALLINT, + retry_delay_s INTEGER ); -- Per-runnable lookup (wipe-on-deploy, list-triggers-for-script). diff --git a/backend/migrations/20260516190545_add_script_trigger_join_all.down.sql b/backend/migrations/20260516190545_add_script_trigger_join_all.down.sql deleted file mode 100644 index b69f7abca6..0000000000 --- a/backend/migrations/20260516190545_add_script_trigger_join_all.down.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE script_trigger - DROP COLUMN join_all; diff --git a/backend/migrations/20260516190545_add_script_trigger_join_all.up.sql b/backend/migrations/20260516190545_add_script_trigger_join_all.up.sql deleted file mode 100644 index b3aa5b31df..0000000000 --- a/backend/migrations/20260516190545_add_script_trigger_join_all.up.sql +++ /dev/null @@ -1,9 +0,0 @@ --- AND-join barrier (`// trigger all`). When true, the subscriber runs only --- once every partition-bearing input has materialized at the same partition --- (plus every reference input exists), rather than firing on any input (OR, --- the default). Stored per script_trigger row — it is a script-level --- property so every row for a given (workspace, runnable) carries the same --- value; this matches the wipe-and-reinsert-on-deploy pattern and keeps the --- subscriber lookup a single query. -ALTER TABLE script_trigger - ADD COLUMN join_all BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/backend/migrations/20260516201317_script_trigger_debounce_s.down.sql b/backend/migrations/20260516201317_script_trigger_debounce_s.down.sql deleted file mode 100644 index fb0950f26f..0000000000 --- a/backend/migrations/20260516201317_script_trigger_debounce_s.down.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE script_trigger - DROP COLUMN debounce_s; diff --git a/backend/migrations/20260516201317_script_trigger_debounce_s.up.sql b/backend/migrations/20260516201317_script_trigger_debounce_s.up.sql deleted file mode 100644 index 7ef44f22c6..0000000000 --- a/backend/migrations/20260516201317_script_trigger_debounce_s.up.sql +++ /dev/null @@ -1,8 +0,0 @@ --- Opt-in debounce window (seconds) for an asset-cascade subscriber edge. --- NULL = no debounce (fan-out, the default/current behaviour). Resolved --- per `// on` edge at deploy as: edge `debounce=` ?? script-level --- `// debounce ` ?? none. Only asset subscriber rows carry it; the --- dispatcher builds DebouncingSettings keyed (subscriber, partition) so --- distinct partitions never collapse and "latest in window" falls out. -ALTER TABLE script_trigger - ADD COLUMN debounce_s INTEGER; diff --git a/backend/migrations/20260523054912_script_trigger_retry.down.sql b/backend/migrations/20260523054912_script_trigger_retry.down.sql deleted file mode 100644 index 49ec6bd823..0000000000 --- a/backend/migrations/20260523054912_script_trigger_retry.down.sql +++ /dev/null @@ -1,3 +0,0 @@ -ALTER TABLE script_trigger - DROP COLUMN retry_count, - DROP COLUMN retry_delay_s; diff --git a/backend/migrations/20260523054912_script_trigger_retry.up.sql b/backend/migrations/20260523054912_script_trigger_retry.up.sql deleted file mode 100644 index 8eb9adca96..0000000000 --- a/backend/migrations/20260523054912_script_trigger_retry.up.sql +++ /dev/null @@ -1,10 +0,0 @@ --- Cascade retry policy (`// retry []`). Stored per --- `script_trigger` row — like `join_all`, it is a script-level property --- (every row for a given runnable carries the same value) but lives --- alongside the edge so the dispatcher reads everything it needs from a --- single query. NULL = no retry (current behaviour); set = re-run on --- failure up to `retry_count` times, waiting `retry_delay_s` seconds --- between attempts (0 = back-to-back). -ALTER TABLE script_trigger - ADD COLUMN retry_count SMALLINT, - ADD COLUMN retry_delay_s INTEGER; diff --git a/backend/migrations/20260523055641_dispatch_event.up.sql b/backend/migrations/20260523055641_dispatch_event.up.sql index f36c1dab45..51403f33ce 100644 --- a/backend/migrations/20260523055641_dispatch_event.up.sql +++ b/backend/migrations/20260523055641_dispatch_event.up.sql @@ -9,13 +9,19 @@ -- retention sweep (monitor.rs delete_expired_jobs_batch -> DELETE FROM -- v2_job WHERE id = ANY(...)) reaps these rows along with their producer. -- No separate cleanup path needed. -CREATE TYPE DISPATCH_OUTCOME AS ENUM ( - 'dispatched', - 'join_pending', - 'skipped' -); +-- +-- Idempotency guards (duplicate_object / IF NOT EXISTS) are load-bearing: +-- re-applying this migration after a squash must be a no-op. +DO $$ BEGIN + CREATE TYPE DISPATCH_OUTCOME AS ENUM ( + 'dispatched', + 'join_pending', + 'skipped' + ); +EXCEPTION WHEN duplicate_object THEN NULL; +END $$; -CREATE TABLE dispatch_event ( +CREATE TABLE IF NOT EXISTS dispatch_event ( id BIGSERIAL PRIMARY KEY, workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, producer_job_id UUID NOT NULL REFERENCES v2_job(id) ON DELETE CASCADE, @@ -41,5 +47,17 @@ CREATE TABLE dispatch_event ( -- Primary access pattern: list events for one producer (the job detail -- panel). Ordered scans by id give chronological order for free. -CREATE INDEX idx_dispatch_event_producer +CREATE INDEX IF NOT EXISTS idx_dispatch_event_producer ON dispatch_event (producer_job_id, id); + +-- Backs the asset-graph edge listing (jobs.rs list_asset_dispatch_edges): +-- WHERE workspace_id = $1 AND subscriber_path LIKE 'prefix%' +-- AND created_at >= $3 +-- ORDER BY created_at DESC, id DESC +-- The (producer_job_id, id) index above doesn't help this access path, so +-- without this one a high-volume dispatch_event seq-scans + sorts. +-- text_pattern_ops makes the anchored LIKE prefix (built as `path_start || '%'`) +-- index-usable regardless of the column collation; created_at DESC matches +-- the ORDER BY so Postgres can satisfy ordering from the index. +CREATE INDEX IF NOT EXISTS idx_dispatch_event_subscriber + ON dispatch_event (workspace_id, subscriber_path text_pattern_ops, created_at DESC); diff --git a/backend/migrations/20260523091727_backfill_asset_from_script_assets.down.sql b/backend/migrations/20260523091727_backfill_asset_from_script_assets.down.sql deleted file mode 100644 index 5a60bddcdf..0000000000 --- a/backend/migrations/20260523091727_backfill_asset_from_script_assets.down.sql +++ /dev/null @@ -1,6 +0,0 @@ --- No-op: the backfill is a one-shot data migration. We can't safely --- distinguish backfilled rows from those a subsequent deploy inserted via --- the normal `insert_static_asset_usage` path (same tuple, no marker), so --- rolling back would risk deleting live data. Leaving the rows in place on --- downgrade is harmless — they'll get rewritten on the next deploy of --- each affected script. diff --git a/backend/migrations/20260523091727_backfill_asset_from_script_assets.up.sql b/backend/migrations/20260523091727_backfill_asset_from_script_assets.up.sql deleted file mode 100644 index ecc77fcf0b..0000000000 --- a/backend/migrations/20260523091727_backfill_asset_from_script_assets.up.sql +++ /dev/null @@ -1,39 +0,0 @@ --- Backfill the `asset` table from `script.assets` JSONB for scripts that --- were deployed before the deploy path started populating asset rows from --- the parsed `ns.assets`. The JSONB column carried the parser's findings --- the whole time; the corresponding asset rows are what `fetch_producer_writes` --- (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. --- --- Idempotent: scoped to (workspace, path) pairs that have ZERO asset rows --- under `usage_kind = 'script'`, so re-running can't duplicate. The --- (workspace_id, path, kind, usage_path, usage_kind) primary key catches --- any residual overlap via `ON CONFLICT DO NOTHING` as a belt-and-braces. --- Skips archived/deleted script versions so we backfill from the *current* --- snapshot, matching what the live deploy path would write. -INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind) -SELECT - s.workspace_id, - a->>'path', - (a->>'kind')::asset_kind, - (a->>'access_type')::asset_access_type, - s.path, - 'script'::asset_usage_kind -FROM script s -CROSS JOIN LATERAL jsonb_array_elements(s.assets) AS a -WHERE s.archived = false - AND s.deleted = false - AND s.assets IS NOT NULL - AND jsonb_typeof(s.assets) = 'array' - AND a->>'path' IS NOT NULL - AND a->>'kind' IS NOT NULL - AND NOT EXISTS ( - SELECT 1 - FROM asset existing - WHERE existing.workspace_id = s.workspace_id - AND existing.usage_kind = 'script' - AND existing.usage_path = s.path - ) -ON CONFLICT DO NOTHING; diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 1ac1638959..31b9815698 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -281,13 +281,12 @@ pub fn merge_assets(assets: Vec) -> Vec { .iter_mut() .find(|x| x.path == asset.path && x.kind == asset.kind) { - // merge access types + // merge access types — a None on either side means ambiguous + // usage (unknown access), which poisons the merge to None; + // otherwise delegate to the shared truth table. existing.access_type = match (asset.access_type, existing.access_type) { (None, _) | (_, None) => None, - (Some(R), Some(W)) | (Some(W), Some(R)) => Some(RW), - (Some(RW), _) | (_, Some(RW)) => Some(RW), - (Some(R), Some(R)) => Some(R), - (Some(W), Some(W)) => Some(W), + (Some(a), Some(b)) => Some(merge_access_types(a, b)), }; // merge columns: union the column sets and merge access types per column existing.columns = merge_column_maps(existing.columns.take(), asset.columns); @@ -391,10 +390,11 @@ fn unquote(s: &str) -> Option<&str> { // `ident=` token — the same assumption `// partitioned` already makes. fn split_trailing_kv_opts(s: &str) -> (&str, BTreeMap) { let mut split_at: Option = None; - let mut pos = 0usize; for tok in s.split_whitespace() { - let tok_start = s[pos..].find(tok).map(|o| pos + o).unwrap_or(pos); - pos = tok_start + tok.len(); + // `split_whitespace` yields slices borrowed from `s`, so the exact + // byte offset is the pointer delta — substring search (`find`) would + // misfire when an earlier token also appears inside a later one. + let tok_start = tok.as_ptr() as usize - s.as_ptr() as usize; if let Some(eq) = tok.find('=') { let key = &tok[..eq]; if !key.is_empty() @@ -489,6 +489,20 @@ fn parse_kv_opts(s: &str) -> BTreeMap { // multiple lines declare them, the first one is kept (last would be // reasonable too, but first matches the file-top convention developers // follow). +// Try to consume `` as a complete word from `rest`. Returns the trailing +// text after the keyword if it matched (empty or whitespace-bounded), +// `None` otherwise. Prevents `partitioned` matching `partition`, `pipelines` +// matching `pipeline`, etc. Mirrors `consumeKeyword` in +// parsePipelineAnnotations.ts. +fn consume_keyword<'a>(rest: &'a str, kw: &str) -> Option<&'a str> { + let after = rest.strip_prefix(kw)?; + if after.is_empty() || after.starts_with(|c: char| c.is_whitespace()) { + Some(after) + } else { + None + } +} + pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { let mut out = PipelineAnnotations::default(); @@ -505,7 +519,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { }; let rest = rest.trim_start(); - if let Some(after_kw) = rest.strip_prefix("pipeline") { + if let Some(after_kw) = consume_keyword(rest, "pipeline") { // Strict: keyword must be the only content on the line. Rejects // `pipeline broken`, `pipelines`, `pipeline-related`, etc. if after_kw.trim().is_empty() { @@ -514,10 +528,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } - if let Some(after_kw) = rest.strip_prefix("partitioned") { - if !after_kw.starts_with(|c: char| c.is_whitespace()) { - continue; - } + if let Some(after_kw) = consume_keyword(rest, "partitioned") { if out.partition.is_none() { if let Some(spec) = parse_partitioned_spec(after_kw.trim()) { out.partition = Some(spec); @@ -526,10 +537,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } - if let Some(after_kw) = rest.strip_prefix("freshness") { - if !after_kw.starts_with(|c: char| c.is_whitespace()) { - continue; - } + if let Some(after_kw) = consume_keyword(rest, "freshness") { let dur = after_kw.trim(); if !dur.is_empty() && out.freshness.is_none() { out.freshness = Some(FreshnessSpec { duration: dur.to_string() }); @@ -537,10 +545,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } - if let Some(after_kw) = rest.strip_prefix("trigger") { - if !after_kw.starts_with(|c: char| c.is_whitespace()) { - continue; - } + if let Some(after_kw) = consume_keyword(rest, "trigger") { match after_kw.trim() { "all" => out.join_mode = JoinMode::All, "any" => out.join_mode = JoinMode::Any, @@ -550,10 +555,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } - if let Some(after_kw) = rest.strip_prefix("debounce") { - if !after_kw.starts_with(|c: char| c.is_whitespace()) { - continue; - } + if let Some(after_kw) = consume_keyword(rest, "debounce") { let dur = after_kw.trim(); if !dur.is_empty() && out.debounce_default.is_none() { out.debounce_default = Some(dur.to_string()); @@ -561,10 +563,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } - if let Some(after_kw) = rest.strip_prefix("tag") { - if !after_kw.starts_with(|c: char| c.is_whitespace()) { - continue; - } + if let Some(after_kw) = consume_keyword(rest, "tag") { let name = after_kw.trim(); if !name.is_empty() && out.tag.is_none() { out.tag = Some(name.to_string()); @@ -572,10 +571,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } - if let Some(after_kw) = rest.strip_prefix("retry") { - if !after_kw.starts_with(|c: char| c.is_whitespace()) { - continue; - } + if let Some(after_kw) = consume_keyword(rest, "retry") { if out.retry.is_none() { if let Some(spec) = parse_retry_spec(after_kw.trim()) { out.retry = Some(spec); @@ -584,10 +580,7 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } - if let Some(after_kw) = rest.strip_prefix("on") { - if !after_kw.starts_with(|c: char| c.is_whitespace()) { - continue; - } + if let Some(after_kw) = consume_keyword(rest, "on") { let spec_text = after_kw.trim(); if spec_text.is_empty() { continue; @@ -1082,6 +1075,33 @@ mod pipeline_annotation_tests { assert_eq!(r.delay.as_deref(), Some("5s")); } + #[test] + fn split_trailing_kv_opts_uses_exact_token_offset() { + // Regression for the token-offset computation in + // `split_trailing_kv_opts`. The asset ref's path token contains the + // exact text of the trailing `debounce=60s` opt as a substring, and + // the same `debounce=60s` text also appears a second time before the + // real opt. Offsetting by the `&str` slice's pointer is exact; a + // substring scan (`find`) is what this guards against regressing to. + let (ref_part, opts) = split_trailing_kv_opts("s3://lake/debounce=60s/raw debounce=60s"); + assert_eq!(ref_part, "s3://lake/debounce=60s/raw"); + assert_eq!(opts.get("debounce").map(String::as_str), Some("60s")); + + // End-to-end through the annotation parser: the ref must parse to the + // full S3 path (not truncated at the embedded `debounce=`), and the + // per-edge debounce override must be picked up from the trailing opt. + let out = parse_pipeline_annotations("// on s3://lake/debounce=60s/raw debounce=90s"); + assert_eq!(out.triggers.len(), 1); + match &out.triggers[0] { + TriggerSpec::Asset { asset_kind, path, debounce } => { + assert_eq!(*asset_kind, AssetKind::S3Object); + assert_eq!(path, "lake/debounce=60s/raw"); + assert_eq!(debounce.as_deref(), Some("90s")); + } + other => panic!("expected asset trigger, got {other:?}"), + } + } + #[test] fn kv_opts_quoted_with_spaces() { let m = parse_kv_opts("a=\"hello world\" b=plain c='single quoted'"); diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index c8ed841e19..7b8d468da8 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -282,6 +282,12 @@ pub async fn migrate( 20260207000002, 20260207000003, 20260207000004, + // Squashed pre-release pipeline migrations: the per-column ALTERs on + // script_trigger and the dispatch_event subscriber index were folded + // back into these two CREATEs, changing their checksum. Both are + // idempotent, so re-applying on an already-migrated DB is a no-op. + 20260423050000, + 20260523055641, ]; for m in migrator.migrations.iter() { if m.migration_type.is_down_migration() { diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 7fac7d2c68..ebdb5bfb54 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -211,13 +211,10 @@ pub fn parse_asset_trigger_ref(s: &str) -> Option<(AssetKind, String)> { pub fn trigger_spec_to_row(spec: &TriggerSpec) -> Option<(ScriptTriggerKind, String)> { match spec { TriggerSpec::Asset { asset_kind, path, .. } => { - let prefix = match asset_kind { - windmill_parser::asset_parser::AssetKind::S3Object => "s3://", - windmill_parser::asset_parser::AssetKind::Resource => "$res:", - windmill_parser::asset_parser::AssetKind::Ducklake => "ducklake://", - windmill_parser::asset_parser::AssetKind::DataTable => "datatable://", - windmill_parser::asset_parser::AssetKind::Volume => "volume://", - }; + // Single source of truth for the canonical prefix lives on the + // common AssetKind; map the parser kind across first. The parser + // enum has no Variable variant, so canonical_prefix is always Some. + let prefix = asset_kind_from_parser(*asset_kind).canonical_prefix()?; Some((ScriptTriggerKind::Asset, format!("{}{}", prefix, path))) } // Schedule joins the native-trigger family — no script_trigger row diff --git a/backend/windmill-common/src/partition.rs b/backend/windmill-common/src/partition.rs index 14bfbdc173..2028438909 100644 --- a/backend/windmill-common/src/partition.rs +++ b/backend/windmill-common/src/partition.rs @@ -14,8 +14,14 @@ //! - `dynamic key=`: extracted from the triggering payload via a //! minimal `$.a.b` JSON path (the realistic per-tenant/shard/event case). +use std::collections::HashMap; + use chrono::{DateTime, NaiveDate, Utc}; use chrono_tz::Tz; +use serde_json::value::RawValue; +use sqlx::types::Json; +use sqlx::PgExecutor; +use uuid::Uuid; use windmill_parser::asset_parser::{PartitionKind, PartitionSpec}; use crate::error::{Error, Result}; @@ -24,6 +30,60 @@ use crate::error::{Error, Result}; /// mirrored at `trigger.partition` for cascade propagation). pub const PARTITION_ARG: &str = "partition"; +/// Persist a freshly resolved `partition` into a job's `v2_job.args`, +/// merging it on top of whatever args already exist. +/// +/// INVARIANT: partition, once resolved, is immutable for the job's +/// lifetime. This is the single place that *establishes* it (run-start +/// resolution); every later args rewrite must preserve it — see +/// [`merge_args_preserving_partition`]. Keeping both the set and the +/// preserve here keeps the JSONB shape and the invariant in one module. +pub async fn set_resolved_partition<'e>( + executor: impl PgExecutor<'e>, + job_id: Uuid, + value: &str, +) -> Result<()> { + sqlx::query!( + "UPDATE v2_job SET args = coalesce(args, '{}'::jsonb) \ + || jsonb_build_object('partition', $1::text) WHERE id = $2", + value, + job_id, + ) + .execute(executor) + .await?; + Ok(()) +} + +/// Replace a job's `v2_job.args` with `new_args`, but carry forward any +/// existing top-level `partition` key so a wholesale rewrite (e.g. the +/// preprocessor swapping in preprocessed args) cannot clobber a resolved +/// partition. +/// +/// INVARIANT: partition, once resolved, is immutable for the job's +/// lifetime (set by [`set_resolved_partition`]); this guards every later +/// args rewrite against dropping it. +pub async fn merge_args_preserving_partition<'e>( + executor: impl PgExecutor<'e>, + job_id: Uuid, + new_args: HashMap>, +) -> Result<()> { + sqlx::query!( + "UPDATE v2_job + SET args = CASE + WHEN args ? 'partition' + THEN $1 || jsonb_build_object('partition', args -> 'partition') + ELSE $1 + END, + preprocessed = TRUE + WHERE id = $2", + Json(new_args) as Json>>, + job_id, + ) + .execute(executor) + .await?; + Ok(()) +} + fn default_format(kind: &PartitionKind) -> &'static str { match kind { PartitionKind::Hourly => "%Y-%m-%dT%H", @@ -40,7 +100,7 @@ fn default_format(kind: &PartitionKind) -> &'static str { /// `spec.start` (anchor: older partitions are not backfilled). Errors on an /// invalid `tz` or `start` rather than silently materializing the wrong /// partition. `Dynamic` is rejected — use [`extract_dynamic_partition`]. -pub fn resolve_time_partition(spec: &PartitionSpec, at: DateTime) -> Result> { +fn resolve_time_partition(spec: &PartitionSpec, at: DateTime) -> Result> { if matches!(spec.kind, PartitionKind::Dynamic { .. }) { return Err(Error::BadRequest( "resolve_time_partition called on a dynamic partition".to_string(), @@ -74,7 +134,7 @@ pub fn resolve_time_partition(spec: &PartitionSpec, at: DateTime) -> Result /// `$.a.b.c` dotted path (no brackets/wildcards/filters — that covers the /// realistic per-tenant/shard/event keys). The leaf must be a string, /// number or bool. -pub fn extract_dynamic_partition(key: &str, payload: &serde_json::Value) -> Result { +fn extract_dynamic_partition(key: &str, payload: &serde_json::Value) -> Result { let path = key.strip_prefix("$").unwrap_or(key); let path = path.strip_prefix('.').unwrap_or(path); let mut cur = payload; diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 6bb58373ea..49e278fdda 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -94,8 +94,8 @@ enum DispatchOutcome { Skipped, } -/// Outcome-specific fields. `record_event` takes the four "always-present" -/// columns positionally and bundles the rest here so each call site only +/// Outcome-specific fields. Event constructors take the four "always-present" +/// columns positionally and bundle the rest here so each call site only /// names what it actually carries. #[derive(Debug, Default)] struct EventOptions<'a> { @@ -107,6 +107,46 @@ struct EventOptions<'a> { reason: Option<&'a str>, } +/// One accumulated `dispatch_event` row. Owned (not borrowed) so the whole +/// dispatch pass can collect rows and flush them in a single batched INSERT +/// at the end, avoiding an N+1 (one INSERT per subscriber × asset write). +#[derive(Debug)] +struct EventRow { + subscriber_path: String, + asset_kind: AssetKind, + asset_path: String, + outcome: DispatchOutcome, + child_job_id: Option, + partition: Option, + received_inputs: Option, + required_inputs: Option, + debounce_s: Option, + reason: Option, +} + +impl EventRow { + fn new( + subscriber_path: &str, + asset_kind: AssetKind, + asset_path: &str, + outcome: DispatchOutcome, + opts: EventOptions<'_>, + ) -> Self { + EventRow { + subscriber_path: subscriber_path.to_string(), + asset_kind, + asset_path: asset_path.to_string(), + outcome, + child_job_id: opts.child_job_id, + partition: opts.partition.map(str::to_string), + received_inputs: opts.received_inputs, + required_inputs: opts.required_inputs, + debounce_s: opts.debounce_s, + reason: opts.reason.map(str::to_string), + } + } +} + /// AND-join slot progress at the moment a partition-bearing input /// arrived. `fired` = all required inputs are now present (the slot has /// been cleared and the subscriber will be dispatched). `received` / @@ -120,19 +160,75 @@ struct JoinSlotStatus { required: i32, } -/// Best-effort insert into `dispatch_event`. Never propagates — the -/// dispatch contract is "logging failures must not retroactively fail -/// the producer's job." -async fn record_event( +/// Outcome of evaluating an AND-join barrier for one (subscriber, input). +/// The caller turns this into a `dispatch_event` row and decides whether to +/// push, all in one place. +enum JoinDecision { + /// Input does not advance the join (recorded as Skipped with `reason`). + Skip(&'static str), + /// Join advanced but is not yet complete (recorded as JoinPending). + Pending { received: i32, required: i32 }, + /// All required inputs are present — push the subscriber. + Fire, +} + +/// Evaluate the AND-join barrier for a partition-bearing subscriber input. +/// Only a partition-bearing input carrying a concrete partition advances the +/// join — a reference input or an unpartitioned producer must never fire a +/// partitioned join (the case-3 silent-wrong guard). On `Err` the caller +/// should log and skip without recording an event. +async fn handle_join( db: &DB, workspace_id: &str, - producer_job_id: Uuid, - subscriber_path: &str, - asset_kind: AssetKind, - asset_path: &str, - outcome: DispatchOutcome, - opts: EventOptions<'_>, -) { + sub_path: &str, + trigger_ref: &str, + partition: Option<&str>, +) -> Result { + if !is_partition_bearing_ref(trigger_ref) { + tracing::debug!( + "AND subscriber {}: non-partition-bearing input {} does not fire the join", + sub_path, + trigger_ref + ); + return Ok(JoinDecision::Skip("case3_non_partition_bearing")); + } + let Some(pv) = partition else { + tracing::warn!( + "AND subscriber {}: partition-bearing input {} arrived with no resolved \ + partition; not dispatching (case-3 guard)", + sub_path, + trigger_ref + ); + return Ok(JoinDecision::Skip("case3_missing_partition")); + }; + match record_and_check_join_slot(db, workspace_id, sub_path, pv, trigger_ref).await? { + JoinSlotStatus { fired: false, received, required } => { + Ok(JoinDecision::Pending { received, required }) + } + JoinSlotStatus { fired: true, .. } => Ok(JoinDecision::Fire), + } +} + +/// Best-effort batched insert into `dispatch_event`. Never propagates — the +/// dispatch contract is "logging failures must not retroactively fail the +/// producer's job." All rows accumulated over a dispatch pass go in one +/// INSERT (UNNEST) to avoid an N+1 across (subscriber × asset write). +async fn flush_events(db: &DB, workspace_id: &str, producer_job_id: Uuid, events: &[EventRow]) { + if events.is_empty() { + return; + } + // Column-oriented arrays for UNNEST. Each Vec is one column across all rows. + let subscriber_paths: Vec = events.iter().map(|e| e.subscriber_path.clone()).collect(); + let asset_kinds: Vec = events.iter().map(|e| e.asset_kind).collect(); + let asset_paths: Vec = events.iter().map(|e| e.asset_path.clone()).collect(); + let outcomes: Vec = events.iter().map(|e| e.outcome).collect(); + let child_job_ids: Vec> = events.iter().map(|e| e.child_job_id).collect(); + let partitions: Vec> = events.iter().map(|e| e.partition.clone()).collect(); + let received_inputs: Vec> = events.iter().map(|e| e.received_inputs).collect(); + let required_inputs: Vec> = events.iter().map(|e| e.required_inputs).collect(); + let debounce_s: Vec> = events.iter().map(|e| e.debounce_s).collect(); + let reasons: Vec> = events.iter().map(|e| e.reason.clone()).collect(); + let res = sqlx::query!( r#"INSERT INTO dispatch_event ( workspace_id, producer_job_id, subscriber_path, @@ -140,26 +236,31 @@ async fn record_event( child_job_id, partition, received_inputs, required_inputs, debounce_s, reason - ) VALUES ($1, $2, $3, $4::ASSET_KIND, $5, $6::DISPATCH_OUTCOME, - $7, $8, $9, $10, $11, $12)"#, + ) + SELECT $1, $2, sp, ak, ap, oc, cj, pt, ri, rq, db, rs + FROM unnest( + $3::text[], $4::ASSET_KIND[], $5::text[], $6::DISPATCH_OUTCOME[], + $7::uuid[], $8::text[], $9::int[], $10::int[], $11::int[], $12::text[] + ) AS t(sp, ak, ap, oc, cj, pt, ri, rq, db, rs)"#, workspace_id, producer_job_id, - subscriber_path, - asset_kind as AssetKind, - asset_path, - outcome as DispatchOutcome, - opts.child_job_id, - opts.partition, - opts.received_inputs, - opts.required_inputs, - opts.debounce_s, - opts.reason, + &subscriber_paths, + asset_kinds as Vec, + &asset_paths, + outcomes as Vec, + &child_job_ids as &[Option], + &partitions as &[Option], + &received_inputs as &[Option], + &required_inputs as &[Option], + &debounce_s as &[Option], + &reasons as &[Option], ) .execute(db) .await; if let Err(e) = res { tracing::error!( - "failed to record dispatch_event for producer {}: {e:#}", + "failed to record {} dispatch_event row(s) for producer {}: {e:#}", + events.len(), producer_job_id ); } @@ -220,8 +321,13 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result } let mut dispatched = Vec::new(); + // Best-effort dispatch_event rows accumulated over the whole pass and + // flushed in one batched INSERT at the end (avoids an N+1 over + // subscriber × asset write). The mid-pass join-slot writes + // (record_and_check_join_slot) are a separate table and unaffected. + let mut events: Vec = Vec::new(); for (asset_kind, asset_path) in writes { - let Some(prefix) = prefix_for(asset_kind) else { + let Some(prefix) = asset_kind.canonical_prefix() else { continue; }; let trigger_ref = format!("{}{}", prefix, asset_path); @@ -230,93 +336,51 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result let Subscriber { path: sub_path, join_all, debounce_s, retry_count, retry_delay_s } = sub; if sub_path == runnable_path { - record_event( - db, - &job.workspace_id, - job.id, + events.push(EventRow::new( &sub_path, asset_kind, &asset_path, DispatchOutcome::Skipped, EventOptions { reason: Some("self_loop"), ..Default::default() }, - ) - .await; + )); continue; } if join_all { - // AND join barrier. Only a partition-bearing input - // (`// on …/{partition}/…`) carrying a concrete partition - // advances the join — a reference input or an - // unpartitioned producer must never fire a partitioned - // join (the case-3 silent-wrong guard). - if !is_partition_bearing_ref(&trigger_ref) { - tracing::debug!( - "AND subscriber {}: non-partition-bearing input {} does not fire the join", - sub_path, - trigger_ref - ); - record_event( - db, - &job.workspace_id, - job.id, - &sub_path, - asset_kind, - &asset_path, - DispatchOutcome::Skipped, - EventOptions { - reason: Some("case3_non_partition_bearing"), - ..Default::default() - }, - ) - .await; - continue; - } - let Some(pv) = partition.as_deref() else { - tracing::warn!( - "AND subscriber {}: partition-bearing input {} arrived with no resolved \ - partition; not dispatching (case-3 guard)", - sub_path, - trigger_ref - ); - record_event( - db, - &job.workspace_id, - job.id, - &sub_path, - asset_kind, - &asset_path, - DispatchOutcome::Skipped, - EventOptions { - reason: Some("case3_missing_partition"), - ..Default::default() - }, - ) - .await; - continue; - }; - match record_and_check_join_slot(db, &job.workspace_id, &sub_path, pv, &trigger_ref) - .await + match handle_join( + db, + &job.workspace_id, + &sub_path, + &trigger_ref, + partition.as_deref(), + ) + .await { - Ok(JoinSlotStatus { fired: false, received, required }) => { - record_event( - db, - &job.workspace_id, - job.id, + Ok(JoinDecision::Skip(reason)) => { + events.push(EventRow::new( + &sub_path, + asset_kind, + &asset_path, + DispatchOutcome::Skipped, + EventOptions { reason: Some(reason), ..Default::default() }, + )); + continue; + } + Ok(JoinDecision::Pending { received, required }) => { + events.push(EventRow::new( &sub_path, asset_kind, &asset_path, DispatchOutcome::JoinPending, EventOptions { - partition: Some(pv), + partition: partition.as_deref(), received_inputs: Some(received), required_inputs: Some(required), ..Default::default() }, - ) - .await; + )); continue; // slot incomplete — wait for the rest } - Ok(JoinSlotStatus { fired: true, .. }) => {} // fall through to push + Ok(JoinDecision::Fire) => {} // fall through to push Err(e) => { tracing::error!("join-slot check failed for {}: {e:#}", sub_path); continue; @@ -339,10 +403,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result .await { Ok(id) => { - record_event( - db, - &job.workspace_id, - job.id, + events.push(EventRow::new( &sub_path, asset_kind, &asset_path, @@ -353,8 +414,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result debounce_s, ..Default::default() }, - ) - .await; + )); dispatched.push(id); } Err(e) => { @@ -364,6 +424,8 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result } } + flush_events(db, &job.workspace_id, job.id, &events).await; + if !dispatched.is_empty() { tracing::info!( "asset-trigger dispatch from job {} ({}): pushed {} downstream jobs", @@ -436,19 +498,6 @@ fn read_partition( serde_json::from_str::(trigger_map?.get(PARTITION_ARG)?.get()).ok() } -fn prefix_for(kind: AssetKind) -> Option<&'static str> { - match kind { - AssetKind::S3Object => Some("s3://"), - AssetKind::Resource => Some("$res:"), - AssetKind::Ducklake => Some("ducklake://"), - AssetKind::DataTable => Some("datatable://"), - AssetKind::Volume => Some("volume://"), - // Deprecated kind from before the parser was unified — has no - // canonical trigger ref and never produced trigger rows. - AssetKind::Variable => None, - } -} - async fn fetch_producer_writes( db: &Pool, workspace_id: &str, @@ -763,7 +812,7 @@ async fn push_subscriber( let mut args: HashMap> = HashMap::new(); let trigger_payload = serde_json::json!({ "kind": "asset", - "asset_kind": serde_json::to_value(&asset_kind).unwrap_or(serde_json::Value::Null), + "asset_kind": serde_json::to_value(&asset_kind).expect("AssetKind serializes"), "asset_path": asset_path, "producer_path": producer_path, "producer_job_id": producer.id.to_string(), diff --git a/backend/windmill-types/src/assets.rs b/backend/windmill-types/src/assets.rs index be20cfae3e..fb624382c4 100644 --- a/backend/windmill-types/src/assets.rs +++ b/backend/windmill-types/src/assets.rs @@ -16,6 +16,23 @@ pub enum AssetKind { Volume, } +impl AssetKind { + /// The canonical URI prefix used in asset trigger refs (e.g. `s3://`, + /// `$res:`). Single source of truth for both trigger-ref construction + /// and runtime cascade dispatch. `Variable` is deprecated and has no + /// canonical ref, so it returns `None`. + pub fn canonical_prefix(&self) -> Option<&'static str> { + match self { + AssetKind::S3Object => Some("s3://"), + AssetKind::Resource => Some("$res:"), + AssetKind::Ducklake => Some("ducklake://"), + AssetKind::DataTable => Some("datatable://"), + AssetKind::Volume => Some("volume://"), + AssetKind::Variable => None, + } + } +} + #[derive( Serialize, Deserialize, Debug, PartialEq, Copy, Clone, Hash, Eq, sqlx::Type, PartialOrd, Ord, )] diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index fdef059c26..ab2cad4da9 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -784,19 +784,11 @@ pub async fn process_completed_job( // by resolve_partition_for_job). Run identity is immutable — // the preprocessor must not change or drop it, or the asset // cascade would read no partition for this producer. - sqlx::query!( - "UPDATE v2_job - SET args = CASE - WHEN args ? 'partition' - THEN $1 || jsonb_build_object('partition', args -> 'partition') - ELSE $1 - END, - preprocessed = TRUE - WHERE id = $2", - Json(preprocessed_args) as Json>>, - job.id + windmill_common::partition::merge_args_preserving_partition( + db, + job.id, + preprocessed_args, ) - .execute(db) .await?; } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 6e9a77597e..984acbe756 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -4385,14 +4385,7 @@ async fn resolve_partition_for_job( // Persist back so dispatch_asset_triggers (which reads the producer's // completed v2_job.args) propagates the same value down the cascade. if let Some(db) = conn.as_sql() { - sqlx::query!( - "UPDATE v2_job SET args = coalesce(args, '{}'::jsonb) \ - || jsonb_build_object('partition', $1::text) WHERE id = $2", - value, - job.id, - ) - .execute(db) - .await?; + windmill_common::partition::set_resolved_partition(db, job.id, &value).await?; } else { tracing::warn!( job_id = %job.id, diff --git a/cli/src/commands/pipeline/pipeline.ts b/cli/src/commands/pipeline/pipeline.ts index 27985af3c7..b4e08f6416 100644 --- a/cli/src/commands/pipeline/pipeline.ts +++ b/cli/src/commands/pipeline/pipeline.ts @@ -10,8 +10,13 @@ import * as log from "../../core/log.ts"; import { GlobalOptions } from "../../types.ts"; // Mirrors the asset-graph endpoint payload (backend/windmill-api-assets). -// Raw-fetched because these routes are newer than the checked-in generated -// client; swap to the generated functions on the next client regen. +// TODO: the checked-in generated client (cli/gen, last regenerated 2025-04) +// predates these routes, so we raw-fetch and hand-roll the types. Once +// `cli/gen` is regenerated (run `cli/gen_wm_client.sh`, which is currently +// >700 openapi.yaml commits stale and would churn the whole client), replace +// `apiGet` + these types with the generated `wmill.getAssetsGraph(...)` +// (operationId getAssetsGraph) and `wmill.listPipelineFolders(...)` +// (operationId listPipelineFolders). type GraphRunnable = { path: string; usage_kind: "script" | "flow" | "job"; @@ -92,6 +97,12 @@ function shortName(scriptPath: string): string { return scriptPath.split("/").pop() ?? scriptPath; } +// Append to a multimap value, creating the bucket on first use. Avoids the +// O(n^2) spread-rebuild pattern (`map.set(k, [...(map.get(k) ?? []), v])`). +function pushTo(map: Map, key: K, val: V): void { + (map.get(key) ?? map.set(key, []).get(key)!).push(val); +} + async function show( opts: GlobalOptions & { json?: boolean }, folder: string, @@ -121,10 +132,7 @@ async function show( for (const e of graph.edges) { if (e.access_type === "w" || e.access_type === "rw") { const uri = assetUri(e.asset_kind, e.asset_path); - writesByScript.set(e.runnable_path, [ - ...(writesByScript.get(e.runnable_path) ?? []), - uri, - ]); + pushTo(writesByScript, e.runnable_path, uri); } } const subsByAsset = new Map(); @@ -137,17 +145,15 @@ async function show( if (t.trigger_kind === "asset") { const at = t as Extract; const uri = assetUri(at.asset_kind, at.asset_path); - subsByAsset.set(uri, [...(subsByAsset.get(uri) ?? []), t.runnable_path]); - subsByScript.set(t.runnable_path, [ - ...(subsByScript.get(t.runnable_path) ?? []), - uri, - ]); + pushTo(subsByAsset, uri, t.runnable_path); + pushTo(subsByScript, t.runnable_path, uri); } else { const nt = t as Exclude; - nativeByScript.set(t.runnable_path, [ - ...(nativeByScript.get(t.runnable_path) ?? []), - { kind: nt.trigger_kind, path: nt.path, missing: nt.missing }, - ]); + pushTo(nativeByScript, t.runnable_path, { + kind: nt.trigger_kind, + path: nt.path, + missing: nt.missing, + }); } } @@ -205,9 +211,16 @@ async function show( .sort(); // UI-first markers (data_upload, webhook) have no trigger row — the - // graph endpoint can't surface them, they live as `// on ` - // annotations in the body. Roots are where sources matter, so fetch just - // those bodies and lift the marker kinds the canvas would show. + // graph endpoint's trigger enum (schedule/email/kafka/mqtt/nats/postgres/ + // sqs/gcp) can't surface them, so they only exist as `// on ` + // annotations in the script body. Roots are where sources matter, so fetch + // just those bodies and lift the marker kinds the canvas would show. + // + // DRIFT RISK: this regex + MARKER_KINDS is a divergent, partial copy of the + // canonical annotation parser. The proper fix is to have the graph endpoint + // emit these UI-only markers as trigger rows (a backend change), after which + // this whole Promise.all body-fetch can be deleted and read straight from + // the response. Until then, keep this list in sync with the canonical parser. const MARKER_KINDS = ["data_upload", "webhook", "email"]; await Promise.all( roots.map(async (p) => { diff --git a/data-pipelines-sync-demo.zip b/data-pipelines-sync-demo.zip deleted file mode 100644 index 9a2776993e..0000000000 Binary files a/data-pipelines-sync-demo.zip and /dev/null differ diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index e5c3605033..9e42eda58d 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -530,7 +530,6 @@ : { inPipeline: false, triggerAssets: [], - schedules: [], nativeTriggers: [] } ) diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetRunsPanel.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetRunsPanel.svelte index 0d0908cffd..f43259b561 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetRunsPanel.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetRunsPanel.svelte @@ -404,6 +404,7 @@
Logs
+ import { workspaceStore } from '$lib/stores' + import { sendUserToast } from '$lib/utils' + import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte' + import type { NativeTriggerKind } from './types' + import { + EmailTriggerService, + GcpTriggerService, + KafkaTriggerService, + MqttTriggerService, + NatsTriggerService, + PostgresTriggerService, + ScheduleService, + SqsTriggerService + } from '$lib/gen' + import KafkaTriggerEditor from '$lib/components/triggers/kafka/KafkaTriggerEditor.svelte' + import MqttTriggerEditor from '$lib/components/triggers/mqtt/MqttTriggerEditor.svelte' + import NatsTriggerEditor from '$lib/components/triggers/nats/NatsTriggerEditor.svelte' + import PostgresTriggerEditor from '$lib/components/triggers/postgres/PostgresTriggerEditor.svelte' + import SqsTriggerEditor from '$lib/components/triggers/sqs/SqsTriggerEditor.svelte' + import GcpTriggerEditor from '$lib/components/triggers/gcp/GcpTriggerEditor.svelte' + import EmailTriggerEditor from '$lib/components/triggers/email/EmailTriggerEditor.svelte' + import ScheduleEditor from '$lib/components/triggers/schedules/ScheduleEditor.svelte' + import WebhookEditor from '$lib/components/triggers/webhook/WebhookEditor.svelte' + + // Owns the native-trigger drawer wiring for the pipeline canvas: the nine + // editor instances, the create/edit dispatch by kind, and the delete + // confirmation flow. The page drives it imperatively via `bind:this` so + // the draft-guard toasts (which depend on the page's drafts map) stay in + // the page while the editor plumbing lives here. + // + // `mountTriggerEditors` gates the eight editor instances: they're + // edit-mode-only on the canvas, so the page unmounts them off edit mode + // (matching the previous `{#if mode === 'edit'}` wrapper). The webhook + // editor stays mounted in every mode — its node is clickable in view mode + // too (informational endpoint URLs/token). + type Props = { onUpdate: () => void; mountTriggerEditors: boolean } + let { onUpdate, mountTriggerEditors }: Props = $props() + + let kafkaEditor: KafkaTriggerEditor | undefined = $state() + let mqttEditor: MqttTriggerEditor | undefined = $state() + let natsEditor: NatsTriggerEditor | undefined = $state() + let postgresEditor: PostgresTriggerEditor | undefined = $state() + let sqsEditor: SqsTriggerEditor | undefined = $state() + let gcpEditor: GcpTriggerEditor | undefined = $state() + let emailEditor: EmailTriggerEditor | undefined = $state() + let scheduleEditor: ScheduleEditor | undefined = $state() + let webhookEditor: WebhookEditor | undefined = $state() + + // Open the create-trigger drawer for `kind`, pre-filling `script_path`. + // Caller is responsible for the draft guard (a trigger row can only point + // at a deployed script). + export function openNew(kind: NativeTriggerKind, scriptPath: string) { + switch (kind) { + case 'schedule': + return scheduleEditor?.openNew(false, scriptPath, undefined, scriptPath) + case 'kafka': + return kafkaEditor?.openNew(false, scriptPath) + case 'mqtt': + return mqttEditor?.openNew(false, scriptPath) + case 'nats': + return natsEditor?.openNew(false, scriptPath) + case 'postgres': + return postgresEditor?.openNew(false, scriptPath) + case 'sqs': + return sqsEditor?.openNew(false, scriptPath) + case 'gcp': + return gcpEditor?.openNew(false, scriptPath) + case 'email': + return emailEditor?.openNew(false, scriptPath) + // webhook has no dedicated editor. + default: + return + } + } + + export function openEdit(kind: NativeTriggerKind, triggerPath: string, scriptPath: string) { + switch (kind) { + case 'schedule': + return scheduleEditor?.openEdit(triggerPath, false, scriptPath) + case 'kafka': + return kafkaEditor?.openEdit(triggerPath, false, scriptPath) + case 'mqtt': + return mqttEditor?.openEdit(triggerPath, false, scriptPath) + case 'nats': + return natsEditor?.openEdit(triggerPath, false, scriptPath) + case 'postgres': + return postgresEditor?.openEdit(triggerPath, false, scriptPath) + case 'sqs': + return sqsEditor?.openEdit(triggerPath, false, scriptPath) + case 'gcp': + return gcpEditor?.openEdit(triggerPath, false, scriptPath) + case 'email': + return emailEditor?.openEdit(triggerPath, false, scriptPath) + default: + return + } + } + + // Open the webhook drawer (endpoint URLs + token flow). Caller owns the + // draft guard. + export function openWebhook(scriptPath: string) { + webhookEditor?.openDrawer(scriptPath, false) + } + + // Delete-trigger confirmation state. Kebab → Delete opens the standard + // ConfirmationModal; the actual delete is dispatched from onConfirmed so + // the dialog stays consistent with the rest of the app. + let triggerDeleteTarget = $state<{ kind: NativeTriggerKind; path: string } | undefined>(undefined) + let triggerDeleteLoading = $state(false) + let triggerDeleteOpen = $derived(triggerDeleteTarget != undefined) + + export function requestDelete(kind: NativeTriggerKind, triggerPath: string) { + triggerDeleteTarget = { kind, path: triggerPath } + } + + async function confirmDeleteAttachedTrigger() { + if (!triggerDeleteTarget || !$workspaceStore) return + const { kind, path: triggerPath } = triggerDeleteTarget + const workspace = $workspaceStore + triggerDeleteLoading = true + try { + switch (kind) { + case 'schedule': + await ScheduleService.deleteSchedule({ workspace, path: triggerPath }) + break + case 'kafka': + await KafkaTriggerService.deleteKafkaTrigger({ workspace, path: triggerPath }) + break + case 'mqtt': + await MqttTriggerService.deleteMqttTrigger({ workspace, path: triggerPath }) + break + case 'nats': + await NatsTriggerService.deleteNatsTrigger({ workspace, path: triggerPath }) + break + case 'postgres': + await PostgresTriggerService.deletePostgresTrigger({ workspace, path: triggerPath }) + break + case 'sqs': + await SqsTriggerService.deleteSqsTrigger({ workspace, path: triggerPath }) + break + case 'gcp': + await GcpTriggerService.deleteGcpTrigger({ workspace, path: triggerPath }) + break + case 'email': + await EmailTriggerService.deleteEmailTrigger({ workspace, path: triggerPath }) + break + default: + return + } + sendUserToast(`Deleted ${kind} trigger "${triggerPath}"`) + triggerDeleteTarget = undefined + onUpdate() + } catch (e: any) { + sendUserToast( + `Could not delete ${kind} trigger "${triggerPath}": ${e?.body ?? e?.message ?? String(e)}`, + true + ) + } finally { + triggerDeleteLoading = false + } + } + + +{#if mountTriggerEditors} + { + if (!triggerDeleteLoading) triggerDeleteTarget = undefined + }} + > + {#if triggerDeleteTarget} +

+ Delete {triggerDeleteTarget.path}? The + // on {triggerDeleteTarget.kind} annotation on the script stays + — the trigger will read as missing on the canvas until you recreate it or remove the annotation. +

+ {/if} + + + + + + + + + + + +{/if} + + diff --git a/frontend/src/lib/components/assets/AssetGraph/deployGraphDiff.ts b/frontend/src/lib/components/assets/AssetGraph/deployGraphDiff.ts index 4d903278a0..3b3c7de6a1 100644 --- a/frontend/src/lib/components/assets/AssetGraph/deployGraphDiff.ts +++ b/frontend/src/lib/components/assets/AssetGraph/deployGraphDiff.ts @@ -1,4 +1,5 @@ import type { AssetGraphResponse } from './types' +import { assetKey, isWriteEdge } from './lib' // Post-deploy graph verification: the canvas previews a draft's edges from // the *frontend* parsers (resolveGraph overlay); the deployed rows are @@ -24,16 +25,14 @@ export type CascadeFacts = { export function extractCascadeFacts(g: AssetGraphResponse, scriptPath: string): CascadeFacts { const writes = new Set() for (const e of g.edges ?? []) { - if (e.runnable_kind !== 'script' || e.runnable_path !== scriptPath) continue - const access = e.access_type ?? 'r' - if (access !== 'w' && access !== 'rw') continue - writes.add(`${e.asset_kind}:${e.asset_path}`) + if (e.runnable_path !== scriptPath || !isWriteEdge(e)) continue + writes.add(assetKey(e)) } const subs = new Set() for (const t of g.triggers ?? []) { if (t.trigger_kind !== 'asset') continue if (t.runnable_kind !== 'script' || t.runnable_path !== scriptPath) continue - subs.add(`${t.asset_kind}:${t.asset_path}`) + subs.add(assetKey(t)) } return { writes, subs } } diff --git a/frontend/src/lib/components/assets/AssetGraph/graphTraversal.test.ts b/frontend/src/lib/components/assets/AssetGraph/graphTraversal.test.ts index b876882dc8..08da865d61 100644 --- a/frontend/src/lib/components/assets/AssetGraph/graphTraversal.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/graphTraversal.test.ts @@ -1,6 +1,11 @@ import { describe, expect, it } from 'vitest' import type { AssetGraphResponse } from './types' -import { computeDownstreamClosure, getDownstreamSubscribers } from './graphTraversal' +import { buildDownstreamMap, computeDownstreamClosure } from './graphTraversal' + +/** Direct (one-hop) subscriber script paths of `scriptPath`. */ +function downstreamSubscribers(g: AssetGraphResponse, scriptPath: string): string[] { + return Array.from(buildDownstreamMap(g).get(scriptPath) ?? []) +} // Minimal graph builder: `writes` are producer→asset write edges, `subs` are // `// on ` subscriptions. All datatable/script for brevity. @@ -37,7 +42,7 @@ describe('buildDownstreamMap', () => { ['c', 'x'] ] ) - expect(getDownstreamSubscribers(g, 'a').sort()).toEqual(['b', 'c']) + expect(downstreamSubscribers(g, 'a').sort()).toEqual(['b', 'c']) }) it('ignores read edges, flow subscribers and self-loops', () => { @@ -57,8 +62,8 @@ describe('buildDownstreamMap', () => { runnable_kind: 'flow', runnable_path: 'f' }) - expect(getDownstreamSubscribers(g, 'a')).toEqual([]) - expect(getDownstreamSubscribers(g, 'r')).toEqual([]) + expect(downstreamSubscribers(g, 'a')).toEqual([]) + expect(downstreamSubscribers(g, 'r')).toEqual([]) }) it('matches on asset kind as well as path', () => { @@ -70,7 +75,7 @@ describe('buildDownstreamMap', () => { runnable_kind: 'script', runnable_path: 'b' }) - expect(getDownstreamSubscribers(g, 'a')).toEqual([]) + expect(downstreamSubscribers(g, 'a')).toEqual([]) }) }) diff --git a/frontend/src/lib/components/assets/AssetGraph/graphTraversal.ts b/frontend/src/lib/components/assets/AssetGraph/graphTraversal.ts index 736d5aa592..6b53020ddb 100644 --- a/frontend/src/lib/components/assets/AssetGraph/graphTraversal.ts +++ b/frontend/src/lib/components/assets/AssetGraph/graphTraversal.ts @@ -1,4 +1,5 @@ import type { AssetGraphResponse } from './types' +import { assetKey, buildAssetSubscribers, isWriteEdge } from './lib' // Execution-DAG traversal over the resolved asset graph (drafts included). // @@ -13,20 +14,11 @@ import type { AssetGraphResponse } from './types' /** `producer script path` → set of subscriber script paths (one hop). */ export function buildDownstreamMap(g: AssetGraphResponse): Map> { - const subscribersByAsset = new Map>() - for (const t of g.triggers ?? []) { - if (t.trigger_kind !== 'asset' || t.runnable_kind !== 'script') continue - const key = `${t.asset_kind}:${t.asset_path}` - const set = subscribersByAsset.get(key) ?? new Set() - set.add(t.runnable_path) - subscribersByAsset.set(key, set) - } + const subscribersByAsset = buildAssetSubscribers(g) const downstream = new Map>() for (const e of g.edges ?? []) { - if (e.runnable_kind !== 'script') continue - const access = e.access_type ?? 'r' - if (access !== 'w' && access !== 'rw') continue - const subs = subscribersByAsset.get(`${e.asset_kind}:${e.asset_path}`) + if (!isWriteEdge(e)) continue + const subs = subscribersByAsset.get(assetKey(e)) if (!subs) continue const merged = downstream.get(e.runnable_path) ?? new Set() for (const s of subs) if (s !== e.runnable_path) merged.add(s) @@ -35,11 +27,6 @@ export function buildDownstreamMap(g: AssetGraphResponse): Map` declarations). + * Flows are excluded, mirroring the backend dispatch policy. + */ +export function buildAssetSubscribers(g: AssetGraphResponse): Map> { + const subscribersByAsset = new Map>() + for (const t of g.triggers ?? []) { + if (t.trigger_kind !== 'asset' || t.runnable_kind !== 'script') continue + const key = assetKey(t) + const set = subscribersByAsset.get(key) ?? new Set() + set.add(t.runnable_path) + subscribersByAsset.set(key, set) + } + return subscribersByAsset +} diff --git a/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.parity.test.ts b/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.parity.test.ts index c03cfe1d78..dfb5385253 100644 --- a/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.parity.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.parity.test.ts @@ -2,7 +2,23 @@ import { readFileSync } from 'node:fs' import { resolve, dirname } from 'node:path' import { fileURLToPath } from 'node:url' import { describe, expect, it } from 'vitest' -import { parsePipelineAnnotations } from './parsePipelineAnnotations' +import { parsePipelineAnnotations, type PipelineAnnotations } from './parsePipelineAnnotations' + +// Every field the TS parser produces, each with an assertion in the per-fixture +// test below. Typed as `Record` so adding a +// field to `PipelineAnnotations` is a compile error here until it's listed — +// forcing the author to also wire up its parity assertion. (Deploy-only Rust +// fields like join_mode / debounce_default are not part of this type, so they +// correctly never appear here.) +const ASSERTED_TS_FIELDS: Record = { + inPipeline: true, + triggerAssets: true, + nativeTriggers: true, + partition: true, + freshness: true, + tag: true, + retry: true +} // Parser-parity guard: this TS parser (drives the live graph preview) and // the Rust `parse_pipeline_annotations` (drives deploy) must stay @@ -14,8 +30,16 @@ import { parsePipelineAnnotations } from './parsePipelineAnnotations' // Rust counterpart: backend/parsers/windmill-parser/tests/ // pipeline_annotations_parity.rs. Extend the corpus when the grammar // changes; a fixture passing on one side and failing on the other is -// exactly the drift this exists to catch. Only fields both parsers produce -// are compared (join_mode / debounce_default are deploy-only, Rust-side). +// exactly the drift this exists to catch. +// +// Intentional divergence — the Rust parser is a superset. It also parses +// `join_mode` and `debounce_default`, which are DEPLOY-ONLY: they affect how +// the backend schedules cascade runs, never the rendered graph, so the TS +// preview parser deliberately doesn't produce them and they are not compared +// here. Every field the TS parser DOES produce is compared, and the +// `ASSERTED_TS_FIELDS` exhaustiveness check above fails the suite if a new TS +// field is added without a matching assertion — so a field can't be parsed on +// the TS side yet silently skipped by this guard. type Fixture = { name: string @@ -48,6 +72,18 @@ describe('parsePipelineAnnotations matches the shared Rust fixture corpus', () = expect(fixtures.length).toBeGreaterThan(0) }) + it('every field the parser emits across the corpus has a parity assertion', () => { + const asserted = new Set(Object.keys(ASSERTED_TS_FIELDS)) + const emitted = new Set() + for (const f of fixtures) { + for (const k of Object.keys(parsePipelineAnnotations(f.code))) emitted.add(k) + } + expect( + [...emitted].filter((k) => !asserted.has(k)), + 'unasserted parser fields' + ).toEqual([]) + }) + for (const f of fixtures) { it(f.name, () => { const got = parsePipelineAnnotations(f.code) diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts index bfa316979e..6d69c564ab 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineHistory.svelte.ts @@ -46,6 +46,10 @@ export function usePipelineHistory( async function load(ws: string, prefix: string, days: number) { const myGen = ++gen + // A folder/days change (or unmount) bumps `gen`; once it diverges from + // this call's `myGen`, every write below this point belongs to a stale + // scope and must be dropped. + const isStale = () => gen !== myGen loading = true error = undefined try { @@ -64,7 +68,7 @@ export function usePipelineHistory( perPage: PER_PAGE, page: pages }) - if (gen !== myGen) return + if (isStale()) return for (const j of rows) { out.push({ id: j.id, @@ -89,15 +93,15 @@ export function usePipelineHistory( } catch (e) { console.warn('failed to load pipeline dispatch edges', e) } - if (gen !== myGen) return + if (isStale()) return events = out edges = edgeRows truncated = sawFullPage } catch (e: any) { - if (gen !== myGen) return + if (isStale()) return error = e?.body ?? e?.message ?? String(e) } finally { - if (gen === myGen) loading = false + if (!isStale()) loading = false } } diff --git a/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts b/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts index 3e09abc7c1..44e501ac3e 100644 --- a/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts +++ b/frontend/src/lib/components/assets/AssetGraph/pipelineTemplates.ts @@ -1,5 +1,6 @@ import type { ScriptLang, AssetKind } from '$lib/gen' import { random_adj } from '$lib/components/random_positive_adjetive' +import { parseDbInputFromAssetSyntax } from '$lib/utils' // What kind of asset the new script will produce. Drives the random output // path scheme and the body skeleton. The output asset is NOT declared in a @@ -77,8 +78,10 @@ const LANG_COMPATIBILITY: Record = { csharp: ['none'], graphql: ['none'], bunnative: ['none'], - nativets: ['none'] -} as any + nativets: ['none'], + ruby: ['none'], + rlang: ['none'] +} export function compatibleOutputKinds(lang: ScriptLang): PipelineOutputKind[] { return LANG_COMPATIBILITY[lang] ?? ['none'] @@ -182,9 +185,12 @@ function s3Key(path: string): string { } // Splits a datatable asset path (`/` or `/.
`) -// into its constituent parts. Mirrors `parseDbInputFromAssetSyntax` in -// $lib/utils.ts: schema is omitted when the table lives in the default -// (`public`) schema, prefixed with `.` only when explicit. +// into its constituent parts. The `.
` grammar is owned by +// `parseDbInputFromAssetSyntax` in $lib/utils.ts (which consumes a full +// `datatable://…` URI): we delegate to it for the schema/table split so the +// two stay in lockstep, and only keep the leading `/` extraction here — +// the util folds the db into `resourcePath`, which is awkward to read back, +// and it has no slashless fallback (the SQL emitters tolerate a bare db). function parseDatatablePath(p: string): { db: string schema: string | undefined @@ -193,10 +199,11 @@ function parseDatatablePath(p: string): { const slash = p.indexOf('/') if (slash < 0) return { db: p, schema: undefined, table: '' } const db = p.slice(0, slash) - const tail = p.slice(slash + 1) - const dot = tail.indexOf('.') - if (dot < 0) return { db, schema: undefined, table: tail } - return { db, schema: tail.slice(0, dot), table: tail.slice(dot + 1) } + // `datatable://` URIs always parse to the `database` variant, so the + // `specificSchema`/`specificTable` accessors below are always present. + const parsed = parseDbInputFromAssetSyntax(`datatable://${p}`) + const schema = parsed?.type === 'database' ? parsed.specificSchema : undefined + return { db, schema, table: parsed?.specificTable ?? '' } } // Schema-qualified table reference for SQL emission. We always emit the diff --git a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts index d6067a347a..6e4f512000 100644 --- a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts +++ b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts @@ -35,6 +35,67 @@ export type ResolveGraphInput = { annotatedNativeKindsByPath: Map> } +/** Mutable bag the sequential passes accumulate the resolved graph into. */ +type Accumulator = { + runnables: AssetGraphResponse['runnables'] + assets: AssetGraphResponse['assets'] + edges: AssetGraphResponse['edges'] + extraTriggers: AssetGraphResponse['triggers'] +} + +/** + * Native trigger kinds with a persisted (non-asset) trigger row pointing at + * `path`. These bind by script path and survive content edits, so they dedup + * against draft/live annotations to avoid showing a real source node next to a + * red "missing" placeholder for the same kind. + */ +function persistedNativeKinds(base: AssetGraphResponse, path: string): Set { + return new Set( + base.triggers + .filter( + (t) => + t.trigger_kind !== 'asset' && t.runnable_kind === 'script' && t.runnable_path === path + ) + .map((t) => t.trigger_kind) + ) +} + +/** `kind:path` keys of persisted asset (`// on `) triggers for `path`. */ +function persistedAssetKeys(base: AssetGraphResponse, path: string): Set { + return new Set( + base.triggers + .filter( + (t) => + t.trigger_kind === 'asset' && t.runnable_kind === 'script' && t.runnable_path === path + ) + .map((t) => (t.trigger_kind === 'asset' ? `${t.asset_kind}:${t.asset_path}` : '')) + ) +} + +/** + * Emit a red "missing" placeholder for each annotated native kind that has no + * matching attached trigger row. `opts.unsaved` marks editor-driven overlays + * (drafts / live buffer); deployed-but-unswept scripts pass `unsaved: false`. + */ +function pushMissingNativeTriggers( + extraTriggers: AssetGraphResponse['triggers'], + annotatedKinds: Iterable, + attachedKinds: Set, + path: string, + opts: { unsaved: boolean } +) { + for (const kind of annotatedKinds) { + if (attachedKinds.has(kind)) continue + extraTriggers.push({ + trigger_kind: kind, + runnable_kind: 'script', + runnable_path: path, + ...(opts.unsaved ? { unsaved: true } : {}), + missing: true + }) + } +} + /** * Merge the persisted base graph with the draft, session-inferred and * open-script live overlays into one `AssetGraphResponse`. @@ -50,41 +111,61 @@ export type ResolveGraphInput = { * `graphWithDraft` `$derived`. See `resolveGraph.test.ts`. */ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse { - const { - base, - drafts, - liveBodyAssets, - liveAnnotations, - inferredWritesByPath, - inferredReadsByPath, - annotatedNativeKindsByPath - } = input + const { base } = input + const ctx = makeContext(input) + + // Pass order mirrors the precedence comment, lowest → highest: + const acc = seedAccumulator(input, ctx) + seedDraftOverlays(acc, input) + applyLiveBufferOverlay(acc, input, ctx) + crossCheckSweptScripts(acc, input) + overlayInferredLineage(acc, input) + + // Drop persisted ASSET triggers for drafted paths — those come from the + // deployed `// on ` annotations, which the draft's live/parsed + // annotations now own. Native triggers (kafka/schedule/…) are kept: they + // bind by `script_path`, which the draft shares, so the attachment is still + // valid regardless of content edits. + const baseTriggers = base.triggers.filter((t) => { + if (t.trigger_kind !== 'asset') return true + if (ctx.isDrafted(t.runnable_kind, t.runnable_path)) return false + if ( + t.runnable_kind === 'script' && + t.runnable_path === ctx.openPath && + ctx.staleForOpen(t.asset_kind, t.asset_path) + ) + return false + return true + }) + return { + ...base, + assets: acc.assets, + runnables: acc.runnables, + edges: acc.edges, + triggers: [...baseTriggers, ...acc.extraTriggers] + } +} + +type ResolveContext = { + draftedPaths: Set + isDrafted: (kind: string, p: string) => boolean + openPath: string | undefined + staleForOpen: (kind: AssetKind, path: string) => boolean +} + +/** Derive the shared "what's drafted / open / stale" predicates once. */ +function makeContext(input: ResolveGraphInput): ResolveContext { + const { drafts, liveBodyAssets, liveAnnotations } = input - // Every draft contributes: a runnable, an output asset, a write edge, - // plus its own seeded schedule trigger (template includes `// on - // schedule "0 * * * *"` by default, picked up through live parse). - // We iterate the whole `drafts` map so multiple concurrent drafts - // all render as their own subgraph at once. - const runnables = [...base.runnables] - const assets = [...base.assets] // A drafted path's content-derived lineage is owned entirely by the draft // overlay (live inference for the active draft, parsed annotations / - // snapshot for inactive ones). Drop the persisted base asset edges for those + // snapshot for inactive ones). Drop persisted base asset edges for those // paths so it shows only the draft's I/O, not the union of the saved version // and the in-flight edits. (Base asset triggers are dropped at the return; // native triggers are kept — they bind by path.) const draftedPaths = new Set(drafts.keys()) const isDrafted = (kind: string, p: string) => kind === 'script' && draftedPaths.has(p) - // Editing a *saved* script does NOT create a `drafts` entry — it stays an - // open selection with live overlays. So the draft filter above doesn't cover - // it: its persisted edges/triggers would linger next to the live edits (e.g. - // renaming an input leaves the old asset linked alongside the new one). For - // the open, live-inferred saved script, treat its current content as - // authoritative and drop persisted asset edges/triggers it no longer - // references. `liveRefKeys` unions the reliable `// on ` annotations - // with the inferred body assets, so an unchanged selection drops nothing - // (every base asset is still referenced). const openPath = liveBodyAssets.scriptPath const openIsSavedEdit = openPath !== undefined && !draftedPaths.has(openPath) const liveRefKeys = new Set() @@ -98,17 +179,39 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse { const staleForOpen = (kind: AssetKind, path: string) => openIsSavedEdit && !liveRefKeys.has(`${kind}:${path}`) + return { draftedPaths, isDrafted, openPath, staleForOpen } +} + +/** Base graph plus the persisted edges that survive the draft/open filter. */ +function seedAccumulator(input: ResolveGraphInput, ctx: ResolveContext): Accumulator { + const { base } = input const edges = base.edges.filter((e) => { - if (isDrafted(e.runnable_kind, e.runnable_path)) return false + if (ctx.isDrafted(e.runnable_kind, e.runnable_path)) return false if ( e.runnable_kind === 'script' && - e.runnable_path === openPath && - staleForOpen(e.asset_kind, e.asset_path) + e.runnable_path === ctx.openPath && + ctx.staleForOpen(e.asset_kind, e.asset_path) ) return false return true }) - const extraTriggers: AssetGraphResponse['triggers'] = [] + return { + runnables: [...base.runnables], + assets: [...base.assets], + edges, + extraTriggers: [] + } +} + +/** + * Every draft contributes: a runnable, output asset(s), a write edge, live + * read lineage for the active draft, plus its seeded asset/native triggers + * (template includes `// on schedule …` by default). Iterates the whole + * `drafts` map so multiple concurrent drafts all render at once. + */ +function seedDraftOverlays(acc: Accumulator, input: ResolveGraphInput) { + const { base, drafts, liveBodyAssets } = input + const { runnables, assets, edges, extraTriggers } = acc for (const [path, d] of drafts) { const parsed = parsePipelineAnnotations(d.script.content) @@ -238,140 +341,110 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse { // they bind by path and are not dropped above. Dedup against those // so the canvas doesn't show the real schedule node next to a red // "missing" placeholder for the same kind. - const persistedNativeKinds = new Set( - base.triggers - .filter( - (t) => - t.trigger_kind !== 'asset' && - t.runnable_kind === 'script' && - t.runnable_path === path - ) - .map((t) => t.trigger_kind) + pushMissingNativeTriggers( + extraTriggers, + parsed.nativeTriggers.map((n) => n.kind), + persistedNativeKinds(base, path), + path, + { unsaved: true } ) - for (const n of parsed.nativeTriggers) { - if (persistedNativeKinds.has(n.kind)) continue - extraTriggers.push({ - trigger_kind: n.kind, - runnable_kind: 'script', - runnable_path: path, - unsaved: true, - missing: true - }) - } } +} - // Live-parsed overlay for the currently-open script — takes precedence - // over the seeded-template triggers for the same path by swapping - // them out. Scoped to one path (only one pane is open at a time). +/** + * Live-parsed overlay for the currently-open script — takes precedence over + * the seeded-template triggers for the same path by swapping them out. Scoped + * to one path (only one pane is open at a time). + */ +function applyLiveBufferOverlay(acc: Accumulator, input: ResolveGraphInput, ctx: ResolveContext) { + const { base, liveAnnotations } = input + const { assets, extraTriggers } = acc const livePath = liveAnnotations.scriptPath - if (livePath) { - // When the open buffer is a draft, its base asset triggers were dropped - // above (draft-owned), so don't dedup live asset triggers against them — - // every live `// on ` is emitted as unsaved. For a non-draft open - // script (viewing a saved one), keep deduping so live re-parse doesn't - // double the persisted triggers. - const persistedAssetKeys = draftedPaths.has(livePath) - ? new Set() - : new Set( - base.triggers - .filter( - (t) => - t.trigger_kind === 'asset' && - t.runnable_kind === 'script' && - t.runnable_path === livePath - ) - .map((t) => (t.trigger_kind === 'asset' ? `${t.asset_kind}:${t.asset_path}` : '')) - ) - // Strip seeded triggers we computed above for the active draft; - // live annotations are authoritative for the open buffer. - for (let i = extraTriggers.length - 1; i >= 0; i--) { - if (extraTriggers[i].runnable_path === livePath) extraTriggers.splice(i, 1) - } - for (const a of liveAnnotations.annotations.triggerAssets) { - const key = `${a.kind}:${a.path}` - if (persistedAssetKeys.has(key)) continue - extraTriggers.push({ - trigger_kind: 'asset', - asset_kind: a.kind, - asset_path: a.path, - runnable_kind: 'script', - runnable_path: livePath, - unsaved: true - }) - // Synthesize the asset node so the new trigger edge has a - // target — without this, typing `// on s3:///...` adds an - // edge to a node that doesn't exist and the canvas silently - // drops it. Mirrors the draft branch above. - if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) { - assets.push({ kind: a.kind, path: a.path }) - } - } - // Native trigger annotations: kinds for which a matching trigger - // row was found in the backend response. If the live buffer - // declares `// on kafka` and at least one kafka_trigger row points - // at this script, the source node is already on the canvas — no - // overlay needed. Otherwise emit a "missing" placeholder so the - // user can either create the trigger row or remove the annotation. - const persistedNativeKinds = new Set( - base.triggers - .filter( - (t) => - t.trigger_kind !== 'asset' && - t.runnable_kind === 'script' && - t.runnable_path === livePath - ) - .map((t) => t.trigger_kind) - ) - for (const n of liveAnnotations.annotations.nativeTriggers) { - if (persistedNativeKinds.has(n.kind)) continue - extraTriggers.push({ - trigger_kind: n.kind, - runnable_kind: 'script', - runnable_path: livePath, - unsaved: true, - missing: true - }) + if (!livePath) return + + // When the open buffer is a draft, its base asset triggers were dropped + // above (draft-owned), so don't dedup live asset triggers against them — + // every live `// on ` is emitted as unsaved. For a non-draft open + // script (viewing a saved one), keep deduping so live re-parse doesn't + // double the persisted triggers. + const assetKeys = ctx.draftedPaths.has(livePath) + ? new Set() + : persistedAssetKeys(base, livePath) + // Strip seeded triggers we computed above for the active draft; + // live annotations are authoritative for the open buffer. + for (let i = extraTriggers.length - 1; i >= 0; i--) { + if (extraTriggers[i].runnable_path === livePath) extraTriggers.splice(i, 1) + } + for (const a of liveAnnotations.annotations.triggerAssets) { + const key = `${a.kind}:${a.path}` + if (assetKeys.has(key)) continue + extraTriggers.push({ + trigger_kind: 'asset', + asset_kind: a.kind, + asset_path: a.path, + runnable_kind: 'script', + runnable_path: livePath, + unsaved: true + }) + // Synthesize the asset node so the new trigger edge has a + // target — without this, typing `// on s3:///...` adds an + // edge to a node that doesn't exist and the canvas silently + // drops it. Mirrors the draft branch above. + if (!assets.some((x) => x.kind === a.kind && x.path === a.path)) { + assets.push({ kind: a.kind, path: a.path }) } } + // Native trigger annotations: kinds for which a matching trigger + // row was found in the backend response. If the live buffer + // declares `// on kafka` and at least one kafka_trigger row points + // at this script, the source node is already on the canvas — no + // overlay needed. Otherwise emit a "missing" placeholder so the + // user can either create the trigger row or remove the annotation. + pushMissingNativeTriggers( + extraTriggers, + liveAnnotations.annotations.nativeTriggers.map((n) => n.kind), + persistedNativeKinds(base, livePath), + livePath, + { unsaved: true } + ) +} - // Cross-check for already-deployed scripts (not the open buffer): if a - // script's persisted body declares `// on kafka` but no matching - // kafka_trigger row points at it, surface a red placeholder. The - // annotated-kinds map is filled by the page-level prefetch sweep - // (one read per script in the folder); drafts and the active editor - // are handled by the loops above. Scripts that haven't been swept - // yet contribute nothing here — they'll surface on the next refetch. - const livePathExcl = livePath +/** + * Cross-check for already-deployed scripts (not the open buffer): if a + * script's persisted body declares `// on kafka` but no matching kafka_trigger + * row points at it, surface a red placeholder. The annotated-kinds map is + * filled by the page-level prefetch sweep (one read per script in the folder); + * drafts and the active editor are handled by the earlier passes. Scripts not + * yet swept contribute nothing — they surface on the next refetch. + */ +function crossCheckSweptScripts(acc: Accumulator, input: ResolveGraphInput) { + const { base, drafts, liveAnnotations, annotatedNativeKindsByPath } = input + const livePath = liveAnnotations.scriptPath for (const [scriptPath, kinds] of annotatedNativeKindsByPath) { if (drafts.has(scriptPath)) continue - if (scriptPath === livePathExcl) continue - const attachedKinds = new Set( - base.triggers - .filter( - (t) => - t.trigger_kind !== 'asset' && - t.runnable_kind === 'script' && - t.runnable_path === scriptPath - ) - .map((t) => t.trigger_kind) + if (scriptPath === livePath) continue + pushMissingNativeTriggers( + acc.extraTriggers, + kinds, + persistedNativeKinds(base, scriptPath), + scriptPath, + { unsaved: false } ) - for (const kind of kinds) { - if (attachedKinds.has(kind)) continue - extraTriggers.push({ - trigger_kind: kind, - runnable_kind: 'script', - runnable_path: scriptPath, - missing: true - }) - } } +} + +/** + * Live body-asset lineage for any persisted script inferred at least once this + * session (maps filled by `handleAssetsChange` + the load prefetch). Drafts + * are handled by `seedDraftOverlays`. For scripts whose deploy didn't persist + * their body assets (e.g. older WASM at save time, or object-form + * writeS3File), this keeps the lineage edge on the canvas across selection + * changes — not just while selected. + */ +function overlayInferredLineage(acc: Accumulator, input: ResolveGraphInput) { + const { base, drafts, inferredWritesByPath, inferredReadsByPath } = input + const { assets, edges } = acc - // Live body-asset lineage for any persisted script inferred at least - // once this session (maps filled by `handleAssetsChange` + the load - // prefetch). Drafts are handled by the loop above. For scripts whose - // deploy didn't persist their body assets (e.g. older WASM at save - // time, or object-form writeS3File), this keeps the lineage edge on - // the canvas across selection changes — not just while selected. const overlayLineage = ( byPath: Map>, access: 'w' | 'r' @@ -407,22 +480,4 @@ export function resolveGraph(input: ResolveGraphInput): AssetGraphResponse { } overlayLineage(inferredWritesByPath, 'w') overlayLineage(inferredReadsByPath, 'r') - - // Drop persisted ASSET triggers for drafted paths — those come from the - // deployed `// on ` annotations, which the draft's live/parsed - // annotations now own. Native triggers (kafka/schedule/…) are kept: they - // bind by `script_path`, which the draft shares, so the attachment is still - // valid regardless of content edits. - const baseTriggers = base.triggers.filter((t) => { - if (t.trigger_kind !== 'asset') return true - if (isDrafted(t.runnable_kind, t.runnable_path)) return false - if ( - t.runnable_kind === 'script' && - t.runnable_path === openPath && - staleForOpen(t.asset_kind, t.asset_path) - ) - return false - return true - }) - return { ...base, assets, runnables, edges, triggers: [...baseTriggers, ...extraTriggers] } } diff --git a/frontend/src/lib/components/runs/DispatchEventsButton.svelte b/frontend/src/lib/components/runs/DispatchEventsButton.svelte index 19455e18de..8edae8efc6 100644 --- a/frontend/src/lib/components/runs/DispatchEventsButton.svelte +++ b/frontend/src/lib/components/runs/DispatchEventsButton.svelte @@ -1,8 +1,7 @@ {#if list.length > 0} diff --git a/frontend/src/lib/components/runs/DispatchEventsPanel.svelte b/frontend/src/lib/components/runs/DispatchEventsPanel.svelte index 545ff35488..5f9ee2d582 100644 --- a/frontend/src/lib/components/runs/DispatchEventsPanel.svelte +++ b/frontend/src/lib/components/runs/DispatchEventsPanel.svelte @@ -1,18 +1,16 @@ {#if list.length > 0} diff --git a/frontend/src/lib/components/runs/DispatchEventsTable.svelte b/frontend/src/lib/components/runs/DispatchEventsTable.svelte index 1ac85b990e..b095c9bda7 100644 --- a/frontend/src/lib/components/runs/DispatchEventsTable.svelte +++ b/frontend/src/lib/components/runs/DispatchEventsTable.svelte @@ -7,6 +7,9 @@ type Props = { events: Event[]; workspace: string } let { events, workspace }: Props = $props() + // These reason strings are produced verbatim by the backend asset-dispatch + // reason enum (backend asset_dispatch.rs) — keep them in sync if that enum + // changes; an unmatched reason falls through to the raw string. function reasonLabel(reason: string | undefined): string { switch (reason) { case 'self_loop': diff --git a/frontend/src/lib/components/runs/useDispatchEvents.svelte.ts b/frontend/src/lib/components/runs/useDispatchEvents.svelte.ts new file mode 100644 index 0000000000..f9a148f7f4 --- /dev/null +++ b/frontend/src/lib/components/runs/useDispatchEvents.svelte.ts @@ -0,0 +1,24 @@ +import { resource } from 'runed' +import { JobService, type ListDispatchEventsResponse } from '$lib/gen' + +export type DispatchEvent = ListDispatchEventsResponse[number] + +// Shared loader for "the jobs this run dispatched". Both the inline button and +// the run-detail panel render the same list, so keep the fetch in one place. +// Returns a getter for the (never-undefined) event list — empty while loading, +// when there's nothing to load, or before workspace/jobId are known. +export function useDispatchEvents( + workspace: () => string, + jobId: () => string +): { readonly list: DispatchEvent[] } { + const events = resource( + () => ({ workspace: workspace(), jobId: jobId() }), + async ({ workspace, jobId }) => + workspace && jobId ? await JobService.listDispatchEvents({ workspace, id: jobId }) : [] + ) + return { + get list() { + return events.current ?? [] + } + } +} diff --git a/frontend/src/lib/components/triggers/TriggerRunnablePicker.svelte b/frontend/src/lib/components/triggers/TriggerRunnablePicker.svelte new file mode 100644 index 0000000000..208645182f --- /dev/null +++ b/frontend/src/lib/components/triggers/TriggerRunnablePicker.svelte @@ -0,0 +1,59 @@ + + +{#if fixedScriptPath != ''} + +{:else} +

+ {promptText} +

+
+ + {@render createButton?.()} +
+{/if} diff --git a/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte index 2f146831b3..34cf4e92c3 100644 --- a/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/email/EmailTriggerEditorInner.svelte @@ -3,9 +3,7 @@ import Drawer from '$lib/components/common/drawer/Drawer.svelte' import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte' import Path from '$lib/components/Path.svelte' - import Required from '$lib/components/Required.svelte' - import ScriptPicker from '$lib/components/ScriptPicker.svelte' - import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte' + import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte' import { EmailTriggerService, type ErrorHandler, @@ -240,8 +238,8 @@ return { noDeployed: !!(s as any)?.no_deployed, overlay: draftFromBackend - ? ({ ...deployedTrigger, ...draftFromBackend } as Record) - : undefined + ? ({ ...deployedTrigger, ...draftFromBackend } as Record) + : undefined } } @@ -390,38 +388,27 @@ {#if !hideTarget}
- {#if fixedScriptPath != ''} - - {:else} -

- Pick a script or flow to be triggered -

-
-
- - - {#if emptyString(script_path)} - - {/if} -
-
- {/if} + + {#snippet createButton()} + {#if emptyString(script_path)} + + {/if} + {/snippet} +
{/if} diff --git a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte index 356983b1fc..f2a22c359c 100644 --- a/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/gcp/GcpTriggerEditorInner.svelte @@ -18,9 +18,7 @@ type TriggerMode } from '$lib/gen' import Section from '$lib/components/Section.svelte' - import ScriptPicker from '$lib/components/ScriptPicker.svelte' - import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte' - import Required from '$lib/components/Required.svelte' + import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte' import GcpTriggerEditorConfigSection from './GcpTriggerEditorConfigSection.svelte' import { untrack, type Snippet } from 'svelte' import TriggerEditorToolbar from '../TriggerEditorToolbar.svelte' @@ -213,11 +211,11 @@ const { draft: draftFromBackend, ...deployedTrigger } = (s ?? {}) as any loadTriggerConfig(deployedTrigger) return { - noDeployed: !!(s as any)?.no_deployed, - overlay: draftFromBackend - ? ({ ...deployedTrigger, ...draftFromBackend } as Record) - : undefined - } + noDeployed: !!(s as any)?.no_deployed, + overlay: draftFromBackend + ? ({ ...deployedTrigger, ...draftFromBackend } as Record) + : undefined + } } catch (error) { sendUserToast(`Could not load GCP Pub/Sub trigger: ${error.body}`, true) return { overlay: undefined, noDeployed: false } @@ -477,24 +475,16 @@ {#if !hideTarget}
- {#if fixedScriptPath != ''} - - {:else} -

- Pick a script or flow to be triggered -

-
- + + {#snippet createButton()} {#if emptyString(script_path)} {/if} -
- {/if} + {/snippet} +
{/if} diff --git a/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte index b20e457cd9..afdd9d7746 100644 --- a/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte @@ -4,9 +4,7 @@ import Drawer from '$lib/components/common/drawer/Drawer.svelte' import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte' import Path from '$lib/components/Path.svelte' - import Required from '$lib/components/Required.svelte' - import ScriptPicker from '$lib/components/ScriptPicker.svelte' - import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte' + import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte' import { KafkaTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen' import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores' import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils' @@ -271,8 +269,8 @@ return { noDeployed: !!(s as any)?.no_deployed, overlay: draftFromBackend - ? ({ ...deployedTrigger, ...draftFromBackend } as Record) - : undefined + ? ({ ...deployedTrigger, ...draftFromBackend } as Record) + : undefined } } @@ -525,24 +523,15 @@ {#if !hideTarget}
- {#if fixedScriptPath != ''} - - {:else} -

- Pick a script or flow to be triggered -

-
- + + {#snippet createButton()} {#if emptyString(script_path)} {/if} -
- {/if} + {/snippet} +
{/if} diff --git a/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte index 469bc0419d..561608bd9f 100644 --- a/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/mqtt/MqttTriggerEditorInner.svelte @@ -4,9 +4,7 @@ import Drawer from '$lib/components/common/drawer/Drawer.svelte' import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte' import Path from '$lib/components/Path.svelte' - import Required from '$lib/components/Required.svelte' - import ScriptPicker from '$lib/components/ScriptPicker.svelte' - import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte' + import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte' import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores' import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils' import { withForkConflictRetry } from '$lib/utils/forkConflict' @@ -269,11 +267,11 @@ const { draft: draftFromBackend, ...deployedTrigger } = (s ?? {}) as any loadTriggerConfig(deployedTrigger) return { - noDeployed: !!(s as any)?.no_deployed, - overlay: draftFromBackend - ? ({ ...deployedTrigger, ...draftFromBackend } as Record) - : undefined - } + noDeployed: !!(s as any)?.no_deployed, + overlay: draftFromBackend + ? ({ ...deployedTrigger, ...draftFromBackend } as Record) + : undefined + } } catch (error) { sendUserToast(`Could not load mqtt trigger: ${error.body}`, true) return { overlay: undefined, noDeployed: false } @@ -500,24 +498,15 @@ {#if !hideTarget}
- {#if fixedScriptPath != ''} - - {:else} -

- Pick a script or flow to be triggered -

-
- + + {#snippet createButton()} {#if emptyString(script_path)}
- {/if} + {/snippet} +
{/if} diff --git a/frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte index d184fb65cf..4450cd6891 100644 --- a/frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/nats/NatsTriggerEditorInner.svelte @@ -3,9 +3,7 @@ import Drawer from '$lib/components/common/drawer/Drawer.svelte' import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte' import Path from '$lib/components/Path.svelte' - import Required from '$lib/components/Required.svelte' - import ScriptPicker from '$lib/components/ScriptPicker.svelte' - import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte' + import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte' import { NatsTriggerService, type ErrorHandler, type Retry, type TriggerMode } from '$lib/gen' import { usedTriggerKinds, userStore, workspaceStore } from '$lib/stores' import { canWrite, capitalize, emptyString, sendUserToast } from '$lib/utils' @@ -267,8 +265,8 @@ return { noDeployed: !!(s as any)?.no_deployed, overlay: draftFromBackend - ? ({ ...deployedTrigger, ...draftFromBackend } as Record) - : undefined + ? ({ ...deployedTrigger, ...draftFromBackend } as Record) + : undefined } } @@ -495,24 +493,15 @@ {#if !hideTarget}
- {#if fixedScriptPath != ''} - - {:else} -

- Pick a script or flow to be triggered -

-
- + + {#snippet createButton()} {#if emptyString(script_path)}
- {/if} + {/snippet} +
{/if} diff --git a/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte index fad6fe18f2..6420d5f9b1 100644 --- a/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/postgres/PostgresTriggerEditorInner.svelte @@ -4,8 +4,7 @@ import DrawerContent from '$lib/components/common/drawer/DrawerContent.svelte' import Path from '$lib/components/Path.svelte' import Required from '$lib/components/Required.svelte' - import ScriptPicker from '$lib/components/ScriptPicker.svelte' - import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte' + import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte' import { PostgresTriggerService, type ErrorHandler, @@ -672,25 +671,17 @@ {#if !hideTarget}
- {#if fixedScriptPath != ''} - - {:else} -

- Pick a script or flow to be triggered -

-
- - + + {#snippet createButton()} {#if emptyString(script_path) && is_flow === false}
{/if} -
- {/if} + {/snippet} +
{/if} diff --git a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte index a8f25c7167..6ca2d13ddd 100644 --- a/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte +++ b/frontend/src/lib/components/triggers/sqs/SqsTriggerEditorInner.svelte @@ -17,9 +17,7 @@ } from '$lib/gen' import SqsTriggerEditorConfigSection from './SqsTriggerEditorConfigSection.svelte' import Section from '$lib/components/Section.svelte' - import ScriptPicker from '$lib/components/ScriptPicker.svelte' - import PipelineLockedRunnableInfo from '$lib/components/triggers/PipelineLockedRunnableInfo.svelte' - import Required from '$lib/components/Required.svelte' + import TriggerRunnablePicker from '$lib/components/triggers/TriggerRunnablePicker.svelte' import { untrack, type Snippet } from 'svelte' import TriggerEditorToolbar from '../TriggerEditorToolbar.svelte' import PermissionedAsLine from '../PermissionedAsLine.svelte' @@ -247,11 +245,11 @@ const { draft: draftFromBackend, ...deployedTrigger } = (s ?? {}) as any loadTriggerConfig(deployedTrigger) return { - noDeployed: !!(s as any)?.no_deployed, - overlay: draftFromBackend - ? ({ ...deployedTrigger, ...draftFromBackend } as Record) - : undefined - } + noDeployed: !!(s as any)?.no_deployed, + overlay: draftFromBackend + ? ({ ...deployedTrigger, ...draftFromBackend } as Record) + : undefined + } } catch (error) { sendUserToast(`Could not load SQS trigger: ${error.body}`, true) return { overlay: undefined, noDeployed: false } @@ -479,24 +477,16 @@ {#if !hideTarget}
- {#if fixedScriptPath != ''} - - {:else} -

- Pick a script or flow to be triggered -

-
- + + {#snippet createButton()} {#if emptyString(script_path)}
- {/if} + {/snippet} +
{/if} diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index ef133de83d..61b08d52c6 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -19,7 +19,6 @@ import PipelineActivityPanel from '$lib/components/assets/AssetGraph/PipelineActivityPanel.svelte' import AssetGraphDetailsPane from '$lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte' import PipelinePickerModal from '$lib/components/assets/AssetGraph/PipelinePickerModal.svelte' - import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte' import { extractWrites, extractReads, @@ -71,17 +70,9 @@ Telescope } from 'lucide-svelte' import { - EmailTriggerService, - GcpTriggerService, JobService, - KafkaTriggerService, - MqttTriggerService, - NatsTriggerService, OpenAPI, - PostgresTriggerService, - ScheduleService, ScriptService, - SqsTriggerService, type AssetKind, type Script, type ScriptLang @@ -95,15 +86,7 @@ import Popover from '$lib/components/meltComponents/Popover.svelte' import HideButton from '$lib/components/apps/editor/settingsPanel/HideButton.svelte' import { inferArgs, inferAssets } from '$lib/infer' - import KafkaTriggerEditor from '$lib/components/triggers/kafka/KafkaTriggerEditor.svelte' - import MqttTriggerEditor from '$lib/components/triggers/mqtt/MqttTriggerEditor.svelte' - import NatsTriggerEditor from '$lib/components/triggers/nats/NatsTriggerEditor.svelte' - import PostgresTriggerEditor from '$lib/components/triggers/postgres/PostgresTriggerEditor.svelte' - import SqsTriggerEditor from '$lib/components/triggers/sqs/SqsTriggerEditor.svelte' - import GcpTriggerEditor from '$lib/components/triggers/gcp/GcpTriggerEditor.svelte' - import EmailTriggerEditor from '$lib/components/triggers/email/EmailTriggerEditor.svelte' - import ScheduleEditor from '$lib/components/triggers/schedules/ScheduleEditor.svelte' - import WebhookEditor from '$lib/components/triggers/webhook/WebhookEditor.svelte' + import PipelineTriggerEditors from '$lib/components/assets/AssetGraph/PipelineTriggerEditors.svelte' // Variables and resources are declarative config, not pipeline assets — // they're hub-shaped (referenced by most runnables) and would swamp the @@ -128,10 +111,7 @@ selection = undefined activeDraftPath = undefined panelHidden = false - liveAnnotations = { - scriptPath: undefined, - annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] } - } + liveAnnotations = EMPTY_LIVE_ANNOTATIONS } const url = new URL(page.url) if (m === 'view') url.searchParams.delete('mode') @@ -242,10 +222,7 @@ panelHidden = false // Same reset as the pane's close button — clears the live // annotation overlay of whichever script was open. - liveAnnotations = { - scriptPath: undefined, - annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] } - } + liveAnnotations = EMPTY_LIVE_ANNOTATIONS } } @@ -503,6 +480,24 @@ content: '' }) + // Canonical "empty" overlay literals, reused both as reset values for the + // live-* state above and as the no-overlay inputs to the deployed graph. + const EMPTY_LIVE_ASSETS = { scriptPath: undefined, assets: [] } + const EMPTY_LIVE_ANNOTATIONS = { + scriptPath: undefined, + annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] } + } + + // Reset every live editor overlay (annotations / body assets / content) + // back to empty, unconditionally. Used by the leave-edit path so a stale + // buffer for the previously-open script can't leak into the view graphs. + // (forgetPath resets these per-path instead — see there.) + function clearLiveOverlays() { + liveAnnotations = EMPTY_LIVE_ANNOTATIONS + liveBodyAssets = EMPTY_LIVE_ASSETS + liveContent = { scriptPath: undefined, content: '' } + } + // Only-add cache of (script_path → body content) populated lazily by // `bodyFetchEffect` for every script in the current folder. We never // remove entries: stale keys (renamed-away, deleted) are simply ignored @@ -524,10 +519,15 @@ // Iteration is gated on `graphRes.current.runnables`, so a path that // gets renamed / deleted disappears from the derived map as soon as // the refetch lands — no manual rekey, no phantom edges. - let inferredWritesByPath = $derived.by(() => { - const out = new Map>() + // Single pass over the graph's scripts producing both the write- and + // read-asset maps (they only differ by extractWrites vs extractReads over + // the same `liveBodyAssets`-vs-cache asset source). Split into two derives + // below so consumers can depend on one without invalidating on the other. + let inferredAssetEdges = $derived.by(() => { + const writes = new Map>() + const reads = new Map>() const g = graphRes.current - if (!g) return out + if (!g) return { writes, reads } const liveAssetsForPath = (path: string) => liveBodyAssets.scriptPath === path ? liveBodyAssets.assets : inferredAssetsByPath.get(path) for (const r of g.runnables) { @@ -535,25 +535,14 @@ const assets = liveAssetsForPath(r.path) if (!assets) continue const w = extractWrites(assets) - if (w.length > 0) out.set(r.path, w) + if (w.length > 0) writes.set(r.path, w) + const rd = extractReads(assets) + if (rd.length > 0) reads.set(r.path, rd) } - return out - }) - let inferredReadsByPath = $derived.by(() => { - const out = new Map>() - const g = graphRes.current - if (!g) return out - const liveAssetsForPath = (path: string) => - liveBodyAssets.scriptPath === path ? liveBodyAssets.assets : inferredAssetsByPath.get(path) - for (const r of g.runnables) { - if (r.usage_kind !== 'script') continue - const assets = liveAssetsForPath(r.path) - if (!assets) continue - const reads = extractReads(assets) - if (reads.length > 0) out.set(r.path, reads) - } - return out + return { writes, reads } }) + let inferredWritesByPath = $derived(inferredAssetEdges.writes) + let inferredReadsByPath = $derived(inferredAssetEdges.reads) // Same derived shape for `// on kafka` etc. annotations. Live buffer // wins for the open script; everyone else is parsed from the // prefetched body content. @@ -947,13 +936,10 @@ selection = undefined } if (liveAnnotations.scriptPath === path) { - liveAnnotations = { - scriptPath: undefined, - annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] } - } + liveAnnotations = EMPTY_LIVE_ANNOTATIONS } if (liveBodyAssets.scriptPath === path) { - liveBodyAssets = { scriptPath: undefined, assets: [] } + liveBodyAssets = EMPTY_LIVE_ASSETS } if (liveContent.scriptPath === path) { liveContent = { scriptPath: undefined, content: '' } @@ -1135,6 +1121,16 @@ // Currently-open draft shape (if any) — fed into the details pane. let activeDraft = $derived(activeDraftPath ? drafts.get(activeDraftPath) : undefined) + // Path of the script currently open in the details pane (draft or + // persisted selection), used wherever run-routing / overlay logic needs + // "the one script the user is editing right now". + let openScriptPath = $derived( + activeDraftPath ?? + (selection?.kind === 'runnable' && selection.runnable_kind === 'script' + ? selection.path + : undefined) + ) + // Entering edit mode with a selection whose path has a draft (e.g. a // deployed script whose unsaved edits were promoted on the switch to // view) re-opens the draft, not the stale deployed version — same @@ -1170,7 +1166,7 @@ // `handleDraftContentChange`, that creates a parent ↔ child feedback // loop ("effect_update_depth_exceeded"). Named functions keep the // prop reference stable so the $effects only re-fire on real - // content changes. + // content changes (e.g. handleContentChange below mutating drafts). function handleAnnotationsChange( scriptPath: string | undefined, annotations: PipelineAnnotations @@ -1365,12 +1361,7 @@ // pane, route through ScriptEditor's Test path — the test panel then // shows logs/result and the user can cancel from there. Same UX as // hitting the Test button directly. - const openPath = - activeDraftPath ?? - (selection?.kind === 'runnable' && selection.runnable_kind === 'script' - ? selection.path - : undefined) - if (openPath === producer.path) { + if (openScriptPath === producer.path) { if (cascade) requestRunCascadeSignal++ else requestRunSignal++ return undefined @@ -1441,11 +1432,7 @@ // drives the read-only pane's run form. Derived from the displayed // graph's triggers so the drafts overlay picks up draft annotations too. let openScriptHasDataUpload = $derived.by(() => { - const path = - activeDraftPath ?? - (selection?.kind === 'runnable' && selection.runnable_kind === 'script' - ? selection.path - : undefined) + const path = openScriptPath if (!path) return false return displayGraph.triggers.some( (t) => @@ -1476,11 +1463,6 @@ // but with no drafts and no live editor buffer. resolveGraph is pure, so // feeding it empty overlays is the cheapest way to share the logic. const EMPTY_DRAFTS: Map = new Map() - const EMPTY_LIVE_ASSETS = { scriptPath: undefined, assets: [] } - const EMPTY_LIVE_ANNOTATIONS = { - scriptPath: undefined, - annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] } - } let deployedGraph = $derived.by(() => resolveGraph({ base: graphRes.current ?? EMPTY_GRAPH, @@ -1504,15 +1486,14 @@ $effect(() => { if (mode === 'edit') return untrack(() => { - if (liveAnnotations.scriptPath != undefined || liveBodyAssets.scriptPath != undefined) { - liveAnnotations = { - scriptPath: undefined, - annotations: { inPipeline: false, triggerAssets: [], nativeTriggers: [] } - } - liveBodyAssets = { scriptPath: undefined, assets: [] } - } - if (liveContent.scriptPath != undefined) { - liveContent = { scriptPath: undefined, content: '' } + // Guard before reassigning so an already-empty overlay doesn't + // needlessly invalidate the graph derives every mode toggle. + if ( + liveAnnotations.scriptPath != undefined || + liveBodyAssets.scriptPath != undefined || + liveContent.scriptPath != undefined + ) { + clearLiveOverlays() } }) }) @@ -1685,9 +1666,15 @@ } // Poll a launched cascade job to a terminal state. Modest fixed cadence — // chains are short and the folder poll is already watching the same jobs - // for the canvas animation. + // for the canvas animation. Capped so a never-terminating job can't pin + // `cascadeRunningRoot` forever and wedge every future cascade: after the + // timeout we throw, which the orchestrator surfaces as a chain failure and + // `runDraftAwareCascade`'s finally clears the running-root guard. + const CASCADE_POLL_INTERVAL_MS = 1000 + const CASCADE_JOB_TIMEOUT_MS = 30 * 60 * 1000 async function waitJobTerminal(jobId: string): Promise<'success' | 'failure'> { - while (true) { + const deadline = Date.now() + CASCADE_JOB_TIMEOUT_MS + while (Date.now() < deadline) { try { const r = await JobService.getCompletedJobResultMaybe({ workspace: $workspaceStore!, @@ -1698,8 +1685,11 @@ } catch { // transient — retry on the next tick } - await new Promise((res) => setTimeout(res, 1000)) + await new Promise((res) => setTimeout(res, CASCADE_POLL_INTERVAL_MS)) } + throw new Error( + `Timed out after ${Math.round(CASCADE_JOB_TIMEOUT_MS / 60000)}min waiting for job ${jobId} to finish` + ) } // "Run + downstream" over a chain that includes drafts: the backend // asset-trigger dispatcher only resolves deployed rows, so the page @@ -1806,11 +1796,7 @@ // in `triggers`) other than self. Flows are excluded because V1 dispatch // only fans out to scripts. let editedScriptDownstreamCount = $derived.by(() => { - const editedPath = - activeDraftPath ?? - (selection?.kind === 'runnable' && selection.runnable_kind === 'script' - ? selection.path - : undefined) + const editedPath = openScriptPath if (!editedPath) return 0 const writes = graphWithDraft.edges.filter( (e) => @@ -1841,20 +1827,10 @@ // folder…" entry in the dropdown otherwise. let pickerModalOpen = $state(false) - // Native trigger editors mounted inline so clicking a "missing" - // placeholder opens the matching drawer with `script_path` pre-filled - // — keeps pipeline drafts intact instead of navigating away. Each - // editor's wrapper lazily mounts its Inner only when `open=true`, so - // holding refs to all seven is cheap. - let kafkaEditor: KafkaTriggerEditor | undefined = $state() - let mqttEditor: MqttTriggerEditor | undefined = $state() - let natsEditor: NatsTriggerEditor | undefined = $state() - let postgresEditor: PostgresTriggerEditor | undefined = $state() - let sqsEditor: SqsTriggerEditor | undefined = $state() - let gcpEditor: GcpTriggerEditor | undefined = $state() - let emailEditor: EmailTriggerEditor | undefined = $state() - let scheduleEditor: ScheduleEditor | undefined = $state() - let webhookEditor: WebhookEditor | undefined = $state() + // Native trigger editor drawers live in ; the page + // drives them imperatively. The draft guards stay here because they depend + // on the drafts map. + let triggerEditors: PipelineTriggerEditors | undefined = $state() // Webhooks have no trigger row to create — clicking the node opens a // drawer with the endpoint URLs + the webhook-specific token creation @@ -1868,7 +1844,7 @@ ) return } - webhookEditor?.openDrawer(scriptPath, false) + triggerEditors?.openWebhook(scriptPath) } // Data upload is a UI-first entry point — no trigger row. Clicking the @@ -1910,112 +1886,19 @@ ) return } - switch (kind) { - case 'schedule': - return scheduleEditor?.openNew(false, scriptPath, undefined, scriptPath) - case 'kafka': - return kafkaEditor?.openNew(false, scriptPath) - case 'mqtt': - return mqttEditor?.openNew(false, scriptPath) - case 'nats': - return natsEditor?.openNew(false, scriptPath) - case 'postgres': - return postgresEditor?.openNew(false, scriptPath) - case 'sqs': - return sqsEditor?.openNew(false, scriptPath) - case 'gcp': - return gcpEditor?.openNew(false, scriptPath) - case 'email': - return emailEditor?.openNew(false, scriptPath) - // webhook has no dedicated editor. - default: - return - } + triggerEditors?.openNew(kind, scriptPath) } // Lock the script-picker to the related script so the user can't // reassign the trigger off this pipeline from the canvas. The trigger // can still be edited everywhere else (TriggersPanel, etc.) where the // picker stays editable. - // Delete-trigger confirmation state. Kebab → Delete opens the standard - // ConfirmationModal; the actual delete is dispatched from onConfirmed - // so the dialog stays consistent with the rest of the app. - let triggerDeleteTarget = $state<{ kind: NativeTriggerKind; path: string } | undefined>(undefined) - let triggerDeleteLoading = $state(false) - let triggerDeleteOpen = $derived(triggerDeleteTarget != undefined) - function deleteAttachedTrigger(kind: NativeTriggerKind, triggerPath: string) { - triggerDeleteTarget = { kind, path: triggerPath } - } - - async function confirmDeleteAttachedTrigger() { - if (!triggerDeleteTarget || !$workspaceStore) return - const { kind, path: triggerPath } = triggerDeleteTarget - const workspace = $workspaceStore - triggerDeleteLoading = true - try { - switch (kind) { - case 'schedule': - await ScheduleService.deleteSchedule({ workspace, path: triggerPath }) - break - case 'kafka': - await KafkaTriggerService.deleteKafkaTrigger({ workspace, path: triggerPath }) - break - case 'mqtt': - await MqttTriggerService.deleteMqttTrigger({ workspace, path: triggerPath }) - break - case 'nats': - await NatsTriggerService.deleteNatsTrigger({ workspace, path: triggerPath }) - break - case 'postgres': - await PostgresTriggerService.deletePostgresTrigger({ workspace, path: triggerPath }) - break - case 'sqs': - await SqsTriggerService.deleteSqsTrigger({ workspace, path: triggerPath }) - break - case 'gcp': - await GcpTriggerService.deleteGcpTrigger({ workspace, path: triggerPath }) - break - case 'email': - await EmailTriggerService.deleteEmailTrigger({ workspace, path: triggerPath }) - break - default: - return - } - sendUserToast(`Deleted ${kind} trigger "${triggerPath}"`) - triggerDeleteTarget = undefined - await graphRes.refetch() - } catch (e: any) { - sendUserToast( - `Could not delete ${kind} trigger "${triggerPath}": ${e?.body ?? e?.message ?? String(e)}`, - true - ) - } finally { - triggerDeleteLoading = false - } + triggerEditors?.requestDelete(kind, triggerPath) } function openEditTriggerDrawer(kind: NativeTriggerKind, triggerPath: string, scriptPath: string) { - switch (kind) { - case 'schedule': - return scheduleEditor?.openEdit(triggerPath, false, scriptPath) - case 'kafka': - return kafkaEditor?.openEdit(triggerPath, false, scriptPath) - case 'mqtt': - return mqttEditor?.openEdit(triggerPath, false, scriptPath) - case 'nats': - return natsEditor?.openEdit(triggerPath, false, scriptPath) - case 'postgres': - return postgresEditor?.openEdit(triggerPath, false, scriptPath) - case 'sqs': - return sqsEditor?.openEdit(triggerPath, false, scriptPath) - case 'gcp': - return gcpEditor?.openEdit(triggerPath, false, scriptPath) - case 'email': - return emailEditor?.openEdit(triggerPath, false, scriptPath) - default: - return - } + triggerEditors?.openEdit(kind, triggerPath, scriptPath) } // Reuse the empty AssetGraphResponse so we can still render the canvas @@ -2456,11 +2339,7 @@ // per-node Run button. The currently-edited script // is whichever path is open in the pane (active // draft, or the persisted-script selection). - const openPath = - activeDraftPath ?? - (selection?.kind === 'runnable' && selection.runnable_kind === 'script' - ? selection.path - : undefined) + const openPath = openScriptPath if (running && openPath) { activeRunnable = { kind: 'script', path: openPath } // Mark the tested runnable as launched-from-here so the @@ -2497,14 +2376,7 @@ // via the explicit "Discard" button in the pane. selection = undefined activeDraftPath = undefined - liveAnnotations = { - scriptPath: undefined, - annotations: { - inPipeline: false, - triggerAssets: [], - nativeTriggers: [] - } - } + liveAnnotations = EMPTY_LIVE_ANNOTATIONS }} onHide={() => (panelHidden = true)} onDiscard={() => { @@ -2583,44 +2455,14 @@ -{#if mode === 'edit'} - { - if (!triggerDeleteLoading) triggerDeleteTarget = undefined - }} - > - {#if triggerDeleteTarget} -

- Delete {triggerDeleteTarget.path}? The - // on {triggerDeleteTarget.kind} annotation on the script stays - — the trigger will read as missing on the canvas until you recreate it or remove the annotation. -

- {/if} - - - - graphRes.refetch()} /> - graphRes.refetch()} /> - graphRes.refetch()} /> - graphRes.refetch()} /> - graphRes.refetch()} /> - graphRes.refetch()} /> - graphRes.refetch()} /> - graphRes.refetch()} /> -{/if} - - + + graphRes.refetch()} +/> {#if leaveModalOpen}