From ba9068b2c0761ce510c07fb997dee51f69e8c464 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 17 May 2026 09:02:21 +0000 Subject: [PATCH] refactor: simplify pipeline code per review (dedup, single-parse, constant) - ParseAssetsOutput::new() collapses the 6-line annotation copy-paste across the 4 asset-parser crates to one call site. - asset_dispatch: parse the cascade trigger object once and pass it to the depth/partition readers instead of deserializing it twice; add a TRIGGER_ARG constant for the previously stringly-typed key (3 sites). - scripts deploy: drop a redundant debounce_default clone. No behavior change; 29 parser + 6 dispatch integration tests green. Co-Authored-By: Claude Opus 4.7 --- .../windmill-parser-py-asset/src/lib.rs | 15 +++---- .../src/asset_parser.rs | 15 +++---- .../windmill-parser-ts-asset/src/lib.rs | 15 +++---- .../windmill-parser-yaml/src/asset_parser.rs | 15 +++---- .../windmill-parser/src/asset_parser.rs | 23 ++++++++++ backend/windmill-api-scripts/src/scripts.rs | 2 +- backend/windmill-queue/src/asset_dispatch.rs | 42 ++++++++++--------- 7 files changed, 66 insertions(+), 61 deletions(-) diff --git a/backend/parsers/windmill-parser-py-asset/src/lib.rs b/backend/parsers/windmill-parser-py-asset/src/lib.rs index 538612fda5..e8a0a1cc40 100644 --- a/backend/parsers/windmill-parser-py-asset/src/lib.rs +++ b/backend/parsers/windmill-parser-py-asset/src/lib.rs @@ -29,16 +29,11 @@ pub fn parse_assets(input: &str) -> anyhow::Result { } let pipeline = parse_pipeline_annotations(input); - Ok(ParseAssetsOutput { - assets: merge_assets(assets_finder.assets), - in_pipeline: pipeline.in_pipeline, - triggers: pipeline.triggers, - partition: pipeline.partition, - freshness: pipeline.freshness, - join_mode: pipeline.join_mode, - debounce_default: pipeline.debounce_default, - ..Default::default() - }) + Ok(ParseAssetsOutput::new( + merge_assets(assets_finder.assets), + Vec::new(), + pipeline, + )) } type VarAssetName = String; diff --git a/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs b/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs index 038266fdc2..38990bd495 100644 --- a/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs +++ b/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs @@ -34,16 +34,11 @@ pub fn parse_assets(input: &str) -> anyhow::Result { } let pipeline = parse_pipeline_annotations(input); - Ok(ParseAssetsOutput { - assets: merge_assets(collector.assets), - in_pipeline: pipeline.in_pipeline, - triggers: pipeline.triggers, - partition: pipeline.partition, - freshness: pipeline.freshness, - join_mode: pipeline.join_mode, - debounce_default: pipeline.debounce_default, - ..Default::default() - }) + Ok(ParseAssetsOutput::new( + merge_assets(collector.assets), + Vec::new(), + pipeline, + )) } /// Visitor that collects S3 asset literals from SQL statements diff --git a/backend/parsers/windmill-parser-ts-asset/src/lib.rs b/backend/parsers/windmill-parser-ts-asset/src/lib.rs index 9a99120c47..b7fe0fea5c 100644 --- a/backend/parsers/windmill-parser-ts-asset/src/lib.rs +++ b/backend/parsers/windmill-parser-ts-asset/src/lib.rs @@ -39,16 +39,11 @@ pub fn parse_assets(code: &str) -> anyhow::Result { AssetsFinder { assets: vec![], sql_queries: vec![], var_identifiers: HashMap::new() }; assets_finder.visit_module_items(&ast); let pipeline = parse_pipeline_annotations(code); - Ok(ParseAssetsOutput { - assets: merge_assets(assets_finder.assets), - sql_queries: assets_finder.sql_queries, - in_pipeline: pipeline.in_pipeline, - triggers: pipeline.triggers, - partition: pipeline.partition, - freshness: pipeline.freshness, - join_mode: pipeline.join_mode, - debounce_default: pipeline.debounce_default, - }) + Ok(ParseAssetsOutput::new( + merge_assets(assets_finder.assets), + assets_finder.sql_queries, + pipeline, + )) } type VarAssetName = String; diff --git a/backend/parsers/windmill-parser-yaml/src/asset_parser.rs b/backend/parsers/windmill-parser-yaml/src/asset_parser.rs index c100c53fce..4dde12ad8e 100644 --- a/backend/parsers/windmill-parser-yaml/src/asset_parser.rs +++ b/backend/parsers/windmill-parser-yaml/src/asset_parser.rs @@ -41,14 +41,9 @@ pub fn parse_assets(input: &str) -> anyhow::Result { } let pipeline = parse_pipeline_annotations(input); - Ok(ParseAssetsOutput { - assets: merge_assets(assets), - in_pipeline: pipeline.in_pipeline, - triggers: pipeline.triggers, - partition: pipeline.partition, - freshness: pipeline.freshness, - join_mode: pipeline.join_mode, - debounce_default: pipeline.debounce_default, - ..Default::default() - }) + Ok(ParseAssetsOutput::new( + merge_assets(assets), + Vec::new(), + pipeline, + )) } diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 536ff31d2b..1cfbbb898b 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -225,6 +225,29 @@ pub struct PipelineAnnotations { pub debounce_default: Option, } +impl ParseAssetsOutput { + /// Build from detected assets/queries plus the script's parsed + /// pipeline annotations, so each language asset-parser does not + /// re-list the per-annotation fields (one call site instead of six + /// lines kept in lockstep across the parser crates). + pub fn new( + assets: Vec, + sql_queries: Vec, + pipeline: PipelineAnnotations, + ) -> Self { + ParseAssetsOutput { + assets, + sql_queries, + in_pipeline: pipeline.in_pipeline, + triggers: pipeline.triggers, + partition: pipeline.partition, + freshness: pipeline.freshness, + join_mode: pipeline.join_mode, + debounce_default: pipeline.debounce_default, + } + } +} + #[derive(Debug, Clone, Serialize)] pub struct DelegateToGitRepoDetails { pub resource: String, diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 2e54fbedb6..19c38cc34b 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -1244,7 +1244,7 @@ async fn create_script_internal<'c>( let pipeline_join_all = !pipeline_annotations.join_mode.is_any(); // Script-level `// debounce ` default; a per-`// on debounce=` // overrides it (precedence resolved per edge below). - let pipeline_debounce_default = pipeline_annotations.debounce_default.clone(); + let pipeline_debounce_default = pipeline_annotations.debounce_default; let pipeline_triggers = pipeline_annotations.triggers; let auto_kind = if in_pipeline { Some("pipeline".to_string()) diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 21fe0d292b..38c6998d76 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -64,6 +64,10 @@ use windmill_common::DB; /// Set by the test panel when the user opts out of the cascade. pub const SKIP_ASSET_DISPATCH_ARG: &str = "_wmill_skip_asset_dispatch"; +/// Arg key holding the cascade trigger object (carries `depth`, `partition`, +/// producer metadata) injected into every dispatched subscriber. +const TRIGGER_ARG: &str = "trigger"; + /// Reserved arg key (under `trigger.depth`) that carries cascade depth. const CHAIN_DEPTH_KEY: &str = "depth"; @@ -105,8 +109,14 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result if read_skip_arg(args.as_ref()) { return Ok(DispatchResult::default()); } - let depth = read_chain_depth(args.as_ref()); - let partition = read_partition(args.as_ref()); + // Parse the cascade `trigger` object once; both depth and the + // propagated partition are read from it. + let trigger_map = args + .as_ref() + .and_then(|a| a.get(TRIGGER_ARG)) + .and_then(|t| serde_json::from_str::>>(t.get()).ok()); + let depth = read_chain_depth(trigger_map.as_ref()); + let partition = read_partition(args.as_ref(), trigger_map.as_ref()); if depth >= MAX_CHAIN_DEPTH { tracing::warn!( "asset-trigger dispatch skipped: chain depth {} >= cap {} (job {}, path {})", @@ -237,17 +247,9 @@ fn read_skip_arg(args: Option<&HashMap>>) -> bool { .unwrap_or(false) } -fn read_chain_depth(args: Option<&HashMap>>) -> i64 { - let Some(args) = args else { - return 0; - }; - let Some(trigger) = args.get("trigger") else { - return 0; - }; - let Ok(map) = serde_json::from_str::>>(trigger.get()) else { - return 0; - }; - map.get(CHAIN_DEPTH_KEY) +fn read_chain_depth(trigger_map: Option<&HashMap>>) -> i64 { + trigger_map + .and_then(|m| m.get(CHAIN_DEPTH_KEY)) .and_then(|v| serde_json::from_str::(v.get()).ok()) .unwrap_or(0) } @@ -257,16 +259,16 @@ fn read_chain_depth(args: Option<&HashMap>>) -> i64 { /// materializes the same partition without re-resolving. Top-level /// `partition` arg (run-start injection) takes precedence over the /// `trigger.partition` carried from an upstream cascade hop. -fn read_partition(args: Option<&HashMap>>) -> Option { - let args = args?; - if let Some(v) = args.get(PARTITION_ARG) { +fn read_partition( + args: Option<&HashMap>>, + trigger_map: Option<&HashMap>>, +) -> Option { + if let Some(v) = args.and_then(|a| a.get(PARTITION_ARG)) { if let Ok(s) = serde_json::from_str::(v.get()) { return Some(s); } } - let trigger = args.get("trigger")?; - let map = serde_json::from_str::>>(trigger.get()).ok()?; - serde_json::from_str::(map.get(PARTITION_ARG)?.get()).ok() + serde_json::from_str::(trigger_map?.get(PARTITION_ARG)?.get()).ok() } fn prefix_for(kind: AssetKind) -> Option<&'static str> { @@ -517,7 +519,7 @@ async fn push_subscriber( CHAIN_DEPTH_KEY: depth, PARTITION_ARG: partition, }); - args.insert("trigger".to_string(), to_raw_value(&trigger_payload)); + args.insert(TRIGGER_ARG.to_string(), to_raw_value(&trigger_payload)); // Carry the producer's resolved partition forward as a top-level arg so // the subscriber's body can read it and the next cascade hop's // `read_partition` picks it up — keeps the whole chain on one partition,