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

handle more error edge cases

Browse Source
This commit is contained in:
Ruben Fiszel
2022-10-29 13:59:50 +02:00
parent 004f4958ef
commit f75823f474
4 changed files with 78 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/worker.rs
+60 -43
View File
@@ -483,11 +483,18 @@ async fn handle_job_error(
.map(|(_, m)| m)
.unwrap_or_else(|_| Map::new());
if let Some(parent_job_id) = job.parent_job {
if job.is_flow_step || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::Flow {
let (flow, job_status_to_update) = if let Some(parent_job_id) = job.parent_job {
(parent_job_id, job.id)
} else {
(job.id, Uuid::nil())
};
let updated_flow = update_flow_status_after_job_completion(
db,
client,
&job,
flow,
&job_status_to_update,
&job.workspace_id,
false,
serde_json::Value::Object(m),
metrics.clone(),
@@ -499,19 +506,21 @@ async fn handle_job_error(
)
.await;
if let Err(err) = updated_flow {
if let Ok(mut tx) = db.begin().await {
if let Ok(Some(parent_job)) =
get_queued_job(parent_job_id, &job.workspace_id, &mut tx).await
{
let _ = add_completed_job_error(
db,
client,
&parent_job,
format!("Unexpected error during flow job error handling:\n{err}"),
err,
metrics,
)
.await;
if let Some(parent_job_id) = job.parent_job {
if let Ok(mut tx) = db.begin().await {
if let Ok(Some(parent_job)) =
get_queued_job(parent_job_id, &job.workspace_id, &mut tx).await
{
let _ = add_completed_job_error(
db,
client,
&parent_job,
format!("Unexpected error during flow job error handling:\n{err}"),
err,
metrics,
)
.await;
}
}
}
}
@@ -632,20 +641,24 @@ async fn handle_queued_job(
Ok(r) => {
add_completed_job(db, client, &job, true, false, r.clone(), logs).await?;
if job.is_flow_step {
update_flow_status_after_job_completion(
db,
client,
&job,
true,
r,
Some(metrics.clone()),
false,
same_worker_tx.clone(),
worker_dir,
worker_config.keep_job_dir,
&worker_config.base_internal_url,
)
.await?;
if let Some(parent_job) = job.parent_job {
update_flow_status_after_job_completion(
db,
client,
parent_job,
&job.id,
&job.workspace_id,
true,
r,
Some(metrics.clone()),
false,
same_worker_tx.clone(),
worker_dir,
worker_config.keep_job_dir,
&worker_config.base_internal_url,
)
.await?;
}
}
}
Err(e) => {
@@ -677,20 +690,24 @@ async fn handle_queued_job(
)
.await?;
if job.is_flow_step {
update_flow_status_after_job_completion(
db,
client,
&job,
false,
serde_json::Value::Object(output_map),
Some(metrics),
false,
same_worker_tx,
worker_dir,
worker_config.keep_job_dir,
&worker_config.base_internal_url,
)
.await?;
if let Some(parent_job) = job.parent_job {
update_flow_status_after_job_completion(
db,
client,
parent_job,
&job.id,
&job.workspace_id,
false,
serde_json::Value::Object(output_map),
Some(metrics),
false,
same_worker_tx,
worker_dir,
worker_config.keep_job_dir,
&worker_config.base_internal_url,
)
.await?;
}
}
}
};
backend/windmill-worker/src/worker_flow.rs
+13 -14
View File
@@ -39,7 +39,9 @@ use windmill_queue::{
pub async fn update_flow_status_after_job_completion(
db: &DB,
client: &windmill_api_client::Client,
job: &QueuedJob,
flow: uuid::Uuid,
job_id_for_status: &Uuid,
w_id: &str,
success: bool,
result: serde_json::Value,
metrics: Option<worker::Metrics>,
@@ -49,13 +51,7 @@ pub async fn update_flow_status_after_job_completion(
keep_job_dir: bool,
base_internal_url: &str,
) -> error::Result<()> {
tracing::debug!("UPDATE FLOW STATUS: {job:?} {success} {result:?}");
let w_id = &job.workspace_id;
let flow = job
.parent_job
.ok_or_else(|| Error::InternalErr(format!("expected parent job")))?;
tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id}");
let mut tx = db.begin().await?;
@@ -131,7 +127,7 @@ pub async fn update_flow_status_after_job_completion(
old_status.step + 1,
FlowStatusModule::Success {
id: module_status.id(),
job: job.id,
job: job_id_for_status.clone(),
flow_jobs,
branch_chosen,
approvers: vec![],
@@ -142,7 +138,7 @@ pub async fn update_flow_status_after_job_completion(
old_status.step,
FlowStatusModule::Failure {
id: module_status.id(),
job: job.id,
job: job_id_for_status.clone(),
flow_jobs,
branch_chosen,
},
@@ -332,11 +328,13 @@ pub async fn update_flow_status_after_job_completion(
let _ = tokio::fs::remove_dir_all(format!("{worker_dir}/{}", flow_job.id)).await;
}
if flow_job.parent_job.is_some() {
if let Some(parent_job) = flow_job.parent_job {
return Ok(update_flow_status_after_job_completion(
db,
client,
&flow_job,
parent_job,
&flow,
w_id,
success,
result,
metrics,
@@ -577,11 +575,12 @@ pub async fn handle_flow(
let flow = serde_json::from_value::<FlowValue>(value)?;
if flow.modules.is_empty() {
let fake_job = QueuedJob { parent_job: Some(flow_job.id), ..flow_job.clone() };
update_flow_status_after_job_completion(
db,
client,
&fake_job,
flow_job.id,
&Uuid::nil(),
flow_job.workspace_id.as_str(),
true,
serde_json::json!({}),
None,
frontend/src/lib/components/ScriptBuilder.svelte
+2 -2
View File
@@ -86,7 +86,7 @@
<div class="flex flex-col h-screen">
<!-- Nav between steps-->
<div class="flex flex-col w-full px-4 py-2 border-b shadow-sm">
<div class="justify-between flex flex-row w-full items-center">
<div class="justify-between flex flex-row w-full items-center overflow-x-auto scrollbar-hidden">
<div class="flex flex-row">
<Breadcrumb
items={['Metadata', 'Code', 'UI Customisation']}
@@ -98,7 +98,7 @@
</Breadcrumb>
</div>
<div class="flex gap-1 flex-row">
<div class="gap-1 flex-row hidden md:flex">
<Button
startIcon={{ icon: faPen }}
variant="contained"
frontend/src/lib/components/ScriptEditor.svelte
+3 -3
View File
@@ -148,7 +148,7 @@
</Pane>
<Pane size={40} minSize={10}>
<Splitpanes horizontal>
<Pane size={30}>
<Pane size={33}>
<div class="w-full bg-gray-100 px-2 text-sm"
>Preview <Tooltip>
To recompute the input schema press <Kbd>Ctrl/Cmd</Kbd> + <Kbd>S</Kbd> or move the focus
@@ -161,7 +161,7 @@
</div>
</div>
</Pane>
<Pane size={70}>
<Pane size={67}>
<div class="px-2 py-1">
{#if testIsLoading}
<Button
@@ -174,7 +174,7 @@
classes: 'animate-spin'
}}
>
'Cancel'
Cancel
</Button>
{:else}
<Button