diff --git a/backend/.sqlx/query-5c609ea0696df96ca02cba7fee359b785515a79dcda3745663e4c4f2cf328389.json b/backend/.sqlx/query-5c609ea0696df96ca02cba7fee359b785515a79dcda3745663e4c4f2cf328389.json new file mode 100644 index 0000000000..03de8472ad --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d.json b/backend/.sqlx/query-6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d.json new file mode 100644 index 0000000000..7f2b4a677a --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb.json b/backend/.sqlx/query-7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb.json new file mode 100644 index 0000000000..cc564952cb --- /dev/null +++ b/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" +} diff --git a/backend/src/main.rs b/backend/src/main.rs index a97320d5d9..57cc02598a 100644 --- a/backend/src/main.rs +++ b/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: {}", diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index ebdb5bfb54..4e241b4b99 100644 --- a/backend/windmill-common/src/assets.rs +++ b/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 ) diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index 49e278fdda..ce05becee0 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/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 _ => 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::(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>>> = + 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, workspace_id: &str, - runnable_path: &str, -) -> Result> { +) -> Result>>> { + 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> = 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