From 5b1fd00aab570cc756fc0f69b592b568bba5c9cb Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 18 Jun 2026 09:51:33 +0000 Subject: [PATCH] feat: move asset-cascade join/debounce/retry to ee-private (free-CE) Join barrier, debounce, and retry become the private windmill_queue::cascade module (cascade_ee in windmill-ee-private); OSS gets cascade_oss no-op fallbacks (plain OR fan-out). Core cascade stays public. Bumps ee-repo-ref. Verified default/private/private,enterprise. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/ee-repo-ref.txt | 2 +- backend/src/monitor.rs | 2 +- backend/windmill-queue/src/asset_dispatch.rs | 239 ++----------------- backend/windmill-queue/src/cascade_oss.rs | 60 +++++ backend/windmill-queue/src/lib.rs | 7 + 5 files changed, 86 insertions(+), 224 deletions(-) create mode 100644 backend/windmill-queue/src/cascade_oss.rs diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 27af163e8d..487fc11b3b 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -1685e543e26b4cd253f2963a0dc8df5852fa8ff7 \ No newline at end of file +f72108d4cfe5673036fd32c5b3ebe33c0121932d \ No newline at end of file diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 3608efb097..be54edf468 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1275,7 +1275,7 @@ pub async fn delete_expired_items(db: &DB) -> () { tracing::error!("Error deleting autoscaling event on CE: {:?}", e); } - if let Err(e) = windmill_queue::asset_dispatch::reap_stale_join_slots(db).await { + if let Err(e) = windmill_queue::cascade::reap_stale_join_slots(db).await { tracing::error!("Error reaping stale join_pending_inputs slots: {:?}", e); } diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index ce05becee0..1238c726df 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -51,12 +51,11 @@ use sqlx::{Pool, Postgres}; use std::collections::HashMap; use std::sync::Arc; use uuid::Uuid; -use windmill_common::assets::{AssetKind, PARTITION_TOKEN}; +use windmill_common::assets::AssetKind; use windmill_common::error::{self, Result}; use windmill_common::get_latest_deployed_hash_for_path; use windmill_common::jobs::{JobKind, JobPayload, JobTriggerKind}; use windmill_common::partition::PARTITION_ARG; -use windmill_common::runnable_settings::DebouncingSettings; use windmill_common::scripts::ScriptHash; use windmill_common::triggers::TriggerMetadata; use windmill_common::users::{get_email_from_permissioned_as, username_to_permissioned_as}; @@ -148,68 +147,6 @@ impl EventRow { } } -/// AND-join slot progress at the moment a partition-bearing input -/// arrived. `fired` = all required inputs are now present (the slot has -/// been cleared and the subscriber will be dispatched). `received` / -/// `required` are the counts observed inside the slot's transaction -/// (monotonic up to that point), surfaced so the dispatch_event row can -/// show partial progress like "2/3". -#[derive(Debug, Clone, Copy)] -struct JoinSlotStatus { - fired: bool, - received: i32, - required: i32, -} - -/// Outcome of evaluating an AND-join barrier for one (subscriber, input). -/// The caller turns this into a `dispatch_event` row and decides whether to -/// push, all in one place. -enum JoinDecision { - /// Input does not advance the join (recorded as Skipped with `reason`). - Skip(&'static str), - /// Join advanced but is not yet complete (recorded as JoinPending). - Pending { received: i32, required: i32 }, - /// All required inputs are present — push the subscriber. - Fire, -} - -/// Evaluate the AND-join barrier for a partition-bearing subscriber input. -/// Only a partition-bearing input carrying a concrete partition advances the -/// join — a reference input or an unpartitioned producer must never fire a -/// partitioned join (the case-3 silent-wrong guard). On `Err` the caller -/// should log and skip without recording an event. -async fn handle_join( - db: &DB, - workspace_id: &str, - sub_path: &str, - trigger_ref: &str, - partition: Option<&str>, -) -> Result { - if !is_partition_bearing_ref(trigger_ref) { - tracing::debug!( - "AND subscriber {}: non-partition-bearing input {} does not fire the join", - sub_path, - trigger_ref - ); - return Ok(JoinDecision::Skip("case3_non_partition_bearing")); - } - let Some(pv) = partition else { - tracing::warn!( - "AND subscriber {}: partition-bearing input {} arrived with no resolved \ - partition; not dispatching (case-3 guard)", - sub_path, - trigger_ref - ); - return Ok(JoinDecision::Skip("case3_missing_partition")); - }; - match record_and_check_join_slot(db, workspace_id, sub_path, pv, trigger_ref).await? { - JoinSlotStatus { fired: false, received, required } => { - Ok(JoinDecision::Pending { received, required }) - } - JoinSlotStatus { fired: true, .. } => Ok(JoinDecision::Fire), - } -} - /// Best-effort batched insert into `dispatch_event`. Never propagates — the /// dispatch contract is "logging failures must not retroactively fail the /// producer's job." All rows accumulated over a dispatch pass go in one @@ -351,7 +288,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result continue; } if join_all { - match handle_join( + match crate::cascade::handle_join( db, &job.workspace_id, &sub_path, @@ -360,7 +297,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result ) .await { - Ok(JoinDecision::Skip(reason)) => { + Ok(crate::cascade::JoinDecision::Skip(reason)) => { events.push(EventRow::new( &sub_path, asset_kind, @@ -370,7 +307,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result )); continue; } - Ok(JoinDecision::Pending { received, required }) => { + Ok(crate::cascade::JoinDecision::Pending { received, required }) => { events.push(EventRow::new( &sub_path, asset_kind, @@ -385,7 +322,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result )); continue; // slot incomplete — wait for the rest } - Ok(JoinDecision::Fire) => {} // fall through to push + Ok(crate::cascade::JoinDecision::Fire) => {} // fall through to push Err(e) => { tracing::error!("join-slot check failed for {}: {e:#}", sub_path); continue; @@ -604,128 +541,6 @@ async fn fetch_subscribers( .collect()) } -/// A `// on ` whose stored ref carries the literal `{partition}` -/// token is partition-bearing — its concrete partition is the AND-join -/// key. Non-token asset refs are reference/presence-only inputs. -fn is_partition_bearing_ref(trigger_ref: &str) -> bool { - trigger_ref.contains(PARTITION_TOKEN) -} - -/// Record an AND input arrival for `(subscriber, partition)` and report -/// whether every partition-bearing input is now present for that -/// partition. The record -> count -> clear sequence runs in one -/// transaction guarded by a transaction-scoped advisory lock keyed on the -/// slot: the same subscriber's last two partition-bearing inputs can -/// complete concurrently on different workers, and a check-then-act on a -/// pooled connection would let both observe "complete" and dispatch -/// twice. Serializing per `(workspace, subscriber, partition)` makes the -/// gate fire exactly once; the lock is released on commit/rollback. -/// Idempotent per input (PK conflict ignored); the slot is cleared on -/// fire so later writes re-accumulate and can re-materialize. -/// -/// `required` = the distinct partition-bearing inputs the subscriber -/// declares (its `{partition}`-token `// on` lines). Reference inputs -/// (no token) and non-asset triggers are presence-only and excluded. -async fn record_and_check_join_slot( - db: &DB, - workspace_id: &str, - subscriber_path: &str, - partition: &str, - trigger_ref: &str, -) -> Result { - let mut tx = db.begin().await?; - let lock_key = format!("{workspace_id}|{subscriber_path}|{partition}"); - sqlx::query!( - "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", - lock_key, - ) - .execute(&mut *tx) - .await?; - sqlx::query!( - r#"INSERT INTO join_pending_inputs - (workspace_id, subscriber_path, partition, trigger_ref) - VALUES ($1, $2, $3, $4) - ON CONFLICT DO NOTHING"#, - workspace_id, - subscriber_path, - partition, - trigger_ref, - ) - .execute(&mut *tx) - .await?; - let required = sqlx::query_scalar!( - r#"SELECT count(DISTINCT trigger_ref) AS "n!" - FROM script_trigger - WHERE workspace_id = $1 - AND runnable_path = $2 - AND trigger_kind = 'asset' - AND runnable_kind = 'script' - AND trigger_ref LIKE '%' || $3 || '%'"#, - workspace_id, - subscriber_path, - PARTITION_TOKEN, - ) - .fetch_one(&mut *tx) - .await?; - let received = sqlx::query_scalar!( - r#"SELECT count(DISTINCT trigger_ref) AS "n!" - FROM join_pending_inputs - WHERE workspace_id = $1 AND subscriber_path = $2 AND partition = $3"#, - workspace_id, - subscriber_path, - partition, - ) - .fetch_one(&mut *tx) - .await?; - let fire = required > 0 && received >= required; - if fire { - sqlx::query!( - r#"DELETE FROM join_pending_inputs - WHERE workspace_id = $1 AND subscriber_path = $2 AND partition = $3"#, - workspace_id, - subscriber_path, - partition, - ) - .execute(&mut *tx) - .await?; - } - tx.commit().await?; - // i64 -> i32: the COUNT(DISTINCT trigger_ref) values are bounded by the - // number of `// on` lines on a single subscriber — fits trivially. - Ok(JoinSlotStatus { fired: fire, received: received as i32, required: required as i32 }) -} - -/// Default time a partial AND-join slot may sit with no new input before -/// it is abandoned. A slot is normally cleared the moment the join fires; -/// this only reaps slots whose inputs never all arrive (an upstream was -/// removed/renamed, a one-off `dynamic` partition key, permanent skew). -/// Conservative so a legitimately slow join is never reaped; a per-join -/// configurable TTL via the annotation is a planned follow-up. -pub const JOIN_SLOT_TTL_SECS: i64 = 60 * 24 * 60 * 60; // 60 days - -/// Reap abandoned AND-join slots. Keyed on the *slot's most recent row* -/// (`HAVING max(received_at)`), never per-row, so a join whose inputs -/// trickle in over a window longer than one input's age is not corrupted -/// mid-accumulation. Called periodically from the monitor loop. -pub async fn reap_stale_join_slots(db: &DB) -> Result<()> { - sqlx::query!( - "DELETE FROM join_pending_inputs jpi - USING ( - SELECT workspace_id, subscriber_path, partition - FROM join_pending_inputs - GROUP BY workspace_id, subscriber_path, partition - HAVING max(received_at) <= now() - ($1::bigint::text || ' s')::interval - ) stale - WHERE jpi.workspace_id = stale.workspace_id - AND jpi.subscriber_path = stale.subscriber_path - AND jpi.partition = stale.partition", - JOIN_SLOT_TTL_SECS, - ) - .execute(db) - .await?; - Ok(()) -} - async fn push_subscriber( db: &DB, producer: &MiniCompletedJob, @@ -757,40 +572,20 @@ async fn push_subscriber( let tag = script.tag; let concurrency_settings = script.runnable_settings.concurrency_settings; - // Debounce is opt-in per subscriber edge (`// debounce` / - // `// on … debounce=`). When set, the window is keyed by - // (subscriber, partition) so distinct partitions never collapse and - // "latest within the window" falls out for free. When not set, the - // subscriber's own script-level debounce settings (if any) apply. - let 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() - }, - _ => script.runnable_settings.debouncing_settings, - }; + // Debounce / retry semantics are a `private` feature (see `cascade`). + // OSS degrades both: debounce falls back to the subscriber's own + // script-level settings, retry is never applied. + let debouncing_settings = crate::cascade::cascade_debouncing_settings( + subscriber_path, + partition, + debounce_s, + script.runnable_settings.debouncing_settings, + ); // Retry is only available via the flow runtime — wrap the script in a - // one-step flow with `Retry { constant: { attempts, seconds } }` when - // the cascade declares `// retry`. The exponential delay and retry_if - // expression are intentionally unused: the annotation grammar is - // count-plus-constant-delay only (parser-light). Empty retry = - // unwrapped `ScriptHash` push, matching the previous behaviour. - let payload = if let Some(count) = retry_count.filter(|c| *c > 0) { - let delay = retry_delay_s.unwrap_or(0).max(0).min(u16::MAX as i32) as u16; - let retry = windmill_common::flows::Retry { - constant: windmill_common::flows::ConstantDelay { - attempts: count as u32, - seconds: delay, - }, - exponential: windmill_common::flows::ExponentialDelay::default(), - retry_if: None, - }; + // one-step flow when the cascade declares one. No retry = + // unwrapped `ScriptHash` push. + let payload = if let Some(retry) = crate::cascade::cascade_retry(retry_count, retry_delay_s) { JobPayload::SingleStepFlow { path: subscriber_path.to_string(), hash: Some(hash), diff --git a/backend/windmill-queue/src/cascade_oss.rs b/backend/windmill-queue/src/cascade_oss.rs new file mode 100644 index 0000000000..be6f93d732 --- /dev/null +++ b/backend/windmill-queue/src/cascade_oss.rs @@ -0,0 +1,60 @@ +//! OSS fallback for the asset-trigger cascade's AND-join / debounce / retry. +//! +//! The richer cascade semantics are a `private` feature (see `cascade_ee`). +//! In the public build they are absent and the cascade degrades to a plain +//! OR fan-out: every join always fires immediately, no slots are recorded or +//! reaped, debounce falls back to the subscriber's own script-level settings, +//! and retry is never applied (the subscriber pushes as a bare `ScriptHash`). +//! The real implementation lives in `windmill-ee-private`. + +use windmill_common::error::Result; +use windmill_common::flows::Retry; +use windmill_common::runnable_settings::DebouncingSettings; +use windmill_common::DB; + +/// Outcome of evaluating an AND-join barrier for one (subscriber, input). +/// Kept identical to the EE definition so the core matcher in +/// `asset_dispatch` compiles against either build. +pub enum JoinDecision { + /// Input does not advance the join (recorded as Skipped with `reason`). + Skip(&'static str), + /// Join advanced but is not yet complete (recorded as JoinPending). + Pending { received: i32, required: i32 }, + /// All required inputs are present — push the subscriber. + Fire, +} + +/// OSS has no AND-join: every input fires immediately (plain OR fan-out). +pub async fn handle_join( + _db: &DB, + _workspace_id: &str, + _sub_path: &str, + _trigger_ref: &str, + _partition: Option<&str>, +) -> Result { + Ok(JoinDecision::Fire) +} + +/// Unused in OSS (no slots are ever recorded), kept for API parity. +pub const JOIN_SLOT_TTL_SECS: i64 = 60 * 24 * 60 * 60; // 60 days + +/// No-op: OSS never records join slots, so there is nothing to reap. +pub async fn reap_stale_join_slots(_db: &DB) -> Result<()> { + Ok(()) +} + +/// No per-edge debounce in OSS — always defer to the subscriber's own +/// script-level debounce settings. +pub fn cascade_debouncing_settings( + _subscriber_path: &str, + _partition: Option<&str>, + _debounce_s: Option, + fallback: DebouncingSettings, +) -> DebouncingSettings { + fallback +} + +/// No per-edge retry in OSS — the subscriber pushes as a bare `ScriptHash`. +pub fn cascade_retry(_retry_count: Option, _retry_delay_s: Option) -> Option { + None +} diff --git a/backend/windmill-queue/src/lib.rs b/backend/windmill-queue/src/lib.rs index 2f73a1a97a..6f689c7026 100644 --- a/backend/windmill-queue/src/lib.rs +++ b/backend/windmill-queue/src/lib.rs @@ -7,6 +7,13 @@ */ pub mod asset_dispatch; +#[cfg(feature = "private")] +pub mod cascade_ee; +pub mod cascade_oss; +#[cfg(feature = "private")] +pub use cascade_ee as cascade; +#[cfg(not(feature = "private"))] +pub use cascade_oss as cascade; pub mod jobs; #[cfg(feature = "private")] pub mod jobs_ee;