diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index c44b8c18e6..6186b0ada1 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -173,17 +173,6 @@ pub async fn migrate(db: &DB) -> Result<(), Error> { let migrator = db.acquire().await?; let mut custom_migrator = CustomMigrator { inner: migrator }; - if let Err(err) = fix_flow_versioning_migration(&mut custom_migrator, db).await { - tracing::error!("Could not apply flow versioning fix migration: {err:#}"); - } - - let db2 = db.clone(); - let _ = tokio::task::spawn(async move { - if let Err(err) = fix_job_completed_index(&db2).await { - tracing::error!("Could not apply job completed index fix migration: {err:#}"); - } - }); - match sqlx::migrate!("../migrations") .run_direct(&mut custom_migrator) .await @@ -199,6 +188,17 @@ pub async fn migrate(db: &DB) -> Result<(), Error> { Err(err) => Err(err), }?; + if let Err(err) = fix_flow_versioning_migration(&mut custom_migrator, db).await { + tracing::error!("Could not apply flow versioning fix migration: {err:#}"); + } + + let db2 = db.clone(); + let _ = tokio::task::spawn(async move { + if let Err(err) = fix_job_completed_index(&db2).await { + tracing::error!("Could not apply job completed index fix migration: {err:#}"); + } + }); + Ok(()) } @@ -302,7 +302,7 @@ macro_rules! run_windmill_migration { .await?; tracing::info!("Finished applying {migration_job_name} migration"); } else { - tracing::info!("migration {migration_job_name} already done"); + tracing::debug!("migration {migration_job_name} already done"); } let _ = sqlx::query("SELECT pg_advisory_unlock(4242)") @@ -311,7 +311,7 @@ macro_rules! run_windmill_migration { tx.commit().await?; tracing::info!("released lock for {migration_job_name}"); } else { - tracing::info!("migration {migration_job_name} already done"); + tracing::debug!("migration {migration_job_name} already done"); } } diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 958e9f13ef..aa9efea9b2 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -101,6 +101,7 @@ pub struct FlowValue { #[serde(skip_serializing_if = "Option::is_none")] #[serde(default)] pub failure_module: Option>, + #[serde(skip_serializing_if = "Option::is_none")] #[serde(default)] pub preprocessor_module: Option>, #[serde(default)] @@ -423,6 +424,7 @@ pub enum FlowModuleValue { path: String, #[serde(skip_serializing_if = "Option::is_none")] hash: Option, + #[serde(deserialize_with = "empty_string_as_none")] tag_override: Option, }, Flow { @@ -480,6 +482,14 @@ fn is_none_or_empty(expr: &Option) -> bool { expr.is_none() || expr.as_ref().unwrap().is_empty() } +fn empty_string_as_none<'de, D>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + let s: Option = Option::deserialize(deserializer)?; + Ok(s.filter(|s| !s.is_empty())) +} + #[derive(Deserialize)] struct UntaggedFlowModuleValue { #[serde(rename = "type")] @@ -488,6 +498,7 @@ struct UntaggedFlowModuleValue { input_transforms: Option>, path: Option, hash: Option, + #[serde(deserialize_with = "empty_string_as_none")] tag_override: Option, iterator: Option, modules: Option>, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 3f92cfc829..a042d41ce4 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -2403,15 +2403,15 @@ async fn push_next_flow_job } tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushing job {i} of {len}"); let payload_tag = match &job_payloads { - ContinuePayload::SingleJob(payload) => payload.clone(), - ContinuePayload::BranchAllJobs(payloads) => payloads[i].clone(), + ContinuePayload::SingleJob(payload) => payload, + ContinuePayload::BranchAllJobs(payloads) => &payloads[i], ContinuePayload::ForloopJobs { flow_value, delete_after_use, .. } => { let mut fv = flow_value.clone(); if let Some(failure_module) = fv.failure_module.as_mut() { failure_module.id_append(&format!("{}-{i}", &status.step.to_string())); } - JobPayloadWithTag { + &JobPayloadWithTag { payload: JobPayload::RawFlow { value: fv, path: Some(format!("{}/forloop-{i}", flow_job.script_path())), @@ -2590,13 +2590,17 @@ async fn push_next_flow_job }; tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}"); - + let tag = if flow_job.tag == "flow" || flow_job.tag == format!("flow-{}", flow_job.workspace_id) { + payload_tag.tag.clone() + } else { + Some(flow_job.tag.clone()) + }; let tx2 = PushIsolationLevel::Transaction(tx); let (uuid, mut inner_tx) = push( &db, tx2, &flow_job.workspace_id, - payload_tag.payload, + payload_tag.payload.clone(), push_args, &flow_job.created_by, &flow_job.email, @@ -2610,11 +2614,7 @@ async fn push_next_flow_job continue_on_same_worker, err, flow_job.visible_to_owner, - if flow_job.tag == "flow" || flow_job.tag == format!("flow-{}", flow_job.workspace_id) { - payload_tag.tag - } else { - Some(flow_job.tag.clone()) - }, + tag, payload_tag.timeout, Some(module.id.clone()), new_job_priority_override, @@ -3761,6 +3761,7 @@ async fn script_to_payload( module: &FlowModule, tag_override: &Option, ) -> Result { + tracing::error!("{tag_override:?}"); let (payload, tag, delete_after_use, script_timeout) = if script_hash.is_none() { let (jp, tag, delete_after_use, script_timeout) = script_path_to_payload(script_path, db, &flow_job.workspace_id, Some(true)).await?; diff --git a/frontend/src/lib/components/flows/content/FlowModuleWorkerTagSelect.svelte b/frontend/src/lib/components/flows/content/FlowModuleWorkerTagSelect.svelte index 1e26f35e82..00c528902e 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleWorkerTagSelect.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleWorkerTagSelect.svelte @@ -33,7 +33,7 @@ {#if tag} {:else} - + {/if} {#each $workerTags ?? [] as tag (tag)}