From 1404484f3887aec1044e5217f84bc7af8763a376 Mon Sep 17 00:00:00 2001 From: Pyra <92104930+pyranota@users.noreply.github.com> Date: Tue, 2 Dec 2025 13:48:14 +0100 Subject: [PATCH] fix(workspace-dependencies): implement better caching (#7273) * fix(workspace-dependencies): implement better caching Signed-off-by: pyranota * Update SQLx metadata * nits Signed-off-by: pyranota * spawn non blocking Signed-off-by: pyranota * nit Signed-off-by: pyranota * fix ci Signed-off-by: pyranota * fix issue Signed-off-by: pyranota --------- Signed-off-by: pyranota Co-authored-by: windmill-internal-app[bot] --- ...9649d0f5db125776fbdffea993a53a09708de.json | 55 ++++++ ...80766003d79884d33f94a79e61f0259807dbc.json | 128 ------------- backend/tests/workspace_dependencies.rs | 4 +- .../src/workspace_dependencies.rs | 73 ++++---- backend/windmill-common/src/cache.rs | 83 ++++++++- .../src/workspace_dependencies.rs | 174 ++++++++++++------ .../src/workspace_dependencies.rs | 124 ++++++++----- 7 files changed, 366 insertions(+), 275 deletions(-) create mode 100644 backend/.sqlx/query-5387fbfd4be55674dcbe9f9b3ce9649d0f5db125776fbdffea993a53a09708de.json delete mode 100644 backend/.sqlx/query-f0e7d28be69c4b922b34b76abb780766003d79884d33f94a79e61f0259807dbc.json diff --git a/backend/.sqlx/query-5387fbfd4be55674dcbe9f9b3ce9649d0f5db125776fbdffea993a53a09708de.json b/backend/.sqlx/query-5387fbfd4be55674dcbe9f9b3ce9649d0f5db125776fbdffea993a53a09708de.json new file mode 100644 index 0000000000..b014854b40 --- /dev/null +++ b/backend/.sqlx/query-5387fbfd4be55674dcbe9f9b3ce9649d0f5db125776fbdffea993a53a09708de.json @@ -0,0 +1,55 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT id FROM workspace_dependencies\n WHERE name IS NOT DISTINCT FROM $1 AND workspace_id = $2 AND archived = false AND language = $3\n LIMIT 1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby" + ] + } + } + } + ] + }, + "nullable": [ + false + ] + }, + "hash": "5387fbfd4be55674dcbe9f9b3ce9649d0f5db125776fbdffea993a53a09708de" +} diff --git a/backend/.sqlx/query-f0e7d28be69c4b922b34b76abb780766003d79884d33f94a79e61f0259807dbc.json b/backend/.sqlx/query-f0e7d28be69c4b922b34b76abb780766003d79884d33f94a79e61f0259807dbc.json deleted file mode 100644 index ab4421f9b4..0000000000 --- a/backend/.sqlx/query-f0e7d28be69c4b922b34b76abb780766003d79884d33f94a79e61f0259807dbc.json +++ /dev/null @@ -1,128 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT id, content, language AS \"language: ScriptLang\", name, description, archived, workspace_id, created_at\n FROM workspace_dependencies\n WHERE name IS NOT DISTINCT FROM $1 AND workspace_id = $2 AND archived = false AND language = $3\n LIMIT 1\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Int8" - }, - { - "ordinal": 1, - "name": "content", - "type_info": "Text" - }, - { - "ordinal": 2, - "name": "language: ScriptLang", - "type_info": { - "Custom": { - "name": "script_lang", - "kind": { - "Enum": [ - "python3", - "deno", - "go", - "bash", - "postgresql", - "nativets", - "bun", - "mysql", - "bigquery", - "snowflake", - "graphql", - "powershell", - "mssql", - "php", - "bunnative", - "rust", - "ansible", - "csharp", - "oracledb", - "nu", - "java", - "duckdb", - "ruby" - ] - } - } - } - }, - { - "ordinal": 3, - "name": "name", - "type_info": "Varchar" - }, - { - "ordinal": 4, - "name": "description", - "type_info": "Text" - }, - { - "ordinal": 5, - "name": "archived", - "type_info": "Bool" - }, - { - "ordinal": 6, - "name": "workspace_id", - "type_info": "Varchar" - }, - { - "ordinal": 7, - "name": "created_at", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - { - "Custom": { - "name": "script_lang", - "kind": { - "Enum": [ - "python3", - "deno", - "go", - "bash", - "postgresql", - "nativets", - "bun", - "mysql", - "bigquery", - "snowflake", - "graphql", - "powershell", - "mssql", - "php", - "bunnative", - "rust", - "ansible", - "csharp", - "oracledb", - "nu", - "java", - "duckdb", - "ruby" - ] - } - } - } - ] - }, - "nullable": [ - false, - false, - false, - true, - false, - false, - false, - false - ] - }, - "hash": "f0e7d28be69c4b922b34b76abb780766003d79884d33f94a79e61f0259807dbc" -} diff --git a/backend/tests/workspace_dependencies.rs b/backend/tests/workspace_dependencies.rs index 6875520125..b41eb7c550 100644 --- a/backend/tests/workspace_dependencies.rs +++ b/backend/tests/workspace_dependencies.rs @@ -64,7 +64,7 @@ mod workspace_dependencies { name: Some("test".to_owned()), description: None, } - .create("", "", "", db) + .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) .await .unwrap(); @@ -151,7 +151,7 @@ mod workspace_dependencies { name: None, // No name = default workspace dependencies description: None, } - .create("", "", "", db) + .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) .await .unwrap(); diff --git a/backend/windmill-api/src/workspace_dependencies.rs b/backend/windmill-api/src/workspace_dependencies.rs index bb511518d2..2330de200a 100644 --- a/backend/windmill-api/src/workspace_dependencies.rs +++ b/backend/windmill-api/src/workspace_dependencies.rs @@ -13,9 +13,8 @@ use windmill_common::{ workspace_dependencies::WorkspaceDependencies, DB, }; -use windmill_worker::{ - scoped_dependency_map, trigger_dependents_to_recompute_dependencies, - workspace_dependencies::NewWorkspaceDependencies, +use windmill_worker::workspace_dependencies::{ + trigger_dependents_to_recompute_dependencies_in_the_background, NewWorkspaceDependencies, }; use crate::db::ApiAuthed; @@ -43,10 +42,12 @@ async fn create( format!( "{}", nwd.create( - &authed.email, - &authed.username, - &username_to_permissioned_as(&authed.username), - &db + ( + authed.email, + username_to_permissioned_as(&authed.username), + authed.username, + ), + db ) .await? ), @@ -93,23 +94,20 @@ async fn archive( let db = &db; WorkspaceDependencies::archive(params.name.clone(), language, &w_id, db).await?; - trigger_dependents_to_recompute_dependencies( - &w_id, - scoped_dependency_map::ScopedDependencyMap::get_dependents( - WorkspaceDependencies::to_path(¶ms.name, language)?.as_str(), - &w_id, - db, - ) - .await?, - None, - None, - &authed.email, - &authed.username, - &username_to_permissioned_as(&authed.username), - db, - vec![], + trigger_dependents_to_recompute_dependencies_in_the_background( + params.name.is_none(), + w_id, + language, + ( + authed.email, + username_to_permissioned_as(&authed.username), + authed.username, + ), + WorkspaceDependencies::to_path(¶ms.name, language)?, + db.clone(), ) - .await + .await; + Ok(()) } #[axum::debug_handler] @@ -125,21 +123,18 @@ async fn delete( let db = &db; WorkspaceDependencies::delete(params.name.clone(), language, &w_id, db).await?; - trigger_dependents_to_recompute_dependencies( - &w_id, - scoped_dependency_map::ScopedDependencyMap::get_dependents( - WorkspaceDependencies::to_path(¶ms.name, language)?.as_str(), - &w_id, - db, - ) - .await?, - None, - None, - &authed.email, - &authed.username, - &username_to_permissioned_as(&authed.username), - db, - vec![], + trigger_dependents_to_recompute_dependencies_in_the_background( + params.name.is_none(), + w_id, + language, + ( + authed.email, + username_to_permissioned_as(&authed.username), + authed.username, + ), + WorkspaceDependencies::to_path(¶ms.name, language)?, + db.clone(), ) - .await + .await; + Ok(()) } diff --git a/backend/windmill-common/src/cache.rs b/backend/windmill-common/src/cache.rs index 8fef4b5e3a..3cfc98dc98 100644 --- a/backend/windmill-common/src/cache.rs +++ b/backend/windmill-common/src/cache.rs @@ -892,6 +892,86 @@ pub mod job { } } +pub mod workspace_dependencies { + use std::{ + future::Future, + time::{Duration, Instant}, + }; + + use crate::{error, scripts::ScriptLang, workspace_dependencies::WorkspaceDependencies, DB}; + + make_static! { + /// Workspace Dependencies by id and workspace cache. + static ref WORKSPACE_DEPENDENCIES: { (i64, String) => WorkspaceDependencies } in "workspace_dependencies" <= 1000; + } + lazy_static::lazy_static! { + /// Cache for checking if default/unnamed workspace dependencies exist for a workspace and language. + /// Cache key: (workspace_id, language) + /// Cache value: (exists: bool, cached_at timestamp) + static ref DEFAULT_WD_EXISTS_CACHE: quick_cache::sync::Cache<(String, ScriptLang), (bool, Instant)> = quick_cache::sync::Cache::new(500); + } + /// Cache timeout for existence checks (10 seconds) + pub const EXISTS_CACHE_TIMEOUT: Duration = Duration::from_secs(10); + + pub fn fetch_workspace_dependencies<'c>( + id: i64, + workspace_id: String, + db: &'c DB, + ) -> impl Future> + 'c { + tracing::debug!(workspace_id = %workspace_id, id, "fetching workspace dependencies"); + WORKSPACE_DEPENDENCIES.get_or_insert_async( + (id, workspace_id.clone()), + WorkspaceDependencies::get(id, workspace_id, db), + ) + } + + pub fn get_cached_is_unnamed_workspace_dependencies_exists<'c>( + language: ScriptLang, + workspace_id: String, + ) -> Option { + let exists_key = (workspace_id.to_string(), language); + if let Some((exists, cached_at)) = DEFAULT_WD_EXISTS_CACHE.get(&exists_key) { + if cached_at.elapsed() < EXISTS_CACHE_TIMEOUT { + tracing::debug!( + workspace_id = %workspace_id, + ?language, + exists, + "cache hit for unnamed workspace dependencies existence" + ); + return Some(exists); + } else { + tracing::debug!( + workspace_id = %workspace_id, + ?language, + "cache expired for unnamed workspace dependencies existence" + ); + DEFAULT_WD_EXISTS_CACHE.remove(&exists_key); + } + } else { + tracing::debug!( + workspace_id = %workspace_id, + ?language, + "cache miss for unnamed workspace dependencies existence" + ); + } + None + } + pub fn set_cached_is_unnamed_workspace_dependencies_exists<'c>( + language: ScriptLang, + workspace_id: String, + exists: bool, + ) { + tracing::debug!( + workspace_id = %workspace_id, + ?language, + exists, + "setting cache for unnamed workspace dependencies existence" + ); + let exists_key = (workspace_id.to_string(), language); + DEFAULT_WD_EXISTS_CACHE.insert(exists_key, (exists, Instant::now())); + } +} + const _: () = { impl Import for RawFlow { fn import(src: &impl Storage) -> error::Result { @@ -1075,7 +1155,8 @@ const _: () = { (ScriptHash, |x| format!("{:016x}", x.0)), ((u8, ScriptHash), |x| format!("{:02x}-{:016x}", x.0, x.1.0)), (FlowNodeId, |x| format!("{:016x}", x.0)), - (AppScriptId, |x| format!("{:016x}", x.0)) + (AppScriptId, |x| format!("{:016x}", x.0)), + ((i64, String), |x| format!("{}-{}", x.1, x.0)) } #[cfg(feature = "scoped_cache")] diff --git a/backend/windmill-common/src/workspace_dependencies.rs b/backend/windmill-common/src/workspace_dependencies.rs index c264f249f7..8060a6ac98 100644 --- a/backend/windmill-common/src/workspace_dependencies.rs +++ b/backend/windmill-common/src/workspace_dependencies.rs @@ -2,9 +2,17 @@ use itertools::Itertools; use regex::Regex; use serde::{Deserialize, Serialize}; use sqlx::PgExecutor; -use std::time::{Duration, Instant}; -use crate::{error, scripts::ScriptLang, utils::calculate_hash, worker::Connection}; +use crate::{ + cache::workspace_dependencies::{ + fetch_workspace_dependencies, get_cached_is_unnamed_workspace_dependencies_exists, + set_cached_is_unnamed_workspace_dependencies_exists, + }, + error, + scripts::ScriptLang, + utils::calculate_hash, + worker::Connection, +}; use phf::phf_set; pub static BLACKLIST: phf::Set<&'static str> = phf_set! { @@ -15,16 +23,8 @@ pub static BLACKLIST: phf::Set<&'static str> = phf_set! { lazy_static::lazy_static! { static ref WMDEBUG_FORCE_V0_WORKSPACE_DEPENDENCIES: bool = std::env::var("WMDEBUG_FORCE_V0_WORKSPACE_DEPENDENCIES").is_ok(); - - /// Simple in-memory cache for workspace dependencies get_latest with 10-second timeout. - /// Cache key: (workspace_id, language, name) - /// Cache value: (Option, cached_at timestamp) - static ref WORKSPACE_DEPENDENCIES_CACHE: quick_cache::sync::Cache<(String, ScriptLang, Option), (Option, Instant)> = quick_cache::sync::Cache::new(1000); } -/// Cache timeout for workspace dependencies -const CACHE_TIMEOUT: Duration = Duration::from_secs(10); - /// Minimum Windmill version required for workspace dependencies feature pub const MIN_VERSION_WORKSPACE_DEPENDENCIES: &str = "1.587.0"; @@ -174,6 +174,42 @@ impl WorkspaceDependencies { .map_err(error::Error::from) } + async fn get_latest_id<'c>( + name: Option, + language: ScriptLang, + workspace_id: &str, + e: impl PgExecutor<'c>, + ) -> error::Result> { + tracing::debug!( + workspace_id = %workspace_id, + ?language, + ?name, + "fetching latest workspace dependencies id" + ); + let result = sqlx::query_scalar!( + r#" + SELECT id FROM workspace_dependencies + WHERE name IS NOT DISTINCT FROM $1 AND workspace_id = $2 AND archived = false AND language = $3 + LIMIT 1 + "#, + name, + workspace_id, + language as ScriptLang + ) + .fetch_optional(e) + .await + .map_err(error::Error::from)?; + + tracing::debug!( + workspace_id = %workspace_id, + ?language, + ?name, + ?result, + "fetched latest workspace dependencies id" + ); + Ok(result) + } + /// Gets the latest version of workspace dependencies by name and language. pub async fn get_latest( name: Option, @@ -185,66 +221,84 @@ impl WorkspaceDependencies { return Ok(None); } - let cache_key = (workspace_id.to_string(), language, name.clone()); - - // Check if cached value is still valid - if let Some((cached_value, cached_at)) = WORKSPACE_DEPENDENCIES_CACHE.get(&cache_key) { - if cached_at.elapsed() < CACHE_TIMEOUT { - return Ok(cached_value); - } - // Expired, remove it - WORKSPACE_DEPENDENCIES_CACHE.remove(&cache_key); + if name.is_none() + && get_cached_is_unnamed_workspace_dependencies_exists( + language, + workspace_id.to_owned(), + ) + .map(|exists| exists == false) + .unwrap_or_default() + { + tracing::debug!( + workspace_id = %workspace_id, + ?language, + "skipping unnamed workspace dependencies fetch - cached as non-existent" + ); + return Ok(None); } - // Fetch and cache - let fetch = Box::pin(async { - match &conn { - Connection::Sql(db) => sqlx::query_as!( - Self, - r#" - SELECT id, content, language AS "language: ScriptLang", name, description, archived, workspace_id, created_at - FROM workspace_dependencies - WHERE name IS NOT DISTINCT FROM $1 AND workspace_id = $2 AND archived = false AND language = $3 - LIMIT 1 - "#, - name, - workspace_id, - language as ScriptLang - ) - .fetch_optional(db) - .await - .map_err(error::Error::from), + // Fetch from database or HTTP + let wd = match &conn { + Connection::Sql(db) => { + let Some(id) = + Self::get_latest_id(name.clone(), language, workspace_id, db).await? + else { + if name.is_none() { + set_cached_is_unnamed_workspace_dependencies_exists( + language, + workspace_id.to_owned(), + false, + ); + } - Connection::Http(http_client) => http_client - .get::>(&format!( - "/api/w/{workspace_id}/agent_workers/workspace_dependencies/get_latest/{}{}", - language.as_str(), - if let Some(ref name_val) = name { - format!("?name={name_val}") - } else { - "".to_owned() - } - )) - .await - .map_err(error::Error::from), + tracing::debug!( + workspace_id = %workspace_id, + ?language, + ?name, + "no latest workspace dependencies found" + ); + return Ok(None); + }; + tracing::debug!( + workspace_id = %workspace_id, + ?language, + ?name, + id, + "fetching workspace dependencies by id from cache or db" + ); + Some(fetch_workspace_dependencies(id, workspace_id.to_owned(), db).await?) } - }); - let (workspace_dependencies_o, ..) = WORKSPACE_DEPENDENCIES_CACHE - .get_or_insert_async(&cache_key, async { - Ok::<_, error::Error>((fetch.await?, Instant::now())) - }) - .await?; + Connection::Http(http_client) => http_client + .get::>(&format!( + "/api/w/{workspace_id}/agent_workers/workspace_dependencies/get_latest/{}{}", + language.as_str(), + if let Some(ref name_val) = name { + format!("?name={name_val}") + } else { + "".to_owned() + } + )) + .await + .map_err(error::Error::from)?, + }; - Ok(workspace_dependencies_o) + if name.is_none() { + set_cached_is_unnamed_workspace_dependencies_exists( + language, + workspace_id.to_owned(), + wd.is_some(), + ); + } + Ok(wd) } /// Gets workspace dependencies by their unique ID. pub async fn get<'c>( id: i64, - workspace_id: &str, + workspace_id: String, e: impl PgExecutor<'c>, - ) -> error::Result> { + ) -> error::Result { sqlx::query_as!( Self, r#" @@ -254,9 +308,9 @@ impl WorkspaceDependencies { LIMIT 1 "#, id, - workspace_id + &workspace_id ) - .fetch_optional(e) + .fetch_one(e) .await .map_err(error::Error::from) } diff --git a/backend/windmill-worker/src/workspace_dependencies.rs b/backend/windmill-worker/src/workspace_dependencies.rs index dd060122a7..281b08a32c 100644 --- a/backend/windmill-worker/src/workspace_dependencies.rs +++ b/backend/windmill-worker/src/workspace_dependencies.rs @@ -1,5 +1,8 @@ use serde::{Deserialize, Serialize}; -use windmill_common::{error, scripts::ScriptLang, workspace_dependencies::WorkspaceDependencies}; +use windmill_common::{ + cache::workspace_dependencies::EXISTS_CACHE_TIMEOUT, error, scripts::ScriptLang, + workspace_dependencies::WorkspaceDependencies, +}; use crate::{ scoped_dependency_map::ScopedDependencyMap, trigger_dependents_to_recompute_dependencies, @@ -25,10 +28,8 @@ impl NewWorkspaceDependencies { /// and rebuilds the dependency map if this is the first unnamed dependency for the workspace. pub async fn create<'c>( self, - email: &str, - created_by: &str, - permissioned_as: &str, - db: &sqlx::Pool, + metadata: (String, String, String), + db: sqlx::Pool, ) -> error::Result { // Check if all workers support workspace dependencies feature windmill_common::workspace_dependencies::min_version_supports_v0_workspace_dependencies() @@ -43,7 +44,7 @@ impl NewWorkspaceDependencies { let setting_name = format!("workspace_dependencies_map_rebuilt:{}", self.workspace_id); let already_rebuilt = windmill_common::global_settings::load_value_from_global_settings( - db, + &db, &setting_name, ) .await? @@ -54,11 +55,11 @@ impl NewWorkspaceDependencies { workspace_id = %self.workspace_id, "Rebuilding workspace dependencies map for first unnamed workspace dependencies" ); - ScopedDependencyMap::rebuild_map_unchecked(&self.workspace_id, db).await?; + ScopedDependencyMap::rebuild_map_unchecked(&self.workspace_id, &db).await?; // Mark as rebuilt by creating the setting windmill_common::global_settings::set_value_in_global_settings( - db, + &db, &setting_name, serde_json::json!({}), ) @@ -111,50 +112,83 @@ impl NewWorkspaceDependencies { .await?; tx.commit().await?; - // Make sure trigger dependents will have latest view. - // NOTE: Uncomment for tests - // #[cfg(test)] - // assert_eq!( - // sqlx::query_scalar!( - // " - // SELECT id FROM workspace_dependencies - // WHERE archived = false - // AND name IS NOT DISTINCT FROM $1 - // AND workspace_id = $2 - // AND language = $3 - // ", - // self.name, - // self.workspace_id, - // self.language as ScriptLang, - // ) - // .fetch_one(db) // Use db - // .await?, - // new_id - // ); - - // It's ok to fail, it will return an error and user will get notified that they should redeploy workspace dependencies - trigger_dependents_to_recompute_dependencies( - &self.workspace_id, - crate::scoped_dependency_map::ScopedDependencyMap::get_dependents( - path.as_str(), - &self.workspace_id, - db, - ) - .await?, - None, - None, - email, - created_by, - permissioned_as, + trigger_dependents_to_recompute_dependencies_in_the_background( + prev_description.is_none() && self.name.is_none(), + self.workspace_id, + self.language, + metadata, + path, db, - vec![], ) - .await?; + .await; Ok(new_id) } } +pub async fn trigger_dependents_to_recompute_dependencies_in_the_background( + wait_for_cache_timeout: bool, + workspace_id: String, + language: ScriptLang, + (email, permissioned_as, created_by): (String, String, String), + path: String, + db: sqlx::Pool, +) { + tokio::spawn(async move { + if wait_for_cache_timeout { + tracing::debug!( + workspace_id = %workspace_id, + language = ?language, + "waiting for cache timeout after creating first unnamed workspace dependencies" + ); + // Wait for cache timeout. + // For context, workers have cache on whether the unnamed workspace dependencies exists or not. + // when we trigger dependents to recompoute dependencies we want to make sure all workers are having cache timed out. + // otherwise it would result into bug, when workers skip fetch of workspace dependencies because they think they don't exist. + tokio::time::sleep(EXISTS_CACHE_TIMEOUT).await; + } + + // It's ok to fail, it will return an error and user will get notified that they should redeploy workspace dependencies + if let Err(e) = trigger_dependents_to_recompute_dependencies( + &workspace_id, + match crate::scoped_dependency_map::ScopedDependencyMap::get_dependents( + path.as_str(), + &workspace_id, + &db, + ) + .await + { + Ok(importers) => importers, + Err(e) => { + tracing::error!( + workspace_id = %workspace_id, + path = %path, + error = %e, + "CRITICAL: failed to get dependents for workspace dependencies - dependent runnables are not being redeployed. Please contact the Windmill team" + ); + return; + } + }, + None, + None, + email.as_str(), + created_by.as_str(), + permissioned_as.as_str(), + &db, + vec![], + ) + .await + { + tracing::error!( + workspace_id = %workspace_id, + path = %path, + error = %e, + "CRITICAL: failed to trigger dependents to recompute dependencies - dependent runnables are not being redeployed. Please contact the Windmill team" + ); + } + }); +} + // Type aliases for backward compatibility pub type RawRequirements = WorkspaceDependencies; pub type NewRawRequirements = NewWorkspaceDependencies;