feat(internal): runnable settings (#7298)

* rework everything again

Signed-off-by: pyranota <pyra@duck.com>

* updcate sqlx

Signed-off-by: pyranota <pyra@duck.com>

* update ref

Signed-off-by: pyranota <pyra@duck.com>

* fix things

Signed-off-by: pyranota <pyra@duck.com>

* fix function

Signed-off-by: pyranota <pyra@duck.com>

* final fixes

Signed-off-by: pyranota <pyra@duck.com>

* update sqlx

Signed-off-by: pyranota <pyra@duck.com>

* fix script creation

Signed-off-by: pyranota <pyra@duck.com>

* address todo

Signed-off-by: pyranota <pyra@duck.com>

* cleanup

Signed-off-by: pyranota <pyra@duck.com>

* remove dbg

Signed-off-by: pyranota <pyra@duck.com>

* cleanup

Signed-off-by: pyranota <pyra@duck.com>

* fix

Signed-off-by: pyranota <pyra@duck.com>

* fixups

Signed-off-by: pyranota <pyra@duck.com>

* fix cargo.toml

Signed-off-by: pyranota <pyra@duck.com>

* update ee repo

Signed-off-by: pyranota <pyra@duck.com>

* fix ci

Signed-off-by: pyranota <pyra@duck.com>

* nit

Signed-off-by: pyranota <pyra@duck.com>

* ref

Signed-off-by: pyranota <pyra@duck.com>

* fix

Signed-off-by: pyranota <pyra@duck.com>

* ee repo

Signed-off-by: pyranota <pyra@duck.com>

* sqlx

Signed-off-by: pyranota <pyra@duck.com>

* ee ref

Signed-off-by: pyranota <pyra@duck.com>

* remove dbg

Signed-off-by: pyranota <pyra@duck.com>

* sqlx

Signed-off-by: pyranota <pyra@duck.com>

* chore: update ee-repo-ref to 505eadbff32d102ea5245a2bef88ce6f1bb95395

This commit updates the EE repository reference after PR #348 was merged in windmill-ee-private.

Previous ee-repo-ref: 195243e56cc0eab55f8890fa57297206bfe2c18c

New ee-repo-ref: 505eadbff32d102ea5245a2bef88ce6f1bb95395

Automated by sync-ee-ref workflow.

* ci: force runnable settings

Signed-off-by: pyranota <pyra@duck.com>

---------

Signed-off-by: pyranota <pyra@duck.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Pyra
2025-12-16 22:17:06 +01:00
committed by GitHub
parent 75fdc2cdc9
commit fe56191422
43 changed files with 1412 additions and 745 deletions
+1
View File
@@ -84,6 +84,7 @@ jobs:
RUST_LOG_STYLE: never
CARGO_NET_GIT_FETCH_WITH_CLI: true
WMDEBUG_FORCE_V0_WORKSPACE_DEPENDENCIES: 1
WMDEBUG_FORCE_RUNNABLE_SETTINGS_V0: 1
run: |
deno --version && bun -v && go version && python3 --version
cd windmill-duckdb-ffi-internal && ./build_dev.sh && cd ..
@@ -1,156 +0,0 @@
{
"db_name": "PostgreSQL",
"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": [
{
"ordinal": 0,
"name": "hash",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "concurrency_key",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "concurrent_limit",
"type_info": "Int4"
},
{
"ordinal": 4,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 5,
"name": "debounce_key",
"type_info": "Varchar"
},
{
"ordinal": 6,
"name": "debounce_delay_s",
"type_info": "Int4"
},
{
"ordinal": 7,
"name": "cache_ttl",
"type_info": "Int4"
},
{
"ordinal": 8,
"name": "cache_ignore_s3_path",
"type_info": "Bool"
},
{
"ordinal": 9,
"name": "language: ScriptLang",
"type_info": {
"Custom": {
"name": "script_lang",
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash",
"postgresql",
"nativets",
"bun",
"mysql",
"bigquery",
"snowflake",
"graphql",
"powershell",
"mssql",
"php",
"bunnative",
"rust",
"ansible",
"csharp",
"oracledb",
"nu",
"java",
"duckdb",
"ruby"
]
}
}
}
},
{
"ordinal": 10,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 11,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 12,
"name": "delete_after_use",
"type_info": "Bool"
},
{
"ordinal": 13,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 14,
"name": "has_preprocessor",
"type_info": "Bool"
},
{
"ordinal": 15,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 16,
"name": "created_by",
"type_info": "Varchar"
},
{
"ordinal": 17,
"name": "path",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Int8",
"Text"
]
},
"nullable": [
false,
true,
true,
true,
true,
true,
true,
true,
true,
false,
true,
true,
true,
true,
true,
true,
false,
false
]
},
"hash": "05b69dcef0f4f649513e186e73089979c49b4b8113ee832ea7539b56a0415f32"
}
@@ -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, cache_ignore_s3_path)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42)",
"query": "WITH inserted_job AS (\n INSERT INTO v2_job (\n id, -- 1\n workspace_id, -- 2\n raw_code, -- 3\n raw_lock, -- 4\n raw_flow, -- 5\n tag, -- 6\n parent_job, -- 7\n created_by, -- 8\n permissioned_as, -- 9\n runnable_id, -- 10\n runnable_path, -- 11\n args, -- 12\n kind, -- 13\n trigger, -- 14\n script_lang, -- 15\n same_worker, -- 16\n pre_run_error, -- 17 \n permissioned_as_email, -- 18\n visible_to_owner, -- 19\n flow_innermost_root_job, -- 20\n root_job, -- 38\n concurrent_limit, -- 21\n concurrency_time_window_s, -- 22\n timeout, -- 23\n flow_step_id, -- 24\n cache_ttl, -- 25\n priority, -- 26\n trigger_kind, -- 39\n script_entrypoint_override, -- 12\n preprocessed -- 27,\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, runnable_settings_handle)\n VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42, $43)",
"describe": {
"columns": [],
"parameters": {
@@ -128,10 +128,11 @@
},
"Bool",
"Varchar",
"Bool"
"Bool",
"Int8"
]
},
"nullable": []
},
"hash": "b179a3f876ca659bed892d464bf51a733cc86a3204fcd9edccda63fddc97dced"
"hash": "14276a040cb4db88d71fccdc3579e8c0bb132b70668301b535872d1632753e30"
}
@@ -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, 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"
}
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO runnable_settings (hash, debouncing_settings, concurrency_settings)\n VALUES ($1, $2, $3)\n ON CONFLICT (hash)\n DO NOTHING",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Int8",
"Int8"
]
},
"nullable": []
},
"hash": "451d9cde90d14071e21ffb5f615052b7ba7fc315fc301ed5c0ff50d9a3ab0d4a"
}
@@ -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 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",
"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 v2_job_queue.runnable_settings_handle,\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": [
{
@@ -35,16 +35,21 @@
},
{
"ordinal": 6,
"name": "runnable_settings_handle",
"type_info": "Int8"
},
{
"ordinal": 7,
"name": "scheduled_for",
"type_info": "Timestamptz"
},
{
"ordinal": 7,
"ordinal": 8,
"name": "runnable_path",
"type_info": "Varchar"
},
{
"ordinal": 8,
"ordinal": 9,
"name": "kind: JobKind",
"type_info": {
"Custom": {
@@ -79,42 +84,42 @@
}
},
{
"ordinal": 9,
"ordinal": 10,
"name": "runnable_id: ScriptHash",
"type_info": "Int8"
},
{
"ordinal": 10,
"ordinal": 11,
"name": "canceled_reason",
"type_info": "Text"
},
{
"ordinal": 11,
"ordinal": 12,
"name": "canceled_by",
"type_info": "Varchar"
},
{
"ordinal": 12,
"ordinal": 13,
"name": "permissioned_as",
"type_info": "Varchar"
},
{
"ordinal": 13,
"ordinal": 14,
"name": "permissioned_as_email",
"type_info": "Varchar"
},
{
"ordinal": 14,
"ordinal": 15,
"name": "flow_status: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 15,
"ordinal": 16,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 16,
"ordinal": 17,
"name": "script_lang: ScriptLang",
"type_info": {
"Custom": {
@@ -150,77 +155,77 @@
}
},
{
"ordinal": 17,
"ordinal": 18,
"name": "same_worker",
"type_info": "Bool"
},
{
"ordinal": 18,
"ordinal": 19,
"name": "pre_run_error",
"type_info": "Text"
},
{
"ordinal": 19,
"ordinal": 20,
"name": "concurrent_limit",
"type_info": "Int4"
},
{
"ordinal": 20,
"ordinal": 21,
"name": "concurrency_time_window_s",
"type_info": "Int4"
},
{
"ordinal": 21,
"ordinal": 22,
"name": "flow_innermost_root_job",
"type_info": "Uuid"
},
{
"ordinal": 22,
"ordinal": 23,
"name": "root_job",
"type_info": "Uuid"
},
{
"ordinal": 23,
"ordinal": 24,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 24,
"ordinal": 25,
"name": "flow_step_id",
"type_info": "Varchar"
},
{
"ordinal": 25,
"ordinal": 26,
"name": "cache_ttl",
"type_info": "Int4"
},
{
"ordinal": 26,
"ordinal": 27,
"name": "cache_ignore_s3_path",
"type_info": "Bool"
},
{
"ordinal": 27,
"ordinal": 28,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 28,
"ordinal": 29,
"name": "preprocessed",
"type_info": "Bool"
},
{
"ordinal": 29,
"ordinal": 30,
"name": "script_entrypoint_override",
"type_info": "Varchar"
},
{
"ordinal": 30,
"ordinal": 31,
"name": "trigger",
"type_info": "Varchar"
},
{
"ordinal": 31,
"ordinal": 32,
"name": "trigger_kind: JobTriggerKind",
"type_info": {
"Custom": {
@@ -246,12 +251,12 @@
}
},
{
"ordinal": 32,
"ordinal": 33,
"name": "visible_to_owner",
"type_info": "Bool"
},
{
"ordinal": 33,
"ordinal": 34,
"name": "permissioned_as_end_user_email",
"type_info": "Text"
}
@@ -268,6 +273,7 @@
true,
false,
true,
true,
false,
true,
false,
@@ -298,5 +304,5 @@
null
]
},
"hash": "6c97ab28ab47b75fb3ff39ea70fa3627f08b61bbd33aecb9ea816f8f78a04ec5"
"hash": "5bf200f2c8db25ddf231b564503c6c70f7f3958564a79bb0c6b3863b1ebb0cbf"
}
@@ -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, runnable_settings_handle)\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, runnable_settings_handle\n\n FROM script WHERE hash = $2 AND workspace_id = $3;\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Int8",
"Text"
]
},
"nullable": []
},
"hash": "5c056ad6cc8967393729288437205c605a24118021fdb2b21b6b61695dc4ff28"
}
@@ -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, 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",
"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, q.runnable_settings_handle\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": [
{
@@ -189,6 +189,11 @@
"ordinal": 20,
"name": "cache_ignore_s3_path",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "runnable_settings_handle",
"type_info": "Int8"
}
],
"parameters": {
@@ -218,8 +223,9 @@
true,
false,
true,
true,
true
]
},
"hash": "7b5ad10af2a9b34fa86429499ea24c0c09c6e7e9ebfa3af90035570133f7c579"
"hash": "a1745a4f525b251d2f5a602ab2b2ede46b4471e21b11f607573a844013911abe"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE((SELECT MIN(started_at) as min_started_at\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id\n WHERE v2_job.runnable_path = $1 AND v2_job.kind != 'dependencies' AND v2_job_queue.running = true AND v2_job_queue.workspace_id = $2 AND v2_job_queue.canceled_by IS NULL AND v2_job.concurrent_limit > 0), $3) as min_started_at, now() AS now",
"query": "SELECT COALESCE((SELECT MIN(started_at) as min_started_at\n FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id LEFT JOIN runnable_settings rs ON rs.hash = v2_job_queue.runnable_settings_handle LEFT JOIN concurrency_settings cs ON cs.hash = rs.concurrency_settings\n WHERE v2_job.runnable_path = $1 AND v2_job.kind != 'dependencies' AND v2_job_queue.running = true AND v2_job_queue.workspace_id = $2 AND v2_job_queue.canceled_by IS NULL AND COALESCE(cs.concurrent_limit, v2_job.concurrent_limit) > 0), $3) as min_started_at, now() AS now",
"describe": {
"columns": [
{
@@ -26,5 +26,5 @@
null
]
},
"hash": "6b6f8f7b4a6b6e7e41a9da8b6dfdbcae842ff252cc355bd91aeeb5e26dcc74f3"
"hash": "a2e52f033120a3f0b64e0a5ba125df7ce0d25096a23f0b655846a1c07b41f620"
}
@@ -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, 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",
"query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, cache_ignore_s3_path, runnable_settings_handle, 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": [
{
@@ -50,6 +50,11 @@
},
{
"ordinal": 9,
"name": "runnable_settings_handle",
"type_info": "Int8"
},
{
"ordinal": 10,
"name": "language: ScriptLang",
"type_info": {
"Custom": {
@@ -85,27 +90,27 @@
}
},
{
"ordinal": 10,
"ordinal": 11,
"name": "dedicated_worker",
"type_info": "Bool"
},
{
"ordinal": 11,
"ordinal": 12,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 12,
"ordinal": 13,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 13,
"ordinal": 14,
"name": "on_behalf_of_email",
"type_info": "Text"
},
{
"ordinal": 14,
"ordinal": 15,
"name": "created_by",
"type_info": "Varchar"
}
@@ -127,6 +132,7 @@
true,
true,
true,
true,
false,
true,
true,
@@ -135,5 +141,5 @@
false
]
},
"hash": "7f9b7ab9bec6a0f745273d0cd5602ceab46a7ec9fd225f7b9d16a2ddb9bad7b3"
"hash": "a33673ebc4d1eb4c3513987dbc43e2c80974598e1d9fe7203145bfc29928ba65"
}
@@ -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, 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",
"query": "SELECT\n id,\n q.runnable_settings_handle,\n q.workspace_id,\n j.runnable_id as \"runnable_id: ScriptHash\",\n scheduled_for,\n parent_job,\n flow_innermost_root_job,\n runnable_path,\n kind as \"kind: JobKind\",\n started_at,\n permissioned_as,\n created_by,\n script_lang as \"script_lang: ScriptLang\",\n permissioned_as_email,\n flow_step_id,\n trigger_kind as \"trigger_kind: JobTriggerKind\",\n trigger,\n q.priority,\n concurrent_limit,\n q.tag,\n cache_ttl,\n cache_ignore_s3_path,\n r.ping as last_ping,\n worker,\n memory_peak,\n running\n FROM v2_job_queue q\n JOIN v2_job j USING (id)\n LEFT JOIN v2_job_runtime r USING (id)\n LEFT JOIN v2_job_status s USING (id)\n WHERE j.id = $1",
"describe": {
"columns": [
{
@@ -10,36 +10,41 @@
},
{
"ordinal": 1,
"name": "runnable_settings_handle",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Varchar"
},
{
"ordinal": 2,
"ordinal": 3,
"name": "runnable_id: ScriptHash",
"type_info": "Int8"
},
{
"ordinal": 3,
"ordinal": 4,
"name": "scheduled_for",
"type_info": "Timestamptz"
},
{
"ordinal": 4,
"ordinal": 5,
"name": "parent_job",
"type_info": "Uuid"
},
{
"ordinal": 5,
"ordinal": 6,
"name": "flow_innermost_root_job",
"type_info": "Uuid"
},
{
"ordinal": 6,
"ordinal": 7,
"name": "runnable_path",
"type_info": "Varchar"
},
{
"ordinal": 7,
"ordinal": 8,
"name": "kind: JobKind",
"type_info": {
"Custom": {
@@ -74,22 +79,22 @@
}
},
{
"ordinal": 8,
"ordinal": 9,
"name": "started_at",
"type_info": "Timestamptz"
},
{
"ordinal": 9,
"ordinal": 10,
"name": "permissioned_as",
"type_info": "Varchar"
},
{
"ordinal": 10,
"ordinal": 11,
"name": "created_by",
"type_info": "Varchar"
},
{
"ordinal": 11,
"ordinal": 12,
"name": "script_lang: ScriptLang",
"type_info": {
"Custom": {
@@ -125,17 +130,17 @@
}
},
{
"ordinal": 12,
"ordinal": 13,
"name": "permissioned_as_email",
"type_info": "Varchar"
},
{
"ordinal": 13,
"ordinal": 14,
"name": "flow_step_id",
"type_info": "Varchar"
},
{
"ordinal": 14,
"ordinal": 15,
"name": "trigger_kind: JobTriggerKind",
"type_info": {
"Custom": {
@@ -161,52 +166,52 @@
}
},
{
"ordinal": 15,
"ordinal": 16,
"name": "trigger",
"type_info": "Varchar"
},
{
"ordinal": 16,
"ordinal": 17,
"name": "priority",
"type_info": "Int2"
},
{
"ordinal": 17,
"ordinal": 18,
"name": "concurrent_limit",
"type_info": "Int4"
},
{
"ordinal": 18,
"ordinal": 19,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 19,
"ordinal": 20,
"name": "cache_ttl",
"type_info": "Int4"
},
{
"ordinal": 20,
"ordinal": 21,
"name": "cache_ignore_s3_path",
"type_info": "Bool"
},
{
"ordinal": 21,
"ordinal": 22,
"name": "last_ping",
"type_info": "Timestamptz"
},
{
"ordinal": 22,
"ordinal": 23,
"name": "worker",
"type_info": "Varchar"
},
{
"ordinal": 23,
"ordinal": 24,
"name": "memory_peak",
"type_info": "Int4"
},
{
"ordinal": 24,
"ordinal": 25,
"name": "running",
"type_info": "Bool"
}
@@ -218,6 +223,7 @@
},
"nullable": [
false,
true,
false,
true,
false,
@@ -244,5 +250,5 @@
false
]
},
"hash": "a84e67035584bbdb02482026b9cc0808086c50f78947d43bb88628a481f41a1d"
"hash": "b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc"
}
@@ -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, 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)",
"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, runnable_settings_handle) 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, $38)",
"describe": {
"columns": [],
"parameters": {
@@ -86,10 +86,11 @@
"Jsonb",
"Varchar",
"Int4",
"Bool"
"Bool",
"Int8"
]
},
"nullable": []
},
"hash": "3d05d9d7e087eb6e1c14c2b8a20598581e6c7493ed99cb9ad1c2ee5d0b212d38"
"hash": "b4eb72b0274cbdce7490f63c36d0d16ee847294fadc138593a1baa417cbb3652"
}
@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "SELECT concurrency_settings, debouncing_settings FROM runnable_settings WHERE hash = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "concurrency_settings",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "debouncing_settings",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Int8"
]
},
"nullable": [
true,
true
]
},
"hash": "ebbbd069e0f33be9609604025d159fe1ecbefc2e9c11f7c4900b7121d4367e01"
}
+1
View File
@@ -15201,6 +15201,7 @@ dependencies = [
"sha1",
"sha2 0.10.9",
"size",
"sql-builder",
"sqlx",
"strum 0.27.2",
"systemstat",
+1
View File
@@ -125,6 +125,7 @@ windmill-autoscaling = { workspace = true, optional = true }
futures.workspace = true
tracing.workspace = true
sqlx.workspace = true
sql-builder.workspace = true
rand.workspace = true
chrono.workspace = true
git-version.workspace = true
+1 -1
View File
@@ -1 +1 @@
d347295041426d03039b747a148a71e3583c3a6b
505eadbff32d102ea5245a2bef88ce6f1bb95395
@@ -0,0 +1,11 @@
ALTER TABLE v2_job_queue
DROP COLUMN runnable_settings_handle;
ALTER TABLE script
DROP COLUMN runnable_settings_handle;
DROP TABLE IF EXISTS job_settings;
DROP TABLE IF EXISTS runnable_settings;
DROP TABLE IF EXISTS concurrency_settings;
DROP TABLE IF EXISTS debouncing_settings;
@@ -0,0 +1,42 @@
CREATE TABLE IF NOT EXISTS concurrency_settings(
hash BIGINT PRIMARY KEY,
concurrency_key VARCHAR(255),
concurrent_limit INTEGER,
concurrency_time_window_s INTEGER
);
CREATE TABLE IF NOT EXISTS debouncing_settings(
hash BIGINT PRIMARY KEY,
debounce_key VARCHAR(255),
debounce_delay_s INTEGER,
max_total_debouncing_time INTEGER,
max_total_debounces_amount INTEGER,
debounce_args_to_accumulate TEXT[]
);
CREATE TABLE IF NOT EXISTS runnable_settings(
hash BIGINT PRIMARY KEY,
debouncing_settings BIGINT DEFAULT NULL,
concurrency_settings BIGINT DEFAULT NULL
);
CREATE TABLE IF NOT EXISTS job_settings(
job_id UUID PRIMARY KEY,
runnable_settings BIGINT DEFAULT NULL
);
ALTER TABLE script
ADD COLUMN runnable_settings_handle BIGINT DEFAULT NULL;
ALTER TABLE v2_job_queue
ADD COLUMN runnable_settings_handle BIGINT DEFAULT NULL;
GRANT ALL ON concurrency_settings TO windmill_admin;
GRANT ALL ON concurrency_settings TO windmill_user;
GRANT ALL ON debouncing_settings TO windmill_admin;
GRANT ALL ON debouncing_settings TO windmill_user;
GRANT ALL ON runnable_settings TO windmill_admin;
GRANT ALL ON runnable_settings TO windmill_user;
GRANT ALL ON job_settings TO windmill_admin;
GRANT ALL ON job_settings TO windmill_user;
+4 -4
View File
@@ -691,8 +691,8 @@ pub async fn run_deployed_relative_imports(
language,
priority: None,
apply_preprocessor: false,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
})
.push(&db2)
.await;
@@ -741,8 +741,8 @@ pub async fn run_preview_relative_imports(
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.push(&db2)
.await;
+6 -6
View File
@@ -52,8 +52,8 @@ mod job_payload {
let result = RunJob::from(JobPayload::ScriptHash {
hash: ScriptHash(123412),
path: "f/system/hello".to_string(),
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
@@ -90,8 +90,8 @@ mod job_payload {
language: ScriptLang::Deno,
priority: None,
apply_preprocessor: true,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
})
.run_until_complete_with(db, false, port, |id| async move {
let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id)
@@ -163,7 +163,7 @@ mod job_payload {
let result = RunJob::from(JobPayload::FlowScript {
id: flow_scripts[0],
language: ScriptLang::Deno,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
@@ -182,7 +182,7 @@ mod job_payload {
let result = RunJob::from(JobPayload::FlowScript {
id: flow_scripts[1],
language: ScriptLang::Deno,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
+10 -10
View File
@@ -189,8 +189,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
@@ -238,8 +238,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
@@ -272,8 +272,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
@@ -311,8 +311,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
@@ -348,8 +348,8 @@ def main():
path: None,
language: ScriptLang::Python3,
lock: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
cache_ttl: None,
cache_ignore_s3_path: None,
dedicated_worker: None,
+24 -24
View File
@@ -188,7 +188,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default()
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
.into(),
is_trigger: None,
assets: None,
@@ -235,7 +235,7 @@ async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
lock: None,
tag: None,
concurrency_settings:
windmill_common::jobs::ConcurrencySettings::default().into(),
windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}
@@ -369,7 +369,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
@@ -425,7 +425,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}.into(),
@@ -465,7 +465,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
@@ -533,7 +533,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
path: None,
lock: None,
tag: None,
concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
is_trigger: None,
assets: None,
}.into(),
@@ -865,8 +865,8 @@ func main(derp string) (string, error) {
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.arg("derp", json!("world"))
.run_until_complete(&db, false, port)
@@ -900,8 +900,8 @@ fn main(world: String) -> Result<String, String> {
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
cache_ttl: None,
dedicated_worker: None,
}))
@@ -978,8 +978,8 @@ echo "hello $msg"
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.arg("msg", json!("world"))
.run_until_complete(&db, false, port)
@@ -1011,8 +1011,8 @@ def main [ msg: string ] {
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.arg("msg", json!("world"))
.run_until_complete(&db, false, port)
@@ -1064,8 +1064,8 @@ def main [
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.arg("a", json!("3"))
.arg("b", json!("null"))
@@ -1126,8 +1126,8 @@ public class Main {
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.arg("a", json!(3))
.arg("b", json!(3.0))
@@ -1161,8 +1161,8 @@ export async function main(a: Date) {
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.run_until_complete(&db, false, port)
@@ -1196,8 +1196,8 @@ export async function main(a: Date) {
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.run_until_complete(&db, false, port)
@@ -1232,8 +1232,8 @@ def main(a: datetime, b: bytes):
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(),
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default().into(),
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
}))
.arg("a", json!("2024-09-24T10:00:00.000Z"))
.arg("b", json!("dGVzdA=="))
+4 -1
View File
@@ -32,6 +32,7 @@ use sql_builder::prelude::*;
use sqlx::{FromRow, Postgres, Transaction};
use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_common::runnable_settings::RunnableSettingsTrait;
use windmill_common::utils::{query_elems_from_hub, WarnAfterExt};
use windmill_common::worker::{to_raw_value, CLOUD_HOSTED, MIN_VERSION_SUPPORTS_DEBOUNCING};
use windmill_common::HUB_BASE_URL;
@@ -1601,7 +1602,9 @@ mod tests {
ConstantDelay, ExponentialDelay, FlowModule, FlowModuleValue, FlowValue,
InputTransform, Retry, StopAfterIf,
},
jobs::{ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings},
runnable_settings::{
ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings,
},
scripts,
};
+80 -78
View File
@@ -36,10 +36,13 @@ use windmill_common::flow_conversations::add_message_to_conversation_tx;
use windmill_common::flow_status::{JobResult, RestartedFrom};
use windmill_common::jobs::{
check_tag_available_for_workspace_internal, format_completed_job_result, format_result,
ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, DynamicInput,
JobTriggerKind, RunInlinePreviewScriptFnParams, ENTRYPOINT_OVERRIDE,
DynamicInput, JobTriggerKind, RunInlinePreviewScriptFnParams, ENTRYPOINT_OVERRIDE,
};
use windmill_common::runnable_settings::{
ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, RunnableSettings,
};
use windmill_common::s3_helpers::{upload_artifact_to_store, BundleFormat};
use windmill_common::scripts::ScriptRunnableSettingsInline;
use windmill_common::triggers::TriggerMetadata;
use windmill_common::utils::{RunnableKind, WarnAfterExt};
use windmill_common::worker::{Connection, CLOUD_HOSTED, TMP_DIR};
@@ -891,7 +894,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, cache_ignore_s3_path, \
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, runnable_settings_handle, \
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",
)
@@ -3373,6 +3376,7 @@ pub struct UnifiedJob {
pub aggregate_wait_time_ms: Option<i64>,
pub preprocessed: Option<bool>,
pub worker: Option<String>,
pub runnable_settings_handle: Option<i64>,
}
const CJ_FIELDS: &[&str] = &[
@@ -3413,6 +3417,7 @@ const CJ_FIELDS: &[&str] = &[
"aggregate_wait_time_ms",
"v2_job.preprocessed",
"v2_job_completed.worker",
"null as runnable_settings_handle",
];
const QJ_FIELDS: &[&str] = &[
@@ -3453,6 +3458,7 @@ const QJ_FIELDS: &[&str] = &[
"aggregate_wait_time_ms",
"v2_job.preprocessed",
"v2_job_queue.worker",
"v2_job_queue.runnable_settings_handle",
];
impl UnifiedJob {
@@ -3517,14 +3523,13 @@ impl<'a> From<UnifiedJob> for Job {
created_by: uj.created_by,
created_at: uj.created_at,
started_at: uj.started_at,
scheduled_for: uj.scheduled_for.unwrap(),
running: uj.running.unwrap(),
script_hash: uj.script_hash,
script_path: uj.script_path,
script_entrypoint_override: None,
args: None,
running: uj.running.unwrap(),
scheduled_for: uj.scheduled_for.unwrap(),
logs: None,
flow_status: None,
workflow_as_code_status: None,
canceled: uj.canceled,
canceled_by: uj.canceled_by,
canceled_reason: None,
@@ -3532,9 +3537,10 @@ impl<'a> From<UnifiedJob> for Job {
job_kind: uj.job_kind,
schedule_path: uj.schedule_path,
permissioned_as: uj.permissioned_as,
flow_status: None,
workflow_as_code_status: None,
is_flow_step: uj.is_flow_step,
language: uj.language,
script_entrypoint_override: None,
same_worker: false,
pre_run_error: None,
email: uj.email,
@@ -3552,6 +3558,7 @@ impl<'a> From<UnifiedJob> for Job {
cache_ignore_s3_path: None,
priority: uj.priority,
preprocessed: uj.preprocessed,
runnable_settings_handle: uj.runnable_settings_handle,
},
)),
t => panic!("job type {} not valid", t),
@@ -4654,6 +4661,13 @@ pub async fn run_workflow_as_code(
let job = not_found_if_none(job, "Queued Job", &job_id.to_string())?;
let JobExtended { inner: job, raw_code, raw_lock, .. } = job;
let (_debouncing_settings, concurrency_settings) =
RunnableSettings::from_runnable_settings_handle(job.runnable_settings_handle, &db)
.await?
.prefetch_cached(&db)
.await?;
let (job_payload, tag, _delete_after_use, timeout, on_behalf_of) = match job.job_kind {
JobKind::Preview => (
JobPayload::Code(RawCode {
@@ -4662,13 +4676,15 @@ pub async fn run_workflow_as_code(
path: job.script_path,
language: job.language.unwrap_or_else(|| ScriptLang::Deno),
lock: raw_lock,
concurrency_settings: windmill_common::jobs::ConcurrencySettingsWithCustom {
custom_concurrency_key: windmill_queue::custom_concurrency_key(&db, &job.id)
.await
.map_err(to_anyhow)?,
concurrent_limit: job.concurrent_limit,
concurrency_time_window_s: job.concurrency_time_window_s,
},
concurrency_settings: concurrency_settings
.maybe_fallback(
windmill_queue::custom_concurrency_key(&db, &job.id)
.await
.map_err(to_anyhow)?,
job.concurrent_limit,
job.concurrency_time_window_s,
)
.into(),
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
dedicated_worker: None,
@@ -5489,11 +5505,6 @@ pub async fn run_wait_result_script_by_hash(
let ScriptHashInfo {
path,
tag,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
debounce_key,
debounce_delay_s,
mut cache_ttl,
mut cache_ignore_s3_path,
language,
@@ -5504,8 +5515,14 @@ pub async fn run_wait_result_script_by_hash(
has_preprocessor,
on_behalf_of_email,
created_by,
runnable_settings:
ScriptRunnableSettingsInline { concurrency_settings, debouncing_settings },
..
} = get_script_info_for_hash(Some(userdb_authed), &db, &w_id, hash).await?;
} = get_script_info_for_hash(Some(userdb_authed), &db, &w_id, hash)
.await?
.prefetch_cached(&db)
.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;
@@ -5539,17 +5556,8 @@ pub async fn run_wait_result_script_by_hash(
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path: path,
concurrency_settings: windmill_common::jobs::ConcurrencySettingsWithCustom {
custom_concurrency_key: concurrency_key,
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
}
.into(),
debouncing_settings: DebouncingSettings {
custom_key: debounce_key,
delay_s: debounce_delay_s,
..Default::default() // TODO
},
concurrency_settings,
debouncing_settings,
cache_ttl,
cache_ignore_s3_path,
language,
@@ -6492,9 +6500,15 @@ async fn add_batch_jobs(
job_kind,
language,
dedicated_worker,
custom_concurrency_key,
concurrent_limit,
concurrent_time_window_s,
ScriptRunnableSettingsInline {
concurrency_settings:
ConcurrencySettings {
concurrent_limit,
concurrency_time_window_s,
concurrency_key: custom_concurrency_key,
},
..
},
timeout,
raw_code,
raw_lock,
@@ -6507,23 +6521,22 @@ async fn add_batch_jobs(
UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let ScriptHashInfo {
hash: script_hash,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
language,
dedicated_worker,
timeout,
runnable_settings,
.. // TODO: consider on_behalf_of_email and created_by for batch jobs
} = get_latest_deployed_hash_for_path(Some(db_authed), db.clone(), &w_id, &path).await?;
} = get_latest_deployed_hash_for_path(Some(db_authed), db.clone(), &w_id, &path)
.await?
.prefetch_cached(&db)
.await?;
(
Some(script_hash),
Some(path),
JobKind::Script,
Some(language),
dedicated_worker,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
runnable_settings,
timeout,
None,
None,
@@ -6544,9 +6557,7 @@ async fn add_batch_jobs(
JobKind::Preview,
rawscript.language,
None,
None,
None,
None,
Default::default(),
None,
Some(rawscript.content),
rawscript.lock,
@@ -6591,19 +6602,20 @@ async fn add_batch_jobs(
add_virtual_items_if_necessary(&mut value.modules);
let flow_status = FlowStatus::new(&value);
(
None, // script_hash
path, // script_path
job_kind, // job_kind
None, // language
None, // dedicated_worker
value.concurrency_settings.concurrency_key.clone(), // custom_concurrency_key
value.concurrency_settings.concurrent_limit.clone(), // concurrent_limit
value.concurrency_settings.concurrency_time_window_s, // concurrency_time_window_s
None, // timeout
None, // raw_code
None, // raw_lock
Some(value), // raw_flow
Some(flow_status), // flow_status
None, // script_hash
path, // script_path
job_kind, // job_kind
None, // language
None, // dedicated_worker
ScriptRunnableSettingsInline {
concurrency_settings: value.concurrency_settings.clone(),
..Default::default()
},
None, // timeout
None, // raw_code
None, // raw_lock
Some(value), // raw_flow
Some(flow_status), // flow_status
)
}
"noop" => (
@@ -6612,9 +6624,7 @@ async fn add_batch_jobs(
JobKind::Noop,
None,
None,
None,
None,
None,
Default::default(),
None,
None,
None,
@@ -6669,7 +6679,7 @@ async fn add_batch_jobs(
username_to_permissioned_as(&authed.username),
authed.email,
concurrent_limit,
concurrent_time_window_s,
concurrency_time_window_s,
timeout,
n,
)
@@ -7063,11 +7073,8 @@ pub async fn run_job_by_hash_inner(
let ScriptHashInfo {
path,
tag,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
debounce_delay_s,
debounce_key,
runnable_settings:
ScriptRunnableSettingsInline { concurrency_settings, debouncing_settings },
mut cache_ttl,
mut cache_ignore_s3_path,
language,
@@ -7079,7 +7086,10 @@ pub async fn run_job_by_hash_inner(
created_by,
delete_after_use,
..
} = get_script_info_for_hash(Some(userdb_authed), &db, &w_id, hash).await?;
} = get_script_info_for_hash(Some(userdb_authed), &db, &w_id, hash)
.await?
.prefetch_cached(&db)
.await?;
check_scopes(&authed, || format!("jobs:run:scripts:{path}"))?;
if let Some(run_query_cache_ttl) = run_query.cache_ttl {
@@ -7115,16 +7125,8 @@ pub async fn run_job_by_hash_inner(
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path: path,
concurrency_settings: ConcurrencySettings {
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
},
debouncing_settings: DebouncingSettings {
custom_key: debounce_key,
delay_s: debounce_delay_s,
..Default::default()
},
concurrency_settings,
debouncing_settings,
cache_ttl,
cache_ignore_s3_path,
language,
+125 -43
View File
@@ -42,9 +42,12 @@ use windmill_worker::{process_relative_imports, scoped_dependency_map::ScopedDep
use windmill_common::{
assets::{clear_asset_usage, insert_asset_usage, AssetUsageKind, AssetWithAltAccessType},
error::to_anyhow,
error::{self, to_anyhow},
runnable_settings::{
min_version_supports_runnable_settings_v0, RunnableSettings, RunnableSettingsTrait,
},
s3_helpers::upload_artifact_to_store,
scripts::hash_script,
scripts::{hash_script, ScriptRunnableSettingsHandle, ScriptRunnableSettingsInline},
utils::{paginate_without_limits, WarnAfterExt},
worker::{CLOUD_HOSTED, MIN_VERSION_SUPPORTS_DEBOUNCING},
};
@@ -71,7 +74,7 @@ use windmill_queue::{schedule::push_scheduled_job, PushIsolationLevel};
const MAX_HASH_HISTORY_LENGTH_STORED: usize = 20;
#[derive(Serialize, sqlx::FromRow)]
pub struct ScriptWDraft {
pub struct ScriptWDraft<SR> {
pub hash: ScriptHash,
pub path: String,
pub summary: String,
@@ -88,10 +91,6 @@ pub struct ScriptWDraft {
#[serde(skip_serializing_if = "Option::is_none")]
pub envs: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ttl: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cache_ignore_s3_path: Option<bool>,
@@ -108,8 +107,6 @@ pub struct ScriptWDraft {
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub visible_to_runner_only: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub no_main_func: Option<bool>,
@@ -120,10 +117,64 @@ pub struct ScriptWDraft {
#[serde(skip_serializing_if = "Option::is_none")]
#[sqlx(json(nullable))]
pub assets: Option<Vec<AssetWithAltAccessType>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
#[serde(flatten)]
#[sqlx(flatten)]
pub runnable_settings: SR,
}
impl ScriptWDraft<ScriptRunnableSettingsHandle> {
pub async fn prefetch_cached<'a>(
self,
db: &DB,
) -> error::Result<ScriptWDraft<ScriptRunnableSettingsInline>> {
let (debouncing_settings, concurrency_settings) =
RunnableSettings::from_runnable_settings_handle(
self.runnable_settings.runnable_settings_handle,
db,
)
.await?
.prefetch_cached(db)
.await?;
Ok(ScriptWDraft {
runnable_settings: ScriptRunnableSettingsInline {
concurrency_settings: concurrency_settings.maybe_fallback(
self.runnable_settings.concurrency_key,
self.runnable_settings.concurrent_limit,
self.runnable_settings.concurrency_time_window_s,
),
debouncing_settings: debouncing_settings.maybe_fallback(
self.runnable_settings.debounce_key,
self.runnable_settings.debounce_delay_s,
),
},
hash: self.hash,
path: self.path,
summary: self.summary,
description: self.description,
content: self.content,
language: self.language,
kind: self.kind,
tag: self.tag,
draft: self.draft,
schema: self.schema,
draft_only: self.draft_only,
envs: self.envs,
cache_ttl: self.cache_ttl,
cache_ignore_s3_path: self.cache_ignore_s3_path,
dedicated_worker: self.dedicated_worker,
ws_error_handler_muted: self.ws_error_handler_muted,
priority: self.priority,
restart_unless_cancelled: self.restart_unless_cancelled,
delete_after_use: self.delete_after_use,
timeout: self.timeout,
visible_to_runner_only: self.visible_to_runner_only,
no_main_func: self.no_main_func,
has_preprocessor: self.has_preprocessor,
on_behalf_of_email: self.on_behalf_of_email,
assets: self.assets,
})
}
}
pub fn global_service() -> Router {
@@ -589,7 +640,7 @@ async fn create_script_internal<'c>(
.to_owned(),
));
};
let clashing_script = sqlx::query_as::<_, Script>(
let clashing_script = sqlx::query_as::<_, Script<ScriptRunnableSettingsHandle>>(
"SELECT * FROM script WHERE path = $1 AND archived = false AND workspace_id = $2",
)
.bind(&ns.path)
@@ -790,6 +841,31 @@ async fn create_script_internal<'c>(
}
};
let runnable_settings_handle = RunnableSettings {
debouncing_settings: ns.debouncing_settings.insert_cached(&db).await?,
concurrency_settings: ns.concurrency_settings.insert_cached(&db).await?,
}
.insert_cached(&db)
.await?;
let (
guarded_concurrent_limit,
guarded_concurrency_time_window_s,
guarded_concurrency_key,
guarded_debounce_key,
guarded_debounce_delay_s,
) = if min_version_supports_runnable_settings_v0().await {
Default::default()
} else {
(
ns.concurrency_settings.concurrent_limit.clone(),
ns.concurrency_settings.concurrency_time_window_s.clone(),
ns.concurrency_settings.concurrency_key.clone(),
ns.debouncing_settings.debounce_key.clone(),
ns.debouncing_settings.debounce_delay_s.clone(),
)
};
// Row lock debounce key for path. We need this to make all updates of runnables sequential and predictable.
tokio::time::timeout(
core::time::Duration::from_secs(60),
@@ -803,8 +879,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, 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)",
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, runnable_settings_handle) \
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, $38)",
&w_id,
&hash.0,
ns.path,
@@ -822,8 +898,8 @@ async fn create_script_internal<'c>(
ns.tag,
ns.draft_only,
envs,
ns.concurrent_limit,
ns.concurrency_time_window_s,
guarded_concurrent_limit,
guarded_concurrency_time_window_s,
ns.cache_ttl,
ns.dedicated_worker,
ns.ws_error_handler_muted.unwrap_or(false),
@@ -831,7 +907,7 @@ async fn create_script_internal<'c>(
ns.restart_unless_cancelled,
ns.delete_after_use,
ns.timeout,
ns.concurrency_key,
guarded_concurrency_key,
ns.visible_to_runner_only,
no_main_func.filter(|x| *x), // should be Some(true) or None
codebase,
@@ -843,9 +919,10 @@ async fn create_script_internal<'c>(
},
validate_schema,
ns.assets.as_ref().and_then(|a| serde_json::to_value(a).ok()),
ns.debounce_key,
ns.debounce_delay_s,
guarded_debounce_key,
guarded_debounce_delay_s,
ns.cache_ignore_s3_path,
runnable_settings_handle
)
.execute(&mut *tx)
.await?;
@@ -1173,18 +1250,20 @@ pub async fn pick_hub_script_by_path(
Ok::<_, Error>((status_code, headers, response))
}
#[axum::debug_handler]
async fn get_script_by_path(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
Query(query): Query<WithStarredInfoQuery>,
) -> JsonResult<ScriptWithStarred> {
) -> JsonResult<ScriptWithStarred<ScriptRunnableSettingsInline>> {
let path = path.to_path();
check_scopes(&authed, || format!("scripts:read:{}", path))?;
let mut tx = user_db.begin(&authed).await?;
let script_o = if query.with_starred_info.unwrap_or(false) {
sqlx::query_as::<_, ScriptWithStarred>(
sqlx::query_as::<_, ScriptWithStarred<ScriptRunnableSettingsHandle>>(
"SELECT s.*, favorite.path IS NOT NULL as starred
FROM script s
LEFT JOIN favorite
@@ -1202,7 +1281,7 @@ async fn get_script_by_path(
.fetch_optional(&mut *tx)
.await?
} else {
sqlx::query_as::<_, ScriptWithStarred>(
sqlx::query_as::<_, ScriptWithStarred<ScriptRunnableSettingsHandle>>(
"SELECT *, NULL as starred FROM script WHERE path = $1 AND workspace_id = $2 ORDER BY created_at DESC LIMIT 1",
)
.bind(path)
@@ -1212,7 +1291,10 @@ async fn get_script_by_path(
};
tx.commit().await?;
let script = not_found_if_none(script_o, "Script", path)?;
let script = not_found_if_none(script_o, "Script", path)?
.prefetch_cached(&db)
.await?;
Ok(Json(script))
}
@@ -1234,15 +1316,16 @@ async fn get_triggers_count(
async fn get_script_by_path_w_draft(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<ScriptWDraft> {
) -> JsonResult<ScriptWDraft<ScriptRunnableSettingsInline>> {
let path = path.to_path();
check_scopes(&authed, || format!("scripts:read:{}", path))?;
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, 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
let script_o = sqlx::query_as::<_, ScriptWDraft<ScriptRunnableSettingsHandle>>(
"SELECT hash, script.path, summary, description, content, language, kind, tag, schema, draft_only, envs, runnable_settings_handle, 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",
@@ -1254,7 +1337,7 @@ async fn get_script_by_path_w_draft(
tx.commit().await?;
let script = not_found_if_none(script_o, "Script", path)?;
Ok(Json(script))
Ok(Json(script.prefetch_cached(&db).await?))
}
async fn get_script_history(
@@ -1686,9 +1769,9 @@ async fn get_script_by_hash_internal<'c>(
workspace_id: &str,
hash: &ScriptHash,
with_starred_info_for_username: Option<&str>,
) -> Result<ScriptWithStarred> {
) -> Result<ScriptWithStarred<ScriptRunnableSettingsHandle>> {
let script_o = if let Some(username) = with_starred_info_for_username {
sqlx::query_as::<_, ScriptWithStarred>(
sqlx::query_as::<_, ScriptWithStarred<ScriptRunnableSettingsHandle>>(
"SELECT s.*, favorite.path IS NOT NULL as starred
FROM script s
LEFT JOIN favorite
@@ -1704,7 +1787,7 @@ async fn get_script_by_hash_internal<'c>(
.fetch_optional(&mut **db)
.await?
} else {
sqlx::query_as::<_, ScriptWithStarred>(
sqlx::query_as::<_, ScriptWithStarred<ScriptRunnableSettingsHandle>>(
"SELECT *, NULL as starred FROM script WHERE hash = $1 AND workspace_id = $2",
)
.bind(hash)
@@ -1728,7 +1811,7 @@ async fn get_script_by_hash(
Query(query): Query<WithStarredInfoQuery>,
Query(query_auth): Query<GetScriptByHashQuery>,
Extension(authed): Extension<ApiAuthed>,
) -> JsonResult<ScriptWithStarred> {
) -> JsonResult<ScriptWithStarred<ScriptRunnableSettingsInline>> {
let mut tx = if query_auth.authed.is_some_and(|x| x) {
user_db.begin(&authed).await?
} else {
@@ -1752,7 +1835,7 @@ async fn get_script_by_hash(
tx.commit().await?;
Ok(Json(r))
Ok(Json(r.prefetch_cached(&db).await?))
}
async fn raw_script_by_hash(
@@ -1887,12 +1970,13 @@ async fn archive_script_by_path(
async fn archive_script_by_hash(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Extension(webhook): Extension<WebhookShared>,
Path((w_id, hash)): Path<(String, ScriptHash)>,
) -> JsonResult<Script> {
) -> JsonResult<Script<ScriptRunnableSettingsInline>> {
let mut tx = user_db.begin(&authed).await?;
let script = sqlx::query_as::<_, Script>(
let script = sqlx::query_as::<_, Script<ScriptRunnableSettingsHandle>>(
"UPDATE script SET archived = true WHERE hash = $1 AND workspace_id = $2 RETURNING *",
)
.bind(&hash.0)
@@ -1931,7 +2015,7 @@ async fn archive_script_by_hash(
WebhookMessage::DeleteScript { workspace: w_id, hash: hash.to_string() },
);
Ok(Json(script))
Ok(Json(script.prefetch_cached(&db).await?))
}
async fn delete_script_by_hash(
@@ -1940,11 +2024,11 @@ async fn delete_script_by_hash(
Extension(webhook): Extension<WebhookShared>,
Extension(db): Extension<DB>,
Path((w_id, hash)): Path<(String, ScriptHash)>,
) -> JsonResult<Script> {
) -> JsonResult<Script<ScriptRunnableSettingsInline>> {
let mut tx = user_db.begin(&authed).await?;
require_admin(authed.is_admin, &authed.username)?;
let script = sqlx::query_as::<_, Script>(
let script = sqlx::query_as::<_, Script<ScriptRunnableSettingsHandle>>(
"UPDATE script SET content = '', archived = true, deleted = true, lock = '', schema = null WHERE hash = $1 AND \
workspace_id = $2 RETURNING *",
)
@@ -1980,7 +2064,7 @@ async fn delete_script_by_hash(
WebhookMessage::DeleteScript { workspace: w_id, hash: hash.to_string() },
);
Ok(Json(script))
Ok(Json(script.prefetch_cached(&db).await?))
}
#[derive(Deserialize)]
@@ -2222,9 +2306,7 @@ async fn delete_scripts_bulk(
/// Validates that script debouncing configuration is supported by all workers
/// Returns an error if debouncing is configured but workers are behind required version
async fn guard_script_from_debounce_data(ns: &NewScript) -> Result<()> {
if !*MIN_VERSION_SUPPORTS_DEBOUNCING.read().await
&& (ns.debounce_key.is_some() || ns.debounce_delay_s.is_some())
{
if !*MIN_VERSION_SUPPORTS_DEBOUNCING.read().await && !ns.debouncing_settings.is_default() {
tracing::warn!(
"Script debouncing configuration rejected: workers are behind minimum required version for debouncing feature"
);
+10 -16
View File
@@ -44,6 +44,8 @@ use axum::{
use http::HeaderName;
use itertools::Itertools;
use windmill_common::runnable_settings::{ConcurrencySettings, DebouncingSettings};
use windmill_common::scripts::ScriptRunnableSettingsHandle;
use windmill_common::utils::require_admin;
use windmill_common::variables::decrypt;
use windmill_common::{
@@ -73,10 +75,6 @@ struct ScriptMetadata {
#[serde(skip_serializing_if = "Option::is_none")]
envs: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
cache_ttl: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
dedicated_worker: Option<bool>,
@@ -99,15 +97,13 @@ struct ScriptMetadata {
#[serde(skip_serializing_if = "Option::is_none")]
pub codebase: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub has_preprocessor: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub on_behalf_of_email: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
#[serde(flatten)]
pub concurrency_settings: ConcurrencySettings,
#[serde(flatten)]
pub debouncing_settings: DebouncingSettings,
}
pub fn is_none_or_false(val: &Option<bool>) -> bool {
@@ -368,7 +364,7 @@ pub(crate) async fn tarball_workspace(
}
{
let scripts = sqlx::query_as::<_, Script>(
let scripts = sqlx::query_as::<_, Script<ScriptRunnableSettingsHandle>>(
"SELECT * FROM script as o WHERE workspace_id = $1 AND archived = false
AND (draft_only IS NULL OR draft_only = false)
AND created_at = (select max(created_at) from script where path = o.path AND \
@@ -379,6 +375,7 @@ pub(crate) async fn tarball_workspace(
.await?;
for script in scripts {
let script = script.prefetch_cached(&db).await?;
let ext = match script.language {
ScriptLang::Python3 => "py",
ScriptLang::Deno => {
@@ -427,8 +424,8 @@ pub(crate) async fn tarball_workspace(
kind: script.kind.to_string(),
lock: script.lock,
envs: script.envs,
concurrent_limit: script.concurrent_limit,
concurrency_time_window_s: script.concurrency_time_window_s,
concurrency_settings: script.runnable_settings.concurrency_settings,
debouncing_settings: script.runnable_settings.debouncing_settings,
cache_ttl: script.cache_ttl,
dedicated_worker: script.dedicated_worker,
ws_error_handler_muted: script.ws_error_handler_muted,
@@ -440,11 +437,8 @@ pub(crate) async fn tarball_workspace(
visible_to_runner_only: script.visible_to_runner_only,
no_main_func: script.no_main_func,
codebase: script.codebase,
concurrency_key: script.concurrency_key,
has_preprocessor: script.has_preprocessor,
on_behalf_of_email: script.on_behalf_of_email,
debounce_key: script.debounce_key,
debounce_delay_s: script.debounce_delay_s,
};
let metadata_str = serde_json::to_string_pretty(&metadata).unwrap();
archive
+29 -27
View File
@@ -41,33 +41,35 @@ pub fn build_agent_http_client(
base_internal_url: Option<String>,
) -> HttpClient {
let client = ClientBuilder::new(
configure_client(reqwest::Client::builder()
.pool_max_idle_per_host(10)
.pool_idle_timeout(Duration::from_secs(60))
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_secs(30)))
.default_headers({
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
"User-Agent", // Replace with your desired header name
"Windmill-Agent/1.0".parse().unwrap(), // Replace with your desired header value
);
let token = format!(
"{}{}_{}",
AGENT_JWT_PREFIX,
worker_suffix,
agent_token
.unwrap_or(AGENT_TOKEN.clone())
.trim_start_matches(AGENT_JWT_PREFIX)
);
headers.insert(
"Authorization",
format!("Bearer {}", token).parse().unwrap(),
);
headers
})
.build()
.expect("Failed to create HTTP client"),
configure_client(
reqwest::Client::builder()
.pool_max_idle_per_host(10)
.pool_idle_timeout(Duration::from_secs(60))
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_secs(30)),
)
.default_headers({
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
"User-Agent", // Replace with your desired header name
"Windmill-Agent/1.0".parse().unwrap(), // Replace with your desired header value
);
let token = format!(
"{}{}_{}",
AGENT_JWT_PREFIX,
worker_suffix,
agent_token
.unwrap_or(AGENT_TOKEN.clone())
.trim_start_matches(AGENT_JWT_PREFIX)
);
headers.insert(
"Authorization",
format!("Bearer {}", token).parse().unwrap(),
);
headers
})
.build()
.expect("Failed to create HTTP client"),
)
.with(RetryTransientMiddleware::new_with_policy(
ExponentialBackoff::builder().build_with_max_retries(5),
+24 -1
View File
@@ -1,4 +1,4 @@
use sqlx::{Acquire, Pool, Postgres, Transaction};
use sqlx::{Acquire, PgConnection, PgExecutor, Pool, Postgres, Transaction};
use crate::audit::AuditAuthor;
@@ -239,3 +239,26 @@ impl UserDB {
Ok(tx)
}
}
pub trait DbExecutor<'b>: Send + Sized + PgExecutor<'b> {
fn executor<'a>(&'a mut self) -> impl PgExecutor<'a>;
fn populate<'a>(&'a mut self) -> impl DbExecutor<'a>;
}
impl<'b> DbExecutor<'b> for &DB {
fn executor<'a>(&'a mut self) -> impl PgExecutor<'a> {
&**self
}
fn populate<'a>(&'a mut self) -> impl DbExecutor<'a> {
&**self
}
}
impl<'b> DbExecutor<'b> for &'b mut PgConnection {
fn executor<'a>(&'a mut self) -> impl PgExecutor<'a> {
&mut **self
}
fn populate<'a>(&'a mut self) -> impl DbExecutor<'a> {
&mut **self
}
}
+1 -1
View File
@@ -11,8 +11,8 @@
//! For users writing scripts to access their infrastructure with firewalls requiring incoming
//! connections to be from whitelisted IP addresses.
use std::time::Duration;
use crate::utils::configure_client;
use std::time::Duration;
pub async fn get_ip() -> anyhow::Result<String> {
tokio::select! {
+1 -1
View File
@@ -24,8 +24,8 @@ use crate::{
cache,
db::DB,
error::{Error, Result as WindmillResult},
jobs::{ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings},
more_serde::{default_empty_string, default_id, default_null, default_true, is_default},
runnable_settings::{ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings},
scripts::{Schema, ScriptHash, ScriptLang},
worker::{to_raw_value, Connection},
};
+7 -3
View File
@@ -1,16 +1,18 @@
#[cfg(feature = "private")]
#[allow(unused)]
pub use crate::git_sync_ee::*;
use url::Url;
#[cfg(not(feature = "private"))]
use sqlx::{Pool, Postgres};
use url::Url;
#[cfg(not(feature = "private"))]
pub async fn get_github_app_token_internal(
_db: &Pool<Postgres>,
_job_token: &str,
) -> crate::error::Result<String> {
return Err(crate::error::Error::BadRequest("Github app authentication is not available on the open source build".to_string()))
return Err(crate::error::Error::BadRequest(
"Github app authentication is not available on the open source build".to_string(),
));
}
pub fn prepend_token_to_github_url(
@@ -20,7 +22,9 @@ pub fn prepend_token_to_github_url(
let url = Url::parse(github_url)?;
if url.host_str() != Some("github.com") {
return Err(crate::error::Error::BadRequest("Invalid: not a github URL".to_string()));
return Err(crate::error::Error::BadRequest(
"Invalid: not a github URL".to_string(),
));
}
Ok(format!(
+15 -119
View File
@@ -24,6 +24,7 @@ use crate::{
flow_status::{FlowStatus, RestartedFrom},
flows::{FlowNodeId, FlowValue, Retry},
get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path,
runnable_settings::{ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings},
scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang},
users::username_to_permissioned_as,
utils::{StripPath, HTTP_CLIENT},
@@ -202,6 +203,8 @@ pub struct QueuedJob {
pub priority: Option<i16>,
#[serde(skip_serializing_if = "Option::is_none")]
pub preprocessed: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub runnable_settings_handle: Option<i64>,
}
impl QueuedJob {
@@ -275,6 +278,7 @@ impl Default for QueuedJob {
cache_ignore_s3_path: None,
priority: None,
preprocessed: None,
runnable_settings_handle: None,
}
}
}
@@ -483,108 +487,6 @@ pub enum JobPayload {
},
}
// TODO: Add validation logic.
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
pub struct DebouncingSettings {
#[serde(
skip_serializing_if = "Option::is_none",
rename = "debounce_key",
alias = "custom_debounce_key"
)]
/// debounce key is usually stored in the db
/// including when:
///
/// 1. User have created custom debounce key from ui or cli
/// 2. User used default one
///
/// in either cases this argument serves as reactive way of overwriting debounce key from the backend.
/// Default: hash(path + step_id + inputs)
pub custom_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none", rename = "debounce_delay_s")]
/// Debouncing delay will be determined by the first job with the key.
/// All subsequent jobs with Some will get debounced.
/// If the job has no delay, it will execute immediately, fully ignoring pending delays.
pub delay_s: Option<i32>,
#[serde(
skip_serializing_if = "Option::is_none",
rename = "max_total_debouncing_time"
)]
pub max_total_time: Option<i32>,
#[serde(
skip_serializing_if = "Option::is_none",
rename = "max_total_debounces_amount"
)]
pub max_total_amount: Option<i32>,
#[serde(
skip_serializing_if = "Option::is_none",
rename = "debounce_args_to_accumulate"
)]
/// top level arguments to preserve
/// For every debounce selected arguments will be saved
/// in the end (when job finally starts) arguments will be appended and passed to runnable
///
/// NOTE: selected args should be the lists.
pub args_to_accumulate: Option<Vec<String>>,
}
impl DebouncingSettings {
pub fn is_default(&self) -> bool {
self == &Self::default()
}
}
#[derive(Debug, Default, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct ConcurrencySettings {
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
}
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow, Default)]
pub struct ConcurrencySettingsWithCustom {
#[serde(skip_serializing_if = "Option::is_none")]
pub custom_concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
}
impl From<ConcurrencySettings> for ConcurrencySettingsWithCustom {
fn from(
ConcurrencySettings { concurrency_key, concurrent_limit, concurrency_time_window_s }: ConcurrencySettings,
) -> Self {
ConcurrencySettingsWithCustom {
custom_concurrency_key: concurrency_key,
concurrency_time_window_s,
concurrent_limit,
}
}
}
impl From<ConcurrencySettingsWithCustom> for ConcurrencySettings {
fn from(
ConcurrencySettingsWithCustom {
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
}: ConcurrencySettingsWithCustom,
) -> Self {
ConcurrencySettings {
concurrency_key: custom_concurrency_key,
concurrency_time_window_s,
concurrent_limit,
}
}
}
#[derive(Clone, Serialize, Deserialize, Debug)]
pub struct SkipHandler {
pub path: String,
@@ -703,11 +605,11 @@ pub async fn script_path_to_payload<'e>(
let ScriptHashInfo {
hash,
tag,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
debounce_key,
debounce_delay_s,
runnable_settings:
super::scripts::ScriptRunnableSettingsInline {
concurrency_settings,
debouncing_settings,
},
cache_ttl,
cache_ignore_s3_path,
language,
@@ -719,7 +621,10 @@ pub async fn script_path_to_payload<'e>(
on_behalf_of_email,
created_by,
..
} = get_latest_deployed_hash_for_path(db_authed, db, w_id, script_path).await?;
} = get_latest_deployed_hash_for_path(db_authed, db.clone(), w_id, script_path)
.await?
.prefetch_cached(&db)
.await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf {
@@ -741,17 +646,8 @@ pub async fn script_path_to_payload<'e>(
priority,
apply_preprocessor: !skip_preprocessor.unwrap_or(false)
&& has_preprocessor.unwrap_or(false),
concurrency_settings: ConcurrencySettingsWithCustom {
custom_concurrency_key: concurrency_key,
concurrent_limit,
concurrency_time_window_s,
}
.into(),
debouncing_settings: DebouncingSettings {
custom_key: debounce_key,
delay_s: debounce_delay_s,
..Default::default()
},
debouncing_settings,
concurrency_settings,
},
tag,
delete_after_use,
+102 -39
View File
@@ -75,6 +75,7 @@ pub mod otel_ee;
pub mod otel_oss;
pub mod queue;
pub mod result_stream;
pub mod runnable_settings;
pub mod s3_helpers;
pub mod schedule;
pub mod schema;
@@ -163,7 +164,7 @@ lazy_static::lazy_static! {
pub static ref DEPLOYED_SCRIPT_HASH_CACHE: Cache<(String, String), ExpiringLatestVersionId> = Cache::new(1000);
pub static ref FLOW_VERSION_CACHE: Cache<(String, String), ExpiringLatestVersionId> = Cache::new(1000);
pub static ref DYNAMIC_INPUT_CACHE: Cache<String, Arc<jobs::DynamicInput>> = Cache::new(1000);
pub static ref DEPLOYED_SCRIPT_INFO_CACHE: Cache<(String, i64), ScriptHashInfo> = Cache::new(1000);
pub static ref DEPLOYED_SCRIPT_INFO_CACHE: Cache<(String, i64), ScriptHashInfo<ScriptRunnableSettingsHandle>> = Cache::new(1000);
pub static ref FLOW_INFO_CACHE: Cache<(String, i64), FlowVersionInfo> = Cache::new(1000);
pub static ref QUIET_LOGS: bool = std::env::var("QUIET_LOGS").map(|s| s.parse::<bool>().unwrap_or(false)).unwrap_or(false);
@@ -344,7 +345,9 @@ pub fn parse_postgres_url(url: &str) -> Result<PostgresUrlComponents, Error> {
let scheme = parsed_url.scheme().to_string();
let username = parsed_url.username().to_string();
let username = urlencoding::decode(&username).map_err(to_anyhow)?.to_string();
let username = urlencoding::decode(&username)
.map_err(to_anyhow)?
.to_string();
let password = parsed_url.password().map(|p| p.to_string());
let password = match password {
Some(p) => Some(urlencoding::decode(&p).map_err(to_anyhow)?.to_string()),
@@ -404,12 +407,9 @@ impl DatabaseUrl {
DatabaseUrl::Static(_) => Ok(()),
}
}
}
static DATABASE_URL_CACHE: tokio::sync::OnceCell<DatabaseUrl> =
tokio::sync::OnceCell::const_new();
static DATABASE_URL_CACHE: tokio::sync::OnceCell<DatabaseUrl> = tokio::sync::OnceCell::const_new();
pub async fn get_database_url() -> Result<DatabaseUrl, Error> {
let database_url = DATABASE_URL_CACHE
@@ -442,22 +442,27 @@ pub async fn get_database_url() -> Result<DatabaseUrl, Error> {
})?;
tracing::info!("iamrds mode detected, generating IAM RDS URL for region: {region}");
#[cfg(all(feature = "enterprise", feature = "private"))]
#[cfg(all(feature = "enterprise", feature = "private"))]
{
let rds_url = db_iam_ee::generate_database_url(&url, &region)
.await
.map_err(|e| {
Error::InternalErr(format!("Failed to generate IAM database URL: {}", e))
})?;
tracing::info!("IAM RDS URL generated successfully");
Ok::<DatabaseUrl, Error>(DatabaseUrl::IamRds(
std::sync::Arc::new(tokio::sync::RwLock::new(rds_url))
))
let rds_url = db_iam_ee::generate_database_url(&url, &region)
.await
.map_err(|e| {
Error::InternalErr(format!(
"Failed to generate IAM database URL: {}",
e
))
})?;
tracing::info!("IAM RDS URL generated successfully");
Ok::<DatabaseUrl, Error>(DatabaseUrl::IamRds(std::sync::Arc::new(
tokio::sync::RwLock::new(rds_url),
)))
}
#[cfg(not(all(feature = "enterprise", feature = "private")))]
{
return Err(Error::BadConfig("IAM RDS authentication is not enabled in OSS mode".to_string()));
return Err(Error::BadConfig(
"IAM RDS authentication is not enabled in OSS mode".to_string(),
));
}
} else {
Ok::<DatabaseUrl, Error>(DatabaseUrl::Static(url.to_string()))
@@ -503,8 +508,7 @@ pub async fn connect_db(
server_mode: bool,
indexer_mode: bool,
worker_mode: bool,
#[cfg(feature = "private")]
mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
#[cfg(feature = "private")] mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> anyhow::Result<sqlx::Pool<sqlx::Postgres>> {
use anyhow::Context;
@@ -529,7 +533,6 @@ pub async fn connect_db(
}
};
let pool = connect(database_url.clone(), max_connections, worker_mode).await?;
#[cfg(all(feature = "enterprise", feature = "private"))]
let pool2 = pool.clone();
@@ -571,7 +574,6 @@ pub async fn connect_db(
}
}
}
}
});
}
@@ -629,7 +631,8 @@ pub async fn connect(
}
})
.connect_with(
sqlx::postgres::PgConnectOptions::from_str(&database_url.as_str().await)?.statement_cache_capacity(400),
sqlx::postgres::PgConnectOptions::from_str(&database_url.as_str().await)?
.statement_cache_capacity(400),
)
.await
.map_err(|err| Error::ConnectingToDatabase(err.to_string()))
@@ -640,7 +643,7 @@ type Tag = String;
pub use db::DB;
use crate::{
auth::{FLOW_PERMS_CACHE, HASH_PERMS_CACHE, PermsCache}, db::{AuthedRef, UserDbWithAuthed}, error::to_anyhow, scripts::ScriptHash
auth::{PermsCache, FLOW_PERMS_CACHE, HASH_PERMS_CACHE}, db::{AuthedRef, UserDbWithAuthed}, error::to_anyhow, runnable_settings::RunnableSettings, scripts::{ScriptHash, ScriptRunnableSettingsHandle, ScriptRunnableSettingsInline}
};
#[derive(Clone)]
@@ -649,16 +652,11 @@ pub struct ExpiringLatestVersionId {
expires_at: std::time::Instant,
}
#[derive(Clone, Debug)]
pub struct ScriptHashInfo {
#[derive(Clone, Debug, sqlx::FromRow)]
pub struct ScriptHashInfo<SR> {
pub path: String,
pub hash: i64,
pub tag: Option<String>,
pub concurrency_key: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
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,
@@ -669,6 +667,50 @@ pub struct ScriptHashInfo {
pub has_preprocessor: Option<bool>,
pub on_behalf_of_email: Option<String>,
pub created_by: String,
#[sqlx(flatten)]
pub runnable_settings: SR,
}
impl ScriptHashInfo<ScriptRunnableSettingsHandle> {
pub async fn prefetch_cached<'a>(self, db: &DB) -> error::Result<ScriptHashInfo<ScriptRunnableSettingsInline>> {
let (debouncing_settings, concurrency_settings) =
RunnableSettings::from_runnable_settings_handle(
self.runnable_settings.runnable_settings_handle,
db,
)
.await?
.prefetch_cached(db)
.await?;
Ok(
ScriptHashInfo {
path: self.path,
hash: self.hash,
tag: self.tag,
cache_ttl: self.cache_ttl,
cache_ignore_s3_path: self.cache_ignore_s3_path,
language: self.language,
dedicated_worker: self.dedicated_worker,
priority: self.priority,
delete_after_use: self.delete_after_use,
timeout: self.timeout,
has_preprocessor: self.has_preprocessor,
on_behalf_of_email: self.on_behalf_of_email,
created_by: self.created_by,
runnable_settings: ScriptRunnableSettingsInline{
concurrency_settings: concurrency_settings
.maybe_fallback(
self.runnable_settings.concurrency_key,
self.runnable_settings.concurrent_limit,
self.runnable_settings.concurrency_time_window_s,
),
debouncing_settings: debouncing_settings
.maybe_fallback(self.runnable_settings.debounce_key, self.runnable_settings.debounce_delay_s),
},
})
}
}
pub fn get_latest_deployed_hash_for_path<'e>(
@@ -676,7 +718,7 @@ pub fn get_latest_deployed_hash_for_path<'e>(
db2: DB,
w_id: &'e str,
script_path: &'e str,
) -> impl Future<Output = error::Result<ScriptHashInfo>> + Send + 'e {
) -> impl Future<Output = error::Result<ScriptHashInfo<ScriptRunnableSettingsHandle>>> + Send + 'e {
async move {
let cache_key = (w_id.to_string(), script_path.to_string());
let mut computed_hash = None;
@@ -759,7 +801,7 @@ pub async fn get_script_info_for_hash<'e, E: sqlx::PgExecutor<'e>>(
db: E,
w_id: &str,
hash: i64,
) -> error::Result<ScriptHashInfo> {
) -> error::Result<ScriptHashInfo<ScriptRunnableSettingsHandle>> {
let key = (w_id.to_string(), hash);
let mut computed_hash = None;
@@ -803,13 +845,32 @@ async fn get_script_info_for_hash_inner<'e, E: sqlx::PgExecutor<'e>>(
db: E,
w_id: &str,
hash: i64,
) -> 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, 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
) -> error::Result<Option<ScriptHashInfo<ScriptRunnableSettingsHandle>>> {
let r = sqlx::query_as::<_, ScriptHashInfo<ScriptRunnableSettingsHandle>>(
"SELECT
hash,
tag,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
debounce_key,
debounce_delay_s,
runnable_settings_handle,
cache_ttl,
cache_ignore_s3_path,
language,
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",
)
.bind(hash)
.bind(w_id)
.fetch_optional(db)
.await?;
Ok(r)
@@ -1031,9 +1092,10 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>(
Option<i32>,
Option<String>,
String,
Option<i64>,
)> {
let r_o = sqlx::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
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, debounce_key, debounce_delay_s, cache_ttl, cache_ignore_s3_path, runnable_settings_handle, 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,
@@ -1061,6 +1123,7 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>(
script.timeout,
script.on_behalf_of_email,
script.created_by,
script.runnable_settings_handle,
))
}
@@ -0,0 +1,149 @@
mod settings;
use itertools::Itertools;
pub use settings::*;
use std::{
future::Future,
hash::{DefaultHasher, Hash, Hasher},
};
use serde::{de::DeserializeOwned, Serialize};
use sqlx::{postgres::PgRow, FromRow, Pool, Postgres};
use crate::{error, make_static};
lazy_static::lazy_static! {
static ref WMDEBUG_FORCE_RUNNABLE_SETTINGS_V0: bool = std::env::var("WMDEBUG_FORCE_RUNNABLE_SETTINGS_V0").is_ok();
pub static ref MIN_VERSION_RUNNABLE_SETTINGS_V0: semver::Version = semver::Version::new(1, 592, 0);
}
pub async fn min_version_supports_runnable_settings_v0() -> bool {
// Check if workers support workspace dependencies feature
if !*WMDEBUG_FORCE_RUNNABLE_SETTINGS_V0
&& !*crate::worker::MIN_VERSION_SUPPORTS_RUNNABLE_SETTINGS_V0
.read()
.await
{
tracing::debug!(
"Internal: min version does not support runnable settings v0, falling back to old system",
);
false
} else {
true
}
}
make_static! {
static ref RUNNABLE_INDIVIDUAL_SETTINGS: { i64 => serde_json::Value } in "runnable_individual_settings" <= 1000;
static ref RUNNABLE_SETTINGS_REFERENCES: { i64 => RunnableSettings } in "runnable_settings_references" <= 1000;
}
pub trait RunnableSettingsTrait:
PartialEq
+ Default
+ Hash
+ for<'r> FromRow<'r, PgRow>
+ Serialize
+ DeserializeOwned
+ private_mod::RunnableSettingsTraitInternal
+ Send
+ Unpin
{
fn is_default(&self) -> bool {
self == &Self::default()
}
/// get [[Self]] from cache or fetch from db
/// if not found, returns Error
fn get<'a>(
hash: i64,
db: &'a Pool<Postgres>,
) -> impl Future<Output = Result<Self, error::Error>>
where
Self: 'a,
{
async move {
let v = RUNNABLE_INDIVIDUAL_SETTINGS
.get_or_insert_async(hash, async {
let r = sqlx::query_as::<Postgres, Self>(&format!(
"SELECT {} FROM {} WHERE hash = $1",
Self::INCLUDE_FIELDS.iter().join(","),
Self::SETTINGS_NAME,
))
.bind(hash)
.fetch_one(db)
.await?;
serde_json::to_value(&r).map_err(error::Error::from)
})
.await?;
// TODO: Less parsing
serde_json::from_value(v).map_err(error::Error::from)
}
}
/// set [[Self]] in db
/// returns optional `hash` that `MUST` be
fn insert_cached<'a>(
&self,
db: &'a Pool<Postgres>,
) -> impl Future<Output = error::Result<Option<i64>>> {
async move {
// If it is default/empty, we optimize by telling "there is no hash for this"
if self.is_default() {
return Ok(None);
}
let hash = {
let mut h = DefaultHasher::new();
self.hash(&mut h);
// Since we use single fs cache, we hash the name as well to prevent from collisions.
Self::SETTINGS_NAME.hash(&mut h);
h.finish() as i64
};
// If already exists in cache, then it is already inserted, we will just ignore the result.
RUNNABLE_INDIVIDUAL_SETTINGS
.get_or_insert_async(hash, async move {
// If it is not in cache
// this means we don't know if it is in db
// in that case, we INSERT to be sure
let sql = format!(
"INSERT INTO {} (hash, {})
VALUES ($1, {})
ON CONFLICT (hash) DO NOTHING",
Self::SETTINGS_NAME,
Self::INCLUDE_FIELDS.iter().join(","),
(1..=Self::INCLUDE_FIELDS.len())
.into_iter()
.map(|i| format!("${}", i + 1))
.join(",")
);
self.bind_arguments(sqlx::query(&sql).bind(hash) /* First bind hash */) /* Then bind user settings */
.execute(db)
.await?;
// Cache existing value for this worker
serde_json::to_value(&self).map_err(error::Error::from)
})
.await?;
Ok(Some(hash))
}
}
}
mod private_mod {
use sqlx::{query::Query, Database, Postgres};
pub type Q<'a> = Query<'a, Postgres, <Postgres as Database>::Arguments<'a>>;
pub trait RunnableSettingsTraitInternal {
const SETTINGS_NAME: &'static str;
const INCLUDE_FIELDS: &'static [&'static str];
fn bind_arguments<'a>(&'a self, q: Q<'a>) -> Q<'a>;
}
}
@@ -0,0 +1,259 @@
use std::{
future::Future,
hash::{Hash, Hasher},
};
use serde::{Deserialize, Serialize};
use sqlx::{Pool, Postgres};
use crate::{
error,
runnable_settings::{
min_version_supports_runnable_settings_v0, private_mod::Q, RunnableSettingsTrait,
},
DB,
};
#[derive(Deserialize, Clone, Copy, Serialize, Default, Hash)]
pub struct RunnableSettings {
pub debouncing_settings: Option<i64>,
pub concurrency_settings: Option<i64>,
}
impl RunnableSettings {
pub async fn prefetch_cached<'a>(
&self,
db: &DB,
) -> error::Result<(DebouncingSettings, ConcurrencySettings)> {
Ok((
if let Some(hash) = self.debouncing_settings {
DebouncingSettings::get(hash, db).await?
} else {
Default::default()
},
if let Some(hash) = self.concurrency_settings {
ConcurrencySettings::get(hash, db).await?
} else {
Default::default()
},
))
}
pub async fn prefetch_cached_from_handle<'a>(
hash: Option<i64>,
db: &'a DB,
) -> error::Result<(DebouncingSettings, ConcurrencySettings)> {
Self::from_runnable_settings_handle(hash, db)
.await?
.prefetch_cached(db)
.await
}
/// Returns error if provided `hash` has no corresponding entry in db
/// If `hash` is None, returnes Default
pub fn from_runnable_settings_handle<'a>(
hash: Option<i64>,
db: &'a DB,
) -> impl Future<Output = error::Result<Self>> + 'a {
async move {
if let Some(hash) = hash {
super::RUNNABLE_SETTINGS_REFERENCES
.get_or_insert_async(hash, async {
sqlx::query_as!(
RunnableSettings,
r#"SELECT concurrency_settings, debouncing_settings FROM runnable_settings WHERE hash = $1"#,
hash
)
.fetch_one(db)
.await
.map_err(error::Error::from)
})
.await
} else {
Ok(Self::default())
}
}
}
pub async fn insert_cached(self, db: &Pool<Postgres>) -> error::Result<Option<i64>> {
if !min_version_supports_runnable_settings_v0().await
|| (self.debouncing_settings.is_none() && self.concurrency_settings.is_none())
{
return Ok(None);
}
let hash = {
let mut h = std::hash::DefaultHasher::new();
self.hash(&mut h);
h.finish() as i64
};
super::RUNNABLE_SETTINGS_REFERENCES
.get_or_insert_async(hash, async {
sqlx::query!(
"INSERT INTO runnable_settings (hash, debouncing_settings, concurrency_settings)
VALUES ($1, $2, $3)
ON CONFLICT (hash)
DO NOTHING",
hash,
self.debouncing_settings,
self.concurrency_settings
)
.execute(db)
.await?;
// .map_err(error::Error::from)
Ok(self)
})
.await?;
Ok(Some(hash))
}
}
// TODO: Add validation logic.
#[derive(
Debug, Clone, Serialize, Deserialize, Default, Hash, PartialEq, sqlx::FromRow, sqlx::Type,
)]
pub struct DebouncingSettings {
#[serde(skip_serializing_if = "Option::is_none", alias = "custom_debounce_key")]
/// debounce key is usually stored in the db
/// including when:
///
/// 1. User have created custom debounce key from ui or cli
/// 2. User used default one
///
/// Default: hash(path + step_id + inputs)
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
/// Debouncing delay will be determined by the first job with the key.
/// All subsequent jobs with Some will get debounced.
/// If the job has no delay, it will execute immediately, fully ignoring pending delays.
pub debounce_delay_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub max_total_debouncing_time: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub max_total_debounces_amount: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
/// top level arguments to preserve
/// For every debounce selected arguments will be saved
/// in the end (when job finally starts) arguments will be appended and passed to runnable
///
/// NOTE: selected args should be the lists.
pub debounce_args_to_accumulate: Option<Vec<String>>,
}
#[derive(
Debug, Default, Clone, Serialize, Deserialize, Hash, PartialEq, sqlx::FromRow, sqlx::Decode,
)]
pub struct ConcurrencySettings {
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
}
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow, Default)]
pub struct ConcurrencySettingsWithCustom {
#[serde(skip_serializing_if = "Option::is_none")]
pub custom_concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
}
impl DebouncingSettings {
pub fn maybe_fallback(
self,
debounce_key: Option<String>,
debounce_delay_s: Option<i32>,
) -> Self {
Self {
debounce_key: self.debounce_key.or(debounce_key),
debounce_delay_s: self.debounce_delay_s.or(debounce_delay_s),
..self
}
}
}
impl ConcurrencySettings {
pub fn maybe_fallback(
self,
concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
) -> Self {
Self {
concurrency_key: self.concurrency_key.or(concurrency_key),
concurrent_limit: self.concurrent_limit.or(concurrent_limit),
concurrency_time_window_s: self.concurrency_time_window_s.or(concurrency_time_window_s),
}
}
}
impl super::private_mod::RunnableSettingsTraitInternal for DebouncingSettings {
const SETTINGS_NAME: &str = "debouncing_settings";
const INCLUDE_FIELDS: &[&str] = &[
"debounce_key",
"debounce_delay_s",
"max_total_debouncing_time",
"max_total_debounces_amount",
"debounce_args_to_accumulate",
];
fn bind_arguments<'a>(&'a self, q: Q<'a>) -> Q<'a> {
q.bind(&self.debounce_key)
.bind(&self.debounce_delay_s)
.bind(&self.max_total_debouncing_time)
.bind(&self.max_total_debounces_amount)
.bind(&self.debounce_args_to_accumulate)
}
}
impl super::RunnableSettingsTrait for DebouncingSettings {}
impl super::private_mod::RunnableSettingsTraitInternal for ConcurrencySettings {
const SETTINGS_NAME: &str = "concurrency_settings";
const INCLUDE_FIELDS: &[&str] = &[
"concurrency_key",
"concurrent_limit",
"concurrency_time_window_s",
];
fn bind_arguments<'a>(&'a self, q: Q<'a>) -> Q<'a> {
q.bind(&self.concurrency_key)
.bind(&self.concurrent_limit)
.bind(&self.concurrency_time_window_s)
}
}
impl super::RunnableSettingsTrait for ConcurrencySettings {}
impl From<ConcurrencySettings> for ConcurrencySettingsWithCustom {
fn from(
ConcurrencySettings { concurrency_key, concurrent_limit, concurrency_time_window_s }: ConcurrencySettings,
) -> Self {
ConcurrencySettingsWithCustom {
custom_concurrency_key: concurrency_key,
concurrency_time_window_s,
concurrent_limit,
}
}
}
impl From<ConcurrencySettingsWithCustom> for ConcurrencySettings {
fn from(
ConcurrencySettingsWithCustom {
custom_concurrency_key,
concurrent_limit,
concurrency_time_window_s,
}: ConcurrencySettingsWithCustom,
) -> Self {
ConcurrencySettings {
concurrency_key: custom_concurrency_key,
concurrency_time_window_s,
concurrent_limit,
}
}
}
+135 -33
View File
@@ -16,6 +16,7 @@ use std::{
use crate::{
assets::AssetWithAltAccessType,
error::{to_anyhow, Error},
runnable_settings::{ConcurrencySettings, DebouncingSettings, RunnableSettings},
utils::http_get_from_hub,
workspace_dependencies::WorkspaceDependenciesAnnotatedRefs,
DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, PRIVATE_HUB_MIN_VERSION,
@@ -348,7 +349,7 @@ pub fn id_to_codebase_info(id: &str) -> CodebaseInfo {
CodebaseInfo { is_tar, is_esm }
}
#[derive(Serialize, sqlx::FromRow, Debug)]
pub struct Script {
pub struct Script<SR> {
pub workspace_id: String,
pub hash: ScriptHash,
pub path: String,
@@ -375,16 +376,6 @@ pub struct Script {
#[serde(skip_serializing_if = "Option::is_none")]
pub envs: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
#[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>,
@@ -413,16 +404,119 @@ pub struct Script {
#[serde(skip_serializing_if = "Option::is_none")]
#[sqlx(json(nullable))]
pub assets: Option<Vec<AssetWithAltAccessType>>,
#[serde(flatten)]
#[sqlx(flatten)]
pub runnable_settings: SR,
}
// Not serializable
#[derive(sqlx::FromRow, Debug, Clone)]
pub struct ScriptRunnableSettingsHandle {
// legacy - for backwards compatibility
// don't add new values.
pub concurrency_key: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub debounce_key: Option<String>,
pub debounce_delay_s: Option<i32>,
// add here as well.
pub runnable_settings_handle: Option<i64>,
}
// Not sqlx queriable
#[derive(Serialize, Debug, Clone, Default)]
pub struct ScriptRunnableSettingsInline {
#[serde(flatten)]
pub concurrency_settings: ConcurrencySettings,
#[serde(flatten)]
pub debouncing_settings: DebouncingSettings,
}
impl Script<ScriptRunnableSettingsHandle> {
pub async fn prefetch_cached<'a>(
self,
db: &DB,
) -> crate::error::Result<Script<ScriptRunnableSettingsInline>> {
let (debouncing_settings, concurrency_settings) =
RunnableSettings::from_runnable_settings_handle(
self.runnable_settings.runnable_settings_handle,
db,
)
.await?
.prefetch_cached(db)
.await?;
Ok(Script {
workspace_id: self.workspace_id,
hash: self.hash,
path: self.path,
parent_hashes: self.parent_hashes,
summary: self.summary,
description: self.description,
content: self.content,
created_by: self.created_by,
created_at: self.created_at,
archived: self.archived,
schema: self.schema,
deleted: self.deleted,
is_template: self.is_template,
extra_perms: self.extra_perms,
lock: self.lock,
lock_error_logs: self.lock_error_logs,
language: self.language,
kind: self.kind,
tag: self.tag,
draft_only: self.draft_only,
envs: self.envs,
dedicated_worker: self.dedicated_worker,
ws_error_handler_muted: self.ws_error_handler_muted,
priority: self.priority,
cache_ttl: self.cache_ttl,
cache_ignore_s3_path: self.cache_ignore_s3_path,
timeout: self.timeout,
delete_after_use: self.delete_after_use,
restart_unless_cancelled: self.restart_unless_cancelled,
visible_to_runner_only: self.visible_to_runner_only,
no_main_func: self.no_main_func,
codebase: self.codebase,
has_preprocessor: self.has_preprocessor,
on_behalf_of_email: self.on_behalf_of_email,
assets: self.assets,
runnable_settings: ScriptRunnableSettingsInline {
concurrency_settings: concurrency_settings.maybe_fallback(
self.runnable_settings.concurrency_key,
self.runnable_settings.concurrent_limit,
self.runnable_settings.concurrency_time_window_s,
),
debouncing_settings: debouncing_settings.maybe_fallback(
self.runnable_settings.debounce_key,
self.runnable_settings.debounce_delay_s,
),
},
})
}
}
#[derive(Serialize, sqlx::FromRow)]
pub struct ScriptWithStarred {
pub struct ScriptWithStarred<SR> {
#[sqlx(flatten)]
#[serde(flatten)]
pub script: Script,
pub script: Script<SR>,
#[serde(skip_serializing_if = "Option::is_none")]
pub starred: Option<bool>,
}
impl ScriptWithStarred<ScriptRunnableSettingsHandle> {
pub async fn prefetch_cached<'a>(
self,
db: &DB,
) -> crate::error::Result<ScriptWithStarred<ScriptRunnableSettingsInline>> {
Ok(ScriptWithStarred {
script: self.script.prefetch_cached(db).await?,
starred: self.starred,
})
}
}
#[derive(Serialize, sqlx::FromRow)]
pub struct ListableScript {
@@ -497,17 +591,10 @@ pub struct NewScript {
pub tag: Option<String>,
pub draft_only: Option<bool>,
pub envs: Option<Vec<String>>,
// NOTE: concurrency and debounce data is inline,
// bc it was this before refactor
// and rust seems to hash it differently
// for backwards compat we keep them inline
pub concurrency_key: Option<String>,
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub debounce_delay_s: Option<i32>,
#[serde(flatten)]
pub concurrency_settings: ConcurrencySettings,
#[serde(flatten)]
pub debouncing_settings: DebouncingSettings,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub dedicated_worker: Option<bool>,
@@ -827,8 +914,8 @@ pub async fn fetch_script_for_update<'a>(
hash: ScriptHash,
w_id: &str,
e: impl sqlx::Executor<'a, Database = sqlx::Postgres>,
) -> crate::error::Result<Option<Script>> {
sqlx::query_as::<_, Script>(
) -> crate::error::Result<Option<Script<ScriptRunnableSettingsHandle>>> {
sqlx::query_as::<_, Script<ScriptRunnableSettingsHandle>>(
"SELECT
workspace_id,
hash,
@@ -857,6 +944,7 @@ pub async fn fetch_script_for_update<'a>(
debounce_key,
debounce_delay_s,
dedicated_worker,
runnable_settings_handle,
ws_error_handler_muted,
priority,
cache_ttl,
@@ -888,6 +976,7 @@ pub async fn clone_script<'c>(
base_hash: ScriptHash,
w_id: &str,
deployment_message: Option<String>,
db: &DB,
tx: &mut sqlx::Transaction<'c, sqlx::Postgres>,
) -> crate::error::Result<ClonedScript> {
let s = if let Some(s) = fetch_script_for_update(base_hash, w_id, &mut **tx).await? {
@@ -899,6 +988,15 @@ pub async fn clone_script<'c>(
)));
};
let (debouncing_settings, concurrency_settings) =
RunnableSettings::from_runnable_settings_handle(
s.runnable_settings.runnable_settings_handle,
db,
)
.await?
.prefetch_cached(db)
.await?;
let ns = NewScript {
path: s.path.clone(),
parent_hash: Some(base_hash),
@@ -915,8 +1013,15 @@ pub async fn clone_script<'c>(
tag: s.tag,
draft_only: s.draft_only,
envs: s.envs,
concurrent_limit: s.concurrent_limit,
concurrency_time_window_s: s.concurrency_time_window_s,
concurrency_settings: concurrency_settings.maybe_fallback(
s.runnable_settings.concurrency_key,
s.runnable_settings.concurrent_limit,
s.runnable_settings.concurrency_time_window_s,
),
debouncing_settings: debouncing_settings.maybe_fallback(
s.runnable_settings.debounce_key,
s.runnable_settings.debounce_delay_s,
),
cache_ttl: s.cache_ttl,
cache_ignore_s3_path: s.cache_ignore_s3_path,
dedicated_worker: s.dedicated_worker,
@@ -926,15 +1031,12 @@ pub async fn clone_script<'c>(
delete_after_use: s.delete_after_use,
restart_unless_cancelled: s.restart_unless_cancelled,
deployment_message,
concurrency_key: s.concurrency_key,
visible_to_runner_only: s.visible_to_runner_only,
no_main_func: s.no_main_func,
codebase: s.codebase,
has_preprocessor: s.has_preprocessor,
on_behalf_of_email: s.on_behalf_of_email,
assets: s.assets,
debounce_delay_s: s.debounce_delay_s,
debounce_key: s.debounce_key,
};
let new_hash = hash_script(&ns);
@@ -953,14 +1055,14 @@ pub async fn clone_script<'c>(
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)
codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s, runnable_settings_handle)
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
codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s, runnable_settings_handle
FROM script WHERE hash = $2 AND workspace_id = $3;
", new_hash, base_hash.0, w_id).execute(&mut **tx).await?;
-2
View File
@@ -1,5 +1,3 @@
#[cfg(feature = "private")]
#[allow(unused)]
pub use crate::teams_ee::*;
+6 -2
View File
@@ -265,6 +265,7 @@ lazy_static::lazy_static! {
.unwrap_or(false);
pub static ref MIN_VERSION: Arc<RwLock<Version>> = Arc::new(RwLock::new(Version::new(0, 0, 0)));
pub static ref MIN_VERSION_SUPPORTS_RUNNABLE_SETTINGS_V0: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
/// Global flag indicating if all workers support workspace dependencies feature (>= 1.583.0)
/// This flag is updated during worker initialization by checking the minimum version across all workers
/// When false, creation of workspace dependencies is forbidden and extraction of external workspace dependencies will error
@@ -411,7 +412,7 @@ fn format_pull_query(peek: String) -> String {
WHERE id = (SELECT id FROM peek)
RETURNING
started_at, scheduled_for,
canceled_by, canceled_reason, worker, cache_ignore_s3_path
canceled_by, canceled_reason, worker, cache_ignore_s3_path, runnable_settings_handle
), r AS NOT MATERIALIZED (
UPDATE v2_job_runtime SET
ping = now()
@@ -437,7 +438,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, q.cache_ignore_s3_path, 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, q.runnable_settings_handle, 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
@@ -1293,6 +1294,9 @@ pub async fn update_min_version(conn: &Connection) -> bool {
tracing::info!("Minimal worker version: {min_version}");
}
*MIN_VERSION_SUPPORTS_RUNNABLE_SETTINGS_V0.write().await =
min_version >= *crate::runnable_settings::MIN_VERSION_RUNNABLE_SETTINGS_V0;
// Workspace dependencies feature requires minimum version across all workers
*MIN_VERSION_SUPPORTS_V0_WORKSPACE_DEPENDENCIES.write().await = min_version
>= Version::parse(crate::workspace_dependencies::MIN_VERSION_WORKSPACE_DEPENDENCIES)
+165 -53
View File
@@ -38,9 +38,10 @@ use windmill_common::audit::AuditAuthor;
use windmill_common::auth::JobPerms;
#[cfg(feature = "benchmark")]
use windmill_common::bench::BenchmarkIter;
use windmill_common::jobs::{
ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, JobTriggerKind,
EMAIL_ERROR_HANDLER_USER_EMAIL,
use windmill_common::jobs::{JobTriggerKind, EMAIL_ERROR_HANDLER_USER_EMAIL};
use windmill_common::runnable_settings::{
ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, RunnableSettings,
RunnableSettingsTrait,
};
use windmill_common::triggers::TriggerMetadata;
use windmill_common::utils::{configure_client, now_from_db};
@@ -1148,7 +1149,14 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
}
}
}
if queued_job.concurrent_limit.is_some() {
if queued_job.concurrent_limit.is_some()
|| RunnableSettings::prefetch_cached_from_handle(queued_job.runnable_settings_handle, db)
.await?
.1
.concurrent_limit
.is_some()
{
let concurrency_key = concurrency_key(db, &queued_job.id).await?;
if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_none() {
tracing::warn!("Concurrency limit is disabled, skipping");
@@ -2008,10 +2016,10 @@ pub struct MiniPulledJob {
pub trigger_kind: Option<JobTriggerKind>,
pub visible_to_owner: bool,
pub permissioned_as_end_user_email: Option<String>,
pub runnable_settings_handle: Option<i64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MiniCompletedJob {
pub id: Uuid,
pub workspace_id: String,
@@ -2035,6 +2043,7 @@ pub struct MiniCompletedJob {
pub tag: String,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub runnable_settings_handle: Option<i64>,
}
impl From<QueuedJobV2> for MiniCompletedJob {
@@ -2061,6 +2070,7 @@ impl From<QueuedJobV2> for MiniCompletedJob {
tag: job.tag,
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
runnable_settings_handle: job.runnable_settings_handle,
}
}
}
@@ -2090,6 +2100,7 @@ impl From<MiniPulledJob> for MiniCompletedJob {
tag: job.tag,
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
runnable_settings_handle: job.runnable_settings_handle,
}
}
}
@@ -2118,6 +2129,7 @@ impl From<Arc<MiniPulledJob>> for MiniCompletedJob {
tag: job.tag.clone(),
cache_ttl: job.cache_ttl,
cache_ignore_s3_path: job.cache_ignore_s3_path,
runnable_settings_handle: job.runnable_settings_handle,
}
}
}
@@ -2212,6 +2224,7 @@ impl MiniPulledJob {
pre_run_error: job.pre_run_error.clone(),
concurrent_limit: job.concurrent_limit.clone(),
concurrency_time_window_s: job.concurrency_time_window_s.clone(),
runnable_settings_handle: job.runnable_settings_handle,
flow_innermost_root_job: job.root_job.clone(), // QueuedJob is taken from v2_as_queue, where root_job corresponds to flow_innermost_root_job in v2_job
root_job: None,
timeout: job.timeout.clone(),
@@ -2261,7 +2274,7 @@ impl MiniPulledJob {
}
}
#[derive(sqlx::FromRow, Debug, Clone, Serialize, Deserialize)]
#[derive(sqlx::FromRow, Debug, Clone)]
pub struct PulledJob {
#[sqlx(flatten)]
pub job: MiniPulledJob,
@@ -2405,6 +2418,7 @@ pub async fn get_mini_pulled_job<'c>(
v2_job.parent_job,
v2_job.created_by,
v2_job_queue.started_at,
v2_job_queue.runnable_settings_handle,
scheduled_for,
runnable_path,
kind as \"kind: JobKind\",
@@ -2464,6 +2478,7 @@ pub struct QueuedJobV2 {
pub tag: String,
pub cache_ttl: Option<i32>,
pub cache_ignore_s3_path: Option<bool>,
pub runnable_settings_handle: Option<i64>,
pub last_ping: Option<chrono::DateTime<chrono::Utc>>,
pub worker: Option<String>,
pub memory_peak: Option<i32>,
@@ -2482,19 +2497,46 @@ pub async fn get_queued_job_v2<'c>(
) -> error::Result<Option<QueuedJobV2>> {
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, 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",
r#"SELECT
id,
q.runnable_settings_handle,
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,
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,
)
.fetch_optional(e)
.await?;
Ok(job)
}
#[derive(Serialize, Deserialize, Debug)]
#[derive(Debug)]
pub struct PulledJobResult {
pub job: Option<PulledJob>,
pub suspended: bool,
@@ -2683,6 +2725,7 @@ impl PulledJobResult {
base_hash,
&pulled_job.workspace_id,
deployment_message,
db,
&mut tx,
)
.await?;
@@ -2802,6 +2845,7 @@ impl PulledJobResult {
}
}
// TODO: Factorize
/// Pull the job from queue
pub async fn pull(
db: &Pool<Postgres>,
@@ -2874,21 +2918,39 @@ pub async fn pull(
}
}
#[cfg(feature = "private")]
let concurrency_settings = if let Some(ref j) = job {
RunnableSettings::from_runnable_settings_handle(j.runnable_settings_handle, db)
.await?
.prefetch_cached(db)
.await?
.1
.maybe_fallback(None, j.concurrent_limit, j.concurrency_time_window_s)
} else {
Default::default()
};
let pulled_job_result = match job {
#[cfg(feature = "private")]
Some(job)
if job.concurrent_limit.is_some()
if concurrency_settings.concurrent_limit.is_some()
// Concurrency limit is available for either enterprise job or dependency job
&& (cfg!(feature = "enterprise") || (job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING)) =>
{
crate::jobs_ee::apply_concurrency_limit(db, pull_loop_count, suspended, job)
.await?
.unwrap_or(PulledJobResult {
job: None,
suspended,
missing_concurrency_key: false,
error_while_preprocessing: None,
})
crate::jobs_ee::apply_concurrency_limit(
db,
pull_loop_count,
suspended,
job,
&concurrency_settings,
)
.await?
.unwrap_or(PulledJobResult {
job: None,
suspended,
missing_concurrency_key: false,
error_while_preprocessing: None,
})
}
_ => PulledJobResult {
job,
@@ -2921,7 +2983,15 @@ pub async fn pull(
});
};
let has_concurent_limit = job.concurrent_limit.is_some();
let concurrency_settings =
RunnableSettings::from_runnable_settings_handle(job.runnable_settings_handle, db)
.await?
.prefetch_cached(db)
.await?
.1
.maybe_fallback(None, job.concurrent_limit, job.concurrency_time_window_s);
let has_concurent_limit = concurrency_settings.concurrent_limit.is_some();
#[cfg(not(feature = "enterprise"))]
if has_concurent_limit && !job.is_dependency() {
@@ -2957,9 +3027,14 @@ pub async fn pull(
if cfg!(feature = "enterprise")
|| (pulled_job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING)
{
if let Some(pulled_job_res) =
crate::jobs_ee::apply_concurrency_limit(db, pull_loop_count, suspended, pulled_job)
.await?
if let Some(pulled_job_res) = crate::jobs_ee::apply_concurrency_limit(
db,
pull_loop_count,
suspended,
pulled_job,
&concurrency_settings,
)
.await?
{
return Ok(pulled_job_res);
}
@@ -3640,7 +3715,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, q.cache_ignore_s3_path
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, q.runnable_settings_handle
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,
@@ -4096,12 +4171,7 @@ pub async fn push<'c, 'd>(
cache_ignore_s3_path,
dedicated_worker,
_low_level_priority,
concurrency_settings:
ConcurrencySettings {
mut concurrency_key,
mut concurrent_limit,
concurrency_time_window_s, //
},
mut concurrency_settings,
debouncing_settings,
} = match job_payload {
JobPayload::ScriptHash {
@@ -4567,7 +4637,7 @@ pub async fn push<'c, 'd>(
concurrency_settings.concurrent_limit = None;
// TODO: May be re-enable?
debouncing_settings.delay_s = None;
debouncing_settings.debounce_delay_s = None;
preprocessed = Some(false);
}
@@ -4708,8 +4778,8 @@ pub async fn push<'c, 'd>(
&& !*WMDEBUG_NO_DJOB_DEBOUNCING
&& *MIN_VERSION_SUPPORTS_DEBOUNCING.read().await,
) {
concurrency_key = Some(format!("dependency:{workspace_id}/{path}"));
concurrent_limit = Some(1);
concurrency_settings.concurrency_key = Some(format!("dependency:{workspace_id}/{path}"));
concurrency_settings.concurrent_limit = Some(1);
}
let final_priority: Option<i16>;
@@ -5034,8 +5104,8 @@ pub async fn push<'c, 'd>(
if schedule_path.is_none() {
if let Some(debounced_job_id) = crate::jobs_ee::maybe_apply_debouncing(
&job_id,
debouncing_settings.delay_s,
debouncing_settings.custom_key,
debouncing_settings.debounce_delay_s,
debouncing_settings.debounce_key.clone(),
workspace_id,
runnable_path.clone(),
&job_kind,
@@ -5056,13 +5126,13 @@ pub async fn push<'c, 'd>(
)
.unzip();
if concurrent_limit.is_some() {
if concurrency_settings.concurrent_limit.is_some() {
insert_concurrency_key(
workspace_id,
&args,
&runnable_path,
job_kind,
concurrency_key,
concurrency_settings.concurrency_key.clone(),
&mut tx,
job_id,
)
@@ -5156,14 +5226,56 @@ pub async fn push<'c, 'd>(
(job_kind, scheduled_for_o)
};
let runnable_settings_handle = RunnableSettings {
debouncing_settings: debouncing_settings.insert_cached(_db).await?,
concurrency_settings: concurrency_settings.insert_cached(_db).await?,
}
.insert_cached(_db)
.await?;
let (guarded_concurrent_limit, guarded_concurrency_time_window_s) =
if windmill_common::runnable_settings::min_version_supports_runnable_settings_v0().await {
(None, None)
} else {
(
concurrency_settings.concurrent_limit,
concurrency_settings.concurrency_time_window_s,
)
};
sqlx::query!(
"WITH inserted_job AS (
INSERT INTO v2_job (id, workspace_id, raw_code, raw_lock, raw_flow, tag, parent_job,
created_by, permissioned_as, runnable_id, runnable_path, args, kind, trigger,
script_lang, same_worker, pre_run_error, permissioned_as_email, visible_to_owner,
flow_innermost_root_job, root_job, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id,
cache_ttl, priority, trigger_kind, script_entrypoint_override, preprocessed)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
INSERT INTO v2_job (
id, -- 1
workspace_id, -- 2
raw_code, -- 3
raw_lock, -- 4
raw_flow, -- 5
tag, -- 6
parent_job, -- 7
created_by, -- 8
permissioned_as, -- 9
runnable_id, -- 10
runnable_path, -- 11
args, -- 12
kind, -- 13
trigger, -- 14
script_lang, -- 15
same_worker, -- 16
pre_run_error, -- 17
permissioned_as_email, -- 18
visible_to_owner, -- 19
flow_innermost_root_job, -- 20
root_job, -- 38
concurrent_limit, -- 21
concurrency_time_window_s, -- 22
timeout, -- 23
flow_step_id, -- 24
cache_ttl, -- 25
priority, -- 26
trigger_kind, -- 39
script_entrypoint_override, -- 12
preprocessed -- 27,
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
$19, $20, $38, $21, $22, $23, $24, $25, $26, $39::job_trigger_kind,
($12::JSONB)->>'_ENTRYPOINT_OVERRIDE', $27)
),
@@ -5176,8 +5288,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, cache_ignore_s3_path)
VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42)",
(workspace_id, id, running, scheduled_for, started_at, tag, priority, cache_ignore_s3_path, runnable_settings_handle)
VALUES ($2, $1, $28, COALESCE($29, now()), CASE WHEN $27 OR $40 THEN now() END, $30, $31, $42, $43)",
job_id,
workspace_id,
raw_code,
@@ -5198,12 +5310,8 @@ pub async fn push<'c, 'd>(
email,
visible_to_owner,
flow_innermost_root_job,
concurrent_limit,
if concurrent_limit.is_some() {
concurrency_time_window_s
} else {
None
},
guarded_concurrent_limit,
guarded_concurrency_time_window_s,
custom_timeout,
flow_step_id,
cache_ttl,
@@ -5224,11 +5332,14 @@ pub async fn push<'c, 'd>(
running,
end_user_email,
cache_ignore_s3_path,
runnable_settings_handle,
)
.execute(&mut *tx)
.warn_after_seconds(1)
.await?;
// RunnableSettings::insert(RunnableType::Job)
// tracing::debug!("Pushing job {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}");
// let uuid = sqlx::query_scalar!(
// "INSERT INTO v2_job_queue
@@ -5734,6 +5845,7 @@ pub async fn get_same_worker_job(
v2_job.flow_step_id,
v2_job.cache_ttl,
v2_job_queue.cache_ignore_s3_path,
v2_job_queue.runnable_settings_handle,
v2_job_queue.priority,
v2_job.preprocessed,
v2_job.script_entrypoint_override,
+18 -12
View File
@@ -20,10 +20,11 @@ use windmill_common::flows::Retry;
use windmill_common::get_flow_version_info_from_version;
use windmill_common::get_latest_flow_version_id_for_path;
use windmill_common::jobs::check_tag_available_for_workspace_internal;
use windmill_common::jobs::ConcurrencySettings;
use windmill_common::jobs::DebouncingSettings;
use windmill_common::jobs::JobPayload;
use windmill_common::jobs::JobTriggerKind;
use windmill_common::runnable_settings::ConcurrencySettings;
use windmill_common::runnable_settings::DebouncingSettings;
use windmill_common::runnable_settings::RunnableSettings;
use windmill_common::schedule::schedule_to_user;
use windmill_common::scripts::ScriptHash;
use windmill_common::triggers::TriggerMetadata;
@@ -101,6 +102,7 @@ async fn get_schedule_metadata<'c>(
timeout,
on_behalf_of_email,
created_by,
_runnable_settings_handle,
) = windmill_common::get_latest_hash_for_path(
&mut **tx,
&schedule.workspace_id,
@@ -319,10 +321,10 @@ pub async fn push_scheduled_job<'c>(
let (
hash,
tag,
custom_concurrency_key,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
custom_debounce_key,
debounce_key,
debounce_delay_s,
cache_ttl,
cache_ignore_s3_path,
@@ -332,6 +334,7 @@ pub async fn push_scheduled_job<'c>(
timeout,
on_behalf_of_email,
created_by,
runnable_settings_handle,
) = windmill_common::get_latest_hash_for_path(
&mut *tx,
&schedule.workspace_id,
@@ -341,6 +344,12 @@ pub async fn push_scheduled_job<'c>(
.warn_after_seconds_with_sql(1, "get_latest_hash_for_path".to_string())
.await?;
let (debouncing_settings, concurrency_settings) =
RunnableSettings::from_runnable_settings_handle(runnable_settings_handle, db)
.await?
.prefetch_cached(db)
.await?;
if schedule.retry.is_some() {
let parsed_retry = serde_json::from_value::<Retry>(schedule.retry.clone().unwrap())
.map_err(|err| {
@@ -393,16 +402,13 @@ pub async fn push_scheduled_job<'c>(
language,
priority,
apply_preprocessor: false,
debouncing_settings: DebouncingSettings {
custom_key: custom_debounce_key,
delay_s: debounce_delay_s,
..Default::default()
},
concurrency_settings: ConcurrencySettings {
concurrency_key: custom_concurrency_key,
debouncing_settings: debouncing_settings
.maybe_fallback(debounce_key, debounce_delay_s),
concurrency_settings: concurrency_settings.maybe_fallback(
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
},
),
},
if schedule.tag.as_ref().is_some_and(|x| x != "") {
schedule.tag.clone()
+11 -6
View File
@@ -2832,10 +2832,14 @@ pub async fn handle_queued_job(
}
#[cfg(not(feature = "enterprise"))]
if job.concurrent_limit.is_some() && !job.kind.is_dependency() {
logs.push_str("---\n");
logs.push_str("WARNING: This job has concurrency limits enabled. Concurrency limits are an EE feature and the setting is ignored.\n");
logs.push_str("---\n");
if let Connection::Sql(db) = conn {
if (job.concurrent_limit.is_some() ||
windmill_common::runnable_settings::RunnableSettings::prefetch_cached_from_handle(job.runnable_settings_handle, db).await?.1.concurrent_limit.is_some())
&& !job.kind.is_dependency() {
logs.push_str("---\n");
logs.push_str("WARNING: This job has concurrency limits enabled. Concurrency limits are an EE feature and the setting is ignored.\n");
logs.push_str("---\n");
}
}
// Only used for testing in tests/relative_imports.rs
@@ -4282,8 +4286,6 @@ pub fn init_worker_internal_server_inline_utils(
script_lang: Some(params.lang),
same_worker: true,
pre_run_error: None,
concurrent_limit: None,
concurrency_time_window_s: None,
flow_innermost_root_job: None,
root_job: None,
timeout: None,
@@ -4297,6 +4299,9 @@ pub fn init_worker_internal_server_inline_utils(
trigger_kind: None,
visible_to_owner: false,
permissioned_as_end_user_email: None,
runnable_settings_handle: None,
concurrent_limit: None,
concurrency_time_window_s: None,
};
Box::pin(async move {
let mut mem_peak: i32 = -1;
+12 -19
View File
@@ -43,10 +43,10 @@ use windmill_common::flow_status::{
};
use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId, StopAfterIf};
use windmill_common::jobs::{
script_path_to_payload, ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings,
JobKind, JobPayload, OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE,
script_path_to_payload, JobKind, JobPayload, OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE,
};
use windmill_common::scripts::ScriptHash;
use windmill_common::runnable_settings::{ConcurrencySettingsWithCustom, DebouncingSettings};
use windmill_common::scripts::{ScriptHash, ScriptRunnableSettingsInline};
use windmill_common::users::username_to_permissioned_as;
use windmill_common::utils::WarnAfterExt;
use windmill_common::worker::to_raw_value;
@@ -4756,13 +4756,14 @@ pub async fn script_to_payload(
timeout,
on_behalf_of_email,
created_by,
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
debounce_key,
debounce_delay_s,
runnable_settings:
ScriptRunnableSettingsInline { concurrency_settings, debouncing_settings },
..
} = get_script_info_for_hash(None, db, &flow_job.workspace_id, hash.0).await?;
} = get_script_info_for_hash(None, db, &flow_job.workspace_id, hash.0)
.await?
.prefetch_cached(&db)
.await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf { email, permissioned_as: username_to_permissioned_as(&created_by) })
} else {
@@ -4776,16 +4777,8 @@ pub async fn script_to_payload(
JobPayload::ScriptHash {
hash,
path: script_path,
debouncing_settings: DebouncingSettings {
custom_key: debounce_key,
delay_s: debounce_delay_s,
..Default::default()
},
concurrency_settings: ConcurrencySettings {
concurrency_key,
concurrent_limit,
concurrency_time_window_s,
},
concurrency_settings,
debouncing_settings,
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,
+1 -1
View File
@@ -274,7 +274,7 @@
REMOTE = "http://127.0.0.1:8000";
REMOTE_LSP = "http://127.0.0.1:3001";
RUSTC_WRAPPER = "${pkgs.sccache}/bin/sccache";
# RUSTC_WRAPPER = "${pkgs.sccache}/bin/sccache";
DENO_PATH = "${pkgs.deno}/bin/deno";
GO_PATH = "${pkgs.go}/bin/go";
PHP_PATH = "${pkgs.php}/bin/php";