feat: opt-in // debounce for asset-cascade subscribers (parser + schema)

Stage E1+E2. Parser: script-level // debounce <dur> + per-// on
debounce=<dur> override (edge wins, else script default, else none =
fan-out, unchanged); TriggerSpec::Asset carries the per-edge override;
split_trailing_kv_opts separates the ref from trailing key=val opts.
Schema/deploy: reversible migration adds script_trigger.debounce_s;
parse_duration_secs (bare int or <n>s|m|h|d, fail-safe on garbage)
resolves the effective per-edge window at deploy and writes it per row.
No reader yet (dispatch wiring is E3) so runtime is unchanged. New unit
tests for the parser directive and duration parsing.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

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"
 }

backend/migrations/20260516201317_script_trigger_debounce_s.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE script_trigger
+  DROP COLUMN debounce_s;

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=<dur>` ?? script-level
+-- `// debounce <dur>` ?? 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;

backend/parsers/windmill-parser-py-asset/src/lib.rs

@@ -36,6 +36,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         partition: pipeline.partition,
         freshness: pipeline.freshness,
         join_mode: pipeline.join_mode,
+        debounce_default: pipeline.debounce_default,
         ..Default::default()
     })
 }

backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs

@@ -41,6 +41,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         partition: pipeline.partition,
         freshness: pipeline.freshness,
         join_mode: pipeline.join_mode,
+        debounce_default: pipeline.debounce_default,
         ..Default::default()
     })
 }

backend/parsers/windmill-parser-ts-asset/src/lib.rs

@@ -47,6 +47,7 @@ pub fn parse_assets(code: &str) -> anyhow::Result<ParseAssetsOutput> {
         partition: pipeline.partition,
         freshness: pipeline.freshness,
         join_mode: pipeline.join_mode,
+        debounce_default: pipeline.debounce_default,
     })
 }
 

backend/parsers/windmill-parser-yaml/src/asset_parser.rs

@@ -48,6 +48,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         partition: pipeline.partition,
         freshness: pipeline.freshness,
         join_mode: pipeline.join_mode,
+        debounce_default: pipeline.debounce_default,
         ..Default::default()
     })
 }

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 <dur>` — 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<String>,
 }
 
 #[derive(Serialize, Debug, PartialEq, Clone)]
 #[serde(tag = "kind", rename_all = "lowercase")]
 pub enum TriggerSpec {
     // Refresh when `<asset>` 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=<dur>` 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<String>,
+    },
     // 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 <kind> <path>` 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<PartitionSpec>,
     pub freshness: Option<FreshnessSpec>,
     pub join_mode: JoinMode,
+    pub debounce_default: Option<String>,
 }
 
 #[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 `<ident>=…` (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<String, String>) {
+    let mut split_at: Option<usize> = 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<String, String> {
     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<TriggerSpec> {
     }
 
     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() {
+        // `<kind> <path>` 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\"";

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 <dur>` 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?;
     }

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<i32>,
 ) -> 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 `<n>` 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<i32> {
+    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:",