diff --git a/backend/.sqlx/query-2ccf7a95ed41083d92d6d9b29f700efb77317837a8bcffc45e23e5ad21e412ef.json b/backend/.sqlx/query-2ccf7a95ed41083d92d6d9b29f700efb77317837a8bcffc45e23e5ad21e412ef.json deleted file mode 100644 index 91b5f2caf2..0000000000 --- a/backend/.sqlx/query-2ccf7a95ed41083d92d6d9b29f700efb77317837a8bcffc45e23e5ad21e412ef.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-91f23fcc27777c279c79e2682fc15c026e55f9ec3799be65a2e8920fe6174a17.json b/backend/.sqlx/query-48fa200f9d1535ef606c5b1c9b6f3e8dff80a4c64857bc57e9fadce2da6b1a5b.json similarity index 69% rename from backend/.sqlx/query-91f23fcc27777c279c79e2682fc15c026e55f9ec3799be65a2e8920fe6174a17.json rename to backend/.sqlx/query-48fa200f9d1535ef606c5b1c9b6f3e8dff80a4c64857bc57e9fadce2da6b1a5b.json index 5ac64ecaed..11464e7f07 100644 --- a/backend/.sqlx/query-91f23fcc27777c279c79e2682fc15c026e55f9ec3799be65a2e8920fe6174a17.json +++ b/backend/.sqlx/query-48fa200f9d1535ef606c5b1c9b6f3e8dff80a4c64857bc57e9fadce2da6b1a5b.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT result as \"result: Json>>\"\n FROM v2_job_completed \n WHERE id = $1", + "query": "SELECT result as \"result: Json>>\"\n FROM v2_job_completed\n WHERE id = $1", "describe": { "columns": [ { @@ -18,5 +18,5 @@ true ] }, - "hash": "91f23fcc27777c279c79e2682fc15c026e55f9ec3799be65a2e8920fe6174a17" + "hash": "48fa200f9d1535ef606c5b1c9b6f3e8dff80a4c64857bc57e9fadce2da6b1a5b" } diff --git a/backend/.sqlx/query-506f90984a8b672aeff50b6d9e4751ea1521c524b9b7f51a48ee26280825ac9a.json b/backend/.sqlx/query-506f90984a8b672aeff50b6d9e4751ea1521c524b9b7f51a48ee26280825ac9a.json deleted file mode 100644 index fca7ee186e..0000000000 --- a/backend/.sqlx/query-506f90984a8b672aeff50b6d9e4751ea1521c524b9b7f51a48ee26280825ac9a.json +++ /dev/null @@ -1,36 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT \n COALESCE(jc.result, NULL) as \"result: sqlx::types::Json>\", \n SUBSTR(rs.stream, $3) AS \"result_stream: Option\", \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>", - "type_info": "Jsonb" - }, - { - "ordinal": 1, - "name": "result_stream: Option", - "type_info": "Text" - }, - { - "ordinal": 2, - "name": "stream_offset", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Text", - "Uuid", - "Int4" - ] - }, - "nullable": [ - null, - null, - null - ] - }, - "hash": "506f90984a8b672aeff50b6d9e4751ea1521c524b9b7f51a48ee26280825ac9a" -} diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 36ddb8ab9f..713ccb9dd3 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - true + null ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-5b8c1803f0ccead11517fbc8a9bdc0227dc3922217fa18f0b71ff0484d65838c.json b/backend/.sqlx/query-5b8c1803f0ccead11517fbc8a9bdc0227dc3922217fa18f0b71ff0484d65838c.json new file mode 100644 index 0000000000..8b07d39a36 --- /dev/null +++ b/backend/.sqlx/query-5b8c1803f0ccead11517fbc8a9bdc0227dc3922217fa18f0b71ff0484d65838c.json @@ -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" +} diff --git a/backend/.sqlx/query-618ec69c9c78f1c9e3539d2770392e3f783f29cd2cc58c0fc87d14ecef32b467.json b/backend/.sqlx/query-618ec69c9c78f1c9e3539d2770392e3f783f29cd2cc58c0fc87d14ecef32b467.json new file mode 100644 index 0000000000..2aa95c8d2c --- /dev/null +++ b/backend/.sqlx/query-618ec69c9c78f1c9e3539d2770392e3f783f29cd2cc58c0fc87d14ecef32b467.json @@ -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" +} diff --git a/backend/.sqlx/query-6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d.json b/backend/.sqlx/query-6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d.json new file mode 100644 index 0000000000..2f14e723e4 --- /dev/null +++ b/backend/.sqlx/query-6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d.json @@ -0,0 +1,42 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n COALESCE(jc.result, jc.result) as \"result: sqlx::types::Json>\",\n jq.running as \"running: Option\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option\",\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>", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "running: Option", + "type_info": "Bool" + }, + { + "ordinal": 2, + "name": "result_stream: Option", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "stream_offset", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text", + "Int4" + ] + }, + "nullable": [ + null, + false, + null, + null + ] + }, + "hash": "6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d" +} diff --git a/backend/.sqlx/query-4f372d047c78532907adf2d2dc114352aa7b5b28dccc50a1231a7f6539397da7.json b/backend/.sqlx/query-70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393.json similarity index 50% rename from backend/.sqlx/query-4f372d047c78532907adf2d2dc114352aa7b5b28dccc50a1231a7f6539397da7.json rename to backend/.sqlx/query-70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393.json index e8377c5ff4..82333559eb 100644 --- a/backend/.sqlx/query-4f372d047c78532907adf2d2dc114352aa7b5b28dccc50a1231a7f6539397da7.json +++ b/backend/.sqlx/query-70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393.json @@ -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>\",\n COALESCE(c.workflow_as_code_status, f.workflow_as_code_status) AS \"workflow_as_code_status: sqlx::types::Json>\",\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\"\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>\",\n COALESCE(c.workflow_as_code_status, f.workflow_as_code_status) AS \"workflow_as_code_status: sqlx::types::Json>\",\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\"\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" } diff --git a/backend/.sqlx/query-9dec888d4b0666d1843fbbc4fb2475fd947f047a9965bc110d7338208b77783d.json b/backend/.sqlx/query-9dec888d4b0666d1843fbbc4fb2475fd947f047a9965bc110d7338208b77783d.json deleted file mode 100644 index 7c49fe890e..0000000000 --- a/backend/.sqlx/query-9dec888d4b0666d1843fbbc4fb2475fd947f047a9965bc110d7338208b77783d.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-b5fbd7893950610f1285662df24f438c9855ba860e23befd88c2544ef86e9133.json b/backend/.sqlx/query-b5fbd7893950610f1285662df24f438c9855ba860e23befd88c2544ef86e9133.json new file mode 100644 index 0000000000..766748fa60 --- /dev/null +++ b/backend/.sqlx/query-b5fbd7893950610f1285662df24f438c9855ba860e23befd88c2544ef86e9133.json @@ -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" +} diff --git a/backend/.sqlx/query-c61f9bad73beb0bc317594f6ca68389c2d0adbef1fb6d73ad23e4153cff44f9e.json b/backend/.sqlx/query-c61f9bad73beb0bc317594f6ca68389c2d0adbef1fb6d73ad23e4153cff44f9e.json deleted file mode 100644 index 22ba8df0fe..0000000000 --- a/backend/.sqlx/query-c61f9bad73beb0bc317594f6ca68389c2d0adbef1fb6d73ad23e4153cff44f9e.json +++ /dev/null @@ -1,42 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT \n COALESCE(jc.result, jc.result) as \"result: sqlx::types::Json>\",\n jq.running as \"running: Option\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option\",\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>", - "type_info": "Jsonb" - }, - { - "ordinal": 1, - "name": "running: Option", - "type_info": "Bool" - }, - { - "ordinal": 2, - "name": "result_stream: Option", - "type_info": "Text" - }, - { - "ordinal": 3, - "name": "stream_offset", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text", - "Int4" - ] - }, - "nullable": [ - null, - false, - null, - null - ] - }, - "hash": "c61f9bad73beb0bc317594f6ca68389c2d0adbef1fb6d73ad23e4153cff44f9e" -} diff --git a/backend/.sqlx/query-c76db7bac7a11b98eaf401de54ceb8d76698d349a2e968b5f06c12c1358d6537.json b/backend/.sqlx/query-c76db7bac7a11b98eaf401de54ceb8d76698d349a2e968b5f06c12c1358d6537.json new file mode 100644 index 0000000000..02fb341cf2 --- /dev/null +++ b/backend/.sqlx/query-c76db7bac7a11b98eaf401de54ceb8d76698d349a2e968b5f06c12c1358d6537.json @@ -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" +} diff --git a/backend/.sqlx/query-ce6f3e803909d55c19169c77d4111bffc0fc93032943369015971373f1f2af68.json b/backend/.sqlx/query-ce6f3e803909d55c19169c77d4111bffc0fc93032943369015971373f1f2af68.json deleted file mode 100644 index 3b5c65095a..0000000000 --- a/backend/.sqlx/query-ce6f3e803909d55c19169c77d4111bffc0fc93032943369015971373f1f2af68.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d.json b/backend/.sqlx/query-d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d.json new file mode 100644 index 0000000000..76c37b7a08 --- /dev/null +++ b/backend/.sqlx/query-d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d.json @@ -0,0 +1,36 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n COALESCE(jc.result, NULL) as \"result: sqlx::types::Json>\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option\",\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>", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "result_stream: Option", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "stream_offset", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Int4" + ] + }, + "nullable": [ + null, + null, + null + ] + }, + "hash": "d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d" +} diff --git a/backend/.sqlx/query-e07660e8d2a265cb6a83f3a2bb8e7e6330f09ab116e9837f6f16f8fdef938004.json b/backend/.sqlx/query-e07660e8d2a265cb6a83f3a2bb8e7e6330f09ab116e9837f6f16f8fdef938004.json deleted file mode 100644 index f03f5c06f6..0000000000 --- a/backend/.sqlx/query-e07660e8d2a265cb6a83f3a2bb8e7e6330f09ab116e9837f6f16f8fdef938004.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-e7348225a27bbdc9607d7c799e7192cd6ce4088467d91a5cbdc019430320d26d.json b/backend/.sqlx/query-e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d.json similarity index 71% rename from backend/.sqlx/query-e7348225a27bbdc9607d7c799e7192cd6ce4088467d91a5cbdc019430320d26d.json rename to backend/.sqlx/query-e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d.json index 000329b19f..c7c7214a87 100644 --- a/backend/.sqlx/query-e7348225a27bbdc9607d7c799e7192cd6ce4088467d91a5cbdc019430320d26d.json +++ b/backend/.sqlx/query-e2474b7855c8b08f927f2b987421e773e537a8eb0a113477764bbab14e1f3a3d.json @@ -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" } diff --git a/backend/Cargo.lock b/backend/Cargo.lock index c5b0e48db7..6c8da40ae9 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15356,6 +15356,7 @@ dependencies = [ "anyhow", "async-recursion", "async-stream", + "async-trait", "aws-config", "aws-sdk-sts", "aws-smithy-types-convert", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index bbb04451e6..775c20ca20 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -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" diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 2ee878c68b..40b748d22c 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -c409ac5e5bd202002648f18e8adb7cdf78b2a83e \ No newline at end of file +ad0f06c836d050b3ddc1937ef3a09f6b49cf5191 \ No newline at end of file diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 5a776e53c7..60d79e3b31 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -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 } \ No newline at end of file +deno_core = { workspace = true, optional = true } diff --git a/backend/windmill-api/src/auth.rs b/backend/windmill-api/src/auth.rs index 01147c0ed8..5f66a0deaf 100644 --- a/backend/windmill-api/src/auth.rs +++ b/backend/windmill-api/src/auth.rs @@ -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) diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 0900962c6d..e5d8c433f1 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -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>, 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, } +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 for Authed { fn from(value: ApiAuthed) -> Self { Self { diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index cbdc5eb5cc..4b9f64ea3c 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -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(
Trigger path: {}
- +
Trigger Type: {}
"#, @@ -1147,7 +1157,7 @@ async fn send_workspace_trigger_failure_email_notification(

{}

- +
Workspace: {}
@@ -1155,27 +1165,21 @@ async fn send_workspace_trigger_failure_email_notification(
Script/Flow Path: {}
- +
Job ID: {}
- +
Error Details: 
{}
- + View Job Details "#, - 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, Path((w_id, id)): Path<(String, Uuid)>, ) -> error::JsonResult { - 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 { 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, Json(preview): Json, ) -> error::Result { - 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::().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::() + .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, Json(raw_flow): Json, ) -> error::Result { - 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::().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::() + .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, 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, ) -> JsonResult { Ok(Json( get_job_update_data( @@ -6201,9 +6249,15 @@ async fn get_job_update_sse( opt_tokened: OptTokened, Extension(db): Extension, 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, ) -> 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>\", v2_job.tag, v2_job_queue.running as \"running: Option\", SUBSTR(rs.stream, $3) AS \"result_stream: Option\", 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>\", jq.running as \"running: Option\", SUBSTR(rs.stream, $3) AS \"result_stream: Option\", @@ -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>\", - SUBSTR(rs.stream, $3) AS \"result_stream: Option\", + "SELECT + COALESCE(jc.result, NULL) as \"result: sqlx::types::Json>\", + SUBSTR(rs.stream, $3) AS \"result_stream: Option\", 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, diff --git a/backend/windmill-api/src/trigger_helpers.rs b/backend/windmill-api/src/trigger_helpers.rs index 3670622128..73ba0250bd 100644 --- a/backend/windmill-api/src/trigger_helpers.rs +++ b/backend/windmill-api/src/trigger_helpers.rs @@ -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 { 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?; diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index 21a698b62d..a0bb07c7ad 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -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 { 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?; diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index d70c190b49..ab17eb81cf 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -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 diff --git a/backend/windmill-common/src/auth.rs b/backend/windmill-common/src/auth.rs index 97aa27e9d8..c2b5b5dffd 100644 --- a/backend/windmill-common/src/auth.rs +++ b/backend/windmill-common/src/auth.rs @@ -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>( + &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>(&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, take: Option) -> bool { let now = Utc::now(); diff --git a/backend/windmill-common/src/db.rs b/backend/windmill-common/src/db.rs index fe0d215c69..46fde40380 100644 --- a/backend/windmill-common/src/db.rs +++ b/backend/windmill-common/src/db.rs @@ -1,8 +1,8 @@ -use sqlx::{Pool, Postgres, Transaction}; +use sqlx::{Acquire, Pool, Postgres, Transaction}; pub type DB = Pool; -#[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, } +#[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, + // (folder name, can write, is owner) + pub folders: &'a Vec<(String, bool, bool)>, + pub scopes: &'a Option>, + pub token_prefix: &'a Option, +} + +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 = 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> { + Box::pin(async move { self.db.clone().begin(self.authed).await }) + } + + fn begin( + self, + ) -> futures_core::future::BoxFuture<'c, Result, sqlx::Error>> { + Box::pin(async move { self.db.clone().begin(self.authed).await }) + } +} + impl UserDB { pub fn new(db: DB) -> Self { Self { db } diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 0c53426eab..1418df1165 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -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>>, + db: DB, w_id: &str, skip_preprocessor: Option, ) -> 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, Option), 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, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index ed35ab84bf..9dd0b09809 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -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 { - 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 { 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, 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> + Send + 'a { +pub fn get_latest_deployed_hash_for_path<'e>( + db: Option>>, + db2: DB, + w_id: &'e str, + script_path: &'e str, +) -> impl Future> + 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> { + 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>>, db: E, w_id: &str, hash: i64, ) -> error::Result { 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> { + 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, } +struct CachedFlowPath(String); + +impl Into 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>>, db: A, + db2: DB, w_id: &'a str, path: &'a str, use_cache: bool, -) -> impl Future> + Send + 'a { +) -> impl Future> + 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> { + 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, diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index c2a753e1a9..2a9981d01f 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -131,6 +131,12 @@ impl FromStr for ScriptLang { #[sqlx(transparent)] pub struct ScriptHash(pub i64); +impl Into for ScriptHash { + fn into(self) -> u64 { + self.0 as u64 + } +} + #[derive(PartialEq, sqlx::Type)] #[sqlx(transparent, no_pg_array)] pub struct ScriptHashes(pub Vec); diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index cd682629c2..d2fbe693e7 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -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, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 92c55f103c..79297b05bf 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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>>\" - 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 { 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 { diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 38a451d8fa..ed417dcfec 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -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(),