From 593da3213fba35fad4a284a4fb2671f0bc565a5b Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 2 Sep 2025 14:34:43 +0000 Subject: [PATCH] add variable caching api --- ...y_resource_cache_on_variable_change.up.sql | 25 +------- backend/src/main.rs | 2 - backend/windmill-api/openapi.yaml | 13 +++- backend/windmill-api/src/apps.rs | 1 + backend/windmill-api/src/resources.rs | 62 ++++++++----------- .../windmill-api/src/var_resource_cache.rs | 14 ++--- backend/windmill-api/src/variables.rs | 41 ++++++------ 7 files changed, 73 insertions(+), 85 deletions(-) diff --git a/backend/migrations/20250902085503_notify_resource_cache_on_variable_change.up.sql b/backend/migrations/20250902085503_notify_resource_cache_on_variable_change.up.sql index 1ee9d2d43b..1f3c0716a3 100644 --- a/backend/migrations/20250902085503_notify_resource_cache_on_variable_change.up.sql +++ b/backend/migrations/20250902085503_notify_resource_cache_on_variable_change.up.sql @@ -1,23 +1,4 @@ -- Add up migration script here -CREATE OR REPLACE FUNCTION notify_var_resource_cache_change() -RETURNS TRIGGER AS $$ -BEGIN - IF TG_TABLE_NAME = 'variable' THEN - PERFORM pg_notify('var_cache_invalidation', - json_build_object( - 'workspace_id', COALESCE(NEW.workspace_id, OLD.workspace_id), - 'path', COALESCE(NEW.path, OLD.path), - 'operation', TG_OP - )::text - ); - END IF; - PERFORM pg_notify('resource_cache_invalidation', - json_build_object( - 'workspace_id', COALESCE(NEW.workspace_id, OLD.workspace_id), - 'path', COALESCE(NEW.path, OLD.path), - 'operation', TG_OP - )::text - ); - RETURN COALESCE(NEW, OLD); -END; -$$ LANGUAGE plpgsql; \ No newline at end of file +DROP TRIGGER variable_cache_invalidate_trigger ON variable; +DROP TRIGGER resource_cache_invalidate_trigger ON resource; +DROP FUNCTION notify_var_resource_cache_change(); \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index 18452f6c22..6ec6e9a1bc 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -1299,8 +1299,6 @@ async fn listen_pg(url: &str) -> Option { "notify_workspace_key_change", "notify_runnable_version_change", "notify_token_invalidation", - "var_cache_invalidation", - "resource_cache_invalidation", ]; #[cfg(feature = "http_trigger")] diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 68a4781225..836ac97cb5 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -3254,6 +3254,12 @@ paths: parameters: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/Path" + - name: allow_cache + description: | + allow getting a cached value for improved performance + in: query + schema: + type: boolean responses: "200": description: variable @@ -4017,6 +4023,11 @@ paths: schema: type: string format: uuid + - name: allow_cache + description: allow getting a cached value for improved performance + in: query + schema: + type: boolean responses: "200": description: resource value @@ -15233,7 +15244,7 @@ components: nu, java, ruby, - duckdb + duckdb, # for related places search: ADD_NEW_LANG ] diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index d20c4052ed..2aef45acb5 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -2517,6 +2517,7 @@ async fn build_args( &path, None, "", + false ) .await?; if res.is_none() { diff --git a/backend/windmill-api/src/resources.rs b/backend/windmill-api/src/resources.rs index cb29f323b5..6381843842 100644 --- a/backend/windmill-api/src/resources.rs +++ b/backend/windmill-api/src/resources.rs @@ -9,10 +9,7 @@ use std::collections::HashMap; use crate::{ - db::{ApiAuthed, DB}, - users::{maybe_refresh_folders, require_owner_of_path, Tokened}, - utils::check_scopes, - webhook_util::{WebhookMessage, WebhookShared}, + db::{ApiAuthed, DB}, users::{maybe_refresh_folders, require_owner_of_path, Tokened}, utils::check_scopes, var_resource_cache::{cache_resource, get_cached_resource}, webhook_util::{WebhookMessage, WebhookShared} }; use axum::{ body::Body, @@ -36,7 +33,6 @@ use windmill_common::{ variables, worker::CLOUD_HOSTED, }; -use crate::var_resource_cache::{get_cached_resource, cache_resource}; pub fn workspaced_service() -> Router { Router::new() @@ -327,28 +323,16 @@ async fn exists_resource( Ok(Json(exists)) } -#[derive(Deserialize)] -struct GetResourceQuery { - allow_cache: Option, -} async fn get_resource_value( authed: ApiAuthed, Extension(user_db): Extension, Extension(db): Extension, - Query(q): Query, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult> { let path = path.to_path(); check_scopes(&authed, || format!("resources:read:{}", path))?; - // Check cache first when explicitly allowed - let allow_cache = q.allow_cache.unwrap_or(false); - if allow_cache { - if let Some(cached_value) = get_cached_resource(&w_id, &path) { - return Ok(Json(Some(cached_value))); - } - } let mut tx = user_db.begin(&authed).await?; @@ -367,12 +351,7 @@ async fn get_resource_value( let value = not_found_if_none(value_o, "Resource", path)?; - // Cache the result if it exists and caching is allowed - if allow_cache { - if let Some(ref val) = value { - cache_resource(&w_id, &path, val.clone()); - } - } + Ok(Json(value)) } @@ -440,7 +419,9 @@ async fn custom_component( #[derive(Deserialize)] struct JobInfo { job_id: Option, + allow_cache: Option, } + async fn get_resource_value_interpolated( authed: ApiAuthed, Extension(user_db): Extension, @@ -452,6 +433,7 @@ async fn get_resource_value_interpolated( let path = path.to_path(); check_scopes(&authed, || format!("resources:read:{}", path))?; + return get_resource_value_interpolated_internal( &authed, Some(user_db), @@ -460,6 +442,7 @@ async fn get_resource_value_interpolated( path, job_info.job_id, token.as_str(), + job_info.allow_cache.unwrap_or(false), ) .await .map(|success| Json(success)); @@ -476,7 +459,13 @@ pub async fn get_resource_value_interpolated_internal( path: &str, job_id: Option, token: &str, + allow_cache: bool, ) -> Result> { + if allow_cache { + if let Some(cached_value) = get_cached_resource(&workspace, &path) { + return Ok(Some(cached_value)); + } + } let mut tx = authed_transaction_or_default(authed, user_db.clone(), db).await?; let value_o = sqlx::query_scalar!( @@ -493,18 +482,20 @@ pub async fn get_resource_value_interpolated_internal( let value = not_found_if_none(value_o, "Resource", path)?; if let Some(value) = value { - Ok(Some( - transform_json_value( - authed, - user_db.clone(), - db, - workspace, - value, - &job_id, - token, - ) - .await?, - )) + let r = transform_json_value( + authed, + user_db.clone(), + db, + workspace, + value, + &job_id, + token, + ) + .await?; + if allow_cache { + cache_resource(&workspace, &path, r.clone()); + } + Ok(Some(r)) } else { Ok(None) } @@ -540,6 +531,7 @@ pub async fn transform_json_value<'c>( username_override: None, token_prefix: None, }), + false ) .await?; Ok(Value::String(v)) diff --git a/backend/windmill-api/src/var_resource_cache.rs b/backend/windmill-api/src/var_resource_cache.rs index bd35647b87..d97416b871 100644 --- a/backend/windmill-api/src/var_resource_cache.rs +++ b/backend/windmill-api/src/var_resource_cache.rs @@ -9,10 +9,9 @@ use quick_cache::sync::Cache; use serde_json::Value; use std::time::{SystemTime, UNIX_EPOCH}; -use windmill_common::variables::ListableVariable; -/// Cache TTL for variables and resources (60 seconds) -const CACHE_TTL_SECS: u64 = 60; +/// Cache TTL for variables and resources (30seconds) +const CACHE_TTL_SECS: u64 = 30; /// Cache entry with timestamp and value (following raw script cache pattern) #[derive(Clone, Debug)] @@ -41,9 +40,10 @@ impl CacheEntry { } } + lazy_static::lazy_static! { /// Cache for individual variable values: key = "workspace_id:path" - pub static ref VARIABLE_CACHE: Cache> = Cache::new(1000); + pub static ref VARIABLE_CACHE: Cache> = Cache::new(1000); /// Cache for resource values: key = "workspace_id:path" pub static ref RESOURCE_CACHE: Cache> = Cache::new(1000); @@ -56,7 +56,7 @@ pub fn cache_key(workspace_id: &str, path: &str) -> String { } /// Get cached variable if available and not expired -pub fn get_cached_variable(workspace_id: &str, path: &str) -> Option { +pub fn get_cached_variable(workspace_id: &str, path: &str) -> Option { let key = cache_key(workspace_id, path); VARIABLE_CACHE.get(&key).and_then(|entry| { if entry.is_expired() { @@ -70,8 +70,8 @@ pub fn get_cached_variable(workspace_id: &str, path: &str) -> Option, include_encrypted: Option, - allow_cache: Option, } async fn get_variable( @@ -147,17 +146,6 @@ async fn get_variable( ) -> JsonResult { let path = path.to_path(); check_scopes(&authed, || format!("variables:read:{}", path))?; - - // Check cache first when explicitly allowed (and for appropriate requests) - let decrypt_secret = q.decrypt_secret.unwrap_or(true); - let allow_cache = q.allow_cache.unwrap_or(false); - let include_encrypted = q.include_encrypted.unwrap_or(false); - - if allow_cache && (!decrypt_secret || include_encrypted) { - if let Some(cached_variable) = get_cached_variable(&w_id, &path) { - return Ok(Json(cached_variable)); - } - } let mut tx = user_db.begin(&authed).await?; @@ -232,25 +220,25 @@ async fn get_variable( variable }; - // Cache the result when explicitly allowed and caching appropriate - if allow_cache && (!decrypt_secret || include_encrypted) { - cache_variable(&w_id, &path, r.clone()); - } Ok(Json(r)) } +#[derive(Deserialize)] +struct GetValueQuery { + allow_cache: Option, +} async fn get_value( authed: ApiAuthed, Extension(user_db): Extension, Extension(db): Extension, - Path((w_id, path)): Path<(String, StripPath)>, + Query(q): Query, ) -> JsonResult { let path = path.to_path(); check_scopes(&authed, || format!("variables:read:{}", path))?; let tx = user_db.begin(&authed).await?; - return get_value_internal(tx, &db, &w_id, &path, &authed) + return get_value_internal(tx, &db, &w_id, &path, &authed, q.allow_cache.unwrap_or(false)) .await .map(Json); } @@ -712,7 +700,16 @@ pub async fn get_value_internal<'c>( w_id: &str, path: &str, audit_author: &impl AuditAuthorable, + allow_cache: bool, ) -> Result { + + + if allow_cache { + if let Some(cached_variable) = get_cached_variable(&w_id, &path) { + return Ok(cached_variable); + } + } + let variable_o = sqlx::query!( "SELECT value, account, (now() > account.expires_at) as is_expired, is_secret, path from variable LEFT JOIN account ON variable.account = account.id WHERE variable.path = $1 AND variable.workspace_id = $2", path, w_id @@ -728,6 +725,8 @@ pub async fn get_value_internal<'c>( unreachable!() }; + + let r = if variable.is_secret { audit_log( &mut *tx, @@ -765,6 +764,12 @@ pub async fn get_value_internal<'c>( variable.value }; + // Cache the result when explicitly allowed and caching appropriate + if allow_cache { + cache_variable(&w_id, &path, audit_author.email(), r.clone()); + } + + Ok(r) }