From 78c8d9b4d709d67e2ec854ac73e2aa1c50e7fc6c Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 18 Apr 2026 15:46:05 +0000 Subject: [PATCH] feat: add asset-trigger dispatcher primitives and save-time signature check Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/windmill-api-scripts/src/scripts.rs | 16 ++ .../windmill-common/src/asset_dispatcher.rs | 148 ++++++++++++++++++ .../windmill-common/src/implicit_triggers.rs | 43 +++++ backend/windmill-common/src/lib.rs | 1 + 4 files changed, 208 insertions(+) create mode 100644 backend/windmill-common/src/asset_dispatcher.rs diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 35b4f3955b..f68aec4fa9 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -999,6 +999,22 @@ async fn create_script_internal<'c>( ) }; + // Stage 5 validation: reject asset-triggered scripts whose signature has + // required args outside the reserved dispatcher set, so the script can + // actually be run by a reactive fire. + { + let asset_annotations = windmill_common::trigger_annotations::parse_trigger_annotations( + &ns.content, + ns.language, + ) + .map_err(|e| Error::BadRequest(e.to_string()))?; + if !asset_annotations.is_empty() { + let schema_raw = ns.schema.as_ref().map(|s| &**s.0.as_ref()); + windmill_common::implicit_triggers::validate_asset_triggered_script_schema(schema_raw) + .map_err(Error::BadRequest)?; + } + } + sqlx::query!( "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \ content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \ diff --git a/backend/windmill-common/src/asset_dispatcher.rs b/backend/windmill-common/src/asset_dispatcher.rs new file mode 100644 index 0000000000..9b8f69b165 --- /dev/null +++ b/backend/windmill-common/src/asset_dispatcher.rs @@ -0,0 +1,148 @@ +//! Reactive dispatcher for asset-change triggers (stage 4). +//! +//! Polls `asset_event` for new rows, matches them against +//! `asset_trigger.subscription_set`, and enqueues downstream runs. +//! +//! This module intentionally exposes **stateless** building blocks (match + +//! policy application + reserved-arg construction); the runtime wiring that +//! turns them into a long-running task with server-ownership and job push +//! lives in `windmill-worker` so that `windmill-common` can stay free of a +//! `windmill-queue` dependency. + +use serde::{Deserialize, Serialize}; +use sqlx::{Pool, Postgres}; + +use crate::{assets::AssetKind, error::Result}; + +/// One pending event picked up from `asset_event`. The dispatcher hands +/// matched rows plus the triggering event down to the enqueue path. +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct PendingAssetEvent { + pub id: i64, + pub workspace_id: String, + pub asset_kind: AssetKind, + pub asset_path: String, + pub partition_key: Option, + pub job_id: uuid::Uuid, + pub at: chrono::DateTime, +} + +/// Row shape returned by the matching query — just what the enqueue path +/// needs, not the full trigger row. +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct MatchedAssetTrigger { + pub workspace_id: String, + pub path: String, + pub script_path: String, + pub is_flow: bool, + pub fires: String, + pub debounce_s: i32, + pub cancel_on_new: bool, + pub owner_script_hash: Option, +} + +/// Fetch events with id greater than `last_id`, ordered by id ascending. +/// Bounded to a single batch so the dispatcher makes forward progress even +/// under burst. +pub async fn fetch_pending_events( + db: &Pool, + last_id: i64, + batch_size: i32, +) -> Result> { + let events = sqlx::query_as::<_, PendingAssetEvent>( + "SELECT id, workspace_id, asset_kind, asset_path, partition_key, job_id, at \ + FROM asset_event \ + WHERE id > $1 \ + ORDER BY id ASC \ + LIMIT $2", + ) + .bind(last_id) + .bind(batch_size) + .fetch_all(db) + .await?; + Ok(events) +} + +/// Select `asset_trigger` rows whose subscription set contains the +/// `(kind, path)` of the given event. Uses JSONB containment so the GIN +/// index on `subscription_set` kicks in. +pub async fn match_triggers_for_event( + db: &Pool, + event: &PendingAssetEvent, +) -> Result> { + let subscription_probe = serde_json::json!({ + "paths": [{ "kind": event.asset_kind, "path": event.asset_path }] + }); + let rows = sqlx::query_as::<_, MatchedAssetTrigger>( + "SELECT workspace_id, path, script_path, is_flow, fires, debounce_s, \ + cancel_on_new, owner_script_hash \ + FROM asset_trigger \ + WHERE workspace_id = $1 \ + AND mode = 'enabled' \ + AND subscription_set @> $2::jsonb", + ) + .bind(&event.workspace_id) + .bind(&subscription_probe) + .fetch_all(db) + .await?; + Ok(rows) +} + +/// Reserved argument payload passed to asset-triggered scripts. The worker +/// fills the corresponding `wm_asset_event` / `wm_partition` / `wm_backfill` +/// parameters at job materialization (stage 5). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AssetEventArg { + pub id: i64, + pub workspace_id: String, + pub asset_kind: AssetKind, + pub asset_path: String, + pub partition_key: Option, + pub job_id: uuid::Uuid, + pub at: chrono::DateTime, +} + +impl From<&PendingAssetEvent> for AssetEventArg { + fn from(e: &PendingAssetEvent) -> Self { + Self { + id: e.id, + workspace_id: e.workspace_id.clone(), + asset_kind: e.asset_kind, + asset_path: e.asset_path.clone(), + partition_key: e.partition_key.clone(), + job_id: e.job_id, + at: e.at, + } + } +} + +/// For `fires = 'all'`, the trigger should only fire when *every* path in +/// the subscription set has seen a newer event than the last fire. This +/// helper checks that condition against `asset_event`. Returns `true` if +/// the trigger is ready to fire. +pub async fn fires_all_ready( + db: &Pool, + workspace_id: &str, + subscription_paths: &[(AssetKind, String)], + last_fired_at: Option>, +) -> Result { + let last = last_fired_at.unwrap_or_else(|| { + chrono::DateTime::::from_timestamp(0, 0).unwrap_or_default() + }); + for (kind, path) in subscription_paths { + let max_at = sqlx::query_scalar::<_, Option>>( + "SELECT MAX(at) FROM asset_event \ + WHERE workspace_id = $1 AND asset_kind = $2 AND asset_path = $3", + ) + .bind(workspace_id) + .bind(kind) + .bind(path) + .fetch_one(db) + .await?; + match max_at { + Some(t) if t > last => continue, + _ => return Ok(false), + } + } + Ok(true) +} diff --git a/backend/windmill-common/src/implicit_triggers.rs b/backend/windmill-common/src/implicit_triggers.rs index 221165ff53..d6c5020a22 100644 --- a/backend/windmill-common/src/implicit_triggers.rs +++ b/backend/windmill-common/src/implicit_triggers.rs @@ -26,6 +26,49 @@ use crate::{ /// work unchanged (`SPLIT_PART(path, '/', 1)` — 'u' / 'f' / 'g'). pub const IMPLICIT_TRIGGER_PATH_SUFFIX: &str = "/__asset_trigger__"; +/// Reserved argument names that asset-triggered scripts may receive from +/// the dispatcher. When a `#trigger: asset` annotation is present, any +/// *other* required-without-default arg makes the script un-fillable and +/// is rejected at save time (stage 5). +pub const RESERVED_ARG_NAMES: &[&str] = &["wm_asset_event", "wm_partition", "wm_backfill"]; + +/// Validate that an asset-triggered script's schema has no required args +/// outside the reserved set. Schema follows the JSON-Schema convention +/// used everywhere else in Windmill: `required: [name,...]` plus +/// `properties: { name: { default?: ... } }`. +/// +/// Returns `Err` with a user-facing message on the first offending arg. +pub fn validate_asset_triggered_script_schema( + schema: Option<&serde_json::value::RawValue>, +) -> std::result::Result<(), String> { + let Some(raw) = schema else { return Ok(()) }; + let Ok(parsed) = serde_json::from_str::(raw.get()) else { + return Ok(()); + }; + let Some(required) = parsed.get("required").and_then(|v| v.as_array()) else { + return Ok(()); + }; + let properties = parsed.get("properties").and_then(|v| v.as_object()); + for arg in required { + let Some(name) = arg.as_str() else { continue }; + if RESERVED_ARG_NAMES.contains(&name) { + continue; + } + let has_default = properties + .and_then(|p| p.get(name)) + .and_then(|v| v.as_object()) + .map(|obj| obj.contains_key("default")) + .unwrap_or(false); + if !has_default { + return Err(format!( + "Script has required parameter `{name}` but is asset-triggered. \ + Give it a default, make it optional, or remove the asset trigger annotation." + )); + } + } + Ok(()) +} + pub fn implicit_trigger_path(script_path: &str) -> String { format!("{script_path}{IMPLICIT_TRIGGER_PATH_SUFFIX}") } diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 80f88251c5..55a130a474 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -34,6 +34,7 @@ pub mod ai_google; pub mod ai_providers; pub mod ai_types; pub mod apps; +pub mod asset_dispatcher; pub mod assets; pub mod audit; pub mod auth;