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

feat(backend): flow streaming (#6520)

Browse Source 
* feat(backend): flow streaming

* all streaming languages + sync api

* sqlx

* fix build

* UI and nits

* nit

* feat: stream last flow step

* sqlx

* nit

* use get for stream endpoints + add snippet in UI

* refactor

* nits

* Update backend/windmill-worker/src/common.rs

Co-authored-by: graphite-app[bot] <96075541+graphite-app[bot]@users.noreply.github.com>

* nits

---------

Co-authored-by: graphite-app[bot] <96075541+graphite-app[bot]@users.noreply.github.com>
This commit is contained in:
hugocasa
2025-09-18 17:49:35 +02:00
committed by GitHub
parent 943ed0e1ca
commit a2b34f3e08
41 changed files with 1031 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0997b46bae6e2374b568e8367898d6fc79c331431326250c1a59674054ceaabd.json
Generated
+22
View File
@@ -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"
}
backend/.sqlx/query-2e589e039986e7a2c75e328868874669b32cbe0dae6822b2d2fad0635c5f6087.json
Generated
+15
View File
@@ -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"
}
backend/.sqlx/query-70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393.json → backend/.sqlx/query-40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773.json
Generated
+13 -7
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n c.id IS NOT NULL AS completed,\n CASE\n WHEN q.id IS NOT NULL THEN (CASE WHEN NOT $5 AND q.running THEN true ELSE null END)\n ELSE false\n END AS running,\n CASE WHEN $7::BOOLEAN THEN NULL ELSE SUBSTR(logs, GREATEST($1 - log_offset, 0)) END AS logs,\n SUBSTR(rs.stream, $8) AS new_result_stream,\n COALESCE(r.memory_peak, c.memory_peak) AS mem_peak,\n COALESCE(c.flow_status, f.flow_status) AS \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n COALESCE(c.workflow_as_code_status, f.workflow_as_code_status) AS \"workflow_as_code_status: sqlx::types::Json<Box<RawValue>>\",\n CASE WHEN $7::BOOLEAN THEN NULL ELSE job_logs.log_offset + CHAR_LENGTH(job_logs.logs) + 1 END AS log_offset,\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset,\n created_by AS \"created_by!\",\n CASE WHEN $4::BOOLEAN THEN (\n SELECT scalar_int FROM job_stats WHERE job_id = $3 AND metric_id = 'progress_perc'\n ) END AS progress,\n rs.stream AS \"result_stream: Option<String>\"\n FROM v2_job j\n LEFT JOIN v2_job_queue q USING (id)\n LEFT JOIN v2_job_runtime r USING (id)\n LEFT JOIN v2_job_status f USING (id)\n LEFT JOIN v2_job_completed c USING (id)\n LEFT JOIN job_result_stream rs ON rs.job_id = $3\n LEFT JOIN job_logs ON job_logs.job_id = $3\n WHERE j.workspace_id = $2 AND j.id = $3\n AND ($6::text[] IS NULL OR j.tag = ANY($6))",
"query": "SELECT\n c.id IS NOT NULL AS completed,\n CASE\n WHEN q.id IS NOT NULL THEN (CASE WHEN NOT $5 AND q.running THEN true ELSE null END)\n ELSE false\n END AS running,\n CASE WHEN $7::BOOLEAN THEN NULL ELSE SUBSTR(logs, GREATEST($1 - log_offset, 0)) END AS logs,\n SUBSTR(rs.stream, $8) AS new_result_stream,\n COALESCE(r.memory_peak, c.memory_peak) AS mem_peak,\n COALESCE(c.flow_status, f.flow_status) AS \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n (COALESCE(c.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<Box<RawValue>>\",\n CASE WHEN $7::BOOLEAN THEN NULL ELSE job_logs.log_offset + CHAR_LENGTH(job_logs.logs) + 1 END AS log_offset,\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset,\n created_by AS \"created_by!\",\n CASE WHEN $4::BOOLEAN THEN (\n SELECT scalar_int FROM job_stats WHERE job_id = $3 AND metric_id = 'progress_perc'\n ) END AS progress,\n rs.stream AS \"result_stream: Option<String>\"\n FROM v2_job j\n LEFT JOIN v2_job_queue q USING (id)\n LEFT JOIN v2_job_runtime r USING (id)\n LEFT JOIN v2_job_status f USING (id)\n LEFT JOIN v2_job_completed c USING (id)\n LEFT JOIN job_result_stream rs ON rs.job_id = $3\n LEFT JOIN job_logs ON job_logs.job_id = $3\n WHERE j.workspace_id = $2 AND j.id = $3\n AND ($6::text[] IS NULL OR j.tag = ANY($6))",
"describe": {
"columns": [
{
@@ -35,31 +35,36 @@
},
{
"ordinal": 6,
"name": "stream_job",
"type_info": "Uuid"
},
{
"ordinal": 7,
"name": "workflow_as_code_status: sqlx::types::Json<Box<RawValue>>",
"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<String>",
"type_info": "Text"
}
@@ -86,10 +91,11 @@
null,
null,
null,
null,
false,
null,
false
]
},
"hash": "70ddcf86865a315934843285e8ec618c47a5acbe400d79840c4a2d86f8886393"
"hash": "40999264f09a781c4393b50c2c41ae5a5e64086198cb67aba72345bb3cdf7773"
}
backend/.sqlx/query-69924462c788dbc8f31aacc7f8ae588d76bf1f25d631833ce4b194818a7d1437.json
Generated
-48
View File
@@ -1,48 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT result as \"result: sqlx::types::Json<Box<RawValue>>\", v2_job.tag,\n v2_job_queue.running as \"running: Option<bool>\", SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\", 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<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 1,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "running: Option<bool>",
"type_info": "Bool"
},
{
"ordinal": 3,
"name": "result_stream: Option<String>",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "stream_offset",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text",
"Uuid",
"Int4"
]
},
"nullable": [
true,
false,
false,
null,
null
]
},
"hash": "69924462c788dbc8f31aacc7f8ae588d76bf1f25d631833ce4b194818a7d1437"
}
backend/.sqlx/query-d8c209b177da2e147a3549c888969478cd80aa157700e5c1f3b9f4d12dd31a1d.json → backend/.sqlx/query-80809d397cf84f7278ebb276078871b371663257a127eb35512695c487066fd7.json
Generated
+16 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n COALESCE(jc.result, NULL) as \"result: sqlx::types::Json<Box<RawValue>>\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset\n FROM (\n SELECT $2::uuid as job_id, $1::text as workspace_id\n ) base\n LEFT JOIN v2_job_completed jc ON jc.id = base.job_id AND jc.workspace_id = base.workspace_id\n LEFT JOIN job_result_stream rs ON rs.job_id = base.job_id\n WHERE base.job_id = $2",
"query": "SELECT\n COALESCE(jc.result, NULL) as \"result: sqlx::types::Json<Box<RawValue>>\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset,\n COALESCE(js.flow_status, jc.flow_status) as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\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<Box<RawValue>>",
"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"
}
backend/.sqlx/query-8126b118704341846e88bd289f1afe83c07b7a8b422f48022994370b3e433f34.json
Generated
+55
View File
@@ -0,0 +1,55 @@
{
"db_name": "PostgreSQL",
"query": "SELECT \n jc.result as \"result: sqlx::types::Json<Box<RawValue>>\",\n v2_job.tag,\n v2_job_queue.running as \"running: Option<bool>\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",\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<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 1,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "running: Option<bool>",
"type_info": "Bool"
},
{
"ordinal": 3,
"name": "result_stream: Option<String>",
"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"
}
backend/.sqlx/query-6907eb134dc5dbf118387e073897f86574c92de16252b2b1c475ab8146e5343d.json → backend/.sqlx/query-a58a345f7082181f89e7f88929b7149791de48bc2e489edb55d63f67702cce05.json
Generated
+10 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n COALESCE(jc.result, jc.result) as \"result: sqlx::types::Json<Box<RawValue>>\",\n jq.running as \"running: Option<bool>\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset\n FROM (\n SELECT $1::uuid as job_id, $2::text as workspace_id\n ) base\n LEFT JOIN v2_job_completed jc ON jc.id = base.job_id AND jc.workspace_id = base.workspace_id\n LEFT JOIN v2_job_queue jq ON jq.id = base.job_id AND jq.workspace_id = base.workspace_id\n LEFT JOIN job_result_stream rs ON rs.job_id = base.job_id\n WHERE base.job_id = $1",
"query": "SELECT\n COALESCE(jc.result, jc.result) as \"result: sqlx::types::Json<Box<RawValue>>\",\n jq.running as \"running: Option<bool>\",\n SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",\n CHAR_LENGTH(rs.stream) + 1 AS stream_offset,\n 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"
}
backend/.sqlx/query-d8ef35b4990eb9b2a306494d5b9acde9f57cfaba047c5b7680e0dccd2c1507df.json
Generated
+40
View File
@@ -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"
}
backend/.sqlx/query-f17f914d2522bf7cb5de9d7ba5557ee0dce940039ab42fd39bf079d87b6cad8a.json
Generated
+29
View File
@@ -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"
}
backend/windmill-api/openapi.yaml
+11 -70
View File
@@ -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: |
backend/windmill-api/src/auth.rs
+1
View File
@@ -415,6 +415,7 @@ pub struct Tokened {
pub token: String,
}
#[derive(Clone, Debug)]
pub struct OptTokened {
#[allow(dead_code)]
pub token: Option<String>,
backend/windmill-api/src/jobs.rs
+358 -75
View File
@@ -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<DB>,
Extension(user_db): Extension<UserDB>,
Path((w_id, flow_path)): Path<(String, StripPath)>,
Query(run_query): Query<RunJobQuery>,
args: RawWebhookArgs,
) -> error::Result<Response> {
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<DB>,
Extension(user_db): Extension<UserDB>,
Path((w_id, script_path)): Path<(String, StripPath)>,
Query(run_query): Query<RunJobQuery>,
args: RawWebhookArgs,
) -> error::Result<Response> {
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<DB>,
Extension(user_db): Extension<UserDB>,
Path((w_id, script_hash)): Path<(String, ScriptHash)>,
Query(run_query): Query<RunJobQuery>,
args: RawWebhookArgs,
) -> error::Result<Response> {
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<Response> {
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<Postgres>,
run_query: RunJobQuery,
@@ -6303,6 +6493,7 @@ pub struct JobUpdateQuery {
pub no_logs: Option<bool>,
pub only_result: Option<bool>,
pub fast: Option<bool>,
pub is_flow: Option<bool>,
}
#[derive(Serialize, Debug)]
@@ -6331,6 +6522,8 @@ pub struct JobUpdate {
pub job: Option<Job>,
#[serde(skip_serializing_if = "Option::is_none")]
pub only_result: Option<Box<serde_json::value::RawValue>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub flow_stream_job_id: Option<Uuid>,
}
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<JobUpdateQuery>,
) -> JsonResult<JobUpdate> {
@@ -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<JobUpdateQuery>,
) -> Response {
let stream = get_job_update_sse_stream(
) -> error::Result<Response> {
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<ApiAuthed>,
opt_tokened: OptTokened,
db: DB,
@@ -6514,18 +6717,18 @@ fn get_job_update_sse_stream(
only_result: Option<bool>,
fast: Option<bool>,
no_logs: Option<bool>,
) -> impl futures::Stream<Item = JobUpdateSSEStream> {
let (tx, rx) = tokio::sync::mpsc::channel(32);
is_flow: Option<bool>,
tx: tokio::sync::mpsc::Sender<JobUpdateSSEStream>,
) -> () {
tokio::spawn(async move {
let mut log_offset = initial_log_offset;
let mut stream_offset = initial_stream_offset;
let mut last_update_hash: Option<String> = 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<Uuid>,
stream_offset: Option<i32>,
) -> error::Result<Option<(Option<String>, Option<i32>)>> {
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<bool>,
no_logs: Option<bool>,
is_flow: Option<bool>,
flow_stream_job_id: Option<Uuid>,
) -> error::Result<JobUpdate> {
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<Box<RawValue>>\", v2_job.tag,
v2_job_queue.running as \"running: Option<bool>\", SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\", CHAR_LENGTH(rs.stream) AS stream_offset
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<Box<RawValue>>\",
v2_job.tag,
v2_job_queue.running as \"running: Option<bool>\",
SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",
CHAR_LENGTH(rs.stream) AS stream_offset,
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<Box<RawValue>>\",
jq.running as \"running: Option<bool>\",
SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",
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<Box<RawValue>>\",
SUBSTR(rs.stream, $3) AS \"result_stream: Option<String>\",
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<Box<RawValue>>\",
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<Box<RawValue>>\",
(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<Box<RawValue>>\",
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<Box<RawValue>>| x.0),
only_result: None,
flow_stream_job_id,
})
}
}
backend/windmill-api/src/scopes.rs
+5 -2
View File
@@ -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! {
backend/windmill-api/src/users.rs
-9
View File
@@ -2131,7 +2131,6 @@ struct Runnable {
workspace: String,
endpoint_async: String,
endpoint_sync: String,
endpoint_openai_sync: String,
summary: String,
description: String,
schema: Option<serde_json::Value>,
@@ -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,
backend/windmill-common/src/flow_status.rs
+3
View File
@@ -43,6 +43,8 @@ pub struct FlowStatus {
pub approval_conditions: Option<ApprovalConditions>,
#[serde(skip_serializing_if = "Option::is_none")]
pub restarted_from: Option<RestartedFrom>,
#[serde(skip_serializing_if = "Option::is_none")]
pub stream_job: Option<Uuid>,
}
#[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,
}
}
backend/windmill-common/src/flows.rs
+11 -7
View File
@@ -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<usize> {
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)]
backend/windmill-common/src/result_stream.rs
+10 -5
View File
@@ -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<String> {
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(())
}
backend/windmill-queue/src/flow_status.rs
+5 -2
View File
@@ -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<Step> {
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(
backend/windmill-queue/src/jobs.rs
+2
View File
@@ -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;
backend/windmill-worker/src/ai_executor.rs
+2 -2
View File
@@ -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#"
backend/windmill-worker/src/ansible_executor.rs
+15 -3
View File
@@ -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
backend/windmill-worker/src/bash_executor.rs
+3
View File
@@ -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?;
backend/windmill-worker/src/bun_executor.rs
+11
View File
@@ -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?;
backend/windmill-worker/src/common.rs
+146
View File
@@ -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<Uuid>,
root_job: Uuid,
visited: Option<HashSet<Uuid>>,
) -> error::Result<bool> {
// 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<Self> {
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,
backend/windmill-worker/src/csharp_executor.rs
+3
View File
@@ -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
backend/windmill-worker/src/deno_executor.rs
+6 -1
View File
@@ -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());
backend/windmill-worker/src/go_executor.rs
+4
View File
@@ -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?;
backend/windmill-worker/src/handle_child.rs
+16 -2
View File
@@ -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<StreamNotifier>,
) -> error::Result<HandleChildResult> {
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<String>,
stream_notifier: Option<StreamNotifier>,
) {
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: {:?}",
backend/windmill-worker/src/java_executor.rs
+2
View File
@@ -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(())
backend/windmill-worker/src/js_eval.rs
+11 -1
View File
@@ -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<StreamNotifier>,
) -> anyhow::Result<Box<RawValue>> {
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<StreamNotifier>,
) -> windmill_common::error::Result<Box<RawValue>> {
let (sender, mut receiver) = oneshot::channel::<IsolateHandle>();
let (append_logs_sender, mut append_logs_receiver) = mpsc::unbounded_channel::<String>();
@@ -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}");
backend/windmill-worker/src/nu_executor.rs
+1
View File
@@ -340,6 +340,7 @@ async fn run<'a>(
false,
&mut Some(occupancy_metrics),
None,
None,
)
.await?;
Ok(())
backend/windmill-worker/src/php_executor.rs
+2
View File
@@ -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
backend/windmill-worker/src/python_executor.rs
+5 -1
View File
@@ -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?;
backend/windmill-worker/src/python_versions.rs
+1
View File
@@ -655,6 +655,7 @@ impl PyV {
false,
occupancy_metrics,
None,
None,
)
.await?;
Ok(())
backend/windmill-worker/src/ruby_executor.rs
+2
View File
@@ -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(())
backend/windmill-worker/src/rust_executor.rs
+6 -1
View File
@@ -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
backend/windmill-worker/src/universal_pkg_installer.rs
+2
View File
@@ -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
{
backend/windmill-worker/src/worker.rs
+4
View File
@@ -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?)
}
backend/windmill-worker/src/worker_flow.rs
+10 -8
View File
@@ -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(
frontend/src/lib/components/JobLoader.svelte
+6
View File
@@ -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()}`
frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte
+167 -76
View File
@@ -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`
<div class="flex flex-col gap-2">
<div class="flex flex-row justify-between">
<div class="text-sm font-normal text-secondary flex flex-row items-center">Request type</div>
<ToggleButtonGroup class="h-[30px] w-auto" bind:selected={webhookType}>
<ToggleButtonGroup class="h-[30px] w-auto" bind:selected={requestType}>
{#snippet children({ item })}
<ToggleButton
label="Async"
@@ -305,44 +375,63 @@ done`
</div>
<div class="flex flex-row justify-between">
<div class="text-sm font-normal text-secondary flex flex-row items-center">Call method</div>
<ToggleButtonGroup class="h-[30px] w-auto" bind:selected={requestType}>
<ToggleButtonGroup class="h-[30px] w-auto" bind:selected={callMethod}>
{#snippet children({ item })}
<ToggleButton
label="POST by path"
value="path"
icon={ArrowUpRight}
{item}
selectedColor="#fb923c"
/>
{#if !isFlow}
<ToggleButton
label="POST by hash"
value="hash"
icon={ArrowUpRight}
selectedColor="#fb923c"
disabled={!hash}
{item}
/>
{/if}
<ToggleButton
label="GET by path"
value="get_path"
label="POST"
icon={ArrowDownRight}
disabled={webhookType !== 'sync'}
{item}
selectedColor="#14b8a6"
value="post"
{item}
/>
<ToggleButton
label="GET"
icon={ArrowUpRight}
selectedColor="#fb923c"
value="get"
{item}
disabled={requestType !== 'sync'}
/>
<ToggleButton
label="SSE"
value="sse"
icon={RssIcon}
selectedColor="#3B82F6"
disabled={requestType !== 'sync'}
tooltip={'Returns an SSE stream. ' +
(isFlow
? 'Only useful if the last step of the flow returns a stream.'
: 'Only useful if the script returns a stream.')}
{item}
/>
{/snippet}
</ToggleButtonGroup>
</div>
{#if !isFlow}
<div class="flex flex-row justify-between">
<div class="text-sm font-normal text-secondary flex flex-row items-center">
Reference type
</div>
<ToggleButtonGroup class="h-[30px] w-auto" bind:selected={runnableId}>
{#snippet children({ item })}
<ToggleButton label="Path" value="path" {item} />
<ToggleButton label="Hash" value="hash" disabled={!hash} {item} />
{/snippet}
</ToggleButtonGroup>
</div>
{/if}
<div class="flex flex-row justify-between">
<div class="text-sm font-normal text-secondary flex flex-row items-center"
>Token configuration</div
>
<ToggleButtonGroup class="h-[30px] w-auto" bind:selected={tokenType}>
{#snippet children({ item })}
<ToggleButton label="Token in Headers" value="headers" {item} />
<ToggleButton
label="Token in Headers"
value="headers"
{item}
disabled={callMethod === 'sse'}
/>
<ToggleButton label="Token in Query" value="query" {item} />
{/snippet}
</ToggleButtonGroup>
@@ -354,10 +443,12 @@ done`
<div>
<Tabs bind:selected={selectedTab}>
<Tab value="rest" size="xs">REST</Tab>
{#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL}
{#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL && callMethod !== 'sse'}
<Tab value="curl" size="xs">Curl</Tab>
{/if}
<Tab value="fetch" size="xs">Fetch</Tab>
<Tab value="fetch" size="xs">
{callMethod === 'sse' ? 'Event Source' : 'Fetch'}
</Tab>
{#snippet content()}
{#key token}
@@ -367,12 +458,12 @@ done`
<ClipboardPanel content={url} />
</Label>
{#if requestType !== 'get_path'}
{#if callMethod !== 'get'}
<Label label="Body">
<ClipboardPanel content={JSON.stringify(cleanedRunnableArgs ?? {}, null, 2)} />
</Label>
{/if}
{#key requestType}
{#key callMethod}
{#key tokenType}
<Label label="Headers">
<ClipboardPanel content={JSON.stringify(headers(), null, 2)} />
@@ -384,8 +475,8 @@ done`
<TabContent value="curl" class="flex flex-col flex-1 h-full">
<div class="relative">
{#key runnableArgs}
{#key requestType}
{#key webhookType}
{#key callMethod}
{#key requestType}
{#key tokenType}
<div
class="flex flex-row flex-1 h-full border p-2 rounded-md overflow-auto relative"
@@ -405,8 +496,8 @@ done`
</TabContent>
<TabContent value="fetch">
{#key runnableArgs}
{#key requestType}
{#key webhookType}
{#key callMethod}
{#key requestType}
{#key tokenType}
{#key token}
<div