fix: correctly run empty flow with preprocessor from UI (#5537)

backend/windmill-queue/src/jobs.rs

@@ -3590,18 +3590,16 @@ pub async fn push<'c, 'd>(
             }?;
             tx = PushIsolationLevel::Transaction(ntx);
 
-            let value = data.value().clone();
+            let mut value = data.value().clone();
             let priority = value.priority;
             let cache_ttl = value.cache_ttl.map(|x| x as i32);
             let custom_concurrency_key = value.concurrency_key.clone();
             let concurrency_time_window_s = value.concurrency_time_window_s;
             let concurrent_limit = value.concurrent_limit;
 
-            // this is a new flow being pushed, status is set to `value`.
-            let mut status = FlowStatus::new(&value);
             let extra = args.extra.get_or_insert_with(HashMap::new);
             if !apply_preprocessor {
-                status.preprocessor_module = None;
+                value.preprocessor_module = None;
                 extra.remove("wm_trigger");
             } else {
                 preprocessed = Some(false);
@@ -3611,6 +3609,10 @@ pub async fn push<'c, 'd>(
                     }))
                 });
             }
+
+            // this is a new flow being pushed, status is set to `value`.
+            let status = FlowStatus::new(&value);
+
             // Keep inserting `value` if not all workers are updated.
             // Starting at `v1.440`, the value is fetched on pull from the version id.
             let value_o = if !*MIN_VERSION_IS_AT_LEAST_1_440.read().await {
@@ -3619,9 +3621,6 @@ pub async fn push<'c, 'd>(
                 if same_worker {
                     value.same_worker = true;
                 }
-                if !apply_preprocessor {
-                    value.preprocessor_module = None;
-                }
                 Some(value)
             } else {
                 // `raw_flow` is fetched on pull, the mutations from the other branch are replaced