test: disable producer cache in asset dispatch tests (isolated-DB safe)

The .remove(WS) approach still raced: #[sqlx::test] gives each test its own DB but they share one workspace id, so the WS-keyed process-global cache clobbered across DBs under concurrent threads. Add an ASSET_PRODUCER_CACHE_DISABLED test hook and set it in the tests so every dispatch reads its own DB. 8/8 pass at --test-threads=10. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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(())
 }
 

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<Postgres>,
     workspace_id: &str,
 ) -> Result<Arc<HashMap<String, Vec<(AssetKind, String)>>>> {
-    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)
 }