diff --git a/backend/.sqlx/query-f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf.json b/backend/.sqlx/query-05b69dcef0f4f649513e186e73089979c49b4b8113ee832ea7539b56a0415f32.json similarity index 87% rename from backend/.sqlx/query-f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf.json rename to backend/.sqlx/query-05b69dcef0f4f649513e186e73089979c49b4b8113ee832ea7539b56a0415f32.json index 4e7b02738e..f221e8da2f 100644 --- a/backend/.sqlx/query-f06ab5e0369b35694fa02c3aac685bd547a1d271eb7401df57fe1774de3211bf.json +++ b/backend/.sqlx/query-05b69dcef0f4f649513e186e73089979c49b4b8113ee832ea7539b56a0415f32.json @@ -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" } diff --git a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json index e7ed0aee65..d29a18c691 100644 --- a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json +++ b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json @@ -46,11 +46,11 @@ ] }, "nullable": [ - false, - false, - false, - false, - false, + true, + true, + true, + true, + true, true, true ] diff --git a/backend/.sqlx/query-23759cb515e926e272bbc8e5d8a0a9d039b99bc2026e381e99ef41cdaf6ea19f.json b/backend/.sqlx/query-23759cb515e926e272bbc8e5d8a0a9d039b99bc2026e381e99ef41cdaf6ea19f.json new file mode 100644 index 0000000000..249c230319 --- /dev/null +++ b/backend/.sqlx/query-23759cb515e926e272bbc8e5d8a0a9d039b99bc2026e381e99ef41cdaf6ea19f.json @@ -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" +} diff --git a/backend/.sqlx/query-0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d.json b/backend/.sqlx/query-3d05d9d7e087eb6e1c14c2b8a20598581e6c7493ed99cb9ad1c2ee5d0b212d38.json similarity index 87% rename from backend/.sqlx/query-0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d.json rename to backend/.sqlx/query-3d05d9d7e087eb6e1c14c2b8a20598581e6c7493ed99cb9ad1c2ee5d0b212d38.json index ecaf828737..521b7cb1c0 100644 --- a/backend/.sqlx/query-0a7132202ecf6c4c10340921644a90d9206c45d92a0423c0bc2396d0d66a0b0d.json +++ b/backend/.sqlx/query-3d05d9d7e087eb6e1c14c2b8a20598581e6c7493ed99cb9ad1c2ee5d0b212d38.json @@ -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" } diff --git a/backend/.sqlx/query-53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6.json b/backend/.sqlx/query-53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6.json deleted file mode 100644 index 902ed458c2..0000000000 --- a/backend/.sqlx/query-53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json b/backend/.sqlx/query-6c97ab28ab47b75fb3ff39ea70fa3627f08b61bbd33aecb9ea816f8f78a04ec5.json similarity index 90% rename from backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json rename to backend/.sqlx/query-6c97ab28ab47b75fb3ff39ea70fa3627f08b61bbd33aecb9ea816f8f78a04ec5.json index fa013ba585..272201898c 100644 --- a/backend/.sqlx/query-3162ec92bb32af47a71cc41172cc740b5dea1304ce4dfdb4d3d0efa4266f38c5.json +++ b/backend/.sqlx/query-6c97ab28ab47b75fb3ff39ea70fa3627f08b61bbd33aecb9ea816f8f78a04ec5.json @@ -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>>\",\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>\",\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>>\",\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>\",\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" } diff --git a/backend/.sqlx/query-6cc922a5bbd348c938a9d1431aaa0f24f078ea814b429d44403aca1e5002e750.json b/backend/.sqlx/query-7b5ad10af2a9b34fa86429499ea24c0c09c6e7e9ebfa3af90035570133f7c579.json similarity index 93% rename from backend/.sqlx/query-6cc922a5bbd348c938a9d1431aaa0f24f078ea814b429d44403aca1e5002e750.json rename to backend/.sqlx/query-7b5ad10af2a9b34fa86429499ea24c0c09c6e7e9ebfa3af90035570133f7c579.json index 4b9bfe326b..5ce7b053fc 100644 --- a/backend/.sqlx/query-6cc922a5bbd348c938a9d1431aaa0f24f078ea814b429d44403aca1e5002e750.json +++ b/backend/.sqlx/query-7b5ad10af2a9b34fa86429499ea24c0c09c6e7e9ebfa3af90035570133f7c579.json @@ -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" } diff --git a/backend/.sqlx/query-27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32.json b/backend/.sqlx/query-7f9b7ab9bec6a0f745273d0cd5602ceab46a7ec9fd225f7b9d16a2ddb9bad7b3.json similarity index 84% rename from backend/.sqlx/query-27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32.json rename to backend/.sqlx/query-7f9b7ab9bec6a0f745273d0cd5602ceab46a7ec9fd225f7b9d16a2ddb9bad7b3.json index c540742fe2..9033ac194b 100644 --- a/backend/.sqlx/query-27a54f8188c25c2c089c818a991ca1c092f67227be217161d6e6617ddbf77b32.json +++ b/backend/.sqlx/query-7f9b7ab9bec6a0f745273d0cd5602ceab46a7ec9fd225f7b9d16a2ddb9bad7b3.json @@ -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" } diff --git a/backend/.sqlx/query-d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd.json b/backend/.sqlx/query-a84e67035584bbdb02482026b9cc0808086c50f78947d43bb88628a481f41a1d.json similarity index 91% rename from backend/.sqlx/query-d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd.json rename to backend/.sqlx/query-a84e67035584bbdb02482026b9cc0808086c50f78947d43bb88628a481f41a1d.json index 73fb2e98d8..8b9b0246b2 100644 --- a/backend/.sqlx/query-d48c9a748746080e9b6cf0366b2ca3516ac5c1d32a98946196b1a42e3f103efd.json +++ b/backend/.sqlx/query-a84e67035584bbdb02482026b9cc0808086c50f78947d43bb88628a481f41a1d.json @@ -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" } diff --git a/backend/.sqlx/query-bf2163c542fb8c4e173167a8f333ef762fecf782424c5b61b89f32918b8d6971.json b/backend/.sqlx/query-b179a3f876ca659bed892d464bf51a733cc86a3204fcd9edccda63fddc97dced.json similarity index 94% rename from backend/.sqlx/query-bf2163c542fb8c4e173167a8f333ef762fecf782424c5b61b89f32918b8d6971.json rename to backend/.sqlx/query-b179a3f876ca659bed892d464bf51a733cc86a3204fcd9edccda63fddc97dced.json index 46db07e1e0..ee004c664c 100644 --- a/backend/.sqlx/query-bf2163c542fb8c4e173167a8f333ef762fecf782424c5b61b89f32918b8d6971.json +++ b/backend/.sqlx/query-b179a3f876ca659bed892d464bf51a733cc86a3204fcd9edccda63fddc97dced.json @@ -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" } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index d57da779de..20e0fa2c17 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -937820a9e572c10c0ce050e1bd5eddb3180fe921 +f03dfeccbe7610c1e3029fb67b47faeba8fac918 \ No newline at end of file diff --git a/backend/migrations/20251204210743_cache_ignore_s3_path.down.sql b/backend/migrations/20251204210743_cache_ignore_s3_path.down.sql new file mode 100644 index 0000000000..cac9727e15 --- /dev/null +++ b/backend/migrations/20251204210743_cache_ignore_s3_path.down.sql @@ -0,0 +1,5 @@ +ALTER TABLE v2_job_queue +DROP COLUMN cache_ignore_s3_path; + +ALTER TABLE script +DROP COLUMN cache_ignore_s3_path; diff --git a/backend/migrations/20251204210743_cache_ignore_s3_path.up.sql b/backend/migrations/20251204210743_cache_ignore_s3_path.up.sql new file mode 100644 index 0000000000..dfbf97a2c8 --- /dev/null +++ b/backend/migrations/20251204210743_cache_ignore_s3_path.up.sql @@ -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; diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 421d1dd8c3..7150eaaf84 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -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(), diff --git a/backend/tests/job_payload.rs b/backend/tests/job_payload.rs index 5eabdc83ce..2ad2bf339d 100644 --- a/backend/tests/job_payload.rs +++ b/backend/tests/job_payload.rs @@ -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(), }) diff --git a/backend/tests/python_jobs.rs b/backend/tests/python_jobs.rs index f6fc35295c..cb5909f183 100644 --- a/backend/tests/python_jobs.rs +++ b/backend/tests/python_jobs.rs @@ -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, }); diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index d18869efb8..59c5f32941 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -201,6 +201,7 @@ async fn test_deno_flow(db: Pool) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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) -> 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 { 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(), diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index afcd9c896d..8875aab7a2 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -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: diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 6ae85b7ad0..d64292765f 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -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, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index add81f0b03..39edaade1d 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -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, pub timeout: Option, pub cache_ttl: Option, + pub cache_ignore_s3_path: Option, pub skip_preprocessor: Option, pub poll_delay_ms: Option, pub memory_id: Option, @@ -3537,6 +3538,7 @@ impl<'a> From 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, diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index f71d6e00ed..fb777b6ed6 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -94,6 +94,8 @@ pub struct ScriptWDraft { #[serde(skip_serializing_if = "Option::is_none")] pub cache_ttl: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub cache_ignore_s3_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub dedicated_worker: Option, #[serde(skip_serializing_if = "Option::is_none")] pub ws_error_handler_muted: Option, @@ -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", diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index e3ae82b172..b580138191 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -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, diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 2d49d708ab..678ec06c68 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -181,6 +181,8 @@ pub struct FlowValue { #[serde(skip_serializing_if = "Option::is_none")] pub cache_ttl: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub cache_ignore_s3_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub early_return: Option, #[serde(skip_serializing_if = "Option::is_none")] // Priority at the flow level @@ -442,6 +444,8 @@ pub struct FlowModule { pub sleep: Option, #[serde(skip_serializing_if = "Option::is_none")] pub cache_ttl: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cache_ignore_s3_path: Option, #[serde( default, deserialize_with = "raw_value_to_input_transform::<_, i32>", @@ -1144,6 +1148,7 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec) { sleep: None, suspend: None, cache_ttl: None, + cache_ignore_s3_path: None, timeout: None, priority: None, delete_after_use: None, diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 3abacd93fc..693343e11b 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -186,6 +186,8 @@ pub struct QueuedJob { #[serde(skip_serializing_if = "Option::is_none")] pub cache_ttl: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub cache_ignore_s3_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub priority: Option, #[serde(skip_serializing_if = "Option::is_none")] pub preprocessed: Option, @@ -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, + cache_ignore_s3_path: Option, dedicated_worker: Option, language: ScriptLang, priority: Option, @@ -366,6 +370,7 @@ pub enum JobPayload { path: String, language: ScriptLang, cache_ttl: Option, + cache_ignore_s3_path: Option, dedicated_worker: Option, concurrency_settings: ConcurrencySettings, }, @@ -448,6 +453,7 @@ pub enum JobPayload { error_handler_args: Option>>, skip_handler: Option, cache_ttl: Option, + cache_ignore_s3_path: Option, priority: Option, tag_override: Option, trigger_path: Option, @@ -584,6 +590,7 @@ pub struct RawCode { pub language: ScriptLang, pub lock: Option, pub cache_ttl: Option, + pub cache_ignore_s3_path: Option, pub dedicated_worker: Option, #[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, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 1f3dbd414b..73390ef6f2 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -659,6 +659,7 @@ pub struct ScriptHashInfo { pub debounce_key: Option, pub debounce_delay_s: Option, pub cache_ttl: Option, + pub cache_ignore_s3_path: Option, pub language: ScriptLang, pub dedicated_worker: Option, pub priority: Option, @@ -804,7 +805,7 @@ async fn get_script_info_for_hash_inner<'e, E: sqlx::PgExecutor<'e>>( ) -> error::Result> { 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, Option, Option, + Option, ScriptLang, Option, Option, @@ -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, diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 25bde2804e..347d69c1a5 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -393,6 +393,8 @@ pub struct Script { #[serde(skip_serializing_if = "Option::is_none")] pub cache_ttl: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub cache_ignore_s3_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub timeout: Option, #[serde(skip_serializing_if = "Option::is_none")] pub delete_after_use: Option, @@ -507,6 +509,7 @@ pub struct NewScript { #[serde(skip_serializing_if = "Option::is_none")] pub debounce_delay_s: Option, pub cache_ttl: Option, + pub cache_ignore_s3_path: Option, pub dedicated_worker: Option, pub ws_error_handler_muted: Option, pub priority: Option, @@ -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 diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 4801b93a78..1c98976bcf 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -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 diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 8f7b2ed7a1..6dce1bb229 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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, pub flow_step_id: Option, pub cache_ttl: Option, + pub cache_ignore_s3_path: Option, pub priority: Option, pub preprocessed: Option, pub script_entrypoint_override: Option, @@ -2009,6 +2013,7 @@ pub struct MiniCompletedJob { pub concurrent_limit: Option, pub tag: String, pub cache_ttl: Option, + pub cache_ignore_s3_path: Option, } impl From for MiniCompletedJob { @@ -2034,6 +2039,7 @@ impl From 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 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> 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, pub tag: String, pub cache_ttl: Option, + pub cache_ignore_s3_path: Option, pub last_ping: Option>, pub worker: Option, pub memory_peak: Option, @@ -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, language: Option, cache_ttl: Option, + cache_ignore_s3_path: Option, dedicated_worker: Option, _low_level_priority: Option, 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, 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, diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 4a74ed80d6..16358b453e 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -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, diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index d0f50ec750..638129e34d 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -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, }, diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 4856e030f7..ed4cfccd26 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -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>>>, 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::(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> { 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::(path.as_str()) + .get_resource_value_interpolated::( + 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, -) -> (Option, HashMap) { - let mut result: HashMap = HashMap::new(); - - let parsed_value = serde_json::from_str::(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>, value: Arc>, - storage: Option, } 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>> { @@ -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 = 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!( diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 31275acbc1..4d8d7506fc 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -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, ) diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 427951f18b..fdbc9ba636 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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, diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index 8d2b022e3b..cb700e2676 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -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 @@
{ - 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)} - - How long to the keep cache valid - - {#if script.cache_ttl} + {#if script.cache_ttl} +
How long to keep the cache valid
+
- {:else} - - {/if} +
+ 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}
diff --git a/frontend/src/lib/components/flows/content/FlowModuleCache.svelte b/frontend/src/lib/components/flows/content/FlowModuleCache.svelte index 72749eb46e..6f1927f1c6 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleCache.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleCache.svelte @@ -44,12 +44,24 @@ right: 'Cache the results for each possible inputs' }} /> - + {#if flowModule.cache_ttl} + + 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} diff --git a/frontend/src/lib/components/flows/content/FlowSettings.svelte b/frontend/src/lib/components/flows/content/FlowSettings.svelte index dc166dd564..41433ab6c8 100644 --- a/frontend/src/lib/components/flows/content/FlowSettings.svelte +++ b/frontend/src/lib/components/flows/content/FlowSettings.svelte @@ -212,12 +212,20 @@
How long to keep the cache valid
- {#if flowStore.val.value.cache_ttl} - - {:else} - - {/if} +
+ 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.' + }} + />
{/if} @@ -466,7 +474,6 @@ {#if customUi?.settingsTabs?.debouncing != false}