Debounced node v0

backend/windmill-types/src/flows.rs

@@ -445,6 +445,8 @@ pub struct FlowModule {
     pub apply_preprocessor: Option<bool>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub pass_flow_input_directly: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub debouncing: Option<DebouncingSettings>,
 }
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
@@ -1117,6 +1119,7 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
             skip_if: None,
             apply_preprocessor: None,
             pass_flow_input_directly: None,
+            debouncing: None,
         });
     }
 }

backend/windmill-worker/src/worker_flow.rs

@@ -3342,8 +3342,43 @@ async fn push_next_flow_job(
     let continue_on_same_worker =
         (flow.same_worker || job_same_worker) && module.suspend.is_none() && module.sleep.is_none();
 
+    // Node-level debouncing: compute debounce key and adjust scheduled_for if configured
+    #[cfg(feature = "enterprise")]
+    let node_debounce_key = if let Some(ref debouncing) = module.debouncing {
+        if matches!(step, Step::Step { .. })
+            && debouncing.debounce_delay_s.filter(|x| *x > 0).is_some()
+        {
+            let delay = debouncing.debounce_delay_s.unwrap();
+            scheduled_for_o = scheduled_for_o.or(Some(
+                chrono::Utc::now() + chrono::Duration::seconds(delay as i64),
+            ));
+            let key = debouncing.debounce_key.clone().unwrap_or_else(|| {
+                format!(
+                    "{}/{}/{}",
+                    &flow_job.workspace_id,
+                    flow_job.runnable_path(),
+                    &module.id
+                )
+            });
+            let key = if key.len() <= 255 {
+                key
+            } else {
+                windmill_common::utils::calculate_hash(&key)
+            };
+            Some(key)
+        } else {
+            None
+        }
+    } else {
+        None
+    };
+    #[cfg(not(feature = "enterprise"))]
+    let _node_debounce_key: Option<String> = None;
+
     /* Finally, push the job into the queue */
     let mut uuids = vec![];
+    #[allow(unused_mut)]
+    let mut debounced_prev_parent_flow: Option<Uuid> = None;
 
     let job_payloads = match job_payloads {
         ContinuePayload::SingleJob(payload) => vec![payload],
@@ -3644,6 +3679,69 @@ async fn push_next_flow_job(
 
         tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushed next flow job: {uuid}");
 
+        // Node-level debouncing: register step job in debounce_key table.
+        // If a previous step job exists for the same key, cancel it (skip it)
+        // and collect the parent flow ID so we can notify it after commit.
+        #[cfg(feature = "enterprise")]
+        if let Some(ref debounce_key) = node_debounce_key {
+            if i == 0 {
+                let prev_step_job_id: Option<Uuid> = sqlx::query_scalar(
+                    "INSERT INTO debounce_key (job_id, key)
+                     VALUES ($1, $2)
+                     ON CONFLICT (key) DO UPDATE SET
+                         previous_job_id = debounce_key.job_id,
+                         job_id = EXCLUDED.job_id,
+                         debounced_times = debounce_key.debounced_times + 1
+                     RETURNING previous_job_id",
+                )
+                .bind(uuid)
+                .bind(debounce_key)
+                .fetch_one(&mut *inner_tx)
+                .await?;
+
+                if let Some(prev_step_job_id) = prev_step_job_id {
+                    tracing::info!(
+                        id = %flow_job.id,
+                        prev_step_job = %prev_step_job_id,
+                        debounce_key = %debounce_key,
+                        "Node debounce: cancelling previous step job"
+                    );
+                    // Complete the previous step job as skipped
+                    let result =
+                        serde_json::to_string(&format!("Debounced by {uuid}")).unwrap_or_default();
+                    sqlx::query(
+                        "WITH completed AS (
+                            INSERT INTO v2_job_completed
+                                (workspace_id, id, started_at, duration_ms, result, status, worker)
+                            SELECT
+                                q.workspace_id, q.id, q.started_at,
+                                (EXTRACT('epoch' FROM now()) - EXTRACT('epoch' FROM COALESCE(q.started_at, now()))) * 1000,
+                                $2::text::jsonb,
+                                'skipped'::job_status,
+                                q.worker
+                            FROM v2_job_queue q
+                            WHERE q.id = $1
+                            ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = EXCLUDED.result
+                        )
+                        DELETE FROM v2_job_queue WHERE id = $1",
+                    )
+                    .bind(prev_step_job_id)
+                    .bind(&result)
+                    .execute(&mut *inner_tx)
+                    .await?;
+
+                    // Get the parent flow of the cancelled step so we can notify it
+                    let parent_flow_id: Option<Uuid> =
+                        sqlx::query_scalar("SELECT parent_job FROM v2_job WHERE id = $1")
+                            .bind(prev_step_job_id)
+                            .fetch_optional(&mut *inner_tx)
+                            .await?;
+
+                    debounced_prev_parent_flow = parent_flow_id;
+                }
+            }
+        }
+
         if value_with_parallel.type_ == "forloopflow"
             && value_with_parallel.parallel.unwrap_or(false)
         {
@@ -3933,6 +4031,39 @@ async fn push_next_flow_job(
     tx.commit().warn_after_seconds(3).await?;
     tracing::info!(id = %flow_job.id, root_id = %job_root, "all next flow jobs pushed: {uuids:?}");
 
+    // After commit: notify the debounced step's parent flow so it processes the
+    // skipped step and stops early (skipping descendants while keeping sibling branches).
+    if let Some(parent_flow_id) = debounced_prev_parent_flow {
+        let debounced_result =
+            serde_json::value::to_raw_value(&serde_json::json!({"debounced": true}))
+                .unwrap_or_else(|_| RawValue::from_string("null".to_string()).unwrap());
+        tracing::info!(
+            id = %flow_job.id,
+            debounced_parent = %parent_flow_id,
+            "Node debounce: notifying debounced flow's parent to stop early"
+        );
+        job_completed_tx
+            .send(
+                SendResultPayload::UpdateFlow(UpdateFlow {
+                    flow: parent_flow_id,
+                    w_id: flow_job.workspace_id.clone(),
+                    success: true,
+                    result: debounced_result,
+                    worker_dir: worker_dir.to_string(),
+                    stop_early_override: Some(true), // skip_if_stopped = true
+                    token: client.token.clone(),
+                }),
+                false,
+            )
+            .warn_after_seconds(3)
+            .await
+            .map_err(|e| {
+                Error::internal_err(format!(
+                    "error sending debounced UpdateFlow to job completed channel: {e:#}"
+                ))
+            })?;
+    }
+
     if continue_on_same_worker || continue_with_runners {
         let flow_runners = if start_runners {
             tracing::info!(id = %flow_job.id, "starting flow runners for module {}", module.id);

frontend/src/lib/components/flows/content/FlowModuleComponent.svelte

@@ -20,6 +20,7 @@
 	import type { FlowEditorContext, FlowGraphAssetContext } from '../types'
 	import FlowModuleScript from './FlowModuleScript.svelte'
 	import FlowModuleEarlyStop from './FlowModuleEarlyStop.svelte'
+	import DebounceLimit from '../DebounceLimit.svelte'
 	import FlowModuleSuspend from './FlowModuleSuspend.svelte'
 	import FlowModuleCache from './FlowModuleCache.svelte'
 	import FlowModuleDeleteAfterUse from './FlowModuleDeleteAfterUse.svelte'
@@ -155,6 +156,12 @@
 	let selected = $state(untrack(() => preprocessorModule) ? 'test' : 'inputs')
 	let advancedSelected = $state('retries')
 	let advancedRuntimeSelected = $state('concurrency')
+
+	$effect(() => {
+		if (advancedSelected === 'debouncing' && !flowModule.debouncing) {
+			flowModule.debouncing = {}
+		}
+	})
 	let s3Kind = $state('s3_client')
 	let validCode = $state(true)
 	let width = $state(1200)
@@ -1126,6 +1133,13 @@
 														label="Suspend"
 													/>
 													<Tab value="sleep" active={Boolean(flowModule.sleep)} label="Sleep" />
+													{#if !parentModule?.value?.type || (parentModule.value.type !== 'forloopflow' && parentModule.value.type !== 'whileloopflow')}
+														<Tab
+															value="debouncing"
+															active={Boolean(flowModule.debouncing?.debounce_delay_s)}
+															label="Debouncing"
+														/>
+													{/if}
 													<Tab
 														value="mock"
 														active={Boolean(flowModule.mock?.enabled)}
@@ -1312,6 +1326,22 @@
 															bind:flowModule
 														/>
 													</div>
+												{:else if advancedSelected === 'debouncing'}
+													<div>
+														{#if flowModule.debouncing}
+															<DebounceLimit
+																size="xs"
+																fontClass="font-medium"
+																bind:debounce_delay_s={flowModule.debouncing.debounce_delay_s}
+																bind:debounce_key={flowModule.debouncing.debounce_key}
+																bind:debounce_args_to_accumulate={flowModule.debouncing.debounce_args_to_accumulate}
+																bind:max_total_debouncing_time={flowModule.debouncing.max_total_debouncing_time}
+																bind:max_total_debounces_amount={flowModule.debouncing.max_total_debounces_amount}
+																schema={flowStateStore.val[flowModule.id]?.schema as any}
+																placeholder={`$workspace/flow/$flow_path-${flowModule.id}`}
+															/>
+														{/if}
+													</div>
 												{:else if advancedSelected === 'mock'}
 													<div>
 														<FlowModuleMockTransitionMessage />

openflow.openapi.yaml

@@ -326,10 +326,30 @@ components:
         retry:
           description: Retry configuration if this step fails
           $ref: '#/components/schemas/Retry'
+        debouncing:
+          description: Debouncing configuration for this step
+          $ref: '#/components/schemas/DebouncingSettings'
       required:
         - value
         - id
 
+    DebouncingSettings:
+      type: object
+      description: Debouncing configuration for a flow module
+      properties:
+        debounce_key:
+          type: string
+        debounce_delay_s:
+          type: integer
+        max_total_debouncing_time:
+          type: integer
+        max_total_debounces_amount:
+          type: integer
+        debounce_args_to_accumulate:
+          type: array
+          items:
+            type: string
+
     InputTransform:
       description: Maps input parameters for a step. Can be a static value or a JavaScript expression that references previous results or flow inputs
       oneOf: