feat: apply opt-in debounce to asset-cascade subscriber dispatch

Stage E3. fetch_subscribers now also returns debounce_s; push_subscriber
builds real DebouncingSettings (delay + a (subscriber, partition) key,
so distinct partitions never collapse and latest-in-window falls out)
instead of ::default() when the edge opted in. Default stays no-debounce
(fan-out — the prior deliberate behaviour, now overridable rather than
reversed). Wiring test asserts the dispatched job carries the configured
window/key and an undebounced edge carries none.

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

backend/.sqlx/query-fdbbbe58fe7ca2a4dee3ef2da6b90fb73b3155b5f8b1cdfa422725b57d9115d0.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n        SELECT runnable_path AS \"runnable_path!\", join_all AS \"join_all!\"\n        FROM script_trigger\n        WHERE workspace_id = $1\n          AND trigger_kind = 'asset'\n          AND trigger_ref = $2\n          AND runnable_kind = 'script'\n        ",
+  "query": "\n        SELECT runnable_path AS \"runnable_path!\", join_all AS \"join_all!\", debounce_s\n        FROM script_trigger\n        WHERE workspace_id = $1\n          AND trigger_kind = 'asset'\n          AND trigger_ref = $2\n          AND runnable_kind = 'script'\n        ",
   "describe": {
     "columns": [
       {
@@ -12,6 +12,11 @@
         "ordinal": 1,
         "name": "join_all!",
         "type_info": "Bool"
+      },
+      {
+        "ordinal": 2,
+        "name": "debounce_s",
+        "type_info": "Int4"
       }
     ],
     "parameters": {
@@ -22,8 +27,9 @@
     },
     "nullable": [
       false,
-      false
+      false,
+      true
     ]
   },
-  "hash": "e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510"
+  "hash": "fdbbbe58fe7ca2a4dee3ef2da6b90fb73b3155b5f8b1cdfa422725b57d9115d0"
 }

backend/tests/asset_trigger_dispatch.rs

@@ -12,6 +12,7 @@ use serde_json::json;
 use sqlx::{Pool, Postgres};
 use uuid::Uuid;
 use windmill_common::jobs::{JobKind, JobPayload};
+use windmill_common::runnable_settings::prefetch_cached_from_handle;
 use windmill_common::scripts::{ScriptHash, ScriptLang};
 use windmill_queue::asset_dispatch::dispatch_asset_triggers;
 use windmill_queue::MiniCompletedJob;
@@ -156,6 +157,27 @@ async fn seed_subscription_and(
     Ok(())
 }
 
