feat: reap abandoned AND-join slots after a TTL (default 60d, per-slot)

join_pending_inputs slots are normally cleared when the join fires;
partial slots whose inputs never all arrive (upstream removed/renamed,
one-off dynamic partition key, permanent skew) would otherwise leak.
windmill_queue::asset_dispatch::reap_stale_join_slots, called from the
monitor's delete_expired_items loop, deletes a (workspace, subscriber,
partition) slot only when its MOST RECENT row is older than
JOIN_SLOT_TTL_SECS (60d) — per-slot, never per-row, so a legitimately
slow join is not corrupted mid-accumulation. Conservative default;
per-join configurable TTL via the annotation is a planned follow-up.
Test covers stale-reaped / fresh-kept / mixed-slot-kept.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

backend/.sqlx/query-3c8a2389c47131ed89ec9069b2ebe15b103cf1344b7c84069e615181508913e9.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM join_pending_inputs jpi\n         USING (\n             SELECT workspace_id, subscriber_path, partition\n             FROM join_pending_inputs\n             GROUP BY workspace_id, subscriber_path, partition\n             HAVING max(received_at) <= now() - ($1::bigint::text || ' s')::interval\n         ) stale\n         WHERE jpi.workspace_id = stale.workspace_id\n           AND jpi.subscriber_path = stale.subscriber_path\n           AND jpi.partition = stale.partition",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "3c8a2389c47131ed89ec9069b2ebe15b103cf1344b7c84069e615181508913e9"
+}

backend/src/monitor.rs

@@ -1133,6 +1133,10 @@ pub async fn delete_expired_items(db: &DB) -> () {
         tracing::error!("Error deleting autoscaling event on CE: {:?}", e);
     }
 
+    if let Err(e) = windmill_queue::asset_dispatch::reap_stale_join_slots(db).await {
+        tracing::error!("Error reaping stale join_pending_inputs slots: {:?}", e);
+    }
+
     match sqlx::query_scalar!(
         "DELETE FROM agent_token_blacklist WHERE expires_at <= now() RETURNING token",
     )

backend/tests/asset_trigger_dispatch.rs

@@ -14,7 +14,7 @@ use uuid::Uuid;
 use windmill_common::jobs::{JobKind, JobPayload};
 use windmill_common::runnable_settings::prefetch_cached_from_handle;
 use windmill_common::scripts::{ScriptHash, ScriptLang};
-use windmill_queue::asset_dispatch::dispatch_asset_triggers;
+use windmill_queue::asset_dispatch::{dispatch_asset_triggers, reap_stale_join_slots};
 use windmill_queue::MiniCompletedJob;
 use windmill_test_utils::{initialize_tracing, ApiServer, RunJob};
 
@@ -767,3 +767,111 @@ async fn fuller_partitioned_join_multihop_pipeline(db: Pool<Postgres>) -> anyhow
 
     Ok(())
 }
