From b83300624190b8b1941a697c25090195a820efd0 Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Fri, 6 Mar 2026 15:31:55 +0100 Subject: [PATCH] Debounced node v0 --- backend/windmill-types/src/flows.rs | 3 + backend/windmill-worker/src/worker_flow.rs | 131 ++++++++++++++++++ .../flows/content/FlowModuleComponent.svelte | 30 ++++ openflow.openapi.yaml | 20 +++ 4 files changed, 184 insertions(+) diff --git a/backend/windmill-types/src/flows.rs b/backend/windmill-types/src/flows.rs index 1449050c24..699e71b105 100644 --- a/backend/windmill-types/src/flows.rs +++ b/backend/windmill-types/src/flows.rs @@ -445,6 +445,8 @@ pub struct FlowModule { pub apply_preprocessor: Option, #[serde(skip_serializing_if = "Option::is_none")] pub pass_flow_input_directly: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub debouncing: Option, } #[derive(Deserialize, Serialize, Debug, Clone)] @@ -1117,6 +1119,7 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec) { skip_if: None, apply_preprocessor: None, pass_flow_input_directly: None, + debouncing: None, }); } } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 33dafbb482..fa83182892 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/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 = None; + /* Finally, push the job into the queue */ let mut uuids = vec![]; + #[allow(unused_mut)] + let mut debounced_prev_parent_flow: Option = 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 = 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 = + 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); diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index 4a43c2c38d..948f6a752f 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/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" /> + {#if !parentModule?.value?.type || (parentModule.value.type !== 'forloopflow' && parentModule.value.type !== 'whileloopflow')} + + {/if} + {:else if advancedSelected === 'debouncing'} +
+ {#if flowModule.debouncing} + + {/if} +
{:else if advancedSelected === 'mock'}
diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 491b9434bf..99f38bb38e 100644 --- a/openflow.openapi.yaml +++ b/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: