starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-11 00:06:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add caching to flows and scripts (#2193)

Browse Source 
* cachettl

* all
This commit is contained in:
Ruben Fiszel
2023-08-29 13:00:20 +02:00
committed by GitHub
parent af2a417902
commit 03e48a4ca5
28 changed files with 457 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-8f0d98661f2a32659f37a123320d14288ce88c9d6e6279bdc36334410023f5e1.json → backend/.sqlx/query-0694f4be1a7966126f20afbd9ee094088db8ce98aefdbe96144c42e38e5bd813.json
Generated
+8 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select tag, concurrent_limit, concurrency_time_window_s from script where hash = $1 AND workspace_id = $2",
"query": "select tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
@@ -17,6 +17,11 @@
"ordinal": 2,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 3,
"name": "cache_ttl",
"type_info": "Int4"
}
],
"parameters": {
@@ -26,10 +31,11 @@
]
},
"nullable": [
true,
true,
true,
true
]
},
"hash": "8f0d98661f2a32659f37a123320d14288ce88c9d6e6279bdc36334410023f5e1"
"hash": "0694f4be1a7966126f20afbd9ee094088db8ce98aefdbe96144c42e38e5bd813"
}
backend/.sqlx/query-9289709b2727425be0e7f1e341d6acd46e1e79d5018f6a18cb41d50ca3f4ba11.json → backend/.sqlx/query-35d2c5708645e60dd5ab56d1e6cfcd197254a534fed178a46de4a2d0f0f5d348.json
Generated
+8 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select hash, tag, concurrent_limit, concurrency_time_window_s from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)",
"query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false AND lock IS not NULL AND lock_error_logs IS NULL)",
"describe": {
"columns": [
{
@@ -22,6 +22,11 @@
"ordinal": 3,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 4,
"name": "cache_ttl",
"type_info": "Int4"
}
],
"parameters": {
@@ -34,8 +39,9 @@
false,
true,
true,
true,
true
]
},
"hash": "9289709b2727425be0e7f1e341d6acd46e1e79d5018f6a18cb41d50ca3f4ba11"
"hash": "35d2c5708645e60dd5ab56d1e6cfcd197254a534fed178a46de4a2d0f0f5d348"
}
backend/.sqlx/query-9ee1423945740c17dedc2fc7ea7e7183380c43d0feeb615c651aa486eef3bd53.json → backend/.sqlx/query-52dbe77fad5729e9e121609eeb6a3573e6d0e1d1c16e29b15c7ad810536e249c.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)",
"query": "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)",
"describe": {
"columns": [],
"parameters": {
@@ -57,10 +57,11 @@
"Bool",
"VarcharArray",
"Int4",
"Int4",
"Int4"
]
},
"nullable": []
},
"hash": "9ee1423945740c17dedc2fc7ea7e7183380c43d0feeb615c651aa486eef3bd53"
"hash": "52dbe77fad5729e9e121609eeb6a3573e6d0e1d1c16e29b15c7ad810536e249c"
}
backend/.sqlx/query-01d5e4e371a507602fca01190965a4f20ff806315a9ed02b372977e2d9479338.json → backend/.sqlx/query-9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b.json
Generated
+4 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id)\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28) RETURNING id",
"query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl)\n VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29) RETURNING id",
"describe": {
"columns": [
{
@@ -80,12 +80,13 @@
"Int4",
"Int4",
"Int4",
"Varchar"
"Varchar",
"Int4"
]
},
"nullable": [
false
]
},
"hash": "01d5e4e371a507602fca01190965a4f20ff806315a9ed02b372977e2d9479338"
"hash": "9e8c3ff3d6b31e366e15beda1e96e03e870ccc3b353401439bc0ed8ff219249b"
}
backend/.sqlx/query-7435be9a71a6b30980d180047486a47d69e36132d40ef0f53448e96a9866842b.json → backend/.sqlx/query-cf21b61aa5b5e47744235c860e2d9b90d7654a4b5d0ec32085b178399f701bfc.json
Generated
+8 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select hash, tag, concurrent_limit, concurrency_time_window_s from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false AND lock IS not NULL AND lock_error_logs IS NULL)",
"query": "select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n deleted = false AND archived = false)",
"describe": {
"columns": [
{
@@ -22,6 +22,11 @@
"ordinal": 3,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 4,
"name": "cache_ttl",
"type_info": "Int4"
}
],
"parameters": {
@@ -34,8 +39,9 @@
false,
true,
true,
true,
true
]
},
"hash": "7435be9a71a6b30980d180047486a47d69e36132d40ef0f53448e96a9866842b"
"hash": "cf21b61aa5b5e47744235c860e2d9b90d7654a4b5d0ec32085b178399f701bfc"
}
backend/.sqlx/query-d164caec6637ab53d8c2e0faad653a458f31a976bb23c3a43d35b8476b066954.json → backend/.sqlx/query-ebac391c83dae67831502fbe3f1034210afce7902a69f7c42bb68160e0464d71.json
Generated
+8 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select path, tag, concurrent_limit, concurrency_time_window_s from script where hash = $1 AND workspace_id = $2",
"query": "select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
@@ -22,6 +22,11 @@
"ordinal": 3,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 4,
"name": "cache_ttl",
"type_info": "Int4"
}
],
"parameters": {
@@ -34,8 +39,9 @@
false,
true,
true,
true,
true
]
},
"hash": "d164caec6637ab53d8c2e0faad653a458f31a976bb23c3a43d35b8476b066954"
"hash": "ebac391c83dae67831502fbe3f1034210afce7902a69f7c42bb68160e0464d71"
}
backend/migrations/20230828202124_add_cache_ttl.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20230828202124_add_cache_ttl.up.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add up migration script here
ALTER TABLE script ADD COLUMN cache_ttl INTEGER;
ALTER TABLE queue ADD COLUMN cache_ttl INTEGER;
backend/tests/worker.rs
+5
View File
@@ -1619,6 +1619,7 @@ func main(derp string) (string, error) {
language: ScriptLang::Go,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
}))
.arg("derp", json!("world"))
.run_until_complete(&db, port)
@@ -1648,6 +1649,7 @@ echo "hello $msg"
language: ScriptLang::Bash,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
}))
.arg("msg", json!("world"))
.run_until_complete(&db, port)
@@ -1675,6 +1677,7 @@ def main():
lock: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
});
let result = run_job_in_new_worker_until_complete(&db, job, port)
@@ -1707,6 +1710,7 @@ def main():
lock: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
});
let result = run_job_in_new_worker_until_complete(&db, job, port)
@@ -1738,6 +1742,7 @@ def main():
lock: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
});
let result = run_job_in_new_worker_until_complete(&db, job, port)
backend/windmill-api/openapi.yaml
+4 -1
View File
@@ -5831,7 +5831,8 @@ components:
type: array
items:
type: integer
cache_ttl:
type: number
required:
- hash
- path
@@ -5905,6 +5906,8 @@ components:
type: array
items:
type: integer
cache_ttl:
type: number
required:
- path
- summary
backend/windmill-api/src/flows.rs
+3 -2
View File
@@ -210,7 +210,7 @@ async fn create_flow(
w_id,
nf.path,
nf.summary,
nf.description,
nf.description.unwrap_or_else(String::new),
nf.value,
&authed.username,
nf.schema.and_then(|x| serde_json::to_string(&x.0).ok()),
@@ -352,7 +352,7 @@ async fn update_flow(
edited_at = now(), schema = $6::text::json, dependency_job = NULL, draft_only = NULL, tag = $9 WHERE path = $7 AND workspace_id = $8",
nf.path,
nf.summary,
nf.description,
nf.description.unwrap_or_else(String::new),
nf.value,
&authed.username,
schema.and_then(|x| serde_json::to_string(&x).ok()),
@@ -752,6 +752,7 @@ mod tests {
concurrent_limit: None,
concurrency_time_window_s: None,
skip_expr: None,
cache_ttl: None,
};
let expect = serde_json::json!({
"modules": [
backend/windmill-api/src/jobs.rs
+18 -7
View File
@@ -305,13 +305,19 @@ pub async fn get_path_for_hash<'c>(
Ok(path)
}
pub async fn get_path_tag_and_limits_for_hash(
pub async fn get_path_tag_limits_cache_for_hash(
db: &DB,
w_id: &str,
hash: i64,
) -> error::Result<(String, Option<String>, Option<i32>, Option<i32>)> {
) -> error::Result<(
String,
Option<String>,
Option<i32>,
Option<i32>,
Option<i32>,
)> {
let script = sqlx::query!(
"select path, tag, concurrent_limit, concurrency_time_window_s from script where hash = $1 AND workspace_id = $2",
"select path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where hash = $1 AND workspace_id = $2",
hash,
w_id
)
@@ -327,6 +333,7 @@ pub async fn get_path_tag_and_limits_for_hash(
script.tag,
script.concurrent_limit,
script.concurrency_time_window_s,
script.cache_ttl,
))
}
@@ -1340,6 +1347,7 @@ impl From<UnifiedJob> for Job {
concurrency_time_window_s: uj.concurrency_time_window_s,
timeout: None,
flow_step_id: None,
cache_ttl: None,
}),
t => panic!("job type {} not valid", t),
}
@@ -2027,8 +2035,8 @@ pub async fn run_wait_result_script_by_hash(
check_queue_too_long(&db, run_query.queue_limit).await?;
let hash = script_hash.0;
let (path, tag, concurrent_limit, concurrency_time_window_s) =
get_path_tag_and_limits_for_hash(&db, &w_id, hash).await?;
let (path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) =
get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
check_scopes(&authed, || format!("run:script/{path}"))?;
let args = run_query.add_include_headers(headers, args.unwrap_or_default());
@@ -2045,6 +2053,7 @@ pub async fn run_wait_result_script_by_hash(
path: path,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
cache_ttl,
},
args,
&authed.username,
@@ -2218,6 +2227,7 @@ async fn run_preview_job(
lock: None,
concurrent_limit: None, // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here
concurrency_time_window_s: None, // TODO(gbouv): same as above
cache_ttl: None,
}),
},
args,
@@ -2350,8 +2360,8 @@ pub async fn run_job_by_hash(
JsonOrForm(args, raw_string): JsonOrForm,
) -> error::Result<(StatusCode, String)> {
let hash = script_hash.0;
let (path, tag, concurrent_limit, concurrency_time_window_s) =
get_path_tag_and_limits_for_hash(&db, &w_id, hash).await?;
let (path, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) =
get_path_tag_limits_cache_for_hash(&db, &w_id, hash).await?;
check_scopes(&authed, || format!("run:script/{path}"))?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
@@ -2369,6 +2379,7 @@ pub async fn run_job_by_hash(
path: path,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
cache_ttl,
},
args,
&authed.username,
backend/windmill-api/src/oauth2.rs
+2 -1
View File
@@ -799,7 +799,7 @@ async fn slack_command(
(JobPayload::Flow(path.to_string()), None)
} else {
let path = path.strip_prefix("script/").unwrap_or_else(|| path);
let (script_hash, tag, concurrent_limit, concurrency_time_window_s) =
let (script_hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) =
windmill_common::get_latest_deployed_hash_for_path(
&db,
&settings.workspace_id,
@@ -812,6 +812,7 @@ async fn slack_command(
path: path.to_owned(),
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
},
tag,
)
backend/windmill-api/src/scripts.rs
+7 -4
View File
@@ -69,6 +69,8 @@ pub struct ScriptWDraft {
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<i32>,
}
pub fn global_service() -> Router {
@@ -370,8 +372,8 @@ async fn create_script(
sqlx::query!(
"INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \
content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \
draft_only, envs, concurrent_limit, concurrency_time_window_s) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)",
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)",
&w_id,
&hash.0,
ns.path,
@@ -390,7 +392,8 @@ async fn create_script(
ns.draft_only,
envs,
ns.concurrent_limit,
ns.concurrency_time_window_s
ns.concurrency_time_window_s,
ns.cache_ttl,
)
.execute(&mut tx)
.await?;
@@ -587,7 +590,7 @@ async fn get_script_by_path_w_draft(
let mut tx = user_db.begin(&authed).await?;
let script_o = sqlx::query_as::<_, ScriptWDraft>(
"SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, draft.value as draft FROM script LEFT JOIN draft ON
"SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, draft.value as draft FROM script LEFT JOIN draft ON
script.path = draft.path AND script.workspace_id = draft.workspace_id AND draft.typ = 'script'
WHERE script.path = $1 AND script.workspace_id = $2 \
AND script.created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND \
backend/windmill-common/src/flows.rs
+3 -1
View File
@@ -60,7 +60,7 @@ pub struct ListableFlow {
pub struct NewFlow {
pub path: String,
pub summary: String,
pub description: String,
pub description: Option<String>,
pub value: serde_json::Value,
pub schema: Option<Schema>,
pub draft_only: Option<bool>,
@@ -83,6 +83,8 @@ pub struct FlowValue {
pub concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub skip_expr: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<u32>,
}
#[derive(Deserialize, Serialize, Debug, Clone)]
backend/windmill-common/src/jobs.rs
+13 -3
View File
@@ -90,6 +90,8 @@ pub struct QueuedJob {
pub timeout: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub flow_step_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<i32>,
}
impl QueuedJob {
@@ -99,6 +101,9 @@ impl QueuedJob {
.map(String::as_str)
.unwrap_or("tmp/main")
}
pub fn is_flow(&self) -> bool {
matches!(self.job_kind, JobKind::Flow | JobKind::FlowPreview)
}
}
impl QueuedJob {
@@ -156,6 +161,7 @@ impl Default for QueuedJob {
concurrency_time_window_s: None,
timeout: None,
flow_step_id: None,
cache_ttl: None,
}
}
}
@@ -170,6 +176,7 @@ pub enum JobPayload {
path: String,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
cache_ttl: Option<i32>,
},
Code(RawCode),
Dependencies {
@@ -202,6 +209,7 @@ pub struct RawCode {
pub lock: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub cache_ttl: Option<i32>,
}
type Tag = String;
@@ -216,7 +224,7 @@ pub async fn script_path_to_payload(
let (job_payload, tag) = if script_path.starts_with("hub/") {
(JobPayload::ScriptHub { path: script_path.to_owned() }, None)
} else {
let (script_hash, tag, concurrent_limit, concurrency_time_window_s) =
let (script_hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) =
get_latest_deployed_hash_for_path(db, w_id, script_path).await?;
(
JobPayload::ScriptHash {
@@ -224,6 +232,7 @@ pub async fn script_path_to_payload(
path: script_path.to_owned(),
concurrent_limit,
concurrency_time_window_s,
cache_ttl: cache_ttl,
},
tag,
)
@@ -235,9 +244,9 @@ pub async fn script_hash_to_tag_and_limits<'c>(
script_hash: &ScriptHash,
db: &mut Transaction<'c, Postgres>,
w_id: &String,
) -> error::Result<(Option<Tag>, Option<i32>, Option<i32>)> {
) -> error::Result<(Option<Tag>, Option<i32>, Option<i32>, Option<i32>)> {
let script = sqlx::query!(
"select tag, concurrent_limit, concurrency_time_window_s from script where hash = $1 AND workspace_id = $2",
"select tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where hash = $1 AND workspace_id = $2",
script_hash.0,
w_id
)
@@ -252,6 +261,7 @@ pub async fn script_hash_to_tag_and_limits<'c>(
script.tag,
script.concurrent_limit,
script.concurrency_time_window_s,
script.cache_ttl,
))
}
backend/windmill-common/src/lib.rs
+18 -4
View File
@@ -163,9 +163,15 @@ pub async fn get_latest_deployed_hash_for_path(
db: &DB,
w_id: &str,
script_path: &str,
) -> error::Result<(scripts::ScriptHash, Option<Tag>, Option<i32>, Option<i32>)> {
) -> error::Result<(
scripts::ScriptHash,
Option<Tag>,
Option<i32>,
Option<i32>,
Option<i32>,
)> {
let r_o = sqlx::query!(
"select hash, tag, concurrent_limit, concurrency_time_window_s from script where path = $1 AND workspace_id = $2 AND
"select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where path = $1 AND workspace_id = $2 AND
created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND
deleted = false AND archived = false AND lock IS not NULL AND lock_error_logs IS NULL)",
script_path,
@@ -181,6 +187,7 @@ pub async fn get_latest_deployed_hash_for_path(
script.tag,
script.concurrent_limit,
script.concurrency_time_window_s,
script.cache_ttl,
))
}
@@ -188,9 +195,15 @@ pub async fn get_latest_hash_for_path<'c>(
db: &mut sqlx::Transaction<'c, sqlx::Postgres>,
w_id: &str,
script_path: &str,
) -> error::Result<(scripts::ScriptHash, Option<Tag>, Option<i32>, Option<i32>)> {
) -> error::Result<(
scripts::ScriptHash,
Option<Tag>,
Option<i32>,
Option<i32>,
Option<i32>,
)> {
let r_o = sqlx::query!(
"select hash, tag, concurrent_limit, concurrency_time_window_s from script where path = $1 AND workspace_id = $2 AND
"select hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl from script where path = $1 AND workspace_id = $2 AND
created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND
deleted = false AND archived = false)",
script_path,
@@ -206,5 +219,6 @@ pub async fn get_latest_hash_for_path<'c>(
script.tag,
script.concurrent_limit,
script.concurrency_time_window_s,
script.cache_ttl,
))
}
backend/windmill-common/src/scripts.rs
+1
View File
@@ -214,6 +214,7 @@ pub struct NewScript {
pub envs: Option<Vec<String>>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub cache_ttl: Option<i32>,
}
#[derive(Deserialize)]
backend/windmill-queue/src/jobs.rs
+43 -6
View File
@@ -1409,8 +1409,15 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
language,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
) = match job_payload {
JobPayload::ScriptHash { hash, path, concurrent_limit, concurrency_time_window_s } => {
JobPayload::ScriptHash {
hash,
path,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
} => {
let language = fetch_scalar_isolated!(sqlx::query_scalar!(
"SELECT language as \"language: ScriptLang\" FROM script WHERE hash = $1 AND workspace_id = $2",
hash.0,
@@ -1431,6 +1438,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
Some(language),
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
)
}
JobPayload::ScriptHub { path } => {
@@ -1444,6 +1452,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
None,
None,
None,
None,
)
}
JobPayload::Code(RawCode {
@@ -1453,6 +1462,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
lock,
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
}) => (
None,
path,
@@ -1462,6 +1472,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
Some(language),
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
),
JobPayload::Dependencies { hash, dependencies, language, path } => (
Some(hash.0),
@@ -1472,6 +1483,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
Some(language),
None,
None,
None,
),
JobPayload::FlowDependencies { path } => {
let value_json = fetch_scalar_isolated!(
@@ -1497,6 +1509,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
None,
None,
None,
None,
)
}
JobPayload::AppDependencies { path, version } => (
@@ -1508,6 +1521,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
None,
None,
None,
None,
),
JobPayload::RawFlow { value, path } => (
None,
@@ -1518,6 +1532,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
None,
value.concurrent_limit.clone(),
value.concurrency_time_window_s,
value.cache_ttl.map(|x| x as i32),
),
JobPayload::Flow(flow) => {
let value_json = fetch_scalar_isolated!(
@@ -1543,10 +1558,31 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
None,
value.concurrent_limit.clone(),
value.concurrency_time_window_s,
value.cache_ttl.map(|x| x as i32),
)
}
JobPayload::Identity => (None, None, None, JobKind::Identity, None, None, None, None),
JobPayload::Noop => (None, None, None, JobKind::Noop, None, None, None, None),
JobPayload::Identity => (
None,
None,
None,
JobKind::Identity,
None,
None,
None,
None,
None,
),
JobPayload::Noop => (
None,
None,
None,
JobKind::Noop,
None,
None,
None,
None,
None,
),
};
let is_running = same_worker;
@@ -1657,8 +1693,8 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
"INSERT INTO queue
(workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for,
script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, \
flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id)
VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28) \
flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl)
VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, now()), $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29) \
RETURNING id",
workspace_id,
job_id,
@@ -1687,7 +1723,8 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
concurrent_limit,
concurrency_time_window_s,
custom_timeout,
flow_step_id
flow_step_id,
cache_ttl
)
.fetch_one(&mut tx)
.await
backend/windmill-queue/src/schedule.rs
+2 -1
View File
@@ -74,7 +74,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
let (payload, tag) = if schedule.is_flow {
(JobPayload::Flow(schedule.script_path), None)
} else {
let (hash, tag, concurrent_limit, concurrency_time_window_s) =
let (hash, tag, concurrent_limit, concurrency_time_window_s, cache_ttl) =
windmill_common::get_latest_hash_for_path(
tx.transaction_mut(),
&schedule.workspace_id,
@@ -87,6 +87,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
path: schedule.script_path,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
cache_ttl: cache_ttl,
},
tag,
)
backend/windmill-worker/src/common.rs
+33 -1
View File
@@ -2,6 +2,7 @@ use async_recursion::async_recursion;
use serde_json::{json, Value};
use sqlx::{Pool, Postgres};
use tokio::{fs::File, io::AsyncReadExt};
use windmill_api_client::{types::CreateResource, Client};
use windmill_common::{
error::{self, Error},
jobs::QueuedJob,
@@ -10,7 +11,12 @@ use windmill_queue::CLOUD_HOSTED;
use anyhow::Result;
use std::{
borrow::Borrow, collections::HashMap, io, os::unix::process::ExitStatusExt, panic,
borrow::Borrow,
collections::{hash_map::DefaultHasher, HashMap},
hash::{Hash, Hasher},
io,
os::unix::process::ExitStatusExt,
panic,
time::Duration,
};
@@ -566,6 +572,32 @@ fn append_with_limit(dst: &mut String, src: &str, limit: &mut usize) {
}
}
pub 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
}
pub async fn save_in_cache(client: &AuthedClient, job: &QueuedJob, cached_path: String, r: &Value) {
let client: &Client = client.get_client();
let expire = chrono::Utc::now().timestamp() + job.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}")
}
}
/* TODO retry this? */
#[tracing::instrument(level = "trace", skip_all)]
async fn append_logs(job_id: uuid::Uuid, logs: impl AsRef<str>, db: impl Borrow<Pool<Postgres>>) {
backend/windmill-worker/src/worker.rs
+165 -167
View File
@@ -14,9 +14,7 @@ use sqlx::{Pool, Postgres};
use windmill_api_client::Client;
use std::{
collections::HashMap, time::Duration,
sync::{Arc, atomic::Ordering},
collections::hash_map::DefaultHasher,
hash::{Hasher, Hash},
sync::{Arc, atomic::Ordering}
};
use tracing::{trace_span, Instrument};
@@ -26,7 +24,7 @@ use windmill_common::{
flows::{FlowModuleValue, FlowValue, FlowModule},
scripts::{ScriptHash, ScriptLang, get_full_hub_script_by_path},
utils::{rd_string, StripPath},
users::SUPERADMIN_SECRET_EMAIL, jobs::{JobKind, QueuedJob, Metrics}, METRICS_ENABLED, IS_READY,
users::SUPERADMIN_SECRET_EMAIL, jobs::{JobKind, QueuedJob, Metrics}, METRICS_ENABLED, IS_READY, DB,
};
use windmill_queue::{canceled_job_to_result, get_queued_job, pull, CLOUD_HOSTED, HTTP_CLIENT, ACCEPTED_TAGS, IS_WORKER_TAGS_DEFINED};
@@ -43,7 +41,6 @@ use tokio::{
use futures::future::FutureExt;
use async_recursion::async_recursion;
use windmill_api_client::types::CreateResource;
#[cfg(feature = "enterprise")]
use rand::Rng;
@@ -55,8 +52,8 @@ use windmill_queue::{add_completed_job, add_completed_job_error,IDLE_WORKERS};
use crate::{
worker_flow::{
handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress,
}, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs, write_file, transform_json_value}, go_executor::{handle_go_job, install_go_dependencies}, js_eval::{transpile_ts, eval_fetch_timeout}, pg_executor::do_postgresql, mysql_executor::do_mysql, graphql_executor::do_graphql, bun_executor::{handle_bun_job, gen_lockfile}, bash_executor::{ANSI_ESCAPE_RE, handle_powershell_job, handle_bash_job}, deno_executor::{handle_deno_job, generate_deno_lock},
handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress,
}, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs, write_file, transform_json_value, save_in_cache, hash_args}, go_executor::{handle_go_job, install_go_dependencies}, js_eval::{transpile_ts, eval_fetch_timeout}, pg_executor::do_postgresql, mysql_executor::do_mysql, graphql_executor::do_graphql, bun_executor::{handle_bun_job, gen_lockfile}, bash_executor::{ANSI_ESCAPE_RE, handle_powershell_job, handle_bash_job}, deno_executor::{handle_deno_job, generate_deno_lock},
};
#[cfg(feature = "enterprise")]
@@ -908,12 +905,6 @@ pub struct JobCompleted {
pub success: bool
}
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
}
pub async fn get_content(job: &QueuedJob, db: &Pool<Postgres>) -> Result<String, Error> {
@@ -960,7 +951,7 @@ async fn do_nativets(job: QueuedJob, logs: String, client: &AuthedClient, code:
#[tracing::instrument(level = "trace", skip_all)]
async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
db: &DB,
client: &AuthedClientBackgroundTask,
worker_name: &str,
worker_dir: &str,
@@ -978,20 +969,63 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
return Err(Error::ExecutionErr(e));
}
let (cache_ttl, step) = if job.is_flow_step {
update_flow_status_in_progress(
let step = if job.is_flow_step {
let r = update_flow_status_in_progress(
db,
&job.workspace_id,
job.parent_job
.ok_or_else(|| Error::InternalErr(format!("expected parent job")))?,
job.id,
)
.await?
.await?;
r
} else {
(None, None)
None
};
let cached_res_path = if job.cache_ttl.is_some() {
let args_hash = hash_args(&job.args.clone().unwrap_or_else(|| json!({})));
let permissioned_as = &job.permissioned_as;
if job.is_flow_step {
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);
Some(format!("{permissioned_as}/cache/{flow_path}/{step}/{args_hash}"))
} else if let Some(script_path) = &job.script_path {
let is_flow = if job.is_flow() { "flow/" } else { "" };
Some(format!("{permissioned_as}/cache/{is_flow}{script_path}/{args_hash}"))
} else {
None
}
} else {
None
};
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;
if let Ok(resource) = resource {
let v = resource.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) {
let result = v.get("value").map(|x| x.to_owned()).unwrap_or_else(|| json!({}));
let logs = "Job skipped because args & path found in cache and not expired".to_string();
process_result(authed_client, job, Ok(result), None, db, worker_dir, job_dir, metrics, same_worker_tx, base_internal_url, rsmq, job_completed_tx, logs).await?;
return Ok(())
}
}
}
};
match job.job_kind {
JobKind::FlowPreview | JobKind::Flow => {
let args = job.args.clone().unwrap_or(Value::Null);
@@ -1019,25 +1053,6 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
set_logs(&logs, &job.id, db).await;
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,
job_id = %job.id,
@@ -1046,32 +1061,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job.id
);
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
};
let result = if let Some(cached_res) = cached_res {
Ok(cached_res)
} else {
match job.job_kind {
let result = match job.job_kind {
JobKind::Dependencies => {
handle_dependency_job(&job, &mut logs, job_dir, db, worker_name, worker_dir, base_internal_url, &client.get_token().await).await
}
@@ -1105,119 +1095,127 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
worker_name )
.await
}
}
};
};
//it's a test job, no need to update the db
if job.workspace_id == "" {
return Ok(());
}
let client = &client.get_authed().await;
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 {
update_flow_status_after_job_completion(
db,
client,
parent_job,
&job.id,
&job.workspace_id,
true,
r,
metrics.clone(),
false,
same_worker_tx.clone(),
worker_dir,
None,
base_internal_url,
rsmq.clone()
)
.await?;
}
} else {
// in the happy path and if job not a flow step, we can delegate updating the completed job in the background
job_completed_tx.send(JobCompleted{job,result:r,logs:logs, success: true}).await.expect("send job completed");
}
}
Err(e) => {
let error_value = match e {
Error::ExitStatus(i) => {
let res = read_result(job_dir).await.ok();
if res.is_some() && res.clone().unwrap().is_object() {
res.unwrap()
} else {
let last_10_log_lines = logs
.lines()
.skip(logs.lines().count().max(13) - 13)
.join("\n")
.to_string()
.replace("\n\n", "\n");
let log_lines = last_10_log_lines
.split("CODE EXECUTION ---")
.last()
.unwrap_or(&logs);
extract_error_value(log_lines, i)
}
}
err @ _ => {
json!({"message": format!("error during execution of the script:\n{}", err), "name": "ExecutionErr"})
}
};
let result =
add_completed_job_error(db, &job, logs, error_value, metrics.clone(), rsmq.clone())
.await?;
if job.is_flow_step {
if let Some(parent_job) = job.parent_job {
update_flow_status_after_job_completion(
db,
client,
parent_job,
&job.id,
&job.workspace_id,
false,
result,
metrics,
false,
same_worker_tx,
worker_dir,
None,
base_internal_url,
rsmq
)
.await?;
}
}
}
};
let client = client.get_authed().await;
process_result(client, job, result, cached_res_path, db, worker_dir, job_dir, metrics, same_worker_tx, base_internal_url, rsmq, job_completed_tx, logs).await?;
}
}
Ok(())
}
async fn process_result<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
client: AuthedClient,
job: QueuedJob,
result: error::Result<serde_json::Value>,
cached_res_path: Option<String>,
db: &DB,
worker_dir: &str,
job_dir: &str,
metrics: Option<Metrics>,
same_worker_tx: Sender<Uuid>,
base_internal_url: &str,
rsmq: Option<R>,
job_completed_tx: Sender<JobCompleted>,
logs: String,
) -> error::Result<()> {
match result {
Ok(r) => {
// println!("bef completed job{:?}", SystemTime::now());
if let Some(cached_path) = cached_res_path {
save_in_cache(&client, &job, cached_path, &r).await;
}
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 {
update_flow_status_after_job_completion(
db,
&client,
parent_job,
&job.id,
&job.workspace_id,
true,
r,
metrics.clone(),
false,
same_worker_tx.clone(),
worker_dir,
None,
base_internal_url,
rsmq.clone()
)
.await?;
}
} else {
// in the happy path and if job not a flow step, we can delegate updating the completed job in the background
job_completed_tx.send(JobCompleted{job,result:r,logs:logs, success: true}).await.expect("send job completed");
}
}
Err(e) => {
let error_value = match e {
Error::ExitStatus(i) => {
let res = read_result(job_dir).await.ok();
if res.is_some() && res.clone().unwrap().is_object() {
res.unwrap()
} else {
let last_10_log_lines = logs
.lines()
.skip(logs.lines().count().max(13) - 13)
.join("\n")
.to_string()
.replace("\n\n", "\n");
let log_lines = last_10_log_lines
.split("CODE EXECUTION ---")
.last()
.unwrap_or(&logs);
extract_error_value(log_lines, i)
}
}
err @ _ => {
json!({"message": format!("error during execution of the script:\n{}", err), "name": "ExecutionErr"})
}
};
let result =
add_completed_job_error(db, &job, logs, error_value, metrics.clone(), rsmq.clone())
.await?;
if job.is_flow_step {
if let Some(parent_job) = job.parent_job {
update_flow_status_after_job_completion(
db,
&client,
parent_job,
&job.id,
&job.workspace_id,
false,
result,
metrics,
false,
same_worker_tx,
worker_dir,
None,
base_internal_url,
rsmq
)
.await?;
}
}
}
};
Ok(())
}
#[tracing::instrument(level = "trace", skip_all)]
async fn handle_code_execution_job(
job: &QueuedJob,
backend/windmill-worker/src/worker_flow.rs
+27 -12
View File
@@ -9,6 +9,7 @@
use std::collections::HashMap;
use std::time::Duration;
use crate::common::{hash_args, save_in_cache};
use crate::js_eval::{eval_timeout, IdContext};
use crate::{AuthedClient, KEEP_JOB_DIR};
use anyhow::Context;
@@ -580,6 +581,16 @@ pub async fn update_flow_status_after_job_completion_internal<
)
.await?;
} else {
if flow_job.cache_ttl.is_some() {
let cached_res_path = {
let args_hash = hash_args(&flow_job.args.clone().unwrap_or_else(|| json!({})));
let permissioned_as = &flow_job.permissioned_as;
let flow_path = flow_job.script_path();
format!("{permissioned_as}/cache/flow/{flow_path}/{args_hash}")
};
save_in_cache(&client, &flow_job, cached_res_path, &nresult).await;
}
add_completed_job(
db,
&flow_job,
@@ -745,28 +756,26 @@ 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<CacheAndStep> {
) -> error::Result<Option<i32>> {
let step = get_step_of_flow_status(db, flow).await?;
let cache_ttl = if let Step::Step(step) = step {
let ttl = sqlx::query_scalar(&format!(
if let Step::Step(step) = step {
sqlx::query(&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
RETURNING (raw_flow->'modules'->{step}->>'cache_ttl')::int as cache_ttl",
WHERE id = $3 AND workspace_id = $4",
))
.bind(json!(job_in_progress.to_string()))
.bind(json!("InProgress"))
.bind(flow)
.bind(w_id)
.fetch_one(db)
.execute(db)
.await?;
(ttl, Some(step))
Ok(Some(step))
} else {
sqlx::query(&format!(
"UPDATE queue
@@ -779,9 +788,8 @@ pub async fn update_flow_status_in_progress(
.bind(w_id)
.execute(db)
.await?;
(None, None)
};
Ok(cache_ttl)
Ok(None)
}
}
pub enum Step {
@@ -1410,6 +1418,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
concurrent_limit: None,
concurrency_time_window_s: None,
skip_expr: None,
cache_ttl: None,
},
path: Some(format!("{}/forloop", flow_job.script_path())),
},
@@ -1777,7 +1786,7 @@ async fn compute_next_flow_transform(
} else {
let hash = script_hash.clone().unwrap();
let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?;
let (tag, concurrent_limit, concurrency_time_window_s) =
let (tag, concurrent_limit, concurrency_time_window_s, cache_ttl) =
script_hash_to_tag_and_limits(&hash, &mut tx, &flow_job.workspace_id).await?;
(
JobPayload::ScriptHash {
@@ -1785,6 +1794,7 @@ async fn compute_next_flow_transform(
path: script_path.to_owned(),
concurrent_limit,
concurrency_time_window_s,
cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(),
},
tag,
)
@@ -1816,6 +1826,7 @@ async fn compute_next_flow_transform(
lock: lock.clone(),
concurrent_limit: *concurrent_limit,
concurrency_time_window_s: *concurrency_time_window_s,
cache_ttl: module.cache_ttl.map(|x| x as i32),
}),
tag: tag.clone(),
}),
@@ -1928,6 +1939,7 @@ async fn compute_next_flow_transform(
concurrent_limit: None,
concurrency_time_window_s: None,
skip_expr: None,
cache_ttl: None,
},
path: Some(format!("{}/loop-{}", flow_job.script_path(), ns.index)),
},
@@ -2002,6 +2014,7 @@ async fn compute_next_flow_transform(
concurrent_limit: None,
concurrency_time_window_s: None,
skip_expr: None,
cache_ttl: None,
},
path: Some(format!(
"{}/branchone-{}",
@@ -2045,6 +2058,7 @@ async fn compute_next_flow_transform(
concurrent_limit: None,
concurrency_time_window_s: None,
skip_expr: None,
cache_ttl: None,
},
path: Some(format!(
"{}/branchall-{}",
@@ -2119,6 +2133,7 @@ async fn compute_next_flow_transform(
concurrent_limit: None,
concurrency_time_window_s: None,
skip_expr: None,
cache_ttl: None,
},
path: Some(format!(
"{}/branchall-{}",
frontend/src/lib/components/ScriptBuilder.svelte
+34 -3
View File
@@ -10,7 +10,7 @@
import ScriptEditor from './ScriptEditor.svelte'
import ScriptSchema from './ScriptSchema.svelte'
import { dirtyStore } from './common/confirmationModal/dirtyStore'
import { Alert, Badge, Button, Drawer, Kbd } from './common'
import { Alert, Badge, Button, Drawer, Kbd, SecondsInput } from './common'
import { faPlus, faSave } from '@fortawesome/free-solid-svg-icons'
import LanguageIcon from './common/languageIcons/LanguageIcon.svelte'
import type { SupportedLanguage } from '$lib/common'
@@ -27,6 +27,7 @@
import { Icon } from 'svelte-awesome'
import { fade } from 'svelte/transition'
import Popover from './Popover.svelte'
import Toggle from './Toggle.svelte'
export let script: NewScript
export let initialPath: string = ''
@@ -168,7 +169,8 @@
tag: script.tag,
envs: script.envs,
concurrent_limit: script.concurrent_limit,
concurrency_time_window_s: script.concurrency_time_window_s
concurrency_time_window_s: script.concurrency_time_window_s,
cache_ttl: script.cache_ttl
}
})
history.replaceState(history.state, '', `/scripts/edit/${script.path}`)
@@ -210,7 +212,8 @@
draft_only: true,
envs: script.envs,
concurrent_limit: script.concurrent_limit,
concurrency_time_window_s: script.concurrency_time_window_s
concurrency_time_window_s: script.concurrency_time_window_s,
cache_ttl: script.cache_ttl
}
})
}
@@ -443,6 +446,34 @@
{:else}
<Loader2 class="animate-spin" />
{/if}
<h2 class="border-b pb-1 mt-10 mb-4 flex items-center gap-4"
>Cache <Toggle
size="xs"
checked={Boolean(script.cache_ttl)}
on:change={() => {
if (script.cache_ttl && script.cache_ttl != undefined) {
script.cache_ttl = undefined
} else {
script.cache_ttl = 300
}
}}
options={{
right: 'Cache the results for each possible inputs'
}}
/></h2
>
<div class="flex gap-x-4 shrink flex-col">
<div class="text-xs">How long to the keep cache valid</div>
{#if script.cache_ttl}
<SecondsInput bind:seconds={script.cache_ttl} />
{:else}
<SecondsInput disabled />
{/if}
</div>
{#if !isCloudHosted()}
<h2 class="border-b pb-1 mt-10 mb-4">
Custom env variables
frontend/src/lib/components/flows/content/FlowModuleCache.svelte
+1 -1
View File
@@ -24,7 +24,7 @@
if (isCacheEnabled && flowModule.cache_ttl != undefined) {
flowModule.cache_ttl = undefined
} else {
flowModule.cache_ttl = 60 * 60 * 24 * 2
flowModule.cache_ttl = 600
}
}}
options={{
frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
+2 -2
View File
@@ -295,8 +295,8 @@
{#if !$selectedId.includes('failure')}
<Tab value="cache">Cache</Tab>
<Tab value="concurrency">Concurrency</Tab>
<Tab value="early-stop">Early Stop/Break</Tab>
<Tab value="suspend">Suspend/Approval</Tab>
<Tab value="early-stop">Early Stop</Tab>
<Tab value="suspend">Suspend</Tab>
<Tab value="sleep">Sleep</Tab>
<Tab value="mock">Mock</Tab>
<Tab value="same_worker">Shared Directory</Tab>
frontend/src/lib/components/flows/content/FlowSettings.svelte
+31
View File
@@ -61,6 +61,7 @@
<Tab value="settings-early-stop">Early Stop</Tab>
<Tab value="settings-worker-group">Worker Group</Tab>
<Tab value="settings-concurrency">Concurrency</Tab>
<Tab value="settings-cache">Cache</Tab>
<svelte:fragment slot="content">
<TabContent value="settings-metadata" class="p-4 h-full">
@@ -238,6 +239,36 @@
}}
/>
</TabContent>
<TabContent value="settings-cache" class="p-4 flex flex-col">
<h2 class="border-b pb-1 mb-4 flex items-center gap-4"
>Cache <Toggle
size="xs"
checked={Boolean($flowStore.value.cache_ttl)}
on:change={() => {
if ($flowStore.value.cache_ttl && $flowStore.value.cache_ttl != undefined) {
$flowStore.value.cache_ttl = undefined
} else {
$flowStore.value.cache_ttl = 300
}
}}
options={{
right: 'Cache the results for each possible inputs'
}}
/></h2
>
<div class="flex gap-x-4 flex-col gap-2">
<div class="text-xs">How long to keep the cache valid</div>
<div>
{#if $flowStore.value.cache_ttl}
<SecondsInput bind:seconds={$flowStore.value.cache_ttl} />
{:else}
<SecondsInput disabled />
{/if}
</div>
</div>
</TabContent>
<TabContent value="settings-worker-group" class="p-4 flex flex-col">
<Alert type="info" title="Worker Group">
When a worker group is defined at the flow level, any steps inside the flow will run
openflow.openapi.yaml
+2
View File
@@ -52,6 +52,8 @@ components:
type: number
skip_expr:
type: string
cache_ttl:
type: number
required:
- modules