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

fix: improve S3 etag caching (#7301)

Browse Source
This commit is contained in:
Diego Imbert
2025-12-05 19:01:27 +01:00
committed by GitHub
parent cb44d94bc6
commit 7b1a8b0dce
38 changed files with 337 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf.json → backend/.sqlx/query-05b69dcef0f4f649513e186e73089979c49b4b8113ee832ea7539b56a0415f32.json
Generated
+16 -10
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by, path from script where hash = $1 AND workspace_id = $2",
"query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, cache_ignore_s3_path, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by, path from script where hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
@@ -45,6 +45,11 @@
},
{
"ordinal": 8,
"name": "cache_ignore_s3_path",
"type_info": "Bool"
},
{
"ordinal": 9,
"name": "language: ScriptLang",
"type_info": {
"Custom": {
@@ -80,42 +85,42 @@
}
},
{
"ordinal": 9,
"ordinal": 10,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 10,
"ordinal": 11,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 11,
"ordinal": 12,
"name": "delete_after_use",
"type_info": "Bool"
},
{
"ordinal": 12,
"ordinal": 13,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 13,
"ordinal": 14,
"name": "has_preprocessor",
"type_info": "Bool"
},
{
"ordinal": 14,
"ordinal": 15,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 15,
"ordinal": 16,
"name": "created_by",
"type_info": "Varchar"
},
{
"ordinal": 16,
"ordinal": 17,
"name": "path",
"type_info": "Varchar"
}
@@ -135,6 +140,7 @@
true,
true,
true,
true,
false,
true,
true,
@@ -146,5 +152,5 @@
false
]
},
"hash": "f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf"
"hash": "05b69dcef0f4f649513e186e73089979c49b4b8113ee832ea7539b56a0415f32"
}
backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json
Generated
+5 -5
View File
@@ -46,11 +46,11 @@
]
},
"nullable": [
false,
false,
false,
false,
false,
true,
true,
true,
true,
true,
true,
true
]
backend/.sqlx/query-23759cb515e926e272bbc8e5d8a0a9d039b99bc2026e381e99ef41cdaf6ea19f.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO script\n (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, cache_ignore_s3_path, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s)\n\n SELECT workspace_id, $1, path, array_prepend($2::bigint, COALESCE(parent_hashes, '{}'::bigint[])), summary, description, content, created_by, schema, is_template, extra_perms, NULL, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, cache_ignore_s3_path, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s\n\n FROM script WHERE hash = $2 AND workspace_id = $3;\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Int8",
"Text"
]
},
"nullable": []
},
"hash": "23759cb515e926e272bbc8e5d8a0a9d039b99bc2026e381e99ef41cdaf6ea19f"
}
backend/.sqlx/query-0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d.json → backend/.sqlx/query-3d05d9d7e087eb6e1c14c2b8a20598581e6c7493ed99cb9ad1c2ee5d0b212d38.json
Generated
+4 -3
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, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36)",
"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, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s, cache_ignore_s3_path) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37)",
"describe": {
"columns": [],
"parameters": {
@@ -85,10 +85,11 @@
"Bool",
"Jsonb",
"Varchar",
"Int4"
"Int4",
"Bool"
]
},
"nullable": []
},
"hash": "0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d"
"hash": "3d05d9d7e087eb6e1c14c2b8a20598581e6c7493ed99cb9ad1c2ee5d0b212d38"
}
backend/.sqlx/query-53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6.json
Generated
-16
View File
@@ -1,16 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO script\n (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, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s)\n\n SELECT workspace_id, $1, path, array_prepend($2::bigint, COALESCE(parent_hashes, '{}'::bigint[])), summary, description, content, created_by, schema, is_template, extra_perms, NULL, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s\n\n FROM script WHERE hash = $2 AND workspace_id = $3;\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Int8",
"Text"
]
},
"nullable": []
},
"hash": "53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6"
}
backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
Generated
+1 -1
View File
@@ -15,7 +15,7 @@
]
},
"nullable": [
null
true
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json → backend/.sqlx/query-6c97ab28ab47b75fb3ff39ea70fa3627f08b61bbd33aecb9ea816f8f78a04ec5.json
Generated
+14 -8
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n visible_to_owner,\n NULL as permissioned_as_end_user_email\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1",
"query": "SELECT \n v2_job_queue.workspace_id,\n v2_job_queue.id,\n v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n v2_job.parent_job,\n v2_job.created_by,\n v2_job_queue.started_at,\n scheduled_for,\n runnable_path,\n kind as \"kind: JobKind\",\n runnable_id as \"runnable_id: ScriptHash\",\n canceled_reason,\n canceled_by,\n permissioned_as,\n permissioned_as_email,\n flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n script_lang as \"script_lang: ScriptLang\",\n same_worker,\n pre_run_error,\n concurrent_limit,\n concurrency_time_window_s,\n flow_innermost_root_job,\n root_job,\n timeout,\n flow_step_id,\n cache_ttl,\n cache_ignore_s3_path,\n v2_job_queue.priority,\n preprocessed,\n script_entrypoint_override,\n trigger,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n visible_to_owner,\n NULL as permissioned_as_end_user_email\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id WHERE v2_job_queue.id = $1",
"describe": {
"columns": [
{
@@ -193,26 +193,31 @@
},
{
"ordinal": 26,
"name": "cache_ignore_s3_path",
"type_info": "Bool"
},
{
"ordinal": 27,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 27,
"ordinal": 28,
"name": "preprocessed",
"type_info": "Bool"
},
{
"ordinal": 28,
"ordinal": 29,
"name": "script_entrypoint_override",
"type_info": "Varchar"
},
{
"ordinal": 29,
"ordinal": 30,
"name": "trigger",
"type_info": "Varchar"
},
{
"ordinal": 30,
"ordinal": 31,
"name": "trigger_kind: JobTriggerKind",
"type_info": {
"Custom": {
@@ -238,12 +243,12 @@
}
},
{
"ordinal": 31,
"ordinal": 32,
"name": "visible_to_owner",
"type_info": "Bool"
},
{
"ordinal": 32,
"ordinal": 33,
"name": "permissioned_as_end_user_email",
"type_info": "Text"
}
@@ -285,9 +290,10 @@
true,
true,
true,
true,
false,
null
]
},
"hash": "3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5"
"hash": "6c97ab28ab47b75fb3ff39ea70fa3627f08b61bbd33aecb9ea816f8f78a04ec5"
}
backend/.sqlx/query-6cc922a5bbd348c938a9d1431aaa0f24f078ea814b429d44403aca1e5002e750.json → backend/.sqlx/query-7b5ad10af2a9b34fa86429499ea24c0c09c6e7e9ebfa3af90035570133f7c579.json
Generated
+8 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n j.id, j.workspace_id, j.runnable_id AS \"runnable_id: ScriptHash\", q.scheduled_for, q.started_at, j.parent_job, j.flow_innermost_root_job, j.runnable_path, j.kind as \"kind!: JobKind\", j.permissioned_as, \n j.created_by, j.script_lang AS \"script_lang: ScriptLang\", j.permissioned_as_email, j.flow_step_id, j.trigger_kind AS \"trigger_kind: JobTriggerKind\", j.trigger, j.priority, j.concurrent_limit, j.tag, j.cache_ttl\n FROM v2_job j LEFT JOIN v2_job_queue q ON j.id = q.id\n WHERE j.id = $1 AND j.workspace_id = $2",
"query": "SELECT \n j.id, j.workspace_id, j.runnable_id AS \"runnable_id: ScriptHash\", q.scheduled_for, q.started_at, j.parent_job, j.flow_innermost_root_job, j.runnable_path, j.kind as \"kind!: JobKind\", j.permissioned_as, \n j.created_by, j.script_lang AS \"script_lang: ScriptLang\", j.permissioned_as_email, j.flow_step_id, j.trigger_kind AS \"trigger_kind: JobTriggerKind\", j.trigger, j.priority, j.concurrent_limit, j.tag, j.cache_ttl, q.cache_ignore_s3_path\n FROM v2_job j LEFT JOIN v2_job_queue q ON j.id = q.id\n WHERE j.id = $1 AND j.workspace_id = $2",
"describe": {
"columns": [
{
@@ -181,6 +181,11 @@
"ordinal": 19,
"name": "cache_ttl",
"type_info": "Int4"
},
{
"ordinal": 20,
"name": "cache_ignore_s3_path",
"type_info": "Bool"
}
],
"parameters": {
@@ -209,8 +214,9 @@
true,
true,
false,
true,
true
]
},
"hash": "6cc922a5bbd348c938a9d1431aaa0f24f078ea814b429d44403aca1e5002e750"
"hash": "7b5ad10af2a9b34fa86429499ea24c0c09c6e7e9ebfa3af90035570133f7c579"
}
backend/.sqlx/query-27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32.json → backend/.sqlx/query-7f9b7ab9bec6a0f745273d0cd5602ceab46a7ec9fd225f7b9d16a2ddb9bad7b3.json
Generated
+13 -7
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script\n WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false)\n ORDER BY created_at DESC LIMIT 1",
"query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, cache_ignore_s3_path, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script\n WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false)\n ORDER BY created_at DESC LIMIT 1",
"describe": {
"columns": [
{
@@ -45,6 +45,11 @@
},
{
"ordinal": 8,
"name": "cache_ignore_s3_path",
"type_info": "Bool"
},
{
"ordinal": 9,
"name": "language: ScriptLang",
"type_info": {
"Custom": {
@@ -80,27 +85,27 @@
}
},
{
"ordinal": 9,
"ordinal": 10,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 10,
"ordinal": 11,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 11,
"ordinal": 12,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 12,
"ordinal": 13,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 13,
"ordinal": 14,
"name": "created_by",
"type_info": "Varchar"
}
@@ -121,6 +126,7 @@
true,
true,
true,
true,
false,
true,
true,
@@ -129,5 +135,5 @@
false
]
},
"hash": "27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32"
"hash": "7f9b7ab9bec6a0f745273d0cd5602ceab46a7ec9fd225f7b9d16a2ddb9bad7b3"
}
backend/.sqlx/query-d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd.json → backend/.sqlx/query-a84e67035584bbdb02482026b9cc0808086c50f78947d43bb88628a481f41a1d.json
Generated
+11 -5
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, q.workspace_id, j.runnable_id as \"runnable_id: ScriptHash\", scheduled_for, parent_job, flow_innermost_root_job, runnable_path, kind as \"kind: JobKind\", started_at, permissioned_as, created_by, script_lang as \"script_lang: ScriptLang\", \n permissioned_as_email, flow_step_id, trigger_kind as \"trigger_kind: JobTriggerKind\", trigger, q.priority, concurrent_limit, q.tag, cache_ttl, r.ping as last_ping, worker, memory_peak, running\n FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n WHERE j.id = $1",
"query": "SELECT id, q.workspace_id, j.runnable_id as \"runnable_id: ScriptHash\", scheduled_for, parent_job, flow_innermost_root_job, runnable_path, kind as \"kind: JobKind\", started_at, permissioned_as, created_by, script_lang as \"script_lang: ScriptLang\", \n permissioned_as_email, flow_step_id, trigger_kind as \"trigger_kind: JobTriggerKind\", trigger, q.priority, concurrent_limit, q.tag, cache_ttl, cache_ignore_s3_path, r.ping as last_ping, worker, memory_peak, running\n FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n WHERE j.id = $1",
"describe": {
"columns": [
{
@@ -184,21 +184,26 @@
},
{
"ordinal": 20,
"name": "cache_ignore_s3_path",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "last_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 21,
"ordinal": 22,
"name": "worker",
"type_info": "Varchar"
},
{
"ordinal": 22,
"ordinal": 23,
"name": "memory_peak",
"type_info": "Int4"
},
{
"ordinal": 23,
"ordinal": 24,
"name": "running",
"type_info": "Bool"
}
@@ -232,8 +237,9 @@
true,
true,
true,
true,
false
]
},
"hash": "d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd"
"hash": "a84e67035584bbdb02482026b9cc0808086c50f78947d43bb88628a481f41a1d"
}
backend/.sqlx/query-bf2163c542fb8c4e173167a8f333ef762fecf782424c5b61b89f32918b8d6971.json → backend/.sqlx/query-b179a3f876ca659bed892d464bf51a733cc86a3204fcd9edccda63fddc97dced.json
Generated
+4 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "WITH inserted_job AS (\n INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)\n ),\n inserted_runtime AS (\n INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)\n ),\n inserted_job_perms AS (\n INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, end_user_email) \n values ($1, $32, $33, $34, $35, $36, $37, $2, $41) \n ON CONFLICT (job_id) DO UPDATE SET email = EXCLUDED.email, username = EXCLUDED.username, is_admin = EXCLUDED.is_admin, is_operator = EXCLUDED.is_operator, folders = EXCLUDED.folders, groups = EXCLUDED.groups, workspace_id = EXCLUDED.workspace_id, end_user_email = EXCLUDED.end_user_email\n )\n INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31)",
"query": "WITH inserted_job AS (\n INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,\n created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,\n script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,\n flow_innermost_root_job, root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,\n cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,\n $19, $20, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind,\n ($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)\n ),\n inserted_runtime AS (\n INSERT INTO v2_job_runtime (id, ping) VALUES ($1, null)\n ),\n inserted_job_perms AS (\n INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, end_user_email) \n values ($1, $32, $33, $34, $35, $36, $37, $2, $41) \n ON CONFLICT (job_id) DO UPDATE SET email = EXCLUDED.email, username = EXCLUDED.username, is_admin = EXCLUDED.is_admin, is_operator = EXCLUDED.is_operator, folders = EXCLUDED.folders, groups = EXCLUDED.groups, workspace_id = EXCLUDED.workspace_id, end_user_email = EXCLUDED.end_user_email\n )\n INSERT INTO v2_job_queue\n (workspace_id, id, running, scheduled_for, started_at, tag, priority, cache_ignore_s3_path)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42)",
"describe": {
"columns": [],
"parameters": {
@@ -124,10 +124,11 @@
}
},
"Bool",
"Varchar"
"Varchar",
"Bool"
]
},
"nullable": []
},
"hash": "bf2163c542fb8c4e173167a8f333ef762fecf782424c5b61b89f32918b8d6971"
"hash": "b179a3f876ca659bed892d464bf51a733cc86a3204fcd9edccda63fddc97dced"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
937820a9e572c10c0ce050e1bd5eddb3180fe921
f03dfeccbe7610c1e3029fb67b47faeba8fac918
backend/migrations/20251204210743_cache_ignore_s3_path.down.sql
+5
View File
@@ -0,0 +1,5 @@
ALTER TABLE v2_job_queue
DROP COLUMN cache_ignore_s3_path;
ALTER TABLE script
DROP COLUMN cache_ignore_s3_path;
backend/migrations/20251204210743_cache_ignore_s3_path.up.sql
+5
View File
@@ -0,0 +1,5 @@
ALTER TABLE script
ADD COLUMN cache_ignore_s3_path BOOLEAN DEFAULT NULL;
ALTER TABLE v2_job_queue
ADD COLUMN cache_ignore_s3_path BOOLEAN DEFAULT NULL;
backend/tests/common/mod.rs
+2
View File
@@ -685,6 +685,7 @@ pub async fn run_deployed_relative_imports(
path: "f/system/test_import".to_string(),
hash: ScriptHash(script.hash),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
language,
priority: None,
@@ -737,6 +738,7 @@ pub async fn run_preview_relative_imports(
language,
lock: None,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
backend/tests/job_payload.rs
+4
View File
@@ -55,6 +55,7 @@ mod job_payload {
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
language: ScriptLang::Deno,
priority: None,
@@ -84,6 +85,7 @@ mod job_payload {
hash: ScriptHash(123413),
path: "f/system/hello_with_preprocessor".to_string(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
language: ScriptLang::Deno,
priority: None,
@@ -163,6 +165,7 @@ mod job_payload {
language: ScriptLang::Deno,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
path: "f/system/hello/test-0".into(),
})
@@ -181,6 +184,7 @@ mod job_payload {
language: ScriptLang::Deno,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
path: "f/system/hello/test-0".into(),
})
backend/tests/python_jobs.rs
+5
View File
@@ -192,6 +192,7 @@ def main():
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
});
@@ -240,6 +241,7 @@ def main():
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
});
@@ -273,6 +275,7 @@ def main():
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
});
@@ -311,6 +314,7 @@ def main():
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
});
@@ -347,6 +351,7 @@ def main():
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
});
backend/tests/worker.rs
+17
View File
@@ -201,6 +201,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -246,6 +247,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -265,6 +267,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -378,6 +381,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -432,6 +436,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -472,6 +477,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -491,6 +497,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -537,6 +544,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -855,6 +863,7 @@ func main(derp string) (string, error) {
lock: None,
language: ScriptLang::Go,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
@@ -890,6 +899,7 @@ fn main(world: String) -> Result<String, String> {
path: None,
lock: None,
language: ScriptLang::Rust,
cache_ignore_s3_path: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
cache_ttl: None,
@@ -966,6 +976,7 @@ echo "hello $msg"
lock: None,
language: ScriptLang::Bash,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
@@ -998,6 +1009,7 @@ def main [ msg: string ] {
lock: None,
language: ScriptLang::Nu,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
@@ -1050,6 +1062,7 @@ def main [
lock: None,
language: ScriptLang::Nu,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
@@ -1111,6 +1124,7 @@ public class Main {
lock: None,
language: ScriptLang::Java,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
@@ -1145,6 +1159,7 @@ export async function main(a: Date) {
lock: None,
language: ScriptLang::Bun,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
@@ -1179,6 +1194,7 @@ export async function main(a: Date) {
lock: None,
language: ScriptLang::Deno,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
@@ -1214,6 +1230,7 @@ def main(a: datetime, b: bytes):
lock: None,
language: ScriptLang::Python3,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
backend/windmill-api/openapi.yaml
+2
View File
@@ -16259,6 +16259,8 @@ components:
type: integer
cache_ttl:
type: number
cache_ignore_s3_path:
type: boolean
dedicated_worker:
type: boolean
ws_error_handler_muted:
backend/windmill-api/src/flows.rs
+7 -4
View File
@@ -1210,10 +1210,8 @@ async fn get_deployment_status(
let status = not_found_if_none(status_o, "DeploymentStatus", path)?;
let deployment_status = DeploymentStatus {
lock_error_logs: status.lock_error_logs,
job_id: status.job_id,
};
let deployment_status =
DeploymentStatus { lock_error_logs: status.lock_error_logs, job_id: status.job_id };
tx.commit().await?;
Ok(Json(deployment_status))
@@ -1634,6 +1632,7 @@ mod tests {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -1666,6 +1665,7 @@ mod tests {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -1698,6 +1698,7 @@ mod tests {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -1729,6 +1730,7 @@ mod tests {
retry: None,
sleep: None,
cache_ttl: None,
cache_ignore_s3_path: None,
mock: None,
timeout: None,
priority: None,
@@ -1742,6 +1744,7 @@ mod tests {
same_worker: false,
skip_expr: None,
cache_ttl: None,
cache_ignore_s3_path: None,
priority: None,
early_return: None,
chat_input_enabled: None,
backend/windmill-api/src/jobs.rs
+13 -1
View File
@@ -887,7 +887,7 @@ macro_rules! get_job_query {
get_job_query!(
@impl "v2_job_queue", ($($opts)*),
"scheduled_for, running, ping as last_ping, suspend, suspend_until, same_worker, pre_run_error, visible_to_owner, \
flow_innermost_root_job AS root_job, flow_leaf_jobs AS leaf_jobs, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl,\
flow_innermost_root_job AS root_job, flow_leaf_jobs AS leaf_jobs, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, cache_ignore_s3_path, \
script_entrypoint_override",
"LEFT JOIN v2_job_runtime ON v2_job_runtime.id = v2_job_queue.id LEFT JOIN v2_job_status ON v2_job_status.id = v2_job_queue.id",
)
@@ -1772,6 +1772,7 @@ pub struct RunJobQuery {
pub tag: Option<String>,
pub timeout: Option<i32>,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub skip_preprocessor: Option<bool>,
pub poll_delay_ms: Option<u64>,
pub memory_id: Option<Uuid>,
@@ -3537,6 +3538,7 @@ impl<'a> From<UnifiedJob> for Job {
timeout: None,
flow_step_id: None,
cache_ttl: None,
cache_ignore_s3_path: None,
priority: uj.priority,
preprocessed: uj.preprocessed,
},
@@ -4594,6 +4596,7 @@ pub async fn run_workflow_as_code(
concurrency_time_window_s: job.concurrency_time_window_s,
},
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
dedicated_worker: None,
// TODO(debouncing): enable for this mode
debouncing_settings: DebouncingSettings::default(),
@@ -5415,6 +5418,7 @@ pub async fn run_wait_result_script_by_hash(
debounce_key,
debounce_delay_s,
mut cache_ttl,
mut cache_ignore_s3_path,
language,
dedicated_worker,
priority,
@@ -5427,6 +5431,7 @@ pub async fn run_wait_result_script_by_hash(
} = get_script_info_for_hash(Some(userdb_authed), &db, &w_id, hash).await?;
if let Some(run_query_cache_ttl) = run_query.cache_ttl {
cache_ttl = Some(run_query_cache_ttl);
cache_ignore_s3_path = run_query.cache_ignore_s3_path;
}
check_scopes(&authed, || format!("jobs:run:scripts:{path}"))?;
@@ -5469,6 +5474,7 @@ pub async fn run_wait_result_script_by_hash(
..Default::default() // TODO
},
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
@@ -5942,6 +5948,7 @@ async fn run_preview_script(
concurrency_settings: ConcurrencySettingsWithCustom::default(), // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here
debouncing_settings: DebouncingSettings::default(), // TODO(pyra): same as for concurrency limits.
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: preview.dedicated_worker,
}),
},
@@ -6059,6 +6066,7 @@ async fn run_bundle_preview_script(
language: preview.language.unwrap_or(ScriptLang::Deno),
lock: preview.lock,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: preview.dedicated_worker,
concurrency_settings: ConcurrencySettingsWithCustom::default(),
debouncing_settings: DebouncingSettings::default(),
@@ -6847,6 +6855,7 @@ async fn run_dynamic_select(
language: dynamic_input.x_windmill_dyn_select_lang,
lock: None,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: ConcurrencySettings::default().into(),
debouncing_settings: DebouncingSettings::default(),
@@ -6929,6 +6938,7 @@ pub async fn run_job_by_hash_inner(
debounce_delay_s,
debounce_key,
mut cache_ttl,
mut cache_ignore_s3_path,
language,
dedicated_worker,
priority,
@@ -6943,6 +6953,7 @@ pub async fn run_job_by_hash_inner(
check_scopes(&authed, || format!("jobs:run:scripts:{path}"))?;
if let Some(run_query_cache_ttl) = run_query.cache_ttl {
cache_ttl = Some(run_query_cache_ttl);
cache_ignore_s3_path = run_query.cache_ignore_s3_path;
}
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tag = run_query.tag.clone().or(tag);
@@ -6984,6 +6995,7 @@ pub async fn run_job_by_hash_inner(
..Default::default()
},
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
backend/windmill-api/src/scripts.rs
+6 -3
View File
@@ -94,6 +94,8 @@ pub struct ScriptWDraft {
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ignore_s3_path: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub dedicated_worker: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub ws_error_handler_muted: Option<bool>,
@@ -801,8 +803,8 @@ async fn create_script_internal<'c>(
content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \
dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \
delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36)",
delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s, cache_ignore_s3_path) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::text::json, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37)",
&w_id,
&hash.0,
ns.path,
@@ -843,6 +845,7 @@ async fn create_script_internal<'c>(
ns.assets.as_ref().and_then(|a| serde_json::to_value(a).ok()),
ns.debounce_key,
ns.debounce_delay_s,
ns.cache_ignore_s3_path,
)
.execute(&mut *tx)
.await?;
@@ -1238,7 +1241,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, cache_ttl, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, has_preprocessor, on_behalf_of_email, assets, debounce_key, debounce_delay_s 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, cache_ignore_s3_path, ws_error_handler_muted, draft.value as draft, dedicated_worker, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, has_preprocessor, on_behalf_of_email, assets, debounce_key, debounce_delay_s 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
ORDER BY script.created_at DESC LIMIT 1",
backend/windmill-api/src/triggers/trigger_helpers.rs
+2
View File
@@ -847,6 +847,7 @@ async fn trigger_script_with_retry_and_error_handler(
concurrency_settings,
debouncing_settings,
cache_ttl,
cache_ignore_s3_path,
priority,
apply_preprocessor,
..
@@ -860,6 +861,7 @@ async fn trigger_script_with_retry_and_error_handler(
error_handler_args,
skip_handler: None,
cache_ttl,
cache_ignore_s3_path,
priority,
tag_override: tag.clone(),
apply_preprocessor,
backend/windmill-common/src/flows.rs
+5
View File
@@ -181,6 +181,8 @@ pub struct FlowValue {
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ignore_s3_path: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub early_return: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
// Priority at the flow level
@@ -442,6 +444,8 @@ pub struct FlowModule {
pub sleep: Option<InputTransform>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ignore_s3_path: Option<bool>,
#[serde(
default,
deserialize_with = "raw_value_to_input_transform::<_, i32>",
@@ -1144,6 +1148,7 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
sleep: None,
suspend: None,
cache_ttl: None,
cache_ignore_s3_path: None,
timeout: None,
priority: None,
delete_after_use: None,
backend/windmill-common/src/jobs.rs
+9
View File
@@ -186,6 +186,8 @@ pub struct QueuedJob {
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ignore_s3_path: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub priority: Option<i16>,
#[serde(skip_serializing_if = "Option::is_none")]
pub preprocessed: Option<bool>,
@@ -259,6 +261,7 @@ impl Default for QueuedJob {
timeout: None,
flow_step_id: None,
cache_ttl: None,
cache_ignore_s3_path: None,
priority: None,
preprocessed: None,
}
@@ -346,6 +349,7 @@ pub enum JobPayload {
hash: ScriptHash,
path: String,
cache_ttl: Option<i32>,
cache_ignore_s3_path: Option<bool>,
dedicated_worker: Option<bool>,
language: ScriptLang,
priority: Option<i16>,
@@ -366,6 +370,7 @@ pub enum JobPayload {
path: String,
language: ScriptLang,
cache_ttl: Option<i32>,
cache_ignore_s3_path: Option<bool>,
dedicated_worker: Option<bool>,
concurrency_settings: ConcurrencySettings,
},
@@ -448,6 +453,7 @@ pub enum JobPayload {
error_handler_args: Option<HashMap<String, Box<RawValue>>>,
skip_handler: Option<SkipHandler>,
cache_ttl: Option<i32>,
cache_ignore_s3_path: Option<bool>,
priority: Option<i16>,
tag_override: Option<String>,
trigger_path: Option<String>,
@@ -584,6 +590,7 @@ pub struct RawCode {
pub language: ScriptLang,
pub lock: Option<String>,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub dedicated_worker: Option<bool>,
#[serde(flatten)]
pub concurrency_settings: ConcurrencySettingsWithCustom,
@@ -691,6 +698,7 @@ pub async fn script_path_to_payload<'e>(
debounce_key,
debounce_delay_s,
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
@@ -716,6 +724,7 @@ pub async fn script_path_to_payload<'e>(
hash: ScriptHash(hash),
path: script_path.to_owned(),
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
backend/windmill-common/src/lib.rs
+5 -2
View File
@@ -659,6 +659,7 @@ pub struct ScriptHashInfo {
pub debounce_key: Option<String>,
pub debounce_delay_s: Option<i32>,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub language: ScriptLang,
pub dedicated_worker: Option<bool>,
pub priority: Option<i16>,
@@ -804,7 +805,7 @@ async fn get_script_info_for_hash_inner<'e, E: sqlx::PgExecutor<'e>>(
) -> error::Result<Option<ScriptHashInfo>> {
let r = sqlx::query_as!(
ScriptHashInfo,
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by, path from script where hash = $1 AND workspace_id = $2",
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, cache_ignore_s3_path, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by, path from script where hash = $1 AND workspace_id = $2",
hash,
w_id
)
@@ -1022,6 +1023,7 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>(
Option<String>,
Option<i32>,
Option<i32>,
Option<bool>,
ScriptLang,
Option<bool>,
Option<i16>,
@@ -1030,7 +1032,7 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>(
String,
)> {
let r_o = sqlx::query!(
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, cache_ignore_s3_path, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script
WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false)
ORDER BY created_at DESC LIMIT 1",
script_path,
@@ -1051,6 +1053,7 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>(
script.debounce_key,
script.debounce_delay_s,
script.cache_ttl,
script.cache_ignore_s3_path,
script.language,
script.dedicated_worker,
script.priority,
backend/windmill-common/src/scripts.rs
+7 -2
View File
@@ -393,6 +393,8 @@ pub struct Script {
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ignore_s3_path: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delete_after_use: Option<bool>,
@@ -507,6 +509,7 @@ pub struct NewScript {
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub dedicated_worker: Option<bool>,
pub ws_error_handler_muted: Option<bool>,
pub priority: Option<i16>,
@@ -857,6 +860,7 @@ pub async fn fetch_script_for_update<'a>(
ws_error_handler_muted,
priority,
cache_ttl,
cache_ignore_s3_path,
timeout,
delete_after_use,
restart_unless_cancelled,
@@ -914,6 +918,7 @@ pub async fn clone_script<'c>(
concurrent_limit: s.concurrent_limit,
concurrency_time_window_s: s.concurrency_time_window_s,
cache_ttl: s.cache_ttl,
cache_ignore_s3_path: s.cache_ignore_s3_path,
dedicated_worker: s.dedicated_worker,
ws_error_handler_muted: s.ws_error_handler_muted,
priority: s.priority,
@@ -945,14 +950,14 @@ pub async fn clone_script<'c>(
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, \
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, cache_ignore_s3_path, \
dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \
delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, \
codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s)
SELECT workspace_id, $1, path, array_prepend($2::bigint, COALESCE(parent_hashes, '{}'::bigint[])), summary, description, \
content, created_by, schema, is_template, extra_perms, NULL, language, kind, tag, \
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \
draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, cache_ignore_s3_path, \
dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \
delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, \
codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s
backend/windmill-common/src/worker.rs
+2 -2
View File
@@ -411,7 +411,7 @@ fn format_pull_query(peek: String) -> String {
WHERE id = (SELECT id FROM peek)
RETURNING
started_at, scheduled_for,
canceled_by, canceled_reason, worker
canceled_by, canceled_reason, worker, cache_ignore_s3_path
), r AS NOT MATERIALIZED (
UPDATE v2_job_runtime SET
ping = now()
@@ -437,7 +437,7 @@ fn format_pull_query(peek: String) -> String {
flow_status, j.script_lang,
j.same_worker, j.pre_run_error, j.visible_to_owner,
j.tag, j.concurrent_limit, j.concurrency_time_window_s, j.flow_innermost_root_job, j.root_job,
j.timeout, j.flow_step_id, j.cache_ttl, j.priority, j.raw_code, j.raw_lock, j.raw_flow,
j.timeout, j.flow_step_id, j.cache_ttl, q.cache_ignore_s3_path, j.priority, j.raw_code, j.raw_lock, j.raw_flow,
j.script_entrypoint_override, j.preprocessed, pj.runnable_path as parent_runnable_path,
COALESCE(p.email, j.permissioned_as_email) as permissioned_as_email, p.username as permissioned_as_username, p.is_admin as permissioned_as_is_admin,
p.is_operator as permissioned_as_is_operator, p.groups as permissioned_as_groups, p.folders as permissioned_as_folders, p.end_user_email as permissioned_as_end_user_email
backend/windmill-queue/src/jobs.rs
+32 -4
View File
@@ -431,6 +431,7 @@ pub async fn push_init_job<'c>(
language: ScriptLang::Bash,
lock: None,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: ConcurrencySettingsWithCustom::default(),
debouncing_settings: DebouncingSettings::default(),
@@ -487,6 +488,7 @@ pub async fn push_periodic_bash_job<'c>(
language: ScriptLang::Bash,
lock: None,
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
concurrency_settings: ConcurrencySettingsWithCustom::default(),
debouncing_settings: DebouncingSettings::default(),
@@ -1322,6 +1324,7 @@ async fn restart_job_if_perpetual_inner(
hash,
path: queued_job.runnable_path.clone().unwrap_or_default(),
cache_ttl: queued_job.cache_ttl,
cache_ignore_s3_path: queued_job.cache_ignore_s3_path,
dedicated_worker: None,
language: queued_job
.script_lang
@@ -1976,6 +1979,7 @@ pub struct MiniPulledJob {
pub timeout: Option<i32>,
pub flow_step_id: Option<String>,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub priority: Option<i16>,
pub preprocessed: Option<bool>,
pub script_entrypoint_override: Option<String>,
@@ -2009,6 +2013,7 @@ pub struct MiniCompletedJob {
pub concurrent_limit: Option<i32>,
pub tag: String,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
}
impl From<QueuedJobV2> for MiniCompletedJob {
@@ -2034,6 +2039,7 @@ impl From<QueuedJobV2> for MiniCompletedJob {
concurrent_limit: job.concurrent_limit,
tag: job.tag,
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
}
}
}
@@ -2062,6 +2068,7 @@ impl From<MiniPulledJob> for MiniCompletedJob {
concurrent_limit: job.concurrent_limit,
tag: job.tag,
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
}
}
}
@@ -2089,6 +2096,7 @@ impl From<Arc<MiniPulledJob>> for MiniCompletedJob {
concurrent_limit: job.concurrent_limit,
tag: job.tag.clone(),
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
}
}
}
@@ -2188,6 +2196,7 @@ impl MiniPulledJob {
timeout: job.timeout.clone(),
flow_step_id: job.flow_step_id.clone(),
cache_ttl: job.cache_ttl.clone(),
cache_ignore_s3_path: job.cache_ignore_s3_path.clone(),
priority: job.priority.clone(),
preprocessed: job.preprocessed.clone(),
script_entrypoint_override: job.script_entrypoint_override.clone(),
@@ -2395,6 +2404,7 @@ pub async fn get_mini_pulled_job<'c>(
timeout,
flow_step_id,
cache_ttl,
cache_ignore_s3_path,
v2_job_queue.priority,
preprocessed,
script_entrypoint_override,
@@ -2432,6 +2442,7 @@ pub struct QueuedJobV2 {
pub concurrent_limit: Option<i32>,
pub tag: String,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub last_ping: Option<chrono::DateTime<chrono::Utc>>,
pub worker: Option<String>,
pub memory_peak: Option<i32>,
@@ -2451,7 +2462,7 @@ pub async fn get_queued_job_v2<'c>(
let job = sqlx::query_as!(
QueuedJobV2,
"SELECT id, q.workspace_id, j.runnable_id as \"runnable_id: ScriptHash\", scheduled_for, parent_job, flow_innermost_root_job, runnable_path, kind as \"kind: JobKind\", started_at, permissioned_as, created_by, script_lang as \"script_lang: ScriptLang\",
permissioned_as_email, flow_step_id, trigger_kind as \"trigger_kind: JobTriggerKind\", trigger, q.priority, concurrent_limit, q.tag, cache_ttl, r.ping as last_ping, worker, memory_peak, running
permissioned_as_email, flow_step_id, trigger_kind as \"trigger_kind: JobTriggerKind\", trigger, q.priority, concurrent_limit, q.tag, cache_ttl, cache_ignore_s3_path, r.ping as last_ping, worker, memory_peak, running
FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)
WHERE j.id = $1",
job_id,
@@ -3608,7 +3619,7 @@ pub fn get_mini_completed_job<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres>
MiniCompletedJob,
"SELECT
j.id, j.workspace_id, j.runnable_id AS \"runnable_id: ScriptHash\", q.scheduled_for, q.started_at, j.parent_job, j.flow_innermost_root_job, j.runnable_path, j.kind as \"kind!: JobKind\", j.permissioned_as,
j.created_by, j.script_lang AS \"script_lang: ScriptLang\", j.permissioned_as_email, j.flow_step_id, j.trigger_kind AS \"trigger_kind: JobTriggerKind\", j.trigger, j.priority, j.concurrent_limit, j.tag, j.cache_ttl
j.created_by, j.script_lang AS \"script_lang: ScriptLang\", j.permissioned_as_email, j.flow_step_id, j.trigger_kind AS \"trigger_kind: JobTriggerKind\", j.trigger, j.priority, j.concurrent_limit, j.tag, j.cache_ttl, q.cache_ignore_s3_path
FROM v2_job j LEFT JOIN v2_job_queue q ON j.id = q.id
WHERE j.id = $1 AND j.workspace_id = $2",
id,
@@ -4043,6 +4054,7 @@ pub async fn push<'c, 'd>(
flow_status: Option<FlowStatus>,
language: Option<ScriptLang>,
cache_ttl: Option<i32>,
cache_ignore_s3_path: Option<bool>,
dedicated_worker: Option<bool>,
_low_level_priority: Option<i16>,
concurrency_settings: ConcurrencySettings,
@@ -4059,6 +4071,7 @@ pub async fn push<'c, 'd>(
flow_status,
language,
cache_ttl,
cache_ignore_s3_path,
dedicated_worker,
_low_level_priority,
concurrency_settings:
@@ -4073,6 +4086,7 @@ pub async fn push<'c, 'd>(
hash,
path,
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
@@ -4092,6 +4106,7 @@ pub async fn push<'c, 'd>(
concurrency_settings,
debouncing_settings,
cache_ttl,
cache_ignore_s3_path,
dedicated_worker,
_low_level_priority: priority,
..Default::default()
@@ -4101,6 +4116,7 @@ pub async fn push<'c, 'd>(
id, // flow_node(id).
language,
cache_ttl,
cache_ignore_s3_path,
dedicated_worker,
path,
concurrency_settings,
@@ -4111,6 +4127,7 @@ pub async fn push<'c, 'd>(
language: Some(language),
concurrency_settings,
cache_ttl,
cache_ignore_s3_path,
dedicated_worker,
..Default::default()
},
@@ -4178,6 +4195,7 @@ pub async fn push<'c, 'd>(
language,
lock,
cache_ttl,
cache_ignore_s3_path,
dedicated_worker,
concurrency_settings,
debouncing_settings,
@@ -4190,6 +4208,7 @@ pub async fn push<'c, 'd>(
concurrency_settings: concurrency_settings.into(),
debouncing_settings,
cache_ttl,
cache_ignore_s3_path,
dedicated_worker,
..Default::default()
},
@@ -4309,6 +4328,7 @@ pub async fn push<'c, 'd>(
job_kind: JobKind::FlowPreview,
flow_status: Some(flow_status),
cache_ttl,
cache_ignore_s3_path: value.cache_ignore_s3_path,
_low_level_priority: priority,
concurrency_settings: value.concurrency_settings.clone(),
debouncing_settings: value.debouncing_settings.clone(),
@@ -4326,6 +4346,7 @@ pub async fn push<'c, 'd>(
skip_handler,
args,
cache_ttl,
cache_ignore_s3_path,
priority,
tag_override,
trigger_path,
@@ -4469,6 +4490,7 @@ pub async fn push<'c, 'd>(
debouncing_settings: debouncing_settings.clone(),
priority,
cache_ttl: cache_ttl.map(|val| val as u32),
cache_ignore_s3_path: cache_ignore_s3_path,
same_worker: false,
early_return: None,
skip_expr: None,
@@ -4484,6 +4506,7 @@ pub async fn push<'c, 'd>(
raw_flow: Some(flow_value),
flow_status: Some(flow_status),
cache_ttl,
cache_ignore_s3_path,
_low_level_priority: priority,
concurrency_settings,
debouncing_settings,
@@ -4511,6 +4534,7 @@ pub async fn push<'c, 'd>(
let mut value = data.value().clone();
let priority = value.priority;
let cache_ttl = value.cache_ttl.map(|x| x as i32);
let cache_ignore_s3_path = value.cache_ignore_s3_path;
let mut concurrency_settings = value.concurrency_settings.clone();
let mut debouncing_settings = value.debouncing_settings.clone();
@@ -4550,6 +4574,7 @@ pub async fn push<'c, 'd>(
raw_flow: value_o,
flow_status: Some(status),
cache_ttl,
cache_ignore_s3_path,
dedicated_worker,
_low_level_priority: priority,
concurrency_settings,
@@ -4621,6 +4646,7 @@ pub async fn push<'c, 'd>(
raw_flow: value_o,
flow_status: Some(restarted_flow_status),
cache_ttl,
cache_ignore_s3_path: value.cache_ignore_s3_path,
_low_level_priority: priority,
concurrency_settings,
debouncing_settings,
@@ -5114,8 +5140,8 @@ pub async fn push<'c, 'd>(
ON CONFLICT (job_id) DO UPDATE SET email = EXCLUDED.email, username = EXCLUDED.username, is_admin = EXCLUDED.is_admin, is_operator = EXCLUDED.is_operator, folders = EXCLUDED.folders, groups = EXCLUDED.groups, workspace_id = EXCLUDED.workspace_id, end_user_email = EXCLUDED.end_user_email
)
INSERT INTO v2_job_queue
(workspace_id, id, running, scheduled_for, started_at, tag, priority)
VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31)",
(workspace_id, id, running, scheduled_for, started_at, tag, priority, cache_ignore_s3_path)
VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42)",
job_id,
workspace_id,
raw_code,
@@ -5161,6 +5187,7 @@ pub async fn push<'c, 'd>(
trigger_kind as Option<JobTriggerKind>,
running,
end_user_email,
cache_ignore_s3_path,
)
.execute(&mut *tx)
.warn_after_seconds(1)
@@ -5667,6 +5694,7 @@ pub async fn get_same_worker_job(
v2_job.timeout,
v2_job.flow_step_id,
v2_job.cache_ttl,
v2_job_queue.cache_ignore_s3_path,
v2_job_queue.priority,
v2_job.preprocessed,
v2_job.script_entrypoint_override,
backend/windmill-queue/src/schedule.rs
+5
View File
@@ -92,6 +92,7 @@ async fn get_schedule_metadata<'c>(
_debounce_key,
_debounce_delay_s,
_cache_ttl,
_cache_ignore_s3_path,
_language,
_dedicated_worker,
_priority,
@@ -261,6 +262,7 @@ pub async fn push_scheduled_job<'c>(
stop_message,
}),
cache_ttl: None,
cache_ignore_s3_path: None,
priority: None,
tag_override: schedule.tag.clone(),
trigger_path: None,
@@ -321,6 +323,7 @@ pub async fn push_scheduled_job<'c>(
custom_debounce_key,
debounce_delay_s,
cache_ttl,
cache_ignore_s3_path,
language,
dedicated_worker,
priority,
@@ -360,6 +363,7 @@ pub async fn push_scheduled_job<'c>(
skip_handler: None,
args: static_args,
cache_ttl,
cache_ignore_s3_path,
priority,
tag_override: schedule.tag.clone(),
trigger_path: None,
@@ -382,6 +386,7 @@ pub async fn push_scheduled_job<'c>(
hash,
path: schedule.script_path.clone(),
cache_ttl,
cache_ignore_s3_path,
dedicated_worker,
language,
priority,
backend/windmill-worker/src/ai/tools.rs
+1
View File
@@ -392,6 +392,7 @@ async fn execute_windmill_tool(
language,
concurrency_settings: concurrency_settings.into(),
cache_ttl: tool_module.cache_ttl.map(|x| x as i32),
cache_ignore_s3_path: tool_module.cache_ignore_s3_path.clone(),
dedicated_worker: None,
path,
},
backend/windmill-worker/src/common.rs
+48 -93
View File
@@ -746,24 +746,48 @@ pub async fn resolve_job_timeout(
}
async fn hash_args(
_db: &DB,
_client: &AuthedClient,
_workspace_id: &str,
#[allow(unused)] db: &DB,
#[allow(unused)] client: &AuthedClient,
#[allow(unused)] workspace_id: &str,
v: &Option<Json<HashMap<String, Box<RawValue>>>>,
hasher: &mut sha2::Sha256,
#[allow(unused)] job_id: &Uuid,
#[allow(unused)] ignore_s3_path: bool,
) {
if let Some(Json(hm)) = v {
for k in hm.keys().sorted() {
hasher.update(k.as_bytes());
let arg_value = hm.get(k).unwrap();
#[cfg(feature = "parquet")]
let (_, arg_additions) =
arg_value_hash_additions(_db, _client, _workspace_id, hm.get(k).unwrap()).await;
hasher.update(arg_value.get().as_bytes());
let etag = match serde_json::from_str::<S3Object>(arg_value.get()).ok() {
Some(s3_object) => {
let s3_resource = get_workspace_s3_resource_path(
db,
client,
workspace_id,
s3_object.storage.as_ref(),
job_id,
)
.await
.ok()
.flatten();
match s3_resource {
Some(s3_resource) => get_etag_or_empty(&s3_resource, s3_object).await,
None => None,
}
}
None => None,
};
#[cfg(feature = "parquet")]
for (_, arg_addition) in arg_additions {
hasher.update(arg_addition.as_bytes());
if let Some(etag) = etag {
hasher.update(etag.as_bytes());
if ignore_s3_path {
continue;
}
}
hasher.update(arg_value.get().as_bytes());
}
}
}
@@ -788,7 +812,16 @@ pub async fn cached_result_path(
_ => {}
}
}
hash_args(db, client, &job.workspace_id, &job.args, &mut hasher).await;
hash_args(
db,
client,
&job.workspace_id,
&job.args,
&mut hasher,
&job.id,
job.cache_ignore_s3_path.unwrap_or(false),
)
.await;
format!("g/results/{:064x}", hasher.finalize())
}
@@ -798,6 +831,7 @@ async fn get_workspace_s3_resource_path(
client: &AuthedClient,
workspace_id: &str,
storage: Option<&String>,
job_id: &Uuid,
) -> windmill_common::error::Result<Option<ObjectStoreResource>> {
use windmill_common::{
job_s3_helpers_oss::get_s3_resource_internal, s3_helpers::StorageResourceType,
@@ -860,7 +894,10 @@ async fn get_workspace_s3_resource_path(
};
let s3_resource_value_raw = client
.get_resource_value::<serde_json::Value>(path.as_str())
.get_resource_value_interpolated::<serde_json::Value>(
path.as_str(),
Some(job_id.to_string()),
)
.await?;
get_s3_resource_internal(
rt,
@@ -872,41 +909,10 @@ async fn get_workspace_s3_resource_path(
.map(Some)
}
#[cfg(feature = "parquet")]
async fn arg_value_hash_additions(
db: &DB,
client: &AuthedClient,
workspace_id: &str,
raw_value: &Box<RawValue>,
) -> (Option<String>, HashMap<String, String>) {
let mut result: HashMap<String, String> = HashMap::new();
let parsed_value = serde_json::from_str::<S3Object>(raw_value.get());
let mut storage = None;
if let Ok(s3_object) = parsed_value {
let s3_resource_opt =
get_workspace_s3_resource_path(db, client, workspace_id, s3_object.storage.as_ref())
.await;
storage = s3_object.storage.clone();
if let Some(mut s3_resource) = s3_resource_opt.ok().flatten() {
let etag = get_etag_or_empty(&mut s3_resource, s3_object.clone()).await;
tracing::warn!("Enriching s3 arg value with etag: {:?}", etag);
result.insert(s3_object.s3.clone(), etag.unwrap_or_default()); // TODO: maybe inject a random value to invalidate the cache?
}
}
return (storage, result);
}
#[derive(Deserialize, Serialize)]
struct CachedResource {
expire: i64,
#[serde(skip_serializing_if = "Option::is_none")]
s3_etags: Option<HashMap<String, String>>,
value: Arc<Box<RawValue>>,
storage: Option<String>,
}
impl CachedResource {
@@ -923,7 +929,6 @@ lazy_static! {
pub async fn get_cached_resource_value_if_valid(
_db: &DB,
client: &AuthedClient,
_job_id: &Uuid,
_workspace_id: &str,
cached_res_path: &str,
) -> Option<Arc<Box<RawValue>>> {
@@ -955,42 +960,6 @@ pub async fn get_cached_resource_value_if_valid(
},
};
#[cfg(feature = "parquet")]
{
let empty_etags = HashMap::new();
let s3_etags = resource.s3_etags.as_ref().unwrap_or(&empty_etags);
let object_store_resource_opt: Option<ObjectStoreResource> = if s3_etags.is_empty() {
None
} else {
get_workspace_s3_resource_path(_db, &client, _workspace_id, resource.storage.as_ref())
.await
.ok()
.flatten()
};
if !s3_etags.is_empty() && object_store_resource_opt.is_none() {
tracing::warn!("Cached result references s3 files that are not retrievable anymore because the workspace S3 resource can't be fetched. Cache will be invalidated");
return None;
}
for (s3_file_key, s3_file_etag) in s3_etags {
if let Some(object_store_resource) = object_store_resource_opt.as_ref() {
let etag = get_etag_or_empty(
object_store_resource,
S3Object {
s3: s3_file_key.clone(),
storage: resource.storage.clone(),
..Default::default()
},
)
.await;
if etag.as_ref() != Some(s3_file_etag) {
tracing::warn!("S3 file etag for '{}' has changed. Value from cache is {:?} while current value from S3 is {:?}. Cache will be invalidated", s3_file_key.clone(), s3_file_etag, etag);
return None;
}
}
}
}
Some(resource.value.clone())
}
@@ -1003,21 +972,7 @@ pub async fn save_in_cache(
) {
let expire = chrono::Utc::now().timestamp() + job.cache_ttl.unwrap() as i64;
#[cfg(feature = "parquet")]
let (storage, s3_etags) =
arg_value_hash_additions(db, _client, job.workspace_id.as_str(), &r).await;
#[cfg(feature = "parquet")]
let s3_etags = if s3_etags.is_empty() {
None
} else {
Some(s3_etags)
};
#[cfg(not(feature = "parquet"))]
let (storage, s3_etags) = (None, None);
let store_cache_resource = CachedResource { expire, s3_etags, value: r, storage };
let store_cache_resource = CachedResource { expire, value: r };
let raw_json = Json(&store_cache_resource);
if let Err(e) = sqlx::query!(
backend/windmill-worker/src/worker.rs
-1
View File
@@ -2727,7 +2727,6 @@ pub async fn handle_queued_job(
let cached_result_maybe = get_cached_resource_value_if_valid(
db,
&client,
&job.id,
&job.workspace_id,
&cached_res_path,
)
backend/windmill-worker/src/worker_flow.rs
+4
View File
@@ -3992,6 +3992,7 @@ async fn compute_next_flow_transform(
language,
concurrency_settings: concurrency_settings.into(),
cache_ttl: module.cache_ttl.map(|x| x as i32),
cache_ignore_s3_path: module.cache_ignore_s3_path.clone(),
dedicated_worker: None,
path,
},
@@ -4641,6 +4642,7 @@ async fn payload_from_simple_module(
id,
language,
cache_ttl: module.cache_ttl.map(|x| x as i32),
cache_ignore_s3_path: module.cache_ignore_s3_path,
dedicated_worker: None,
path: inner_path,
concurrency_settings: concurrency_settings.into(),
@@ -4672,6 +4674,7 @@ pub fn raw_script_to_payload(
language,
lock,
cache_ttl: module.cache_ttl.map(|x| x as i32),
cache_ignore_s3_path: module.cache_ignore_s3_path,
dedicated_worker: None,
concurrency_settings,
// TODO: Should this have debouncing?
@@ -4776,6 +4779,7 @@ pub async fn script_to_payload(
concurrency_time_window_s,
},
cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(),
cache_ignore_s3_path: module.cache_ignore_s3_path,
language,
dedicated_worker,
priority,
frontend/src/lib/components/ScriptBuilder.svelte
+22 -19
View File
@@ -523,6 +523,7 @@
debounce_key: emptyString(script.debounce_key) ? undefined : script.debounce_key,
debounce_delay_s: script.debounce_delay_s,
cache_ttl: script.cache_ttl,
cache_ignore_s3_path: script.cache_ignore_s3_path,
ws_error_handler_muted: script.ws_error_handler_muted,
priority: script.priority,
restart_unless_cancelled: script.restart_unless_cancelled,
@@ -666,6 +667,7 @@
debounce_key: emptyString(script.debounce_key) ? undefined : script.debounce_key,
debounce_delay_s: script.debounce_delay_s,
cache_ttl: script.cache_ttl,
cache_ignore_s3_path: script.cache_ignore_s3_path,
ws_error_handler_muted: script.ws_error_handler_muted,
priority: script.priority,
restart_unless_cancelled: script.restart_unless_cancelled,
@@ -1305,27 +1307,28 @@
<div class="flex gap-2 shrink flex-col">
<Toggle
size="sm"
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'
}}
bind:checked={
() => !!script.cache_ttl, (v) => (script.cache_ttl = v ? 300 : undefined)
}
options={{ right: 'Cache the results for each possible inputs' }}
/>
{#if Boolean(script.cache_ttl)}
<span class="text-xs font-semibold text-emphasis leading-none mt-2">
How long to the keep cache valid
</span>
{#if script.cache_ttl}
{#if script.cache_ttl}
<div class="text-2xs text-secondary">How long to keep the cache valid</div>
<div class="-mt-5">
<SecondsInput bind:seconds={script.cache_ttl} />
{:else}
<SecondsInput disabled />
{/if}
</div>
<Toggle
size="2xs"
bind:checked={
() => script.cache_ignore_s3_path,
(v) => (script.cache_ignore_s3_path = v || undefined)
}
options={{
right: 'Ignore S3 Object paths for caching purposes',
rightTooltip:
'If two S3 objects passed as input have the same content, they will hit the same cache entry, regardless of their path.'
}}
/>
{/if}
</div>
</Section>
frontend/src/lib/components/flows/content/FlowModuleCache.svelte
+19 -7
View File
@@ -44,12 +44,24 @@
right: 'Cache the results for each possible inputs'
}}
/>
<Label label="How long to keep cache valid">
{#if flowModule.cache_ttl}
<SecondsInput bind:seconds={flowModule.cache_ttl} />
{:else}
<SecondsInput disabled />
{/if}
</Label>
{#if flowModule.cache_ttl}
<Label label="How long to keep cache valid">
<div class="-mt-5">
<SecondsInput bind:seconds={flowModule.cache_ttl} />
</div>
</Label>
<Toggle
size="2xs"
bind:checked={
() => flowModule.cache_ignore_s3_path,
(v) => (flowModule.cache_ignore_s3_path = v || undefined)
}
options={{
right: 'Ignore S3 Object paths for caching purposes',
rightTooltip:
'If two S3 objects passed as input have the same content, they will hit the same cache entry, regardless of their path.'
}}
/>
{/if}
{/if}
</Section>
frontend/src/lib/components/flows/content/FlowSettings.svelte
+13 -6
View File
@@ -212,12 +212,20 @@
<div class="flex gap-x-4 flex-col gap-1 mt-2" transition:slide={{ duration: 120 }}>
<div class="text-2xs text-secondary">How long to keep the cache valid</div>
<div class="-mt-5">
{#if flowStore.val.value.cache_ttl}
<SecondsInput bind:seconds={flowStore.val.value.cache_ttl} />
{:else}
<SecondsInput disabled />
{/if}
<SecondsInput bind:seconds={flowStore.val.value.cache_ttl} />
</div>
<Toggle
size="2xs"
bind:checked={
() => flowStore.val.value.cache_ignore_s3_path,
(v) => (flowStore.val.value.cache_ignore_s3_path = v || undefined)
}
options={{
right: 'Ignore S3 Object paths for caching purposes',
rightTooltip:
'If two S3 objects passed as input have the same content, they will hit the same cache entry, regardless of their path.'
}}
/>
</div>
{/if}
</div>
@@ -466,7 +474,6 @@
{#if customUi?.settingsTabs?.debouncing != false}
<DebounceLimit
size="xs"
color="nord"
fontClass="font-medium"
bind:debounce_delay_s={flowStore.val.value.debounce_delay_s}
bind:debounce_key={flowStore.val.value.debounce_key}
openflow.openapi.yaml
+4
View File
@@ -62,6 +62,8 @@ components:
type: string
cache_ttl:
type: number
cache_ignore_s3_path:
type: boolean
flow_env:
type: object
additionalProperties:
@@ -208,6 +210,8 @@ components:
$ref: "#/components/schemas/InputTransform"
cache_ttl:
type: number
cache_ignore_s3_path:
type: boolean
timeout:
$ref: "#/components/schemas/InputTransform"
delete_after_use: