refactor: move concurrency and debouncing settings to their own structs

backend/tests/common/mod.rs

@@ -684,16 +684,13 @@ pub async fn run_deployed_relative_imports(
             let job = RunJob::from(JobPayload::ScriptHash {
                 path: "f/system/test_import".to_string(),
                 hash: ScriptHash(script.hash),
-                custom_concurrency_key: None,
-                concurrent_limit: None,
-                concurrency_time_window_s: None,
                 cache_ttl: None,
                 dedicated_worker: None,
                 language,
                 priority: None,
                 apply_preprocessor: false,
-                custom_debounce_key: None,
-                debounce_delay_s: None,
+                concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
+                debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
             })
             .push(&db2)
             .await;
@@ -739,13 +736,10 @@ pub async fn run_preview_relative_imports(
                 path: Some("f/system/test_import".to_string()),
                 language,
                 lock: None,
-                custom_concurrency_key: None,
-                concurrent_limit: None,
-                concurrency_time_window_s: None,
                 cache_ttl: None,
                 dedicated_worker: None,
-                custom_debounce_key: None,
-                debounce_delay_s: None,
+                concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
+                debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
             }))
             .push(&db2)
             .await;

backend/tests/job_payload.rs

@@ -86,16 +86,13 @@ mod job_payload {
             let job = RunJob::from(JobPayload::ScriptHash {
                 hash: ScriptHash(123413),
                 path: "f/system/hello_with_preprocessor".to_string(),
-                custom_concurrency_key: None,
-                concurrent_limit: None,
-                concurrency_time_window_s: None,
                 cache_ttl: None,
                 dedicated_worker: None,
                 language: ScriptLang::Deno,
                 priority: None,
                 apply_preprocessor: true,
-                custom_debounce_key: None,
-                debounce_delay_s: None,
+                concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
+                debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
             })
             .run_until_complete_with(db, false, port, |id| async move {
                 let job = sqlx::query!("SELECT preprocessed FROM v2_job WHERE id = $1", id)
@@ -167,9 +164,7 @@ mod job_payload {
             let result = RunJob::from(JobPayload::FlowScript {
                 id: flow_scripts[0],
                 language: ScriptLang::Deno,
-                custom_concurrency_key: None,
-                concurrent_limit: None,
-                concurrency_time_window_s: None,
+                concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
                 cache_ttl: None,
                 dedicated_worker: None,
                 path: "f/system/hello/test-0".into(),
@@ -187,9 +182,7 @@ mod job_payload {
             let result = RunJob::from(JobPayload::FlowScript {
                 id: flow_scripts[1],
                 language: ScriptLang::Deno,
-                custom_concurrency_key: None,
-                concurrent_limit: None,
-                concurrency_time_window_s: None,
+                concurrency_settings: windmill_common::jobs::ConcurrencySettings::default(),
                 cache_ttl: None,
                 dedicated_worker: None,
                 path: "f/system/hello/test-0".into(),

backend/tests/python_jobs.rs

@@ -25,7 +25,10 @@ def main():
         &db,
         content,
         ScriptLang::Python3,
-        vec!["# workspace-dependencies-mode: manual\n# py: 3.11.11","tiny==0.1.3"],
+        vec![
+            "# workspace-dependencies-mode: manual\n# py: 3.11.11",
+            "tiny==0.1.3",
+        ],
     )
     .await?;
     Ok(())
@@ -186,11 +189,8 @@ def main():
         path: None,
         language: ScriptLang::Python3,
         lock: None,
-        custom_concurrency_key: None,
-        concurrent_limit: None,
-        concurrency_time_window_s: None,
-        custom_debounce_key: None,
-        debounce_delay_s: None,
+        concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
+        debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
         cache_ttl: None,
         dedicated_worker: None,
     });
@@ -237,11 +237,8 @@ def main():
             path: None,
             language: ScriptLang::Python3,
             lock: None,
-            custom_concurrency_key: None,
-            concurrent_limit: None,
-            concurrency_time_window_s: None,
-            custom_debounce_key: None,
-            debounce_delay_s: None,
+            concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
+            debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
             cache_ttl: None,
             dedicated_worker: None,
         });
@@ -273,11 +270,8 @@ def main():
             path: None,
             language: ScriptLang::Python3,
             lock: None,
-            custom_concurrency_key: None,
-            concurrent_limit: None,
-            concurrency_time_window_s: None,
-            custom_debounce_key: None,
-            debounce_delay_s: None,
+            concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
+            debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
             cache_ttl: None,
             dedicated_worker: None,
         });
@@ -314,11 +308,8 @@ def main():
         path: None,
         language: ScriptLang::Python3,
         lock: None,
-        custom_concurrency_key: None,
-        concurrent_limit: None,
-        concurrency_time_window_s: None,
-        custom_debounce_key: None,
-        debounce_delay_s: None,
+        concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
+        debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
         cache_ttl: None,
         dedicated_worker: None,
     });
@@ -353,11 +344,8 @@ def main():
         path: None,
         language: ScriptLang::Python3,
         lock: None,
-        custom_concurrency_key: None,
-        concurrent_limit: None,
-        concurrency_time_window_s: None,
-        custom_debounce_key: None,
-        debounce_delay_s: None,
+        concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(),
+        debouncing_settings: windmill_common::jobs::DebouncingSettings::default(),
         cache_ttl: None,
         dedicated_worker: None,
     });

backend/windmill-api/src/flows.rs

@@ -715,13 +715,12 @@ async fn get_flow_version_by_id(
     let mut tx = user_db.begin(&authed).await?;
 
     // First, fetch the path to perform authorization check early
-    let path: Option<String> = sqlx::query_scalar(
-        "SELECT path FROM flow_version WHERE id = $1 AND workspace_id = $2",
-    )
-    .bind(version)
-    .bind(&w_id)
-    .fetch_optional(&mut *tx)
-    .await?;
+    let path: Option<String> =
+        sqlx::query_scalar("SELECT path FROM flow_version WHERE id = $1 AND workspace_id = $2")
+            .bind(version)
+            .bind(&w_id)
+            .fetch_optional(&mut *tx)
+            .await?;
 
     let path = not_found_if_none(
         path,
@@ -788,13 +787,12 @@ async fn update_flow_history(
     let mut tx = user_db.begin(&authed).await?;
 
     // Fetch path and perform authorization check early
-    let path: Option<String> = sqlx::query_scalar(
-        "SELECT path FROM flow_version WHERE workspace_id = $1 AND id = $2",
-    )
-    .bind(&w_id)
-    .bind(version)
-    .fetch_optional(&mut *tx)
-    .await?;
+    let path: Option<String> =
+        sqlx::query_scalar("SELECT path FROM flow_version WHERE workspace_id = $1 AND id = $2")
+            .bind(&w_id)
+            .bind(version)
+            .fetch_optional(&mut *tx)
+            .await?;
 
     let path = not_found_if_none(
         path,
@@ -1440,10 +1438,9 @@ async fn archive_flow_by_path(
 /// Validates that flow debouncing configuration is supported by all workers
 /// Returns an error if debouncing is configured but workers are behind required version
 async fn guard_flow_from_debounce_data(nf: &NewFlow) -> Result<()> {
-    if !*MIN_VERSION_SUPPORTS_DEBOUNCING.read().await && {
-        let flow_value = nf.parse_flow_value()?;
-        flow_value.debounce_key.is_some() || flow_value.debounce_delay_s.is_some()
-    } {
+    if !*MIN_VERSION_SUPPORTS_DEBOUNCING.read().await
+        && !nf.parse_flow_value()?.debouncing_settings.is_default()
+    {
         tracing::warn!(
             "Flow debouncing configuration rejected: workers are behind minimum required version for debouncing feature"
         );
@@ -1557,11 +1554,13 @@ mod tests {
 
     use std::{collections::HashMap, time::Duration};
 
+    use aws_sdk_config::operation::stop_configuration_recorder::builders::StopConfigurationRecorderOutputBuilder;
     use windmill_common::{
         flows::{
             ConstantDelay, ExponentialDelay, FlowModule, FlowModuleValue, FlowValue,
             InputTransform, Retry, StopAfterIf,
         },
+        jobs::{ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings},
         scripts,
     };
 
@@ -1612,11 +1611,9 @@ mod tests {
                         path: None,
                         lock: None,
                         tag: None,
-                        custom_concurrency_key: None,
-                        concurrent_limit: None,
-                        concurrency_time_window_s: None,
                         is_trigger: None,
                         assets: None,
+                        concurrency_settings: ConcurrencySettingsWithCustom::default(),
                     }),
                     stop_after_if: Some(StopAfterIf {
                         expr: "foo = 'bar'".to_string(),
@@ -1702,17 +1699,14 @@ mod tests {
             })),
             preprocessor_module: None,
             same_worker: false,
-            concurrent_limit: None,
-            concurrency_time_window_s: None,
             skip_expr: None,
             cache_ttl: None,
             priority: None,
             early_return: None,
-            concurrency_key: None,
             chat_input_enabled: None,
             flow_env: None,
-            debounce_key: None,
-            debounce_delay_s: None,
+            concurrency_settings: ConcurrencySettings::default(),
+            debouncing_settings: DebouncingSettings::default(),
         };
         let expect = serde_json::json!({
           "modules": [