From 027fa6ff095c786ebd69f79aafd61e4f9414de2b Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 4 Jun 2023 14:39:10 +0200 Subject: [PATCH] feat: add cache as a primitive for flows (#1671) * feat: add cache as a primitive for flows * fix failure module --- backend/sqlx-data.json | 50 +++++-- backend/tests/worker.rs | 9 +- backend/windmill-api/openapi.yaml | 109 ++++++++++---- backend/windmill-api/src/flows.rs | 4 + backend/windmill-api/src/resources.rs | 14 +- backend/windmill-common/src/flows.rs | 2 + backend/windmill-queue/src/jobs.rs | 1 + backend/windmill-worker/src/js_eval.rs | 7 +- backend/windmill-worker/src/worker.rs | 141 +++++++++++++----- backend/windmill-worker/src/worker_flow.rs | 18 ++- .../flows/content/FlowModuleCache.svelte | 42 ++++++ .../flows/content/FlowModuleComponent.svelte | 7 + .../flows/content/FlowModuleHeader.svelte | 13 +- .../flows/content/FlowModuleSuspend.svelte | 7 - .../flows/map/FlowModuleSchemaItem.svelte | 14 +- .../lib/components/flows/map/MapItem.svelte | 3 +- .../propertyPicker/ObjectViewer.svelte | 2 +- .../(root)/(logged)/resources/+page.svelte | 27 +++- openflow.openapi.yaml | 2 + 19 files changed, 359 insertions(+), 113 deletions(-) create mode 100644 frontend/src/lib/components/flows/content/FlowModuleCache.svelte diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index f64be6a551..ab5d80c0f8 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -2888,6 +2888,22 @@ }, "query": "SELECT EXISTS(SELECT 1 FROM usr WHERE workspace_id = $1 AND username = $2)" }, + "726f77c2c1ef63eece7a1176bcbb916987fc29b04b889544a064dc7cd5eab36f": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Jsonb", + "Text", + "Varchar" + ] + } + }, + "query": "INSERT INTO resource\n (workspace_id, path, value, description, resource_type)\n VALUES ($1, $2, $3, $4, $5) ON CONFLICT (workspace_id, path)\n DO UPDATE SET value = $3, description = $4, resource_type = $5" + }, "73d3ed17fd0723ba75722f394904f6ee306b59aaf8ecfcf56484d38541343f06": { "describe": { "columns": [], @@ -3587,6 +3603,24 @@ }, "query": "SELECT lockfile FROM pip_resolution_cache WHERE hash = $1" }, + "873fde22f7947882edae7d15bc54e8df105d5e241eeb842a83e3444be0d2736d": { + "describe": { + "columns": [ + { + "name": "path", + "ordinal": 0, + "type_info": "Varchar" + } + ], + "nullable": [ + false + ], + "parameters": { + "Left": [] + } + }, + "query": "DELETE FROM resource WHERE resource_type = 'cache' AND to_timestamp((value->>'expire')::int) < now() RETURNING path" + }, "8876fa929ffb175cd976a2bca1195704aa9fe7215013ae29e49ef15cb201ba57": { "describe": { "columns": [], @@ -3675,22 +3709,6 @@ }, "query": "DELETE FROM usr_to_group WHERE workspace_id = $1" }, - "8a80333c2fbf7b50fed305882de6e4ffda985d5c648cd617add6c9e6a9c03f34": { - "describe": { - "columns": [], - "nullable": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Jsonb", - "Text", - "Varchar" - ] - } - }, - "query": "INSERT INTO resource\n (workspace_id, path, value, description, resource_type)\n VALUES ($1, $2, $3, $4, $5)" - }, "8c0131a9cc61f2daa258d49767242bcaab6bb34a977ff7fb0c18aa9202d11f47": { "describe": { "columns": [], diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index ff499ce957..1fd608b842 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1015,6 +1015,7 @@ async fn test_deno_flow(db: Pool) { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }, FlowModule { id: "b".to_string(), @@ -1043,6 +1044,7 @@ async fn test_deno_flow(db: Pool) { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }], }, stop_after_if: Default::default(), @@ -1050,6 +1052,7 @@ async fn test_deno_flow(db: Pool) { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }, ], same_worker: false, @@ -1144,6 +1147,7 @@ async fn test_deno_flow_same_worker(db: Pool) { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }, FlowModule { id: "b".to_string(), @@ -1183,6 +1187,7 @@ async fn test_deno_flow_same_worker(db: Pool) { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }, FlowModule { id: "e".to_string(), @@ -1209,6 +1214,7 @@ async fn test_deno_flow_same_worker(db: Pool) { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }, ], }, @@ -1217,7 +1223,7 @@ async fn test_deno_flow_same_worker(db: Pool) { suspend: Default::default(), retry: None, sleep: None, - + cache_ttl: None, }, FlowModule { id: "c".to_string(), @@ -1252,6 +1258,7 @@ async fn test_deno_flow_same_worker(db: Pool) { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }, ], same_worker: true, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index bbaed4befe..c855771ab0 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -569,7 +569,6 @@ paths: - summary - kind - /users/whoami: get: summary: get current global whoami (if logged in) @@ -1572,6 +1571,10 @@ paths: - resource parameters: - $ref: "#/components/parameters/WorkspaceId" + - name: update_if_exists + in: query + schema: + type: boolean requestBody: description: new resource required: true @@ -2113,7 +2116,8 @@ paths: schema: type: string - name: first_parent_hash - description: mask to filter scripts whom first direct parent has exact hash + description: + mask to filter scripts whom first direct parent has exact hash in: query schema: type: string @@ -2196,7 +2200,6 @@ paths: items: type: string - /w/{workspace}/drafts/create: post: summary: create draft @@ -2230,7 +2233,6 @@ paths: schema: type: string - /w/{workspace}/scripts/create: post: summary: create script @@ -2257,7 +2259,9 @@ paths: /workers/custom_tags: get: - summary: get all instance custom tags (tags are used to dispatch jobs to different worker groups) + summary: + get all instance custom tags (tags are used to dispatch jobs to + different worker groups) operationId: getCustomTags tags: - worker @@ -2391,7 +2395,8 @@ paths: /w/{workspace}/scripts/delete/h/{hash}: post: - summary: delete script by hash (erase content but keep hash, require admin) + summary: + delete script by hash (erase content but keep hash, require admin) operationId: deleteScriptByHash tags: - script @@ -2457,8 +2462,6 @@ paths: schema: $ref: "#/components/schemas/NewScriptWithDraft" - - /w/{workspace}/scripts/raw/p/{path}: get: summary: raw script by path @@ -2478,7 +2481,9 @@ paths: /scripts_u/tokened_raw/{workspace}/{token}/{path}: get: - summary: raw script by path with a token (mostly used by lsp to be used with import maps to resolve scripts) + summary: + raw script by path with a token (mostly used by lsp to be used with + import maps to resolve scripts) operationId: rawScriptByPathTokened tags: - script @@ -2583,14 +2588,16 @@ paths: type: string format: date-time - name: scheduled_in_secs - description: schedule the script to execute in the number of seconds starting now + description: + schedule the script to execute in the number of seconds starting now in: query schema: type: integer - $ref: "#/components/parameters/ParentJob" - $ref: "#/components/parameters/NewJobId" - name: invisible_to_owner - description: make the run invisible to the the script owner (default false) + description: + make the run invisible to the the script owner (default false) in: query schema: type: boolean @@ -2826,7 +2833,6 @@ paths: draft: $ref: "#/components/schemas/Flow" - /w/{workspace}/flows/exists/{path}: get: summary: exists flow by path @@ -3165,7 +3171,6 @@ paths: schema: $ref: "#/components/schemas/AppWithLastVersionWDraft" - /w/{workspace}/apps_u/public_app/{path}: get: summary: get public app by secret @@ -3414,7 +3419,8 @@ paths: type: string format: date-time - name: scheduled_in_secs - description: schedule the script to execute in the number of seconds starting now + description: + schedule the script to execute in the number of seconds starting now in: query schema: type: integer @@ -3422,7 +3428,8 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/IncludeHeader" - name: invisible_to_owner - description: make the run invisible to the the flow owner (default false) + description: + make the run invisible to the the flow owner (default false) in: query schema: type: boolean @@ -3460,7 +3467,8 @@ paths: type: string format: date-time - name: scheduled_in_secs - description: schedule the script to execute in the number of seconds starting now + description: + schedule the script to execute in the number of seconds starting now in: query schema: type: integer @@ -3468,7 +3476,8 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/IncludeHeader" - name: invisible_to_owner - description: make the run invisible to the the script owner (default false) + description: + make the run invisible to the the script owner (default false) in: query schema: type: boolean @@ -3499,7 +3508,8 @@ paths: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/IncludeHeader" - name: invisible_to_owner - description: make the run invisible to the the script owner (default false) + description: + make the run invisible to the the script owner (default false) in: query schema: type: boolean @@ -3532,7 +3542,8 @@ paths: - $ref: "#/components/parameters/WorkspaceId" - $ref: "#/components/parameters/IncludeHeader" - name: invisible_to_owner - description: make the run invisible to the the script owner (default false) + description: + make the run invisible to the the script owner (default false) in: query schema: type: boolean @@ -3883,7 +3894,8 @@ paths: /w/{workspace}/jobs/resume_urls/{id}/{resume_id}: get: - summary: get resume urls given a job_id, resume_id and a nonce to resume a flow + summary: + get resume urls given a job_id, resume_id and a nonce to resume a flow operationId: getResumeUrls tags: - job @@ -4335,7 +4347,8 @@ paths: - $ref: "#/components/parameters/WorkspaceId" - name: only_member_of in: query - description: only list the groups the user is member of (default false) + description: + only list the groups the user is member of (default false) schema: type: boolean responses: @@ -4523,7 +4536,8 @@ paths: - $ref: "#/components/parameters/WorkspaceId" - name: only_member_of in: query - description: only list the folders the user is member of (default false) + description: + only list the folders the user is member of (default false) schema: type: boolean responses: @@ -4762,7 +4776,17 @@ paths: schema: type: string enum: - [script, group_, resource, schedule, variable, flow, folder, app, raw_app] + [ + script, + group_, + resource, + schedule, + variable, + flow, + folder, + app, + raw_app, + ] responses: "200": description: acls @@ -4788,7 +4812,17 @@ paths: schema: type: string enum: - [script, group_, resource, schedule, variable, flow, folder, app, raw_app] + [ + script, + group_, + resource, + schedule, + variable, + flow, + folder, + app, + raw_app, + ] requestBody: description: acl to add required: true @@ -4825,7 +4859,17 @@ paths: schema: type: string enum: - [script, group_, resource, schedule, variable, flow, folder, app, raw_app] + [ + script, + group_, + resource, + schedule, + variable, + flow, + folder, + app, + raw_app, + ] requestBody: description: acl to add required: true @@ -5131,7 +5175,8 @@ components: type: integer PerPage: name: per_page - description: number of items to return for a given page (default 30, max 100) + description: + number of items to return for a given page (default 30, max 100) in: query schema: type: integer @@ -5159,8 +5204,9 @@ components: NewJobId: name: job_id description: - The job id to assign to the created job. if missing, job is chosen randomly using the ULID scheme. - If a job id already exists in the queue or as a completed job, the request to create one will fail (Bad Request) + The job id to assign to the created job. if missing, job is chosen + randomly using the ULID scheme. If a job id already exists in the queue + or as a completed job, the request to create one will fail (Bad Request) in: query schema: type: string @@ -5240,7 +5286,8 @@ components: type: boolean ArgsFilter: name: args - description: filter on jobs containing those args as a json subset (@> in postgres) + description: + filter on jobs containing those args as a json subset (@> in postgres) in: query schema: type: string @@ -5252,7 +5299,8 @@ components: type: string ResultFilter: name: result - description: filter on jobs containing those result as a json subset (@> in postgres) + description: + filter on jobs containing those result as a json subset (@> in postgres) in: query schema: type: string @@ -5451,7 +5499,6 @@ components: required: - hash - ScriptArgs: type: object additionalProperties: {} diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index fbf86230f3..388c19aa48 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -652,6 +652,7 @@ mod tests { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }, FlowModule { id: "b".to_string(), @@ -671,6 +672,7 @@ mod tests { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }, FlowModule { id: "c".to_string(), @@ -688,6 +690,7 @@ mod tests { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }, ], failure_module: Some(FlowModule { @@ -705,6 +708,7 @@ mod tests { suspend: Default::default(), retry: None, sleep: None, + cache_ttl: None }), same_worker: false, }; diff --git a/backend/windmill-api/src/resources.rs b/backend/windmill-api/src/resources.rs index 22c827935c..1fe4288e30 100644 --- a/backend/windmill-api/src/resources.rs +++ b/backend/windmill-api/src/resources.rs @@ -260,23 +260,33 @@ async fn check_path_conflict<'c>( return Ok(()); } +#[derive(Deserialize)] +struct CreateResourceQuery { + update_if_exists: Option +} async fn create_resource( authed: Authed, Extension(user_db): Extension, Extension(webhook): Extension, Extension(db): Extension, Path(w_id): Path, + Query(q): Query, Json(resource): Json, ) -> Result<(StatusCode, String)> { let authed = maybe_refresh_folders(&resource.path, &w_id, authed, &db).await; let mut tx = user_db.begin(&authed).await?; - check_path_conflict(&mut tx, &w_id, &resource.path).await?; + let update_if_exists = q.update_if_exists.unwrap_or(false); + if !update_if_exists { + check_path_conflict(&mut tx, &w_id, &resource.path).await?; + } + sqlx::query!( "INSERT INTO resource (workspace_id, path, value, description, resource_type) - VALUES ($1, $2, $3, $4, $5)", + VALUES ($1, $2, $3, $4, $5) ON CONFLICT (workspace_id, path) + DO UPDATE SET value = $3, description = $4, resource_type = $5", w_id, resource.path, resource.value, diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 64a44f2d93..82fb9ad637 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -169,6 +169,8 @@ pub struct FlowModule { pub retry: Option, #[serde(skip_serializing_if = "Option::is_none")] pub sleep: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cache_ttl: Option, } impl FlowModule { diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 23547f2079..ae243a73ec 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -572,6 +572,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>( retry: None, sleep: None, suspend: None, + cache_ttl: None }); raw_flow = Some(FlowValue { modules, ..flow.clone() }); } diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index 47a5c029dc..04a9e318af 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -330,12 +330,9 @@ async fn op_resource( if let Some(client) = client { let result = client .get_client() - .get_resource(&client.workspace, path) + .get_resource_value(&client.workspace, path) .await?; - Ok(result - .into_inner() - .value - .unwrap_or_else(|| serde_json::json!({}))) + Ok(result.into_inner()) } else { anyhow::bail!("No client found in op state"); } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index d84d2d9fe7..1a3795632e 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -17,10 +17,12 @@ use std::{ borrow::Borrow, collections::HashMap, io, os::unix::process::ExitStatusExt, panic, process::Stdio, time::{Duration}, sync::{Arc, atomic::Ordering}, + collections::hash_map::DefaultHasher, + hash::{Hasher, Hash}, }; + use tracing::{trace_span, Instrument}; use uuid::Uuid; - use windmill_common::{ error::{self, to_anyhow, Error}, flows::{FlowModuleValue, FlowValue}, @@ -48,6 +50,7 @@ use futures::{ }; use async_recursion::async_recursion; +use windmill_api_client::types::CreateResource; #[cfg(feature = "enterprise")] use rand::Rng; @@ -877,6 +880,13 @@ struct JobCompleted { result: serde_json::Value, logs: String, } + +fn hash_args(v: &serde_json::Value) -> i64 { + let mut dh = DefaultHasher::new(); + serde_json::to_string(v).unwrap().hash(&mut dh); + dh.finish() as i64 +} + #[tracing::instrument(level = "trace", skip_all)] async fn handle_queued_job( job: QueuedJob, @@ -920,7 +930,7 @@ async fn handle_queued_job( logs.push_str(&log_str); } - if job.is_flow_step { + let (cache_ttl, step) = if job.is_flow_step { update_flow_status_in_progress( db, &job.workspace_id, @@ -928,8 +938,27 @@ async fn handle_queued_job( .ok_or_else(|| Error::InternalErr(format!("expected parent job")))?, job.id, ) - .await?; - } + .await? + } else { + (None, None) + }; + + let cached_res_path = if cache_ttl.is_some() { + let flow_path = sqlx::query_scalar!( + "SELECT script_path FROM queue WHERE id = $1", + &job.parent_job.unwrap() + ) + .fetch_one(db) + .await + .map_err(|e| Error::InternalErr(format!("fetching step flow status: {e}")))? + .ok_or_else(|| Error::InternalErr(format!("Expected script_path")))?; + let step = step.unwrap_or(-1); + let args_hash = hash_args(&job.args.clone().unwrap_or_else(|| json!({}))); + let permissioned_as = &job.permissioned_as; + Some(format!("{permissioned_as}/cache/{flow_path}/{step}/{args_hash}")) + } else { + None + }; tracing::debug!( worker = %worker_name, @@ -939,39 +968,65 @@ async fn handle_queued_job( job.id ); - logs.push_str(&format!("job {} on worker {}\n", &job.id, &worker_name)); - let result = match job.job_kind { - JobKind::Dependencies => { - handle_dependency_job(&job, &mut logs, job_dir, db, worker_name, worker_dir).await - } - JobKind::FlowDependencies => { - handle_flow_dependency_job(&job, &mut logs, job_dir, db, worker_name, worker_dir) - .await - .map(|()| Value::Null) - } - JobKind::Identity => match job.args.clone() { - Some(Value::Object(args)) - if args.len() == 1 && args.contains_key("previous_result") => - { - Ok(args.get("previous_result").unwrap().clone()) - } - args @ _ => Ok(args.unwrap_or_else(|| Value::Null)), - }, - _ => { - handle_code_execution_job( - &job, - db, - client, - job_dir, - worker_dir, - &mut logs, - base_internal_url, - worker_name - ) - .await - } + let cached_res = if let Some(cached_res_path) = cached_res_path.clone() { + let authed_client = client.get_authed().await; + let client: &Client = authed_client.get_client(); + let resource = client.get_resource_value(&job.workspace_id, &cached_res_path).await; + resource.ok() + .and_then(|x| { + let v = x.into_inner(); + if let Some(o) = v.as_object() { + let expire = o.get("expire"); + if expire.is_some() && expire.unwrap().as_i64().map(|x| x > chrono::Utc::now().timestamp()).unwrap_or(false) { + v.get("value").map(|x| x.to_owned()) + } else { + None + } + } else { + None + } + }) + } else { + None }; + logs.push_str(&format!("job {} on worker {}\n", &job.id, &worker_name)); + let result = if let Some(cached_res) = cached_res { + Ok(cached_res) + } else { + match job.job_kind { + JobKind::Dependencies => { + handle_dependency_job(&job, &mut logs, job_dir, db, worker_name, worker_dir).await + } + JobKind::FlowDependencies => { + handle_flow_dependency_job(&job, &mut logs, job_dir, db, worker_name, worker_dir) + .await + .map(|()| Value::Null) + } + JobKind::Identity => match job.args.clone() { + Some(Value::Object(args)) + if args.len() == 1 && args.contains_key("previous_result") => + { + Ok(args.get("previous_result").unwrap().clone()) + } + args @ _ => Ok(args.unwrap_or_else(|| Value::Null)), + }, + _ => { + handle_code_execution_job( + &job, + db, + client, + job_dir, + worker_dir, + &mut logs, + base_internal_url, + worker_name + ) + .await + } + } + }; + //it's a test job, no need to update the db if job.workspace_id == "" { return Ok(()); @@ -980,6 +1035,22 @@ async fn handle_queued_job( match result { Ok(r) => { // println!("bef completed job{:?}", SystemTime::now()); + if let Some(cached_path) = cached_res_path { + let client: &Client = client.get_client(); + let expire = chrono::Utc::now().timestamp() + cache_ttl.unwrap() as i64; + let cr = &CreateResource { + path: cached_path, + description: None, + resource_type: "cache".to_string(), + value: serde_json::json!({ + "value": r, + "expire": expire + }) + }; + if let Err(e) = client.create_resource(&job.workspace_id, Some(true), cr).await { + tracing::error!("Error creating cache resource {e}") + } + } if job.is_flow_step { add_completed_job(db, &job, true, false, r.clone(), logs, rsmq.clone()).await?; if let Some(parent_job) = job.parent_job { diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index cc4a8c4e98..207ad8f35b 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -728,25 +728,28 @@ async fn compute_bool_from_expr( } } +type CacheAndStep = (Option, Option); pub async fn update_flow_status_in_progress( db: &DB, w_id: &str, flow: Uuid, job_in_progress: Uuid, -) -> error::Result<()> { +) -> error::Result { let step = get_step_of_flow_status(db, flow).await?; - if let Step::Step(step) = step { - sqlx::query(&format!( + let cache_ttl = if let Step::Step(step) = step { + let ttl = sqlx::query_scalar(&format!( "UPDATE queue SET flow_status = jsonb_set(jsonb_set(flow_status, '{{modules, {step}, job}}', $1), '{{modules, {step}, type}}', $2) - WHERE id = $3 AND workspace_id = $4", + WHERE id = $3 AND workspace_id = $4 + RETURNING (raw_flow->'modules'->{step}->>'cache_ttl')::int as cache_ttl", )) .bind(json!(job_in_progress.to_string())) .bind(json!("InProgress")) .bind(flow) .bind(w_id) - .execute(db) + .fetch_one(db) .await?; + (ttl, Some(step)) } else { sqlx::query(&format!( "UPDATE queue @@ -759,8 +762,9 @@ pub async fn update_flow_status_in_progress( .bind(w_id) .execute(db) .await?; - } - Ok(()) + (None, None) + }; + Ok(cache_ttl) } pub enum Step { diff --git a/frontend/src/lib/components/flows/content/FlowModuleCache.svelte b/frontend/src/lib/components/flows/content/FlowModuleCache.svelte new file mode 100644 index 0000000000..b9309c7d94 --- /dev/null +++ b/frontend/src/lib/components/flows/content/FlowModuleCache.svelte @@ -0,0 +1,42 @@ + + +

+ Cache + + If defined, the result of the step will be cached for the number of seconds defined such that if + this step were to be re-triggered with the same input it would retrieve and return its cached + value instead of recomputing it. + +

+ { + if (isCacheEnabled && flowModule.cache_ttl != undefined) { + flowModule.cache_ttl = undefined + } else { + flowModule.cache_ttl = 60 * 60 * 24 * 2 + } + }} + options={{ + right: 'Cache the results for each possible inputs' + }} +/> +
+ How long to keep cache valid + + {#if flowModule.cache_ttl} + + {:else} + + {/if} +
diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index 6c76ccf857..0e5e1d8681 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -18,6 +18,7 @@ import FlowModuleScript from './FlowModuleScript.svelte' import FlowModuleEarlyStop from './FlowModuleEarlyStop.svelte' import FlowModuleSuspend from './FlowModuleSuspend.svelte' + import FlowModuleCache from './FlowModuleCache.svelte' import FlowRetries from './FlowRetries.svelte' import { getStepPropPicker } from '../previousResults' import { deepEqual } from 'fast-equals' @@ -133,6 +134,7 @@ on:toggleSuspend={() => selectAdvanced('suspend')} on:toggleSleep={() => selectAdvanced('sleep')} on:toggleRetry={() => selectAdvanced('retries')} + on:toggleCache={() => selectAdvanced('cache')} on:toggleStopAfterIf={() => selectAdvanced('early-stop')} on:fork={async () => { const [module, state] = await fork(flowModule) @@ -256,6 +258,7 @@ Retries {#if !$selectedId.includes('failure')} + Cache Early Stop/Break Suspend Sleep @@ -275,6 +278,10 @@
+ {:else if advancedSelected === 'cache'} +
+ +
{:else if advancedSelected === 'same_worker'}
diff --git a/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte b/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte index c1a05632fb..3ee1a773eb 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte @@ -3,7 +3,7 @@ import { WorkerService, type FlowModule } from '$lib/gen' import { faCodeBranch, faPen, faSave } from '@fortawesome/free-solid-svg-icons' import { createEventDispatcher, getContext } from 'svelte' - import { Bed, PhoneIncoming, Repeat, Square } from 'lucide-svelte' + import { Bed, Database, PhoneIncoming, Repeat, Square } from 'lucide-svelte' import Popover from '../../Popover.svelte' import type { FlowEditorContext } from '../types' import { sendUserToast } from '$lib/utils' @@ -39,6 +39,17 @@ Retries + dispatch('toggleCache')} + > + + Cache + Retries {/if} + {#if cache} + +
+ +
+ Cached + + {/if} {#if earlyStop}
) { diff --git a/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte b/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte index 9a89332e34..e387d78572 100644 --- a/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte +++ b/frontend/src/lib/components/propertyPicker/ObjectViewer.svelte @@ -11,7 +11,7 @@ export let level = 0 export let currentPath: string = '' export let pureViewer = false - export let collapsed = level % 3 == 0 || Array.isArray(json) + export let collapsed = (level != 0 && level % 3 == 0) || Array.isArray(json) export let rawKey = false export let topBrackets = false export let topLevelNode = false diff --git a/frontend/src/routes/(root)/(logged)/resources/+page.svelte b/frontend/src/routes/(root)/(logged)/resources/+page.svelte index 9633fa320a..9fec7e13d1 100644 --- a/frontend/src/routes/(root)/(logged)/resources/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/resources/+page.svelte @@ -98,12 +98,20 @@ $: preFilteredType = typeFilter == undefined ? preFilteredItemsOwners?.filter((x) => - tab == 'states' ? x.resource_type == 'state' : x.resource_type != 'state' + tab == 'states' + ? x.resource_type == 'state' + : tab == 'cache' + ? x.resource_type == 'cache' + : x.resource_type != 'state' && x.resource_type != 'cache' ) : preFilteredItemsOwners?.filter( (x) => x.resource_type == typeFilter && - (tab == 'states' ? x.resource_type == 'state' : x.resource_type != 'state') + (tab == 'states' + ? x.resource_type == 'state' + : tab == 'cache' + ? x.resource_type == 'cache' + : x.resource_type != 'state' && x.resource_type != 'cache') ) async function loadResources(): Promise { @@ -207,7 +215,7 @@ } let disableCustomPrefix = false - let tab: 'workspace' | 'types' | 'states' = 'workspace' + let tab: 'workspace' | 'types' | 'states' | 'cache' = 'workspace' let inferrer: Drawer | undefined = undefined let inferrerJson = '' @@ -370,13 +378,22 @@
+ +
+ Cache + + Cached results are actually resources (but excluded from the Workspace tab for clarity). + Cache are used by flows's step to cache result to avoid recomputing unnecessarily + +
+ - {#if tab == 'workspace' || tab == 'states'} + {#if tab == 'workspace' || tab == 'states' || tab == 'cache'}
- {#if tab != 'states'} + {#if tab != 'states' && tab != 'cache'}