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

fix: check perms more tightly for running jobs (#6532)

Browse Source 
* all

* masterpiece

* masterpiece

* masterpiece

* masterpiece

* masterpiece

* masterpiece

* masterpiece

* all

* masterpiece

* masterpiece

* masterpiece

* masterpiece

* masterpiece

* masterpiece

* nits
This commit is contained in:
Ruben Fiszel
2025-09-05 06:07:45 +00:00
committed by GitHub
parent 8ae43d86be
commit 0647202a4f
34 changed files with 753 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-2ccf7a95ed41083d92d6d9b29f700efb77317837a8bcffc45e23e5ad21e412ef.json
Generated
-12
View File
@@ -1,12 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM _sqlx_migrations WHERE \n version=20250131115248 OR version=20250902085503 OR version=20250201145630 OR \n version=20250201145631 OR version=20250201145632",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "2ccf7a95ed41083d92d6d9b29f700efb77317837a8bcffc45e23e5ad21e412ef"
}
backend/.sqlx/query-91f23fcc27777c279c79e2682fc15c026e55f9ec3799be65a2e8920fe6174a17.json → backend/.sqlx/query-48fa200f9d1535ef606c5b1c9b6f3e8dff80a4c64857bc57e9fadce2da6b1a5b.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT result as \"result: Json<HashMap<String, Box<RawValue>>>\"\n FROM v2_job_completed \n WHERE id = $1",
"query": "SELECT result as \"result: Json<HashMap<String, Box<RawValue>>>\"\n FROM v2_job_completed\n WHERE id = $1",
"describe": {
"columns": [
{
@@ -18,5 +18,5 @@
true
]
},
"hash": "91f23fcc27777c279c79e2682fc15c026e55f9ec3799be65a2e8920fe6174a17"
"hash": "48fa200f9d1535ef606c5b1c9b6f3e8dff80a4c64857bc57e9fadce2da6b1a5b"
}
backend/.sqlx/query-506f90984a8b672aeff50b6d9e4751ea1521c524b9b7f51a48ee26280825ac9a.json
Generated
-36
View File
@@ -1,36 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n COALESCE(jc.result, NULL) as \"result: sqlx::types::Json<Box<RawValue>>\", \n SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\", \n CHAR_LENGTH(rs.stream) + 1 AS stream_offset\n FROM (\n SELECT $2::uuid as job_id, $1::text as workspace_id\n ) base\n LEFT JOIN v2_job_completed jc ON jc.id = base.job_id AND jc.workspace_id = base.workspace_id\n LEFT JOIN job_result_stream rs ON rs.job_id = base.job_id\n WHERE base.job_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "result: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 1,
"name": "result_stream: Option<String>",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "stream_offset",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Uuid",
"Int4"
]
},
"nullable": [
null,
null,
null
]
},
"hash": "506f90984a8b672aeff50b6d9e4751ea1521c524b9b7f51a48ee26280825ac9a"
}
backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
Generated
+1 -1
View File
@@ -15,7 +15,7 @@
]
},
"nullable": [
true
null
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
backend/.sqlx/query-5b8c1803f0ccead11517fbc8a9bdc0227dc3922217fa18f0b71ff0484d65838c.json
Generated
+18
View File
@@ -0,0 +1,18 @@
{
"db_name": "PostgreSQL",
"query": "WITH job_result AS (\n SELECT result\n FROM v2_job_completed\n WHERE id = $1\n ),\n updated_queue AS (\n UPDATE v2_job_queue\n SET running = false,\n tag = COALESCE($3, tag)\n WHERE id = $2\n )\n UPDATE v2_job\n SET\n tag = COALESCE($3, tag),\n concurrent_limit = COALESCE($4, concurrent_limit),\n concurrency_time_window_s = COALESCE($5, concurrency_time_window_s),\n args = COALESCE(\n CASE\n WHEN job_result.result IS NULL THEN NULL\n WHEN jsonb_typeof(job_result.result) = 'object'\n THEN job_result.result\n WHEN jsonb_typeof(job_result.result) = 'null'\n THEN NULL\n ELSE jsonb_build_object('value', job_result.result)\n END,\n '{}'::jsonb\n ),\n preprocessed = TRUE\n FROM job_result\n WHERE v2_job.id = $2;\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Uuid",
"Varchar",
"Int4",
"Int4"
]
},
"nullable": []
},
"hash": "5b8c1803f0ccead11517fbc8a9bdc0227dc3922217fa18f0b71ff0484d65838c"
}
backend/.sqlx/query-618ec69c9c78f1c9e3539d2770392e3f783f29cd2cc58c0fc87d14ecef32b467.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM _sqlx_migrations WHERE\n version=20250131115248 OR version=20250902085503 OR version=20250201145630 OR\n version=20250201145631 OR version=20250201145632",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "618ec69c9c78f1c9e3539d2770392e3f783f29cd2cc58c0fc87d14ecef32b467"
}
backend/.sqlx/query-6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d.json
Generated
+42
View File
@@ -0,0 +1,42 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n COALESCE(jc.result, jc.result) as \"result: sqlx::types::Json<Box<RawValue>>\",\n jq.running as \"running: Option<bool>\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset\n FROM (\n SELECT $1::uuid as job_id, $2::text as workspace_id\n ) base\n LEFT JOIN v2_job_completed jc ON jc.id = base.job_id AND jc.workspace_id = base.workspace_id\n LEFT JOIN v2_job_queue jq ON jq.id = base.job_id AND jq.workspace_id = base.workspace_id\n LEFT JOIN job_result_stream rs ON rs.job_id = base.job_id\n WHERE base.job_id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "result: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 1,
"name": "running: Option<bool>",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "result_stream: Option<String>",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "stream_offset",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Uuid",
"Text",
"Int4"
]
},
"nullable": [
null,
false,
null,
null
]
},
"hash": "6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d"
}
backend/.sqlx/query-4f372d047c78532907adf2d2dc114352aa7b5b28dccc50a1231a7f6539397da7.json → backend/.sqlx/query-70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n c.id IS NOT NULL AS completed,\n CASE \n WHEN q.id IS NOT NULL THEN (CASE WHEN NOT $5 AND q.running THEN true ELSE null END)\n ELSE false\n END AS running,\n CASE WHEN $7::BOOLEAN THEN NULL ELSE SUBSTR(logs, GREATEST($1 - log_offset, 0)) END AS logs,\n SUBSTR(rs.stream, $8) AS new_result_stream,\n COALESCE(r.memory_peak, c.memory_peak) AS mem_peak,\n COALESCE(c.flow_status, f.flow_status) AS \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n COALESCE(c.workflow_as_code_status, f.workflow_as_code_status) AS \"workflow_as_code_status: sqlx::types::Json<Box<RawValue>>\",\n CASE WHEN $7::BOOLEAN THEN NULL ELSE job_logs.log_offset + CHAR_LENGTH(job_logs.logs) + 1 END AS log_offset,\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset,\n created_by AS \"created_by!\",\n CASE WHEN $4::BOOLEAN THEN (\n SELECT scalar_int FROM job_stats WHERE job_id = $3 AND metric_id = 'progress_perc'\n ) END AS progress,\n rs.stream AS \"result_stream: Option<String>\"\n FROM v2_job j\n LEFT JOIN v2_job_queue q USING (id)\n LEFT JOIN v2_job_runtime r USING (id)\n LEFT JOIN v2_job_status f USING (id)\n LEFT JOIN v2_job_completed c USING (id)\n LEFT JOIN job_result_stream rs ON rs.job_id = $3\n LEFT JOIN job_logs ON job_logs.job_id = $3\n WHERE j.workspace_id = $2 AND j.id = $3\n AND ($6::text[] IS NULL OR j.tag = ANY($6))",
"query": "SELECT\n c.id IS NOT NULL AS completed,\n CASE\n WHEN q.id IS NOT NULL THEN (CASE WHEN NOT $5 AND q.running THEN true ELSE null END)\n ELSE false\n END AS running,\n CASE WHEN $7::BOOLEAN THEN NULL ELSE SUBSTR(logs, GREATEST($1 - log_offset, 0)) END AS logs,\n SUBSTR(rs.stream, $8) AS new_result_stream,\n COALESCE(r.memory_peak, c.memory_peak) AS mem_peak,\n COALESCE(c.flow_status, f.flow_status) AS \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n COALESCE(c.workflow_as_code_status, f.workflow_as_code_status) AS \"workflow_as_code_status: sqlx::types::Json<Box<RawValue>>\",\n CASE WHEN $7::BOOLEAN THEN NULL ELSE job_logs.log_offset + CHAR_LENGTH(job_logs.logs) + 1 END AS log_offset,\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset,\n created_by AS \"created_by!\",\n CASE WHEN $4::BOOLEAN THEN (\n SELECT scalar_int FROM job_stats WHERE job_id = $3 AND metric_id = 'progress_perc'\n ) END AS progress,\n rs.stream AS \"result_stream: Option<String>\"\n FROM v2_job j\n LEFT JOIN v2_job_queue q USING (id)\n LEFT JOIN v2_job_runtime r USING (id)\n LEFT JOIN v2_job_status f USING (id)\n LEFT JOIN v2_job_completed c USING (id)\n LEFT JOIN job_result_stream rs ON rs.job_id = $3\n LEFT JOIN job_logs ON job_logs.job_id = $3\n WHERE j.workspace_id = $2 AND j.id = $3\n AND ($6::text[] IS NULL OR j.tag = ANY($6))",
"describe": {
"columns": [
{
@@ -91,5 +91,5 @@
false
]
},
"hash": "4f372d047c78532907adf2d2dc114352aa7b5b28dccc50a1231a7f6539397da7"
"hash": "70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393"
}
backend/.sqlx/query-9dec888d4b0666d1843fbbc4fb2475fd947f047a9965bc110d7338208b77783d.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT flow_version.id from flow\n INNER JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 and flow.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "9dec888d4b0666d1843fbbc4fb2475fd947f047a9965bc110d7338208b77783d"
}
backend/.sqlx/query-b5fbd7893950610f1285662df24f438c9855ba860e23befd88c2544ef86e9133.json
Generated
+17
View File
@@ -0,0 +1,17 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO script\n (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets)\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, $4, 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\n\n FROM script WHERE hash = $2 AND workspace_id = $3;\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Int8",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "b5fbd7893950610f1285662df24f438c9855ba860e23befd88c2544ef86e9133"
}
backend/.sqlx/query-c61f9bad73beb0bc317594f6ca68389c2d0adbef1fb6d73ad23e4153cff44f9e.json
Generated
-42
View File
@@ -1,42 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n COALESCE(jc.result, jc.result) as \"result: sqlx::types::Json<Box<RawValue>>\",\n jq.running as \"running: Option<bool>\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset\n FROM (\n SELECT $1::uuid as job_id, $2::text as workspace_id\n ) base\n LEFT JOIN v2_job_completed jc ON jc.id = base.job_id AND jc.workspace_id = base.workspace_id\n LEFT JOIN v2_job_queue jq ON jq.id = base.job_id AND jq.workspace_id = base.workspace_id\n LEFT JOIN job_result_stream rs ON rs.job_id = base.job_id\n WHERE base.job_id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "result: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 1,
"name": "running: Option<bool>",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "result_stream: Option<String>",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "stream_offset",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Uuid",
"Text",
"Int4"
]
},
"nullable": [
null,
false,
null,
null
]
},
"hash": "c61f9bad73beb0bc317594f6ca68389c2d0adbef1fb6d73ad23e4153cff44f9e"
}
backend/.sqlx/query-c76db7bac7a11b98eaf401de54ceb8d76698d349a2e968b5f06c12c1358d6537.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT flow_version.id from flow\n INNER JOIN flow_version\n ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]\n WHERE flow.path = $1 and flow.workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "c76db7bac7a11b98eaf401de54ceb8d76698d349a2e968b5f06c12c1358d6537"
}
backend/.sqlx/query-ce6f3e803909d55c19169c77d4111bffc0fc93032943369015971373f1f2af68.json
Generated
-17
View File
@@ -1,17 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO script\n (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets) \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, $4, 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 \n\n FROM script WHERE hash = $2 AND workspace_id = $3;\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int8",
"Int8",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "ce6f3e803909d55c19169c77d4111bffc0fc93032943369015971373f1f2af68"
}
backend/.sqlx/query-d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d.json
Generated
+36
View File
@@ -0,0 +1,36 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n COALESCE(jc.result, NULL) as \"result: sqlx::types::Json<Box<RawValue>>\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset\n FROM (\n SELECT $2::uuid as job_id, $1::text as workspace_id\n ) base\n LEFT JOIN v2_job_completed jc ON jc.id = base.job_id AND jc.workspace_id = base.workspace_id\n LEFT JOIN job_result_stream rs ON rs.job_id = base.job_id\n WHERE base.job_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "result: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 1,
"name": "result_stream: Option<String>",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "stream_offset",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Uuid",
"Int4"
]
},
"nullable": [
null,
null,
null
]
},
"hash": "d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d"
}
backend/.sqlx/query-e07660e8d2a265cb6a83f3a2bb8e7e6330f09ab116e9837f6f16f8fdef938004.json
Generated
-18
View File
@@ -1,18 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH job_result AS (\n SELECT result \n FROM v2_job_completed \n WHERE id = $1\n ),\n updated_queue AS (\n UPDATE v2_job_queue\n SET running = false,\n tag = COALESCE($3, tag)\n WHERE id = $2\n )\n UPDATE v2_job \n SET \n tag = COALESCE($3, tag),\n concurrent_limit = COALESCE($4, concurrent_limit),\n concurrency_time_window_s = COALESCE($5, concurrency_time_window_s),\n args = COALESCE(\n CASE \n WHEN job_result.result IS NULL THEN NULL\n WHEN jsonb_typeof(job_result.result) = 'object' \n THEN job_result.result\n WHEN jsonb_typeof(job_result.result) = 'null'\n THEN NULL\n ELSE jsonb_build_object('value', job_result.result)\n END, \n '{}'::jsonb\n ),\n preprocessed = TRUE\n FROM job_result\n WHERE v2_job.id = $2;\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Uuid",
"Varchar",
"Int4",
"Int4"
]
},
"nullable": []
},
"hash": "e07660e8d2a265cb6a83f3a2bb8e7e6330f09ab116e9837f6f16f8fdef938004"
}
backend/.sqlx/query-e7348225a27bbdc9607d7c799e7192cd6ce4088467d91a5cbdc019430320d26d.json → backend/.sqlx/query-e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n value->'preprocessor_module'->'value' as \"preprocessor_module: _\",\n schema as \"schema: _\"\n FROM flow_version\n WHERE \n path = $1\n AND workspace_id = $2\n ORDER BY created_at DESC\n LIMIT 1",
"query": "SELECT\n value->'preprocessor_module'->'value' as \"preprocessor_module: _\",\n schema as \"schema: _\"\n FROM flow_version\n WHERE\n path = $1\n AND workspace_id = $2\n ORDER BY created_at DESC\n LIMIT 1",
"describe": {
"columns": [
{
@@ -25,5 +25,5 @@
true
]
},
"hash": "e7348225a27bbdc9607d7c799e7192cd6ce4088467d91a5cbdc019430320d26d"
"hash": "e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d"
}
backend/Cargo.lock
Generated
+1
View File
@@ -15356,6 +15356,7 @@ dependencies = [
"anyhow",
"async-recursion",
"async-stream",
"async-trait",
"aws-config",
"aws-sdk-sts",
"aws-smithy-types-convert",
backend/Cargo.toml
+2 -2
View File
@@ -103,7 +103,7 @@ all_languages = ["python", "deno_core", "rust", "mysql", "oracledb", "duckdb", "
all_languages_windows = ["python", "deno_core", "rust", "mysql", "oracledb", "mssql", "bigquery", "csharp", "nu", "php", "java"]
[patch.crates-io]
object_store = { git = "https://github.com/apache/arrow-rs-object-store", rev = "36752c975d4f29e20b57c91f81a10872dcd48ae7" }
object_store = { git = "https://github.com/apache/arrow-rs-object-store", rev = "36752c975d4f29e20b57c91f81a10872dcd48ae7" }
[dependencies]
anyhow.workspace = true
@@ -359,7 +359,7 @@ nu-parser = { version = "0.101.0", default-features = false }
process-wrap = { version = "8.2.1", features = ["tokio1"] }
datafusion = "47.0.0"
object_store = { git = "https://github.com/apache/arrow-rs-object-store", rev = "36752c975d4f29e20b57c91f81a10872dcd48ae7", features = ["aws", "azure", "gcp"] }
object_store = { git = "https://github.com/apache/arrow-rs-object-store", rev = "36752c975d4f29e20b57c91f81a10872dcd48ae7", features = ["aws", "azure", "gcp"] }
openidconnect = { version = "4.0.0-rc.1" }
aws-config = "^1"
aws-sdk-sqs = "=1.77.0"
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
c409ac5e5bd202002648f18e8adb7cdf78b2a83e
ad0f06c836d050b3ddc1937ef3a09f6b49cf5191
backend/windmill-api/Cargo.toml
+2 -2
View File
@@ -139,7 +139,7 @@ thiserror = { workspace = true, optional = true }
rust_decimal = { workspace = true, optional = true }
rust-postgres-native-tls = { workspace = true, optional = true}
rumqttc = { workspace = true, optional = true }
aws-sdk-sqs = { workspace = true, optional = true }
aws-sdk-sqs = { workspace = true, optional = true }
aws-sdk-sso = { workspace = true, optional = true }
aws-sdk-ssooidc = { workspace = true, optional = true }
aws-sdk-sts = { workspace = true, optional = true }
@@ -156,4 +156,4 @@ deno_error = { workspace = true, optional = true }
deno_core = { workspace = true, optional = true }
backon = {workspace = true, optional = true}
[build-dependencies]
deno_core = { workspace = true, optional = true }
deno_core = { workspace = true, optional = true }
backend/windmill-api/src/auth.rs
+1 -1
View File
@@ -31,8 +31,8 @@ use windmill_common::{
lazy_static::lazy_static! {
// Global auth cache accessible from main.rs for direct invalidation
pub static ref AUTH_CACHE: Cache<(String, String), ExpiringAuthCache> = Cache::new(300);
}
}
// Global function to invalidate a specific token from cache
pub fn invalidate_token_from_cache(token: &str) {
// Remove all cache entries for this token (across all workspaces)
backend/windmill-api/src/db.rs
+19 -4
View File
@@ -17,7 +17,7 @@ use tokio::task::JoinHandle;
use windmill_audit::audit_oss::{AuditAuthor, AuditAuthorable};
pub use windmill_common::db::DB;
use windmill_common::{
db::{Authable, Authed},
db::{Authable, Authed, AuthedRef},
error::Error,
utils::generate_lock_id,
};
@@ -48,7 +48,7 @@ lazy_static::lazy_static! {
(20250429211554, include_str!(
"../../migrations/20250429211554_create_indices_on_queue.up.sql"
).replace("public.", "")),
(20241006144414, include_str!(
(20241006144414, include_str!(
"../../custom_migrations/grant_all_current_schema.sql"
).to_string()),
(20221105003256, "DELETE FROM workspace_invite WHERE workspace_id = 'demo' AND email = 'ruben@windmill.dev';".to_string()),
@@ -201,8 +201,8 @@ pub async fn migrate(db: &DB) -> Result<Option<JoinHandle<()>>, Error> {
let mut custom_migrator = CustomMigrator { inner: migrator };
if let Err(err) = sqlx::query!(
"DELETE FROM _sqlx_migrations WHERE
version=20250131115248 OR version=20250902085503 OR version=20250201145630 OR
"DELETE FROM _sqlx_migrations WHERE
version=20250131115248 OR version=20250902085503 OR version=20250201145630 OR
version=20250201145631 OR version=20250201145632"
)
.execute(db)
@@ -243,6 +243,21 @@ pub struct ApiAuthed {
pub token_prefix: Option<String>,
}
impl ApiAuthed {
pub fn to_authed_ref<'e>(&'e self) -> AuthedRef<'e> {
AuthedRef {
email: &self.email,
username: &self.username,
is_admin: &self.is_admin,
is_operator: &self.is_operator,
groups: &self.groups,
folders: &self.folders,
scopes: &self.scopes,
token_prefix: &self.token_prefix,
}
}
}
impl From<ApiAuthed> for Authed {
fn from(value: ApiAuthed) -> Self {
Self {
backend/windmill-api/src/jobs.rs
+140 -71
View File
@@ -28,6 +28,7 @@ use tower::ServiceBuilder;
#[cfg(all(feature = "enterprise", feature = "smtp"))]
use windmill_common::auth::is_super_admin_email;
use windmill_common::auth::TOKEN_PREFIX_LEN;
use windmill_common::db::UserDbWithAuthed;
use windmill_common::error::JsonResult;
use windmill_common::flow_status::{JobResult, RestartedFrom};
use windmill_common::jobs::{
@@ -177,14 +178,20 @@ pub fn workspaced_service() -> Router {
.layer(ce_headers.clone()),
)
.route("/run/preview", post(run_preview_script))
.route("/run_wait_result/preview", post(run_wait_result_preview_script))
.route(
"/run_wait_result/preview",
post(run_wait_result_preview_script),
)
.route(
"/run/preview_bundle",
post(run_bundle_preview_script).layer(axum::extract::DefaultBodyLimit::disable()),
)
.route("/add_batch_jobs/:n", post(add_batch_jobs))
.route("/run/preview_flow", post(run_preview_flow_job))
.route("/run_wait_result/preview_flow", post(run_wait_result_preview_flow))
.route(
"/run_wait_result/preview_flow",
post(run_wait_result_preview_flow),
)
.route("/list", get(list_jobs))
.route(
"/list_selected_job_groups",
@@ -274,7 +281,10 @@ pub fn workspace_unauthed_service() -> Router {
.route("/get_root_job_id/:id", get(get_root_job))
.route("/get/:id", get(get_job))
.route("/get_logs/:id", get(get_job_logs))
.route("/get_completed_logs_tail/:id", get(get_completed_job_logs_tail))
.route(
"/get_completed_logs_tail/:id",
get(get_completed_job_logs_tail),
)
.route("/get_args/:id", get(get_args))
.route("/get_flow_debug_info/:id", get(get_flow_job_debug_info))
.route("/completed/get/:id", get(get_completed_job))
@@ -1105,7 +1115,7 @@ async fn send_workspace_trigger_failure_email_notification(
<div class="section">
<span class="label">Trigger path:</span> {}
</div>
<div class="section">
<span class="label">Trigger Type:</span> {}
</div>"#,
@@ -1147,7 +1157,7 @@ async fn send_workspace_trigger_failure_email_notification(
</head>
<body>
<h1>{}</h1>
<div class="section">
<span class="label">Workspace:</span> {}
</div>
@@ -1155,27 +1165,21 @@ async fn send_workspace_trigger_failure_email_notification(
<div class="section">
<span class="label">Script/Flow Path:</span> {}
</div>
<div class="section">
<span class="label">Job ID:</span> {}
</div>
<div class="section">
<span class="label">Error Details:</span>
<pre>{}</pre>
</div>
<a href="{}" class="button">View Job Details</a>
</body>
</html>
"#,
email_title,
w_id,
trigger_info,
runnable_path,
&job_id,
error_details,
job_url
email_title, w_id, trigger_info, runnable_path, &job_id, error_details, job_url
);
if let Err(e) = send_email_html(
@@ -1364,7 +1368,6 @@ async fn get_completed_job_logs_tail(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::JsonResult<String> {
let tags = opt_authed
.as_ref()
.map(|authed| get_scope_tags(authed).map(|v| v.iter().map(|s| s.to_string()).collect_vec()))
@@ -1390,7 +1393,7 @@ async fn get_completed_job_logs_tail(
"As a non logged in user, you can only see jobs ran by anonymous users".to_string(),
));
}
let logs = record.logs.unwrap_or_default();
Ok(Json(logs))
} else {
@@ -3814,7 +3817,7 @@ pub async fn run_flow_by_path_inner(
let flow_path = flow_path.to_path();
check_scopes(&authed, || format!("jobs:run:flows:{flow_path}"))?;
let mut tx = user_db.clone().begin(&authed).await?;
let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let FlowVersionInfo {
version,
@@ -3825,9 +3828,15 @@ pub async fn run_flow_by_path_inner(
edited_by,
early_return,
..
} = get_latest_flow_version_info_for_path(&mut *tx, &w_id, &flow_path, true).await?;
drop(tx);
} = get_latest_flow_version_info_for_path(
Some(userdb_authed),
&db,
db.clone(),
&w_id,
&flow_path,
true,
)
.await?;
let tag = run_query.tag.clone().or(tag);
@@ -3882,7 +3891,7 @@ pub async fn run_flow_by_path_inner(
None,
None,
push_authed.as_ref(),
false
false,
)
.await?;
tx.commit().await?;
@@ -3978,7 +3987,7 @@ pub async fn restart_flow(
None,
completed_job.priority,
Some(&authed.clone().into()),
false
false,
)
.await?;
tx.commit().await?;
@@ -4024,10 +4033,15 @@ pub async fn run_script_by_path_inner(
let script_path = script_path.to_path();
check_scopes(&authed, || format!("jobs:run:scripts:{script_path}"))?;
let mut tx = user_db.clone().begin(&authed).await?;
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) =
script_path_to_payload(script_path, &mut *tx, &w_id, run_query.skip_preprocessor).await?;
drop(tx);
let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) = script_path_to_payload(
script_path,
Some(userdb_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let tag = run_query.tag.clone().or(tag);
@@ -4148,10 +4162,12 @@ pub async fn run_workflow_as_code(
None,
),
JobKind::Script => {
let mut tx = user_db.clone().begin(&authed).await?;
let userdb_authed =
UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
script_path_to_payload(
job.script_path(),
&mut *tx,
Some(userdb_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
@@ -4707,10 +4723,16 @@ pub async fn run_wait_result_job_by_path_get(
check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
let mut tx = user_db.clone().begin(&authed).await?;
let (job_payload, tag, delete_after_use, timeout, on_behalf_authed) =
script_path_to_payload(script_path, &mut *tx, &w_id, run_query.skip_preprocessor).await?;
drop(tx);
let user_db_with_authed =
UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let (job_payload, tag, delete_after_use, timeout, on_behalf_authed) = script_path_to_payload(
script_path,
Some(user_db_with_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
@@ -4854,15 +4876,15 @@ pub async fn run_wait_result_script_by_path_internal(
) -> error::Result<Response> {
check_queue_too_long(&db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?;
let mut tx = user_db.clone().begin(&authed).await?;
let db_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) = script_path_to_payload(
script_path.to_path(),
&mut *tx,
Some(db_authed),
db.clone(),
&w_id,
run_query.skip_preprocessor,
)
.await?;
drop(tx);
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
@@ -4945,7 +4967,7 @@ pub async fn run_wait_result_script_by_hash(
check_queue_too_long(&db, run_query.queue_limit).await?;
let hash = script_hash.0;
let mut tx = user_db.clone().begin(&authed).await?;
let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let ScriptHashInfo {
path,
tag,
@@ -4962,7 +4984,7 @@ pub async fn run_wait_result_script_by_hash(
on_behalf_of_email,
created_by,
..
} = get_script_info_for_hash(&mut *tx, &w_id, hash).await?;
} = get_script_info_for_hash(Some(userdb_authed), &db, &w_id, hash).await?;
if let Some(run_query_cache_ttl) = run_query.cache_ttl {
cache_ttl = Some(run_query_cache_ttl);
}
@@ -5080,7 +5102,7 @@ pub async fn run_wait_result_flow_by_path_internal(
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let mut tx = user_db.clone().begin(&authed).await?;
let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let FlowVersionInfo {
tag,
@@ -5090,8 +5112,15 @@ pub async fn run_wait_result_flow_by_path_internal(
on_behalf_of_email,
edited_by,
version,
} = get_latest_flow_version_info_for_path(&mut *tx, &w_id, &flow_path, true).await?;
drop(tx);
} = get_latest_flow_version_info_for_path(
Some(userdb_authed),
&db,
db.clone(),
&w_id,
&flow_path,
true,
)
.await?;
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;
@@ -5231,16 +5260,18 @@ async fn run_wait_result_preview_script(
Query(run_query): Query<RunJobQuery>,
Json(preview): Json<Preview>,
) -> error::Result<Response> {
let (_status_code, uuid) = run_preview_script(
authed.clone(),
Extension(db.clone()),
Extension(user_db.clone()),
Path(w_id.clone()),
Query(run_query.clone()),
Json(preview)
).await?;
let uuid = uuid.parse::<Uuid>().map_err(|_| Error::BadRequest("Invalid UUID".to_string()))?;
authed.clone(),
Extension(db.clone()),
Extension(user_db.clone()),
Path(w_id.clone()),
Query(run_query.clone()),
Json(preview),
)
.await?;
let uuid = uuid
.parse::<Uuid>()
.map_err(|_| Error::BadRequest("Invalid UUID".to_string()))?;
let result = run_wait_result(&db, uuid, w_id, None, &authed.username).await;
return result;
}
@@ -5613,7 +5644,8 @@ async fn add_batch_jobs(
) = match batch_info.kind.as_str() {
"script" => {
if let Some(path) = batch_info.path {
let mut tx = user_db.clone().begin(&authed).await?;
let db_authed =
UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let ScriptHashInfo {
hash: script_hash,
concurrency_key,
@@ -5623,7 +5655,7 @@ async fn add_batch_jobs(
dedicated_worker,
timeout,
.. // TODO: consider on_behalf_of_email and created_by for batch jobs
} = get_latest_deployed_hash_for_path(&mut *tx, &w_id, &path).await?;
} = get_latest_deployed_hash_for_path(Some(db_authed), db.clone(), &w_id, &path).await?;
(
Some(script_hash),
Some(path),
@@ -5919,8 +5951,18 @@ async fn run_wait_result_preview_flow(
Query(run_query): Query<RunJobQuery>,
Json(raw_flow): Json<PreviewFlow>,
) -> error::Result<Response> {
let (_status_code, uuid) = run_preview_flow_job(authed.clone(), Extension(db.clone()), Extension(user_db.clone()), Path(w_id.clone()), Query(run_query.clone()), Json(raw_flow)).await?;
let uuid = uuid.parse::<Uuid>().map_err(|_| Error::BadRequest("Invalid UUID".to_string()))?;
let (_status_code, uuid) = run_preview_flow_job(
authed.clone(),
Extension(db.clone()),
Extension(user_db.clone()),
Path(w_id.clone()),
Query(run_query.clone()),
Json(raw_flow),
)
.await?;
let uuid = uuid
.parse::<Uuid>()
.map_err(|_| Error::BadRequest("Invalid UUID".to_string()))?;
let result = run_wait_result(&db, uuid, w_id, None, &authed.username).await;
return result;
}
@@ -5962,7 +6004,7 @@ pub async fn run_job_by_hash_inner(
check_license_key_valid().await?;
let hash = script_hash.0;
let mut tx = user_db.clone().begin(&authed).await?;
let userdb_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
let ScriptHashInfo {
path,
tag,
@@ -5979,7 +6021,7 @@ pub async fn run_job_by_hash_inner(
created_by,
delete_after_use,
..
} = get_script_info_for_hash(&mut *tx, &w_id, hash).await?;
} = get_script_info_for_hash(Some(userdb_authed), &db, &w_id, hash).await?;
check_scopes(&authed, || format!("jobs:run:scripts:{path}"))?;
if let Some(run_query_cache_ttl) = run_query.cache_ttl {
@@ -6172,9 +6214,15 @@ async fn get_job_update(
opt_tokened: OptTokened,
Extension(db): Extension<DB>,
Path((w_id, job_id)): Path<(String, Uuid)>,
Query(JobUpdateQuery { log_offset, stream_offset, get_progress, running, only_result, no_logs, .. }): Query<
JobUpdateQuery,
>,
Query(JobUpdateQuery {
log_offset,
stream_offset,
get_progress,
running,
only_result,
no_logs,
..
}): Query<JobUpdateQuery>,
) -> JsonResult<JobUpdate> {
Ok(Json(
get_job_update_data(
@@ -6201,9 +6249,15 @@ async fn get_job_update_sse(
opt_tokened: OptTokened,
Extension(db): Extension<DB>,
Path((w_id, job_id)): Path<(String, Uuid)>,
Query(JobUpdateQuery { log_offset, stream_offset, get_progress, running, no_logs, only_result, fast }): Query<
JobUpdateQuery,
>,
Query(JobUpdateQuery {
log_offset,
stream_offset,
get_progress,
running,
no_logs,
only_result,
fast,
}): Query<JobUpdateQuery>,
) -> Response {
let stream = get_job_update_sse_stream(
opt_authed,
@@ -6477,7 +6531,7 @@ async fn get_job_update_data(
if only_result.unwrap_or(false) {
let result = if let Some(tags) = tags {
let r =
let r =
sqlx::query!(
"SELECT result as \"result: sqlx::types::Json<Box<RawValue>>\", v2_job.tag,
v2_job_queue.running as \"running: Option<bool>\", SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\", CHAR_LENGTH(rs.stream) AS stream_offset
@@ -6502,11 +6556,16 @@ async fn get_job_update_data(
)));
}
let running = r.running.as_ref().map(|x| *x);
(r.result.map(|x| x.0), running, r.result_stream.flatten(), r.stream_offset)
(
r.result.map(|x| x.0),
running,
r.result_stream.flatten(),
r.stream_offset,
)
} else {
if running.is_some_and(|x| !x) {
let r = sqlx::query!(
"SELECT
"SELECT
COALESCE(jc.result, jc.result) as \"result: sqlx::types::Json<Box<RawValue>>\",
jq.running as \"running: Option<bool>\",
SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",
@@ -6524,15 +6583,20 @@ async fn get_job_update_data(
).fetch_optional(db).await?;
if let Some(r) = r {
let running = r.running.as_ref().map(|x| *x);
(r.result.map(|x| x.0), running, r.result_stream.flatten(), r.stream_offset)
(
r.result.map(|x| x.0),
running,
r.result_stream.flatten(),
r.stream_offset,
)
} else {
(None, None, None, None)
}
} else {
let q = sqlx::query!(
"SELECT
COALESCE(jc.result, NULL) as \"result: sqlx::types::Json<Box<RawValue>>\",
SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",
"SELECT
COALESCE(jc.result, NULL) as \"result: sqlx::types::Json<Box<RawValue>>\",
SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",
CHAR_LENGTH(rs.stream) + 1 AS stream_offset
FROM (
SELECT $2::uuid as job_id, $1::text as workspace_id
@@ -6547,7 +6611,12 @@ async fn get_job_update_data(
.fetch_optional(db)
.await?;
if let Some(r) = q {
(r.result.map(|x| x.0), running, r.result_stream.flatten(), r.stream_offset)
(
r.result.map(|x| x.0),
running,
r.result_stream.flatten(),
r.stream_offset,
)
} else {
(None, None, None, None)
}
@@ -6571,7 +6640,7 @@ async fn get_job_update_data(
let record = sqlx::query!(
"SELECT
c.id IS NOT NULL AS completed,
CASE
CASE
WHEN q.id IS NOT NULL THEN (CASE WHEN NOT $5 AND q.running THEN true ELSE null END)
ELSE false
END AS running,
backend/windmill-api/src/trigger_helpers.rs
+29 -9
View File
@@ -6,7 +6,7 @@ use serde_json::value::RawValue;
use std::collections::HashMap;
use uuid::Uuid;
use windmill_common::{
db::UserDB,
db::{UserDB, UserDbWithAuthed},
error::Result,
flows::{FlowModuleValue, Retry},
get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path,
@@ -89,7 +89,8 @@ impl ScriptId {
async fn get_script_hash(self, workspace_id: &str, db: &DB) -> Result<i64> {
let hash = match self {
ScriptId::ScriptPath(path) => {
let info = get_latest_deployed_hash_for_path(db, workspace_id, &path).await?;
let info = get_latest_deployed_hash_for_path(None, db.clone(), workspace_id, &path)
.await?;
info.hash
}
ScriptId::ScriptHash(hash) => hash.0,
@@ -251,8 +252,15 @@ pub async fn get_runnable_format(
)
}
RunnableId::FlowPath(path) => {
let FlowVersionInfo { version, .. } =
get_latest_flow_version_info_for_path(db, workspace_id, &path, true).await?;
let FlowVersionInfo { version, .. } = get_latest_flow_version_info_for_path(
None,
db,
db.clone(),
workspace_id,
&path,
true,
)
.await?;
let key = (
HubOrWorkspaceId::WorkspaceId(workspace_id.to_string()),
@@ -273,7 +281,7 @@ pub async fn get_runnable_format(
value->'preprocessor_module'->'value' as \"preprocessor_module: _\",
schema as \"schema: _\"
FROM flow_version
WHERE
WHERE
path = $1
AND workspace_id = $2
ORDER BY created_at DESC
@@ -293,8 +301,13 @@ pub async fn get_runnable_format(
let hash = if let Some(hash) = hash {
hash.0
} else {
let script_hash =
get_latest_deployed_hash_for_path(db, workspace_id, &path).await?;
let script_hash = get_latest_deployed_hash_for_path(
None,
db.clone(),
workspace_id,
&path,
)
.await?;
script_hash.hash
};
let script_info = get_script_info(db, workspace_id, hash).await?;
@@ -701,8 +714,15 @@ async fn trigger_script_with_retry_and_error_handler(
let error_handler_args = error_handler_args.map(|args| args.0.clone());
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) = {
let mut tx = user_db.clone().begin(&authed).await?;
script_path_to_payload(script_path, &mut *tx, &workspace_id, Some(false)).await?
let db_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
script_path_to_payload(
script_path,
Some(db_authed),
db.clone(),
&workspace_id,
Some(false),
)
.await?
};
check_tag_available_for_workspace(&db, &workspace_id, &tag, &authed).await?;
backend/windmill-api/src/triggers/trigger_helpers.rs
+29 -9
View File
@@ -8,7 +8,7 @@ use std::collections::HashMap;
use std::future::Future;
use uuid::Uuid;
use windmill_common::{
db::UserDB,
db::{UserDB, UserDbWithAuthed},
error::Result,
flows::{FlowModuleValue, Retry},
get_latest_deployed_hash_for_path, get_latest_flow_version_info_for_path,
@@ -89,7 +89,8 @@ impl ScriptId {
async fn get_script_hash(self, workspace_id: &str, db: &DB) -> Result<i64> {
let hash = match self {
ScriptId::ScriptPath(path) => {
let info = get_latest_deployed_hash_for_path(db, workspace_id, &path).await?;
let info = get_latest_deployed_hash_for_path(None, db.clone(), workspace_id, &path)
.await?;
info.hash
}
ScriptId::ScriptHash(hash) => hash.0,
@@ -251,8 +252,15 @@ pub async fn get_runnable_format(
)
}
RunnableId::FlowPath(path) => {
let FlowVersionInfo { version, .. } =
get_latest_flow_version_info_for_path(db, workspace_id, &path, true).await?;
let FlowVersionInfo { version, .. } = get_latest_flow_version_info_for_path(
None,
db,
db.clone(),
workspace_id,
&path,
true,
)
.await?;
let key = (
HubOrWorkspaceId::WorkspaceId(workspace_id.to_string()),
@@ -273,7 +281,7 @@ pub async fn get_runnable_format(
value->'preprocessor_module'->'value' as \"preprocessor_module: _\",
schema as \"schema: _\"
FROM flow_version
WHERE
WHERE
path = $1
AND workspace_id = $2
ORDER BY created_at DESC
@@ -293,8 +301,13 @@ pub async fn get_runnable_format(
let hash = if let Some(hash) = hash {
hash.0
} else {
let script_hash =
get_latest_deployed_hash_for_path(db, workspace_id, &path).await?;
let script_hash = get_latest_deployed_hash_for_path(
None,
db.clone(),
workspace_id,
&path,
)
.await?;
script_hash.hash
};
let script_info = get_script_info(db, workspace_id, hash).await?;
@@ -717,8 +730,15 @@ async fn trigger_script_with_retry_and_error_handler(
let error_handler_args = error_handler_args.map(|args| args.0.clone());
let (job_payload, tag, delete_after_use, timeout, on_behalf_of) = {
let mut tx = user_db.clone().begin(&authed).await?;
script_path_to_payload(script_path, &mut *tx, &workspace_id, Some(false)).await?
let db_authed = UserDbWithAuthed { db: user_db.clone(), authed: &authed.to_authed_ref() };
script_path_to_payload(
script_path,
Some(db_authed),
db.clone(),
&workspace_id,
Some(false),
)
.await?
};
check_tag_available_for_workspace(&db, &workspace_id, &tag, &authed).await?;
backend/windmill-common/Cargo.toml
+2 -1
View File
@@ -15,7 +15,7 @@ loki = ["dep:tracing-loki"]
benchmark = []
parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts", "dep:aws-smithy-types-convert", "dep:datafusion"]
aws_auth = ["dep:aws-sdk-sts", "dep:aws-config"]
otel = ["dep:opentelemetry-semantic-conventions", "dep:opentelemetry-otlp", "dep:opentelemetry_sdk",
otel = ["dep:opentelemetry-semantic-conventions", "dep:opentelemetry-otlp", "dep:opentelemetry_sdk",
"dep:opentelemetry", "dep:tracing-opentelemetry", "dep:opentelemetry-appender-tracing", "dep:tonic"]
smtp = ["dep:mail-send"]
scoped_cache = []
@@ -36,6 +36,7 @@ serde_json.workspace = true
chrono.workspace = true
chrono-tz.workspace = true
hex.workspace = true
async-trait.workspace = true
reqwest-middleware = { workspace = true }
reqwest-retry = { workspace = true }
rand.workspace = true
backend/windmill-common/src/auth.rs
+76 -1
View File
@@ -1,10 +1,16 @@
use std::{
hash::DefaultHasher,
sync::atomic::{AtomicI64, Ordering},
};
use anyhow::Context;
use chrono::{DateTime, Duration, Utc};
use quick_cache::sync::Cache;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::{
db::Authed,
db::{Authed, AuthedRef},
error::{Error, Result},
jwt,
users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SYNC_EMAIL},
@@ -20,6 +26,75 @@ pub struct IdToken {
pub const TOKEN_PREFIX_LEN: usize = 10;
lazy_static::lazy_static! {
// Cache for script hash permissions - (ApiAuthed hash, script_hash) -> permission result
pub static ref HASH_PERMS_CACHE: PermsCache = PermsCache::new();
pub static ref FLOW_PERMS_CACHE: PermsCache = PermsCache::new();
}
pub struct PermsCache(Cache<(u64, u64), ()>, AtomicI64);
use std::hash::Hash;
use std::hash::Hasher;
impl PermsCache {
pub fn compute_hash(authed: &AuthedRef) -> u64 {
let mut hasher = DefaultHasher::new();
authed.username.hash(&mut hasher);
authed.folders.hash(&mut hasher);
authed.groups.hash(&mut hasher);
authed.is_admin.hash(&mut hasher);
hasher.finish()
}
}
pub const PERMS_CACHE_EXPIRATION_SECONDS: i64 = 60 * 60;
impl PermsCache {
pub fn new() -> Self {
PermsCache(
Cache::new(10000),
AtomicI64::new(chrono::Utc::now().timestamp() as i64),
)
}
pub fn check_perms_in_cache<'e, T: Into<u64>>(
&self,
authed: &'e AuthedRef<'e>,
key: T,
) -> (bool, u64) {
// Clear cache every hour
if self.1.load(Ordering::Relaxed)
< chrono::Utc::now().timestamp() - PERMS_CACHE_EXPIRATION_SECONDS
{
self.0.clear();
self.1
.store(chrono::Utc::now().timestamp() as i64, Ordering::Relaxed);
}
// Create hash of the ApiAuthed struct for caching
let authed_hash = Self::compute_hash(authed);
let key = key.into();
tracing::debug!(
"Checking cache for authed hash {authed_hash} and script hash {}",
key
);
// Check cache first
if let Some(_) = self.0.get(&(authed_hash, key)) {
tracing::debug!("Cached result for authed hash {authed_hash}",);
return (true, authed_hash);
}
return (false, authed_hash);
}
pub fn insert<'e, T: Into<u64>>(&self, authed_hash: u64, key: T) {
let key = key.into();
tracing::debug!("Inserting authed hash {authed_hash} and key {}", key);
self.0.insert((authed_hash, key), ());
}
}
pub fn has_expired(expiration_time: DateTime<Utc>, take: Option<Duration>) -> bool {
let now = Utc::now();
backend/windmill-common/src/db.rs
+59 -2
View File
@@ -1,8 +1,8 @@
use sqlx::{Pool, Postgres, Transaction};
use sqlx::{Acquire, Pool, Postgres, Transaction};
pub type DB = Pool<Postgres>;
#[derive(Clone, Debug)]
#[derive(Clone, Debug, Hash)]
pub struct Authed {
pub email: String,
pub username: String,
@@ -15,6 +15,43 @@ pub struct Authed {
pub token_prefix: Option<String>,
}
#[derive(Clone, Debug, Hash)]
pub struct AuthedRef<'a> {
pub email: &'a str,
pub username: &'a str,
pub is_admin: &'a bool,
pub is_operator: &'a bool,
pub groups: &'a Vec<String>,
// (folder name, can write, is owner)
pub folders: &'a Vec<(String, bool, bool)>,
pub scopes: &'a Option<Vec<String>>,
pub token_prefix: &'a Option<String>,
}
impl Authable for AuthedRef<'_> {
fn email(&self) -> &str {
self.email
}
fn username(&self) -> &str {
self.username
}
fn is_admin(&self) -> bool {
*self.is_admin
}
fn is_operator(&self) -> bool {
*self.is_operator
}
fn groups(&self) -> &[String] {
self.groups
}
fn folders(&self) -> &[(String, bool, bool)] {
self.folders
}
fn scopes(&self) -> Option<&[String]> {
self.scopes.as_ref().map(|x| x.as_slice())
}
}
#[derive(Clone)]
pub struct UserDB {
db: DB,
@@ -64,6 +101,26 @@ lazy_static::lazy_static! {
pub static ref PG_SCHEMA: Option<String> = std::env::var("PG_SCHEMA").ok();
}
pub struct UserDbWithAuthed<'c, T: Authable + Sync> {
pub authed: &'c T,
pub db: UserDB,
}
impl<'c, 'd, T: Authable + Sync> Acquire<'c> for &'c UserDbWithAuthed<'d, T> {
type Database = Postgres;
type Connection = Transaction<'c, Postgres>;
fn acquire(self) -> futures_core::future::BoxFuture<'c, Result<Self::Connection, sqlx::Error>> {
Box::pin(async move { self.db.clone().begin(self.authed).await })
}
fn begin(
self,
) -> futures_core::future::BoxFuture<'c, Result<Transaction<'c, Postgres>, sqlx::Error>> {
Box::pin(async move { self.db.clone().begin(self.authed).await })
}
}
impl UserDB {
pub fn new(db: DB) -> Self {
Self { db }
backend/windmill-common/src/jobs.rs
+15 -7
View File
@@ -5,7 +5,7 @@ use futures_core::Stream;
use indexmap::IndexMap;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sqlx::{types::Json, Postgres};
use sqlx::types::Json;
use tokio::io::AsyncReadExt;
use uuid::Uuid;
@@ -17,7 +17,7 @@ pub const EMAIL_ERROR_HANDLER_USER_EMAIL: &str = "email_error_handler@windmill.d
use crate::{
apps::AppScriptId,
auth::is_super_admin_email,
db::DB,
db::{AuthedRef, UserDbWithAuthed, DB},
error::{self, to_anyhow, Error},
flow_status::{FlowStatus, RestartedFrom},
flows::{FlowNodeId, FlowValue, Retry},
@@ -459,9 +459,10 @@ pub fn get_has_preprocessor_from_content_and_lang(
Ok(has_preprocessor)
}
pub async fn script_path_to_payload<'e, A: sqlx::Acquire<'e, Database = Postgres> + Send>(
pub async fn script_path_to_payload<'e>(
script_path: &str,
db: A,
db_authed: Option<UserDbWithAuthed<'e, AuthedRef<'e>>>,
db: DB,
w_id: &str,
skip_preprocessor: Option<bool>,
) -> error::Result<(
@@ -508,7 +509,7 @@ pub async fn script_path_to_payload<'e, A: sqlx::Acquire<'e, Database = Postgres
on_behalf_of_email,
created_by,
..
} = get_latest_deployed_hash_for_path(db, w_id, script_path).await?;
} = get_latest_deployed_hash_for_path(db_authed, db, w_id, script_path).await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf {
@@ -554,11 +555,18 @@ pub async fn get_payload_tag_from_prefixed_path(
w_id: &str,
) -> Result<(JobPayload, Option<String>, Option<OnBehalfOf>), Error> {
let (payload, tag, _, _, on_behalf_of) = if path.starts_with("script/") {
script_path_to_payload(path.strip_prefix("script/").unwrap(), db, w_id, Some(true)).await?
script_path_to_payload(
path.strip_prefix("script/").unwrap(),
None,
db.clone(),
w_id,
Some(true),
)
.await?
} else if path.starts_with("flow/") {
let path = path.strip_prefix("flow/").unwrap().to_string();
let FlowVersionInfo { dedicated_worker, tag, version, .. } =
get_latest_flow_version_info_for_path(db, w_id, &path, true).await?;
get_latest_flow_version_info_for_path(None, db, db.clone(), w_id, &path, true).await?;
(
JobPayload::Flow { path, dedicated_worker, apply_preprocessor: false, version },
tag,
backend/windmill-common/src/lib.rs
+190 -67
View File
@@ -9,6 +9,7 @@
use quick_cache::sync::Cache;
use std::{
future::Future,
hash::{Hash, Hasher},
net::SocketAddr,
str::FromStr,
sync::{
@@ -22,7 +23,7 @@ use tokio::sync::broadcast;
use ee_oss::CriticalErrorChannel;
use error::Error;
use scripts::ScriptLang;
use sqlx::Postgres;
use sqlx::{Acquire, Postgres};
pub mod agent_workers;
pub mod apps;
@@ -51,19 +52,19 @@ pub mod job_s3_helpers_ee;
#[cfg(feature = "parquet")]
pub mod job_s3_helpers_oss;
#[cfg(all(feature = "enterprise", feature = "openidconnect", feature = "private"))]
pub mod oidc_ee;
#[cfg(all(feature = "enterprise", feature = "openidconnect"))]
pub mod oidc_oss;
pub mod triggers;
pub mod jobs;
pub mod jwt;
pub mod more_serde;
pub mod oauth2;
#[cfg(all(feature = "enterprise", feature = "openidconnect", feature = "private"))]
pub mod oidc_ee;
#[cfg(all(feature = "enterprise", feature = "openidconnect"))]
pub mod oidc_oss;
#[cfg(feature = "private")]
pub mod otel_ee;
pub mod otel_oss;
pub mod queue;
pub mod result_stream;
pub mod s3_helpers;
pub mod schedule;
pub mod schema;
@@ -72,17 +73,17 @@ pub mod server;
#[cfg(feature = "private")]
pub mod stats_ee;
pub mod stats_oss;
pub mod stream;
#[cfg(feature = "private")]
pub mod teams_ee;
pub mod teams_oss;
pub mod tracing_init;
pub mod triggers;
pub mod users;
pub mod utils;
pub mod variables;
pub mod worker;
pub mod workspaces;
pub mod result_stream;
pub mod stream;
pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;
pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 5;
@@ -287,12 +288,16 @@ pub struct PostgresUrlComponents {
}
pub fn parse_postgres_url(url: &str) -> Result<PostgresUrlComponents, Error> {
let parsed_url = url::Url::parse(url).map_err(|_| Error::BadConfig("Invalid PostgreSQL URL".to_string()))?;
let parsed_url =
url::Url::parse(url).map_err(|_| Error::BadConfig("Invalid PostgreSQL URL".to_string()))?;
let scheme = parsed_url.scheme().to_string();
let username = parsed_url.username().to_string();
let password = parsed_url.password().map(|p| p.to_string());
let host = parsed_url.host_str().ok_or_else(|| Error::BadConfig("Missing host in PostgreSQL URL".to_string()))?.to_string();
let host = parsed_url
.host_str()
.ok_or_else(|| Error::BadConfig("Missing host in PostgreSQL URL".to_string()))?
.to_string();
let port = parsed_url.port();
let database = parsed_url.path().trim_start_matches('/').to_string();
let mut ssl_mode = None;
@@ -304,7 +309,11 @@ pub fn parse_postgres_url(url: &str) -> Result<PostgresUrlComponents, Error> {
Ok(PostgresUrlComponents {
scheme,
username: if username.is_empty() { None } else { Some(username) },
username: if username.is_empty() {
None
} else {
Some(username)
},
password,
host,
port,
@@ -377,8 +386,8 @@ pub async fn connect(
max_connections: u32,
worker_mode: bool,
) -> Result<sqlx::Pool<sqlx::Postgres>, error::Error> {
use std::time::Duration;
use sqlx::Executor;
use std::time::Duration;
sqlx::postgres::PgPoolOptions::new()
.min_connections((max_connections / 5).clamp(3, max_connections))
.max_connections(max_connections)
@@ -386,30 +395,39 @@ pub async fn connect(
.after_connect(move |conn, _| {
if worker_mode {
Box::pin(async move {
if let Err(e) = conn.execute(r#"
if let Err(e) = conn
.execute(
r#"
SET enable_seqscan = OFF;
SET statement_timeout = '5min';
SET idle_in_transaction_session_timeout = '10min';
SET tcp_keepalives_idle = 300;
SET tcp_keepalives_interval = 60;
SET tcp_keepalives_count = 10;"#)
.await {
tracing::error!("Error setting postgres settings: {}", e);
}
SET tcp_keepalives_count = 10;"#,
)
.await
{
tracing::error!("Error setting postgres settings: {}", e);
}
Ok(())
})
} else {
Box::pin(async move {
if let Err(e) = conn.execute(r#"
Box::pin(async move {
if let Err(e) = conn
.execute(
r#"
SET statement_timeout = '5min';
SET idle_in_transaction_session_timeout = '10min';
SET tcp_keepalives_idle = 300;
SET tcp_keepalives_interval = 60;
SET tcp_keepalives_count = 10;"#)
.await {
SET tcp_keepalives_count = 10;"#,
)
.await
{
tracing::error!("Error setting postgres settings: {}", e);
}
Ok(()) })
}
Ok(())
})
}
})
.connect_with(
@@ -423,6 +441,12 @@ type Tag = String;
pub use db::DB;
use crate::{
auth::{PermsCache, FLOW_PERMS_CACHE, HASH_PERMS_CACHE},
db::{AuthedRef, UserDbWithAuthed},
scripts::ScriptHash,
};
#[derive(Clone)]
pub struct ExpiringLatestVersionId {
id: i64,
@@ -448,21 +472,25 @@ pub struct ScriptHashInfo {
pub created_by: String,
}
pub fn get_latest_deployed_hash_for_path<
'a,
'e,
E: sqlx::Acquire<'e, Database = Postgres> + Send + 'a,
>(
db: E,
w_id: &'a str,
script_path: &'a str,
) -> impl Future<Output = error::Result<ScriptHashInfo>> + Send + 'a {
pub fn get_latest_deployed_hash_for_path<'e>(
db: Option<UserDbWithAuthed<'e, AuthedRef<'e>>>,
db2: DB,
w_id: &'e str,
script_path: &'e str,
) -> impl Future<Output = error::Result<ScriptHashInfo>> + Send + 'e {
async move {
let mut conn = db.acquire().await?;
let cache_key = (w_id.to_string(), script_path.to_string());
let mut computed_hash = None;
let hash = match DEPLOYED_SCRIPT_HASH_CACHE.get(&cache_key) {
Some(cached_hash) if cached_hash.expires_at > std::time::Instant::now() => {
Some(cached_hash)
if cached_hash.expires_at > std::time::Instant::now()
&& db.as_ref().is_none_or(|x| {
let r = HASH_PERMS_CACHE
.check_perms_in_cache(x.authed, ScriptHash(cached_hash.id));
computed_hash = Some(r.1);
return r.0;
}) =>
{
tracing::debug!(
"Using cached script hash {} for {script_path}",
cached_hash.id
@@ -471,16 +499,23 @@ pub fn get_latest_deployed_hash_for_path<
}
_ => {
tracing::debug!("Fetching script hash for {script_path}");
let hash = sqlx::query_scalar!(
"select hash from script where path = $1 AND workspace_id = $2 AND deleted = false AND lock IS not NULL AND lock_error_logs IS NULL ORDER BY created_at DESC LIMIT 1",
script_path,
w_id
)
.fetch_optional(&mut *conn)
.await?;
let hash = if let Some(db) = db {
let authed = db.authed;
let mut conn = db.acquire().await?;
let hash = get_latest_script_hash(&mut *conn, script_path, w_id).await?;
if let Some(hash) = hash {
HASH_PERMS_CACHE.insert(
computed_hash.unwrap_or_else(|| PermsCache::compute_hash(authed)),
ScriptHash(hash),
);
}
hash
} else {
let mut conn = db2.acquire().await?;
get_latest_script_hash(&mut *conn, script_path, w_id).await?
};
let hash = utils::not_found_if_none(hash, "script", script_path)?;
DEPLOYED_SCRIPT_HASH_CACHE.insert(
cache_key,
ExpiringLatestVersionId {
@@ -493,32 +528,60 @@ pub fn get_latest_deployed_hash_for_path<
}
};
get_script_info_for_hash(&mut *conn, w_id, hash).await
get_script_info_for_hash(None, &db2, w_id, hash).await
}
}
pub async fn get_latest_script_hash<'e, E: sqlx::PgExecutor<'e>>(
db: E,
script_path: &'e str,
w_id: &'e str,
) -> error::Result<Option<i64>> {
let hash = sqlx::query_scalar!(
"select hash from script where path = $1 AND workspace_id = $2 AND deleted = false AND lock IS not NULL AND lock_error_logs IS NULL ORDER BY created_at DESC LIMIT 1",
script_path,
w_id
)
.fetch_optional(db)
.await?;
return Ok(hash);
}
pub async fn get_script_info_for_hash<'e, E: sqlx::PgExecutor<'e>>(
db_authed: Option<UserDbWithAuthed<'e, AuthedRef<'e>>>,
db: E,
w_id: &str,
hash: i64,
) -> error::Result<ScriptHashInfo> {
let key = (w_id.to_string(), hash);
let mut computed_hash = None;
match DEPLOYED_SCRIPT_INFO_CACHE.get(&key) {
Some(info) => {
Some(info)
if db_authed.as_ref().is_none_or(|x| {
let r = HASH_PERMS_CACHE.check_perms_in_cache(x.authed, scripts::ScriptHash(hash));
computed_hash = Some(r.1);
return r.0;
}) =>
{
tracing::debug!("Using cached deployed script info for {hash}");
Ok(info)
}
_ => {
tracing::debug!("Fetching deployed script info for {hash}");
let info = sqlx::query_as!(
ScriptHashInfo,
"select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by, path from script where hash = $1 AND workspace_id = $2",
hash,
w_id
)
.fetch_optional(db)
.await?;
let info = if let Some(db_authed) = db_authed {
let mut conn = db_authed.acquire().await?;
let hash_info = get_script_info_for_hash_inner(&mut *conn, w_id, hash).await?;
if hash_info.is_some() {
HASH_PERMS_CACHE.insert(
computed_hash.unwrap_or_else(|| PermsCache::compute_hash(db_authed.authed)),
ScriptHash(hash),
);
}
hash_info
} else {
get_script_info_for_hash_inner(db, w_id, hash).await?
};
let info = utils::not_found_if_none(info, "script", &hash.to_string())?;
@@ -529,6 +592,21 @@ pub async fn get_script_info_for_hash<'e, E: sqlx::PgExecutor<'e>>(
}
}
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, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, delete_after_use, timeout, has_preprocessor, on_behalf_of_email, created_by, path from script where hash = $1 AND workspace_id = $2",
hash,
w_id
)
.fetch_optional(db)
.await?;
Ok(r)
}
#[derive(Clone)]
pub struct FlowVersionInfo {
pub version: i64,
@@ -540,44 +618,70 @@ pub struct FlowVersionInfo {
pub dedicated_worker: Option<bool>,
}
struct CachedFlowPath(String);
impl Into<u64> for CachedFlowPath {
fn into(self) -> u64 {
let mut hasher = std::collections::hash_map::DefaultHasher::new();
self.0.hash(&mut hasher);
hasher.finish()
}
}
pub fn get_latest_flow_version_info_for_path<
'a,
'e,
A: sqlx::Acquire<'e, Database = Postgres> + Send + 'a,
>(
db_authed: Option<UserDbWithAuthed<'e, AuthedRef<'e>>>,
db: A,
db2: DB,
w_id: &'a str,
path: &'a str,
use_cache: bool,
) -> impl Future<Output = error::Result<FlowVersionInfo>> + Send + 'a {
) -> impl Future<Output = error::Result<FlowVersionInfo>> + Send + 'a
where
'e: 'a,
{
// as instructed in the docstring of sqlx::Acquire
async move {
let mut conn = db.acquire().await?;
let cache_key = (w_id.to_string(), path.to_string());
let cached_version = if use_cache {
FLOW_VERSION_CACHE.get(&cache_key)
} else {
None
};
let mut computed_hash: Option<_> = None;
let version = match cached_version {
Some(cached_version) if cached_version.expires_at > std::time::Instant::now() => {
Some(cached_version)
if cached_version.expires_at > std::time::Instant::now()
&& db_authed.as_ref().is_none_or(|x| {
let r = FLOW_PERMS_CACHE
.check_perms_in_cache(x.authed, CachedFlowPath(path.to_string()));
computed_hash = Some(r.1);
return r.0;
}) =>
{
tracing::debug!("Using cached flow version {} for {path}", cached_version.id);
cached_version.id
}
_ => {
tracing::debug!("Fetching flow version for {path}");
let version = sqlx::query_scalar!(
"SELECT flow_version.id from flow
INNER JOIN flow_version
ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
WHERE flow.path = $1 and flow.workspace_id = $2",
path,
w_id
)
.fetch_optional(&mut *conn)
.await?;
let version = if let Some(db_authed) = db_authed {
let mut conn = db_authed.acquire().await?;
let r = get_latest_flow_version_for_path(&mut *conn, w_id, path).await?;
if r.is_some() {
FLOW_PERMS_CACHE.insert(
computed_hash
.unwrap_or_else(|| PermsCache::compute_hash(db_authed.authed)),
CachedFlowPath(path.to_string()),
);
}
r
} else {
let mut conn = db.acquire().await?;
get_latest_flow_version_for_path(&mut *conn, w_id, path).await?
};
let version = utils::not_found_if_none(version, "flow", path)?;
@@ -602,6 +706,7 @@ pub fn get_latest_flow_version_info_for_path<
}
_ => {
tracing::debug!("Fetching flow version info for {version} ({path})");
let mut conn = db2.acquire().await?;
let info = sqlx::query_as!(
FlowVersionInfo,
"SELECT tag, dedicated_worker, flow_version.value->>'early_return' as early_return, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor, on_behalf_of_email, edited_by, flow_version.id AS version
@@ -626,6 +731,24 @@ pub fn get_latest_flow_version_info_for_path<
}
}
async fn get_latest_flow_version_for_path<'e, E: sqlx::PgExecutor<'e>>(
db: E,
w_id: &str,
path: &str,
) -> error::Result<Option<i64>> {
let version = sqlx::query_scalar!(
"SELECT flow_version.id from flow
INNER JOIN flow_version
ON flow_version.id = flow.versions[array_upper(flow.versions, 1)]
WHERE flow.path = $1 and flow.workspace_id = $2",
path,
w_id
)
.fetch_optional(db)
.await?;
Ok(version)
}
pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>(
db: E,
w_id: &str,
backend/windmill-common/src/scripts.rs
+6
View File
@@ -131,6 +131,12 @@ impl FromStr for ScriptLang {
#[sqlx(transparent)]
pub struct ScriptHash(pub i64);
impl Into<u64> for ScriptHash {
fn into(self) -> u64 {
self.0 as u64
}
}
#[derive(PartialEq, sqlx::Type)]
#[sqlx(transparent, no_pg_array)]
pub struct ScriptHashes(pub Vec<i64>);
backend/windmill-queue/src/schedule.rs
+2
View File
@@ -120,7 +120,9 @@ pub async fn push_scheduled_job<'c>(
let FlowVersionInfo {
version, tag, dedicated_worker, on_behalf_of_email, edited_by, ..
} = get_latest_flow_version_info_for_path(
None,
&mut *tx,
db.clone(),
&schedule.workspace_id,
&schedule.script_path,
false,
backend/windmill-worker/src/worker_flow.rs
+19 -13
View File
@@ -1062,7 +1062,7 @@ pub async fn update_flow_status_after_job_completion_internal(
if require_args {
let args = sqlx::query_scalar!(
"SELECT result as \"result: Json<HashMap<String, Box<RawValue>>>\"
FROM v2_job_completed
FROM v2_job_completed
WHERE id = $1",
job_id_for_status
)
@@ -1109,8 +1109,8 @@ pub async fn update_flow_status_after_job_completion_internal(
// let concurrency_key = tag_and_concurrency_key.and_then(|tc| tc.concurrency_key.map(|ck| interpolate_args(&ck, &args, &workspace_id)));
sqlx::query!(
"WITH job_result AS (
SELECT result
FROM v2_job_completed
SELECT result
FROM v2_job_completed
WHERE id = $1
),
updated_queue AS (
@@ -1119,20 +1119,20 @@ pub async fn update_flow_status_after_job_completion_internal(
tag = COALESCE($3, tag)
WHERE id = $2
)
UPDATE v2_job
SET
UPDATE v2_job
SET
tag = COALESCE($3, tag),
concurrent_limit = COALESCE($4, concurrent_limit),
concurrency_time_window_s = COALESCE($5, concurrency_time_window_s),
args = COALESCE(
CASE
CASE
WHEN job_result.result IS NULL THEN NULL
WHEN jsonb_typeof(job_result.result) = 'object'
WHEN jsonb_typeof(job_result.result) = 'object'
THEN job_result.result
WHEN jsonb_typeof(job_result.result) = 'null'
THEN NULL
ELSE jsonb_build_object('value', job_result.result)
END,
END,
'{}'::jsonb
),
preprocessed = TRUE
@@ -4176,7 +4176,7 @@ async fn flow_to_payload(
db: &DB,
) -> Result<JobPayloadWithTag, Error> {
let FlowVersionInfo { version, on_behalf_of_email, edited_by, tag, .. } =
get_latest_flow_version_info_for_path(db, w_id, &path, true).await?;
get_latest_flow_version_info_for_path(None, db, db.clone(), w_id, &path, true).await?;
let on_behalf_of = if let Some(email) = on_behalf_of_email {
Some(OnBehalfOf { email, permissioned_as: username_to_permissioned_as(&edited_by) })
} else {
@@ -4202,8 +4202,14 @@ pub async fn script_to_payload(
tag_override
};
let (payload, tag, delete_after_use, script_timeout, on_behalf_of) = if script_hash.is_none() {
let (jp, tag, delete_after_use, script_timeout, on_behalf_of) =
script_path_to_payload(&script_path, db, &flow_job.workspace_id, Some(true)).await?;
let (jp, tag, delete_after_use, script_timeout, on_behalf_of) = script_path_to_payload(
&script_path,
None,
db.clone(),
&flow_job.workspace_id,
Some(true),
)
.await?;
(
jp,
tag_override.to_owned().or(tag),
@@ -4213,7 +4219,7 @@ pub async fn script_to_payload(
)
} else {
let hash = script_hash.unwrap();
let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?;
let ScriptHashInfo {
tag,
concurrency_key,
@@ -4228,7 +4234,7 @@ pub async fn script_to_payload(
on_behalf_of_email,
created_by,
..
} = get_script_info_for_hash(&mut *tx, &flow_job.workspace_id, hash.0).await?;
} = get_script_info_for_hash(None, db, &flow_job.workspace_id, hash.0).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 {
backend/windmill-worker/src/worker_lockfiles.rs
+5 -3
View File
@@ -422,14 +422,14 @@ pub async fn handle_dependency_job(
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)
codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets)
SELECT workspace_id, $1, path, array_prepend($2::bigint, COALESCE(parent_hashes, '{}'::bigint[])), summary, description, \
content, created_by, schema, is_template, extra_perms, $4, 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
codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets
FROM script WHERE hash = $2 AND workspace_id = $3;
",
@@ -665,7 +665,9 @@ pub async fn trigger_dependents_to_recompute_dependencies(
let kind = s.importer_kind.clone().unwrap_or_default();
let job_payload = if kind == "script" {
let r = get_latest_deployed_hash_for_path(db, w_id, s.importer_path.as_str()).await;
let r =
get_latest_deployed_hash_for_path(None, db.clone(), w_id, s.importer_path.as_str())
.await;
match r {
Ok(r) => JobPayload::Dependencies {
path: s.importer_path.clone(),