From e1a7c75e192b72b3b0d854c1901653e0b9386bf2 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 8 May 2026 07:43:46 +0200 Subject: [PATCH] perf(flows): cache resolved flow_env per flow execution (#9079) * perf(flows): cache resolved flow_env per flow execution * perf(flows): tighten flow_env cache cap to 1024 and clarify memory note * perf(flows): don't cache transient flow_env resolution failures --- backend/Cargo.lock | 1 + backend/windmill-worker/Cargo.toml | 1 + backend/windmill-worker/src/worker_flow.rs | 85 ++++++++++++++++++---- 3 files changed, 72 insertions(+), 15 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index c6bf144004..323155c29b 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -17895,6 +17895,7 @@ dependencies = [ "process-wrap", "prometheus", "prost", + "quick_cache", "rand 0.9.0", "rcgen", "regex", diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index ad3d87d8df..f1c7fa6c99 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -93,6 +93,7 @@ itertools.workspace = true regex.workspace = true prometheus = { workspace = true, optional = true } lazy_static.workspace = true +quick_cache.workspace = true chrono.workspace = true dotenv.workspace = true rand.workspace = true # TODO: Remove. only used by token creation hack. diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 3a07d1bfe4..91c6cf9ec7 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -81,6 +81,33 @@ use windmill_audit::ActionKind; use windmill_common::audit::AuditAuthor; use windmill_queue::{canceled_job_to_result, push}; +lazy_static::lazy_static! { + /// Per-worker LRU cache of resolved `flow_env` values, keyed by flow job id. + /// `update_flow_status_after_job_completion_internal` runs once per child-step + /// completion, and re-resolving `$var:`/`$res:` references each time was the + /// dominant heap-allocation source under flow-heavy load. The cache aligns + /// predicate evaluation with `handle_flow`'s input-transform path, which + /// already resolves once at flow entry — predicates were the only place still + /// re-resolving on every read. + /// + /// Per-worker scope (caveat): the cache lives in process memory, not the DB, + /// so different workers processing children of the same flow each compute + /// their own snapshot on first miss. If a `$var:`/`$res:` value mutates + /// mid-flow, predicate eval on different workers can observe different + /// values for the same flow run. For typical use (env values configured at + /// flow start, read-only during execution) this is invisible. Cross-worker + /// determinism would require persisting the resolved env in `v2_job_status`; + /// see follow-up notes. + /// + /// Entries become dead weight once a flow completes and are evicted by LRU + /// pressure. Bounded at 1024 entries; per-entry footprint depends on the + /// env's contents (literals are small but a single resolved `$res:` can be + /// tens of KB), so worst-case memory scales with workload mix rather than + /// being fixed. + static ref RESOLVED_FLOW_ENV_CACHE: quick_cache::sync::Cache>>> = + quick_cache::sync::Cache::new(1024); +} + #[derive(Debug)] pub struct SchedulePushZombieError(pub String); @@ -480,8 +507,26 @@ pub async fn update_flow_status_after_job_completion_internal( .failure_module .as_ref() .is_some_and(|fm| retry_uses_flow_env(fm)); - let resolved_flow_env: Option>> = if needs_flow_env { - resolve_flow_env_for_status_update(db, client, flow, w_id, flow_value).await + // The resolved env is constant for a flow's lifetime, so cache it by flow + // job id and reuse across child-step completions. Cache miss falls back to + // the existing resolve+persist path; same-worker subsequent completions are + // a single Arc::clone away. Transient-failure fallbacks (`is_cacheable == + // false`) bypass the insert so a single API blip doesn't poison the rest + // of the flow run. + let resolved_flow_env: Option>>> = if needs_flow_env { + if let Some(cached) = RESOLVED_FLOW_ENV_CACHE.get(&flow) { + Some(cached) + } else { + resolve_flow_env_for_status_update(db, client, flow, w_id, flow_value) + .await + .map(|(env, is_cacheable)| { + let arc = Arc::new(env); + if is_cacheable { + RESOLVED_FLOW_ENV_CACHE.insert(flow, arc.clone()); + } + arc + }) + } } else { None }; @@ -632,7 +677,7 @@ pub async fn update_flow_status_after_job_completion_internal( let bool_res = compute_bool_from_expr( &expr, Marc::new(args), - resolved_flow_env.as_ref(), + resolved_flow_env.as_deref(), result.clone(), all_iters, id_ctx.as_ref(), @@ -916,7 +961,7 @@ pub async fn update_flow_status_after_job_completion_internal( &mut stop_early_err_msg, &mut nresult, args, - resolved_flow_env.as_ref(), + resolved_flow_env.as_deref(), flow, &old_status, ) @@ -1136,7 +1181,7 @@ pub async fn update_flow_status_after_job_completion_internal( &mut stop_early_err_msg, &mut nresult, args, - resolved_flow_env.as_ref(), + resolved_flow_env.as_deref(), flow, &old_status, ) @@ -1219,7 +1264,7 @@ pub async fn update_flow_status_after_job_completion_internal( &old_status.retry, result.clone(), Marc::new(args), - resolved_flow_env.as_ref(), + resolved_flow_env.as_deref(), Some(client), ) .await? @@ -1601,7 +1646,7 @@ pub async fn update_flow_status_after_job_completion_internal( &old_status.retry, result.clone(), Marc::new(args), - resolved_flow_env.as_ref(), + resolved_flow_env.as_deref(), Some(client), ) .await? @@ -2387,13 +2432,19 @@ async fn fetch_root_flow_id(db: &DB, flow_id: Uuid) -> Uuid { // `update_flow_status_after_job_completion_internal`: take the current flow's // `flow_env` if present, otherwise inherit from the root flow, then interpolate // any `$var:`/`$res:` references via `transform_json`. +// +// Returns `Some((env, is_cacheable))`. `is_cacheable` is `false` only when the +// returned env is a partially-resolved fallback after a transient error (e.g. +// `transform_json` failed mid-resolve, or the mini job fetch failed). Callers +// must not cache `is_cacheable == false` results — doing so would freeze the +// transient failure for the rest of the flow run. async fn resolve_flow_env_for_status_update( db: &DB, client: &AuthedClient, flow_job_id: Uuid, workspace_id: &str, flow_value: &FlowValue, -) -> Option>> { +) -> Option<(HashMap>, bool)> { // Fetch the env source. For the inherited path, we first need to know whether the // flow even has a parent — `fetch_root_flow_env` runs a recursive CTE on `v2_job` // and is wasted work for top-level flows with no own `flow_env`. The mini job we @@ -2418,21 +2469,23 @@ async fn resolve_flow_env_for_status_update( (env, Some(mini)) }; if env.is_empty() { - return Some(env); + return Some((env, true)); } // Skip the DB roundtrip + `transform_json` when nothing in env needs interpolation. // This is the common case: a flow_env containing only literal values. if !crate::common::map_needs_resolution(&env) { - return Some(env); + return Some((env, true)); } let mini = match mini { Some(m) => m, None => match get_mini_pulled_job(db, &flow_job_id).await { Ok(Some(j)) => j, - Ok(None) => return Some(env), + // Don't cache: we couldn't fetch the job, so the env we'd return is the + // pre-interpolation literal — caching that would freeze the broken state. + Ok(None) => return Some((env, false)), Err(e) => { tracing::warn!("Failed to fetch flow job to resolve flow_env: {e:#}"); - return Some(env); + return Some((env, false)); } }, }; @@ -2445,11 +2498,13 @@ async fn resolve_flow_env_for_status_update( ) .await { - Ok(Some(resolved)) => Some(resolved), - Ok(None) => Some(env), + Ok(Some(resolved)) => Some((resolved, true)), + Ok(None) => Some((env, true)), + // Don't cache transient resolution failures: one variable/resource API blip + // would otherwise poison the literal-only env for the rest of the flow run. Err(e) => { tracing::warn!("Failed to resolve flow_env references in status update: {e:#}"); - Some(env) + Some((env, false)) } } }