diff --git a/backend/.sqlx/query-28e4b886452f77af784e966649d07c736aa8a8eb2f1db500c8a4adfe39a26112.json b/backend/.sqlx/query-c6f8883cf68cc91d3992bb4a889aaa91f113513e9b13a52205fd2356ec690c13.json similarity index 83% rename from backend/.sqlx/query-28e4b886452f77af784e966649d07c736aa8a8eb2f1db500c8a4adfe39a26112.json rename to backend/.sqlx/query-c6f8883cf68cc91d3992bb4a889aaa91f113513e9b13a52205fd2356ec690c13.json index c10709ce12..25f8b750fa 100644 --- a/backend/.sqlx/query-28e4b886452f77af784e966649d07c736aa8a8eb2f1db500c8a4adfe39a26112.json +++ b/backend/.sqlx/query-c6f8883cf68cc91d3992bb4a889aaa91f113513e9b13a52205fd2356ec690c13.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO script_trigger\n (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref, join_all)\n VALUES ($1, $2, $3, $4, $5, $6)", + "query": "INSERT INTO script_trigger\n (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref, join_all,\n debounce_s)\n VALUES ($1, $2, $3, $4, $5, $6, $7)", "describe": { "columns": [], "parameters": { @@ -39,10 +39,11 @@ } }, "Text", - "Bool" + "Bool", + "Int4" ] }, "nullable": [] }, - "hash": "28e4b886452f77af784e966649d07c736aa8a8eb2f1db500c8a4adfe39a26112" + "hash": "c6f8883cf68cc91d3992bb4a889aaa91f113513e9b13a52205fd2356ec690c13" } diff --git a/backend/migrations/20260516201317_script_trigger_debounce_s.down.sql b/backend/migrations/20260516201317_script_trigger_debounce_s.down.sql new file mode 100644 index 0000000000..fb0950f26f --- /dev/null +++ b/backend/migrations/20260516201317_script_trigger_debounce_s.down.sql @@ -0,0 +1,2 @@ +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 new file mode 100644 index 0000000000..7ef44f22c6 --- /dev/null +++ b/backend/migrations/20260516201317_script_trigger_debounce_s.up.sql @@ -0,0 +1,8 @@ +-- 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/parsers/windmill-parser-py-asset/src/lib.rs b/backend/parsers/windmill-parser-py-asset/src/lib.rs index 4014f9e1c5..538612fda5 100644 --- a/backend/parsers/windmill-parser-py-asset/src/lib.rs +++ b/backend/parsers/windmill-parser-py-asset/src/lib.rs @@ -36,6 +36,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result { partition: pipeline.partition, freshness: pipeline.freshness, join_mode: pipeline.join_mode, + debounce_default: pipeline.debounce_default, ..Default::default() }) } 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 4ab03c344b..038266fdc2 100644 --- a/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs +++ b/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs @@ -41,6 +41,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result { partition: pipeline.partition, freshness: pipeline.freshness, join_mode: pipeline.join_mode, + debounce_default: pipeline.debounce_default, ..Default::default() }) } diff --git a/backend/parsers/windmill-parser-ts-asset/src/lib.rs b/backend/parsers/windmill-parser-ts-asset/src/lib.rs index 639c8cf147..9a99120c47 100644 --- a/backend/parsers/windmill-parser-ts-asset/src/lib.rs +++ b/backend/parsers/windmill-parser-ts-asset/src/lib.rs @@ -47,6 +47,7 @@ pub fn parse_assets(code: &str) -> anyhow::Result { partition: pipeline.partition, freshness: pipeline.freshness, join_mode: pipeline.join_mode, + debounce_default: pipeline.debounce_default, }) } diff --git a/backend/parsers/windmill-parser-yaml/src/asset_parser.rs b/backend/parsers/windmill-parser-yaml/src/asset_parser.rs index f7b757d184..c100c53fce 100644 --- a/backend/parsers/windmill-parser-yaml/src/asset_parser.rs +++ b/backend/parsers/windmill-parser-yaml/src/asset_parser.rs @@ -48,6 +48,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result { partition: pipeline.partition, freshness: pipeline.freshness, join_mode: pipeline.join_mode, + debounce_default: pipeline.debounce_default, ..Default::default() }) } diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index f63ebfa5cb..536ff31d2b 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -88,30 +88,62 @@ pub struct ParseAssetsOutput { // subscriber's trigger rows. #[serde(skip_serializing_if = "JoinMode::is_any", default)] pub join_mode: JoinMode, + // `// debounce ` — script-level default debounce window for this + // script's asset inputs. A per-`// on … debounce=` overrides it. Raw + // duration string, parsed to seconds at deploy (parser-light, like + // freshness). Absent = no debounce (fan-out, current behaviour). + #[serde(skip_serializing_if = "Option::is_none", default)] + pub debounce_default: Option, } #[derive(Serialize, Debug, PartialEq, Clone)] #[serde(tag = "kind", rename_all = "lowercase")] pub enum TriggerSpec { // Refresh when `` changes. Kind comes from parse_asset_syntax so - // it matches the `asset` table. - Asset { asset_kind: AssetKind, path: String }, + // it matches the `asset` table. `debounce` is the optional per-input + // `// on … debounce=` override (raw duration string); it takes + // precedence over the script-level `// debounce` default, resolved at + // deploy. + Asset { + asset_kind: AssetKind, + path: String, + #[serde(skip_serializing_if = "Option::is_none", default)] + debounce: Option, + }, // Refresh on cron. The raw expression is passed through as-is so the // existing schedule subsystem can validate it. - Schedule { cron: String }, + Schedule { + cron: String, + }, // `// on ` style. The `path` is a workspace-relative // reference to a trigger row already configured in the corresponding // trigger table (http_trigger, email_trigger, kafka_trigger, …). Keeps // the annotation terse; auth/broker/topic details live in the trigger's // own UI. - Webhook { path: String }, - Email { path: String }, - Kafka { path: String }, - Mqtt { path: String }, - Nats { path: String }, - Postgres { path: String }, - Sqs { path: String }, - Gcp { path: String }, + Webhook { + path: String, + }, + Email { + path: String, + }, + Kafka { + path: String, + }, + Mqtt { + path: String, + }, + Nats { + path: String, + }, + Postgres { + path: String, + }, + Sqs { + path: String, + }, + Gcp { + path: String, + }, } impl TriggerSpec { @@ -190,6 +222,7 @@ pub struct PipelineAnnotations { pub partition: Option, pub freshness: Option, pub join_mode: JoinMode, + pub debounce_default: Option, } #[derive(Debug, Clone, Serialize)] @@ -310,6 +343,34 @@ fn unquote(s: &str) -> Option<&str> { // values run until the next whitespace; quoted values consume until the // matching quote. Malformed pairs (missing `=` or empty key) are skipped // rather than aborting the whole annotation. +// Split a `// on` right-hand side into the trigger ref and any trailing +// `key=value` opts. The opts section starts at the first whitespace token +// shaped like `=…` (e.g. `debounce=60s`); everything before is the +// asset/kind ref. Asset refs aren't expected to contain a space then an +// `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(); + if let Some(eq) = tok.find('=') { + let key = &tok[..eq]; + if !key.is_empty() + && key.starts_with(|c: char| c.is_ascii_alphabetic() || c == '_') + && key.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') + { + split_at = Some(tok_start); + break; + } + } + } + match split_at { + Some(i) => (s[..i].trim_end(), parse_kv_opts(&s[i..])), + None => (s.trim_end(), BTreeMap::new()), + } +} + fn parse_kv_opts(s: &str) -> BTreeMap { let mut out = BTreeMap::new(); let mut chars = s.chars().peekable(); @@ -460,6 +521,17 @@ 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; + } + let dur = after_kw.trim(); + if !dur.is_empty() && out.debounce_default.is_none() { + out.debounce_default = Some(dur.to_string()); + } + continue; + } + if let Some(after_kw) = rest.strip_prefix("on") { if !after_kw.starts_with(|c: char| c.is_whitespace()) { continue; @@ -468,7 +540,18 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { if spec_text.is_empty() { continue; } - if let Some(trig) = parse_trigger_spec(spec_text) { + // Split off trailing `key=value` opts (e.g. `debounce=60s`) + // from the asset/kind ref. The per-input debounce override is + // only meaningful for asset inputs (cascade fan-out); other + // trigger kinds ignore it. + let (ref_part, opts) = split_trailing_kv_opts(spec_text); + if let Some(mut trig) = parse_trigger_spec(ref_part) { + if let TriggerSpec::Asset { debounce, .. } = &mut trig { + *debounce = opts + .get("debounce") + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()); + } if !out.triggers.contains(&trig) { out.triggers.push(trig); } @@ -545,7 +628,8 @@ fn parse_trigger_spec(s: &str) -> Option { } let (kind, path) = parse_asset_syntax(s, false)?; - Some(TriggerSpec::Asset { asset_kind: kind, path: path.to_string() }) + // `debounce` is attached by the caller from the `// on` line's opts. + Some(TriggerSpec::Asset { asset_kind: kind, path: path.to_string(), debounce: None }) } #[cfg(test)] @@ -690,6 +774,62 @@ mod pipeline_annotation_tests { assert!(!out.triggers[2].is_partition_bearing()); } + fn asset_debounce(t: &TriggerSpec) -> Option<&str> { + match t { + TriggerSpec::Asset { debounce, .. } => debounce.as_deref(), + _ => None, + } + } + + #[test] + fn debounce_none_by_default() { + let out = parse_pipeline_annotations("// on s3://a/b"); + assert_eq!(out.debounce_default, None); + assert_eq!(asset_debounce(&out.triggers[0]), None); + } + + #[test] + fn script_level_debounce_default() { + let out = parse_pipeline_annotations("// debounce 30s\n// on s3://a/b"); + assert_eq!(out.debounce_default.as_deref(), Some("30s")); + // Per-edge override absent — precedence (edge ?? default) is + // resolved at deploy, so the Asset itself stays None here. + assert_eq!(asset_debounce(&out.triggers[0]), None); + } + + #[test] + fn on_level_debounce_override() { + let out = parse_pipeline_annotations( + "// debounce 30s\n\ + // on s3://lake/{partition}/raw.parquet debounce=60s\n\ + // on $res:f/cfg", + ); + assert_eq!(out.debounce_default.as_deref(), Some("30s")); + assert_eq!(out.triggers.len(), 2); + assert_eq!(asset_debounce(&out.triggers[0]), Some("60s")); + // ref still parses correctly with the trailing opt stripped. + assert!(out.triggers[0].is_partition_bearing()); + assert_eq!(asset_debounce(&out.triggers[1]), None); + } + + #[test] + fn debounce_keyword_strictness_and_first_wins() { + // `debounced` (no space) must not match; empty ignored; first wins. + let out = parse_pipeline_annotations( + "// debounced nope\n// debounce\n// debounce 1m\n// debounce 5m", + ); + assert_eq!(out.debounce_default.as_deref(), Some("1m")); + } + + #[test] + fn on_kv_split_preserves_non_asset_and_spaced_refs() { + // ` ` ref with a trailing opt still parses; the opt is + // simply not carried for non-asset triggers. + let out = parse_pipeline_annotations("// on webhook f/foo debounce=30s"); + assert_eq!(out.triggers.len(), 1); + assert!(matches!(out.triggers[0], TriggerSpec::Webhook { .. })); + } + #[test] fn partitioned_daily() { let code = "// partitioned daily tz=\"UTC\" format=\"YYYY-MM-DD\" start=\"2024-01-01\""; diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index c07ad22e7f..2e54fbedb6 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -45,8 +45,8 @@ use windmill_common::{ assets::{ clear_script_triggers, clear_static_asset_usage, clear_static_asset_usage_by_script_hash, delete_managed_pipeline_schedule, insert_script_trigger, insert_static_asset_usage, - parse_pipeline_annotations, reconcile_pipeline_schedule, trigger_spec_to_row, - AssetUsageKind, AssetWithAltAccessType, TriggerSpec, + parse_duration_secs, parse_pipeline_annotations, reconcile_pipeline_schedule, + trigger_spec_to_row, AssetUsageKind, AssetWithAltAccessType, TriggerSpec, }, error::{self, to_anyhow}, min_version::{MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2}, @@ -1242,6 +1242,9 @@ async fn create_script_internal<'c>( let in_pipeline = pipeline_annotations.in_pipeline; // `// trigger all` → AND join barrier (else OR, the default). 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_triggers = pipeline_annotations.triggers; let auto_kind = if in_pipeline { Some("pipeline".to_string()) @@ -1561,6 +1564,16 @@ async fn create_script_internal<'c>( clear_script_triggers(&mut *tx, &w_id, &ns.path, AssetUsageKind::Script).await?; for spec in &pipeline_triggers { let (trigger_kind, trigger_ref) = trigger_spec_to_row(spec); + // Effective debounce for this edge: per-`// on debounce=` wins, + // else the script-level `// debounce` default. Debounce only + // applies to asset-cascade edges; other trigger kinds get none. + let debounce_s = match spec { + TriggerSpec::Asset { debounce: Some(d), .. } => parse_duration_secs(d), + TriggerSpec::Asset { .. } => pipeline_debounce_default + .as_deref() + .and_then(parse_duration_secs), + _ => None, + }; insert_script_trigger( &mut *tx, &w_id, @@ -1569,6 +1582,7 @@ async fn create_script_internal<'c>( trigger_kind, &trigger_ref, pipeline_join_all, + debounce_s, ) .await?; } diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index d5bc0d7752..6a6a3ff617 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -207,23 +207,76 @@ pub async fn insert_script_trigger<'e>( trigger_kind: ScriptTriggerKind, trigger_ref: &str, join_all: bool, + debounce_s: Option, ) -> error::Result<()> { sqlx::query!( r#"INSERT INTO script_trigger - (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref, join_all) - VALUES ($1, $2, $3, $4, $5, $6)"#, + (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref, join_all, + debounce_s) + VALUES ($1, $2, $3, $4, $5, $6, $7)"#, workspace_id, runnable_kind as AssetUsageKind, runnable_path, trigger_kind as ScriptTriggerKind, trigger_ref, join_all, + debounce_s, ) .execute(executor) .await?; Ok(()) } +/// Parse a debounce duration into whole seconds. Accepts a bare integer +/// (seconds) or an `` with an `s`/`m`/`h`/`d` suffix (e.g. `30s`, +/// `5m`, `2h`, `1d`). Returns `None` for empty / malformed / non-positive +/// input — the caller treats `None` as "no debounce" (fan-out), so a typo +/// fails safe rather than silently debouncing. +pub fn parse_duration_secs(s: &str) -> Option { + let s = s.trim(); + if s.is_empty() { + return None; + } + let (num, mult): (&str, i64) = match s.as_bytes().last() { + Some(b's') => (&s[..s.len() - 1], 1), + Some(b'm') => (&s[..s.len() - 1], 60), + Some(b'h') => (&s[..s.len() - 1], 3600), + Some(b'd') => (&s[..s.len() - 1], 86400), + Some(c) if c.is_ascii_digit() => (s, 1), + _ => return None, + }; + let n: i64 = num.trim().parse().ok()?; + let secs = n.checked_mul(mult)?; + if secs <= 0 || secs > i32::MAX as i64 { + return None; + } + Some(secs as i32) +} + +#[cfg(test)] +mod debounce_duration_tests { + use super::parse_duration_secs; + + #[test] + fn parses_units_and_bare_seconds() { + assert_eq!(parse_duration_secs("60"), Some(60)); + assert_eq!(parse_duration_secs("30s"), Some(30)); + assert_eq!(parse_duration_secs("5m"), Some(300)); + assert_eq!(parse_duration_secs("2h"), Some(7200)); + assert_eq!(parse_duration_secs(" 1d "), Some(86400)); + } + + #[test] + fn rejects_garbage_and_nonpositive() { + assert_eq!(parse_duration_secs(""), None); + assert_eq!(parse_duration_secs("abc"), None); + assert_eq!(parse_duration_secs("0"), None); + assert_eq!(parse_duration_secs("-5"), None); + assert_eq!(parse_duration_secs("10x"), None); + assert_eq!(parse_duration_secs("s"), None); + } +} + // Inverse of trigger_spec_to_row for the Asset variant: parses a stored // trigger_ref (e.g. `s3://foo`, `$res:bar`) back into the (kind, path) pair // used as a graph node id. Returns None for refs that don't match any known @@ -238,7 +291,7 @@ pub fn parse_asset_trigger_ref(s: &str) -> Option<(AssetKind, String)> { // trigger_ref matches what downstream lookups expect. pub fn trigger_spec_to_row(spec: &TriggerSpec) -> (ScriptTriggerKind, String) { match spec { - TriggerSpec::Asset { asset_kind, path } => { + TriggerSpec::Asset { asset_kind, path, .. } => { let prefix = match asset_kind { windmill_parser::asset_parser::AssetKind::S3Object => "s3://", windmill_parser::asset_parser::AssetKind::Resource => "$res:",