kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add cache as a primitive for flows (#1671)

Browse Source 
* feat: add cache as a primitive for flows

* fix failure module
This commit is contained in:
Ruben Fiszel
2023-06-04 14:39:10 +02:00
committed by GitHub
parent a46f117538
commit 027fa6ff09
19 changed files with 359 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
backend/sqlx-data.json
View File

@@ -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": [],

9
backend/tests/worker.rs
View File

@@ -1015,6 +1015,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None
},
],
},
@@ -1217,7 +1223,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
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<Postgres>) {
suspend: Default::default(),
retry: None,
sleep: None,
cache_ttl: None
},
],
same_worker: true,

109
backend/windmill-api/openapi.yaml
View File

@@ -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: {}

4
backend/windmill-api/src/flows.rs
View File

@@ -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,
};

14
backend/windmill-api/src/resources.rs
View File

@@ -260,23 +260,33 @@ async fn check_path_conflict<'c>(
return Ok(());
}
#[derive(Deserialize)]
struct CreateResourceQuery {
update_if_exists: Option<bool>
}
async fn create_resource(
authed: Authed,
Extension(user_db): Extension<UserDB>,
Extension(webhook): Extension<WebhookShared>,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Query(q): Query<CreateResourceQuery>,
Json(resource): Json<CreateResource>,
) -> 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,

2
backend/windmill-common/src/flows.rs
View File

@@ -169,6 +169,8 @@ pub struct FlowModule {
pub retry: Option<Retry>,
#[serde(skip_serializing_if = "Option::is_none")]
pub sleep: Option<InputTransform>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<u32>,
}
impl FlowModule {

1
backend/windmill-queue/src/jobs.rs
View File

@@ -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() });
}

7
backend/windmill-worker/src/js_eval.rs
View File

@@ -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");
}

141
backend/windmill-worker/src/worker.rs
View File

