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

cache: better cache system (#4951)

Browse Source 
* 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
This commit is contained in:
Lucas Abel
2024-12-20 11:19:57 +01:00
committed by GitHub
parent 9d9fa34851
commit 12d18beb32
14 changed files with 625 additions and 424 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/backend-test.yml
+1 -1
View File
@@ -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
backend/.sqlx/query-0bf123446bebbc357c58a53a9319f4954dbf3225e91cbe999e5b264c1a747664.json
Generated
+72
View File
@@ -0,0 +1,72 @@
{
"db_name": "PostgreSQL",
"query": "SELECT content AS \"content!: String\",\n lock AS \"lock: String\", language AS \"language: Option<ScriptLang>\", envs AS \"envs: Vec<String>\", codebase AS \"codebase: String\" FROM script WHERE hash = $1 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "content!: String",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "lock: String",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "language: Option<ScriptLang>",
"type_info": {
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp"
]
}
}
}
},
{
"ordinal": 3,
"name": "envs: Vec<String>",
"type_info": "VarcharArray"
},
{
"ordinal": 4,
"name": "codebase: String",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
false,
true,
false,
true,
true
]
},
"hash": "0bf123446bebbc357c58a53a9319f4954dbf3225e91cbe999e5b264c1a747664"
}
backend/.sqlx/query-3ee2d60ca93eeaf02ab0ee96aca399ec055b044a06284c0ab19b67d97f803894.json
Generated
+34
View File
@@ -0,0 +1,34 @@
{
"db_name": "PostgreSQL",
"query": "SELECT code AS \"raw_code: String\", lock AS \"raw_lock: String\", flow AS \"raw_flow: Json<Box<RawValue>>\" FROM flow_node WHERE id = $1 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "raw_code: String",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "raw_lock: String",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "raw_flow: Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
true,
true,
true
]
},
"hash": "3ee2d60ca93eeaf02ab0ee96aca399ec055b044a06284c0ab19b67d97f803894"
}
backend/.sqlx/query-b7f2ed32e933b65fa5455928c71f61068ad7dfee8352a76122f5af893ffe6517.json
Generated
+72
View File
@@ -0,0 +1,72 @@
{
"db_name": "PostgreSQL",
"query": "SELECT content AS \"content!: String\",\n lock AS \"lock: String\", language AS \"language: Option<ScriptLang>\", envs AS \"envs: Vec<String>\", codebase AS \"codebase: String\" FROM script WHERE hash = $1 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "content!: String",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "lock: String",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "language: Option<ScriptLang>",
"type_info": {
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp"
]
}
}
}
},
{
"ordinal": 3,
"name": "envs: Vec<String>",
"type_info": "VarcharArray"
},
{
"ordinal": 4,
"name": "codebase: String",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
false,
true,
false,
true,
true
]
},
"hash": "b7f2ed32e933b65fa5455928c71f61068ad7dfee8352a76122f5af893ffe6517"
}
backend/Cargo.lock
Generated
+1
View File
@@ -10871,6 +10871,7 @@ dependencies = [
"serde_json",
"sha2 0.10.8",
"sqlx",
"tempfile",
"thiserror 2.0.8",
"tikv-jemalloc-ctl",
"tokio",
backend/Cargo.toml
+1
View File
@@ -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]
backend/tests/worker.rs
+3 -5
View File
@@ -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 {
backend/windmill-api/src/apps.rs
+3 -6
View File
@@ -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!(),
backend/windmill-common/Cargo.toml
+2
View File
@@ -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 }
backend/windmill-common/src/cache.rs
+427 -403
View File
File diff suppressed because it is too large Load Diff
backend/windmill-common/src/flows.rs
+1 -1
View File
@@ -840,7 +840,7 @@ pub async fn resolve_modules(
if let Some(id) = modules_node {
*modules = cache::flow::fetch_flow(e, id)
.await
.and_then(|data| Ok(data.value()?.modules.clone()))?;
.map(|data| data.value().modules.clone())?;
}
for module in modules.iter_mut() {
Box::pin(resolve_module(
backend/windmill-queue/src/jobs.rs
+5 -5
View File
@@ -2868,7 +2868,7 @@ pub async fn push<'c, 'd>(
),
JobPayload::FlowNode { id, path } => {
let data = cache::flow::fetch_flow(_db, id).await?;
let value = data.value()?;
let value = data.value();
let status = Some(FlowStatus::new(value));
// Keep inserting `value` if not all workers are updated.
// Starting at `v1.440`, the value is fetched on pull from the flow node id.
@@ -3020,7 +3020,7 @@ pub async fn push<'c, 'd>(
// The version has been inserted only within the transaction.
let data = cache::flow::fetch_version(&mut *ntx, version).await?;
tx = PushIsolationLevel::Transaction(ntx);
Some(data.value()?.clone())
Some(data.value().clone())
} else {
// `raw_flow` is fetched on pull.
None
@@ -3218,7 +3218,7 @@ pub async fn push<'c, 'd>(
}?;
tx = PushIsolationLevel::Transaction(ntx);
let value = data.value()?.clone();
let value = data.value().clone();
let priority = value.priority;
let cache_ttl = value.cache_ttl.map(|x| x as i32);
let custom_concurrency_key = value.concurrency_key.clone();
@@ -3311,7 +3311,7 @@ pub async fn push<'c, 'd>(
user_states,
preprocessor_module: None,
};
let value = flow_data.value()?;
let value = flow_data.value();
let priority = value.priority;
let concurrency_key = value.concurrency_key.clone();
let concurrent_limit = value.concurrent_limit;
@@ -3781,7 +3781,7 @@ async fn restarted_flows_resolution(
let flow_data = cache::job::fetch_flow(db, row.job_kind, row.script_hash)
.or_else(|_| cache::job::fetch_preview_flow(db, &completed_flow_id, row.raw_flow))
.await?;
let flow_value = flow_data.value()?;
let flow_value = flow_data.value();
let flow_status = row
.flow_status
.as_ref()
backend/windmill-worker/src/worker_flow.rs
+2 -2
View File
@@ -250,7 +250,7 @@ pub async fn update_flow_status_after_job_completion_internal(
let flow_data = cache::job::fetch_flow(db, job_kind, script_hash)
.or_else(|_| cache::job::fetch_preview_flow(db, &flow, raw_flow))
.await?;
let flow_value = flow_data.value()?;
let flow_value = flow_data.value();
let module_step = Step::from_i32_and_len(old_status.step, old_status.modules.len());
let current_module = match module_step {
@@ -1499,7 +1499,7 @@ pub async fn handle_flow(
worker_dir: &str,
job_completed_tx: Sender<SendResult>,
) -> anyhow::Result<()> {
let flow = flow_data.value()?;
let flow = flow_data.value();
let status = flow_job
.parse_flow_status()
.with_context(|| "Unable to parse flow status")?;
backend/windmill-worker/src/worker_lockfiles.rs
+1 -1
View File
@@ -603,7 +603,7 @@ pub async fn handle_flow_dependency_job(
_ => return Err(Error::InternalErr("expected script hash".into())),
},
}
.value()?
.value()
.clone();
let mut tx = db.begin().await?;