diff --git a/backend/tests/asset_trigger_dispatch.rs b/backend/tests/asset_trigger_dispatch.rs index 894a98e353..e83f6c9fd7 100644 --- a/backend/tests/asset_trigger_dispatch.rs +++ b/backend/tests/asset_trigger_dispatch.rs @@ -75,12 +75,13 @@ async fn seed_asset_write( .bind(producer_path) .execute(db) .await?; - // The dispatcher caches the per-workspace producer set, normally - // invalidated at deploy via the notify_event poller. These tests seed - // `asset` rows directly and run no poller, so invalidate here to mirror - // what a deploy would do — otherwise a stale cache hides freshly-seeded - // producers and nothing dispatches. - windmill_queue::asset_dispatch::ASSET_PRODUCER_WRITES_CACHE.remove(WS); + // These tests use #[sqlx::test] isolated DBs that all share one workspace + // id, so the process-global producer cache (keyed by workspace) would + // clobber across DBs under concurrent test threads. Disable it so every + // dispatch reads the test's own DB. (Production invalidates via the + // notify_event poller instead.) + windmill_queue::asset_dispatch::ASSET_PRODUCER_CACHE_DISABLED + .store(true, std::sync::atomic::Ordering::Relaxed); Ok(()) } diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 1238c726df..4ce41747b2 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -457,6 +457,14 @@ lazy_static::lazy_static! { quick_cache::sync::Cache::new(1000); } +/// Test hook: disables the producer-writes cache so every dispatch reads the +/// current DB. Integration tests use `#[sqlx::test]` isolated DBs that all +/// share one workspace id, so a process-global cache keyed by workspace would +/// clobber across DBs under concurrent test threads. Always `false` in +/// production (the cache is invalidated via the notify_event poller instead). +pub static ASSET_PRODUCER_CACHE_DISABLED: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(false); + /// Load (cached) the producer→writes map for a workspace. The single load /// query replaces the per-completion producer lookup; once cached, every /// completion in the workspace is served from memory until invalidation. @@ -464,8 +472,11 @@ async fn workspace_producer_writes( db: &Pool, workspace_id: &str, ) -> Result>>> { - if let Some(map) = ASSET_PRODUCER_WRITES_CACHE.get(workspace_id) { - return Ok(map); + let use_cache = !ASSET_PRODUCER_CACHE_DISABLED.load(std::sync::atomic::Ordering::Relaxed); + if use_cache { + if let Some(map) = ASSET_PRODUCER_WRITES_CACHE.get(workspace_id) { + return Ok(map); + } } let rows = sqlx::query!( r#" @@ -487,7 +498,9 @@ async fn workspace_producer_writes( map.entry(r.usage_path).or_default().push((r.kind, r.path)); } let map = Arc::new(map); - ASSET_PRODUCER_WRITES_CACHE.insert(workspace_id.to_string(), map.clone()); + if use_cache { + ASSET_PRODUCER_WRITES_CACHE.insert(workspace_id.to_string(), map.clone()); + } Ok(map) }