kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(backend): get_result_by_id do a downward pass to find node at any depth (#1249)

Browse Source 
* downwardRec

* downwardRec

* any node

* any node

* any node
This commit is contained in:
Ruben Fiszel
2023-03-01 11:33:48 +01:00
committed by GitHub
parent 9afa65de65
commit 10f4bf16c2
3 changed files with 109 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
backend/windmill-queue/src/jobs.rs
View File

@@ -6,7 +6,10 @@
* LICENSE-AGPL for a copy of the license.
*/
use std::{collections::HashMap, str::FromStr};
use std::{
collections::{HashMap, VecDeque},
str::FromStr,
};
use anyhow::Context;
use reqwest::Client;
@@ -17,7 +20,7 @@ use ulid::Ulid;
use uuid::Uuid;
use windmill_audit::{audit_log, ActionKind};
use windmill_common::{
error::{self, Error},
error::{self, to_anyhow, 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},
@@ -150,6 +153,41 @@ 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,
@@ -159,10 +197,12 @@ pub async fn get_result_by_id(
) -> 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 ",
"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,
)
@@ -174,6 +214,7 @@ pub async fn get_result_by_id(
.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| {
@@ -195,10 +236,17 @@ pub async fn get_result_by_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 result_id = windmill_common::utils::not_found_if_none(
result_id,
"Flow result by id",