starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-20 16:03:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: replace asset-cascade depth cap with cycle detection

Browse Source 
The hardcoded MAX_CHAIN_DEPTH=5 truncated legitimate deep pipelines (silently — the check returned before event logging). Replace it with per-edge cycle detection: carry the producer lineage in trigger.chain and skip only a subscriber already in the chain, recording a visible cycle_detected dispatch_event. Acyclic pipelines of any depth now cascade fully; a high MAX_CHAIN_LEN backstop guards against runaway. Tests + UI label updated; 8/8 pass at --test-threads=10.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-06-18 15:41:16 +00:00
co-authored by Claude Opus 4.8
parent 21d1ff0416
commit b31f37ef47
3 changed files with 82 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/asset_trigger_dispatch.rs
+34 -24
View File
@@ -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<Postgres>, args: serde_json::Value) -> anyhow::Result<Uuid> {
let id = Uuid::new_v4();
@@ -283,8 +283,8 @@ async fn clear_dispatched(db: &Pool<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>,
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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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;
backend/windmill-queue/src/asset_dispatch.rs
+46 -23
View File
@@ -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<DispatchResult>
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::<HashMap<String, Box<RawValue>>>(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<DispatchResult>
));
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<DispatchResult>
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<String, Box<RawValue>>>) -> bool {
.unwrap_or(false)
}
fn read_chain_depth(trigger_map: Option<&HashMap<String, Box<RawValue>>>) -> i64 {
fn read_chain(trigger_map: Option<&HashMap<String, Box<RawValue>>>) -> Vec<String> {
trigger_map
.and_then(|m| m.get(CHAIN_DEPTH_KEY))
.and_then(|v| serde_json::from_str::<i64>(v.get()).ok())
.unwrap_or(0)
.and_then(|m| m.get(CHAIN_KEY))
.and_then(|v| serde_json::from_str::<Vec<String>>(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<i32>,
retry_count: Option<i16>,
@@ -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));
frontend/src/lib/components/runs/DispatchEventsTable.svelte
+2
View File
@@ -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 ?? ''
}