fix(backend): in flows, workspace scripts should use their set tags instead of the default one

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");
 
             }
         }

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<Box<FlowModule>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(default)]
     pub preprocessor_module: Option<Box<FlowModule>>,
     #[serde(default)]
@@ -423,6 +424,7 @@ pub enum FlowModuleValue {
         path: String,
         #[serde(skip_serializing_if = "Option::is_none")]
         hash: Option<ScriptHash>,
+        #[serde(deserialize_with = "empty_string_as_none")]
         tag_override: Option<String>,
     },
     Flow {
@@ -480,6 +482,14 @@ fn is_none_or_empty(expr: &Option<String>) -> bool {
     expr.is_none() || expr.as_ref().unwrap().is_empty()
 }
 
+fn empty_string_as_none<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let s: Option<String> = 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<HashMap<String, InputTransform>>,
     path: Option<String>,
     hash: Option<ScriptHash>,
+    #[serde(deserialize_with = "empty_string_as_none")]
     tag_override: Option<String>,
     iterator: Option<InputTransform>,
     modules: Option<Vec<FlowModule>>,

backend/windmill-worker/src/worker_flow.rs

@@ -2403,15 +2403,15 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         }
         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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
         };
 
         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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             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<String>,
 ) -> Result<JobPayloadWithTag, Error> {
+    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?;

frontend/src/lib/components/flows/content/FlowModuleWorkerTagSelect.svelte

@@ -33,7 +33,7 @@
 					{#if tag}
 						<option value="">reset to default</option>
 					{:else}
-						<option value="" disabled selected>Tag</option>
+						<option value={undefined} disabled selected>Tag</option>
 					{/if}
 					{#each $workerTags ?? [] as tag (tag)}
 						<option value={tag}>{tag}</option>