cache: better cache system (#4951)

* cache: consider source as untrusted * cache: better scoped cache * cache: non-lazy flow value parse error * fixup! cache: consider source as untrusted * fixup! cache: consider source as untrusted * fixup! cache: better scoped cache
windmill-labs · Dec 20, 2024 · 83d24cb · 83d24cb
1 parent 71d527c
commit 83d24cb
Show file tree

Hide file tree

Showing 14 changed files with 599 additions and 398 deletions.
diff --git a/.github/workflows/backend-test.yml b/.github/workflows/backend-test.yml
@@ -67,4 +67,4 @@ jobs:
           DISABLE_EMBEDDING=true RUST_LOG=info PYTHON_PATH=$(which python)
           DENO_PATH=$(which deno) BUN_PATH=$(which bun) GO_PATH=$(which go)
           UV_PATH=$(which uv) cargo test --features
-          enterprise,deno_core,license,python,rust --all -- --nocapture
+          enterprise,deno_core,license,python,rust,scoped_cache --all -- --nocapture
diff --git a/backend/.sqlx/query-0bf123446bebbc357c58a53a9319f4954dbf3225e91cbe999e5b264c1a747664.json b/backend/.sqlx/query-0bf123446bebbc357c58a53a9319f4954dbf3225e91cbe999e5b264c1a747664.json
diff --git a/backend/.sqlx/query-3ee2d60ca93eeaf02ab0ee96aca399ec055b044a06284c0ab19b67d97f803894.json b/backend/.sqlx/query-3ee2d60ca93eeaf02ab0ee96aca399ec055b044a06284c0ab19b67d97f803894.json
diff --git a/backend/.sqlx/query-b7f2ed32e933b65fa5455928c71f61068ad7dfee8352a76122f5af893ffe6517.json b/backend/.sqlx/query-b7f2ed32e933b65fa5455928c71f61068ad7dfee8352a76122f5af893ffe6517.json
diff --git a/backend/Cargo.lock b/backend/Cargo.lock
diff --git a/backend/Cargo.toml b/backend/Cargo.toml
@@ -79,6 +79,7 @@ oauth2 = ["windmill-api/oauth2"]
 http_trigger = ["windmill-api/http_trigger"]
 zip = ["windmill-api/zip"]
 static_frontend = ["windmill-api/static_frontend"]
+scoped_cache = ["windmill-common/scoped_cache"]
 
 
 [dependencies]

diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs
@@ -3814,7 +3814,7 @@ mod job_payload {
         let flow_data = cache::flow::fetch_version_lite(&db, 1443253234253454)
             .await
             .unwrap();
-        let flow_value = flow_data.value().unwrap();
+        let flow_value = flow_data.value();
         let flow_scripts = {
             async fn load(db: &Pool<Postgres>, modules: &[FlowModule]) -> Vec<FlowNodeId> {
                 let mut res = vec![];
@@ -3825,9 +3825,7 @@ mod job_payload {
                         FlowModuleValue::FlowScript { id, .. } => res.push(id),
                         FlowModuleValue::ForloopFlow { modules_node: Some(flow_node), .. } => {
                             let flow_data = cache::flow::fetch_flow(db, flow_node).await.unwrap();
-                            res.extend(
-                                Box::pin(load(db, &flow_data.value().unwrap().modules)).await,
-                            );
+                            res.extend(Box::pin(load(db, &flow_data.value().modules)).await);
                         }
                         _ => {}
                     }
@@ -3904,7 +3902,7 @@ mod job_payload {
         let flow_data = cache::flow::fetch_version_lite(&db, 1443253234253454)
             .await
             .unwrap();
-        let flow_value = flow_data.value().unwrap();
+        let flow_value = flow_data.value();
         let forloop_module =
             serde_json::from_str::<FlowModuleValue>(flow_value.modules[0].value.get()).unwrap();
         let FlowModuleValue::ForloopFlow { modules_node: Some(id), .. } = forloop_module else {

diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs
@@ -1322,10 +1322,7 @@ async fn execute_component(
             // 2. Otherwise, always fetch the policy from the database.
             let policy = if let Some(id) = payload.version {
                 let cache = cache::anon!({ u64 => Arc<Policy> } in "policy" <= 1000);
-                arc_policy = policy_fut
-                    .map_ok(sqlx::types::Json) // cache as json.
-                    .cached(cache, id as u64, |sqlx::types::Json(x)| Arc::new(x))
-                    .await?;
+                arc_policy = policy_fut.map_ok(Arc::new).cached(cache, id as u64).await?;
                 &*arc_policy
             } else {
                 policy = policy_fut.await?;
@@ -1352,8 +1349,8 @@ async fn execute_component(
                     )
                     .fetch_one(&db)
                     .map_err(Into::<Error>::into)
-                    .map_ok(sqlx::types::Json) // cache as json.
-                    .cached(cache, *id as u64, |sqlx::types::Json(x)| Arc::new(x))
+                    .map_ok(Arc::new)
+                    .cached(cache, *id as u64)
                     .await?
                 }
                 _ => unreachable!(),

diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml
@@ -15,6 +15,7 @@ parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts"]
 otel = ["dep:opentelemetry-semantic-conventions", "dep:opentelemetry-otlp", "dep:opentelemetry_sdk", 
     "dep:opentelemetry", "dep:tracing-opentelemetry", "dep:opentelemetry-appender-tracing", "dep:tonic"]
 smtp = ["dep:mail-send"]
+scoped_cache = []
 
 [lib]
 name = "windmill_common"
@@ -65,6 +66,7 @@ croner = "2.0.6"
 quick_cache.workspace = true
 pin-project-lite.workspace = true
 futures.workspace = true
+tempfile.workspace = true
 
 opentelemetry-semantic-conventions = { workspace = true, optional = true }
 opentelemetry-otlp = { workspace = true, optional = true }