+
+/// The TTL reaper deletes abandoned AND-join slots, but keyed on the
+/// slot's MOST RECENT row: a slot still receiving input (newest row
+/// fresh) is never reaped even if it also has rows older than the TTL.
+/// This per-slot (not per-row) property is the correctness point — it
+/// prevents corrupting a join whose inputs trickle in slowly.
+#[sqlx::test(fixtures("base"))]
+async fn reaper_clears_only_stale_join_slots(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    // Seed a join_pending_inputs row with an explicit age (days old).
+    async fn seed_slot_row(
+        db: &Pool<Postgres>,
+        sub: &str,
+        part: &str,
+        tref: &str,
+        age_days: i64,
+    ) -> anyhow::Result<()> {
+        sqlx::query(
+            r#"INSERT INTO join_pending_inputs
+                 (workspace_id, subscriber_path, partition, trigger_ref, received_at)
+               VALUES ($1, $2, $3, $4, now() - ($5::bigint::text || ' d')::interval)"#,
+        )
+        .bind(WS)
+        .bind(sub)
+        .bind(part)
+        .bind(tref)
+        .bind(age_days)
+        .execute(db)
+        .await?;
+        Ok(())
+    }
+    async fn slot_count(db: &Pool<Postgres>, sub: &str) -> anyhow::Result<i64> {
+        Ok(sqlx::query_scalar!(
+            r#"SELECT count(*) AS "n!" FROM join_pending_inputs
+               WHERE workspace_id = $1 AND subscriber_path = $2"#,
+            WS,
+            sub,
+        )
+        .fetch_one(db)
+        .await?)
+    }
+
+    // Stale: every row older than the 60d TTL → reaped.
+    seed_slot_row(
+        &db,
+        "u/test-user/sub-stale",
+        "p1",
+        "s3://x/{partition}/a",
+        61,
+    )
+    .await?;
+    seed_slot_row(
+        &db,
+        "u/test-user/sub-stale",
+        "p1",
+        "s3://x/{partition}/b",
+        90,
+    )
+    .await?;
+    // Fresh: recent → kept.
+    seed_slot_row(
+        &db,
+        "u/test-user/sub-fresh",
+        "p1",
+        "s3://y/{partition}/a",
+        0,
+    )
+    .await?;
+    // Mixed: one ancient row + one fresh row in the SAME slot. max(received_at)
+    // is fresh, so the whole slot must be kept (the correctness property).
+    seed_slot_row(
+        &db,
+        "u/test-user/sub-mixed",
+        "p1",
+        "s3://z/{partition}/a",
+        120,
+    )
+    .await?;
+    seed_slot_row(
+        &db,
+        "u/test-user/sub-mixed",
+        "p1",
+        "s3://z/{partition}/b",
+        0,
+    )
+    .await?;
+
+    reap_stale_join_slots(&db).await?;
+
+    assert_eq!(
+        slot_count(&db, "u/test-user/sub-stale").await?,
+        0,
+        "stale slot reaped"
+    );
+    assert_eq!(
+        slot_count(&db, "u/test-user/sub-fresh").await?,
+        1,
+        "fresh slot kept"
+    );
+    assert_eq!(
+        slot_count(&db, "u/test-user/sub-mixed").await?,
+        2,
+        "slot with a recent row must be kept entirely (per-slot, not per-row)"
+    );
+
+    Ok(())
+}

backend/windmill-queue/src/asset_dispatch.rs

@@ -435,6 +435,37 @@ async fn record_and_check_join_slot(
     Ok(fire)
 }
 
+/// Default time a partial AND-join slot may sit with no new input before
+/// it is abandoned. A slot is normally cleared the moment the join fires;
+/// this only reaps slots whose inputs never all arrive (an upstream was
+/// removed/renamed, a one-off `dynamic` partition key, permanent skew).
+/// Conservative so a legitimately slow join is never reaped; a per-join
+/// configurable TTL via the annotation is a planned follow-up.
+pub const JOIN_SLOT_TTL_SECS: i64 = 60 * 24 * 60 * 60; // 60 days
+
+/// Reap abandoned AND-join slots. Keyed on the *slot's most recent row*
+/// (`HAVING max(received_at)`), never per-row, so a join whose inputs
+/// trickle in over a window longer than one input's age is not corrupted
+/// mid-accumulation. Called periodically from the monitor loop.
+pub async fn reap_stale_join_slots(db: &DB) -> Result<()> {
+    sqlx::query!(
+        "DELETE FROM join_pending_inputs jpi
+         USING (
+             SELECT workspace_id, subscriber_path, partition
+             FROM join_pending_inputs
+             GROUP BY workspace_id, subscriber_path, partition
+             HAVING max(received_at) <= now() - ($1::bigint::text || ' s')::interval
+         ) stale
+         WHERE jpi.workspace_id = stale.workspace_id
+           AND jpi.subscriber_path = stale.subscriber_path
+           AND jpi.partition = stale.partition",
+        JOIN_SLOT_TTL_SECS,
+    )
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
 async fn push_subscriber(
     db: &DB,
     producer: &MiniCompletedJob,