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

@@ -17895,6 +17895,7 @@ dependencies = [
  "process-wrap",
  "prometheus",
  "prost",
+ "quick_cache",
  "rand 0.9.0",
  "rcgen",
  "regex",

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.

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<Uuid, Arc<HashMap<String, Box<RawValue>>>> =
+        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<HashMap<String, Box<RawValue>>> = 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<Arc<HashMap<String, Box<RawValue>>>> = 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<HashMap<String, Box<RawValue>>> {
+) -> Option<(HashMap<String, Box<RawValue>>, 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))
         }
     }
 }