From 99e81c5b509cf8f3bc2ef0e14f5eb59da84bf826 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 16 May 2026 20:24:35 +0000 Subject: [PATCH] feat: apply opt-in debounce to asset-cascade subscriber dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- ...0fb73b3155b5f8b1cdfa422725b57d9115d0.json} | 12 ++- backend/tests/asset_trigger_dispatch.rs | 79 +++++++++++++++++++ backend/windmill-queue/src/asset_dispatch.rs | 35 +++++--- 3 files changed, 113 insertions(+), 13 deletions(-) rename backend/.sqlx/{query-e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510.json => query-fdbbbe58fe7ca2a4dee3ef2da6b90fb73b3155b5f8b1cdfa422725b57d9115d0.json} (52%) diff --git a/backend/.sqlx/query-e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510.json b/backend/.sqlx/query-fdbbbe58fe7ca2a4dee3ef2da6b90fb73b3155b5f8b1cdfa422725b57d9115d0.json similarity index 52% rename from backend/.sqlx/query-e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510.json rename to backend/.sqlx/query-fdbbbe58fe7ca2a4dee3ef2da6b90fb73b3155b5f8b1cdfa422725b57d9115d0.json index ca6a1683e1..8065afc73b 100644 --- a/backend/.sqlx/query-e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510.json +++ b/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" } diff --git a/backend/tests/asset_trigger_dispatch.rs b/backend/tests/asset_trigger_dispatch.rs index dad3ad013d..c5c257a42e 100644 --- a/backend/tests/asset_trigger_dispatch.rs +++ b/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, + 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) -> 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, +) -> 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, + path: &str, + ) -> anyhow::Result<(Option, Option)> { + 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(()) +} diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 89bf2858a9..43e05fed0b 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -130,7 +130,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result }; 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 runnable_path, depth + 1, partition.as_deref(), + debounce_s, ) .await { @@ -309,14 +310,15 @@ async fn fetch_subscribers( db: &Pool, workspace_id: &str, trigger_ref: &str, -) -> Result> { +) -> Result)>> { // 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, ) -> Result { 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, };