perf: gate asset dispatch on a cached per-workspace producer set

Cache the producer-path→writes map per workspace and invalidate it from the asset-clear paths via the notify_event polling system, so a top-level script/preview completion that isn't an asset producer costs an in-memory lookup instead of a per-completion query.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/.sqlx/query-5c609ea0696df96ca02cba7fee359b785515a79dcda3745663e4c4f2cf328389.json

@@ -0,0 +1,48 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT\n            usage_path AS \"usage_path!\",\n            kind AS \"kind!: AssetKind\",\n            path AS \"path!\"\n        FROM asset\n        WHERE workspace_id = $1\n          AND usage_kind = 'script'\n          AND usage_access_type IN ('w', 'rw')\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "usage_path!",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "kind!: AssetKind",
+        "type_info": {
+          "Custom": {
+            "name": "asset_kind",
+            "kind": {
+              "Enum": [
+                "s3object",
+                "resource",
+                "variable",
+                "ducklake",
+                "datatable",
+                "volume"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 2,
+        "name": "path!",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "5c609ea0696df96ca02cba7fee359b785515a79dcda3745663e4c4f2cf328389"
+}

backend/.sqlx/query-6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH del AS (\n             DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script'\n               AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1)\n           )\n           INSERT INTO notify_event (channel, payload)\n           VALUES ('notify_asset_producer_change', $1)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d"
+}

backend/.sqlx/query-7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb.json

@@ -0,0 +1,27 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH del AS (\n             DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3\n           )\n           INSERT INTO notify_event (channel, payload)\n           SELECT 'notify_asset_producer_change', $1 WHERE $3 = 'script'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        {
+          "Custom": {
+            "name": "asset_usage_kind",
+            "kind": {
+              "Enum": [
+                "script",
+                "flow",
+                "job"
+              ]
+            }
+          }
+        }
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb"
+}

backend/src/main.rs

@@ -1646,6 +1646,13 @@ async fn process_notify_event(
             );
             windmill_common::variables::CUSTOM_ENVS_CACHE.remove(payload);
         }
+        "notify_asset_producer_change" => {
+            tracing::debug!(
+                "Asset producer change for workspace {}, invalidating producer-writes cache",
+                payload
+            );
+            windmill_queue::asset_dispatch::ASSET_PRODUCER_WRITES_CACHE.remove(payload);
+        }
         "notify_workspace_key_change" => {
             tracing::info!(
                 "Workspace key change detected, invalidating workspace key cache: {}",

backend/windmill-common/src/assets.rs

@@ -56,8 +56,21 @@ pub async fn clear_static_asset_usage<'e>(
     usage_path: &str,
     usage_kind: AssetUsageKind,
 ) -> error::Result<()> {
+    // Invalidate the per-workspace producer-writes cache that gates the
+    // asset-trigger dispatch hook (windmill-queue
+    // asset_dispatch::ASSET_PRODUCER_WRITES_CACHE), but only for script
+    // usage. The notify is unconditional (not gated on rows deleted) so a
+    // deploy where the script *gains* its first asset still invalidates;
+    // emitting it in the same statement keeps it atomic with the change and
+    // visible to pollers only on commit. Flow usage doesn't affect the
+    // script producer set, so the INSERT is skipped via the `$3 = 'script'`
+    // guard.
     sqlx::query!(
-        r#"DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3"#,
+        r#"WITH del AS (
+             DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3
+           )
+           INSERT INTO notify_event (channel, payload)
+           SELECT 'notify_asset_producer_change', $1 WHERE $3 = 'script'"#,
         workspace_id,
         usage_path,
         usage_kind as AssetUsageKind
@@ -72,8 +85,16 @@ pub async fn clear_static_asset_usage_by_script_hash<'e>(
     workspace_id: &str,
     script_hash: ScriptHash,
 ) -> error::Result<()> {
+    // Always script usage → unconditionally invalidate the producer-writes
+    // cache for this workspace (see clear_static_asset_usage), atomically
+    // with the delete.
     sqlx::query!(
-        "DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script' AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1)",
+        r#"WITH del AS (
+             DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script'
+               AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1)
+           )
+           INSERT INTO notify_event (channel, payload)
+           VALUES ('notify_asset_producer_change', $1)"#,
         workspace_id,
         script_hash.0
     )

backend/windmill-queue/src/asset_dispatch.rs

@@ -49,6 +49,7 @@ use serde_json::value::RawValue;
 use sqlx::types::Json;
 use sqlx::{Pool, Postgres};
 use std::collections::HashMap;
+use std::sync::Arc;
 use uuid::Uuid;
 use windmill_common::assets::{AssetKind, PARTITION_TOKEN};
 use windmill_common::error::{self, Result};
@@ -288,14 +289,18 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result<DispatchResult>
         _ => return Ok(DispatchResult::default()),
     };
 
