diff --git a/backend/tests/asset_trigger_dispatch.rs b/backend/tests/asset_trigger_dispatch.rs index e83f6c9fd7..ec61c05e8c 100644 --- a/backend/tests/asset_trigger_dispatch.rs +++ b/backend/tests/asset_trigger_dispatch.rs @@ -4,7 +4,7 @@ //! `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 +//! cycle guard, 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. @@ -107,7 +107,7 @@ async fn seed_subscription( /// 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 +/// test inject `_wmill_skip_asset_dispatch` or `trigger.chain` 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(); @@ -283,8 +283,8 @@ async fn clear_dispatched(db: &Pool) -> anyhow::Result<()> { /// 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 +/// - a subscriber already in the lineage is skipped (cycle guard) +/// - the lineage chain accumulates the producer path each hop /// - self-loop subscriber is filtered /// - producer with parent_job is ineligible /// - producer with `Flow` kind is ineligible @@ -346,7 +346,11 @@ async fn end_to_end_asset_dispatch(db: Pool) -> anyhow::Result<()> { 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_trig["chain"], + json!([PRODUCER]), + "lineage starts with the producer on the first hop" + ); assert_eq!(s3_row.1.as_deref(), Some(PRODUCER)); let res_row = by_path @@ -365,22 +369,28 @@ async fn end_to_end_asset_dispatch(db: Pool) -> anyhow::Result<()> { 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?; + // cycle guard: a subscriber already in the lineage is skipped, but its + // siblings still dispatch (only the cyclic edge is cut). + let id = seed_producer_job(&db, json!({ "trigger": { "chain": [SUB_S3] } })).await?; let r = dispatch_asset_triggers(&db, &make_mini(id, PRODUCER)).await; - assert_eq!(r.dispatched.len(), 0, "depth at cap blocked"); + assert_eq!( + r.dispatched.len(), + 1, + "cyclic subscriber (already in lineage) skipped; sibling still dispatched" + ); - // depth just below cap fans out, new run carries depth + 1 + // lineage accumulates: a fresh producer extends the chain with its own path clear_dispatched(&db).await?; - let id = seed_producer_job(&db, json!({ "trigger": { "depth": 4 } })).await?; + let id = seed_producer_job(&db, json!({ "trigger": { "chain": ["f/upstream"] } })).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"); + assert_eq!( + row.2.as_ref().unwrap()["trigger"]["chain"], + json!(["f/upstream", PRODUCER]), + "lineage accumulates the producer path" + ); } // producer with parent_job (flow step) is ineligible @@ -773,12 +783,12 @@ async fn fuller_partitioned_join_multihop_pipeline(db: Pool) -> anyhow seed_subscription(&db, DN, "script", "s3://lake/{partition}/c").await?; // Helper: assert exactly one dispatch to `path` carrying partition - // `part` at chain depth `depth`. + // `part`, with a cascade lineage of `chain_len` producer paths. async fn assert_hop( db: &Pool, path: &str, part: &str, - depth: i64, + chain_len: usize, ) -> anyhow::Result<()> { let rows = fetch_dispatched(db).await?; let hits: Vec<_> = rows.iter().filter(|r| r.0 == path).collect(); @@ -791,9 +801,9 @@ async fn fuller_partitioned_join_multihop_pipeline(db: Pool) -> anyhow "{path} trigger.partition" ); assert_eq!( - args["trigger"]["depth"].as_i64(), - Some(depth), - "{path} chain depth" + args["trigger"]["chain"].as_array().map(|c| c.len()), + Some(chain_len), + "{path} lineage length" ); Ok(()) } @@ -802,7 +812,7 @@ async fn fuller_partitioned_join_multihop_pipeline(db: Pool) -> anyhow let pa = seed_producer_job_path( &db, PA, - json!({ "partition": "day1", "trigger": { "depth": 1 } }), + json!({ "partition": "day1", "trigger": { "chain": ["s0"] } }), ) .await?; let r = dispatch_asset_triggers(&db, &make_mini(pa, PA)).await; @@ -813,7 +823,7 @@ async fn fuller_partitioned_join_multihop_pipeline(db: Pool) -> anyhow let pb = seed_producer_job_path( &db, PB, - json!({ "partition": "day1", "trigger": { "depth": 1 } }), + json!({ "partition": "day1", "trigger": { "chain": ["s0"] } }), ) .await?; let r = dispatch_asset_triggers(&db, &make_mini(pb, PB)).await; @@ -825,7 +835,7 @@ async fn fuller_partitioned_join_multihop_pipeline(db: Pool) -> anyhow let jn = seed_producer_job_path( &db, JN, - json!({ "partition": "day1", "trigger": { "depth": 2 } }), + json!({ "partition": "day1", "trigger": { "chain": ["s0", PB] } }), ) .await?; let r = dispatch_asset_triggers(&db, &make_mini(jn, JN)).await; @@ -837,7 +847,7 @@ async fn fuller_partitioned_join_multihop_pipeline(db: Pool) -> anyhow let cn = seed_producer_job_path( &db, CN, - json!({ "partition": "day1", "trigger": { "depth": 3 } }), + json!({ "partition": "day1", "trigger": { "chain": ["s0", PB, JN] } }), ) .await?; let r = dispatch_asset_triggers(&db, &make_mini(cn, CN)).await; @@ -850,7 +860,7 @@ async fn fuller_partitioned_join_multihop_pipeline(db: Pool) -> anyhow let pa2 = seed_producer_job_path( &db, PA, - json!({ "partition": "day2", "trigger": { "depth": 1 } }), + json!({ "partition": "day2", "trigger": { "chain": ["s0"] } }), ) .await?; let r = dispatch_asset_triggers(&db, &make_mini(pa2, PA)).await; diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 4ce41747b2..7647ffc864 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -19,14 +19,14 @@ //! - 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). +//! (self-loop) or already appears in the cascade lineage +//! (`trigger.chain`) — cycle detection, which bounds the cascade +//! without capping legitimate depth. //! //! Args sent to subscribers: //! ```json @@ -37,7 +37,7 @@ //! "asset_path": "...", //! "producer_path": "...", //! "producer_job_id": "...", -//! "depth": 1 +//! "chain": ["f/a/producer0", "f/a/producer1"] //! } //! } //! ``` @@ -66,15 +66,20 @@ use windmill_common::DB; /// Set by the test panel when the user opts out of the cascade. pub const SKIP_ASSET_DISPATCH_ARG: &str = "_wmill_skip_asset_dispatch"; -/// Arg key holding the cascade trigger object (carries `depth`, `partition`, +/// Arg key holding the cascade trigger object (carries `chain`, `partition`, /// producer metadata) injected into every dispatched subscriber. const TRIGGER_ARG: &str = "trigger"; -/// Reserved arg key (under `trigger.depth`) that carries cascade depth. -const CHAIN_DEPTH_KEY: &str = "depth"; +/// Arg key (under `trigger.chain`) carrying the cascade lineage: the ordered +/// list of producer paths already run in this chain. Used to detect cycles +/// (a producer re-appearing) and stop only the cyclic edge — so deep but +/// *acyclic* pipelines are never truncated. +const CHAIN_KEY: &str = "chain"; -/// Cap cascade depth so a misconfigured ring doesn't fan out unbounded. -const MAX_CHAIN_DEPTH: i64 = 5; +/// Safety backstop on lineage length. Cycle detection already bounds an +/// acyclic cascade (a path can't repeat), so this only guards against a +/// runaway from a bug. Set far above any real pipeline depth. +const MAX_CHAIN_LEN: usize = 1000; /// Returned to the caller (the worker's completed-job hook) so logs can /// reference the dispatched ids. @@ -243,24 +248,28 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result if read_skip_arg(args.as_ref()) { return Ok(DispatchResult::default()); } - // Parse the cascade `trigger` object once; both depth and the - // propagated partition are read from it. + // Parse the cascade `trigger` object once; both the lineage chain and + // the propagated partition are read from it. let trigger_map = args .as_ref() .and_then(|a| a.get(TRIGGER_ARG)) .and_then(|t| serde_json::from_str::>>(t.get()).ok()); - let depth = read_chain_depth(trigger_map.as_ref()); + let chain = read_chain(trigger_map.as_ref()); let partition = read_partition(args.as_ref(), trigger_map.as_ref()); - if depth >= MAX_CHAIN_DEPTH { + if chain.len() >= MAX_CHAIN_LEN { tracing::warn!( - "asset-trigger dispatch skipped: chain depth {} >= cap {} (job {}, path {})", - depth, - MAX_CHAIN_DEPTH, + "asset-trigger dispatch skipped: cascade lineage length {} >= backstop {} (job {}, path {})", + chain.len(), + MAX_CHAIN_LEN, job.id, runnable_path ); return Ok(DispatchResult::default()); } + // Lineage propagated to any subscriber pushed from this producer: the + // ancestors that already ran, plus this producer. + let mut next_chain = chain.clone(); + next_chain.push(runnable_path.to_string()); let mut dispatched = Vec::new(); // Best-effort dispatch_event rows accumulated over the whole pass and @@ -287,6 +296,20 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result )); continue; } + // Cycle guard: a subscriber already in this producer's lineage + // would re-enter the chain (A→…→A), looping forever. Stop only + // this edge — sibling branches still dispatch, and acyclic chains + // of any depth are unaffected. + if chain.iter().any(|p| p == &sub_path) { + events.push(EventRow::new( + &sub_path, + asset_kind, + &asset_path, + DispatchOutcome::Skipped, + EventOptions { reason: Some("cycle_detected"), ..Default::default() }, + )); + continue; + } if join_all { match crate::cascade::handle_join( db, @@ -336,7 +359,7 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result asset_kind, &asset_path, runnable_path, - depth + 1, + &next_chain, partition.as_deref(), debounce_s, retry_count, @@ -416,11 +439,11 @@ fn read_skip_arg(args: Option<&HashMap>>) -> bool { .unwrap_or(false) } -fn read_chain_depth(trigger_map: Option<&HashMap>>) -> i64 { +fn read_chain(trigger_map: Option<&HashMap>>) -> Vec { trigger_map - .and_then(|m| m.get(CHAIN_DEPTH_KEY)) - .and_then(|v| serde_json::from_str::(v.get()).ok()) - .unwrap_or(0) + .and_then(|m| m.get(CHAIN_KEY)) + .and_then(|v| serde_json::from_str::>(v.get()).ok()) + .unwrap_or_default() } /// The partition value the producer ran with, if any. Resolved once at the @@ -561,7 +584,7 @@ async fn push_subscriber( asset_kind: AssetKind, asset_path: &str, producer_path: &str, - depth: i64, + chain: &[String], partition: Option<&str>, debounce_s: Option, retry_count: Option, @@ -656,7 +679,7 @@ async fn push_subscriber( "asset_path": asset_path, "producer_path": producer_path, "producer_job_id": producer.id.to_string(), - CHAIN_DEPTH_KEY: depth, + CHAIN_KEY: chain, PARTITION_ARG: partition, }); args.insert(TRIGGER_ARG.to_string(), to_raw_value(&trigger_payload)); diff --git a/frontend/src/lib/components/runs/DispatchEventsTable.svelte b/frontend/src/lib/components/runs/DispatchEventsTable.svelte index b095c9bda7..828437fee6 100644 --- a/frontend/src/lib/components/runs/DispatchEventsTable.svelte +++ b/frontend/src/lib/components/runs/DispatchEventsTable.svelte @@ -18,6 +18,8 @@ return 'reference input — only {partition}-bearing edges advance the join' case 'case3_missing_partition': return 'producer ran without a resolved partition' + case 'cycle_detected': + return 'subscriber already ran upstream in this chain (cycle)' default: return reason ?? '' }