@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: QueuedJob,
@@ -920,7 +930,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
.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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
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 {

18
backend/windmill-worker/src/worker_flow.rs
View File

@@ -728,25 +728,28 @@ async fn compute_bool_from_expr(
}
}
type CacheAndStep = (Option<i32>, Option<i32>);
pub async fn update_flow_status_in_progress(
db: &DB,
w_id: &str,
flow: Uuid,
job_in_progress: Uuid,
) -> error::Result<()> {
) -> error::Result<CacheAndStep> {
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 {

42
frontend/src/lib/components/flows/content/FlowModuleCache.svelte Normal file
View File

@@ -0,0 +1,42 @@
<script lang="ts">
import Toggle from '$lib/components/Toggle.svelte'
import Tooltip from '$lib/components/Tooltip.svelte'
import type { FlowModule } from '$lib/gen'
import { SecondsInput } from '../../common'
export let flowModule: FlowModule
$: isCacheEnabled = Boolean(flowModule.cache_ttl)
</script>
<h2>
Cache
<Tooltip documentationLink="https://docs.windmill.dev/docs/flows/flow_approval">
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.
</Tooltip>
</h2>
<Toggle
checked={isCacheEnabled}
on:change={() => {
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'
}}
/>
<div class="mb-4">
<span class="text-xs font-bold">How long to keep cache valid</span>
{#if flowModule.cache_ttl}
<SecondsInput bind:seconds={flowModule.cache_ttl} />
{:else}
<SecondsInput disabled />
{/if}
</div>

7
frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
View File

@@ -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 @@
<Tabs bind:selected={advancedSelected}>
<Tab value="retries">Retries</Tab>
{#if !$selectedId.includes('failure')}
<Tab value="cache">Cache</Tab>
<Tab value="early-stop">Early Stop/Break</Tab>
<Tab value="suspend">Suspend</Tab>
<Tab value="sleep">Sleep</Tab>
@@ -275,6 +278,10 @@
<div>
<FlowModuleSleep previousModuleId={previousModule?.id} bind:flowModule />
</div>
{:else if advancedSelected === 'cache'}
<div>
<FlowModuleCache bind:flowModule />
</div>
{:else if advancedSelected === 'same_worker'}
<div>
<Alert type="info" title="Share a directory between steps">

13
frontend/src/lib/components/flows/content/FlowModuleHeader.svelte
View File

@@ -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 @@
<Repeat size={14} />
<svelte:fragment slot="text">Retries</svelte:fragment>
</Popover>
<Popover
placement="bottom"
class="center-center rounded border p-2
{module.cache_ttl != undefined
? 'bg-blue-100 text-blue-800 border-blue-300 hover:bg-blue-200'
: 'bg-white text-gray-800 border-gray-300 hover:bg-gray-100'}"
on:click={() => dispatch('toggleCache')}
>
<Database size={14} />
<svelte:fragment slot="text">Cache</svelte:fragment>
</Popover>
<Popover
placement="bottom"
class="center-center rounded border p-2

7
frontend/src/lib/components/flows/content/FlowModuleSuspend.svelte
View File

@@ -3,17 +3,10 @@
import Tooltip from '$lib/components/Tooltip.svelte'
import type { FlowModule } from '$lib/gen'
import { emptySchema } from '$lib/utils'
import { SecondsInput } from '../../common'
export let flowModule: FlowModule
let schema = emptySchema()
schema.properties['sleep'] = {
type: 'number',
description: 'Sleep time in seconds'
}
$: isSuspendEnabled = Boolean(flowModule.suspend)
</script>

14
frontend/src/lib/components/flows/map/FlowModuleSchemaItem.svelte
View File

@@ -2,13 +2,14 @@
import Badge from '$lib/components/common/badge/Badge.svelte'
import Popover from '$lib/components/Popover.svelte'
import { classNames } from '$lib/utils'
import { Bed, Move, PhoneIncoming, Repeat, Square, X } from 'lucide-svelte'
import { Bed, Database, Move, PhoneIncoming, Repeat, Square, X } from 'lucide-svelte'
import { createEventDispatcher } from 'svelte'
import { fade } from 'svelte/transition'
export let selected: boolean = false
export let deletable: boolean = false
export let retry: boolean = false
export let cache: boolean = false
export let earlyStop: boolean = false
export let suspend: boolean = false
export let sleep: boolean = false
@@ -43,6 +44,17 @@
<svelte:fragment slot="text">Retries</svelte:fragment>
</Popover>
{/if}
{#if cache}
<Popover notClickable>
<div
transition:fade|local={{ duration: 200 }}
class="center-center rounded border bg-white border-gray-400 text-gray-700 px-1 py-0.5"
>
<Database size={14} />
</div>
<svelte:fragment slot="text">Cached</svelte:fragment>
</Popover>
{/if}
{#if earlyStop}
<Popover notClickable>
<div

3
frontend/src/lib/components/flows/map/MapItem.svelte
View File

@@ -41,7 +41,8 @@
retry: mod.retry?.constant != undefined || mod.retry?.exponential != undefined,
earlyStop: mod.stop_after_if != undefined,
suspend: Boolean(mod.suspend),
sleep: Boolean(mod.sleep)
sleep: Boolean(mod.sleep),
cache: Boolean(mod.cache_ttl)
}
function onDelete(event: CustomEvent<MouseEvent>) {

2
frontend/src/lib/components/propertyPicker/ObjectViewer.svelte
View File

@@ -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

27
frontend/src/routes/(root)/(logged)/resources/+page.svelte
View File

@@ -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<void> {
@@ -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 @@
</Tooltip>
</div>
</Tab>
<Tab size="md" value="cache">
<div class="flex gap-2 items-center my-1">
Cache
<Tooltip>
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
</Tooltip>
</div>
</Tab>
</Tabs>
{#if tab == 'workspace' || tab == 'states'}
{#if tab == 'workspace' || tab == 'states' || tab == 'cache'}
<div class="pt-2">
<input placeholder="Search Resource" bind:value={filter} class="input mt-1" />
</div>
<ListFilters bind:selectedFilter={ownerFilter} filters={owners} />
{#if tab != 'states'}
{#if tab != 'states' && tab != 'cache'}
<ListFilters
queryName="app_filter"
bind:selectedFilter={typeFilter}

2
openflow.openapi.yaml
View File

@@ -88,6 +88,8 @@ components:
- expr
sleep:
$ref: "#/components/schemas/InputTransform"
cache_ttl:
type: number
summary:
type: string
suspend: