diff --git a/backend/.sqlx/query-2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c.json b/backend/.sqlx/query-2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c.json new file mode 100644 index 0000000000..8153ca9932 --- /dev/null +++ b/backend/.sqlx/query-2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM schedule\n WHERE workspace_id = $1\n AND managed_by_runnable_path = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "2a934656604281fd66aad72e08cf7f3268d5fc4144184ed8379ca0d3a638bc1c" +} diff --git a/backend/.sqlx/query-3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57.json b/backend/.sqlx/query-3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57.json new file mode 100644 index 0000000000..a236063636 --- /dev/null +++ b/backend/.sqlx/query-3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT runnable_path AS \"runnable_path!\"\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": [ + { + "ordinal": 0, + "name": "runnable_path!", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57" +} diff --git a/backend/.sqlx/query-58e8e13acd9f7ff951f37d555beab84dcea21e0a38906de164889cf2dacf2e43.json b/backend/.sqlx/query-58e8e13acd9f7ff951f37d555beab84dcea21e0a38906de164889cf2dacf2e43.json new file mode 100644 index 0000000000..3a09f1ee8d --- /dev/null +++ b/backend/.sqlx/query-58e8e13acd9f7ff951f37d555beab84dcea21e0a38906de164889cf2dacf2e43.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT args AS \"args!: Json>>\"\n FROM v2_job\n WHERE workspace_id = $1 AND id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "args!: Json>>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "Uuid" + ] + }, + "nullable": [ + true + ] + }, + "hash": "58e8e13acd9f7ff951f37d555beab84dcea21e0a38906de164889cf2dacf2e43" +} diff --git a/backend/.sqlx/query-7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b.json b/backend/.sqlx/query-7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b.json new file mode 100644 index 0000000000..dece68de29 --- /dev/null +++ b/backend/.sqlx/query-7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b.json @@ -0,0 +1,47 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO script_trigger\n (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref)\n VALUES ($1, $2, $3, $4, $5)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + { + "Custom": { + "name": "asset_usage_kind", + "kind": { + "Enum": [ + "script", + "flow", + "job" + ] + } + } + }, + "Varchar", + { + "Custom": { + "name": "script_trigger_kind", + "kind": { + "Enum": [ + "asset", + "schedule", + "webhook", + "email", + "kafka", + "mqtt", + "nats", + "postgres", + "sqs", + "gcp" + ] + } + } + }, + "Text" + ] + }, + "nullable": [] + }, + "hash": "7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b" +} diff --git a/backend/.sqlx/query-818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05.json b/backend/.sqlx/query-818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05.json new file mode 100644 index 0000000000..6d84a4158f --- /dev/null +++ b/backend/.sqlx/query-818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM schedule\n WHERE workspace_id = $1\n AND managed_by_runnable_path = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "818188e13fd4da80070697027934503130d042aafe2229bf3e64b925d3e93c05" +} diff --git a/backend/.sqlx/query-8bce3f969b4bbbcf3dc1b8c671ab76e947e491da84fbaf5f130f58fbc851594e.json b/backend/.sqlx/query-8bce3f969b4bbbcf3dc1b8c671ab76e947e491da84fbaf5f130f58fbc851594e.json new file mode 100644 index 0000000000..fe484078c2 --- /dev/null +++ b/backend/.sqlx/query-8bce3f969b4bbbcf3dc1b8c671ab76e947e491da84fbaf5f130f58fbc851594e.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM script_trigger\n WHERE workspace_id = $1 AND runnable_kind = $2 AND runnable_path = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + { + "Custom": { + "name": "asset_usage_kind", + "kind": { + "Enum": [ + "script", + "flow", + "job" + ] + } + } + }, + "Text" + ] + }, + "nullable": [] + }, + "hash": "8bce3f969b4bbbcf3dc1b8c671ab76e947e491da84fbaf5f130f58fbc851594e" +} diff --git a/backend/.sqlx/query-a9a99880d870266f474878dd6ef541df988da527d30f663ef6f764f0c3d70d4b.json b/backend/.sqlx/query-a9a99880d870266f474878dd6ef541df988da527d30f663ef6f764f0c3d70d4b.json new file mode 100644 index 0000000000..d12d305b92 --- /dev/null +++ b/backend/.sqlx/query-a9a99880d870266f474878dd6ef541df988da527d30f663ef6f764f0c3d70d4b.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT path AS \"path!\"\n FROM script\n WHERE workspace_id = $1\n AND auto_kind = 'pipeline'\n AND archived = false\n AND deleted = false\n AND ($2::text IS NULL OR path LIKE $2)\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path!", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "a9a99880d870266f474878dd6ef541df988da527d30f663ef6f764f0c3d70d4b" +} diff --git a/backend/.sqlx/query-abb36bfddf707c7897b3936e982725f973eeb6ba5abbd61416060fb7d675593b.json b/backend/.sqlx/query-abb36bfddf707c7897b3936e982725f973eeb6ba5abbd61416060fb7d675593b.json new file mode 100644 index 0000000000..7795458cef --- /dev/null +++ b/backend/.sqlx/query-abb36bfddf707c7897b3936e982725f973eeb6ba5abbd61416060fb7d675593b.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n substring(path from '^f/([^/]+)/') AS \"folder!\",\n COUNT(*) AS \"script_count!\"\n FROM script\n WHERE workspace_id = $1\n AND auto_kind = 'pipeline'\n AND archived = false\n AND deleted = false\n AND path LIKE 'f/%'\n GROUP BY substring(path from '^f/([^/]+)/')\n ORDER BY substring(path from '^f/([^/]+)/')\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "folder!", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "script_count!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "abb36bfddf707c7897b3936e982725f973eeb6ba5abbd61416060fb7d675593b" +} diff --git a/backend/.sqlx/query-c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df.json b/backend/.sqlx/query-c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df.json new file mode 100644 index 0000000000..2a122623fc --- /dev/null +++ b/backend/.sqlx/query-c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO schedule (\n workspace_id, path, schedule, timezone, edited_by, script_path,\n is_flow, enabled, email, permissioned_as,\n ws_error_handler_muted, no_flow_overlap, cron_version,\n managed_by_runnable_path\n )\n VALUES ($1, $2, $3, 'UTC', $4, $2, $5, true, $6, $7, false, false, 'v2', $2)\n ON CONFLICT (workspace_id, path) DO UPDATE\n SET schedule = EXCLUDED.schedule,\n edited_at = now(),\n edited_by = EXCLUDED.edited_by,\n managed_by_runnable_path = EXCLUDED.managed_by_runnable_path\n WHERE schedule.managed_by_runnable_path = EXCLUDED.managed_by_runnable_path\n OR schedule.managed_by_runnable_path IS NULL AND schedule.script_path = EXCLUDED.script_path\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar", + "Varchar", + "Bool", + "Varchar", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "c0a3643369481c056214b61fb8d42f3e935111904a9d05f86c5b6d44704662df" +} diff --git a/backend/.sqlx/query-c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08.json b/backend/.sqlx/query-c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08.json new file mode 100644 index 0000000000..8376b31a9b --- /dev/null +++ b/backend/.sqlx/query-c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08.json @@ -0,0 +1,52 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n runnable_kind AS \"runnable_kind!: AssetUsageKind\",\n runnable_path AS \"runnable_path!\",\n trigger_kind::text AS \"trigger_kind!\",\n trigger_ref AS \"trigger_ref!\"\n FROM script_trigger\n WHERE workspace_id = $1\n AND ($2::text IS NULL OR runnable_path LIKE $2)\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "runnable_kind!: AssetUsageKind", + "type_info": { + "Custom": { + "name": "asset_usage_kind", + "kind": { + "Enum": [ + "script", + "flow", + "job" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "runnable_path!", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "trigger_kind!", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "trigger_ref!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + false, + null, + false + ] + }, + "hash": "c17d1607191ccb033c48f9f620db980d2c02ed435b56b0ff0d11c49a63154e08" +} diff --git a/backend/.sqlx/query-de06f44bad94710f14e9be4c0a6e6080e3c4faae5052500b93cc24b6fe556f2b.json b/backend/.sqlx/query-de06f44bad94710f14e9be4c0a6e6080e3c4faae5052500b93cc24b6fe556f2b.json new file mode 100644 index 0000000000..2ba317edc6 --- /dev/null +++ b/backend/.sqlx/query-de06f44bad94710f14e9be4c0a6e6080e3c4faae5052500b93cc24b6fe556f2b.json @@ -0,0 +1,43 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n kind AS \"kind!: AssetKind\",\n path AS \"path!\"\n FROM asset\n WHERE workspace_id = $1\n AND usage_kind = 'script'\n AND usage_path = $2\n AND usage_access_type IN ('w', 'rw')\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "kind!: AssetKind", + "type_info": { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "path!", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "de06f44bad94710f14e9be4c0a6e6080e3c4faae5052500b93cc24b6fe556f2b" +} diff --git a/backend/migrations/20260510174213_asset_trigger_dispatch.down.sql b/backend/migrations/20260510174213_asset_trigger_dispatch.down.sql new file mode 100644 index 0000000000..76e5b7050c --- /dev/null +++ b/backend/migrations/20260510174213_asset_trigger_dispatch.down.sql @@ -0,0 +1,3 @@ +-- Postgres has no ALTER TYPE ... DROP VALUE for enums. The 'asset' value +-- stays even on rollback; this is consistent with how other job_trigger_kind +-- values were added (see 20250323162033_add-missing-trigger-kind-...). diff --git a/backend/migrations/20260510174213_asset_trigger_dispatch.up.sql b/backend/migrations/20260510174213_asset_trigger_dispatch.up.sql new file mode 100644 index 0000000000..30cdb58ec3 --- /dev/null +++ b/backend/migrations/20260510174213_asset_trigger_dispatch.up.sql @@ -0,0 +1,5 @@ +-- Add 'asset' as a job_trigger_kind so jobs that get dispatched as a +-- consequence of an upstream pipeline script writing an asset can be +-- attributed via v2_job.trigger_kind = 'asset'. The producer's runnable +-- path goes into v2_job.trigger. +ALTER TYPE job_trigger_kind ADD VALUE IF NOT EXISTS 'asset'; diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index c6a7de2601..21fdaa2435 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -23677,6 +23677,7 @@ components: - azure - google - github + - asset TriggerMode: description: job trigger mode diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs new file mode 100644 index 0000000000..3155d8019c --- /dev/null +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -0,0 +1,400 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2022 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +//! Runtime fan-out for asset-triggered scripts. +//! +//! When a producer pipeline script (`// pipeline`) writes an asset and a +//! downstream script subscribes to that asset via `// on s3://...`, this +//! module pushes a job for each subscriber after the producer's job +//! completes successfully. +//! +//! Eligibility (V1, narrow on purpose): +//! - Producer kind is `Script` or `Preview`. Flows defer. +//! - Producer is top-level (no `parent_job`, no `flow_step_id`). +//! - Producer succeeded. +//! - The producer's args do not contain `_wmill_skip_asset_dispatch: true`. +//! - Cascade depth (carried in args under `trigger.depth`) is below +//! `MAX_CHAIN_DEPTH`. +//! +//! Subscribers (V1): +//! - Only `script` runnables. Flow subscribers defer. +//! - The subscriber must have at least one non-archived script row. +//! - A subscriber is skipped if its path equals the producer's path +//! (avoids trivial self-loops). +//! +//! Args sent to subscribers: +//! ```json +//! { +//! "trigger": { +//! "kind": "asset", +//! "asset_kind": "s3object", +//! "asset_path": "...", +//! "producer_path": "...", +//! "producer_job_id": "...", +//! "depth": 1 +//! } +//! } +//! ``` +//! +//! Errors are logged but never bubble up to fail the producer's job. + +use crate::{push, MiniCompletedJob, PushArgs, PushIsolationLevel}; +use serde_json::value::RawValue; +use sqlx::types::Json; +use sqlx::{Pool, Postgres}; +use std::collections::HashMap; +use uuid::Uuid; +use windmill_common::assets::{parse_asset_trigger_ref, AssetKind}; +use windmill_common::error::{self, Result}; +use windmill_common::get_latest_hash_for_path; +use windmill_common::jobs::{JobKind, JobPayload, JobTriggerKind}; +use windmill_common::runnable_settings::{ConcurrencySettings, DebouncingSettings}; +use windmill_common::triggers::TriggerMetadata; +use windmill_common::users::username_to_permissioned_as; +use windmill_common::worker::to_raw_value; +use windmill_common::DB; + +/// Reserved arg key that suppresses asset-trigger dispatch for a single run. +/// Set by the test panel when the user opts out of the cascade. +pub const SKIP_ASSET_DISPATCH_ARG: &str = "_wmill_skip_asset_dispatch"; + +/// Reserved arg key (under `trigger.depth`) that carries cascade depth. +const CHAIN_DEPTH_KEY: &str = "depth"; + +/// Cap cascade depth so a misconfigured ring doesn't fan out unbounded. +const MAX_CHAIN_DEPTH: i64 = 5; + +/// Returned to the caller (the worker's completed-job hook) so logs can +/// reference the dispatched ids. +#[derive(Debug, Default)] +pub struct DispatchResult { + pub dispatched: Vec, +} + +/// Top-level entry. Returns `Ok(default)` and logs on any internal failure +/// rather than propagating, because dispatch is best-effort and must not +/// retroactively fail the producer. +pub async fn dispatch_asset_triggers(db: &DB, job: &MiniCompletedJob) -> DispatchResult { + match try_dispatch(db, job).await { + Ok(r) => r, + Err(e) => { + tracing::error!("asset-trigger dispatch failed for job {}: {e:#}", job.id); + DispatchResult::default() + } + } +} + +async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result { + if !is_eligible_kind(job) { + return Ok(DispatchResult::default()); + } + let runnable_path = match job.runnable_path.as_deref() { + Some(p) if !p.is_empty() => p, + _ => return Ok(DispatchResult::default()), + }; + + // Args were moved from v2_job to v2_job_completed by add_completed_job + // before dispatch runs. Fetch from v2_job_completed. + let args = fetch_args(db, &job.workspace_id, job.id).await?; + if read_skip_arg(args.as_ref()) { + return Ok(DispatchResult::default()); + } + let depth = read_chain_depth(args.as_ref()); + if depth >= MAX_CHAIN_DEPTH { + tracing::warn!( + "asset-trigger dispatch skipped: chain depth {} >= cap {} (job {}, path {})", + depth, + MAX_CHAIN_DEPTH, + job.id, + runnable_path + ); + return Ok(DispatchResult::default()); + } + + let writes = fetch_producer_writes(db, &job.workspace_id, runnable_path).await?; + if writes.is_empty() { + return Ok(DispatchResult::default()); + } + + let mut dispatched = Vec::new(); + for (asset_kind, asset_path) in writes { + let Some(prefix) = prefix_for(asset_kind) else { + continue; + }; + let trigger_ref = format!("{}{}", prefix, asset_path); + let subs = fetch_subscribers(db, &job.workspace_id, &trigger_ref).await?; + for sub_path in subs { + if sub_path == runnable_path { + continue; + } + match push_subscriber( + db, + job, + &sub_path, + asset_kind, + &asset_path, + runnable_path, + depth + 1, + ) + .await + { + Ok(id) => dispatched.push(id), + Err(e) => { + tracing::error!("failed to push asset-triggered job for {}: {e:#}", sub_path) + } + } + } + } + + if !dispatched.is_empty() { + tracing::info!( + "asset-trigger dispatch from job {} ({}): pushed {} downstream jobs", + job.id, + runnable_path, + dispatched.len() + ); + } + Ok(DispatchResult { dispatched }) +} + +fn is_eligible_kind(job: &MiniCompletedJob) -> bool { + if !matches!(job.kind, JobKind::Script | JobKind::Preview) { + return false; + } + if job.parent_job.is_some() || job.flow_step_id.is_some() { + return false; + } + true +} + +async fn fetch_args( + db: &Pool, + workspace_id: &str, + job_id: Uuid, +) -> Result>>> { + // Read from v2_job because args live there permanently — v2_job_completed + // is the *result* row and doesn't carry args. The producer's v2_job row + // is still present at dispatch time (deletion happens later in the + // completion pipeline, after this hook). + let row = sqlx::query!( + r#"SELECT args AS "args!: Json>>" + FROM v2_job + WHERE workspace_id = $1 AND id = $2"#, + workspace_id, + job_id, + ) + .fetch_optional(db) + .await?; + Ok(row.map(|r| r.args.0)) +} + +fn read_skip_arg(args: Option<&HashMap>>) -> bool { + args.and_then(|a| a.get(SKIP_ASSET_DISPATCH_ARG)) + .and_then(|v| serde_json::from_str::(v.get()).ok()) + .unwrap_or(false) +} + +fn read_chain_depth(args: Option<&HashMap>>) -> i64 { + let Some(args) = args else { + return 0; + }; + let Some(trigger) = args.get("trigger") else { + return 0; + }; + let Ok(map) = serde_json::from_str::>>(trigger.get()) else { + return 0; + }; + map.get(CHAIN_DEPTH_KEY) + .and_then(|v| serde_json::from_str::(v.get()).ok()) + .unwrap_or(0) +} + +fn prefix_for(kind: AssetKind) -> Option<&'static str> { + match kind { + AssetKind::S3Object => Some("s3://"), + AssetKind::Resource => Some("$res:"), + AssetKind::Ducklake => Some("ducklake://"), + AssetKind::DataTable => Some("datatable://"), + AssetKind::Volume => Some("volume://"), + // Deprecated kind from before the parser was unified — has no + // canonical trigger ref and never produced trigger rows. + AssetKind::Variable => None, + } +} + +async fn fetch_producer_writes( + db: &Pool, + workspace_id: &str, + runnable_path: &str, +) -> Result> { + let rows = sqlx::query!( + r#" + SELECT + kind AS "kind!: AssetKind", + path AS "path!" + FROM asset + WHERE workspace_id = $1 + AND usage_kind = 'script' + AND usage_path = $2 + AND usage_access_type IN ('w', 'rw') + "#, + workspace_id, + runnable_path, + ) + .fetch_all(db) + .await?; + Ok(rows.into_iter().map(|r| (r.kind, r.path)).collect()) +} + +async fn fetch_subscribers( + db: &Pool, + workspace_id: &str, + trigger_ref: &str, +) -> 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. + let rows = sqlx::query!( + r#" + SELECT runnable_path AS "runnable_path!" + FROM script_trigger + WHERE workspace_id = $1 + AND trigger_kind = 'asset' + AND trigger_ref = $2 + AND runnable_kind = 'script' + "#, + workspace_id, + trigger_ref, + ) + .fetch_all(db) + .await?; + // Sanity check: parsing the stored trigger_ref must succeed. If it + // doesn't, the row is corrupt; skip it loudly rather than silently. + if parse_asset_trigger_ref(trigger_ref).is_none() { + tracing::warn!( + "asset-trigger dispatch: trigger_ref {} did not round-trip through parse_asset_trigger_ref", + trigger_ref + ); + } + Ok(rows.into_iter().map(|r| r.runnable_path).collect()) +} + +async fn push_subscriber( + db: &DB, + producer: &MiniCompletedJob, + subscriber_path: &str, + asset_kind: AssetKind, + asset_path: &str, + producer_path: &str, + depth: i64, +) -> Result { + let ( + hash, + tag, + _concurrency_key, + _concurrent_limit, + _concurrency_time_window_s, + _debounce_key, + _debounce_delay_s, + cache_ttl, + cache_ignore_s3_path, + language, + dedicated_worker, + priority, + _timeout, + on_behalf_of_email, + created_by, + _runnable_settings_handle, + labels, + ) = get_latest_hash_for_path(db, &producer.workspace_id, subscriber_path, false).await?; + + let payload = JobPayload::ScriptHash { + hash, + path: subscriber_path.to_string(), + cache_ttl, + cache_ignore_s3_path, + dedicated_worker, + 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(), + concurrency_settings: ConcurrencySettings::default(), + labels, + }; + + // Subscriber's own on_behalf_of_email controls identity when set; + // otherwise we run as the producer. This keeps the asset cascade + // attributable to whoever originally wrote the asset, while still + // honoring scripts that explicitly opted into a service-account email. + let (permissioned_as, email) = if let Some(obo) = on_behalf_of_email { + (username_to_permissioned_as(&created_by), obo) + } else { + ( + producer.permissioned_as.clone(), + producer.permissioned_as_email.clone(), + ) + }; + + let mut args: HashMap> = HashMap::new(); + let trigger_payload = serde_json::json!({ + "kind": "asset", + "asset_kind": serde_json::to_value(&asset_kind).unwrap_or(serde_json::Value::Null), + "asset_path": asset_path, + "producer_path": producer_path, + "producer_job_id": producer.id.to_string(), + CHAIN_DEPTH_KEY: depth, + }); + args.insert("trigger".to_string(), to_raw_value(&trigger_payload)); + + // Attribute the dispatched run to a synthetic user so audit logs reflect + // it came from the asset cascade, not the original human runner. + let pseudo_user = format!("asset-{producer_path}"); + + let tx = PushIsolationLevel::IsolatedRoot(db.clone()); + let (id, tx) = push( + db, + tx, + &producer.workspace_id, + payload, + PushArgs { args: &args, extra: None }, + &pseudo_user, + &email, + permissioned_as, + Some(producer_path), + None, + Some(producer_path.to_string()), + None, + None, + None, + None, + false, + false, + None, + true, + tag, + None, + None, + None, + None, + false, + None, + Some(TriggerMetadata::new( + Some(producer_path.to_string()), + JobTriggerKind::Asset, + )), + None, + ) + .await + .map_err(|e| error::Error::internal_err(format!("push asset-triggered job: {e:#}")))?; + tx.commit().await?; + Ok(id) +} diff --git a/backend/windmill-queue/src/lib.rs b/backend/windmill-queue/src/lib.rs index ffd102c081..7a00dfb4df 100644 --- a/backend/windmill-queue/src/lib.rs +++ b/backend/windmill-queue/src/lib.rs @@ -6,6 +6,7 @@ * LICENSE-AGPL for a copy of the license. */ +pub mod asset_dispatch; pub mod jobs; #[cfg(feature = "private")] pub mod jobs_ee; diff --git a/backend/windmill-types/src/jobs.rs b/backend/windmill-types/src/jobs.rs index 03e38d9db3..babdc92ae2 100644 --- a/backend/windmill-types/src/jobs.rs +++ b/backend/windmill-types/src/jobs.rs @@ -43,6 +43,9 @@ pub enum JobTriggerKind { #[serde(rename = "ci_test")] #[sqlx(rename = "ci_test")] CiTest, + // A run dispatched because an upstream pipeline script wrote an asset + // this runnable subscribes to via `// on s3://...` annotations. + Asset, } impl std::fmt::Display for JobTriggerKind { @@ -64,6 +67,7 @@ impl std::fmt::Display for JobTriggerKind { JobTriggerKind::Google => "google", JobTriggerKind::Github => "github", JobTriggerKind::CiTest => "ci_test", + JobTriggerKind::Asset => "asset", }; write!(f, "{}", kind) } diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 9eac0a5dcd..d070800131 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -35,9 +35,9 @@ use windmill_common::{ use windmill_common::bench::{BenchmarkInfo, BenchmarkIter}; use windmill_queue::{ - append_logs, get_mini_completed_job, is_pre_shaped_wm_failure_result, CanceledBy, FlowRunners, - JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson, WrappedError, INIT_SCRIPT_TAG, - MANUAL_FAILURE_ERROR_NAME, + append_logs, asset_dispatch, get_mini_completed_job, is_pre_shaped_wm_failure_result, + CanceledBy, FlowRunners, JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson, + WrappedError, INIT_SCRIPT_TAG, MANUAL_FAILURE_ERROR_NAME, }; use serde_json::{json, value::RawValue, Value}; @@ -789,6 +789,12 @@ pub async fn process_completed_job( from_cache.unwrap_or(false), ) .await?; + + // Asset-trigger fan-out: best-effort, never propagates errors. + // Internal eligibility checks gate to top-level Script/Preview runs; + // see windmill_queue::asset_dispatch. + asset_dispatch::dispatch_asset_triggers(db, &job).await; + drop(job); add_time!(bench, "add_completed_job END"); diff --git a/frontend/src/lib/components/triggers/utils.ts b/frontend/src/lib/components/triggers/utils.ts index 711d84aea9..de9c0f21c0 100644 --- a/frontend/src/lib/components/triggers/utils.ts +++ b/frontend/src/lib/components/triggers/utils.ts @@ -81,7 +81,8 @@ export const jobTriggerKinds: JobTriggerKind[] = [ 'gcp', 'azure', 'google', - 'github' + 'github', + 'asset' ] export type Trigger = {