starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-09 16:05:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

introduce root_job and leaf_jobs for efficient result_by_id

Browse Source
This commit is contained in:
Ruben Fiszel
2023-03-03 12:44:44 +01:00
parent 304dea4b74
commit 58300eb6ac
15 changed files with 272 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20230303064952_add_root_leaf_jobs.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20230303064952_add_root_leaf_jobs.up.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add up migration script here
ALTER TABLE queue ADD COLUMN root_job uuid;
ALTER TABLE queue ADD COLUMN leaf_jobs jsonb;
backend/sqlx-data.json
+204 -158
View File
@@ -645,27 +645,6 @@
},
"query": "SELECT * FROM workspace_settings WHERE workspace_id = $1"
},
"176afff6dfceac701d499663f15fb7f569fd9d449406df4c5ecc811414429d42": {
"describe": {
"columns": [
{
"name": "flow_status",
"ordinal": 0,
"type_info": "Jsonb"
}
],
"nullable": [
null
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
}
},
"query": "SELECT flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2 \n UNION ALL SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2 "
},
"187a27d18f78f068319a0ec684f71a76c49dd09c5c74c2c777e35233c4f5a281": {
"describe": {
"columns": [
@@ -826,75 +805,6 @@
},
"query": "\n SELECT id, flow_status, suspend, script_path\n FROM queue\n WHERE id = $1\n "
},
"1e35c39bc786d638252e5483ca4efae9a041f7e845341f8bfd715ddd9e899499": {
"describe": {
"columns": [
{
"name": "id",
"ordinal": 0,
"type_info": "Uuid"
}
],
"nullable": [
false
],
"parameters": {
"Left": [
"Varchar",
"Uuid",
"Bool",
"Uuid",
"Varchar",
"Varchar",
"Timestamptz",
"Int8",
"Varchar",
"Text",
"Text",
"Jsonb",
{
"Custom": {
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies"
]
},
"name": "job_kind"
}
},
"Varchar",
"Jsonb",
"Jsonb",
"Bool",
{
"Custom": {
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash"
]
},
"name": "script_lang"
}
},
"Bool",
"Text",
"Varchar",
"Bool"
]
}
},
"query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22) RETURNING id"
},
"1eaf8d677d520c7f2f303a731de6b6d939918e41ad0d1c748d80db3fd33cb9d3": {
"describe": {
"columns": [],
@@ -1176,6 +1086,20 @@
},
"query": "SELECT set_config('session.folders_read', $1, true)"
},
"2a3ebe1b0eae5b2164894321e138cc4dc0293788aeb98d05d95d18dfc708d6a6": {
"describe": {
"columns": [],
"nullable": [],
"parameters": {
"Left": [
"Text",
"Jsonb",
"Uuid"
]
}
},
"query": "\n UPDATE queue\n SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)\n WHERE COALESCE((SELECT root_job FROM queue WHERE id = $3), $3) = id\n "
},
"2a4be8334db7d39f3d954193a8b0169cc4a4a07e081d2fa61d8764879d6a8ff5": {
"describe": {
"columns": [],
@@ -1395,6 +1319,27 @@
},
"query": "SELECT EXISTS(SELECT 1 FROM resource_type WHERE name = $1 AND (workspace_id = $2 OR workspace_id = 'admins'))"
},
"39e423f16f2ffb74df27760345a5340f2f89409f2a16247ac92a1949181d0cf0": {
"describe": {
"columns": [
{
"name": "leaf_jobs",
"ordinal": 0,
"type_info": "Jsonb"
}
],
"nullable": [
true
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
}
},
"query": "SELECT leaf_jobs FROM queue WHERE id = (SELECT root_job FROM queue WHERE id = $1) AND workspace_id = $2"
},
"39f1b99319f48acb95fd4247f70baaf7a38062aa8b4f70fac10215edbbd2d41c": {
"describe": {
"columns": [],
@@ -1893,6 +1838,28 @@
},
"query": "DELETE FROM usr_to_group WHERE group_ = $1 AND workspace_id = $2"
},
"52a96fa4034dec3edb0655c1cd125e6dacc63c10bbc6c0ea1789809807d720a3": {
"describe": {
"columns": [
{
"name": "?column?",
"ordinal": 0,
"type_info": "Jsonb"
}
],
"nullable": [
null
],
"parameters": {
"Left": [
"Text",
"Uuid",
"Text"
]
}
},
"query": "SELECT leaf_jobs->$1::text FROM queue WHERE id = $2 AND workspace_id = $3"
},
"52c8b4350235bdaab4df79e517d5e42a61a4e1e209d120b2c8bb31ebb7ce1e56": {
"describe": {
"columns": [],
@@ -2661,6 +2628,27 @@
},
"query": "SELECT EXISTS(SELECT 1 FROM usr WHERE workspace_id = $1 AND username = $2)"
},
"7165e50c7de4437ea1969aae4db51dfe040ce3b2f8dcf7eb123831697fcdbfd0": {
"describe": {
"columns": [
{
"name": "leaf_jobs",
"ordinal": 0,
"type_info": "Jsonb"
}
],
"nullable": [
true
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
}
},
"query": "SELECT leaf_jobs FROM queue WHERE id = $1 AND workspace_id = $2"
},
"72098030cab635723a9cecf8b3b1448e69a8afd68342850ef6376352d2897723": {
"describe": {
"columns": [],
@@ -3657,33 +3645,6 @@
},
"query": "SELECT is_admin FROM usr where username = $1 AND workspace_id = $2 AND disabled = false"
},
"92ce38eaaef9e793a49788bf3b7fc88340e83da57b785364a477f08c776f66f3": {
"describe": {
"columns": [
{
"name": "flow_status",
"ordinal": 0,
"type_info": "Jsonb"
},
{
"name": "parent_job",
"ordinal": 1,
"type_info": "Uuid"
}
],
"nullable": [
null,
null
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
}
},
"query": "SELECT flow_status, parent_job FROM completed_job WHERE id = $1 AND workspace_id = $2 \n UNION ALL SELECT flow_status, parent_job FROM queue WHERE id = $1 AND workspace_id = $2 "
},
"930ad84a4db26fa6d2c8447d447099e944a0ea7b1266b4d02cee620fe3d761a5": {
"describe": {
"columns": [],
@@ -3830,6 +3791,26 @@
},
"query": "UPDATE schedule SET enabled = $1, email = $2 WHERE path = $3 AND workspace_id = $4 RETURNING *"
},
"971175f6169857c3e1cdc08ac8aeed57300b7792e1797a9cdd73c9b3967cd7b9": {
"describe": {
"columns": [
{
"name": "root_job",
"ordinal": 0,
"type_info": "Uuid"
}
],
"nullable": [
true
],
"parameters": {
"Left": [
"Uuid"
]
}
},
"query": "SELECT root_job FROM queue WHERE id = $1"
},
"97e364c703bdcdfb5345810659cbe0477a28b8199ef0b297f9a22c88a43b6b5c": {
"describe": {
"columns": [
@@ -4009,6 +3990,76 @@
},
"query": "SELECT content FROM script WHERE path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND archived = false AND workspace_id = $2)"
},
"a1c41bbeb2d64fa1e7dfd2ed053191a1de5d786ae8c22e225e450865ecac94e9": {
"describe": {
"columns": [
{
"name": "id",
"ordinal": 0,
"type_info": "Uuid"
}
],
"nullable": [
false
],
"parameters": {
"Left": [
"Varchar",
"Uuid",
"Bool",
"Uuid",
"Varchar",
"Varchar",
"Timestamptz",
"Int8",
"Varchar",
"Text",
"Text",
"Jsonb",
{
"Custom": {
"kind": {
"Enum": [
"script",
"preview",
"flow",
"dependencies",
"flowpreview",
"script_hub",
"identity",
"flowdependencies"
]
},
"name": "job_kind"
}
},
"Varchar",
"Jsonb",
"Jsonb",
"Bool",
{
"Custom": {
"kind": {
"Enum": [
"python3",
"deno",
"go",
"bash"
]
},
"name": "script_lang"
}
},
"Bool",
"Text",
"Varchar",
"Bool",
"Uuid"
]
}
},
"query": "INSERT INTO queue\n (workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for, \n script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23) RETURNING id"
},
"a227548b6604c56bfc15eb780bd8ee72a89dc6701a50f5048e928bd87baa7b9a": {
"describe": {
"columns": [
@@ -4119,6 +4170,28 @@
},
"query": "UPDATE flow SET dependency_job = $1 WHERE path = $2 AND workspace_id = $3"
},
"a5f9fb82791103e2bbaf9cb6d87e8c50495d12d87f8ed83382068203a8dd7a67": {
"describe": {
"columns": [
{
"name": "?column?",
"ordinal": 0,
"type_info": "Jsonb"
}
],
"nullable": [
null
],
"parameters": {
"Left": [
"Text",
"Uuid",
"Text"
]
}
},
"query": "SELECT leaf_jobs->$1::text FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3"
},
"a6145b0482c9e5da245059a80b1563cad20318fd2dd8aef33f9ca97de1826b8b": {
"describe": {
"columns": [],
@@ -4821,26 +4894,6 @@
},
"query": "UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND workspace_id = $3 AND is_flow IS false RETURNING *"
},
"c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc": {
"describe": {
"columns": [
{
"name": "customer_id",
"ordinal": 0,
"type_info": "Varchar"
}
],
"nullable": [
true
],
"parameters": {
"Left": [
"Text"
]
}
},
"query": "SELECT customer_id FROM workspace_settings WHERE workspace_id = $1"
},
"c1149f74ef0cca4a48be80f523d339d78e8e20e2bfacac9ad1b6b75c5ae370fb": {
"describe": {
"columns": [],
@@ -4875,27 +4928,6 @@
},
"query": "SELECT result FROM completed_job WHERE id = $1"
},
"c2d0e44faab6981a21ca28dfd6f4eef9dfcafb471852e701c0bbc8ae11344325": {
"describe": {
"columns": [
{
"name": "parent_job",
"ordinal": 0,
"type_info": "Uuid"
}
],
"nullable": [
null
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
}
},
"query": "SELECT parent_job FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT parent_job FROM queue WHERE id = $1 AND workspace_id = $2"
},
"c2d6cb56c1dea4498e2aab9ea9301dbbaa127602a38f57f5add4108fdc209b1a": {
"describe": {
"columns": [
@@ -5816,6 +5848,20 @@
},
"query": "DELETE FROM password WHERE email = $1"
},
"e990488d2238fe229a213c975102b85ea2e159142bb2d34bc83835322496f2b9": {
"describe": {
"columns": [],
"nullable": [],
"parameters": {
"Left": [
"Text",
"Jsonb",
"Uuid"
]
}
},
"query": "\n UPDATE queue\n SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)\n WHERE (SELECT root_job FROM queue WHERE id = $3) = id\n "
},
"e9c0e331c16312bf086b17c91466c5389d41454fd3f18d73c2e9554845ee9a72": {
"describe": {
"columns": [],
backend/tests/worker.rs
+1
View File
@@ -836,6 +836,7 @@ impl RunJob {
/* scheduled_for_o */ None,
/* schedule_path */ None,
/* parent_job */ None,
/* root job */ None,
/* is_flow_step */ false,
/* running */ false,
None,
backend/windmill-api/openapi.yaml
-5
View File
@@ -2538,11 +2538,6 @@ paths:
required: true
schema:
type: string
- name: skip_direct
description: Skip checking that the node is part of the given flow.
in: query
schema:
type: boolean
responses:
"200":
description: job result
backend/windmill-api/src/apps.rs
+1
View File
@@ -670,6 +670,7 @@ async fn execute_component(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-api/src/flows.rs
+2
View File
@@ -237,6 +237,7 @@ async fn create_flow(
None,
None,
None,
None,
false,
false,
None,
@@ -393,6 +394,7 @@ async fn update_flow(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-api/src/jobs.rs
+12 -8
View File
@@ -102,10 +102,9 @@ pub fn global_service() -> Router {
async fn get_result_by_id(
Extension(db): Extension<DB>,
Query(ResultByIdQuery { skip_direct }): Query<ResultByIdQuery>,
Path((w_id, flow_id, node_id)): Path<(String, String, String)>,
Path((w_id, flow_id, node_id)): Path<(String, Uuid, String)>,
) -> windmill_common::error::JsonResult<serde_json::Value> {
let res = windmill_queue::get_result_by_id(db, skip_direct, w_id, flow_id, node_id).await?;
let res = windmill_queue::get_result_by_id(db, w_id, flow_id, node_id).await?;
Ok(Json(res))
}
@@ -177,11 +176,6 @@ async fn get_job(
Ok(Json(job))
}
#[derive(Deserialize)]
pub struct ResultByIdQuery {
pub skip_direct: bool,
}
pub async fn get_job_by_id<'c>(
mut tx: Transaction<'c, Postgres>,
w_id: &str,
@@ -1078,6 +1072,8 @@ impl From<UnifiedJob> for Job {
visible_to_owner: uj.visible_to_owner,
suspend: uj.suspend,
mem_peak: uj.mem_peak,
root_job: None,
leaf_jobs: None,
}),
t => panic!("job type {} not valid", t),
}
@@ -1173,6 +1169,7 @@ pub async fn run_flow_by_path(
scheduled_for,
None,
run_query.parent_job,
run_query.parent_job,
false,
false,
None,
@@ -1208,6 +1205,7 @@ pub async fn run_job_by_path(
scheduled_for,
None,
run_query.parent_job,
run_query.parent_job,
false,
false,
None,
@@ -1360,6 +1358,7 @@ pub async fn run_wait_result_job_by_path(
scheduled_for,
None,
run_query.parent_job,
run_query.parent_job,
false,
false,
None,
@@ -1406,6 +1405,7 @@ pub async fn run_wait_result_job_by_hash(
scheduled_for,
None,
run_query.parent_job,
run_query.parent_job,
false,
false,
None,
@@ -1451,6 +1451,7 @@ pub async fn run_wait_result_flow_by_path(
scheduled_for,
None,
run_query.parent_job,
run_query.parent_job,
false,
false,
None,
@@ -1513,6 +1514,7 @@ async fn run_preview_job(
scheduled_for,
None,
None,
None,
false,
false,
None,
@@ -1546,6 +1548,7 @@ async fn run_preview_flow_job(
scheduled_for,
None,
None,
None,
false,
false,
None,
@@ -1581,6 +1584,7 @@ pub async fn run_job_by_hash(
scheduled_for,
None,
run_query.parent_job,
run_query.parent_job,
false,
false,
None,
backend/windmill-api/src/oauth2.rs
+1
View File
@@ -785,6 +785,7 @@ async fn slack_command(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-api/src/scripts.rs
+1
View File
@@ -377,6 +377,7 @@ async fn create_script(
None,
None,
None,
None,
false,
false,
None,
backend/windmill-common/src/flow_status.rs
+1 -1
View File
@@ -125,7 +125,7 @@ pub enum FlowStatusModule {
},
}
#[derive(Debug, Clone)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum JobResult {
SingleJob(Uuid),
ListJob(Vec<Uuid>),
backend/windmill-queue/src/jobs.rs
+26 -98
View File
@@ -6,10 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
use std::{
collections::{HashMap, VecDeque},
str::FromStr,
};
use std::collections::HashMap;
use anyhow::Context;
use reqwest::Client;
@@ -20,7 +17,7 @@ use ulid::Ulid;
use uuid::Uuid;
use windmill_audit::{audit_log, ActionKind};
use windmill_common::{
error::{self, to_anyhow, Error},
error::{self, Error},
flow_status::{FlowStatus, JobResult, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL},
flows::{FlowModule, FlowModuleValue, FlowValue},
scripts::{get_full_hub_script_by_path, HubScript, ScriptHash, ScriptLang},
@@ -153,102 +150,26 @@ pub async fn pull(
Ok(job)
}
pub async fn find_recursively_downward(
db: &Pool<Postgres>,
w_id: &str,
flow_id: Uuid,
node_id: &str,
) -> windmill_common::error::Result<Option<JobResult>> {
let mut bfs_stack = VecDeque::new();
bfs_stack.push_back(flow_id);
while bfs_stack.len() > 0 {
let parent_id = bfs_stack.pop_front().unwrap();
let job = sqlx::query_scalar!(
"SELECT flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2
UNION ALL SELECT flow_status FROM queue WHERE id = $1 AND workspace_id = $2 ",
parent_id,
w_id
)
.fetch_optional(db)
.await?
.flatten();
if let Some(r) = job {
let status = serde_json::from_value::<FlowStatus>(r).map_err(to_anyhow)?;
for m in status.modules.iter() {
let id = m.id();
if id == node_id {
return Ok(m.job_result());
}
if let Some(job_id) = m.job() {
bfs_stack.push_back(job_id);
}
}
}
}
Ok(None)
}
pub async fn get_result_by_id(
db: Pool<Postgres>,
mut skip_direct: bool,
w_id: String,
flow_id: String,
flow_id: Uuid,
node_id: String,
) -> error::Result<serde_json::Value> {
let mut result_id: Option<JobResult> = None;
let mut parent_id = Uuid::from_str(&flow_id).ok();
let mut lparent_id = parent_id.clone();
while result_id.is_none() && parent_id.is_some() {
if !skip_direct {
let r = sqlx::query!(
"SELECT flow_status, parent_job FROM completed_job WHERE id = $1 AND workspace_id = $2
UNION ALL SELECT flow_status, parent_job FROM queue WHERE id = $1 AND workspace_id = $2 ",
parent_id.unwrap(),
w_id,
)
.fetch_optional(&db)
.await?;
if let Some(r) = r {
let value = r
.flow_status
.as_ref()
.ok_or_else(|| Error::InternalErr(format!("requiring a flow status value")))?
.to_owned();
lparent_id = parent_id;
parent_id = r.parent_job;
let status_o = serde_json::from_value::<FlowStatus>(value).ok();
result_id = status_o.and_then(|status| {
status
.modules
.iter()
.find(|m| m.id() == node_id)
.and_then(|m| m.job_result())
});
} else {
parent_id = None;
}
} else {
let q_parent = sqlx::query_scalar!(
"SELECT parent_job FROM completed_job WHERE id = $1 AND workspace_id = $2 UNION ALL SELECT parent_job FROM queue WHERE id = $1 AND workspace_id = $2",
parent_id.unwrap(),
w_id,
)
.fetch_optional(&db)
.await?
.flatten();
lparent_id = parent_id;
parent_id = q_parent;
skip_direct = false
}
}
// we could not find the node going upward from the flow by looking at all the jobs (in progress or completed)
// we now look downward from the flow root to the all the children completed job for a job that might hide itself
// in a deep non-direct parent job such as in nested branches
if result_id.is_none() && lparent_id.is_some() {
result_id = find_recursively_downward(&db, &w_id, lparent_id.unwrap(), &node_id).await?;
}
let job_result: Option<JobResult> = sqlx::query_scalar!(
"SELECT leaf_jobs->$1::text FROM queue WHERE COALESCE((SELECT root_job FROM queue WHERE id = $2), $2) = id AND workspace_id = $3",
node_id,
flow_id,
w_id,
)
.fetch_optional(&db)
.await?
.flatten()
.map(|x| serde_json::from_value(x).ok())
.flatten();
let result_id = windmill_common::utils::not_found_if_none(
result_id,
job_result,
"Flow result by id",
format!("{}, {}", flow_id, node_id),
)?;
@@ -330,6 +251,7 @@ pub async fn push<'c>(
scheduled_for_o: Option<chrono::DateTime<chrono::Utc>>,
schedule_path: Option<String>,
parent_job: Option<Uuid>,
root_job: Option<Uuid>,
is_flow_step: bool,
mut same_worker: bool,
pre_run_error: Option<&windmill_common::error::Error>,
@@ -582,12 +504,13 @@ pub async fn push<'c>(
.unwrap_or_else(|| (None, None));
let flow_status = raw_flow.as_ref().map(FlowStatus::new);
let uuid = sqlx::query_scalar!(
"INSERT INTO queue
(workspace_id, id, running, parent_job, created_by, permissioned_as, scheduled_for,
script_hash, script_path, raw_code, raw_lock, args, job_kind, schedule_path, raw_flow, \
flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22) \
flow_status, is_flow_step, language, started_at, same_worker, pre_run_error, email, visible_to_owner, root_job)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, CASE WHEN $3 THEN now() END, $19, $20, $21, $22, $23) \
RETURNING id",
workspace_id,
job_id,
@@ -610,7 +533,8 @@ pub async fn push<'c>(
same_worker,
pre_run_error.map(|e| e.to_string()),
email,
visible_to_owner
visible_to_owner,
root_job
)
.fetch_one(&mut tx)
.await
@@ -723,6 +647,10 @@ pub struct QueuedJob {
pub suspend: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub mem_peak: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub root_job: Option<Uuid>,
#[serde(skip_serializing_if = "Option::is_none")]
pub leaf_jobs: Option<serde_json::Value>,
}
impl QueuedJob {
backend/windmill-queue/src/schedule.rs
+1
View File
@@ -85,6 +85,7 @@ pub async fn push_scheduled_job<'c>(
Some(next),
Some(schedule.path.clone()),
None,
None,
false,
false,
None,
backend/windmill-worker/src/js_eval.rs
+1 -1
View File
@@ -315,7 +315,7 @@ async fn op_get_id(args: Vec<String>) -> Result<Option<serde_json::Value>, anyho
let client = windmill_api_client::create_client(base_url, token.clone());
let result = client
.result_by_id(workspace, flow_job_id, node_id, Some(true))
.result_by_id(workspace, flow_job_id, node_id)
.await
.map_or(None, |e| Some(e.into_inner()));
backend/windmill-worker/src/worker_flow.rs
+17
View File
@@ -326,6 +326,21 @@ pub async fn update_flow_status_after_job_completion(
)
.execute(&mut tx)
.await?;
if let Some(job_result) = new_status.job_result() {
sqlx::query!(
"
UPDATE queue
SET leaf_jobs = JSONB_SET(coalesce(leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)
WHERE COALESCE((SELECT root_job FROM queue WHERE id = $3), $3) = id
",
new_status.id(),
json!(job_result),
flow
)
.execute(&mut tx)
.await?;
}
}
}
@@ -1237,6 +1252,7 @@ async fn push_next_flow_job(
Ok(v) => (Some(v), None),
Err(e) => (None, Some(e)),
};
let root_job = flow_job.root_job.or_else(|| Some(flow_job.id));
let (uuid, inner_tx) = push(
tx,
&flow_job.workspace_id,
@@ -1248,6 +1264,7 @@ async fn push_next_flow_job(
scheduled_for_o,
flow_job.schedule_path.clone(),
Some(flow_job.id),
root_job,
true,
continue_on_same_worker,
err,