From 124ccf5509c7f2d0029ddebde4595567a57f22ab Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 16 May 2026 19:53:22 +0000 Subject: [PATCH] feat: AND-join barrier for partitioned pipeline subscribers Stage D: a // trigger all subscriber no longer fires on any input. New join_pending_inputs slot table keyed (workspace, subscriber, partition); fetch_subscribers now returns join_all and the dispatch loop records each partition-bearing input arrival, pushing the subscriber once only when every partition-bearing input it declares is present for that partition. Per-partition slots, cleared on fire (re-accumulate, no double-fire), skew-immune (unlike debounce). Case-3 guard: an unpartitioned producer or a reference (non-{partition}) input never fires a partitioned join. Integration test covers wait/fire/isolation/no-double-fire. Co-Authored-By: Claude Opus 4.7 --- ...0bbc18da7e1ed758cab017fba2598c9467b7f.json | 17 +++ ...de10b1ca1785b66a68a260d4aaf2957fbcc66.json | 24 ++++ ...2bfc65f9786616e8f86570f7b91764e8b7c57.json | 23 --- ...d993e37d8a8bc67cd90682a7971760442b90a.json | 24 ++++ ...aeebc22883bf1d10dcbce41cccebb392ffd69.json | 16 +++ ...f13b151e845593d5a7ad63ffe433d3da60510.json | 29 ++++ ...0260516194247_join_pending_inputs.down.sql | 1 + .../20260516194247_join_pending_inputs.up.sql | 19 +++ backend/tests/asset_trigger_dispatch.rs | 108 ++++++++++++++ backend/windmill-common/src/assets.rs | 2 +- backend/windmill-queue/src/asset_dispatch.rs | 132 +++++++++++++++++- 11 files changed, 366 insertions(+), 29 deletions(-) create mode 100644 backend/.sqlx/query-0e7fe0e1d7aa2072a3431d081080bbc18da7e1ed758cab017fba2598c9467b7f.json create mode 100644 backend/.sqlx/query-3c5a387c2fed905838b0c1d2e0ade10b1ca1785b66a68a260d4aaf2957fbcc66.json delete mode 100644 backend/.sqlx/query-3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57.json create mode 100644 backend/.sqlx/query-3eb137e83c0aa6389b2893d59acd993e37d8a8bc67cd90682a7971760442b90a.json create mode 100644 backend/.sqlx/query-80f2d2f20e93b5e05ecd1fe5afeaeebc22883bf1d10dcbce41cccebb392ffd69.json create mode 100644 backend/.sqlx/query-e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510.json create mode 100644 backend/migrations/20260516194247_join_pending_inputs.down.sql create mode 100644 backend/migrations/20260516194247_join_pending_inputs.up.sql diff --git a/backend/.sqlx/query-0e7fe0e1d7aa2072a3431d081080bbc18da7e1ed758cab017fba2598c9467b7f.json b/backend/.sqlx/query-0e7fe0e1d7aa2072a3431d081080bbc18da7e1ed758cab017fba2598c9467b7f.json new file mode 100644 index 0000000000..e1386cf4d6 --- /dev/null +++ b/backend/.sqlx/query-0e7fe0e1d7aa2072a3431d081080bbc18da7e1ed758cab017fba2598c9467b7f.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO join_pending_inputs\n (workspace_id, subscriber_path, partition, trigger_ref)\n VALUES ($1, $2, $3, $4)\n ON CONFLICT DO NOTHING", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "0e7fe0e1d7aa2072a3431d081080bbc18da7e1ed758cab017fba2598c9467b7f" +} diff --git a/backend/.sqlx/query-3c5a387c2fed905838b0c1d2e0ade10b1ca1785b66a68a260d4aaf2957fbcc66.json b/backend/.sqlx/query-3c5a387c2fed905838b0c1d2e0ade10b1ca1785b66a68a260d4aaf2957fbcc66.json new file mode 100644 index 0000000000..71770359cc --- /dev/null +++ b/backend/.sqlx/query-3c5a387c2fed905838b0c1d2e0ade10b1ca1785b66a68a260d4aaf2957fbcc66.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT count(DISTINCT trigger_ref) AS \"n!\"\n FROM join_pending_inputs\n WHERE workspace_id = $1 AND subscriber_path = $2 AND partition = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "n!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "3c5a387c2fed905838b0c1d2e0ade10b1ca1785b66a68a260d4aaf2957fbcc66" +} diff --git a/backend/.sqlx/query-3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57.json b/backend/.sqlx/query-3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57.json deleted file mode 100644 index a236063636..0000000000 --- a/backend/.sqlx/query-3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "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-3eb137e83c0aa6389b2893d59acd993e37d8a8bc67cd90682a7971760442b90a.json b/backend/.sqlx/query-3eb137e83c0aa6389b2893d59acd993e37d8a8bc67cd90682a7971760442b90a.json new file mode 100644 index 0000000000..9b65eb8a18 --- /dev/null +++ b/backend/.sqlx/query-3eb137e83c0aa6389b2893d59acd993e37d8a8bc67cd90682a7971760442b90a.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT count(DISTINCT trigger_ref) AS \"n!\"\n FROM script_trigger\n WHERE workspace_id = $1\n AND runnable_path = $2\n AND trigger_kind = 'asset'\n AND runnable_kind = 'script'\n AND trigger_ref LIKE '%' || $3 || '%'", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "n!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "3eb137e83c0aa6389b2893d59acd993e37d8a8bc67cd90682a7971760442b90a" +} diff --git a/backend/.sqlx/query-80f2d2f20e93b5e05ecd1fe5afeaeebc22883bf1d10dcbce41cccebb392ffd69.json b/backend/.sqlx/query-80f2d2f20e93b5e05ecd1fe5afeaeebc22883bf1d10dcbce41cccebb392ffd69.json new file mode 100644 index 0000000000..f0f7df5289 --- /dev/null +++ b/backend/.sqlx/query-80f2d2f20e93b5e05ecd1fe5afeaeebc22883bf1d10dcbce41cccebb392ffd69.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM join_pending_inputs\n WHERE workspace_id = $1 AND subscriber_path = $2 AND partition = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "80f2d2f20e93b5e05ecd1fe5afeaeebc22883bf1d10dcbce41cccebb392ffd69" +} diff --git a/backend/.sqlx/query-e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510.json b/backend/.sqlx/query-e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510.json new file mode 100644 index 0000000000..ca6a1683e1 --- /dev/null +++ b/backend/.sqlx/query-e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT runnable_path AS \"runnable_path!\", join_all AS \"join_all!\"\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" + }, + { + "ordinal": 1, + "name": "join_all!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "e7a82dee0433e1116e749a5e81cf13b151e845593d5a7ad63ffe433d3da60510" +} diff --git a/backend/migrations/20260516194247_join_pending_inputs.down.sql b/backend/migrations/20260516194247_join_pending_inputs.down.sql new file mode 100644 index 0000000000..6e67e88e1b --- /dev/null +++ b/backend/migrations/20260516194247_join_pending_inputs.down.sql @@ -0,0 +1 @@ +DROP TABLE join_pending_inputs; diff --git a/backend/migrations/20260516194247_join_pending_inputs.up.sql b/backend/migrations/20260516194247_join_pending_inputs.up.sql new file mode 100644 index 0000000000..cbda6dd16e --- /dev/null +++ b/backend/migrations/20260516194247_join_pending_inputs.up.sql @@ -0,0 +1,19 @@ +-- AND-join barrier slot state. For a `// trigger all` subscriber, each +-- partition-bearing input arrival (an `// on` asset whose declared path +-- contains the `{partition}` token) is recorded against the +-- (subscriber, partition) slot. The subscriber is dispatched once, for a +-- given partition, only when every partition-bearing input it declares has +-- arrived for that partition — skew-immune (unlike a debounce). The slot +-- is cleared on fire so later writes re-accumulate and can re-materialize. +-- +-- trigger_ref stores the literal `{partition}`-token form (lineage is +-- partition-agnostic; the concrete value is the `partition` column), +-- matching how script_trigger / asset rows store it. +CREATE TABLE join_pending_inputs ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + subscriber_path VARCHAR(255) NOT NULL, + partition TEXT NOT NULL, + trigger_ref TEXT NOT NULL, + received_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (workspace_id, subscriber_path, partition, trigger_ref) +); diff --git a/backend/tests/asset_trigger_dispatch.rs b/backend/tests/asset_trigger_dispatch.rs index b75926cdbb..dad3ad013d 100644 --- a/backend/tests/asset_trigger_dispatch.rs +++ b/backend/tests/asset_trigger_dispatch.rs @@ -114,6 +114,48 @@ async fn seed_producer_job(db: &Pool, args: serde_json::Value) -> anyh Ok(id) } +/// Like `seed_producer_job` but for an arbitrary runnable path (the +/// AND-join test needs two distinct producers). +async fn seed_producer_job_path( + db: &Pool, + path: &str, + 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, + path, + args, + ) + .execute(db) + .await?; + Ok(id) +} + +/// Seed an asset subscription flagged as an AND join (`// trigger all`). +async fn seed_subscription_and( + db: &Pool, + subscriber_path: &str, + trigger_ref: &str, +) -> anyhow::Result<()> { + sqlx::query( + r#"INSERT INTO script_trigger + (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref, join_all) + VALUES ($1, 'script'::asset_usage_kind, $2, 'asset'::script_trigger_kind, $3, TRUE)"#, + ) + .bind(WS) + .bind(subscriber_path) + .bind(trigger_ref) + .execute(db) + .await?; + Ok(()) +} + fn make_mini(id: Uuid, runnable_path: &str) -> MiniCompletedJob { MiniCompletedJob { id, @@ -389,3 +431,69 @@ async fn partition_dynamic_resolved_persisted_and_propagated( Ok(()) } + +/// Stage D: an AND-join subscriber (`// trigger all`) with two +/// partition-bearing inputs must NOT dispatch until both inputs have +/// arrived for the *same* partition; then it fires exactly once. Slots +/// are per-partition and cleared on fire (re-accumulate, no double-fire). +#[sqlx::test(fixtures("base"))] +async fn and_join_waits_for_all_partition_inputs(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + + const PROD_A: &str = "u/test-user/prod-a"; + const PROD_B: &str = "u/test-user/prod-b"; + const SUB_J: &str = "u/test-user/sub-join"; + + seed_script(&db, SUB_J, "echo join-subscriber", "bash").await?; + // Two partition-bearing producers, one input each (literal token form). + seed_asset_write(&db, PROD_A, "s3object", "lake/{partition}/a").await?; + seed_asset_write(&db, PROD_B, "s3object", "lake/{partition}/b").await?; + seed_subscription_and(&db, SUB_J, "s3://lake/{partition}/a").await?; + seed_subscription_and(&db, SUB_J, "s3://lake/{partition}/b").await?; + + // Input A for partition "acme" → slot 1/2, must NOT dispatch. + let a_acme = seed_producer_job_path(&db, PROD_A, json!({ "partition": "acme" })).await?; + let r = dispatch_asset_triggers(&db, &make_mini(a_acme, PROD_A)).await; + assert!( + r.dispatched.is_empty(), + "AND join must wait: only 1 of 2 inputs present" + ); + assert!( + fetch_dispatched(&db).await?.is_empty(), + "no subscriber job pushed yet" + ); + + // A different partition for input B must open its OWN slot, not + // complete acme's. + let b_globex = seed_producer_job_path(&db, PROD_B, json!({ "partition": "globex" })).await?; + let r = dispatch_asset_triggers(&db, &make_mini(b_globex, PROD_B)).await; + assert!( + r.dispatched.is_empty(), + "different partition opens a separate slot, does not complete acme" + ); + + // Input B for "acme" → acme slot now 2/2 → dispatch exactly once. + let b_acme = seed_producer_job_path(&db, PROD_B, json!({ "partition": "acme" })).await?; + let r = dispatch_asset_triggers(&db, &make_mini(b_acme, PROD_B)).await; + assert_eq!(r.dispatched.len(), 1, "AND join fires once both inputs in"); + + let rows = fetch_dispatched(&db).await?; + assert_eq!(rows.len(), 1); + let sub = &rows[0]; + assert_eq!(sub.0, SUB_J); + let args = sub.2.as_ref().unwrap(); + assert_eq!(args["partition"], json!("acme")); + assert_eq!(args["trigger"]["partition"], json!("acme")); + + // Slot cleared on fire: re-arrival of A/acme alone is 1/2 again, no + // double-fire. + clear_dispatched(&db).await?; + let a_acme2 = seed_producer_job_path(&db, PROD_A, json!({ "partition": "acme" })).await?; + let r = dispatch_asset_triggers(&db, &make_mini(a_acme2, PROD_A)).await; + assert!( + r.dispatched.is_empty(), + "slot was cleared on fire; single input must not re-fire" + ); + + Ok(()) +} diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index daaeda87fc..d5bc0d7752 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -2,7 +2,7 @@ use sqlx::PgExecutor; use crate::{error, scripts::ScriptHash}; -pub use windmill_parser::asset_parser::{parse_pipeline_annotations, TriggerSpec}; +pub use windmill_parser::asset_parser::{parse_pipeline_annotations, TriggerSpec, PARTITION_TOKEN}; pub use windmill_types::assets::*; #[derive(sqlx::Type, Debug, Clone, Copy, PartialEq)] diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 442280a184..89bf2858a9 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -49,7 +49,7 @@ 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::assets::{parse_asset_trigger_ref, AssetKind, PARTITION_TOKEN}; use windmill_common::error::{self, Result}; use windmill_common::get_latest_hash_for_path; use windmill_common::jobs::{JobKind, JobPayload, JobTriggerKind}; @@ -130,10 +130,44 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result }; let trigger_ref = format!("{}{}", prefix, asset_path); let subs = fetch_subscribers(db, &job.workspace_id, &trigger_ref).await?; - for sub_path in subs { + for (sub_path, join_all) in subs { if sub_path == runnable_path { continue; } + if join_all { + // AND join barrier. Only a partition-bearing input + // (`// on …/{partition}/…`) carrying a concrete partition + // advances the join — a reference input or an + // unpartitioned producer must never fire a partitioned + // join (the case-3 silent-wrong guard). + if !is_partition_bearing_ref(&trigger_ref) { + tracing::debug!( + "AND subscriber {}: non-partition-bearing input {} does not fire the join", + sub_path, + trigger_ref + ); + continue; + } + let Some(pv) = partition.as_deref() else { + tracing::warn!( + "AND subscriber {}: partition-bearing input {} arrived with no resolved \ + partition; not dispatching (case-3 guard)", + sub_path, + trigger_ref + ); + continue; + }; + match record_and_check_join_slot(db, &job.workspace_id, &sub_path, pv, &trigger_ref) + .await + { + Ok(false) => continue, // slot incomplete — wait for the rest + Ok(true) => {} // all inputs present for this partition + Err(e) => { + tracing::error!("join-slot check failed for {}: {e:#}", sub_path); + continue; + } + } + } match push_subscriber( db, job, @@ -275,13 +309,14 @@ async fn fetch_subscribers( db: &Pool, workspace_id: &str, trigger_ref: &str, -) -> Result> { +) -> 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. + // `join_all` is the subscriber's `// trigger all` (AND join) flag. let rows = sqlx::query!( r#" - SELECT runnable_path AS "runnable_path!" + SELECT runnable_path AS "runnable_path!", join_all AS "join_all!" FROM script_trigger WHERE workspace_id = $1 AND trigger_kind = 'asset' @@ -301,7 +336,94 @@ async fn fetch_subscribers( trigger_ref ); } - Ok(rows.into_iter().map(|r| r.runnable_path).collect()) + Ok(rows + .into_iter() + .map(|r| (r.runnable_path, r.join_all)) + .collect()) +} + +/// A `// on ` whose stored ref carries the literal `{partition}` +/// token is partition-bearing — its concrete partition is the AND-join +/// key. Non-token asset refs are reference/presence-only inputs. +fn is_partition_bearing_ref(trigger_ref: &str) -> bool { + trigger_ref.contains(PARTITION_TOKEN) +} + +/// Distinct partition-bearing asset inputs an AND subscriber declares +/// (its `{partition}`-token `// on` lines). Reference inputs (no token) +/// and non-asset triggers are presence-only and do not gate the join in +/// v1, so they are excluded from the required set. +async fn count_required_join_inputs( + db: &DB, + workspace_id: &str, + subscriber_path: &str, +) -> Result { + let n = sqlx::query_scalar!( + r#"SELECT count(DISTINCT trigger_ref) AS "n!" + FROM script_trigger + WHERE workspace_id = $1 + AND runnable_path = $2 + AND trigger_kind = 'asset' + AND runnable_kind = 'script' + AND trigger_ref LIKE '%' || $3 || '%'"#, + workspace_id, + subscriber_path, + PARTITION_TOKEN, + ) + .fetch_one(db) + .await?; + Ok(n) +} + +/// Record an AND input arrival for `(subscriber, partition)` and report +/// whether every partition-bearing input is now present for that +/// partition. Idempotent per input (PK conflict ignored), so a re-fired +/// upstream doesn't double-count. On completion the slot is cleared so +/// later writes re-accumulate and can re-materialize the partition. +async fn record_and_check_join_slot( + db: &DB, + workspace_id: &str, + subscriber_path: &str, + partition: &str, + trigger_ref: &str, +) -> Result { + sqlx::query!( + r#"INSERT INTO join_pending_inputs + (workspace_id, subscriber_path, partition, trigger_ref) + VALUES ($1, $2, $3, $4) + ON CONFLICT DO NOTHING"#, + workspace_id, + subscriber_path, + partition, + trigger_ref, + ) + .execute(db) + .await?; + let required = count_required_join_inputs(db, workspace_id, subscriber_path).await?; + let received = sqlx::query_scalar!( + r#"SELECT count(DISTINCT trigger_ref) AS "n!" + FROM join_pending_inputs + WHERE workspace_id = $1 AND subscriber_path = $2 AND partition = $3"#, + workspace_id, + subscriber_path, + partition, + ) + .fetch_one(db) + .await?; + if required > 0 && received >= required { + sqlx::query!( + r#"DELETE FROM join_pending_inputs + WHERE workspace_id = $1 AND subscriber_path = $2 AND partition = $3"#, + workspace_id, + subscriber_path, + partition, + ) + .execute(db) + .await?; + Ok(true) + } else { + Ok(false) + } } async fn push_subscriber(