diff --git a/backend/.sqlx/query-0997b46bae6e2374b568e8367898d6fc79c331431326250c1a59674054ceaabd.json b/backend/.sqlx/query-0997b46bae6e2374b568e8367898d6fc79c331431326250c1a59674054ceaabd.json new file mode 100644 index 0000000000..ef5348cf50 --- /dev/null +++ b/backend/.sqlx/query-0997b46bae6e2374b568e8367898d6fc79c331431326250c1a59674054ceaabd.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT parent_job FROM v2_job WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "parent_job", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + true + ] + }, + "hash": "0997b46bae6e2374b568e8367898d6fc79c331431326250c1a59674054ceaabd" +} diff --git a/backend/.sqlx/query-2e589e039986e7a2c75e328868874669b32cbe0dae6822b2d2fad0635c5f6087.json b/backend/.sqlx/query-2e589e039986e7a2c75e328868874669b32cbe0dae6822b2d2fad0635c5f6087.json new file mode 100644 index 0000000000..6ba559e644 --- /dev/null +++ b/backend/.sqlx/query-2e589e039986e7a2c75e328868874669b32cbe0dae6822b2d2fad0635c5f6087.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE v2_job_status\n SET flow_status = jsonb_set(flow_status, array['stream_job'], to_jsonb($1::UUID::TEXT))\n WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "2e589e039986e7a2c75e328868874669b32cbe0dae6822b2d2fad0635c5f6087" +} diff --git a/backend/.sqlx/query-70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393.json b/backend/.sqlx/query-40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773.json similarity index 61% rename from backend/.sqlx/query-70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393.json rename to backend/.sqlx/query-40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773.json index 82333559eb..e7d66a3d75 100644 --- a/backend/.sqlx/query-70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393.json +++ b/backend/.sqlx/query-40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773.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.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": [ { @@ -35,31 +35,36 @@ }, { "ordinal": 6, + "name": "stream_job", + "type_info": "Uuid" + }, + { + "ordinal": 7, "name": "workflow_as_code_status: sqlx::types::Json>", "type_info": "Jsonb" }, { - "ordinal": 7, + "ordinal": 8, "name": "log_offset", "type_info": "Int4" }, { - "ordinal": 8, + "ordinal": 9, "name": "stream_offset", "type_info": "Int4" }, { - "ordinal": 9, + "ordinal": 10, "name": "created_by!", "type_info": "Varchar" }, { - "ordinal": 10, + "ordinal": 11, "name": "progress", "type_info": "Int4" }, { - "ordinal": 11, + "ordinal": 12, "name": "result_stream: Option", "type_info": "Text" } @@ -86,10 +91,11 @@ null, null, null, + null, false, null, false ] }, - "hash": "70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393" + "hash": "40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773" } diff --git a/backend/.sqlx/query-69924462c788dbc8f31aacc7f8ae588d76bf1f25d631833ce4b194818a7d1437.json b/backend/.sqlx/query-69924462c788dbc8f31aacc7f8ae588d76bf1f25d631833ce4b194818a7d1437.json deleted file mode 100644 index f25cfd9eba..0000000000 --- a/backend/.sqlx/query-69924462c788dbc8f31aacc7f8ae588d76bf1f25d631833ce4b194818a7d1437.json +++ /dev/null @@ -1,48 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT result as \"result: sqlx::types::Json>\", v2_job.tag,\n v2_job_queue.running as \"running: Option\", SUBSTR(rs.stream, $3) AS \"result_stream: Option\", CHAR_LENGTH(rs.stream) AS stream_offset\n FROM v2_job\n LEFT JOIN v2_job_queue USING (id)\n LEFT JOIN v2_job_completed 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" - } - ], - "parameters": { - "Left": [ - "Text", - "Uuid", - "Int4" - ] - }, - "nullable": [ - true, - false, - false, - null, - null - ] - }, - "hash": "69924462c788dbc8f31aacc7f8ae588d76bf1f25d631833ce4b194818a7d1437" -} diff --git a/backend/.sqlx/query-d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d.json b/backend/.sqlx/query-80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7.json similarity index 53% rename from backend/.sqlx/query-d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d.json rename to backend/.sqlx/query-80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7.json index 76c37b7a08..7254b7e84f 100644 --- a/backend/.sqlx/query-d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d.json +++ b/backend/.sqlx/query-80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7.json @@ -1,6 +1,6 @@ { "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", + "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": [ { @@ -17,20 +17,33 @@ "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" + "Int4", + "Bool" ] }, "nullable": [ + null, + null, null, null, null ] }, - "hash": "d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d" + "hash": "80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7" } diff --git a/backend/.sqlx/query-8126b118704341846e88bd289f1afe83c07b7a8b422f48022994370b3e433f34.json b/backend/.sqlx/query-8126b118704341846e88bd289f1afe83c07b7a8b422f48022994370b3e433f34.json new file mode 100644 index 0000000000..abc63b0e90 --- /dev/null +++ b/backend/.sqlx/query-8126b118704341846e88bd289f1afe83c07b7a8b422f48022994370b3e433f34.json @@ -0,0 +1,55 @@ +{ + "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-6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d.json b/backend/.sqlx/query-a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05.json similarity index 51% rename from backend/.sqlx/query-6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d.json rename to backend/.sqlx/query-a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05.json index 2f14e723e4..4cca74dd4b 100644 --- a/backend/.sqlx/query-6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d.json +++ b/backend/.sqlx/query-a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05.json @@ -1,6 +1,6 @@ { "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", + "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": [ { @@ -22,21 +22,28 @@ "ordinal": 3, "name": "stream_offset", "type_info": "Int4" + }, + { + "ordinal": 4, + "name": "stream_job", + "type_info": "Uuid" } ], "parameters": { "Left": [ "Uuid", "Text", - "Int4" + "Int4", + "Bool" ] }, "nullable": [ null, false, null, + null, null ] }, - "hash": "6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d" + "hash": "a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05" } diff --git a/backend/.sqlx/query-d8ef35b4990eb9b2a306494d5b9acde9f57cfaba047c5b7680e0dccd2c1507df.json b/backend/.sqlx/query-d8ef35b4990eb9b2a306494d5b9acde9f57cfaba047c5b7680e0dccd2c1507df.json new file mode 100644 index 0000000000..43ce3fada4 --- /dev/null +++ b/backend/.sqlx/query-d8ef35b4990eb9b2a306494d5b9acde9f57cfaba047c5b7680e0dccd2c1507df.json @@ -0,0 +1,40 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT \n (flow_status->'step')::integer as step,\n jsonb_array_length(flow_status->'modules') as len,\n flow_status->'modules'->-1->>'branch_chosen' IS NOT NULL as is_branch_one,\n parent_job as ppp_job\n FROM v2_job \n LEFT JOIN v2_job_status USING (id)\n WHERE v2_job.id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "step", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "len", + "type_info": "Int4" + }, + { + "ordinal": 2, + "name": "is_branch_one", + "type_info": "Bool" + }, + { + "ordinal": 3, + "name": "ppp_job", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null, + null, + null, + true + ] + }, + "hash": "d8ef35b4990eb9b2a306494d5b9acde9f57cfaba047c5b7680e0dccd2c1507df" +} diff --git a/backend/.sqlx/query-f17f914d2522bf7cb5de9d7ba5557ee0dce940039ab42fd39bf079d87b6cad8a.json b/backend/.sqlx/query-f17f914d2522bf7cb5de9d7ba5557ee0dce940039ab42fd39bf079d87b6cad8a.json new file mode 100644 index 0000000000..5afdd9726a --- /dev/null +++ b/backend/.sqlx/query-f17f914d2522bf7cb5de9d7ba5557ee0dce940039ab42fd39bf079d87b6cad8a.json @@ -0,0 +1,29 @@ +{ + "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/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index b6ca1b56c8..59f8ab321b 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -1306,8 +1306,6 @@ paths: type: string endpoint_sync: type: string - endpoint_openai_sync: - type: string summary: type: string description: @@ -1318,7 +1316,6 @@ paths: - workspace - endpoint_async - endpoint_sync - - endpoint_openai_sync - summary - kind @@ -5606,35 +5603,6 @@ paths: type: string format: uuid - /w/{workspace}/jobs/openai_sync/p/{path}: - post: - summary: run script by path in openai format - operationId: openaiSyncScriptByPath - tags: - - job - parameters: - - $ref: "#/components/parameters/WorkspaceId" - - $ref: "#/components/parameters/ScriptPath" - - $ref: "#/components/parameters/ParentJob" - - $ref: "#/components/parameters/NewJobId" - - $ref: "#/components/parameters/IncludeHeader" - - $ref: "#/components/parameters/QueueLimit" - - requestBody: - description: script args - required: true - content: - application/json: - schema: - $ref: "#/components/schemas/ScriptArgs" - - responses: - "200": - description: job result - content: - application/json: - schema: {} - /w/{workspace}/jobs/run_wait_result/p/{path}: post: summary: run script by path @@ -5650,6 +5618,7 @@ paths: - $ref: "#/components/parameters/NewJobId" - $ref: "#/components/parameters/IncludeHeader" - $ref: "#/components/parameters/QueueLimit" + - $ref: "#/components/parameters/SkipPreprocessor" requestBody: description: script args @@ -5681,34 +5650,7 @@ paths: - $ref: "#/components/parameters/IncludeHeader" - $ref: "#/components/parameters/QueueLimit" - $ref: "#/components/parameters/Payload" - - responses: - "200": - description: job result - content: - application/json: - schema: {} - - /w/{workspace}/jobs/openai_sync/f/{path}: - post: - summary: run flow by path and wait until completion in openai format - operationId: openaiSyncFlowByPath - tags: - - job - parameters: - - $ref: "#/components/parameters/WorkspaceId" - - $ref: "#/components/parameters/ScriptPath" - - $ref: "#/components/parameters/IncludeHeader" - - $ref: "#/components/parameters/QueueLimit" - - $ref: "#/components/parameters/NewJobId" - - requestBody: - description: script args - required: true - content: - application/json: - schema: - $ref: "#/components/schemas/ScriptArgs" + - $ref: "#/components/parameters/SkipPreprocessor" responses: "200": @@ -5729,6 +5671,7 @@ paths: - $ref: "#/components/parameters/IncludeHeader" - $ref: "#/components/parameters/QueueLimit" - $ref: "#/components/parameters/NewJobId" + - $ref: "#/components/parameters/SkipPreprocessor" requestBody: description: script args @@ -7114,11 +7057,7 @@ paths: in: query schema: type: integer - - name: skip_preprocessor - description: skip the preprocessor - in: query - schema: - type: boolean + - $ref: "#/components/parameters/SkipPreprocessor" - $ref: "#/components/parameters/ParentJob" - $ref: "#/components/parameters/WorkerTag" - $ref: "#/components/parameters/NewJobId" @@ -7276,11 +7215,7 @@ paths: in: query schema: type: integer - - name: skip_preprocessor - description: skip the preprocessor - in: query - schema: - type: boolean + - $ref: "#/components/parameters/SkipPreprocessor" - $ref: "#/components/parameters/ParentJob" - $ref: "#/components/parameters/WorkerTag" - $ref: "#/components/parameters/CacheTtl" @@ -14137,6 +14072,12 @@ components: in: query schema: type: string + SkipPreprocessor: + name: skip_preprocessor + description: skip the preprocessor + in: query + schema: + type: boolean Payload: name: payload description: | diff --git a/backend/windmill-api/src/auth.rs b/backend/windmill-api/src/auth.rs index 5f66a0deaf..00148a2961 100644 --- a/backend/windmill-api/src/auth.rs +++ b/backend/windmill-api/src/auth.rs @@ -415,6 +415,7 @@ pub struct Tokened { pub token: String, } +#[derive(Clone, Debug)] pub struct OptTokened { #[allow(dead_code)] pub token: Option, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index b8d51c3020..b00a08cdb5 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -44,6 +44,7 @@ use windmill_common::{email_oss::send_email_html, server::load_smtp_config}; use windmill_common::scripts::PREVIEW_IS_CODEBASE_HASH; use windmill_common::variables::get_workspace_key; +use crate::triggers::trigger_helpers::ScriptId; use crate::{ add_webhook_allowed_origin, args::{self, RawWebhookArgs}, @@ -171,6 +172,27 @@ pub fn workspaced_service() -> Router { .layer(cors.clone()) .layer(ce_headers.clone()), ) + .route( + "/run_and_stream/f/*script_path", + get(stream_flow_by_path) + .head(|| async { "" }) + .layer(cors.clone()) + .layer(ce_headers.clone()), + ) + .route( + "/run_and_stream/p/*script_path", + get(stream_script_by_path) + .head(|| async { "" }) + .layer(cors.clone()) + .layer(ce_headers.clone()), + ) + .route( + "/run_and_stream/h/:hash", + get(stream_script_by_hash) + .head(|| async { "" }) + .layer(cors.clone()) + .layer(ce_headers.clone()), + ) .route( "/run/h/:hash", post(run_job_by_hash) @@ -5145,6 +5167,174 @@ pub async fn run_wait_result_flow_by_path( .await } +pub async fn stream_flow_by_path( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Path((w_id, flow_path)): Path<(String, StripPath)>, + Query(run_query): Query, + args: RawWebhookArgs, +) -> error::Result { + stream_job( + authed, + db, + user_db, + w_id, + RunnableId::from_flow_path(flow_path.to_path()), + args, + run_query, + ) + .await +} + +pub async fn stream_script_by_path( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Path((w_id, script_path)): Path<(String, StripPath)>, + Query(run_query): Query, + args: RawWebhookArgs, +) -> error::Result { + stream_job( + authed, + db, + user_db, + w_id, + RunnableId::from_script_path(script_path.to_path()), + args, + run_query, + ) + .await +} + +pub async fn stream_script_by_hash( + authed: ApiAuthed, + Extension(db): Extension, + Extension(user_db): Extension, + Path((w_id, script_hash)): Path<(String, ScriptHash)>, + Query(run_query): Query, + args: RawWebhookArgs, +) -> error::Result { + stream_job( + authed, + db, + user_db, + w_id, + RunnableId::from_script_hash(script_hash), + args, + run_query, + ) + .await +} + +pub async fn stream_job( + authed: ApiAuthed, + db: DB, + user_db: UserDB, + w_id: String, + runnable_id: RunnableId, + args: RawWebhookArgs, + run_query: RunJobQuery, +) -> error::Result { + let payload_r = run_query.payload.clone().map(decode_payload).map(|x| { + x.map_err(|e| Error::internal_err(format!("Impossible to decode query payload: {e:#?}"))) + }); + + let payload_args = if let Some(payload) = payload_r { + payload? + } else { + HashMap::new() + }; + + let mut args = args.process_args(&authed, &db, &w_id, None).await?; + args.body = args::Body::HashMap(payload_args); + + let args = args + .to_args_from_runnable(&db, &w_id, runnable_id.clone(), run_query.skip_preprocessor) + .await?; + + let uuid = match runnable_id { + RunnableId::ScriptId(ScriptId::ScriptPath(script_path)) + | RunnableId::HubScript(script_path) => { + run_script_by_path_inner( + authed.clone(), + db.clone(), + user_db, + w_id.clone(), + StripPath(script_path), + run_query, + args, + ) + .await? + .0 + } + RunnableId::ScriptId(ScriptId::ScriptHash(script_hash)) => { + run_job_by_hash_inner( + authed.clone(), + db.clone(), + user_db, + w_id.clone(), + script_hash, + run_query, + args, + ) + .await? + .0 + } + RunnableId::FlowPath(flow_path) => { + run_flow_by_path_inner( + authed.clone(), + db.clone(), + user_db, + w_id.clone(), + StripPath(flow_path), + run_query, + args, + ) + .await? + .0 + } + }; + + let opt_authed = Some(authed.clone()); + let opt_tokened = OptTokened { token: None }; // ignored when authed is some + let (tx, rx) = tokio::sync::mpsc::channel(32); + + let stream = tokio_stream::wrappers::ReceiverStream::new(rx).map(|x| { + format!( + "data: {}\n\n", + serde_json::to_string(&x).unwrap_or_default() + ) + }); + + start_job_update_sse_stream( + opt_authed, + opt_tokened, + db, + w_id, + uuid, + None, + None, + None, + None, + Some(true), + Some(true), + None, + None, + tx, + ); + + let body = axum::body::Body::from_stream(stream.map(Result::<_, std::convert::Infallible>::Ok)); + + Ok(Response::builder() + .status(200) + .header("Content-Type", "text/event-stream") + .header("Cache-Control", "no-cache") + .header("Connection", "keep-alive") + .body(body) + .unwrap()) +} + pub async fn run_wait_result_flow_by_path_internal( db: sqlx::Pool, run_query: RunJobQuery, @@ -6303,6 +6493,7 @@ pub struct JobUpdateQuery { pub no_logs: Option, pub only_result: Option, pub fast: Option, + pub is_flow: Option, } #[derive(Serialize, Debug)] @@ -6331,6 +6522,8 @@ pub struct JobUpdate { pub job: Option, #[serde(skip_serializing_if = "Option::is_none")] pub only_result: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub flow_stream_job_id: Option, } impl JobUpdate { @@ -6349,6 +6542,7 @@ impl Hash for JobUpdate { self.mem_peak.hash(state); self.progress.hash(state); self.stream_offset.hash(state); + self.flow_stream_job_id.hash(state); if !self.completed.unwrap_or(false) { self.flow_status.as_ref().map(|x| x.get().hash(state)); self.workflow_as_code_status @@ -6421,6 +6615,7 @@ async fn get_job_update( running, only_result, no_logs, + is_flow, .. }): Query, ) -> JsonResult { @@ -6439,6 +6634,8 @@ async fn get_job_update( false, only_result, no_logs, + is_flow, + None, ) .await?, )) @@ -6457,9 +6654,12 @@ async fn get_job_update_sse( no_logs, only_result, fast, + is_flow, }): Query, -) -> Response { - let stream = get_job_update_sse_stream( +) -> error::Result { + let (tx, rx) = tokio::sync::mpsc::channel(32); + + start_job_update_sse_stream( opt_authed, opt_tokened, db, @@ -6472,8 +6672,11 @@ async fn get_job_update_sse( only_result, fast, no_logs, - ) - .map(|x| { + is_flow, + tx, + ); + + let stream = tokio_stream::wrappers::ReceiverStream::new(rx).map(|x| { format!( "data: {}\n\n", serde_json::to_string(&x).unwrap_or_default() @@ -6482,26 +6685,26 @@ async fn get_job_update_sse( let body = axum::body::Body::from_stream(stream.map(Result::<_, std::convert::Infallible>::Ok)); - Response::builder() + Ok(Response::builder() .status(200) .header("Content-Type", "text/event-stream") .header("Cache-Control", "no-cache") .header("Connection", "keep-alive") .body(body) - .unwrap() + .unwrap()) } #[derive(Serialize)] #[serde(tag = "type", rename_all = "lowercase")] enum JobUpdateSSEStream { Update(JobUpdate), - Error(String), + Error { error: String }, NotFound, Timeout, Ping, } -fn get_job_update_sse_stream( +fn start_job_update_sse_stream( opt_authed: Option, opt_tokened: OptTokened, db: DB, @@ -6514,18 +6717,18 @@ fn get_job_update_sse_stream( only_result: Option, fast: Option, no_logs: Option, -) -> impl futures::Stream { - let (tx, rx) = tokio::sync::mpsc::channel(32); - + is_flow: Option, + tx: tokio::sync::mpsc::Sender, +) -> () { tokio::spawn(async move { let mut log_offset = initial_log_offset; let mut stream_offset = initial_stream_offset; let mut last_update_hash: Option = None; + let mut flow_stream_job_id = None; // Send initial update immediately let mut running = running; let mut mem_peak = 0; - match get_job_update_data( &opt_authed, &opt_tokened, @@ -6540,6 +6743,8 @@ fn get_job_update_sse_stream( true, only_result, no_logs, + is_flow, + flow_stream_job_id, ) .await { @@ -6566,6 +6771,9 @@ fn get_job_update_sse_stream( update.stream_offset = None; } } + if update.flow_stream_job_id.is_some() { + flow_stream_job_id = update.flow_stream_job_id; + } if tx.send(JobUpdateSSEStream::Update(update)).await.is_err() { tracing::warn!("Failed to send initial job update for job {job_id}"); return; @@ -6576,7 +6784,7 @@ fn get_job_update_sse_stream( } Err(e) => { if tx - .send(JobUpdateSSEStream::Error(e.to_string())) + .send(JobUpdateSSEStream::Error { error: e.to_string() }) .await .is_err() { @@ -6630,6 +6838,8 @@ fn get_job_update_sse_stream( true, only_result, no_logs, + is_flow, + flow_stream_job_id, ) .await { @@ -6665,6 +6875,13 @@ fn get_job_update_sse_stream( update.stream_offset = None; } } + if update.flow_stream_job_id.is_some() { + if flow_stream_job_id.is_none() { + flow_stream_job_id = update.flow_stream_job_id; + } else { + update.flow_stream_job_id = None; + } + } if let Some(new_mem_peak) = update.mem_peak { if new_mem_peak != mem_peak { mem_peak = new_mem_peak; @@ -6693,8 +6910,29 @@ fn get_job_update_sse_stream( } } }); +} - tokio_stream::wrappers::ReceiverStream::new(rx) +async fn get_flow_stream_delta( + db: &DB, + flow_stream_job_id: Option, + stream_offset: Option, +) -> 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", + 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))) + } else { + Ok(None) + } + } else { + Ok(None) + } } async fn get_job_update_data( @@ -6711,6 +6949,8 @@ async fn get_job_update_data( get_full_job_on_completion: bool, only_result: Option, no_logs: Option, + is_flow: Option, + flow_stream_job_id: Option, ) -> error::Result { let tags = if log_view { log_job_view( @@ -6729,115 +6969,146 @@ async fn get_job_update_data( None }; + let ignore_flow_stream_job_id = is_flow.is_some_and(|x| !x) || flow_stream_job_id.is_some(); + if only_result.unwrap_or(false) { - let result = if let Some(tags) = tags { - 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 + 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 + 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, + 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 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 WHERE v2_job.id = $2 AND v2_job.workspace_id = $1", - w_id, - job_id, - stream_offset.unwrap_or(0), - ) - .fetch_optional(db) - .await? - .ok_or_else(|| Error::NotFound(format!("Job not found: {}", job_id)))?; - - if !tags.contains(&r.tag.as_str()) { - return Err(Error::NotAuthorized(format!( - "Job tag {} is not in the scope tags: {}", - r.tag, - tags.join(", ") - ))); - } - let running = r.running.as_ref().map(|x| *x); - ( - r.result.map(|x| x.0), - running, - r.result_stream.flatten(), - r.stream_offset, + w_id, + job_id, + stream_offset.unwrap_or(0), + ignore_flow_stream_job_id, ) - } else { - if running.is_some_and(|x| !x) { - let r = sqlx::query!( + .fetch_optional(db) + .await? + .ok_or_else(|| Error::NotFound(format!("Job not found: {}", job_id)))?; + + if !tags.contains(&r.tag.as_str()) { + return Err(Error::NotAuthorized(format!( + "Job tag {} is not in the scope tags: {}", + r.tag, + tags.join(", ") + ))); + } + let running = r.running.as_ref().map(|x| *x); + ( + r.result.map(|x| x.0), + running, + r.result_stream.flatten(), + r.stream_offset, + r.stream_job, + ) + } else { + if running.is_some_and(|x| !x) { + let r = sqlx::query!( "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 + CHAR_LENGTH(rs.stream) + 1 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 ) base 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 WHERE base.job_id = $1", job_id, w_id, stream_offset.unwrap_or(0), + ignore_flow_stream_job_id, ).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, - ) + 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.stream_job, + ) + } else { + (None, None, None, None, None) + } } else { - (None, None, None, None) - } - } else { - let q = sqlx::query!( + let q = sqlx::query!( "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 + CHAR_LENGTH(rs.stream) + 1 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 ( SELECT $2::uuid as job_id, $1::text as workspace_id ) 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 WHERE base.job_id = $2", w_id, job_id, stream_offset.unwrap_or(0), + ignore_flow_stream_job_id, ) .fetch_optional(db) .await?; - if let Some(r) = q { - ( - r.result.map(|x| x.0), - running, - r.result_stream.flatten(), - r.stream_offset, - ) - } else { - (None, None, None, None) + if let Some(r) = q { + ( + r.result.map(|x| x.0), + running, + r.result_stream.flatten(), + r.stream_offset, + r.stream_job, + ) + } else { + (None, None, None, None, None) + } } - } - }; + }; + + let flow_stream_job_id = flow_stream_job_id.or(new_flow_stream_job_id); + + let flow_stream_delta = + get_flow_stream_delta(db, flow_stream_job_id, stream_offset).await?; + + if let Some((flow_result_stream, flow_stream_offset)) = flow_stream_delta { + result_stream = flow_result_stream; + new_stream_offset = flow_stream_offset; + } + Ok(JobUpdate { - running: result.1, - completed: if result.0.is_some() { Some(true) } else { None }, + running, + completed: if result.is_some() { Some(true) } else { None }, log_offset: None, new_logs: None, - new_result_stream: result.2, - stream_offset: result.3, + new_result_stream: result_stream, + stream_offset: new_stream_offset, mem_peak: None, progress: None, job: None, flow_status: None, workflow_as_code_status: None, - only_result: result.0, + only_result: result, + flow_stream_job_id, }) } else { - let record = sqlx::query!( + let mut record = sqlx::query!( "SELECT c.id IS NOT NULL AS completed, CASE @@ -6848,6 +7119,7 @@ async fn get_job_update_data( SUBSTR(rs.stream, $8) 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, @@ -6891,6 +7163,16 @@ async fn get_job_update_data( None }; + let flow_stream_job_id = flow_stream_job_id.or(record.stream_job); + + let flow_stream_delta = + get_flow_stream_delta(db, flow_stream_job_id, stream_offset).await?; + + if let Some((new_result_stream, stream_offset)) = flow_stream_delta { + record.new_result_stream = new_result_stream; + record.stream_offset = stream_offset; + } + Ok(JobUpdate { running: record.running, completed: record.completed, @@ -6908,6 +7190,7 @@ async fn get_job_update_data( .flow_status .map(|x: sqlx::types::Json>| x.0), only_result: None, + flow_stream_job_id, }) } } diff --git a/backend/windmill-api/src/scopes.rs b/backend/windmill-api/src/scopes.rs index ef20131659..7984e4b627 100644 --- a/backend/windmill-api/src/scopes.rs +++ b/backend/windmill-api/src/scopes.rs @@ -484,21 +484,24 @@ pub fn check_route_access( ))) } -const SCRIPT_JOBS: [&'static str; 6] = [ +const SCRIPT_JOBS: [&'static str; 8] = [ "jobs/run/p", "jobs/run/h", "jobs/run_wait_result/p", "jobs/run_wait_result/h", "jobs/run/preview_bundle", "jobs/run/preview", + "jobs/run_and_stream/p", + "jobs/run_and_stream/h", ]; -const FLOW_JOBS: [&'static str; 5] = [ +const FLOW_JOBS: [&'static str; 6] = [ "jobs/run/f", "jobs/run_wait_result/f", "jobs/run/preview_flow", "jobs/restart/f", "jobs/flow/resume", + "jobs/run_and_stream/f", ]; lazy_static::lazy_static! { diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index 6ded25f923..949d563ac6 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -2131,7 +2131,6 @@ struct Runnable { workspace: String, endpoint_async: String, endpoint_sync: String, - endpoint_openai_sync: String, summary: String, description: String, schema: Option, @@ -2183,10 +2182,6 @@ async fn get_all_runnables( "/w/{}/jobs/run_wait_result/f/{}", &f.workspace, &f.path ), - endpoint_openai_sync: format!( - "/w/{}/jobs/openai_sync/f/{}", - &f.workspace, &f.path - ), summary: f.summary, description: f.description, schema: f.schema, @@ -2212,10 +2207,6 @@ async fn get_all_runnables( "/w/{}/jobs/run_wait_result/p/{}", &s.workspace, &s.path ), - endpoint_openai_sync: format!( - "/w/{}/jobs/openai_sync/p/{}", - &s.workspace, &s.path - ), summary: s.summary, description: s.description, schema: s.schema, diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index 1ae95843b7..8b16e6f427 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/backend/windmill-common/src/flow_status.rs @@ -43,6 +43,8 @@ pub struct FlowStatus { pub approval_conditions: Option, #[serde(skip_serializing_if = "Option::is_none")] pub restarted_from: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub stream_job: Option, } #[derive(Serialize, Deserialize, Debug, Clone, Default)] @@ -442,6 +444,7 @@ impl FlowStatus { retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, restarted_from: None, user_states: HashMap::new(), + stream_job: None, } } diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 962ffd6c11..4f93812f6a 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -146,10 +146,10 @@ impl FlowValue { .preprocessor_module .as_deref() .with_context(|| format!("no preprocessor module")), - Step::Step(i) => self + Step::Step { idx, .. } => self .modules - .get(i) - .with_context(|| format!("no module found at index: {i}")), + .get(idx) + .with_context(|| format!("no module found at index: {idx}")), Step::FailureStep => self .failure_module .as_deref() @@ -162,7 +162,7 @@ impl FlowValue { #[derive(Debug, Copy, Clone)] pub enum Step { - Step(usize), + Step { idx: usize, len: usize }, PreprocessorStep, FailureStep, } @@ -172,7 +172,7 @@ impl Step { if step < 0 { Step::PreprocessorStep } else if (step as usize) < len { - Step::Step(step as usize) + Step::Step { idx: step as usize, len } } else { Step::FailureStep } @@ -180,13 +180,13 @@ impl Step { pub fn get_step_index(&self) -> Option { match self { - Step::Step(index) => Some(*index), + Step::Step { idx, .. } => Some(*idx), _ => None, } } pub fn is_index_step(&self) -> bool { - matches!(self, Step::Step(_)) + matches!(self, Step::Step { .. }) } pub fn is_preprocessor_step(&self) -> bool { @@ -196,6 +196,10 @@ impl Step { pub fn is_failure_step(&self) -> bool { matches!(self, Step::FailureStep) } + + pub fn is_last_step(&self) -> bool { + matches!(self, Step::Step { idx, len } if *idx == len - 1) + } } #[derive(Default, Deserialize, Serialize, Debug, Clone)] diff --git a/backend/windmill-common/src/result_stream.rs b/backend/windmill-common/src/result_stream.rs index 4daa0cbd40..d44933891d 100644 --- a/backend/windmill-common/src/result_stream.rs +++ b/backend/windmill-common/src/result_stream.rs @@ -1,5 +1,5 @@ -use uuid::Uuid; use crate::{error, DB}; +use uuid::Uuid; pub const STREAM_PREFIX: &str = "WM_STREAM: "; @@ -14,9 +14,12 @@ pub fn extract_stream_from_logs(line: &str) -> Option { None } - - -pub async fn append_result_stream_db(db: &DB, workspace_id: &str, job_id: &Uuid, nstream: &str) -> error::Result<()> { +pub async fn append_result_stream_db( + db: &DB, + workspace_id: &str, + job_id: &Uuid, + nstream: &str, +) -> error::Result<()> { if !nstream.is_empty() { sqlx::query!( r#" @@ -27,7 +30,9 @@ pub async fn append_result_stream_db(db: &DB, workspace_id: &str, job_id: &Uuid, workspace_id, job_id, nstream, - ).execute(db).await?; + ) + .execute(db) + .await?; } Ok(()) } diff --git a/backend/windmill-queue/src/flow_status.rs b/backend/windmill-queue/src/flow_status.rs index 50af234149..e6b64c659d 100644 --- a/backend/windmill-queue/src/flow_status.rs +++ b/backend/windmill-queue/src/flow_status.rs @@ -1,6 +1,9 @@ use uuid::Uuid; use windmill_common::{ - error::{self, Error}, flows::Step, utils::WarnAfterExt, DB + error::{self, Error}, + flows::Step, + utils::WarnAfterExt, + DB, }; pub async fn update_flow_status_in_progress( @@ -11,7 +14,7 @@ pub async fn update_flow_status_in_progress( ) -> error::Result { let step = get_step_of_flow_status(db, flow).await?; match step { - Step::Step(step) => { + Step::Step { idx: step, .. } => { sqlx::query!( "UPDATE v2_job_status SET flow_status = jsonb_set( diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 487a7be632..7a7815dc6c 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -4098,6 +4098,7 @@ pub async fn push<'c, 'd>( }), user_states, preprocessor_module: None, + stream_job: None, } } _ => { @@ -4352,6 +4353,7 @@ pub async fn push<'c, 'd>( }), user_states, preprocessor_module: None, + stream_job: None, }; let value = flow_data.value(); let priority = value.priority; diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 9d5ac8aa66..2eddb657b5 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -326,7 +326,7 @@ async fn update_flow_status_module_with_actions( ) -> Result<(), Error> { let step = get_step_of_flow_status(db, parent_job.to_owned()).await?; match step { - Step::Step(step) => { + Step::Step { idx: step, .. } => { sqlx::query!( r#" UPDATE v2_job_status SET @@ -356,7 +356,7 @@ async fn update_flow_status_module_with_actions_success( ) -> Result<(), Error> { let step = get_step_of_flow_status(db, parent_job.to_owned()).await?; match step { - Step::Step(step) => { + Step::Step { idx: step, .. } => { // Append the new bool to the existing array, or create a new array if it doesn't exist sqlx::query!( r#" diff --git a/backend/windmill-worker/src/ansible_executor.rs b/backend/windmill-worker/src/ansible_executor.rs index 72640eb9da..04567b6217 100644 --- a/backend/windmill-worker/src/ansible_executor.rs +++ b/backend/windmill-worker/src/ansible_executor.rs @@ -93,6 +93,7 @@ async fn clone_repo( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -112,7 +113,8 @@ async fn clone_repo( .stdout(Stdio::piped()) .stderr(Stdio::piped()); - let checkout_cmd_child = start_child_process(checkout_cmd, GIT_PATH.as_str(), false).await?; + let checkout_cmd_child = + start_child_process(checkout_cmd, GIT_PATH.as_str(), false).await?; handle_child( job_id, conn, @@ -127,6 +129,7 @@ async fn clone_repo( false, &mut Some(occupancy_metrics), None, + None, ) .await?; } @@ -232,6 +235,7 @@ async fn clone_repo_without_history( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -249,7 +253,8 @@ async fn clone_repo_without_history( .stdout(Stdio::piped()) .stderr(Stdio::piped()); - let add_remote_cmd_child = start_child_process(add_remote_cmd, GIT_PATH.as_str(), false).await?; + let add_remote_cmd_child = + start_child_process(add_remote_cmd, GIT_PATH.as_str(), false).await?; handle_child( job_id, conn, @@ -264,6 +269,7 @@ async fn clone_repo_without_history( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -296,6 +302,7 @@ async fn clone_repo_without_history( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -328,6 +335,7 @@ async fn clone_repo_without_history( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -462,6 +470,7 @@ pub async fn install_galaxy_collections( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -484,7 +493,8 @@ pub async fn install_galaxy_collections( .stdout(Stdio::piped()) .stderr(Stdio::piped()); - let child = start_child_process(galaxy_collections_cmd, ANSIBLE_GALAXY_PATH.as_str(), false).await?; + let child = + start_child_process(galaxy_collections_cmd, ANSIBLE_GALAXY_PATH.as_str(), false).await?; handle_child( job_id, conn, @@ -499,6 +509,7 @@ pub async fn install_galaxy_collections( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -1133,6 +1144,7 @@ fi false, &mut Some(occupancy_metrics), None, + None, ) .await?; read_and_check_result(job_dir).await diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index b224957958..e55ee7eeaa 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -231,6 +231,7 @@ exit $exit_status true, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -647,6 +648,7 @@ pub async fn handle_powershell_job( false, &mut Some(occupancy_metrics), None, + None, ) .await?; } @@ -871,6 +873,7 @@ $env:PSModulePath = \"{};$PSModulePathBackup\"", false, &mut Some(occupancy_metrics), None, + None, ) .await?; diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index dce084bdf4..4016010b63 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -18,6 +18,7 @@ use crate::{ common::{ create_args_and_out_file, get_reserved_variables, parse_npm_config, read_file, read_file_content, read_result, start_child_process, write_file_binary, OccupancyMetrics, + StreamNotifier, }, handle_child::handle_child, BUNFIG_INSTALL_SCOPES, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR, BUN_NO_CACHE, BUN_PATH, @@ -168,6 +169,7 @@ pub async fn gen_bun_lockfile( false, occupancy_metrics, None, + None, ) .await?; } else { @@ -374,6 +376,7 @@ pub async fn install_bun_lockfile( false, occupancy_metrics, None, + None, ) .await?; } else { @@ -546,6 +549,7 @@ pub async fn generate_wrapper_mjs( false, occupancy_metrics, None, + None, ) .await?; fs::rename( @@ -597,6 +601,7 @@ pub async fn generate_bun_bundle( false, occupancy_metrics, None, + None, ) .await?; } else { @@ -1308,6 +1313,8 @@ try {{ append_logs(&job.id, &job.workspace_id, format!("{init_logs}\n"), conn).await; + let stream_notifier = StreamNotifier::new(conn, job); + let result = crate::js_eval::eval_fetch_timeout( env_code, inner_content.clone(), @@ -1323,6 +1330,7 @@ try {{ &job.workspace_id, false, occupancy_metrics, + stream_notifier, ) .await?; tracing::info!( @@ -1465,6 +1473,8 @@ try {{ .await? }; + let stream_notifier = StreamNotifier::new(conn, job); + let handle_result = handle_child( &job.id, conn, @@ -1479,6 +1489,7 @@ try {{ false, &mut Some(occupancy_metrics), None, + stream_notifier, ) .await?; diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index fbdac8909e..a90d14ca46 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -14,6 +14,7 @@ use sqlx::{Pool, Postgres}; use tokio::process::Command; use tokio::{fs::File, io::AsyncReadExt}; +use windmill_common::flows::Step; #[cfg(feature = "parquet")] use windmill_common::s3_helpers::{ get_etag_or_empty, LargeFileStorage, ObjectStoreResource, S3Object, @@ -32,8 +33,10 @@ use windmill_common::{ use anyhow::{anyhow, Result}; use windmill_parser_sql::{s3_mode_extension, S3ModeArgs, S3ModeFormat}; +use windmill_queue::flow_status::get_step_of_flow_status; use windmill_queue::MiniPulledJob; +use std::collections::HashSet; use std::path::Path; use std::{collections::HashMap, sync::Arc, time::Duration}; @@ -1065,6 +1068,149 @@ pub fn get_root_job_id(job: &MiniPulledJob) -> uuid::Uuid { .unwrap_or(job.id) } +#[derive(Clone)] +pub struct StreamNotifier { + db: DB, + job_id: uuid::Uuid, + parent_job: uuid::Uuid, + root_job: uuid::Uuid, +} + +#[async_recursion] +async fn check_if_nested_step_is_last( + db: &DB, + parent_job: Uuid, + parent_of_parent_job: Option, + root_job: Uuid, + visited: Option>, +) -> error::Result { + // Initialize or use the provided visited set for cycle detection + let mut visited = visited.unwrap_or_else(HashSet::new); + + // Check for cycles - if we've already visited this job, return false to break the recursion + if !visited.insert(parent_job) { + return Ok(false); + } + + // get parent of parent job to get step of parent job + let parent_of_parent_job = parent_of_parent_job.or(sqlx::query_scalar!( + "SELECT parent_job FROM v2_job WHERE id = $1", + parent_job + ) + .fetch_one(db) + .await?); + if let Some(parent_of_parent_job) = parent_of_parent_job { + // Check for cycles again with the parent_of_parent_job + if !visited.insert(parent_of_parent_job) { + return Ok(false); + } + + let r = sqlx::query!( + r#"SELECT + (flow_status->'step')::integer as step, + jsonb_array_length(flow_status->'modules') as len, + flow_status->'modules'->-1->>'branch_chosen' IS NOT NULL as is_branch_one, + parent_job as ppp_job + FROM v2_job + LEFT JOIN v2_job_status USING (id) + WHERE v2_job.id = $1"#, + parent_of_parent_job + ) + .fetch_one(db) + .await + .map_err(|e| Error::internal_err(format!("fetching step flow status: {e:#}")))?; + + if let Some(step) = r.step { + let step = Step::from_i32_and_len(step, r.len.unwrap_or(0) as usize); + + // if parent job is last and a branch one and + // - root_job is equal to parent of parent job, return true + // - root job is not equal to parent of parent job, recursively check if the parent of parent job is a branch one and last + if step.is_last_step() && r.is_branch_one.unwrap_or(false) { + if parent_of_parent_job == root_job { + return Ok(true); + } else { + return check_if_nested_step_is_last( + db, + parent_of_parent_job, + r.ppp_job, + root_job, + Some(visited), + ) + .await; + } + } + } + } + + Ok(false) +} + +impl StreamNotifier { + pub fn new(conn: &Connection, job: &MiniPulledJob) -> Option { + let root_job = get_root_job_id(job); + if job.is_flow_step() && job.parent_job.is_some() { + match conn { + Connection::Sql(db) => Some(Self { + db: db.clone(), + parent_job: job.parent_job.unwrap(), + job_id: job.id, + root_job, + }), + Connection::Http(_) => { + tracing::warn!( + "Flow job streaming is only supported for workers connected to a database" + ); + None + } + } + } else { + None + } + } + + async fn update_flow_status_with_stream_job_inner( + db: DB, + parent_job: Uuid, + job_id: Uuid, + root_job: Uuid, + ) -> Result<(), Error> { + let step = get_step_of_flow_status(&db, parent_job).await?; + + if step.is_last_step() + && (parent_job == root_job + || check_if_nested_step_is_last(&db, parent_job, None, root_job, None).await?) + { + sqlx::query!(r#" + UPDATE v2_job_status + SET flow_status = jsonb_set(flow_status, array['stream_job'], to_jsonb($1::UUID::TEXT)) + WHERE id = $2"#, + job_id, + root_job + ) + .execute(&db) + .await?; + } + + Ok(()) + } + + pub fn update_flow_status_with_stream_job(&self) -> () { + let db = self.db.clone(); + let parent_job = self.parent_job; + let job_id = self.job_id; + let root_job = self.root_job; + tokio::spawn(async move { + if let Err(err) = + Self::update_flow_status_with_stream_job_inner(db, parent_job, job_id, root_job) + .await + { + tracing::error!("Could not notify about stream job {}: {err:#?}", parent_job); + } + }); + } +} + #[derive(Clone)] pub struct S3ModeWorkerData { pub client: AuthedClient, diff --git a/backend/windmill-worker/src/csharp_executor.rs b/backend/windmill-worker/src/csharp_executor.rs index 84cc3184a3..51bdf2f8e8 100644 --- a/backend/windmill-worker/src/csharp_executor.rs +++ b/backend/windmill-worker/src/csharp_executor.rs @@ -129,6 +129,7 @@ pub async fn generate_nuget_lockfile( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -386,6 +387,7 @@ async fn build_cs_proj( false, &mut Some(occupancy_metrics), None, + None, ) .await?; append_logs(job_id, w_id, "\n\n", conn).await; @@ -643,6 +645,7 @@ pub async fn handle_csharp_job( false, &mut Some(occupancy_metrics), None, + None, ) .await?; read_result(job_dir, None).await diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index 431a457d74..0e15e5ba12 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -8,7 +8,7 @@ use windmill_queue::{append_logs, CanceledBy, MiniPulledJob}; use crate::{ common::{ create_args_and_out_file, get_reserved_variables, parse_npm_config, read_file, read_result, - start_child_process, OccupancyMetrics, + start_child_process, OccupancyMetrics, StreamNotifier, }, handle_child::handle_child, DENO_CACHE_DIR, DENO_PATH, DISABLE_NSJAIL, HOME_ENV, NPM_CONFIG_REGISTRY, PATH_ENV, TZ_ENV, @@ -161,6 +161,7 @@ pub async fn generate_deno_lock( false, occupancy_metrics, None, + None, ) .await?; } else { @@ -417,6 +418,9 @@ try {{ .stderr(Stdio::piped()); start_child_process(deno_cmd, DENO_PATH.as_str(), false).await? }; + + let stream_notifier = StreamNotifier::new(conn, job); + // logs.push_str(format!("prepare: {:?}\n", start.elapsed().as_micros()).as_str()); // start = Instant::now(); let handle_result = handle_child( @@ -433,6 +437,7 @@ try {{ false, &mut Some(occupancy_metrics), None, + stream_notifier, ) .await?; // logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str()); diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index 85adbf7fea..7a758051ab 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -274,6 +274,7 @@ func Run(req Req) (interface{{}}, error){{ false, &mut Some(occupation_metrics), None, + None, ) .await?; @@ -422,6 +423,7 @@ func Run(req Req) (interface{{}}, error){{ false, &mut Some(occupation_metrics), None, + None, ) .await?; @@ -507,6 +509,7 @@ pub async fn install_go_dependencies( false, &mut Some(occupation_metrics), None, + None, ) .await?; @@ -622,6 +625,7 @@ pub async fn install_go_dependencies( false, &mut Some(occupation_metrics), None, + None, ) .await?; diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 11078c7892..6d36b8ca0d 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::AtomicU32; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::Arc; use std::{io, panic, time::Duration}; @@ -52,7 +52,7 @@ use futures::{ stream, StreamExt, }; -use crate::common::{resolve_job_timeout, OccupancyMetrics}; +use crate::common::{resolve_job_timeout, OccupancyMetrics, StreamNotifier}; use crate::job_logger::{append_job_logs, append_result_stream, append_with_limit}; use crate::job_logger_oss::process_streaming_log_lines; use crate::worker_utils::{ping_job_status, update_worker_ping_from_job}; @@ -114,6 +114,7 @@ pub async fn handle_child( occupancy_metrics: &mut Option<&mut OccupancyMetrics>, // Do not print logs to output, but instead save to string. pipe_stdout: Option<&mut String>, + stream_notifier: Option, ) -> error::Result { let start = Instant::now(); @@ -315,6 +316,7 @@ pub async fn handle_child( &mut rx2, child_name, &mut stream_result, + stream_notifier, ) .instrument(trace_span!("child_lines")); @@ -354,6 +356,7 @@ pub async fn write_lines( rx2: &mut broadcast::Receiver<()>, child_name: &str, stream_result: &mut Vec, + stream_notifier: Option, ) { let max_log_size = if *CLOUD_HOSTED { MAX_RESULT_SIZE @@ -384,6 +387,7 @@ pub async fn write_lines( let mut pipe_stdout = pipe_stdout; + let is_stream = Arc::new(AtomicBool::new(false)); while let Some(line) = output.by_ref().next().await { let do_write_ = do_write.shared(); @@ -410,6 +414,7 @@ pub async fn write_lines( let job_id = job_id.clone(); let mut nstream = String::new(); + while let Some(line) = read_lines.next().await { match line { Ok(line) => { @@ -479,8 +484,17 @@ pub async fn write_lines( let w_id = w_id.to_string(); let job_id = job_id.clone(); let pg_log_total_size = pg_log_total_size.clone(); + let stream_notifier = stream_notifier.clone(); + let is_stream = is_stream.clone(); (do_write, write_result) = tokio::spawn(async move { if !nstream.is_empty() { + if let Some(stream_notifier) = stream_notifier { + if !is_stream.load(Ordering::SeqCst) { + 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 { tracing::error!( "Unable to send result stream for job {job_id}. Error was: {:?}", diff --git a/backend/windmill-worker/src/java_executor.rs b/backend/windmill-worker/src/java_executor.rs index 6aa12d5024..d133308489 100644 --- a/backend/windmill-worker/src/java_executor.rs +++ b/backend/windmill-worker/src/java_executor.rs @@ -534,6 +534,7 @@ async fn compile<'a>( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -718,6 +719,7 @@ async fn run<'a>( false, &mut Some(occupancy_metrics), None, + None, ) .await?; Ok(()) diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index 7f6fa8146c..cbf19b5946 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -48,7 +48,7 @@ use windmill_common::worker::{write_file, TMP_DIR}; use windmill_common::flow_status::JobResult; use windmill_queue::CanceledBy; -use crate::common::OccupancyMetrics; +use crate::common::{OccupancyMetrics, StreamNotifier}; use windmill_common::client::AuthedClient; #[cfg(feature = "deno_core")] @@ -769,6 +769,7 @@ pub async fn eval_fetch_timeout( _w_id: &str, _load_client: bool, _occupation_metrics: &mut OccupancyMetrics, + _stream_notifier: Option, ) -> anyhow::Result> { use serde_json::value::to_raw_value; Ok(to_raw_value("require deno_core").unwrap()) @@ -790,6 +791,7 @@ pub async fn eval_fetch_timeout( w_id: &str, load_client: bool, occupation_metrics: &mut OccupancyMetrics, + stream_notifier: Option, ) -> windmill_common::error::Result> { let (sender, mut receiver) = oneshot::channel::(); let (append_logs_sender, mut append_logs_receiver) = mpsc::unbounded_channel::(); @@ -940,10 +942,18 @@ pub async fn eval_fetch_timeout( } let handle = tokio::spawn(async move { let mut result_stream = String::new(); + let mut is_stream = false; while let Some(log) = log_receiver.recv().await { use windmill_common::result_stream::extract_stream_from_logs; if let Some(stream) = extract_stream_from_logs(&log.trim_end_matches("\n")) { + if let Some(sn) = stream_notifier.as_ref() { + if !is_stream { + is_stream = true; + sn.update_flow_status_with_stream_job(); + } + } + result_stream.push_str(&stream); if let Err(e) = result_stream_sender.send(stream) { tracing::error!("failed to send result stream: {e}"); diff --git a/backend/windmill-worker/src/nu_executor.rs b/backend/windmill-worker/src/nu_executor.rs index a52270e745..2497757d9d 100644 --- a/backend/windmill-worker/src/nu_executor.rs +++ b/backend/windmill-worker/src/nu_executor.rs @@ -340,6 +340,7 @@ async fn run<'a>( false, &mut Some(occupancy_metrics), None, + None, ) .await?; Ok(()) diff --git a/backend/windmill-worker/src/php_executor.rs b/backend/windmill-worker/src/php_executor.rs index cdfd4cfe6f..b272d631da 100644 --- a/backend/windmill-worker/src/php_executor.rs +++ b/backend/windmill-worker/src/php_executor.rs @@ -106,6 +106,7 @@ pub async fn composer_install( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -343,6 +344,7 @@ try {{ false, &mut Some(occupancy_metrics), None, + None, ) .await?; read_result(job_dir, None).await diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index c41089652b..57936969b8 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -121,7 +121,7 @@ use windmill_common::s3_helpers::OBJECT_STORE_SETTINGS; use crate::{ common::{ create_args_and_out_file, get_reserved_variables, read_file, read_result, - start_child_process, OccupancyMetrics, + start_child_process, OccupancyMetrics, StreamNotifier, }, handle_child::handle_child, worker_utils::ping_job_status, @@ -386,6 +386,7 @@ pub async fn uv_pip_compile( false, occupancy_metrics, None, + None, ) .await .map_err(|e| { @@ -864,6 +865,8 @@ mount {{ start_child_process(python_cmd, &python_path, false).await? }; + let stream_notifier = StreamNotifier::new(conn, job); + let handle_result = handle_child( &job.id, conn, @@ -878,6 +881,7 @@ mount {{ false, &mut Some(occupancy_metrics), None, + stream_notifier, ) .await?; diff --git a/backend/windmill-worker/src/python_versions.rs b/backend/windmill-worker/src/python_versions.rs index 3d71e9f9ec..2c5b828313 100644 --- a/backend/windmill-worker/src/python_versions.rs +++ b/backend/windmill-worker/src/python_versions.rs @@ -655,6 +655,7 @@ impl PyV { false, occupancy_metrics, None, + None, ) .await?; Ok(()) diff --git a/backend/windmill-worker/src/ruby_executor.rs b/backend/windmill-worker/src/ruby_executor.rs index fb087a9d24..b635a6bade 100644 --- a/backend/windmill-worker/src/ruby_executor.rs +++ b/backend/windmill-worker/src/ruby_executor.rs @@ -420,6 +420,7 @@ Your Gemfile syntax will continue to work as-is." &mut None, // Some(&mut stdout), None, + None, ) .await?; @@ -871,6 +872,7 @@ mount {{ false, &mut Some(occupancy_metrics), None, + None, ) .await?; Ok(()) diff --git a/backend/windmill-worker/src/rust_executor.rs b/backend/windmill-worker/src/rust_executor.rs index f47399c54a..6a05de459c 100644 --- a/backend/windmill-worker/src/rust_executor.rs +++ b/backend/windmill-worker/src/rust_executor.rs @@ -175,7 +175,8 @@ pub async fn generate_cargo_lockfile( std::env::var("TMP").unwrap_or_else(|_| "C:\\tmp".to_string()), ); } - let gen_lockfile_process = start_child_process(gen_lockfile_cmd, CARGO_PATH.as_str(), false).await?; + let gen_lockfile_process = + start_child_process(gen_lockfile_cmd, CARGO_PATH.as_str(), false).await?; handle_child( job_id, conn, @@ -190,6 +191,7 @@ pub async fn generate_cargo_lockfile( false, &mut Some(occupancy_metrics), None, + None, ) .await?; @@ -298,6 +300,7 @@ async fn get_build_dir( false, &mut None, None, + None, ) .await } @@ -410,6 +413,7 @@ pub async fn build_rust_crate( false, &mut Some(occupancy_metrics), None, + None, ) .await?; append_logs(&job.id, &job.workspace_id, "\n\n", conn).await; @@ -599,6 +603,7 @@ pub async fn handle_rust_job( false, &mut Some(occupancy_metrics), None, + None, ) .await?; read_result(job_dir, None).await diff --git a/backend/windmill-worker/src/universal_pkg_installer.rs b/backend/windmill-worker/src/universal_pkg_installer.rs index dc4b5dab2a..3f3198339a 100644 --- a/backend/windmill-worker/src/universal_pkg_installer.rs +++ b/backend/windmill-worker/src/universal_pkg_installer.rs @@ -159,6 +159,7 @@ pub async fn par_install_language_dependencies_all_at_once< false, &mut None, pipe_stdout, + None, ) .await { @@ -587,6 +588,7 @@ async fn try_install_one_detached<'a, T: Clone + std::marker::Send + Sync + 'a + false, &mut None, None, + None, ) .await { diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 6532c1bb22..2ae1a42f74 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -101,6 +101,7 @@ use tokio::{ use rand::Rng; use crate::ai_executor::handle_ai_agent_job; +use crate::common::StreamNotifier; use crate::{ agent_workers::{queue_init_job, queue_periodic_job}, bash_executor::{handle_bash_job, handle_powershell_job}, @@ -2229,6 +2230,8 @@ async fn do_nativets( job.args.as_ref() }; + let stream_notifier = StreamNotifier::new(conn, job); + Ok(eval_fetch_timeout( env_code, code.clone(), @@ -2244,6 +2247,7 @@ async fn do_nativets( &job.workspace_id, true, occupancy_metrics, + stream_notifier, ) .await?) } diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 0ccebcf477..201924e825 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -328,7 +328,7 @@ pub async fn update_flow_status_after_job_completion_internal( let module_step = Step::from_i32_and_len(old_status.step, old_status.modules.len()); let current_module = match module_step { - Step::Step(i) => flow_value.modules.get(i), + Step::Step { idx: i, .. } => flow_value.modules.get(i), _ => None, }; @@ -342,7 +342,7 @@ pub async fn update_flow_status_after_job_completion_internal( .as_ref() .ok_or_else(|| Error::internal_err(format!("preprocessor module not found")))?, Step::FailureStep => &old_status.failure_module.module_status, - Step::Step(i) => old_status + Step::Step { idx: i, .. } => old_status .modules .get(i as usize) .ok_or_else(|| Error::internal_err(format!("module {i} not found")))?, @@ -1984,7 +1984,7 @@ async fn push_next_flow_job( tracing::info!(id = %flow_job.id, root_id = %job_root, step = ?step, "pushing next flow job"); let mut status_module = match step { - Step::Step(i) => status + Step::Step { idx: i, .. } => status .modules .get(i) .cloned() @@ -2035,7 +2035,7 @@ async fn push_next_flow_job( }))); } - if matches!(step, Step::Step(0)) { + if matches!(step, Step::Step { idx: 0, .. }) { if !flow_job.is_flow_step() && flow_job.schedule_path().is_some() { let schedule_path = flow_job.schedule_path(); let no_flow_overlap = sqlx::query_scalar!( @@ -2125,7 +2125,7 @@ async fn push_next_flow_job( let arc_last_job_result = if status_module.is_failure() { // if job is being retried, pass the result of its previous failure last_job_result.unwrap_or_else(|| Arc::new(to_raw_value(&json!("{}")))) - } else if matches!(step, Step::Step(0)) || step.is_preprocessor_step() { + } else if matches!(step, Step::Step { idx: 0, .. }) || step.is_preprocessor_step() { // if it's the first job executed in the flow, pass the flow args Arc::new(to_raw_value(&flow_job.args)) } else { @@ -2444,7 +2444,9 @@ async fn push_next_flow_job( let current_id = &module.id; let mut previous_id = match step { - Step::Step(i) if i >= 1 => flow.modules.get(i - 1).map(|m| m.id.clone()).unwrap(), + Step::Step { idx: i, .. } if i >= 1 => { + flow.modules.get(i - 1).map(|m| m.id.clone()).unwrap() + } _ => String::new(), }; @@ -2458,7 +2460,7 @@ async fn push_next_flow_job( ) { None } else { - let sleep_input_transform = if let Step::Step(i) = step { + let sleep_input_transform = if let Step::Step { idx: i, .. } = step { i.checked_sub(1) .and_then(|i| flow.modules.get(i)) .and_then(|m| m.sleep.clone()) @@ -3272,7 +3274,7 @@ async fn push_next_flow_job( .warn_after_seconds(3) .await?; } - Step::Step(i) => { + Step::Step { idx: i, .. } => { sqlx::query!( "UPDATE v2_job_status SET flow_status = JSONB_SET( diff --git a/frontend/src/lib/components/JobLoader.svelte b/frontend/src/lib/components/JobLoader.svelte index 89281c7619..14e675328e 100644 --- a/frontend/src/lib/components/JobLoader.svelte +++ b/frontend/src/lib/components/JobLoader.svelte @@ -669,6 +669,12 @@ if (resultStreamOffset) { params.set('stream_offset', resultStreamOffset.toString()) } + if (job) { + params.set( + 'is_flow', + (job.job_kind === 'flow' || job.job_kind === 'flowpreview').toString() + ) + } const sseUrl = `/api/w/${workspace}/jobs_u/getupdate_sse/${id}?${params.toString()}` diff --git a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte index 311108dc92..4271728c1d 100644 --- a/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte +++ b/frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte @@ -10,7 +10,7 @@ } from '$lib/consts' import bash from 'svelte-highlight/languages/bash' import { Tabs, Tab, TabContent, Button } from '$lib/components/common' - import { ArrowDownRight, ArrowUpRight, Clipboard } from 'lucide-svelte' + import { ArrowDownRight, ArrowUpRight, Clipboard, RssIcon } from 'lucide-svelte' import { Highlight } from 'svelte-highlight' import { typescript } from 'svelte-highlight/languages' import ClipboardPanel from '../../details/ClipboardPanel.svelte' @@ -44,24 +44,43 @@ let webhooks: { async: { - hash?: string - path: string + get: {} + post: { + hash?: string + path: string + } + sse: {} } sync: { - hash?: string - path: string - get_path?: string + get: { + path: string + } + post: { + hash?: string + path: string + } + sse: { + hash?: string + path: string + } } } = $derived(isFlow ? computeFlowWebhooks(path) : computeScriptWebhooks(hash, path)) let selectedTab: string = $state('rest') let userSettings: UserSettings | undefined = $state() - let webhookType = $state(DEFAULT_WEBHOOK_TYPE) as 'async' | 'sync' - let requestType = $state(isFlow ? 'path' : 'path') as 'hash' | 'path' | 'get_path' + let requestType = $state(DEFAULT_WEBHOOK_TYPE) as 'async' | 'sync' + let callMethod = $state('post') as 'get' | 'post' | 'sse' + let runnableId = $state('path') as 'hash' | 'path' let tokenType = $state('headers') as 'query' | 'headers' $effect(() => { - if (webhookType === 'async' && requestType === 'get_path') { - requestType = hash ? 'hash' : 'path' + if (requestType === 'async' && (callMethod === 'get' || callMethod === 'sse')) { + callMethod = 'post' + } + }) + + $effect(() => { + if (callMethod === 'sse' && tokenType === 'headers') { + tokenType = 'query' } }) @@ -72,15 +91,15 @@ : runnableArgs }) let url: string = $derived( - webhooks[webhookType][requestType] + + webhooks[requestType][callMethod][runnableId] + (tokenType === 'query' ? `?token=${token}${ - requestType === 'get_path' + callMethod === 'get' || callMethod === 'sse' ? `&payload=${encodeURIComponent(btoa(JSON.stringify(cleanedRunnableArgs ?? {})))}` : '' }` : `${ - requestType === 'get_path' + callMethod === 'get' ? `?payload=${encodeURIComponent(btoa(JSON.stringify(cleanedRunnableArgs ?? {})))}` : '' }`) @@ -90,13 +109,25 @@ let webhookBase = `${location.origin}${base}/api/w/${$workspaceStore}/jobs` return { async: { - hash: `${webhookBase}/run/h/${hash}`, - path: `${webhookBase}/run/p/${path}` + get: {}, + post: { + hash: `${webhookBase}/run/h/${hash}`, + path: `${webhookBase}/run/p/${path}` + }, + sse: {} }, sync: { - hash: `${webhookBase}/run_wait_result/h/${hash}`, - path: `${webhookBase}/run_wait_result/p/${path}`, - get_path: `${webhookBase}/run_wait_result/p/${path}` + get: { + path: `${webhookBase}/run_wait_result/p/${path}` + }, + post: { + hash: `${webhookBase}/run_wait_result/h/${hash}`, + path: `${webhookBase}/run_wait_result/p/${path}` + }, + sse: { + hash: `${webhookBase}/run_and_stream/h/${hash}`, + path: `${webhookBase}/run_and_stream/p/${path}` + } } } } @@ -106,20 +137,32 @@ let urlAsync = `${webhooksBase}/run/f/${path}` let urlSync = `${webhooksBase}/run_wait_result/f/${path}` + let urlStream = `${webhooksBase}/run_and_stream/f/${path}` return { async: { - path: urlAsync + get: {}, + post: { + path: urlAsync + }, + sse: {} }, sync: { - path: urlSync, - get_path: urlSync + get: { + path: urlSync + }, + post: { + path: urlSync + }, + sse: { + path: urlStream + } } } } function headers() { const headers = {} - if (requestType != 'get_path') { + if (callMethod === 'post') { headers['Content-Type'] = 'application/json' } @@ -130,7 +173,34 @@ } function fetchCode() { - if (webhookType === 'sync') { + if (callMethod === 'sse') { + return ` +import { EventSource } from "eventsource"; + +export async function main() { + const endpoint = \`${url}\`; + + return new Promise((resolve, reject) => { + const eventSource = new EventSource(endpoint); + + eventSource.onmessage = (event) => { + const data = JSON.parse(event.data); + console.log(data); + if (data.completed) { + eventSource.close(); + resolve(); + } + }; + + eventSource.onerror = (error) => { + console.error('EventSource error:', error); + eventSource.close(); + reject(error); + }; + }); +}` + } + if (requestType === 'sync') { return ` export async function main() { const jobTriggerResponse = await triggerJob(); @@ -140,7 +210,7 @@ export async function main() { async function triggerJob() { ${ - requestType === 'get_path' + callMethod === 'get' ? '// Payload is a base64 encoded string of the arguments' : `const body = JSON.stringify(${JSON.stringify( cleanedRunnableArgs ?? {}, @@ -151,16 +221,15 @@ async function triggerJob() { const endpoint = \`${url}\`; return await fetch(endpoint, { - method: '${requestType === 'get_path' ? 'GET' : 'POST'}', + method: '${callMethod === 'get' ? 'GET' : 'POST'}', headers: ${JSON.stringify(headers(), null, 2).replaceAll('\n', '\n\t\t')}${ - requestType === 'get_path' ? '' : `,\n\t\tbody` + callMethod === 'get' ? '' : `,\n\t\tbody` } }); }` - } - - // Main function - let mainFunction = ` + } else { + // Main function + let mainFunction = ` export async function main() { const jobTriggerResponse = await triggerJob(); const UUID = await jobTriggerResponse.text(); @@ -168,8 +237,8 @@ export async function main() { return jobCompletionData; }` - // triggerJob function - let triggerJobFunction = ` + // triggerJob function + let triggerJobFunction = ` async function triggerJob() { const body = JSON.stringify(${JSON.stringify(cleanedRunnableArgs ?? {}, null, 2).replaceAll( '\n', @@ -178,14 +247,14 @@ async function triggerJob() { const endpoint = \`${url}\`; return await fetch(endpoint, { - method: '${requestType === 'get_path' ? 'GET' : 'POST'}', + method: '${callMethod === 'get' ? 'GET' : 'POST'}', headers: ${JSON.stringify(headers(), null, 2).replaceAll('\n', '\n\t\t')}, body }); }` - // waitForJobCompletion function - let waitForJobCompletionFunction = ` + // waitForJobCompletion function + let waitForJobCompletionFunction = ` function waitForJobCompletion(UUID) { return new Promise(async (resolve, reject) => { try { @@ -214,22 +283,23 @@ function waitForJobCompletion(UUID) { }); }` - // Combine and return - return `${mainFunction}\n\n${triggerJobFunction}\n\n${waitForJobCompletionFunction}` + // Combine and return + return `${mainFunction}\n\n${triggerJobFunction}\n\n${waitForJobCompletionFunction}` + } } function curlCode() { return `TOKEN='${token}' -${requestType !== 'get_path' ? `BODY='${JSON.stringify(cleanedRunnableArgs ?? {})}'` : ''} +${callMethod !== 'get' ? `BODY='${JSON.stringify(cleanedRunnableArgs ?? {})}'` : ''} URL='${url}' -${webhookType === 'sync' ? 'RESULT' : 'UUID'}=$(curl -s ${ - requestType != 'get_path' ? "-H 'Content-Type: application/json'" : '' +${requestType === 'sync' ? 'RESULT' : 'UUID'}=$(curl -s ${ + callMethod != 'get' ? "-H 'Content-Type: application/json'" : '' } ${tokenType === 'headers' ? `-H "Authorization: Bearer $TOKEN"` : ''} -X ${ - requestType === 'get_path' ? 'GET' : 'POST' - } ${requestType !== 'get_path' ? `-d "$BODY" ` : ''}$URL) + callMethod === 'get' ? 'GET' : 'POST' + } ${callMethod !== 'get' ? `-d "$BODY" ` : ''}$URL) ${ - webhookType === 'sync' + requestType === 'sync' ? 'echo -E $RESULT | jq' : ` URL="${location.origin}/api/w/${$workspaceStore}/jobs_u/completed/get_result_maybe/$UUID" @@ -286,7 +356,7 @@ done`
Request type
- + {#snippet children({ item })}
Call method
- + {#snippet children({ item })} - {#if !isFlow} - - {/if} - - + + {/snippet}
+ {#if !isFlow} +
+
+ Reference type +
+ + {#snippet children({ item })} + + + {/snippet} + +
+ {/if}
Token configuration
{#snippet children({ item })} - + {/snippet} @@ -354,10 +443,12 @@ done`
REST - {#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL} + {#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL && callMethod !== 'sse'} Curl {/if} - Fetch + + {callMethod === 'sse' ? 'Event Source' : 'Fetch'} + {#snippet content()} {#key token} @@ -367,12 +458,12 @@ done` - {#if requestType !== 'get_path'} + {#if callMethod !== 'get'} {/if} - {#key requestType} + {#key callMethod} {#key tokenType}