From 638b947d002f8a412da11654f9f37e702b04de3b Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 13 May 2026 16:34:44 +0000 Subject: [PATCH] test: cover asset-trigger dispatch end-to-end through worker --- backend/tests/asset_trigger_dispatch.rs | 309 +++++++++++++++++++ backend/windmill-queue/src/asset_dispatch.rs | 1 + 2 files changed, 310 insertions(+) create mode 100644 backend/tests/asset_trigger_dispatch.rs diff --git a/backend/tests/asset_trigger_dispatch.rs b/backend/tests/asset_trigger_dispatch.rs new file mode 100644 index 0000000000..f952350dbc --- /dev/null +++ b/backend/tests/asset_trigger_dispatch.rs @@ -0,0 +1,309 @@ +//! End-to-end test for asset-trigger dispatch. +//! +//! Runs a real Bash producer through a worker, lets the +//! `result_processor` hook fire `dispatch_asset_triggers`, and then makes +//! several follow-up calls into `dispatch_asset_triggers` against the same +//! seeded graph to cover the eligibility branches (self-loop, skip arg, +//! depth cap, flow subscriber, ineligible job kinds). Direct calls share +//! the same workspace so we exercise the real query paths against real +//! `asset` / `script_trigger` rows produced by deploy-equivalent seeding. + +use serde_json::json; +use sqlx::{Pool, Postgres}; +use uuid::Uuid; +use windmill_common::jobs::{JobKind, JobPayload}; +use windmill_common::scripts::{ScriptHash, ScriptLang}; +use windmill_queue::asset_dispatch::dispatch_asset_triggers; +use windmill_queue::MiniCompletedJob; +use windmill_test_utils::{initialize_tracing, ApiServer, RunJob}; + +const WS: &str = "test-workspace"; +const PRODUCER: &str = "u/test-user/producer"; +const SUB_S3: &str = "u/test-user/sub-s3"; +const SUB_RES: &str = "u/test-user/sub-res"; +const SUB_FLOW: &str = "u/test-user/sub-flow"; + +// ── Seeding helpers ─────────────────────────────────────────────────────── + +async fn seed_script( + db: &Pool, + path: &str, + content: &str, + language: &str, +) -> anyhow::Result { + // Hash needs to be unique per (workspace, hash). Derive from path so + // sibling scripts in the same test don't collide. + let mut h = 0i64; + for b in path.bytes() { + h = h.wrapping_mul(31).wrapping_add(b as i64); + } + sqlx::query( + r#"INSERT INTO script (workspace_id, hash, path, summary, description, content, + created_by, language, tag, lock) + VALUES ($1, $2, $3, '', '', $4, 'test-user', $5::script_lang, 'deno', '') + ON CONFLICT DO NOTHING"#, + ) + .bind(WS) + .bind(h) + .bind(path) + .bind(content) + .bind(language) + .execute(db) + .await?; + Ok(h) +} + +async fn seed_asset_write( + db: &Pool, + producer_path: &str, + kind: &str, + asset_path: &str, +) -> anyhow::Result<()> { + sqlx::query( + r#"INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind) + VALUES ($1, $2, $3::asset_kind, 'w'::asset_access_type, $4, 'script'::asset_usage_kind) + ON CONFLICT DO NOTHING"#, + ) + .bind(WS) + .bind(asset_path) + .bind(kind) + .bind(producer_path) + .execute(db) + .await?; + Ok(()) +} + +async fn seed_subscription( + db: &Pool, + subscriber_path: &str, + subscriber_kind: &str, + trigger_ref: &str, +) -> anyhow::Result<()> { + sqlx::query( + r#"INSERT INTO script_trigger + (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref) + VALUES ($1, $2::asset_usage_kind, $3, 'asset'::script_trigger_kind, $4)"#, + ) + .bind(WS) + .bind(subscriber_kind) + .bind(subscriber_path) + .bind(trigger_ref) + .execute(db) + .await?; + Ok(()) +} + +/// Insert a synthetic producer `v2_job` row used by the direct +/// `dispatch_asset_triggers` calls in the edge-case section. `args` lets the +/// test inject `_wmill_skip_asset_dispatch` or `trigger.depth` so the +/// dispatcher's arg-driven branches are exercised against real rows. +async fn seed_producer_job(db: &Pool, args: serde_json::Value) -> anyhow::Result { + let id = Uuid::new_v4(); + sqlx::query!( + r#"INSERT INTO v2_job (id, workspace_id, kind, runnable_path, args, created_by, + permissioned_as, permissioned_as_email, tag, script_lang) + VALUES ($1, $2, 'script'::job_kind, $3, $4, 'test-user', + 'u/test-user', 'test@windmill.dev', 'deno', 'bash'::script_lang)"#, + id, + WS, + PRODUCER, + args, + ) + .execute(db) + .await?; + Ok(id) +} + +fn make_mini(id: Uuid, runnable_path: &str) -> MiniCompletedJob { + MiniCompletedJob { + id, + workspace_id: WS.to_string(), + runnable_id: Some(ScriptHash(1)), + scheduled_for: chrono::Utc::now(), + parent_job: None, + flow_innermost_root_job: None, + runnable_path: Some(runnable_path.to_string()), + kind: JobKind::Script, + started_at: Some(chrono::Utc::now()), + permissioned_as: "u/test-user".to_string(), + created_by: "test-user".to_string(), + script_lang: Some(ScriptLang::Bash), + permissioned_as_email: "test@windmill.dev".to_string(), + flow_step_id: None, + trigger_kind: None, + trigger: None, + priority: None, + concurrent_limit: None, + tag: "deno".to_string(), + cache_ttl: None, + cache_ignore_s3_path: None, + runnable_settings_handle: None, + } +} + +/// Read `v2_job` rows that were created by asset dispatch (filtered by +/// `trigger_kind = 'asset'` so dep-jobs / other infra rows don't leak in). +async fn fetch_dispatched( + db: &Pool, +) -> anyhow::Result, Option)>> { + let rows = sqlx::query!( + r#"SELECT runnable_path AS "runnable_path!", trigger, + args AS "args: sqlx::types::Json" + FROM v2_job + WHERE workspace_id = $1 AND trigger_kind = 'asset' + ORDER BY runnable_path"#, + WS, + ) + .fetch_all(db) + .await?; + Ok(rows + .into_iter() + .map(|r| (r.runnable_path, r.trigger, r.args.map(|j| j.0))) + .collect()) +} + +async fn clear_dispatched(db: &Pool) -> anyhow::Result<()> { + sqlx::query!( + "DELETE FROM v2_job WHERE workspace_id = $1 AND trigger_kind = 'asset'", + WS, + ) + .execute(db) + .await?; + Ok(()) +} + +// ── The test ───────────────────────────────────────────────────────────── + +/// One end-to-end test that: +/// 1. seeds a producer that writes two asset kinds (s3 + resource) with three +/// subscribers (two script subs + one flow sub that must be skipped), +/// 2. runs the producer through a real worker and asserts the +/// `result_processor` hook fired and pushed the right jobs with the +/// right trigger metadata, +/// 3. then drives `dispatch_asset_triggers` directly against the same +/// seeded graph to cover the arg-driven and eligibility branches that +/// can't be reached by varying the producer's runtime args alone: +/// - skip arg suppresses dispatch +/// - cascade depth caps fan-out +/// - depth increments correctly just below the cap +/// - self-loop subscriber is filtered +/// - producer with parent_job is ineligible +/// - producer with `Flow` kind is ineligible +#[sqlx::test(fixtures("base"))] +async fn end_to_end_asset_dispatch(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + + // ── Seed the graph ────────────────────────────────────────────────── + let producer_hash = seed_script(&db, PRODUCER, "echo producer", "bash").await?; + seed_script(&db, SUB_S3, "echo s3-subscriber", "bash").await?; + seed_script(&db, SUB_RES, "echo res-subscriber", "bash").await?; + // Self-loop subscriber: same path as producer → must be filtered. + seed_subscription(&db, PRODUCER, "script", "s3://f/blob").await?; + // Flow subscriber on the same asset → V1 hard-filters runnable_kind='flow'. + seed_subscription(&db, SUB_FLOW, "flow", "s3://f/blob").await?; + // Legit subscribers. + seed_subscription(&db, SUB_S3, "script", "s3://f/blob").await?; + seed_subscription(&db, SUB_RES, "script", "$res:f/cfg").await?; + // Two writes from one producer, distinct kinds. + seed_asset_write(&db, PRODUCER, "s3object", "f/blob").await?; + seed_asset_write(&db, PRODUCER, "resource", "f/cfg").await?; + + // ── 1. Real worker run: the hook must fire after producer success ─── + let job = JobPayload::ScriptHash { + path: PRODUCER.to_string(), + hash: ScriptHash(producer_hash), + cache_ttl: None, + cache_ignore_s3_path: None, + dedicated_worker: None, + language: ScriptLang::Bash, + priority: None, + apply_preprocessor: false, + concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(), + debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(), + labels: None, + }; + let completed = RunJob::from(job).run_until_complete(&db, false, port).await; + assert!( + completed.success, + "producer must succeed for dispatch to fire" + ); + + let mut rows = fetch_dispatched(&db).await?; + rows.sort_by(|a, b| a.0.cmp(&b.0)); + assert_eq!( + rows.len(), + 2, + "expected dispatch to the two legit script subscribers (flow sub filtered, self-loop filtered)" + ); + let by_path: std::collections::HashMap<_, _> = rows.iter().map(|r| (r.0.as_str(), r)).collect(); + + let s3_row = by_path + .get(SUB_S3) + .expect("s3 subscriber should have a job"); + let s3_trig = s3_row.2.as_ref().unwrap().get("trigger").unwrap(); + assert_eq!(s3_trig["kind"], "asset"); + assert_eq!(s3_trig["asset_kind"], "s3object"); + assert_eq!(s3_trig["asset_path"], "f/blob"); + assert_eq!(s3_trig["producer_path"], PRODUCER); + assert_eq!(s3_trig["depth"], 1, "depth starts at 1 on first hop"); + assert_eq!(s3_row.1.as_deref(), Some(PRODUCER)); + + let res_row = by_path + .get(SUB_RES) + .expect("resource subscriber should have a job"); + let res_trig = res_row.2.as_ref().unwrap().get("trigger").unwrap(); + assert_eq!(res_trig["asset_kind"], "resource"); + assert_eq!(res_trig["asset_path"], "f/cfg"); + + // ── 2. Direct calls to dispatch_asset_triggers for arg / eligibility + // branches that can't be reached via the runtime path ────────── + clear_dispatched(&db).await?; + + // skip arg suppresses dispatch + let id = seed_producer_job(&db, json!({ "_wmill_skip_asset_dispatch": true })).await?; + let r = dispatch_asset_triggers(&db, &make_mini(id, PRODUCER)).await; + assert_eq!(r.dispatched.len(), 0, "skip arg suppressed dispatch"); + + // depth at cap (MAX_CHAIN_DEPTH = 5) blocks further fan-out + let id = seed_producer_job(&db, json!({ "trigger": { "depth": 5 } })).await?; + let r = dispatch_asset_triggers(&db, &make_mini(id, PRODUCER)).await; + assert_eq!(r.dispatched.len(), 0, "depth at cap blocked"); + + // depth just below cap fans out, new run carries depth + 1 + clear_dispatched(&db).await?; + let id = seed_producer_job(&db, json!({ "trigger": { "depth": 4 } })).await?; + let r = dispatch_asset_triggers(&db, &make_mini(id, PRODUCER)).await; + assert_eq!(r.dispatched.len(), 2); + let rows = fetch_dispatched(&db).await?; + for row in &rows { + let depth = row.2.as_ref().unwrap()["trigger"]["depth"] + .as_i64() + .unwrap(); + assert_eq!(depth, 5, "depth should increment to 5"); + } + + // producer with parent_job (flow step) is ineligible + clear_dispatched(&db).await?; + let id = seed_producer_job(&db, json!({})).await?; + let mut mini = make_mini(id, PRODUCER); + mini.parent_job = Some(Uuid::new_v4()); + let r = dispatch_asset_triggers(&db, &mini).await; + assert_eq!(r.dispatched.len(), 0, "flow step producer ineligible"); + + // producer with kind=Flow is ineligible + let id = seed_producer_job(&db, json!({})).await?; + let mut mini = make_mini(id, PRODUCER); + mini.kind = JobKind::Flow; + let r = dispatch_asset_triggers(&db, &mini).await; + assert_eq!(r.dispatched.len(), 0, "flow producer ineligible"); + + // Sanity: the eligible direct call (clean producer, no args) still fires — + // proves the assertions above are negative cases, not a broken setup. + let id = seed_producer_job(&db, json!({})).await?; + let r = dispatch_asset_triggers(&db, &make_mini(id, PRODUCER)).await; + assert_eq!(r.dispatched.len(), 2, "clean direct call still dispatches"); + + Ok(()) +} diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 3155d8019c..59eaba58e0 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -398,3 +398,4 @@ async fn push_subscriber( tx.commit().await?; Ok(id) } +