-    // Cheapest filter first: this hook runs on every top-level script
-    // completion, and the overwhelmingly common case is a script that
-    // writes no asset — one indexed query and out, before touching the
-    // job's args JSONB.
-    let writes = fetch_producer_writes(db, &job.workspace_id, runnable_path).await?;
-    if writes.is_empty() {
+    // Producer gate (cached): this hook fires on every top-level
+    // script/preview completion, and the overwhelmingly common case is a
+    // script that writes no asset. The per-workspace producer→writes map is
+    // cached and invalidated by a trigger on `asset`, so a non-producer
+    // completion costs one in-memory lookup and zero queries. The map is
+    // keyed on the deploy-time `asset` table by path, so an undeployed/new
+    // preview (no asset rows for its path) is a non-producer and never
+    // cascades — same as the previous per-completion lookup.
+    let producers = workspace_producer_writes(db, &job.workspace_id).await?;
+    let Some(writes) = producers.get(runnable_path).cloned() else {
         return Ok(DispatchResult::default());
-    }
+    };
 
     let args = fetch_args(db, &job.workspace_id, job.id).await?;
     if read_skip_arg(args.as_ref()) {
@@ -498,28 +503,55 @@ fn read_partition(
     serde_json::from_str::<String>(trigger_map?.get(PARTITION_ARG)?.get()).ok()
 }
 
-async fn fetch_producer_writes(
+lazy_static::lazy_static! {
+    /// Per-workspace map of producer script path → the assets it writes
+    /// (`usage_access_type IN ('w','rw')`). Serves both the producer gate
+    /// (is this path a producer?) and the writes themselves, so a completion
+    /// that isn't a producer costs a single in-memory lookup and zero
+    /// queries — the dispatch hook fires on every top-level script/preview
+    /// completion instance-wide, the overwhelming majority of which write no
+    /// asset. An empty map means the workspace has no asset producers (no
+    /// pipelines). Invalidated per workspace by `notify_asset_producer_change`
+    /// (a trigger on `asset`) through the polling notify system; until the
+    /// next poll a freshly-deployed producer may not cascade (sub-poll lag,
+    /// acceptable for a data pipeline).
+    pub static ref ASSET_PRODUCER_WRITES_CACHE:
+        quick_cache::sync::Cache<String, Arc<HashMap<String, Vec<(AssetKind, String)>>>> =
+        quick_cache::sync::Cache::new(1000);
+}
+
+/// 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.
+async fn workspace_producer_writes(
     db: &Pool<Postgres>,
     workspace_id: &str,
-    runnable_path: &str,
-) -> Result<Vec<(AssetKind, String)>> {
+) -> Result<Arc<HashMap<String, Vec<(AssetKind, String)>>>> {
+    if let Some(map) = ASSET_PRODUCER_WRITES_CACHE.get(workspace_id) {
+        return Ok(map);
+    }
     let rows = sqlx::query!(
         r#"
         SELECT
+            usage_path AS "usage_path!",
             kind AS "kind!: AssetKind",
             path AS "path!"
         FROM asset
         WHERE workspace_id = $1
           AND usage_kind = 'script'
-          AND usage_path = $2
           AND usage_access_type IN ('w', 'rw')
         "#,
         workspace_id,
-        runnable_path,
     )
     .fetch_all(db)
     .await?;
-    Ok(rows.into_iter().map(|r| (r.kind, r.path)).collect())
+    let mut map: HashMap<String, Vec<(AssetKind, String)>> = HashMap::new();
+    for r in rows {
+        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());
+    Ok(map)
 }
 
 /// A subscriber row resolved from `script_trigger`. Bundles the per-edge