From 2eeb5689669fd0081ab7f4d71d1ab01eb615a1bc Mon Sep 17 00:00:00 2001 From: pyranota Date: Tue, 2 Dec 2025 12:34:31 +0100 Subject: [PATCH] fix(workspace-dependencies): implement better caching Signed-off-by: pyranota --- .../src/workspace_dependencies.rs | 21 +++ backend/windmill-common/src/cache.rs | 83 ++++++++- .../src/workspace_dependencies.rs | 164 +++++++++++------- .../src/workspace_dependencies.rs | 37 ++-- 4 files changed, 222 insertions(+), 83 deletions(-) diff --git a/backend/windmill-api/src/workspace_dependencies.rs b/backend/windmill-api/src/workspace_dependencies.rs index bb511518d2..fb043be071 100644 --- a/backend/windmill-api/src/workspace_dependencies.rs +++ b/backend/windmill-api/src/workspace_dependencies.rs @@ -6,6 +6,7 @@ use axum::{ use http::StatusCode; use serde::Deserialize; use windmill_common::{ + cache::workspace_dependencies::EXISTS_CACHE_TIMEOUT, error::{self, JsonResult}, scripts::ScriptLang, users::username_to_permissioned_as, @@ -93,6 +94,16 @@ async fn archive( let db = &db; WorkspaceDependencies::archive(params.name.clone(), language, &w_id, db).await?; + if params.name.is_none() { + tracing::debug!( + workspace_id = %w_id, + ?language, + "waiting for cache timeout after archiving unnamed workspace dependencies" + ); + // for context read [[NewWorkspaceDependencies::create]] + tokio::time::sleep(EXISTS_CACHE_TIMEOUT).await; + } + trigger_dependents_to_recompute_dependencies( &w_id, scoped_dependency_map::ScopedDependencyMap::get_dependents( @@ -125,6 +136,16 @@ async fn delete( let db = &db; WorkspaceDependencies::delete(params.name.clone(), language, &w_id, db).await?; + if params.name.is_none() { + tracing::debug!( + workspace_id = %w_id, + ?language, + "waiting for cache timeout after deleting unnamed workspace dependencies" + ); + // for context read [[NewWorkspaceDependencies::create]] + tokio::time::sleep(EXISTS_CACHE_TIMEOUT).await; + } + trigger_dependents_to_recompute_dependencies( &w_id, scoped_dependency_map::ScopedDependencyMap::get_dependents( 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..77e334e911 100644 --- a/backend/windmill-common/src/workspace_dependencies.rs +++ b/backend/windmill-common/src/workspace_dependencies.rs @@ -2,9 +2,18 @@ use itertools::Itertools; use regex::Regex; use serde::{Deserialize, Serialize}; use sqlx::PgExecutor; -use std::time::{Duration, Instant}; +use std::time::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 +24,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 +175,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 +222,71 @@ 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() + { + 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), - - 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), + // 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 { + 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 +296,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..233b2368cd 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, @@ -111,26 +114,18 @@ 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 - // ); + if prev_description.is_none() && self.name.is_none() { + tracing::debug!( + workspace_id = %self.workspace_id, + language = ?self.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 trigger_dependents_to_recompute_dependencies(