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

backend: rework raw values fetching logic (#4770)

Browse Source
This commit is contained in:
Lucas Abel
2024-11-21 13:59:42 +01:00
committed by GitHub
parent 9d1a50818f
commit 002f0b92ea
15 changed files with 355 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0722f461da9edb169c3c73423440eaa3e1b9f2ca3b7033bd7ed978a3b4ddf3fc.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT raw_flow AS \"raw_flow!: Json<Box<JsonRawValue>>\"\n FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "raw_flow!: Json<Box<JsonRawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true
]
},
"hash": "0722f461da9edb169c3c73423440eaa3e1b9f2ca3b7033bd7ed978a3b4ddf3fc"
}
backend/.sqlx/query-217a5291438d23597b2c7f05d2c481f406364d56a129089a268f6423c548bca6.json
Generated
+35
View File
@@ -0,0 +1,35 @@
{
"db_name": "PostgreSQL",
"query": "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json<Box<JsonRawValue>>\"\n FROM queue WHERE id = $1 AND workspace_id = $2 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "raw_code",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "raw_lock",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "raw_flow: Json<Box<JsonRawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
true,
true
]
},
"hash": "217a5291438d23597b2c7f05d2c481f406364d56a129089a268f6423c548bca6"
}
backend/.sqlx/query-45950064cce9f53f73a01ddcd6911ec677297009b71041d39019c4700a571c0f.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM completed_job WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "45950064cce9f53f73a01ddcd6911ec677297009b71041d39019c4700a571c0f"
}
backend/.sqlx/query-b3a9ab85a78edf292ddb0af33f7435bbcc84f63c9079e29d84043f258eba208e.json
Generated
+29
View File
@@ -0,0 +1,29 @@
{
"db_name": "PostgreSQL",
"query": "SELECT\n flow_status AS \"flow_status!: Json<Box<RawValue>>\",\n coalesce(job.raw_flow, queue.raw_flow)->'modules'->(flow_status->'step')::int AS \"module: Json<Box<RawValue>>\"\n FROM queue LEFT JOIN job USING(id, workspace_id) WHERE id = $1 AND workspace_id = $2 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "flow_status!: Json<Box<RawValue>>",
"type_info": "Jsonb"
},
{
"ordinal": 1,
"name": "module: Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
null
]
},
"hash": "b3a9ab85a78edf292ddb0af33f7435bbcc84f63c9079e29d84043f258eba208e"
}
backend/.sqlx/query-da861c06f6448f68069ba3be9b677641c5c6b2c1e4960c74db27dbaffcbd4c40.json
Generated
+35
View File
@@ -0,0 +1,35 @@
{
"db_name": "PostgreSQL",
"query": "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json<Box<RawValue>>\"\n FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "raw_code",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "raw_lock",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "raw_flow: Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
true,
true
]
},
"hash": "da861c06f6448f68069ba3be9b677641c5c6b2c1e4960c74db27dbaffcbd4c40"
}
backend/.sqlx/query-e1923bc755bd6b8cc871ae9381a97d3c63a0fce3dd57c93f48a2136f7395fa3a.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM job WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "e1923bc755bd6b8cc871ae9381a97d3c63a0fce3dd57c93f48a2136f7395fa3a"
}
backend/.sqlx/query-f24303a2386575a750a48a21d1b6cdecd6e64a0415c60ad1495f100c5d85bbe1.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT raw_flow AS \"raw_flow!: Json<Box<sqlx::types::JsonRawValue>>\"\n FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "raw_flow!: Json<Box<sqlx::types::JsonRawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true
]
},
"hash": "f24303a2386575a750a48a21d1b6cdecd6e64a0415c60ad1495f100c5d85bbe1"
}
backend/.sqlx/query-fb1a32318b35ec5c8129eb3660b79eb5e6d1e01fcf01cc05d3c7ddf47295c2f5.json
Generated
+35
View File
@@ -0,0 +1,35 @@
{
"db_name": "PostgreSQL",
"query": "SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json<Box<JsonRawValue>>\"\n FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "raw_code",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "raw_lock",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "raw_flow: Json<Box<JsonRawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
true,
true,
true
]
},
"hash": "fb1a32318b35ec5c8129eb3660b79eb5e6d1e01fcf01cc05d3c7ddf47295c2f5"
}
backend/Cargo.lock
Generated
+1
View File
@@ -11110,6 +11110,7 @@ dependencies = [
"chrono",
"chrono-tz 0.10.0",
"cron",
"futures",
"futures-core",
"hex",
"hmac",
backend/windmill-common/src/flows.rs
-27
View File
@@ -16,7 +16,6 @@ use rand::Rng;
use serde::{Deserialize, Serialize, Serializer};
use crate::{
error::Error,
more_serde::{default_empty_string, default_id, default_null, default_true, is_default},
scripts::{Schema, ScriptHash, ScriptLang},
};
@@ -652,29 +651,3 @@ pub fn add_virtual_items_if_necessary(modules: &mut Vec<FlowModule>) {
});
}
}
pub async fn has_failure_module<'c>(flow: sqlx::types::Uuid, db: &sqlx::Pool<sqlx::Postgres>, completed: bool) -> Result<bool, Error> {
if completed {
sqlx::query_scalar!(
"SELECT raw_flow->'failure_module' != 'null'::jsonb
FROM completed_job_view
WHERE id = $1",
flow
)
} else {
sqlx::query_scalar!(
"SELECT raw_flow->'failure_module' != 'null'::jsonb
FROM queue_view
WHERE id = $1",
flow
)
}
.fetch_one(db)
.await
.map_err(|e| {
Error::InternalErr(format!(
"error during retrieval of has_failure_module: {e:#}"
))
})
.map(|v| v.unwrap_or(false))
}
backend/windmill-common/src/worker.rs
+6 -4
View File
@@ -119,8 +119,9 @@ pub async fn make_suspended_pull_query(wc: &WorkerConfig) {
canceled_reason, last_ping, job_kind, schedule_path, permissioned_as,
flow_status, is_flow_step, language, suspend, suspend_until,
same_worker, pre_run_error, email, visible_to_owner, mem_peak,
root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s,
timeout, flow_step_id, cache_ttl, priority", wc.worker_tags.iter().map(|x| format!("'{x}'")).join(", "));
root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s,
timeout, flow_step_id, cache_ttl, priority,
raw_code, raw_lock, raw_flow", wc.worker_tags.iter().map(|x| format!("'{x}'")).join(", "));
let mut l = WORKER_SUSPENDED_PULL_QUERY.write().await;
*l = query;
}
@@ -150,8 +151,9 @@ pub async fn make_pull_query(wc: &WorkerConfig) {
canceled_reason, last_ping, job_kind, schedule_path, permissioned_as,
flow_status, is_flow_step, language, suspend, suspend_until,
same_worker, pre_run_error, email, visible_to_owner, mem_peak,
root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s,
timeout, flow_step_id, cache_ttl, priority", tags.tags.iter().map(|x| format!("'{x}'")).join(", "));
root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s,
timeout, flow_step_id, cache_ttl, priority,
raw_code, raw_lock, raw_flow", tags.tags.iter().map(|x| format!("'{x}'")).join(", "));
queries.push(query);
}
backend/windmill-queue/Cargo.toml
+1
View File
@@ -37,6 +37,7 @@ cron.workspace = true
rsmq_async.workspace = true
tokio.workspace = true
futures-core.workspace = true
futures.workspace = true
itertools.workspace = true
async-recursion.workspace = true
bigdecimal.workspace = true
backend/windmill-queue/src/jobs.rs
+71 -28
View File
@@ -17,6 +17,7 @@ use axum::{
response::{IntoResponse, Response},
};
use chrono::{DateTime, Duration, Utc};
use futures::future::TryFutureExt;
use itertools::Itertools;
#[cfg(feature = "prometheus")]
use prometheus::IntCounter;
@@ -71,9 +72,6 @@ use windmill_common::BASE_URL;
#[cfg(feature = "cloud")]
use windmill_common::users::SUPERADMIN_SYNC_EMAIL;
#[cfg(feature = "enterprise")]
use windmill_common::flows::has_failure_module;
#[cfg(feature = "enterprise")]
use windmill_common::worker::CLOUD_HOSTED;
@@ -584,12 +582,21 @@ pub async fn add_completed_job<
let (raw_code, raw_lock, raw_flow) = if !*MIN_VERSION_IS_AT_LEAST_1_427.read().await {
sqlx::query!(
"SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json<Box<JsonRawValue>>\"
FROM queue_view WHERE id = $1 AND workspace_id = $2 LIMIT 1",
FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1",
&job_id, &queued_job.workspace_id
)
.fetch_one(db)
.map_ok(|record| (record.raw_code, record.raw_lock, record.raw_flow))
.or_else(|_| {
sqlx::query!(
"SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json<Box<JsonRawValue>>\"
FROM queue WHERE id = $1 AND workspace_id = $2 LIMIT 1",
&job_id, &queued_job.workspace_id
)
.fetch_one(db)
.map_ok(|record| (record.raw_code, record.raw_lock, record.raw_flow))
})
.await
.map(|record| (record.raw_code, record.raw_lock, record.raw_flow))
.unwrap_or_default()
} else {
(None, None, None)
@@ -923,6 +930,18 @@ pub async fn add_completed_job<
#[cfg(feature = "enterprise")]
if !success {
async fn has_failure_module(db: &Pool<Postgres>, job_id: Uuid) -> bool {
sqlx::query_scalar!("SELECT raw_flow->'failure_module' != 'null'::jsonb FROM job WHERE id = $1", job_id)
.fetch_one(db)
.or_else(|_|
sqlx::query_scalar!("SELECT raw_flow->'failure_module' != 'null'::jsonb FROM completed_job WHERE id = $1", job_id)
.fetch_one(db)
)
.await
.unwrap_or(Some(false))
.unwrap_or(false)
}
if queued_job.email == ERROR_HANDLER_USER_EMAIL {
let base_url = BASE_URL.read().await;
let w_id = &queued_job.workspace_id;
@@ -964,7 +983,7 @@ pub async fn add_completed_job<
} else if !skip_downstream_error_handlers
&& (matches!(queued_job.job_kind, JobKind::Script)
|| matches!(queued_job.job_kind, JobKind::Flow)
&& !has_failure_module(job_id, db, true).await.unwrap_or(false))
&& !has_failure_module(db, job_id).await)
&& queued_job.parent_job.is_none()
{
let result = serde_json::from_str(
@@ -1779,11 +1798,27 @@ async fn handle_successful_schedule<
Ok(())
}
#[derive(sqlx::FromRow)]
pub struct PulledJob {
#[sqlx(flatten)]
pub job: QueuedJob,
pub raw_code: Option<String>,
pub raw_lock: Option<String>,
pub raw_flow: Option<Json<Box<RawValue>>>,
}
impl std::ops::Deref for PulledJob {
type Target = QueuedJob;
fn deref(&self) -> &Self::Target {
&self.job
}
}
pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
db: &Pool<Postgres>,
rsmq: Option<R>,
suspend_first: bool,
) -> windmill_common::error::Result<(Option<QueuedJob>, bool)> {
) -> windmill_common::error::Result<(Option<PulledJob>, bool)> {
loop {
let (job, suspended) = pull_single_job_and_mark_as_running_no_concurrency_limit(
db,
@@ -1792,11 +1827,11 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
)
.await?;
if job.is_none() {
let Some(job) = job else {
return Ok((None, suspended));
}
};
let has_concurent_limit = job.as_ref().unwrap().concurrent_limit.is_some();
let has_concurent_limit = job.concurrent_limit.is_some();
#[cfg(not(feature = "enterprise"))]
if has_concurent_limit {
@@ -1807,7 +1842,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
let has_concurent_limit = false;
// concurrency check. If more than X jobs for this path are already running, we re-queue and pull another job from the queue
let pulled_job = job.unwrap();
let pulled_job = job;
if pulled_job.script_path.is_none() || !has_concurent_limit || pulled_job.canceled {
#[cfg(feature = "prometheus")]
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
@@ -1987,7 +2022,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
let job_log_event = format!(
"\nRe-scheduled job to {estimated_next_schedule_timestamp} due to concurrency limits with key {job_concurrency_key} and limit {job_custom_concurrent_limit} in the last {job_custom_concurrency_time_window_s} seconds",
);
let _ = append_logs(&job_uuid, pulled_job.workspace_id, job_log_event, db).await;
let _ = append_logs(&job_uuid, &pulled_job.workspace_id, job_log_event, db).await;
if rsmq.is_some() {
// if let Some(ref mut rsmq) = tx.rsmq {
// if using redis, only one message at a time can be poped from the queue. Process only this message and move to the next elligible job
@@ -2041,8 +2076,8 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
db: &Pool<Postgres>,
rsmq: Option<R>,
suspend_first: bool,
) -> windmill_common::error::Result<(Option<QueuedJob>, bool)> {
let job_and_suspended: (Option<QueuedJob>, bool) = if let Some(mut rsmq) = rsmq {
) -> windmill_common::error::Result<(Option<PulledJob>, bool)> {
let job_and_suspended: (Option<PulledJob>, bool) = if let Some(mut rsmq) = rsmq {
#[cfg(feature = "benchmark")]
let instant = Instant::now();
@@ -2074,7 +2109,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
.map_err(|_| anyhow::anyhow!("Failed to parsed Redis message"))?,
);
let m2r = sqlx::query_as::<_, QueuedJob>(
let m2r = sqlx::query_as::<_, PulledJob>(
"UPDATE queue
SET running = true
, started_at = coalesce(started_at, now())
@@ -2087,7 +2122,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
flow_status, is_flow_step, language, suspend, suspend_until,
same_worker, pre_run_error, email, visible_to_owner, mem_peak,
root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s,
timeout, flow_step_id, cache_ttl, priority",
timeout, flow_step_id, cache_ttl, priority, raw_code, raw_lock, raw_flow",
)
.bind(uuid)
.fetch_optional(db)
@@ -2121,7 +2156,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
let r = if suspend_first {
// tracing::info!("Pulling job with query: {}", query);
sqlx::query_as::<_, QueuedJob>(&query)
sqlx::query_as::<_, PulledJob>(&query)
.fetch_optional(db)
.await?
} else {
@@ -2130,7 +2165,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
if r.is_none() {
// #[cfg(feature = "benchmark")]
// let instant = Instant::now();
let mut highest_priority_job: Option<QueuedJob> = None;
let mut highest_priority_job: Option<PulledJob> = None;
let queries = WORKER_PULL_QUERIES.read().await;
@@ -2141,7 +2176,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
for query in queries.iter() {
// tracing::info!("Pulling job with query: {}", query);
let r = sqlx::query_as::<_, QueuedJob>(query)
let r = sqlx::query_as::<_, PulledJob>(query)
.fetch_optional(db)
.await?;
@@ -2581,7 +2616,7 @@ pub async fn job_is_complete(db: &DB, id: Uuid, w_id: &str) -> error::Result<boo
.unwrap_or(false))
}
pub async fn get_queued_job_tx<'c>(
async fn get_queued_job_tx<'c>(
id: Uuid,
w_id: &str,
tx: &mut Transaction<'c, Postgres>,
@@ -4112,7 +4147,14 @@ async fn restarted_flows_resolution(
),
Error,
> {
let completed_job = sqlx::query_as::<_, CompletedJob>(
#[derive(sqlx::FromRow)]
struct CompletedJobWithRawFlow {
#[sqlx(flatten)]
completed_job: CompletedJob,
raw_flow: Option<Json<Box<RawValue>>>,
}
let CompletedJobWithRawFlow { completed_job, raw_flow } = sqlx::query_as::<_, CompletedJobWithRawFlow>(
"SELECT *, null as labels FROM completed_job WHERE id = $1 and workspace_id = $2",
)
.bind(completed_flow_id)
@@ -4127,22 +4169,23 @@ async fn restarted_flows_resolution(
})?;
let flow_value = if let Some(flow_value) = flow_value_if_any {
flow_value
Some(flow_value)
} else if let Some(raw_flow) = raw_flow.as_ref() {
serde_json::from_str::<FlowValue>(raw_flow.get()).ok()
} else {
sqlx::query_scalar!(
"SELECT raw_flow AS \"raw_flow!: Json<Box<JsonRawValue>>\"
FROM completed_job_view WHERE id = $1 AND workspace_id = $2 LIMIT 1",
FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1",
&completed_flow_id, workspace_id
)
.fetch_one(db)
.await
.ok()
.and_then(|raw_flow| serde_json::from_str::<FlowValue>(raw_flow.get()).ok())
.ok_or(Error::InternalErr(format!(
"Unable to parse raw definition for job {} in workspace {}",
completed_flow_id, workspace_id,
)))?
};
}.ok_or(Error::InternalErr(format!(
"Unable to parse raw definition for job {} in workspace {}",
completed_flow_id, workspace_id,
)))?;
let flow_status = completed_job
.parse_flow_status()
.ok_or(Error::InternalErr(format!(
backend/windmill-worker/src/worker.rs
+18 -9
View File
@@ -54,8 +54,8 @@ use windmill_common::{
};
use windmill_queue::{
append_logs, canceled_job_to_result, empty_result, pull, push, CanceledBy, PushArgs,
PushIsolationLevel, HTTP_CLIENT,
append_logs, canceled_job_to_result, empty_result, pull, push, CanceledBy, PulledJob,
PushArgs, PushIsolationLevel, HTTP_CLIENT,
};
#[cfg(feature = "prometheus")]
@@ -1190,7 +1190,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
"received {} from same worker channel",
same_worker_job.job_id
);
let r = sqlx::query_as::<_, QueuedJob>(
let r = sqlx::query_as::<_, PulledJob>(
"UPDATE queue SET last_ping = now() WHERE id = $1 RETURNING *",
)
.bind(same_worker_job.job_id)
@@ -1327,7 +1327,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
};
if let Some(key) = key_o {
if let Some(dedicated_worker_tx) = dedicated_workers.get(&key) {
if let Err(e) = dedicated_worker_tx.send(Arc::new(job)).await {
if let Err(e) = dedicated_worker_tx.send(Arc::new(job.job)).await {
tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}");
}
@@ -1346,7 +1346,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
add_time!(bench, "send job completed START");
job_completed_tx
.send(JobCompleted {
job: Arc::new(job),
job: Arc::new(job.job),
success: true,
result: Arc::new(empty_result()),
mem_peak: 0,
@@ -1464,10 +1464,14 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let tag = job.tag.clone();
let is_init_script: bool = job.tag.as_str() == INIT_SCRIPT_TAG;
let PulledJob { job, raw_code, raw_lock, raw_flow } = job;
let arc_job = Arc::new(job);
add_time!(bench, "handle_queued_job START");
match handle_queued_job(
arc_job.clone(),
raw_code,
raw_lock,
raw_flow,
db,
&authed_client,
&hostname,
@@ -1754,6 +1758,9 @@ pub struct PreviousResult<'a> {
#[tracing::instrument(name = "job", level = "info", skip_all, fields(job_id = %job.id))]
async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: Arc<QueuedJob>,
raw_code: Option<String>,
raw_lock: Option<String>,
raw_flow: Option<Json<Box<RawValue>>>,
db: &DB,
client: &AuthedClientBackgroundTask,
hostname: &str,
@@ -1829,16 +1836,18 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
None
};
let (raw_code, raw_lock, raw_flow) =
sqlx::query!(
let (raw_code, raw_lock, raw_flow) = match (raw_code, raw_lock, raw_flow) {
(None, None, None) => sqlx::query!(
"SELECT raw_code, raw_lock, raw_flow AS \"raw_flow: Json<Box<RawValue>>\"
FROM queue_view WHERE id = $1 AND workspace_id = $2 LIMIT 1",
FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1",
&job.id, job.workspace_id
)
.fetch_one(db)
.await
.map(|record| (record.raw_code, record.raw_lock, record.raw_flow))
.unwrap_or_default();
.unwrap_or_default(),
(raw_code, raw_lock, raw_flow) => (raw_code, raw_lock, raw_flow),
};
let cached_res_path = if job.cache_ttl.is_some() {
let version_hash = if let Some(h) = job.script_hash {
backend/windmill-worker/src/worker_flow.rs
+34 -27
View File
@@ -48,17 +48,17 @@ use windmill_common::{
Approval, BranchAllStatus, BranchChosen, FlowStatus, FlowStatusModule, RetryStatus,
MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
},
flows::{has_failure_module, FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
};
use windmill_queue::schedule::get_schedule_opt;
use windmill_queue::{
add_completed_job, add_completed_job_error, append_logs, handle_maybe_scheduled_job,
CanceledBy, PushArgs, PushIsolationLevel, WrappedError,
CanceledBy, PulledJob, PushArgs, PushIsolationLevel, WrappedError,
};
type DB = sqlx::Pool<sqlx::Postgres>;
use windmill_queue::{canceled_job_to_result, get_queued_job_tx, push, QueueTransaction};
use windmill_queue::{canceled_job_to_result, push, QueueTransaction};
// #[instrument(level = "trace", skip_all)]
pub async fn update_flow_status_after_job_completion<
@@ -214,8 +214,8 @@ pub async fn update_flow_status_after_job_completion_internal<
let (old_status, current_module) = sqlx::query!(
"SELECT
flow_status AS \"flow_status!: Json<Box<RawValue>>\",
raw_flow->'modules'->(flow_status->'step')::int AS \"module: Json<Box<RawValue>>\"
FROM queue_view WHERE id = $1 AND workspace_id = $2 LIMIT 1",
coalesce(job.raw_flow, queue.raw_flow)->'modules'->(flow_status->'step')::int AS \"module: Json<Box<RawValue>>\"
FROM queue LEFT JOIN job USING(id, workspace_id) WHERE id = $1 AND workspace_id = $2 LIMIT 1",
flow, w_id
)
.fetch_one(db)
@@ -290,17 +290,15 @@ pub async fn update_flow_status_after_job_completion_internal<
Step::Step(i) => Some(i),
};
let is_flow = if let Some(step) = step {
sqlx::query_scalar!(
"SELECT raw_flow->'modules'->($1)::text->'value'->>'type' = 'flow' FROM queue_view WHERE id = $2 LIMIT 1",
step as i32, flow
)
.fetch_one(db)
.await
.map_err(|e| {
Error::InternalErr(format!("error during retrieval of step's type: {e:#}"))
})?
.unwrap_or(false)
let is_flow = if let Some(_) = step {
#[derive(Deserialize)]
struct GetType<'j> { r#type: &'j str }
current_module
.as_ref()
.map(|module| serde_json::from_str::<GetType>(module.value.get()).map(|v| v.r#type == "flow"))
.unwrap_or(Ok(false))
.unwrap_or(false)
} else {
false
};
@@ -915,8 +913,12 @@ pub async fn update_flow_status_after_job_completion_internal<
.context("remove flow status retry")?;
}
let flow_job = get_queued_job_tx(flow, w_id, tx.transaction_mut())
.await?
let flow_job = sqlx::query_as::<_, PulledJob>("SELECT * FROM queue WHERE id = $1 AND workspace_id = $2")
.bind(flow)
.bind(w_id)
.fetch_optional(&mut tx)
.await
.map_err(Into::<Error>::into)?
.ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?;
tx.commit().await?;
@@ -926,14 +928,19 @@ pub async fn update_flow_status_after_job_completion_internal<
.unwrap_or_else(|| "none".to_string());
tracing::info!(id = %flow_job.id, root_id = %job_root, "update flow status");
let raw_flow = sqlx::query_scalar!(
"SELECT raw_flow AS \"raw_flow!: Json<Box<sqlx::types::JsonRawValue>>\"
FROM queue_view WHERE id = $1 AND workspace_id = $2 LIMIT 1",
&flow_job.id, w_id
)
.fetch_one(db)
.await
.ok();
let PulledJob { job: flow_job, raw_flow, .. } = flow_job;
let raw_flow = if let Some(raw_flow) = raw_flow {
Some(raw_flow)
} else {
sqlx::query_scalar!(
"SELECT raw_flow AS \"raw_flow!: Json<Box<sqlx::types::JsonRawValue>>\"
FROM job WHERE id = $1 AND workspace_id = $2 LIMIT 1",
&flow_job.id, w_id
)
.fetch_one(db)
.await
.ok()
};
let flow_value = raw_flow
.as_ref()
.and_then(|raw_flow| serde_json::from_str::<FlowValue>(raw_flow.get()).ok());
@@ -965,7 +972,7 @@ pub async fn update_flow_status_after_job_completion_internal<
false
if !is_failure_step
&& !skip_error_handler
&& has_failure_module(flow, db, false).await? =>
&& flow_value.as_ref().map(|v| v.failure_module.is_some()).unwrap_or(false) =>
{
true
}