starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-23 16:00:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(workspace-dependencies): implement better caching

Browse Source 
Signed-off-by: pyranota <pyra@duck.com>
This commit is contained in:
pyranota
2025-12-02 12:34:31 +01:00
parent 16ef690cac
commit 2eeb568966
4 changed files with 222 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/workspace_dependencies.rs
+21
View File
@@ -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(
backend/windmill-common/src/cache.rs
+82 -1
View File
@@ -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<Output = error::Result<WorkspaceDependencies>> + '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<bool> {
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<Self> {
@@ -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")]
backend/windmill-common/src/workspace_dependencies.rs
+103 -61
View File
@@ -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<WorkspaceDependencies>, cached_at timestamp)
static ref WORKSPACE_DEPENDENCIES_CACHE: quick_cache::sync::Cache<(String, ScriptLang, Option<String>), (Option<WorkspaceDependencies>, 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<String>,
language: ScriptLang,
workspace_id: &str,
e: impl PgExecutor<'c>,
) -> error::Result<Option<i64>> {
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<String>,
@@ -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::<Option<WorkspaceDependencies>>(&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::<Option<WorkspaceDependencies>>(&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<Option<Self>> {
) -> error::Result<Self> {
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)
}
backend/windmill-worker/src/workspace_dependencies.rs
+16 -21
View File
@@ -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(