From 715cb08fcdec6b87920173476d110ea7d9cca629 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 10 Dec 2023 13:29:51 +0100 Subject: [PATCH] fix: make cleanup_module backcompatible --- backend/windmill-common/src/flow_status.rs | 16 ++----- backend/windmill-queue/src/jobs.rs | 14 +----- backend/windmill-worker/src/worker_flow.rs | 54 +++++++++++++--------- python-client/wmill/pyproject.toml | 2 +- 4 files changed, 37 insertions(+), 49 deletions(-) diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index ea293c7c06..a64ec16966 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/backend/windmill-common/src/flow_status.rs @@ -29,6 +29,7 @@ pub struct FlowStatus { pub step: i32, pub modules: Vec, pub failure_module: FlowStatusModuleWParent, + #[serde(default)] pub cleanup_module: FlowCleanupModule, #[serde(default)] #[serde(skip_serializing_if = "is_retry_default")] @@ -97,13 +98,11 @@ pub struct FlowStatusModuleWParent { pub module_status: FlowStatusModule, } -#[derive(Serialize, Deserialize, Debug, Clone)] +#[derive(Serialize, Deserialize, Debug, Clone, Default)] pub struct FlowCleanupModule { #[serde(default)] #[serde(skip_serializing_if = "Vec::is_empty")] pub flow_jobs_to_clean: Vec, - #[serde(flatten)] - pub module_status: FlowStatusModule, } #[derive(Serialize, Deserialize, Debug, Clone)] @@ -228,16 +227,7 @@ impl FlowStatus { .unwrap_or_else(|| "failure".to_string()), }, }, - cleanup_module: FlowCleanupModule { - flow_jobs_to_clean: vec![], - module_status: FlowStatusModule::WaitingForPriorSteps { - id: f - .failure_module - .as_ref() - .map(|x| x.id.clone()) - .unwrap_or_else(|| "cleanup".to_string()), - }, - }, + cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] }, retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, restarted_from: None, } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index d1a94cbe00..64b92d443c 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -2449,12 +2449,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection id: "failure".to_string(), }, }, - cleanup_module: FlowCleanupModule { - module_status: FlowStatusModule::WaitingForPriorSteps { - id: "cleanup".to_string(), - }, - flow_jobs_to_clean: vec![], - }, + cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] }, // retry status is reset retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, // TODO: for now, flows with approval conditions aren't supported for restart @@ -2540,12 +2535,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection id: "failure".to_string(), }, }, - cleanup_module: FlowCleanupModule { - module_status: FlowStatusModule::WaitingForPriorSteps { - id: "cleanup".to_string(), - }, - flow_jobs_to_clean: vec![], - }, + cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] }, // retry status is reset retry: RetryStatus { fail_count: 0, failed_jobs: vec![] }, // TODO: for now, flows with approval conditions aren't supported for restart diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index f5c340afdb..b9bcd3bffd 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -27,7 +27,7 @@ use tokio::sync::mpsc::Sender; use tracing::instrument; use uuid::Uuid; use windmill_common::flow_status::{ - ApprovalConditions, FlowCleanupModule, FlowStatusModuleWParent, Iterator, JobResult, + ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult, }; use windmill_common::flows::add_virtual_items_if_necessary; use windmill_common::jobs::{ @@ -168,7 +168,15 @@ pub async fn update_flow_status_after_job_completion_internal< rsmq: Option, worker_name: &str, ) -> error::Result> { - let (should_continue_flow, flow_job, stop_early, skip_if_stop_early, nresult, is_failure_step) = { + let ( + should_continue_flow, + flow_job, + stop_early, + skip_if_stop_early, + nresult, + is_failure_step, + cleanup_module, + ) = { // tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}"); let old_status_json = sqlx::query_scalar!( @@ -574,6 +582,7 @@ pub async fn update_flow_status_after_job_completion_internal< skip_if_stop_early, nresult, is_failure_step, + old_status.cleanup_module, ) }; @@ -591,8 +600,7 @@ pub async fn update_flow_status_after_job_completion_internal< #[cfg(feature = "enterprise")] if flow_job.parent_job.is_none() { // run the cleanup step only when the root job is complete - let cleanup_module = retrieve_cleanup_module(flow, db).await?; - if cleanup_module.flow_jobs_to_clean.len() > 0 { + if !cleanup_module.flow_jobs_to_clean.is_empty() { tracing::debug!( "Cleaning up jobs arguments, result and logs as they were marked as delete_after_use {:?}", cleanup_module.flow_jobs_to_clean @@ -837,26 +845,26 @@ async fn has_failure_module<'c>( .map(|v| v.unwrap_or(false)) } -async fn retrieve_cleanup_module<'c>(flow_uuid: Uuid, db: &DB) -> Result { - tracing::warn!("Retrieving cleanup module of flow {}", flow_uuid); - let raw_value = sqlx::query_scalar!( - "SELECT flow_status->'cleanup_module' as cleanup_module - FROM queue - WHERE id = $1", - flow_uuid, - ) - .fetch_one(db) - .await - .map_err(|e| Error::InternalErr(format!("error during retrieval of cleanup module: {e}")))?; +// async fn retrieve_cleanup_module<'c>(flow_uuid: Uuid, db: &DB) -> Result { +// tracing::warn!("Retrieving cleanup module of flow {}", flow_uuid); +// let raw_value = sqlx::query_scalar!( +// "SELECT flow_status->'cleanup_module' as cleanup_module +// FROM queue +// WHERE id = $1", +// flow_uuid, +// ) +// .fetch_one(db) +// .await +// .map_err(|e| Error::InternalErr(format!("error during retrieval of cleanup module: {e}")))?; - raw_value - .clone() - .and_then(|rv| serde_json::from_value::(rv).ok()) - .ok_or(Error::InternalErr(format!( - "Unable to parse flow cleanup module {:?}", - raw_value - ))) -} +// raw_value +// .clone() +// .and_then(|rv| serde_json::from_value::(rv).ok()) +// .ok_or(Error::InternalErr(format!( +// "Unable to parse flow cleanup module {:?}", +// raw_value +// ))) +// } fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u16, Duration)> { (status.fail_count <= MAX_RETRY_ATTEMPTS) diff --git a/python-client/wmill/pyproject.toml b/python-client/wmill/pyproject.toml index 164760502f..c8754d98be 100644 --- a/python-client/wmill/pyproject.toml +++ b/python-client/wmill/pyproject.toml @@ -16,7 +16,7 @@ include = ["wmill/py.typed"] [tool.poetry.dependencies] python = "^3.7" -httpx = "^0.25" +httpx = "^0.24" [build-system] requires = ["poetry>=1.0.2", "poetry-dynamic-versioning"]