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 <noreply@anthropic.com>

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"
+}

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"
+}

backend/.sqlx/query-3d66da60d4da1663bacce789d2b2bfc65f9786616e8f86570f7b91764e8b7c57.json

@@ -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"
-}

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"
+}

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"
+}

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"
+}

backend/migrations/20260516194247_join_pending_inputs.down.sql

@@ -0,0 +1 @@
+DROP TABLE join_pending_inputs;

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)
+);

backend/tests/asset_trigger_dispatch.rs

@@ -114,6 +114,48 @@ async fn seed_producer_job(db: &Pool<Postgres>, 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<Postgres>,
+    path: &str,
+    args: serde_json::Value,
+) -> anyhow::Result<Uuid> {
+    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<Postgres>,
+    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<Postgres>) -> 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(())
+}

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)]

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<DispatchResult>
         };
         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<Postgres>,
     workspace_id: &str,
     trigger_ref: &str,
-) -> Result<Vec<String>> {
+) -> Result<Vec<(String, bool)>> {
     // 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 <asset>` 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<i64> {
+    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<bool> {
+    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(