+/// Seed an asset subscription with an opt-in debounce window (seconds).
+async fn seed_subscription_debounced(
+    db: &Pool<Postgres>,
+    subscriber_path: &str,
+    trigger_ref: &str,
+    debounce_s: i32,
+) -> anyhow::Result<()> {
+    sqlx::query(
+        r#"INSERT INTO script_trigger
+             (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref, debounce_s)
+           VALUES ($1, 'script'::asset_usage_kind, $2, 'asset'::script_trigger_kind, $3, $4)"#,
+    )
+    .bind(WS)
+    .bind(subscriber_path)
+    .bind(trigger_ref)
+    .bind(debounce_s)
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
 fn make_mini(id: Uuid, runnable_path: &str) -> MiniCompletedJob {
     MiniCompletedJob {
         id,
@@ -497,3 +519,60 @@ async fn and_join_waits_for_all_partition_inputs(db: Pool<Postgres>) -> anyhow::
 
     Ok(())
 }
+
+/// Stage E3: a subscriber whose edge has a debounce window gets real
+/// DebouncingSettings (delay + a (subscriber, partition) key) on the
+/// dispatched job; an undebounced subscriber on the same asset gets none
+/// (fan-out, unchanged). Asserts the wiring fetch→push→payload→handle;
+/// the actual window-collapse is the queue subsystem's own concern.
+#[sqlx::test(fixtures("base"))]
+async fn debounce_setting_applied_to_dispatched_subscriber(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    seed_script(&db, SUB_S3, "echo debounced", "bash").await?;
+    seed_script(&db, SUB_RES, "echo plain", "bash").await?;
+    seed_asset_write(&db, PRODUCER, "s3object", "f/blob").await?;
+    seed_subscription_debounced(&db, SUB_S3, "s3://f/blob", 30).await?;
+    seed_subscription(&db, SUB_RES, "script", "s3://f/blob").await?;
+
+    let id = seed_producer_job(&db, json!({})).await?;
+    let r = dispatch_asset_triggers(&db, &make_mini(id, PRODUCER)).await;
+    assert_eq!(r.dispatched.len(), 2, "both subscribers dispatched");
+
+    async fn debounce_of(
+        db: &Pool<Postgres>,
+        path: &str,
+    ) -> anyhow::Result<(Option<i32>, Option<String>)> {
+        let handle = sqlx::query_scalar!(
+            r#"SELECT q.runnable_settings_handle
+                 FROM v2_job j JOIN v2_job_queue q ON q.id = j.id
+                 WHERE j.workspace_id = $1 AND j.runnable_path = $2
+                   AND j.trigger_kind = 'asset'"#,
+            WS,
+            path,
+        )
+        .fetch_one(db)
+        .await?;
+        let (deb, _conc) = prefetch_cached_from_handle(handle, db).await?;
+        Ok((deb.debounce_delay_s, deb.debounce_key))
+    }
+
+    let (deb_delay, deb_key) = debounce_of(&db, SUB_S3).await?;
+    assert_eq!(deb_delay, Some(30), "debounced edge → 30s window");
+    assert!(
+        deb_key
+            .as_deref()
+            .is_some_and(|k| k.starts_with("asset-cascade:")),
+        "debounce key is scoped to the (subscriber, partition) cascade slot, got {deb_key:?}"
+    );
+
+    let (plain_delay, _) = debounce_of(&db, SUB_RES).await?;
+    assert_eq!(
+        plain_delay, None,
+        "undebounced edge → no debounce (fan-out)"
+    );
+
+    Ok(())
+}

backend/windmill-queue/src/asset_dispatch.rs

@@ -130,7 +130,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult>
         };
         let trigger_ref = format!("{}{}", prefix, asset_path);
         let subs = fetch_subscribers(db, &job.workspace_id, &trigger_ref).await?;
-        for (sub_path, join_all) in subs {
+        for (sub_path, join_all, debounce_s) in subs {
             if sub_path == runnable_path {
                 continue;
             }
@@ -177,6 +177,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult>
                 runnable_path,
                 depth + 1,
                 partition.as_deref(),
+                debounce_s,
             )
             .await
             {
@@ -309,14 +310,15 @@ async fn fetch_subscribers(
     db: &Pool<Postgres>,
     workspace_id: &str,
     trigger_ref: &str,
-) -> Result<Vec<(String, bool)>> {
+) -> Result<Vec<(String, bool, Option<i32>)>> {
     // V1: script subscribers only. Flow subscribers (`runnable_kind = 'flow'`)
     // are intentionally excluded — wiring them is straightforward but the
     // payload shape and permissioning need their own pass.
-    // `join_all` is the subscriber's `// trigger all` (AND join) flag.
+    // `join_all` = `// trigger all` (AND join); `debounce_s` = the opt-in
+    // debounce window resolved at deploy (NULL = fan-out, the default).
     let rows = sqlx::query!(
         r#"
-        SELECT runnable_path AS "runnable_path!", join_all AS "join_all!"
+        SELECT runnable_path AS "runnable_path!", join_all AS "join_all!", debounce_s
         FROM script_trigger
         WHERE workspace_id = $1
           AND trigger_kind = 'asset'
@@ -338,7 +340,7 @@ async fn fetch_subscribers(
     }
     Ok(rows
         .into_iter()
-        .map(|r| (r.runnable_path, r.join_all))
+        .map(|r| (r.runnable_path, r.join_all, r.debounce_s))
         .collect())
 }
 
@@ -435,6 +437,7 @@ async fn push_subscriber(
     producer_path: &str,
     depth: i64,
     partition: Option<&str>,
+    debounce_s: Option<i32>,
 ) -> Result<Uuid> {
     let (
         hash,
@@ -465,11 +468,23 @@ async fn push_subscriber(
         language,
         priority,
         apply_preprocessor: false,
-        // V1: skip debouncing/concurrency for asset-triggered runs. The
-        // trigger fan-out is the user's intent — we don't want a noisy
-        // upstream's writes to silently drop downstream runs because a
-        // debounce key collides. Revisit if we see lots of dups.
-        debouncing_settings: DebouncingSettings::default(),
+        // Debounce is opt-in per subscriber edge (`// debounce` /
+        // `// on … debounce=`). Default = none (fan-out — the user's
+        // intent unless they ask otherwise). When set, the window is
+        // keyed by (subscriber, partition) so distinct partitions never
+        // collapse and "latest within the window" falls out for free.
+        debouncing_settings: match debounce_s {
+            Some(s) if s > 0 => DebouncingSettings {
+                debounce_key: Some(format!(
+                    "asset-cascade:{}:{}",
+                    subscriber_path,
+                    partition.unwrap_or("")
+                )),
+                debounce_delay_s: Some(s),
+                ..DebouncingSettings::default()
+            },
+            _ => DebouncingSettings::default(),
+        },
         concurrency_settings: ConcurrencySettings::default(),
         labels,
     };