From bc67842f7ac63bb9d2db9d18e103e75f6821661e Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 17 May 2026 09:43:39 +0000 Subject: [PATCH] feat: reap abandoned AND-join slots after a TTL (default 60d, per-slot) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- ...be15b103cf1344b7c84069e615181508913e9.json | 14 +++ backend/src/monitor.rs | 4 + backend/tests/asset_trigger_dispatch.rs | 110 +++++++++++++++++- backend/windmill-queue/src/asset_dispatch.rs | 31 +++++ 4 files changed, 158 insertions(+), 1 deletion(-) create mode 100644 backend/.sqlx/query-3c8a2389c47131ed89ec9069b2ebe15b103cf1344b7c84069e615181508913e9.json diff --git a/backend/.sqlx/query-3c8a2389c47131ed89ec9069b2ebe15b103cf1344b7c84069e615181508913e9.json b/backend/.sqlx/query-3c8a2389c47131ed89ec9069b2ebe15b103cf1344b7c84069e615181508913e9.json new file mode 100644 index 0000000000..650876cc23 --- /dev/null +++ b/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" +} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index d5ea145cfc..8e5aea5e1b 100644 --- a/backend/src/monitor.rs +++ b/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", ) diff --git a/backend/tests/asset_trigger_dispatch.rs b/backend/tests/asset_trigger_dispatch.rs index d351060e0f..8c5b6eeb3c 100644 --- a/backend/tests/asset_trigger_dispatch.rs +++ b/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) -> 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) -> anyhow::Result<()> { + initialize_tracing().await; + + // Seed a join_pending_inputs row with an explicit age (days old). + async fn seed_slot_row( + db: &Pool, + 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, sub: &str) -> anyhow::Result { + 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(()) +} diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 38c6998d76..4db46ef933 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/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,