From a08268337b2a3b0282a160176cca782c7655d1cc Mon Sep 17 00:00:00 2001 From: hugocasa Date: Fri, 26 Sep 2025 13:26:50 +0200 Subject: [PATCH] feat(backend): job result stream optimization (#6673) * feat(backend): job result stream optimization * get offset locally instead of from db * fix: agent worker result stream * update ref * nit * remove foreign key on job * fix build --- ...04c3a36aa284fab3b7e638dbca12ac846b82b.json | 5 + ...10fffb37599f2a703885ee723cb9290fed13e.json | 5 + ...1ae5a5e64086198cb67aba72345bb3cdf7773.json | 101 ------------------ ...73917ed13269f2ee20b86df79fca2c8efe672.json | 5 + ...4e96c69d10e3970e38151c97d1fca0cb7388d.json | 49 +++++++++ ...03df52520b7606378ecca267e88383a45b49b.json | 16 --- ...de88d2615e7e6f1267f2fb12ccc325d4e4148.json | 14 +++ ...9ffe41d162b482adb04b7b898a19576af2a5e.json | 5 + ...aa669a153ab0c0bb2807cd4f7fd405afa6f69.json | 5 + ...259bd989a9d769a6a300d37ab116960438882.json | 101 ++++++++++++++++++ ...a0dbc200d45960fb8dea867492c8c7dd096d5.json | 5 + ...9cf5e8484c147e457044d6b075323b163ebaa.json | 12 --- ...871b371663257a127eb35512695c487066fd7.json | 49 --------- ...afe83c07b7a8b422f48022994370b3e433f34.json | 55 ---------- ...dbb34af450041c675b30b708fed3ef9e01d2d.json | 5 + ...63078922340294f1c6515400fb2c896666fde.json | 17 +++ ...72b273d14a093bb29ec12dda8bbfaecd49a35.json | 55 ++++++++++ ...7149791de48bc2e489edb55d63f67702cce05.json | 49 --------- ...72889ff257db0f0e3e9252adda2e4ef2039d6.json | 5 + ...03873e7dfd1cf9fe984c769afb5a8bdb48d49.json | 49 +++++++++ ...33b38deb9c7fa17bd3836c9ad8884f27be84a.json | 5 + ...5693dedf34366e51044ddf30695995d194090.json | 14 +++ ...589fc104647cae7abf37ddce3ef7109726261.json | 29 +++++ ...57ee0dce940039ab42fd39bf079d87b6cad8a.json | 29 ----- ...87b578bc63792cdb0f7ddaeb82e9771287c22.json | 5 + ...0737b249a64c8d9e8f7dbcc46fce98e874571.json | 12 +++ backend/ee-repo-ref.txt | 2 +- ...50953_new_job_result_stream_table.down.sql | 4 + ...4150953_new_job_result_stream_table.up.sql | 13 +++ backend/src/monitor.rs | 13 ++- backend/update_sqlx.sh | 10 +- backend/windmill-api/src/jobs.rs | 82 ++++++++++---- backend/windmill-common/src/result_stream.rs | 13 ++- backend/windmill-queue/src/jobs.rs | 11 ++ backend/windmill-worker/src/bun_executor.rs | 4 + .../windmill-worker/src/dedicated_worker.rs | 6 +- backend/windmill-worker/src/deno_executor.rs | 4 + backend/windmill-worker/src/handle_child.rs | 16 ++- backend/windmill-worker/src/job_logger.rs | 18 +++- backend/windmill-worker/src/js_eval.rs | 22 ++-- .../windmill-worker/src/python_executor.rs | 3 + .../windmill-worker/src/result_processor.rs | 19 +++- backend/windmill-worker/src/worker.rs | 13 +++ backend/windmill-worker/src/worker_flow.rs | 2 + frontend/src/lib/components/JobLoader.svelte | 4 +- 45 files changed, 605 insertions(+), 355 deletions(-) delete mode 100644 backend/.sqlx/query-40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773.json create mode 100644 backend/.sqlx/query-4b056d33215b3a1e9849bb66ce84e96c69d10e3970e38151c97d1fca0cb7388d.json delete mode 100644 backend/.sqlx/query-4bc533074c720820cebff8d97a203df52520b7606378ecca267e88383a45b49b.json create mode 100644 backend/.sqlx/query-5adeb6989648ca42431c96069c4de88d2615e7e6f1267f2fb12ccc325d4e4148.json create mode 100644 backend/.sqlx/query-7b524ee24bb78e494a93c5ea205259bd989a9d769a6a300d37ab116960438882.json delete mode 100644 backend/.sqlx/query-807c920bff25f56b10e88900d879cf5e8484c147e457044d6b075323b163ebaa.json delete mode 100644 backend/.sqlx/query-80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7.json delete mode 100644 backend/.sqlx/query-8126b118704341846e88bd289f1afe83c07b7a8b422f48022994370b3e433f34.json create mode 100644 backend/.sqlx/query-9bfc2a821b25641af48b0e3954163078922340294f1c6515400fb2c896666fde.json create mode 100644 backend/.sqlx/query-9c0bbd44902d8eee393236f7c2372b273d14a093bb29ec12dda8bbfaecd49a35.json delete mode 100644 backend/.sqlx/query-a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05.json create mode 100644 backend/.sqlx/query-a76eec5797ca8f97e63ed5542bf03873e7dfd1cf9fe984c769afb5a8bdb48d49.json create mode 100644 backend/.sqlx/query-bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090.json create mode 100644 backend/.sqlx/query-dd92bce0ddbfbf06834807aab4d589fc104647cae7abf37ddce3ef7109726261.json delete mode 100644 backend/.sqlx/query-f17f914d2522bf7cb5de9d7ba5557ee0dce940039ab42fd39bf079d87b6cad8a.json create mode 100644 backend/.sqlx/query-f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571.json create mode 100644 backend/migrations/20250924150953_new_job_result_stream_table.down.sql create mode 100644 backend/migrations/20250924150953_new_job_result_stream_table.up.sql diff --git a/backend/.sqlx/query-193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b.json b/backend/.sqlx/query-193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b.json index 2d4df84899..76a51667e9 100644 --- a/backend/.sqlx/query-193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b.json +++ b/backend/.sqlx/query-193d292c5ed44bf5266ad52c83704c3a36aa284fab3b7e638dbca12ac846b82b.json @@ -37,6 +37,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json b/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json index 5c5d9b68ba..3d3a708086 100644 --- a/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json +++ b/backend/.sqlx/query-3f08ffbb5c71b873a9e164ecb0b10fffb37599f2a703885ee723cb9290fed13e.json @@ -26,6 +26,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773.json b/backend/.sqlx/query-40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773.json deleted file mode 100644 index e7d66a3d75..0000000000 --- a/backend/.sqlx/query-40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773.json +++ /dev/null @@ -1,101 +0,0 @@ -{ - "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.flow_status, f.flow_status)->>'stream_job')::uuid AS stream_job,\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": [ - { - "ordinal": 0, - "name": "completed", - "type_info": "Bool" - }, - { - "ordinal": 1, - "name": "running", - "type_info": "Bool" - }, - { - "ordinal": 2, - "name": "logs", - "type_info": "Text" - }, - { - "ordinal": 3, - "name": "new_result_stream", - "type_info": "Text" - }, - { - "ordinal": 4, - "name": "mem_peak", - "type_info": "Int4" - }, - { - "ordinal": 5, - "name": "flow_status: sqlx::types::Json>", - "type_info": "Jsonb" - }, - { - "ordinal": 6, - "name": "stream_job", - "type_info": "Uuid" - }, - { - "ordinal": 7, - "name": "workflow_as_code_status: sqlx::types::Json>", - "type_info": "Jsonb" - }, - { - "ordinal": 8, - "name": "log_offset", - "type_info": "Int4" - }, - { - "ordinal": 9, - "name": "stream_offset", - "type_info": "Int4" - }, - { - "ordinal": 10, - "name": "created_by!", - "type_info": "Varchar" - }, - { - "ordinal": 11, - "name": "progress", - "type_info": "Int4" - }, - { - "ordinal": 12, - "name": "result_stream: Option", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Int4", - "Text", - "Uuid", - "Bool", - "Bool", - "TextArray", - "Bool", - "Int4" - ] - }, - "nullable": [ - null, - null, - null, - null, - null, - null, - null, - null, - null, - null, - false, - null, - false - ] - }, - "hash": "40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773" -} diff --git a/backend/.sqlx/query-488dd591096b2b47787afdc3a1d73917ed13269f2ee20b86df79fca2c8efe672.json b/backend/.sqlx/query-488dd591096b2b47787afdc3a1d73917ed13269f2ee20b86df79fca2c8efe672.json index af477b4db6..e1120042ff 100644 --- a/backend/.sqlx/query-488dd591096b2b47787afdc3a1d73917ed13269f2ee20b86df79fca2c8efe672.json +++ b/backend/.sqlx/query-488dd591096b2b47787afdc3a1d73917ed13269f2ee20b86df79fca2c8efe672.json @@ -38,6 +38,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-4b056d33215b3a1e9849bb66ce84e96c69d10e3970e38151c97d1fca0cb7388d.json b/backend/.sqlx/query-4b056d33215b3a1e9849bb66ce84e96c69d10e3970e38151c97d1fca0cb7388d.json new file mode 100644 index 0000000000..da7115eb81 --- /dev/null +++ b/backend/.sqlx/query-4b056d33215b3a1e9849bb66ce84e96c69d10e3970e38151c97d1fca0cb7388d.json @@ -0,0 +1,49 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH result_stream AS (\n SELECT \n string_agg(stream, '' order by idx asc) as stream, \n job_id, \n max(idx) + 1 as offset \n FROM job_result_stream_v2\n WHERE job_id = $2 AND idx >= $3\n GROUP BY job_id\n )\n SELECT\n COALESCE(jc.result, NULL) as \"result: sqlx::types::Json>\",\n rs.stream AS \"result_stream: Option\",\n rs.offset AS stream_offset,\n COALESCE(js.flow_status, jc.flow_status) as \"flow_status: sqlx::types::Json>\",\n CASE WHEN $4 THEN NULL ELSE (COALESCE(js.flow_status, jc.flow_status)->>'stream_job')::uuid END as stream_job\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 v2_job_status js ON js.id = base.job_id\n LEFT JOIN 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" + }, + { + "ordinal": 3, + "name": "flow_status: sqlx::types::Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 4, + "name": "stream_job", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Int4", + "Bool" + ] + }, + "nullable": [ + null, + null, + null, + null, + null + ] + }, + "hash": "4b056d33215b3a1e9849bb66ce84e96c69d10e3970e38151c97d1fca0cb7388d" +} diff --git a/backend/.sqlx/query-4bc533074c720820cebff8d97a203df52520b7606378ecca267e88383a45b49b.json b/backend/.sqlx/query-4bc533074c720820cebff8d97a203df52520b7606378ecca267e88383a45b49b.json deleted file mode 100644 index 8c9de521d2..0000000000 --- a/backend/.sqlx/query-4bc533074c720820cebff8d97a203df52520b7606378ecca267e88383a45b49b.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO job_result_stream (workspace_id, job_id, stream)\n VALUES ($1, $2, $3)\n ON CONFLICT (job_id) DO UPDATE SET stream = job_result_stream.stream || $3\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Uuid", - "Text" - ] - }, - "nullable": [] - }, - "hash": "4bc533074c720820cebff8d97a203df52520b7606378ecca267e88383a45b49b" -} diff --git a/backend/.sqlx/query-5adeb6989648ca42431c96069c4de88d2615e7e6f1267f2fb12ccc325d4e4148.json b/backend/.sqlx/query-5adeb6989648ca42431c96069c4de88d2615e7e6f1267f2fb12ccc325d4e4148.json new file mode 100644 index 0000000000..859f2b23a4 --- /dev/null +++ b/backend/.sqlx/query-5adeb6989648ca42431c96069c4de88d2615e7e6f1267f2fb12ccc325d4e4148.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM job_result_stream_v2 WHERE job_id = ANY($1)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [] + }, + "hash": "5adeb6989648ca42431c96069c4de88d2615e7e6f1267f2fb12ccc325d4e4148" +} diff --git a/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json b/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json index 63363e74ff..b68c9ca97d 100644 --- a/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json +++ b/backend/.sqlx/query-7274e9489b18d7ab82bad1fbff89ffe41d162b482adb04b7b898a19576af2a5e.json @@ -96,6 +96,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json b/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json index e4a0fec1b2..f9761e083d 100644 --- a/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json +++ b/backend/.sqlx/query-7af1cf089022fc1c3597b270b69aa669a153ab0c0bb2807cd4f7fd405afa6f69.json @@ -36,6 +36,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-7b524ee24bb78e494a93c5ea205259bd989a9d769a6a300d37ab116960438882.json b/backend/.sqlx/query-7b524ee24bb78e494a93c5ea205259bd989a9d769a6a300d37ab116960438882.json new file mode 100644 index 0000000000..59f972f50d --- /dev/null +++ b/backend/.sqlx/query-7b524ee24bb78e494a93c5ea205259bd989a9d769a6a300d37ab116960438882.json @@ -0,0 +1,101 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH result_stream AS (\n SELECT \n string_agg(stream, '' order by idx asc) as stream, \n job_id, \n max(idx) + 1 as offset \n FROM job_result_stream_v2\n WHERE job_id = $3 AND idx >= $8\n GROUP BY job_id\n )\n 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 rs.stream 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.flow_status, f.flow_status)->>'stream_job')::uuid AS stream_job,\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 rs.offset 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 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": [ + { + "ordinal": 0, + "name": "completed", + "type_info": "Bool" + }, + { + "ordinal": 1, + "name": "running", + "type_info": "Bool" + }, + { + "ordinal": 2, + "name": "logs", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "new_result_stream", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "mem_peak", + "type_info": "Int4" + }, + { + "ordinal": 5, + "name": "flow_status: sqlx::types::Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 6, + "name": "stream_job", + "type_info": "Uuid" + }, + { + "ordinal": 7, + "name": "workflow_as_code_status: sqlx::types::Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 8, + "name": "log_offset", + "type_info": "Int4" + }, + { + "ordinal": 9, + "name": "stream_offset", + "type_info": "Int4" + }, + { + "ordinal": 10, + "name": "created_by!", + "type_info": "Varchar" + }, + { + "ordinal": 11, + "name": "progress", + "type_info": "Int4" + }, + { + "ordinal": 12, + "name": "result_stream: Option", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int4", + "Text", + "Uuid", + "Bool", + "Bool", + "TextArray", + "Bool", + "Int4" + ] + }, + "nullable": [ + null, + null, + null, + null, + null, + null, + null, + null, + null, + null, + false, + null, + null + ] + }, + "hash": "7b524ee24bb78e494a93c5ea205259bd989a9d769a6a300d37ab116960438882" +} diff --git a/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json b/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json index 5500f83561..111d536b18 100644 --- a/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json +++ b/backend/.sqlx/query-805d633de90fee335f1726284eda0dbc200d45960fb8dea867492c8c7dd096d5.json @@ -36,6 +36,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-807c920bff25f56b10e88900d879cf5e8484c147e457044d6b075323b163ebaa.json b/backend/.sqlx/query-807c920bff25f56b10e88900d879cf5e8484c147e457044d6b075323b163ebaa.json deleted file mode 100644 index 1d56449d18..0000000000 --- a/backend/.sqlx/query-807c920bff25f56b10e88900d879cf5e8484c147e457044d6b075323b163ebaa.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "VACUUM v2_job, v2_job_completed, job_result_stream, job_stats, job_logs, concurrency_key, log_file, metrics", - "describe": { - "columns": [], - "parameters": { - "Left": [] - }, - "nullable": [] - }, - "hash": "807c920bff25f56b10e88900d879cf5e8484c147e457044d6b075323b163ebaa" -} diff --git a/backend/.sqlx/query-80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7.json b/backend/.sqlx/query-80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7.json deleted file mode 100644 index 7254b7e84f..0000000000 --- a/backend/.sqlx/query-80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7.json +++ /dev/null @@ -1,49 +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 COALESCE(js.flow_status, jc.flow_status) as \"flow_status: sqlx::types::Json>\",\n CASE WHEN $4 THEN NULL ELSE (COALESCE(js.flow_status, jc.flow_status)->>'stream_job')::uuid END as stream_job\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 v2_job_status js ON js.id = base.job_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" - }, - { - "ordinal": 3, - "name": "flow_status: sqlx::types::Json>", - "type_info": "Jsonb" - }, - { - "ordinal": 4, - "name": "stream_job", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Text", - "Uuid", - "Int4", - "Bool" - ] - }, - "nullable": [ - null, - null, - null, - null, - null - ] - }, - "hash": "80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7" -} diff --git a/backend/.sqlx/query-8126b118704341846e88bd289f1afe83c07b7a8b422f48022994370b3e433f34.json b/backend/.sqlx/query-8126b118704341846e88bd289f1afe83c07b7a8b422f48022994370b3e433f34.json deleted file mode 100644 index abc63b0e90..0000000000 --- a/backend/.sqlx/query-8126b118704341846e88bd289f1afe83c07b7a8b422f48022994370b3e433f34.json +++ /dev/null @@ -1,55 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT \n jc.result as \"result: sqlx::types::Json>\",\n v2_job.tag,\n v2_job_queue.running as \"running: Option\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option\",\n CHAR_LENGTH(rs.stream) AS stream_offset,\n CASE WHEN $4 THEN NULL ELSE (COALESCE(js.flow_status, jc.flow_status)->>'stream_job')::uuid END as stream_job\n FROM v2_job\n LEFT JOIN v2_job_queue USING (id)\n LEFT JOIN v2_job_completed jc USING (id)\n LEFT JOIN v2_job_status js USING (id)\n LEFT JOIN job_result_stream rs ON rs.job_id = $2\n WHERE v2_job.id = $2 AND v2_job.workspace_id = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "result: sqlx::types::Json>", - "type_info": "Jsonb" - }, - { - "ordinal": 1, - "name": "tag", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "running: Option", - "type_info": "Bool" - }, - { - "ordinal": 3, - "name": "result_stream: Option", - "type_info": "Text" - }, - { - "ordinal": 4, - "name": "stream_offset", - "type_info": "Int4" - }, - { - "ordinal": 5, - "name": "stream_job", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Text", - "Uuid", - "Int4", - "Bool" - ] - }, - "nullable": [ - true, - false, - false, - null, - null, - null - ] - }, - "hash": "8126b118704341846e88bd289f1afe83c07b7a8b422f48022994370b3e433f34" -} diff --git a/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json b/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json index bcc10fcf4a..63d8282ef0 100644 --- a/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json +++ b/backend/.sqlx/query-92c7c961198e506426bf3f97a8ddbb34af450041c675b30b708fed3ef9e01d2d.json @@ -26,6 +26,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-9bfc2a821b25641af48b0e3954163078922340294f1c6515400fb2c896666fde.json b/backend/.sqlx/query-9bfc2a821b25641af48b0e3954163078922340294f1c6515400fb2c896666fde.json new file mode 100644 index 0000000000..de16cf3146 --- /dev/null +++ b/backend/.sqlx/query-9bfc2a821b25641af48b0e3954163078922340294f1c6515400fb2c896666fde.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO job_result_stream_v2 (workspace_id, job_id, stream, idx)\n VALUES (\n $1, \n $2,\n $3, \n $4\n )\n ON CONFLICT (job_id, idx) DO UPDATE SET stream = job_result_stream_v2.stream || $3\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Text", + "Int4" + ] + }, + "nullable": [] + }, + "hash": "9bfc2a821b25641af48b0e3954163078922340294f1c6515400fb2c896666fde" +} diff --git a/backend/.sqlx/query-9c0bbd44902d8eee393236f7c2372b273d14a093bb29ec12dda8bbfaecd49a35.json b/backend/.sqlx/query-9c0bbd44902d8eee393236f7c2372b273d14a093bb29ec12dda8bbfaecd49a35.json new file mode 100644 index 0000000000..5740a1bda5 --- /dev/null +++ b/backend/.sqlx/query-9c0bbd44902d8eee393236f7c2372b273d14a093bb29ec12dda8bbfaecd49a35.json @@ -0,0 +1,55 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH result_stream AS (\n SELECT \n string_agg(stream, '' order by idx asc) as stream, \n job_id, \n max(idx) + 1 as offset \n FROM job_result_stream_v2\n WHERE job_id = $2 AND idx >= $3\n GROUP BY job_id\n )\n SELECT \n jc.result as \"result: sqlx::types::Json>\",\n v2_job.tag,\n v2_job_queue.running as \"running: Option\",\n rs.stream AS \"result_stream: Option\",\n rs.offset AS stream_offset,\n CASE WHEN $4 THEN NULL ELSE (COALESCE(js.flow_status, jc.flow_status)->>'stream_job')::uuid END as stream_job\n FROM v2_job\n LEFT JOIN v2_job_queue USING (id)\n LEFT JOIN v2_job_completed jc USING (id)\n LEFT JOIN v2_job_status js USING (id)\n LEFT JOIN result_stream rs ON rs.job_id = $2\n WHERE v2_job.id = $2 AND v2_job.workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "result: sqlx::types::Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "running: Option", + "type_info": "Bool" + }, + { + "ordinal": 3, + "name": "result_stream: Option", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "stream_offset", + "type_info": "Int4" + }, + { + "ordinal": 5, + "name": "stream_job", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Int4", + "Bool" + ] + }, + "nullable": [ + true, + false, + false, + null, + null, + null + ] + }, + "hash": "9c0bbd44902d8eee393236f7c2372b273d14a093bb29ec12dda8bbfaecd49a35" +} diff --git a/backend/.sqlx/query-a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05.json b/backend/.sqlx/query-a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05.json deleted file mode 100644 index 4cca74dd4b..0000000000 --- a/backend/.sqlx/query-a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05.json +++ /dev/null @@ -1,49 +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 CASE WHEN $4 THEN NULL ELSE (COALESCE(js.flow_status, jc.flow_status)->>'stream_job')::uuid END as stream_job\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 v2_job_status js ON js.id = base.job_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" - }, - { - "ordinal": 4, - "name": "stream_job", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text", - "Int4", - "Bool" - ] - }, - "nullable": [ - null, - false, - null, - null, - null - ] - }, - "hash": "a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05" -} diff --git a/backend/.sqlx/query-a6608d47b96d851eb7b04d2e4b472889ff257db0f0e3e9252adda2e4ef2039d6.json b/backend/.sqlx/query-a6608d47b96d851eb7b04d2e4b472889ff257db0f0e3e9252adda2e4ef2039d6.json index 8fea4c40ec..48c153591f 100644 --- a/backend/.sqlx/query-a6608d47b96d851eb7b04d2e4b472889ff257db0f0e3e9252adda2e4ef2039d6.json +++ b/backend/.sqlx/query-a6608d47b96d851eb7b04d2e4b472889ff257db0f0e3e9252adda2e4ef2039d6.json @@ -96,6 +96,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-a76eec5797ca8f97e63ed5542bf03873e7dfd1cf9fe984c769afb5a8bdb48d49.json b/backend/.sqlx/query-a76eec5797ca8f97e63ed5542bf03873e7dfd1cf9fe984c769afb5a8bdb48d49.json new file mode 100644 index 0000000000..ddf7d06e68 --- /dev/null +++ b/backend/.sqlx/query-a76eec5797ca8f97e63ed5542bf03873e7dfd1cf9fe984c769afb5a8bdb48d49.json @@ -0,0 +1,49 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH result_stream AS (\n SELECT \n string_agg(stream, '' order by idx asc) as stream, \n job_id, \n max(idx) + 1 as offset \n FROM job_result_stream_v2\n WHERE job_id = $1 AND idx >= $3\n GROUP BY job_id\n )\n SELECT\n COALESCE(jc.result, jc.result) as \"result: sqlx::types::Json>\",\n jq.running as \"running: Option\",\n rs.stream AS \"result_stream: Option\",\n rs.offset AS stream_offset,\n CASE WHEN $4 THEN NULL ELSE (COALESCE(js.flow_status, jc.flow_status)->>'stream_job')::uuid END as stream_job\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 v2_job_status js ON js.id = base.job_id\n LEFT JOIN 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" + }, + { + "ordinal": 4, + "name": "stream_job", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text", + "Int4", + "Bool" + ] + }, + "nullable": [ + null, + false, + null, + null, + null + ] + }, + "hash": "a76eec5797ca8f97e63ed5542bf03873e7dfd1cf9fe984c769afb5a8bdb48d49" +} diff --git a/backend/.sqlx/query-b3c02fd225a6aa78785d466e7f033b38deb9c7fa17bd3836c9ad8884f27be84a.json b/backend/.sqlx/query-b3c02fd225a6aa78785d466e7f033b38deb9c7fa17bd3836c9ad8884f27be84a.json index 5a6aa7dd69..4756fcf815 100644 --- a/backend/.sqlx/query-b3c02fd225a6aa78785d466e7f033b38deb9c7fa17bd3836c9ad8884f27be84a.json +++ b/backend/.sqlx/query-b3c02fd225a6aa78785d466e7f033b38deb9c7fa17bd3836c9ad8884f27be84a.json @@ -31,6 +31,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090.json b/backend/.sqlx/query-bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090.json new file mode 100644 index 0000000000..d9ca7cfa52 --- /dev/null +++ b/backend/.sqlx/query-bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM job_result_stream_v2 WHERE job_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "bb46e5dcf5490ef3511faa131ad5693dedf34366e51044ddf30695995d194090" +} diff --git a/backend/.sqlx/query-dd92bce0ddbfbf06834807aab4d589fc104647cae7abf37ddce3ef7109726261.json b/backend/.sqlx/query-dd92bce0ddbfbf06834807aab4d589fc104647cae7abf37ddce3ef7109726261.json new file mode 100644 index 0000000000..d30b0bae62 --- /dev/null +++ b/backend/.sqlx/query-dd92bce0ddbfbf06834807aab4d589fc104647cae7abf37ddce3ef7109726261.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n string_agg(stream, '' order by idx asc) as stream, \n max(idx) + 1 as offset \n FROM job_result_stream_v2\n WHERE job_id = $2 AND idx >= $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "stream", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "offset", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Int4", + "Uuid" + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "dd92bce0ddbfbf06834807aab4d589fc104647cae7abf37ddce3ef7109726261" +} diff --git a/backend/.sqlx/query-f17f914d2522bf7cb5de9d7ba5557ee0dce940039ab42fd39bf079d87b6cad8a.json b/backend/.sqlx/query-f17f914d2522bf7cb5de9d7ba5557ee0dce940039ab42fd39bf079d87b6cad8a.json deleted file mode 100644 index 5afdd9726a..0000000000 --- a/backend/.sqlx/query-f17f914d2522bf7cb5de9d7ba5557ee0dce940039ab42fd39bf079d87b6cad8a.json +++ /dev/null @@ -1,29 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT SUBSTR(rs.stream, $1) AS new_result_stream, CHAR_LENGTH(rs.stream) + 1 AS stream_offset FROM job_result_stream rs WHERE rs.job_id = $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "new_result_stream", - "type_info": "Text" - }, - { - "ordinal": 1, - "name": "stream_offset", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int4", - "Uuid" - ] - }, - "nullable": [ - null, - null - ] - }, - "hash": "f17f914d2522bf7cb5de9d7ba5557ee0dce940039ab42fd39bf079d87b6cad8a" -} diff --git a/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json b/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json index 28c0756b0f..13a5f5d76b 100644 --- a/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json +++ b/backend/.sqlx/query-f3f96e066716e61042519a645d487b578bc63792cdb0f7ddaeb82e9771287c22.json @@ -66,6 +66,11 @@ "appdependencies", "deploymentcallback", "singlescriptflow", + "scripttriggertransformer", + "flowtriggertransformer", + "scriptwithpreprocessing", + "scriptwithpreprocessor", + "flowwithpreprocessing", "flowscript", "flownode", "appscript", diff --git a/backend/.sqlx/query-f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571.json b/backend/.sqlx/query-f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571.json new file mode 100644 index 0000000000..70fbb752eb --- /dev/null +++ b/backend/.sqlx/query-f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, concurrency_key, log_file, metrics", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "f8ac5379ecfbff7b8ae75c821680737b249a64c8d9e8f7dbcc46fce98e874571" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index b6e5157568..7fa3c692b1 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -aa4c7936b99d0de2f7af31cde79dfc7b1b482c29 \ No newline at end of file +ce85973ea37c951bb7e3260ae1c262fbc2011f76 \ No newline at end of file diff --git a/backend/migrations/20250924150953_new_job_result_stream_table.down.sql b/backend/migrations/20250924150953_new_job_result_stream_table.down.sql new file mode 100644 index 0000000000..5d3cb6c6fe --- /dev/null +++ b/backend/migrations/20250924150953_new_job_result_stream_table.down.sql @@ -0,0 +1,4 @@ +-- Add down migration script here +DROP TABLE job_result_stream_v2; + +ALTER TABLE job_result_stream ADD CONSTRAINT fk_job_result_stream_job_id FOREIGN KEY (job_id) REFERENCES v2_job_queue(id) ON DELETE CASCADE; \ No newline at end of file diff --git a/backend/migrations/20250924150953_new_job_result_stream_table.up.sql b/backend/migrations/20250924150953_new_job_result_stream_table.up.sql new file mode 100644 index 0000000000..c5230e0645 --- /dev/null +++ b/backend/migrations/20250924150953_new_job_result_stream_table.up.sql @@ -0,0 +1,13 @@ +-- Add up migration script here +CREATE TABLE job_result_stream_v2 ( + job_id UUID NOT NULL, + workspace_id TEXT NOT NULL, + stream TEXT NOT NULL, + idx INT NOT NULL, + PRIMARY KEY (job_id, idx) +); + +GRANT ALL ON TABLE job_result_stream_v2 TO windmill_admin; +GRANT ALL ON TABLE job_result_stream_v2 TO windmill_user; + +ALTER TABLE job_result_stream DROP CONSTRAINT fk_job_result_stream_job_id; \ No newline at end of file diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index a0900da40e..c292c085d1 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -950,6 +950,17 @@ pub async fn delete_expired_items(db: &DB) -> () { { tracing::error!("Error deleting job: {:?}", e); } + + // should already be deleted but just in case + if let Err(e) = sqlx::query!( + "DELETE FROM job_result_stream_v2 WHERE job_id = ANY($1)", + &deleted_jobs + ) + .execute(&mut *tx) + .await + { + tracing::error!("Error deleting job result stream: {:?}", e); + } } } Err(e) => { @@ -1654,7 +1665,7 @@ pub async fn monitor_db( } async fn vacuuming_tables(db: &Pool) -> error::Result<()> { - sqlx::query!("VACUUM v2_job, v2_job_completed, job_result_stream, job_stats, job_logs, concurrency_key, log_file, metrics") + sqlx::query!("VACUUM v2_job, v2_job_completed, job_result_stream_v2, job_stats, job_logs, concurrency_key, log_file, metrics") .execute(db) .await?; Ok(()) diff --git a/backend/update_sqlx.sh b/backend/update_sqlx.sh index 77b772c970..8ab84c6306 100755 --- a/backend/update_sqlx.sh +++ b/backend/update_sqlx.sh @@ -21,9 +21,17 @@ if [[ "$(uname)" == "Darwin" ]]; then sed -i '' 's/^samael = { version="0.0.14", features = \["xmlsec"\] }/#samael = { version="0.0.14", features = ["xmlsec"] }/' Cargo.toml # Uncomment the git-based samael dependency sed -i '' 's/^# \(samael = { git="https:\/\/github.com\/njaremko\/samael", rev="464d015e3ae393e4b5dd00b4d6baa1b617de0dd6", features = \["xmlsec"\] }\)/\1/' Cargo.toml + + # Run cargo sqlx prepare with deno_core_mac + echo "Running cargo sqlx prepare with deno_core_mac..." + cargo sqlx prepare --workspace -- --all-targets --features all_sqlx_features,private,deno_core_mac +else + # Run cargo sqlx prepare + echo "Running cargo sqlx prepare..." + cargo sqlx prepare --workspace -- --all-targets --features all_sqlx_features,private fi -cargo sqlx prepare --workspace -- --all-targets --features all_sqlx_features,private + # Undo the samael changes on macOS if [[ "$(uname)" == "Darwin" ]]; then diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 596e404629..e7ed78b257 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -6990,14 +6990,20 @@ async fn get_flow_stream_delta( ) -> error::Result, Option)>> { if let Some(job_id) = flow_stream_job_id { let record = sqlx::query!( - "SELECT SUBSTR(rs.stream, $1) AS new_result_stream, CHAR_LENGTH(rs.stream) + 1 AS stream_offset FROM job_result_stream rs WHERE rs.job_id = $2", + " + SELECT + string_agg(stream, '' order by idx asc) as stream, + max(idx) + 1 as offset + FROM job_result_stream_v2 + WHERE job_id = $2 AND idx >= $1 + ", stream_offset.unwrap_or(0), job_id, ) .fetch_optional(db) .await?; if let Some(record) = record { - Ok(Some((record.new_result_stream, record.stream_offset))) + Ok(Some((record.stream, record.offset))) } else { Ok(None) } @@ -7046,18 +7052,28 @@ async fn get_job_update_data( let (result, running, mut result_stream, mut new_stream_offset, new_flow_stream_job_id) = if let Some(tags) = tags { let r = sqlx::query!( - "SELECT + " + WITH result_stream AS ( + SELECT + string_agg(stream, '' order by idx asc) as stream, + job_id, + max(idx) + 1 as offset + FROM job_result_stream_v2 + WHERE job_id = $2 AND idx >= $3 + GROUP BY job_id + ) + SELECT jc.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, + rs.stream AS \"result_stream: Option\", + rs.offset AS stream_offset, CASE WHEN $4 THEN NULL ELSE (COALESCE(js.flow_status, jc.flow_status)->>'stream_job')::uuid END as stream_job FROM v2_job LEFT JOIN v2_job_queue USING (id) LEFT JOIN v2_job_completed jc USING (id) LEFT JOIN v2_job_status js USING (id) - LEFT JOIN job_result_stream rs ON rs.job_id = $2 + LEFT JOIN result_stream rs ON rs.job_id = $2 WHERE v2_job.id = $2 AND v2_job.workspace_id = $1", w_id, job_id, @@ -7086,11 +7102,21 @@ async fn get_job_update_data( } else { if running.is_some_and(|x| !x) { let r = sqlx::query!( - "SELECT + " + WITH result_stream AS ( + SELECT + string_agg(stream, '' order by idx asc) as stream, + job_id, + max(idx) + 1 as offset + FROM job_result_stream_v2 + WHERE job_id = $1 AND idx >= $3 + GROUP BY job_id + ) + SELECT COALESCE(jc.result, jc.result) as \"result: sqlx::types::Json>\", jq.running as \"running: Option\", - SUBSTR(rs.stream, $3) AS \"result_stream: Option\", - CHAR_LENGTH(rs.stream) + 1 AS stream_offset, + rs.stream AS \"result_stream: Option\", + rs.offset AS stream_offset, CASE WHEN $4 THEN NULL ELSE (COALESCE(js.flow_status, jc.flow_status)->>'stream_job')::uuid END as stream_job FROM ( SELECT $1::uuid as job_id, $2::text as workspace_id @@ -7098,7 +7124,7 @@ async fn get_job_update_data( LEFT JOIN v2_job_completed jc ON jc.id = base.job_id AND jc.workspace_id = base.workspace_id LEFT JOIN v2_job_queue jq ON jq.id = base.job_id AND jq.workspace_id = base.workspace_id LEFT JOIN v2_job_status js ON js.id = base.job_id - LEFT JOIN job_result_stream rs ON rs.job_id = base.job_id + LEFT JOIN result_stream rs ON rs.job_id = base.job_id WHERE base.job_id = $1", job_id, w_id, @@ -7119,10 +7145,20 @@ async fn get_job_update_data( } } else { let q = sqlx::query!( - "SELECT + " + WITH result_stream AS ( + SELECT + string_agg(stream, '' order by idx asc) as stream, + job_id, + max(idx) + 1 as offset + FROM job_result_stream_v2 + WHERE job_id = $2 AND idx >= $3 + GROUP BY job_id + ) + 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, + rs.stream AS \"result_stream: Option\", + rs.offset AS stream_offset, COALESCE(js.flow_status, jc.flow_status) as \"flow_status: sqlx::types::Json>\", CASE WHEN $4 THEN NULL ELSE (COALESCE(js.flow_status, jc.flow_status)->>'stream_job')::uuid END as stream_job FROM ( @@ -7130,7 +7166,7 @@ async fn get_job_update_data( ) base LEFT JOIN v2_job_completed jc ON jc.id = base.job_id AND jc.workspace_id = base.workspace_id LEFT JOIN v2_job_status js ON js.id = base.job_id - LEFT JOIN job_result_stream rs ON rs.job_id = base.job_id + LEFT JOIN result_stream rs ON rs.job_id = base.job_id WHERE base.job_id = $2", w_id, job_id, @@ -7180,20 +7216,30 @@ async fn get_job_update_data( }) } else { let mut record = sqlx::query!( - "SELECT + " + WITH result_stream AS ( + SELECT + string_agg(stream, '' order by idx asc) as stream, + job_id, + max(idx) + 1 as offset + FROM job_result_stream_v2 + WHERE job_id = $3 AND idx >= $8 + GROUP BY job_id + ) + SELECT c.id IS NOT NULL AS completed, 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, CASE WHEN $7::BOOLEAN THEN NULL ELSE SUBSTR(logs, GREATEST($1 - log_offset, 0)) END AS logs, - SUBSTR(rs.stream, $8) AS new_result_stream, + rs.stream AS new_result_stream, COALESCE(r.memory_peak, c.memory_peak) AS mem_peak, COALESCE(c.flow_status, f.flow_status) AS \"flow_status: sqlx::types::Json>\", (COALESCE(c.flow_status, f.flow_status)->>'stream_job')::uuid AS stream_job, COALESCE(c.workflow_as_code_status, f.workflow_as_code_status) AS \"workflow_as_code_status: sqlx::types::Json>\", CASE WHEN $7::BOOLEAN THEN NULL ELSE job_logs.log_offset + CHAR_LENGTH(job_logs.logs) + 1 END AS log_offset, - CHAR_LENGTH(rs.stream) + 1 AS stream_offset, + rs.offset AS stream_offset, created_by AS \"created_by!\", CASE WHEN $4::BOOLEAN THEN ( SELECT scalar_int FROM job_stats WHERE job_id = $3 AND metric_id = 'progress_perc' @@ -7204,7 +7250,7 @@ async fn get_job_update_data( LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status f USING (id) LEFT JOIN v2_job_completed c USING (id) - LEFT JOIN job_result_stream rs ON rs.job_id = $3 + LEFT JOIN result_stream rs ON rs.job_id = $3 LEFT JOIN job_logs ON job_logs.job_id = $3 WHERE j.workspace_id = $2 AND j.id = $3 AND ($6::text[] IS NULL OR j.tag = ANY($6))", diff --git a/backend/windmill-common/src/result_stream.rs b/backend/windmill-common/src/result_stream.rs index d44933891d..02bfa75b92 100644 --- a/backend/windmill-common/src/result_stream.rs +++ b/backend/windmill-common/src/result_stream.rs @@ -19,17 +19,24 @@ pub async fn append_result_stream_db( workspace_id: &str, job_id: &Uuid, nstream: &str, + offset: i32, ) -> error::Result<()> { if !nstream.is_empty() { sqlx::query!( r#" - INSERT INTO job_result_stream (workspace_id, job_id, stream) - VALUES ($1, $2, $3) - ON CONFLICT (job_id) DO UPDATE SET stream = job_result_stream.stream || $3 + INSERT INTO job_result_stream_v2 (workspace_id, job_id, stream, idx) + VALUES ( + $1, + $2, + $3, + $4 + ) + ON CONFLICT (job_id, idx) DO UPDATE SET stream = job_result_stream_v2.stream || $3 "#, workspace_id, job_id, nstream, + offset ) .execute(db) .await?; diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 7c6a3b3dcd..8a22e9d4cb 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -147,6 +147,7 @@ pub struct JobCompleted { pub token: String, pub canceled_by: Option, pub duration: Option, + pub has_stream: bool, } pub async fn cancel_single_job<'c>( @@ -737,6 +738,7 @@ pub async fn add_completed_job_error( canceled_by, flow_is_done, duration, + false, ) .await?; Ok(result) @@ -758,6 +760,7 @@ pub async fn add_completed_job( canceled_by: Option, flow_is_done: bool, duration: Option, + has_stream: bool, ) -> Result<(Uuid, i64), Error> { // tracing::error!("Start"); // let start = tokio::time::Instant::now(); @@ -782,6 +785,7 @@ pub async fn add_completed_job( &canceled_by, flow_is_done, duration, + has_stream, ) }) .retry( @@ -834,6 +838,7 @@ async fn commit_completed_job( canceled_by: &Option, flow_is_done: bool, duration: Option, + has_stream: bool, ) -> windmill_common::error::Result<(Option, i64, bool)> { // let start = std::time::Instant::now(); @@ -1116,6 +1121,12 @@ async fn commit_completed_job( .execute(&mut *tx) .await?; + if !success || has_stream { + sqlx::query!("DELETE FROM job_result_stream_v2 WHERE job_id = $1", job_id) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; tracing::info!( diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 4016010b63..492b8f3beb 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -858,6 +858,7 @@ pub async fn handle_bun_job( new_args: &mut Option>>, occupancy_metrics: &mut OccupancyMetrics, precomputed_agent_info: Option, + has_stream: &mut bool, ) -> error::Result> { let mut annotation = windmill_common::worker::TypeScriptAnnotations::parse(inner_content); @@ -1331,6 +1332,7 @@ try {{ false, occupancy_metrics, stream_notifier, + has_stream, ) .await?; tracing::info!( @@ -1493,6 +1495,8 @@ try {{ ) .await?; + *has_stream = handle_result.result_stream.is_some(); + if apply_preprocessor { let args = read_file(&format!("{job_dir}/args.json")) .await diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index 1805b22230..142300e307 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -186,14 +186,14 @@ pub async fn handle_dedicated_process( let result = Arc::new(result); append_logs(&job.id, &job.workspace_id, logs.clone(), &db.into()).await; if line.starts_with("wm_res[success]:") { - job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None }, true).await.unwrap() + job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None, has_stream: false }, true).await.unwrap() } else { - job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None }, true).await.unwrap() + job_completed_tx.send_job(JobCompleted { job , result, result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None, has_stream: false }, true).await.unwrap() } }, Err(e) => { tracing::error!("Could not deserialize job result `{line}`: {e:?}"); - job_completed_tx.send_job(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))), result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None }, true).await.unwrap(); + job_completed_tx.send_job(JobCompleted { job , result: Arc::new(to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")}))), result_columns: None, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string(), duration: None, preprocessed_args: None, has_stream: false }, true).await.unwrap(); }, }; logs = init_log.clone(); diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 0e15e5ba12..3433eef3ae 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -194,6 +194,7 @@ pub async fn handle_deno_job( envs: HashMap, new_args: &mut Option>>, occupancy_metrics: &mut OccupancyMetrics, + has_stream: &mut bool, ) -> error::Result> { // let mut start = Instant::now(); let logs1 = "\n\n--- DENO CODE EXECUTION ---\n".to_string(); @@ -440,6 +441,9 @@ try {{ stream_notifier, ) .await?; + + *has_stream = handle_result.result_stream.is_some(); + // logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str()); if let Err(e) = tokio::fs::remove_dir_all(format!("{DENO_CACHE_DIR}/gen/file/{job_dir}")).await { diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 6d36b8ca0d..6c8f94d6ac 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -29,7 +29,7 @@ use windmill_queue::{append_logs, CanceledBy}; use std::os::unix::process::ExitStatusExt; use std::process::ExitStatus; -use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU32, Ordering}; use std::sync::Arc; use std::{io, panic, time::Duration}; @@ -388,6 +388,7 @@ pub async fn write_lines( let mut pipe_stdout = pipe_stdout; let is_stream = Arc::new(AtomicBool::new(false)); + let offset = Arc::new(AtomicI32::new(0)); while let Some(line) = output.by_ref().next().await { let do_write_ = do_write.shared(); @@ -486,6 +487,7 @@ pub async fn write_lines( let pg_log_total_size = pg_log_total_size.clone(); let stream_notifier = stream_notifier.clone(); let is_stream = is_stream.clone(); + let offset = offset.clone(); (do_write, write_result) = tokio::spawn(async move { if !nstream.is_empty() { if let Some(stream_notifier) = stream_notifier { @@ -493,9 +495,17 @@ pub async fn write_lines( is_stream.store(true, Ordering::SeqCst); stream_notifier.update_flow_status_with_stream_job(); } - } + }; - if let Err(err) = append_result_stream(&conn, &w_id, &job_id, &nstream).await { + if let Err(err) = append_result_stream( + &conn, + &w_id, + &job_id, + &nstream, + offset.fetch_add(1, Ordering::SeqCst), + ) + .await + { tracing::error!( "Unable to send result stream for job {job_id}. Error was: {:?}", err diff --git a/backend/windmill-worker/src/job_logger.rs b/backend/windmill-worker/src/job_logger.rs index 263234cee9..11e7785ee5 100644 --- a/backend/windmill-worker/src/job_logger.rs +++ b/backend/windmill-worker/src/job_logger.rs @@ -8,6 +8,7 @@ use windmill_common::worker::{Connection, CLOUD_HOSTED}; use windmill_common::{error, DB}; use windmill_queue::append_logs; +use serde::Serialize; use std::sync::atomic::AtomicU32; use std::sync::Arc; @@ -62,22 +63,33 @@ pub async fn append_job_logs( } } +#[derive(Serialize)] +struct ResultStreamBody<'a> { + result_stream: &'a str, + offset: i32, +} + pub async fn append_result_stream( conn: &Connection, workspace_id: &str, job_id: &Uuid, nstream: &str, + offset: i32, ) -> error::Result<()> { match conn { Connection::Sql(db) => { - append_result_stream_db(db, workspace_id, job_id, nstream).await?; + append_result_stream_db(db, workspace_id, job_id, nstream, offset).await?; } Connection::Http(client) => { + let body = ResultStreamBody { result_stream: nstream, offset }; if let Err(e) = client .post::<_, String>( - &format!("/api/w/{}/agent_workers/push_logs/{}", workspace_id, job_id), + &format!( + "/api/w/{}/agent_workers/push_result_stream/{}", + workspace_id, job_id + ), None, - &nstream, + &body, ) .await { diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index cbf19b5946..b935b017f9 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -770,6 +770,7 @@ pub async fn eval_fetch_timeout( _load_client: bool, _occupation_metrics: &mut OccupancyMetrics, _stream_notifier: Option, + _has_stream: &mut bool, ) -> anyhow::Result> { use serde_json::value::to_raw_value; Ok(to_raw_value("require deno_core").unwrap()) @@ -792,6 +793,7 @@ pub async fn eval_fetch_timeout( load_client: bool, occupation_metrics: &mut OccupancyMetrics, stream_notifier: Option, + has_stream: &mut bool, ) -> windmill_common::error::Result> { let (sender, mut receiver) = oneshot::channel::(); let (append_logs_sender, mut append_logs_receiver) = mpsc::unbounded_channel::(); @@ -808,9 +810,14 @@ pub async fn eval_fetch_timeout( let conn_ = conn.clone(); let w_id_ = w_id.to_string(); tokio::spawn(async move { + let mut offset = 0; while let Some(stream) = result_stream_receiver.recv().await { use crate::job_logger::append_result_stream; - if let Err(e) = append_result_stream(&conn_, &w_id_, &job_id, &stream).await { + let curr_offset = offset.clone(); + offset += 1; + if let Err(e) = + append_result_stream(&conn_, &w_id_, &job_id, &stream, curr_offset).await + { tracing::error!("failed to append result stream: {e}"); } } @@ -978,22 +985,24 @@ pub async fn eval_fetch_timeout( drop(js_runtime); if let Ok(r) = r { match handle.await { - Ok(Some(logs)) => Ok(merge_result_stream(r, Some(logs)).await), - Ok(None) => Ok(r), + Ok(Some(logs)) => { + Ok(merge_result_stream(r, Some(logs)).await.map(|r| (r, true))) + } + Ok(None) => Ok(r.map(|r| (r, false))), Err(e) => Err(Error::ExecutionErr(e.to_string())), } } else { - r + r.map(|r| r.map(|r| (r, false))) } // r }; let r = runtime.block_on(future)?; // tracing::info!("total: {:?}", instant.elapsed()); - r as windmill_common::error::Result> + r as windmill_common::error::Result<(Box, bool)> }); - let res = run_future_with_polling_update_job_poller( + let (res, new_has_stream) = run_future_with_polling_update_job_poller( job_id, job_timeout, conn, @@ -1012,6 +1021,7 @@ pub async fn eval_fetch_timeout( } e })?; + *has_stream = new_has_stream; *mem_peak = (res.get().len() / 1000) as i32; Ok(res) } diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index b31de6e184..5b742feb4b 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -544,6 +544,7 @@ pub async fn handle_python_job( new_args: &mut Option>>, occupancy_metrics: &mut OccupancyMetrics, precomputed_agent_info: Option, + has_stream: &mut bool, ) -> windmill_common::error::Result> { let script_path = crate::common::use_flow_root_path(job.runnable_path()); @@ -885,6 +886,8 @@ mount {{ ) .await?; + *has_stream = handle_result.result_stream.is_some(); + if apply_preprocessor { let args = read_file(&format!("{job_dir}/args.json")) .await diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 144c8d9101..0577eb6638 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -17,7 +17,14 @@ use windmill_common::otel_oss::FutureExt; use uuid::Uuid; use windmill_common::{ - add_time, error::{self, Error}, flow_status::{FlowJobDuration}, jobs::JobKind, utils::WarnAfterExt, worker::{to_raw_value, Connection, WORKER_GROUP}, worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, KillpillSender, DB + add_time, + error::{self, Error}, + flow_status::FlowJobDuration, + jobs::JobKind, + utils::WarnAfterExt, + worker::{to_raw_value, Connection, WORKER_GROUP}, + worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, + KillpillSender, DB, }; #[cfg(feature = "benchmark")] @@ -405,6 +412,7 @@ pub async fn process_result( preprocessed_args: Option>>, conn: &Connection, duration: Option, + has_stream: bool, ) -> error::Result { match result { Ok(result) => { @@ -421,6 +429,7 @@ pub async fn process_result( cached_res_path, token: token.to_string(), duration, + has_stream, }, ) .with_context(windmill_common::otel_oss::otel_ctx()) @@ -483,6 +492,7 @@ pub async fn process_result( cached_res_path, token: token.to_string(), duration, + has_stream, }, ) .with_context(windmill_common::otel_oss::otel_ctx()) @@ -557,6 +567,7 @@ pub async fn process_completed_job( duration, result_columns, preprocessed_args, + has_stream, .. }: JobCompleted, client: &AuthedClient, @@ -621,6 +632,7 @@ pub async fn process_completed_job( canceled_by, false, duration, + has_stream, ) .await?; drop(job); @@ -679,7 +691,10 @@ pub async fn process_completed_job( &job.workspace_id, false, Arc::new(serde_json::value::to_raw_value(&result).unwrap()), - duration.map(|x| FlowJobDuration { started_at: job.started_at.unwrap(), duration_ms: x }), + duration.map(|x| FlowJobDuration { + started_at: job.started_at.unwrap(), + duration_ms: x, + }), false, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(), &worker_dir, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 06a295f820..dea40e8bf8 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -765,6 +765,7 @@ pub async fn handle_all_job_kind_error( cached_res_path: None, token: authed_client.token.clone(), duration: None, + has_stream: false, }, false, ) @@ -1687,6 +1688,7 @@ pub async fn run_worker( token: "".to_string(), canceled_by: None, duration: None, + has_stream: false, }, true, ) @@ -2229,6 +2231,7 @@ async fn do_nativets( canceled_by: &mut Option, worker_name: &str, occupancy_metrics: &mut OccupancyMetrics, + has_stream: &mut bool, ) -> windmill_common::error::Result> { let args = build_args_map(job, client, conn).await?.map(Json); let job_args = if args.is_some() { @@ -2255,6 +2258,7 @@ async fn do_nativets( true, occupancy_metrics, stream_notifier, + has_stream, ) .await?) } @@ -2408,6 +2412,7 @@ pub async fn handle_queued_job( cached_res_path: None, token: client.token.clone(), duration: None, + has_stream: false, }, true, ) @@ -2491,6 +2496,7 @@ pub async fn handle_queued_job( let mut column_order: Option> = None; let mut new_args: Option>> = None; + let mut has_stream = false; let result = match job.kind { JobKind::Dependencies => match conn { Connection::Sql(db) => { @@ -2614,6 +2620,7 @@ pub async fn handle_queued_job( occupancy_metrics, killpill_rx, precomputed_agent_info, + &mut has_stream, ) .await; occupancy_metrics.total_duration_of_running_jobs += @@ -2646,6 +2653,7 @@ pub async fn handle_queued_job( new_args, conn, Some(started.elapsed().as_millis() as i64), + has_stream, ) .await } @@ -2833,6 +2841,7 @@ async fn handle_code_execution_job( occupancy_metrics: &mut OccupancyMetrics, killpill_rx: &mut tokio::sync::broadcast::Receiver<()>, precomputed_agent_info: Option, + has_stream: &mut bool, ) -> error::Result> { let script_hash = || { job.runnable_id @@ -3172,6 +3181,7 @@ async fn handle_code_execution_job( canceled_by, worker_name, occupancy_metrics, + has_stream, ) .await?; return Ok(result); @@ -3245,6 +3255,7 @@ mount {{ new_args, occupancy_metrics, precomputed_agent_info, + has_stream, ) .await } @@ -3264,6 +3275,7 @@ mount {{ envs, new_args, occupancy_metrics, + has_stream, ) .await } @@ -3286,6 +3298,7 @@ mount {{ new_args, occupancy_metrics, precomputed_agent_info, + has_stream, ) .await } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index afd2b6e850..8aafb17817 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1473,6 +1473,7 @@ pub async fn update_flow_status_after_job_completion_internal( None, true, None, + false, ) .await?; duration @@ -1492,6 +1493,7 @@ pub async fn update_flow_status_after_job_completion_internal( None, true, None, + false, ) .await?; duration diff --git a/frontend/src/lib/components/JobLoader.svelte b/frontend/src/lib/components/JobLoader.svelte index ecdd8df69f..0341f9c764 100644 --- a/frontend/src/lib/components/JobLoader.svelte +++ b/frontend/src/lib/components/JobLoader.svelte @@ -254,9 +254,6 @@ if (logOffset == 0) { logOffset = job?.logs?.length ? job.logs?.length + 1 : 0 } - if (resultStreamOffset == 0) { - resultStreamOffset = job?.result_stream?.length ? job.result_stream?.length + 1 : 0 - } } export async function getLogs() { if (job) { @@ -496,6 +493,7 @@ id, running: job.running, logOffset: logOffset, + streamOffset: resultStreamOffset, getProgress: getProgress })