From fcbfd99ba962a566f1b5132ef3862c72d86ffd3a Mon Sep 17 00:00:00 2001 From: Pyra <92104930+pyranota@users.noreply.github.com> Date: Thu, 4 Dec 2025 11:11:54 +0100 Subject: [PATCH] refactor: move concurrency and debouncing settings to their own structs (#7286) * move concurrency and debouncing settings to it's own structs Signed-off-by: pyranota * cleanup queue::jobs::pull Signed-off-by: pyranota * done Signed-off-by: pyranota * migrated the rest Signed-off-by: pyranota * implement todo! * impmenet todo! v2 * more cleanup * fix all Signed-off-by: pyranota * remove unrelated code Signed-off-by: pyranota * nits Signed-off-by: pyranota * do not serialize rawcode Signed-off-by: pyranota * cleanup Signed-off-by: pyranota * fixes Signed-off-by: pyranota * update ee repo ref Signed-off-by: pyranota * update ee repo Signed-off-by: pyranota * fix tests Signed-off-by: pyranota * Update ee-repo-ref.txt --------- Signed-off-by: pyranota Co-authored-by: windmill-internal-app[bot] --- backend/ee-repo-ref.txt | 2 +- backend/tests/common/mod.rs | 14 +- backend/tests/job_payload.rs | 22 +- backend/tests/python_jobs.rs | 40 +- backend/tests/worker.rs | 89 +- backend/windmill-api/src/flows.rs | 45 +- backend/windmill-api/src/jobs.rs | 101 +-- .../src/triggers/trigger_helpers.rs | 14 +- backend/windmill-common/src/flows.rs | 54 +- backend/windmill-common/src/jobs.rs | 190 ++++- backend/windmill-common/src/scripts.rs | 4 + backend/windmill-queue/src/jobs.rs | 791 +++++++----------- backend/windmill-queue/src/schedule.rs | 31 +- backend/windmill-worker/src/ai/tools.rs | 22 +- backend/windmill-worker/src/worker_flow.rs | 72 +- .../windmill-worker/src/worker_lockfiles.rs | 16 +- backend/windmill-worker/src/worker_utils.rs | 14 +- 17 files changed, 677 insertions(+), 844 deletions(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 9cc9af9a72..d57da779de 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -d54c99a1e4a08ff30c35737bd282d6d8d2b49e1f +937820a9e572c10c0ce050e1bd5eddb3180fe921 diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 129fd86d63..421d1dd8c3 100644 --- a/backend/tests/common/mod.rs +++ b/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; diff --git a/backend/tests/job_payload.rs b/backend/tests/job_payload.rs index 22e713329b..5eabdc83ce 100644 --- a/backend/tests/job_payload.rs +++ b/backend/tests/job_payload.rs @@ -52,11 +52,8 @@ mod job_payload { let result = RunJob::from(JobPayload::ScriptHash { hash: ScriptHash(123412), path: "f/system/hello".to_string(), - 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, language: ScriptLang::Deno, @@ -86,16 +83,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 +161,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 +179,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(), diff --git a/backend/tests/python_jobs.rs b/backend/tests/python_jobs.rs index 549e052b65..f6fc35295c 100644 --- a/backend/tests/python_jobs.rs +++ b/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, }); diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index bb18d6e397..d18869efb8 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -188,9 +188,8 @@ async fn test_deno_flow(db: Pool) -> anyhow::Result<()> { path: None, lock: None, tag: None, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, + concurrency_settings: windmill_common::jobs::ConcurrencySettings::default() + .into(), is_trigger: None, assets: None, } @@ -234,9 +233,8 @@ async fn test_deno_flow(db: Pool) -> anyhow::Result<()> { path: None, lock: None, tag: None, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, + concurrency_settings: + windmill_common::jobs::ConcurrencySettings::default().into(), is_trigger: None, assets: None, } @@ -368,9 +366,7 @@ async fn test_deno_flow_same_worker(db: Pool) -> anyhow::Result<()> { path: None, lock: None, tag: None, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, + concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(), is_trigger: None, assets: None, @@ -425,9 +421,7 @@ async fn test_deno_flow_same_worker(db: Pool) -> anyhow::Result<()> { path: None, lock: None, tag: None, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, + concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(), is_trigger: None, assets: None, }.into(), @@ -466,9 +460,7 @@ async fn test_deno_flow_same_worker(db: Pool) -> anyhow::Result<()> { path: None, lock: None, tag: None, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, + concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(), is_trigger: None, assets: None, @@ -534,9 +526,7 @@ async fn test_deno_flow_same_worker(db: Pool) -> anyhow::Result<()> { path: None, lock: None, tag: None, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, + concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(), is_trigger: None, assets: None, }.into(), @@ -864,13 +854,10 @@ func main(derp string) (string, error) { path: None, lock: None, language: ScriptLang::Go, - 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(), })) .arg("derp", json!("world")) .run_until_complete(&db, false, port) @@ -903,13 +890,10 @@ fn main(world: String) -> Result { path: None, lock: None, language: ScriptLang::Rust, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, + concurrency_settings: windmill_common::jobs::ConcurrencySettings::default().into(), + debouncing_settings: windmill_common::jobs::DebouncingSettings::default(), cache_ttl: None, dedicated_worker: None, - custom_debounce_key: None, - debounce_delay_s: None, })) .arg("world", json!("Hyrule")) .run_until_complete(&db, false, port) @@ -981,13 +965,10 @@ echo "hello $msg" path: None, lock: None, language: ScriptLang::Bash, - 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(), })) .arg("msg", json!("world")) .run_until_complete(&db, false, port) @@ -1016,13 +997,10 @@ def main [ msg: string ] { path: None, lock: None, language: ScriptLang::Nu, - 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(), })) .arg("msg", json!("world")) .run_until_complete(&db, false, port) @@ -1071,13 +1049,10 @@ def main [ path: None, lock: None, language: ScriptLang::Nu, - 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(), })) .arg("a", json!("3")) .arg("b", json!("null")) @@ -1135,13 +1110,10 @@ public class Main { path: None, lock: None, language: ScriptLang::Java, - 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(), })) .arg("a", json!(3)) .arg("b", json!(3.0)) @@ -1172,13 +1144,10 @@ export async function main(a: Date) { path: None, lock: None, language: ScriptLang::Bun, - 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(), })) .arg("a", json!("2024-09-24T10:00:00.000Z")) .run_until_complete(&db, false, port) @@ -1209,13 +1178,10 @@ export async function main(a: Date) { path: None, lock: None, language: ScriptLang::Deno, - 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(), })) .arg("a", json!("2024-09-24T10:00:00.000Z")) .run_until_complete(&db, false, port) @@ -1247,13 +1213,10 @@ def main(a: datetime, b: bytes): path: None, lock: None, language: ScriptLang::Python3, - 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(), })) .arg("a", json!("2024-09-24T10:00:00.000Z")) .arg("b", json!("dGVzdA==")) diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 7495dbd75b..6ae85b7ad0 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -729,13 +729,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 = 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 = + 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, @@ -802,13 +801,12 @@ async fn update_flow_history( let mut tx = user_db.begin(&authed).await?; // Fetch path and perform authorization check early - let path: Option = 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 = + 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, @@ -1482,10 +1480,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" ); @@ -1604,6 +1601,7 @@ mod tests { ConstantDelay, ExponentialDelay, FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, StopAfterIf, }, + jobs::{ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings}, scripts, }; @@ -1654,11 +1652,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(), @@ -1744,17 +1740,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": [ diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 011c55cee0..add81f0b03 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -35,7 +35,8 @@ use windmill_common::flow_conversations::add_message_to_conversation_tx; use windmill_common::flow_status::{JobResult, RestartedFrom}; use windmill_common::jobs::{ check_tag_available_for_workspace_internal, format_completed_job_result, format_result, - DynamicInput, JobTriggerKind, ENTRYPOINT_OVERRIDE, + ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, DynamicInput, + JobTriggerKind, ENTRYPOINT_OVERRIDE, }; use windmill_common::s3_helpers::{upload_artifact_to_store, BundleFormat}; use windmill_common::utils::{RunnableKind, WarnAfterExt}; @@ -284,10 +285,7 @@ pub fn workspaced_service() -> Router { "/completed/import", post(crate::jobs_export::import_completed_jobs).layer(cors.clone()), ) - .route( - "/delete", - post(crate::jobs_export::delete_jobs), - ) + .route("/delete", post(crate::jobs_export::delete_jobs)) .route( "/completed/get/:id", get(get_completed_job).layer(cors.clone()), @@ -4588,16 +4586,17 @@ pub async fn run_workflow_as_code( path: job.script_path, language: job.language.unwrap_or_else(|| ScriptLang::Deno), lock: raw_lock, - custom_concurrency_key: windmill_queue::custom_concurrency_key(&db, &job.id) - .await - .map_err(to_anyhow)?, - concurrent_limit: job.concurrent_limit, - concurrency_time_window_s: job.concurrency_time_window_s, + concurrency_settings: windmill_common::jobs::ConcurrencySettingsWithCustom { + custom_concurrency_key: windmill_queue::custom_concurrency_key(&db, &job.id) + .await + .map_err(to_anyhow)?, + concurrent_limit: job.concurrent_limit, + concurrency_time_window_s: job.concurrency_time_window_s, + }, cache_ttl: job.cache_ttl, dedicated_worker: None, // TODO(debouncing): enable for this mode - custom_debounce_key: None, - debounce_delay_s: None, + debouncing_settings: DebouncingSettings::default(), }), Some(job.tag.clone()), None, @@ -5458,11 +5457,17 @@ pub async fn run_wait_result_script_by_hash( JobPayload::ScriptHash { hash: ScriptHash(hash), path: path, - custom_concurrency_key: concurrency_key, - concurrent_limit: concurrent_limit, - concurrency_time_window_s: concurrency_time_window_s, - custom_debounce_key: debounce_key, - debounce_delay_s, + concurrency_settings: windmill_common::jobs::ConcurrencySettingsWithCustom { + custom_concurrency_key: concurrency_key, + concurrent_limit: concurrent_limit, + concurrency_time_window_s: concurrency_time_window_s, + } + .into(), + debouncing_settings: DebouncingSettings { + custom_key: debounce_key, + delay_s: debounce_delay_s, + ..Default::default() // TODO + }, cache_ttl, language, dedicated_worker, @@ -5934,11 +5939,8 @@ async fn run_preview_script( path: preview.path, language: preview.language.unwrap_or(ScriptLang::Deno), lock: preview.lock, - custom_concurrency_key: None, - concurrent_limit: None, // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here - concurrency_time_window_s: None, // TODO(gbouv): same as above - custom_debounce_key: None, // TODO(pyra): same as for concurrency limits. - debounce_delay_s: None, + concurrency_settings: ConcurrencySettingsWithCustom::default(), // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here + debouncing_settings: DebouncingSettings::default(), // TODO(pyra): same as for concurrency limits. cache_ttl: None, dedicated_worker: preview.dedicated_worker, }), @@ -6056,13 +6058,10 @@ async fn run_bundle_preview_script( path: preview.path, language: preview.language.unwrap_or(ScriptLang::Deno), lock: preview.lock, - concurrent_limit: None, - concurrency_time_window_s: None, cache_ttl: None, dedicated_worker: preview.dedicated_worker, - custom_concurrency_key: None, - custom_debounce_key: None, - debounce_delay_s: None, + concurrency_settings: ConcurrencySettingsWithCustom::default(), + debouncing_settings: DebouncingSettings::default(), }), PushArgs::from(&args), authed.display_username(), @@ -6466,19 +6465,19 @@ async fn add_batch_jobs( add_virtual_items_if_necessary(&mut value.modules); let flow_status = FlowStatus::new(&value); ( - None, // script_hash - path, // script_path - job_kind, // job_kind - None, // language - None, // dedicated_worker - value.concurrency_key.clone(), // custom_concurrency_key - value.concurrent_limit.clone(), // concurrent_limit - value.concurrency_time_window_s, // concurrency_time_window_s - None, // timeout - None, // raw_code - None, // raw_lock - Some(value), // raw_flow - Some(flow_status), // flow_status + None, // script_hash + path, // script_path + job_kind, // job_kind + None, // language + None, // dedicated_worker + value.concurrency_settings.concurrency_key.clone(), // custom_concurrency_key + value.concurrency_settings.concurrent_limit.clone(), // concurrent_limit + value.concurrency_settings.concurrency_time_window_s, // concurrency_time_window_s + None, // timeout + None, // raw_code + None, // raw_lock + Some(value), // raw_flow + Some(flow_status), // flow_status ) } "noop" => ( @@ -6847,13 +6846,10 @@ async fn run_dynamic_select( path: None, language: dynamic_input.x_windmill_dyn_select_lang, 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: ConcurrencySettings::default().into(), + debouncing_settings: DebouncingSettings::default(), }), PushArgs::from(&request.args.unwrap_or_default()), authed.display_username(), @@ -6977,11 +6973,16 @@ pub async fn run_job_by_hash_inner( JobPayload::ScriptHash { hash: ScriptHash(hash), path: path, - custom_concurrency_key: concurrency_key, - concurrent_limit: concurrent_limit, - concurrency_time_window_s: concurrency_time_window_s, - custom_debounce_key: debounce_key, - debounce_delay_s, + concurrency_settings: ConcurrencySettings { + concurrency_key, + concurrent_limit, + concurrency_time_window_s, + }, + debouncing_settings: DebouncingSettings { + custom_key: debounce_key, + delay_s: debounce_delay_s, + ..Default::default() + }, cache_ttl, language, dedicated_worker, diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index e592c43bb3..e3ae82b172 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -844,11 +844,8 @@ async fn trigger_script_with_retry_and_error_handler( JobPayload::ScriptHash { hash, path, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, - custom_debounce_key, - debounce_delay_s, + concurrency_settings, + debouncing_settings, cache_ttl, priority, apply_preprocessor, @@ -862,16 +859,13 @@ async fn trigger_script_with_retry_and_error_handler( error_handler_path, error_handler_args, skip_handler: None, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, cache_ttl, priority, tag_override: tag.clone(), apply_preprocessor, trigger_path: Some(trigger_path), - custom_debounce_key, - debounce_delay_s, + concurrency_settings, + debouncing_settings, }, _ => { return Err(windmill_common::error::Error::internal_err(format!( diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 9134da3c41..2d49d708ab 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -24,6 +24,7 @@ use crate::{ cache, db::DB, error::{Error, Result as WindmillResult}, + jobs::{ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings}, more_serde::{default_empty_string, default_id, default_null, default_true, is_default}, scripts::{Schema, ScriptHash, ScriptLang}, worker::{to_raw_value, Connection}, @@ -171,18 +172,10 @@ pub struct FlowValue { #[serde(default)] #[serde(skip_serializing_if = "is_default")] pub same_worker: bool, - - #[serde(skip_serializing_if = "Option::is_none")] - pub concurrency_key: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub concurrent_limit: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub concurrency_time_window_s: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub debounce_key: Option, - #[serde(skip_serializing_if = "Option::is_none")] - pub debounce_delay_s: Option, - + #[serde(flatten)] + pub concurrency_settings: ConcurrencySettings, + #[serde(flatten)] + pub debouncing_settings: DebouncingSettings, #[serde(skip_serializing_if = "Option::is_none")] pub skip_expr: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -919,12 +912,8 @@ pub enum FlowModuleValue { #[serde(skip_serializing_if = "is_none_or_empty")] tag: Option, language: ScriptLang, - #[serde(skip_serializing_if = "Option::is_none")] - custom_concurrency_key: Option, - #[serde(skip_serializing_if = "Option::is_none")] - concurrent_limit: Option, - #[serde(skip_serializing_if = "Option::is_none")] - concurrency_time_window_s: Option, + #[serde(flatten)] + concurrency_settings: ConcurrencySettingsWithCustom, #[serde(skip_serializing_if = "Option::is_none")] is_trigger: Option, #[serde(skip_serializing_if = "is_none_or_empty_vec")] @@ -945,12 +934,8 @@ pub enum FlowModuleValue { #[serde(skip_serializing_if = "is_none_or_empty")] tag: Option, language: ScriptLang, - #[serde(skip_serializing_if = "Option::is_none")] - custom_concurrency_key: Option, - #[serde(skip_serializing_if = "Option::is_none")] - concurrent_limit: Option, - #[serde(skip_serializing_if = "Option::is_none")] - concurrency_time_window_s: Option, + #[serde(flatten)] + concurrency_settings: ConcurrencySettingsWithCustom, #[serde(skip_serializing_if = "Option::is_none")] is_trigger: Option, #[serde(skip_serializing_if = "is_none_or_empty_vec")] @@ -989,9 +974,6 @@ struct UntaggedFlowModuleValue { lock: Option, tag: Option, language: Option, - custom_concurrency_key: Option, - concurrent_limit: Option, - concurrency_time_window_s: Option, is_trigger: Option, id: Option, default_node: Option, @@ -1000,6 +982,8 @@ struct UntaggedFlowModuleValue { tools: Option>, pass_flow_input_directly: Option, squash: Option, + #[serde(flatten)] + concurrency_settings: ConcurrencySettingsWithCustom, } impl<'de> Deserialize<'de> for FlowModuleValue { @@ -1074,9 +1058,7 @@ impl<'de> Deserialize<'de> for FlowModuleValue { language: untagged .language .ok_or_else(|| serde::de::Error::missing_field("language"))?, - custom_concurrency_key: untagged.custom_concurrency_key, - concurrent_limit: untagged.concurrent_limit, - concurrency_time_window_s: untagged.concurrency_time_window_s, + concurrency_settings: untagged.concurrency_settings, is_trigger: untagged.is_trigger, assets: untagged.assets, }), @@ -1089,9 +1071,7 @@ impl<'de> Deserialize<'de> for FlowModuleValue { language: untagged .language .ok_or_else(|| serde::de::Error::missing_field("language"))?, - custom_concurrency_key: untagged.custom_concurrency_key, - concurrent_limit: untagged.concurrent_limit, - concurrency_time_window_s: untagged.concurrency_time_window_s, + concurrency_settings: untagged.concurrency_settings, is_trigger: untagged.is_trigger, assets: untagged.assets, }), @@ -1234,11 +1214,9 @@ pub async fn resolve_module( id, tag, language, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, is_trigger, assets, + concurrency_settings, } = std::mem::replace(&mut val, Identity) else { unreachable!() @@ -1258,11 +1236,9 @@ pub async fn resolve_module( path: None, tag, language, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, is_trigger, assets, + concurrency_settings, }; } ForloopFlow { modules, modules_node, .. } | WhileloopFlow { modules, modules_node, .. } => { diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 7cd332cb7e..3abacd93fc 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -73,7 +73,7 @@ impl std::fmt::Display for JobTriggerKind { } } -#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Copy, Clone)] +#[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Copy, Clone, Default)] #[sqlx(type_name = "JOB_KIND", rename_all = "lowercase")] #[serde(rename_all(serialize = "lowercase", deserialize = "lowercase"))] pub enum JobKind { @@ -88,6 +88,7 @@ pub enum JobKind { Identity, FlowDependencies, AppDependencies, + #[default] Noop, DeploymentCallback, FlowScript, @@ -344,23 +345,13 @@ pub enum JobPayload { ScriptHash { hash: ScriptHash, path: String, - /// Override default concurrency key - custom_concurrency_key: Option, - /// How many jobs can run at the same time - concurrent_limit: Option, - /// In seconds - concurrency_time_window_s: Option, - /// If not set, will be inferred from the hash(path + step_id + inputs) - custom_debounce_key: Option, - /// Debouncing delay will be determined by the first job with the key. - /// All subsequent jobs with Some will get debounced. - /// If the job has no delay, it will execute immediately, fully ignoring pending delays. - debounce_delay_s: Option, cache_ttl: Option, dedicated_worker: Option, language: ScriptLang, priority: Option, apply_preprocessor: bool, + concurrency_settings: ConcurrencySettings, + debouncing_settings: DebouncingSettings, }, /// Execute flow step (can be subflow only). @@ -372,13 +363,11 @@ pub enum JobPayload { /// Execute flow step FlowScript { id: FlowNodeId, // flow_node(id). + path: String, language: ScriptLang, - custom_concurrency_key: Option, - concurrent_limit: Option, - concurrency_time_window_s: Option, cache_ttl: Option, dedicated_worker: Option, - path: String, + concurrency_settings: ConcurrencySettings, }, /// Inline App Script @@ -458,19 +447,17 @@ pub enum JobPayload { error_handler_path: Option, error_handler_args: Option>>, skip_handler: Option, - custom_concurrency_key: Option, - concurrent_limit: Option, - concurrency_time_window_s: Option, - custom_debounce_key: Option, - debounce_delay_s: Option, cache_ttl: Option, priority: Option, tag_override: Option, trigger_path: Option, apply_preprocessor: bool, + concurrency_settings: ConcurrencySettings, + debouncing_settings: DebouncingSettings, }, DeploymentCallback { path: String, + // debouncing_settings: Option, }, Identity, Noop, @@ -479,6 +466,108 @@ pub enum JobPayload { }, } +// TODO: Add validation logic. +#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)] +pub struct DebouncingSettings { + #[serde( + skip_serializing_if = "Option::is_none", + rename = "debounce_key", + alias = "custom_debounce_key" + )] + /// debounce key is usually stored in the db + /// including when: + /// + /// 1. User have created custom debounce key from ui or cli + /// 2. User used default one + /// + /// in either cases this argument serves as reactive way of overwriting debounce key from the backend. + /// Default: hash(path + step_id + inputs) + pub custom_key: Option, + + #[serde(skip_serializing_if = "Option::is_none", rename = "debounce_delay_s")] + /// Debouncing delay will be determined by the first job with the key. + /// All subsequent jobs with Some will get debounced. + /// If the job has no delay, it will execute immediately, fully ignoring pending delays. + pub delay_s: Option, + + #[serde( + skip_serializing_if = "Option::is_none", + rename = "max_total_debouncing_time" + )] + pub max_total_time: Option, + + #[serde( + skip_serializing_if = "Option::is_none", + rename = "max_total_debounces_amount" + )] + pub max_total_amount: Option, + + #[serde( + skip_serializing_if = "Option::is_none", + rename = "debounce_args_to_accumulate" + )] + /// top level arguments to preserve + /// For every debounce selected arguments will be saved + /// in the end (when job finally starts) arguments will be appended and passed to runnable + /// + /// NOTE: selected args should be the lists. + pub args_to_accumulate: Option>, +} + +impl DebouncingSettings { + pub fn is_default(&self) -> bool { + self == &Self::default() + } +} + +#[derive(Debug, Default, Clone, Serialize, Deserialize, sqlx::FromRow)] +pub struct ConcurrencySettings { + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrent_limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_time_window_s: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow, Default)] +pub struct ConcurrencySettingsWithCustom { + #[serde(skip_serializing_if = "Option::is_none")] + pub custom_concurrency_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrent_limit: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_time_window_s: Option, +} + +impl From for ConcurrencySettingsWithCustom { + fn from( + ConcurrencySettings { concurrency_key, concurrent_limit, concurrency_time_window_s }: ConcurrencySettings, + ) -> Self { + ConcurrencySettingsWithCustom { + custom_concurrency_key: concurrency_key, + concurrency_time_window_s, + concurrent_limit, + } + } +} + +impl From for ConcurrencySettings { + fn from( + ConcurrencySettingsWithCustom { + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + }: ConcurrencySettingsWithCustom, + ) -> Self { + ConcurrencySettings { + concurrency_key: custom_concurrency_key, + concurrency_time_window_s, + concurrent_limit, + } + } +} + #[derive(Clone, Serialize, Deserialize, Debug)] pub struct SkipHandler { pub path: String, @@ -487,20 +576,49 @@ pub struct SkipHandler { pub stop_message: String, } -#[derive(Clone, Serialize, Deserialize, Debug, Default)] +#[derive(Clone, Deserialize, Debug, Default)] pub struct RawCode { pub content: String, pub path: Option, pub hash: Option, pub language: ScriptLang, pub lock: Option, - pub custom_concurrency_key: Option, - pub concurrent_limit: Option, - pub concurrency_time_window_s: Option, - pub custom_debounce_key: Option, - pub debounce_delay_s: Option, pub cache_ttl: Option, pub dedicated_worker: Option, + #[serde(flatten)] + pub concurrency_settings: ConcurrencySettingsWithCustom, + #[serde(flatten)] + // NOTE: Since we can only deserialize the struct, + // even though the older versions pass `custom_debounce_key` to RawCode, + // we can still have `debounce_key` in DebouncingSettings + // we just add alias `custom_debounce_key` + // however, serializing this settings will produce `debounce_key` + pub debouncing_settings: DebouncingSettings, +} + +impl JobPayload { + pub fn job_kind(&self) -> JobKind { + match self { + JobPayload::Noop => JobKind::Noop, + JobPayload::Identity => JobKind::Identity, + JobPayload::Code { .. } => JobKind::Preview, + JobPayload::AIAgent { .. } => JobKind::AIAgent, + JobPayload::FlowNode { .. } => JobKind::FlowNode, + JobPayload::ScriptHash { .. } => JobKind::Script, + JobPayload::AppScript { .. } => JobKind::AppScript, + JobPayload::RawFlow { .. } => JobKind::FlowPreview, + JobPayload::ScriptHub { .. } => JobKind::Script_Hub, + JobPayload::FlowScript { .. } => JobKind::FlowScript, + JobPayload::Dependencies { .. } => JobKind::Dependencies, + JobPayload::SingleStepFlow { .. } => JobKind::SingleStepFlow, + JobPayload::AppDependencies { .. } => JobKind::AppDependencies, + JobPayload::FlowDependencies { .. } => JobKind::FlowDependencies, + JobPayload::RawScriptDependencies { .. } => JobKind::Dependencies, + JobPayload::RawFlowDependencies { .. } => JobKind::FlowDependencies, + JobPayload::DeploymentCallback { .. } => JobKind::DeploymentCallback, + JobPayload::Flow { .. } | JobPayload::RestartedFlow { .. } => JobKind::Flow, + } + } } type Tag = String; @@ -597,17 +715,23 @@ pub async fn script_path_to_payload<'e>( JobPayload::ScriptHash { hash: ScriptHash(hash), path: script_path.to_owned(), - custom_concurrency_key: concurrency_key, - concurrent_limit, - concurrency_time_window_s, - custom_debounce_key: debounce_key, - debounce_delay_s, cache_ttl, language, dedicated_worker, priority, apply_preprocessor: !skip_preprocessor.unwrap_or(false) && has_preprocessor.unwrap_or(false), + concurrency_settings: ConcurrencySettingsWithCustom { + custom_concurrency_key: concurrency_key, + concurrent_limit, + concurrency_time_window_s, + } + .into(), + debouncing_settings: DebouncingSettings { + custom_key: debounce_key, + delay_s: debounce_delay_s, + ..Default::default() + }, }, tag, delete_after_use, diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 191fe7343a..25bde2804e 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -495,6 +495,10 @@ pub struct NewScript { pub tag: Option, pub draft_only: Option, pub envs: Option>, + // NOTE: concurrency and debounce data is inline, + // bc it was this before refactor + // and rust seems to hash it differently + // for backwards compat we keep them inline pub concurrency_key: Option, pub concurrent_limit: Option, pub concurrency_time_window_s: Option, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 059bfc69b8..8f7b2ed7a1 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -14,9 +14,9 @@ use async_recursion::async_recursion; use chrono::{DateTime, Utc}; use futures::future::TryFutureExt; use itertools::Itertools; -use quick_cache::sync::Cache; #[cfg(feature = "prometheus")] use prometheus::IntCounter; +use quick_cache::sync::Cache; use regex::Regex; use reqwest::Client; use serde::Deserialize; @@ -38,7 +38,10 @@ use windmill_common::add_time; use windmill_common::auth::JobPerms; #[cfg(feature = "benchmark")] use windmill_common::bench::BenchmarkIter; -use windmill_common::jobs::{JobTriggerKind, EMAIL_ERROR_HANDLER_USER_EMAIL}; +use windmill_common::jobs::{ + ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, JobTriggerKind, + EMAIL_ERROR_HANDLER_USER_EMAIL, +}; use windmill_common::utils::{configure_client, now_from_db}; use windmill_common::worker::{Connection, MIN_VERSION_SUPPORTS_DEBOUNCING, SCRIPT_TOKEN_EXPIRY}; @@ -168,7 +171,6 @@ pub async fn cancel_single_job<'c>( db: &Pool, force_cancel: bool, ) -> error::Result<(Transaction<'c, Postgres>, Option)> { - let id = job_running.id; if force_cancel || (job_running.parent_job.is_none() && !job_running.running) { let username = username.to_string(); @@ -236,7 +238,6 @@ pub async fn cancel_job<'c>( //TODO fetch mini completed job instead of QueuedJob let job = get_queued_job_v2(&mut *tx, &id).await?; - if job.is_none() { return Ok((tx, None)); } @@ -249,7 +250,6 @@ pub async fn cancel_job<'c>( )); } - if job.workspace_id != w_id { return Err(Error::BadRequest( "You are not authorized to cancel this job belonging to another workspace".to_string(), @@ -324,16 +324,8 @@ ORDER BY depth, id tracing::info!("Found {} child jobs to cancel", jobs_to_cancel.len()); } - let (ntx, _) = cancel_single_job( - username, - reason.clone(), - job, - w_id, - tx, - db, - force_cancel, - ) - .await?; + let (ntx, _) = + cancel_single_job(username, reason.clone(), job, w_id, tx, db, force_cancel).await?; tx = ntx; if !force_cancel { @@ -357,16 +349,9 @@ ORDER BY depth, id let job = get_queued_job_v2(&mut *tx, &job_id).await?; if let Some(job) = job { - let (ntx, _) = cancel_single_job( - username, - reason.clone(), - job, - w_id, - tx, - db, - force_cancel, - ) - .await?; + let (ntx, _) = + cancel_single_job(username, reason.clone(), job, w_id, tx, db, force_cancel) + .await?; tx = ntx; } } @@ -445,13 +430,10 @@ pub async fn push_init_job<'c>( path: Some(format!("{INIT_SCRIPT_PATH_PREFIX}{worker_name}")), language: ScriptLang::Bash, lock: None, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, - custom_debounce_key: None, - debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, + concurrency_settings: ConcurrencySettingsWithCustom::default(), + debouncing_settings: DebouncingSettings::default(), }), PushArgs::from(&ehm), worker_name, @@ -504,13 +486,10 @@ pub async fn push_periodic_bash_job<'c>( )), language: ScriptLang::Bash, lock: None, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, - custom_debounce_key: None, - debounce_delay_s: None, cache_ttl: None, dedicated_worker: None, + concurrency_settings: ConcurrencySettingsWithCustom::default(), + debouncing_settings: DebouncingSettings::default(), }), PushArgs::from(&ehm), worker_name, @@ -867,7 +846,6 @@ pub async fn add_completed_job( restart_job_if_perpetual(db, completed_job, &canceled_by).await?; - // tracing::error!("4 {:?}", start.elapsed()); Ok((completed_job.id, duration)) @@ -1343,9 +1321,6 @@ async fn restart_job_if_perpetual_inner( JobPayload::ScriptHash { hash, path: queued_job.runnable_path.clone().unwrap_or_default(), - custom_concurrency_key: custom_concurrency_key(db, &queued_job.id).await?, - concurrent_limit: None, - concurrency_time_window_s: None, cache_ttl: queued_job.cache_ttl, dedicated_worker: None, language: queued_job @@ -1354,9 +1329,13 @@ async fn restart_job_if_perpetual_inner( .unwrap_or_else(|| ScriptLang::Deno), priority: queued_job.priority, apply_preprocessor: false, + concurrency_settings: ConcurrencySettings { + concurrency_key: custom_concurrency_key(db, &queued_job.id).await?, + concurrent_limit: None, + concurrency_time_window_s: None, + }, // TODO(debouncing): handle properly - custom_debounce_key: None, - debounce_delay_s: None, + debouncing_settings: DebouncingSettings::default(), }, PushArgs::from(&args.0), &queued_job.created_by, @@ -1442,7 +1421,6 @@ fn apply_completed_job_cloud_usage( tracing::error!("Failed to get team plan status to update usage for workspace {w_id}: {err:#}"); } }; - }).await; if let Err(_) = result { @@ -1571,8 +1549,9 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> (cached.0.clone(), cached.1.clone(), cached.2) } else { // Cache expired, fetch from database - let row_result = sqlx::query_as::<_, (Option, Option>>, bool)>( - r#" + let row_result = + sqlx::query_as::<_, (Option, Option>>, bool)>( + r#" SELECT error_handler, error_handler_extra_args, @@ -1582,25 +1561,33 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> WHERE workspace_id = $1 "#, - ) - .bind(&w_id) - .fetch_optional(db) - .await - .context("fetching error handler info from workspace_settings")? - .ok_or_else(|| Error::internal_err(format!("no workspace settings for id {w_id}")))?; + ) + .bind(&w_id) + .fetch_optional(db) + .await + .context("fetching error handler info from workspace_settings")? + .ok_or_else(|| { + Error::internal_err(format!("no workspace settings for id {w_id}")) + })?; // Update cache with 60s TTL let expiry = now + 60; WORKSPACE_ERROR_HANDLER_CACHE.insert( w_id.clone(), - (row_result.0.clone(), row_result.1.clone(), row_result.2, expiry) + ( + row_result.0.clone(), + row_result.1.clone(), + row_result.2, + expiry, + ), ); row_result } } else { // Cache miss, fetch from database - let row_result = sqlx::query_as::<_, (Option, Option>>, bool)>( - r#" + let row_result = + sqlx::query_as::<_, (Option, Option>>, bool)>( + r#" SELECT error_handler, error_handler_extra_args, @@ -1610,18 +1597,25 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync> WHERE workspace_id = $1 "#, - ) - .bind(&w_id) - .fetch_optional(db) - .await - .context("fetching error handler info from workspace_settings")? - .ok_or_else(|| Error::internal_err(format!("no workspace settings for id {w_id}")))?; + ) + .bind(&w_id) + .fetch_optional(db) + .await + .context("fetching error handler info from workspace_settings")? + .ok_or_else(|| { + Error::internal_err(format!("no workspace settings for id {w_id}")) + })?; // Store in cache with 60s TTL let expiry = now + 60; WORKSPACE_ERROR_HANDLER_CACHE.insert( w_id.clone(), - (row_result.0.clone(), row_result.1.clone(), row_result.2, expiry) + ( + row_result.0.clone(), + row_result.1.clone(), + row_result.2, + expiry, + ), ); row_result }; @@ -1991,7 +1985,6 @@ pub struct MiniPulledJob { pub permissioned_as_end_user_email: Option, } - #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MiniCompletedJob { @@ -2041,7 +2034,6 @@ impl From for MiniCompletedJob { concurrent_limit: job.concurrent_limit, tag: job.tag, cache_ttl: job.cache_ttl, - } } } @@ -2116,11 +2108,16 @@ impl MiniCompletedJob { pub fn is_dependency(&self) -> bool { self.kind.is_dependency() } - } -fn schedule_path(trigger_kind: &Option, trigger: &Option) -> Option { - if trigger_kind.as_ref().is_some_and(|t| matches!(t, JobTriggerKind::Schedule)) { +fn schedule_path( + trigger_kind: &Option, + trigger: &Option, +) -> Option { + if trigger_kind + .as_ref() + .is_some_and(|t| matches!(t, JobTriggerKind::Schedule)) + { trigger.clone() } else { None @@ -2413,7 +2410,6 @@ pub async fn get_mini_pulled_job<'c>( Ok(job) } - pub struct QueuedJobV2 { pub id: Uuid, pub workspace_id: String, @@ -2449,7 +2445,9 @@ impl QueuedJobV2 { } pub async fn get_queued_job_v2<'c>( - e: impl PgExecutor<'c>, job_id: &Uuid) -> error::Result> { + e: impl PgExecutor<'c>, + job_id: &Uuid, +) -> error::Result> { let job = sqlx::query_as!( QueuedJobV2, "SELECT id, q.workspace_id, j.runnable_id as \"runnable_id: ScriptHash\", scheduled_for, parent_job, flow_innermost_root_job, runnable_path, kind as \"kind: JobKind\", started_at, permissioned_as, created_by, script_lang as \"script_lang: ScriptLang\", @@ -2469,7 +2467,7 @@ pub struct PulledJobResult { pub job: Option, pub suspended: bool, pub missing_concurrency_key: bool, - pub error_while_preprocessing: Option + pub error_while_preprocessing: Option, } #[derive(thiserror::Error, Debug)] @@ -2480,7 +2478,6 @@ pub enum PulledJobResultToJobErr { ErrorWhilePreprocessing(JobCompleted), } - impl PulledJobResult { pub fn to_pulled_job(self) -> Result, PulledJobResultToJobErr> { match self { @@ -2761,7 +2758,6 @@ impl PulledJobResult { } } else { tracing::warn!("Debouncing is not supported on this version of Windmill. Minimum version required for debouncing support."); - } // This will unblock pusher. tx.commit().await?; @@ -2853,21 +2849,21 @@ pub async fn pull( // Concurrency limit is available for either enterprise job or dependency job && (cfg!(feature = "enterprise") || (job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING)) => { - crate::jobs_ee::apply_concurrency_limit( - db, - pull_loop_count, - suspended, - job, - ) - .await? - .unwrap_or(PulledJobResult { - job: None, - suspended, - missing_concurrency_key: false, - error_while_preprocessing: None, - }) + crate::jobs_ee::apply_concurrency_limit(db, pull_loop_count, suspended, job) + .await? + .unwrap_or(PulledJobResult { + job: None, + suspended, + missing_concurrency_key: false, + error_while_preprocessing: None, + }) } - _ => PulledJobResult { job, suspended, missing_concurrency_key: false, error_while_preprocessing: None }, + _ => PulledJobResult { + job, + suspended, + missing_concurrency_key: false, + error_while_preprocessing: None, + }, }; Ok::<_, Error>(pulled_job_result) @@ -2885,7 +2881,12 @@ pub async fn pull( ) .await?; let Some(job) = job else { - return Ok(PulledJobResult { job: None, suspended, missing_concurrency_key: false, error_while_preprocessing: None }); + return Ok(PulledJobResult { + job: None, + suspended, + missing_concurrency_key: false, + error_while_preprocessing: None, + }); }; let has_concurent_limit = job.concurrent_limit.is_some(); @@ -2896,7 +2897,10 @@ pub async fn pull( } #[cfg(not(feature = "enterprise"))] - let has_concurent_limit = job.is_dependency() && job.concurrent_limit.is_some() && cfg!(feature = "private") && !*WMDEBUG_NO_DJOB_DEBOUNCING; + let has_concurent_limit = job.is_dependency() + && job.concurrent_limit.is_some() + && cfg!(feature = "private") + && !*WMDEBUG_NO_DJOB_DEBOUNCING; // if we don't have private flag, we don't have concurrency limit // concurrency check. If more than X jobs for this path are already running, we re-queue and pull another job from the queue @@ -3593,13 +3597,13 @@ pub async fn job_is_complete(db: &DB, id: Uuid, w_id: &str) -> error::Result + Send + 'a, ->(id: &'a Uuid, w_id: &'a str, db: A) -> impl Future>> + Send + 'a { - async move { - let mut conn = db.acquire().await?; +pub fn get_mini_completed_job<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres> + Send + 'a>( + id: &'a Uuid, + w_id: &'a str, + db: A, +) -> impl Future>> + Send + 'a { + async move { + let mut conn = db.acquire().await?; sqlx::query_as!( MiniCompletedJob, "SELECT @@ -3616,7 +3620,6 @@ A: sqlx::Acquire<'e, Database = Postgres> + Send + 'a, } } - pub enum PushIsolationLevel<'c> { IsolatedRoot(DB), Isolated(UserDB, Authed), @@ -3828,8 +3831,7 @@ pub async fn push<'c, 'd>( mut email: &str, mut permissioned_as: String, token_prefix: Option<&str>, - #[allow(unused_mut)] - mut scheduled_for_o: Option>, + #[allow(unused_mut)] mut scheduled_for_o: Option>, schedule_path: Option, parent_job: Option, root_job: Option, @@ -4031,87 +4033,87 @@ pub async fn push<'c, 'd>( } } + #[derive(Default)] + struct JobPayloadUntagged { + runnable_id: Option, + runnable_path: Option, + raw_code_tuple: Option<(String, Option)>, // (content, lock) + job_kind: JobKind, + raw_flow: Option, + flow_status: Option, + language: Option, + cache_ttl: Option, + dedicated_worker: Option, + _low_level_priority: Option, + concurrency_settings: ConcurrencySettings, + debouncing_settings: DebouncingSettings, + } let mut preprocessed = None; - #[allow(unused)] - let ( - script_hash, - script_path, + #[allow(unused)] + let JobPayloadUntagged { + runnable_id, + runnable_path, raw_code_tuple, job_kind, raw_flow, flow_status, language, - mut custom_concurrency_key, - mut concurrent_limit, - concurrency_time_window_s, cache_ttl, dedicated_worker, _low_level_priority, - custom_debounce_key, - debounce_delay_s, - ) = match job_payload { + concurrency_settings: + ConcurrencySettings { + mut concurrency_key, + mut concurrent_limit, + concurrency_time_window_s, // + }, + debouncing_settings, + } = match job_payload { JobPayload::ScriptHash { hash, path, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, cache_ttl, language, dedicated_worker, priority, apply_preprocessor, - custom_debounce_key, - debounce_delay_s, + concurrency_settings, + debouncing_settings, } => { if apply_preprocessor { preprocessed = Some(false); } - ( - Some(hash.0), - Some(path), - None, - JobKind::Script, - None, - None, - Some(language), - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + JobPayloadUntagged { + runnable_id: Some(hash.0), + runnable_path: Some(path), + job_kind: JobKind::Script, + language: Some(language), + concurrency_settings, + debouncing_settings, cache_ttl, dedicated_worker, - priority, - custom_debounce_key, - debounce_delay_s, - ) + _low_level_priority: priority, + ..Default::default() + } } JobPayload::FlowScript { id, // flow_node(id). language, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, cache_ttl, dedicated_worker, path, - } => ( - Some(id.0), - Some(path), - None, - JobKind::FlowScript, - None, - None, - Some(language), - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, + } => JobPayloadUntagged { + runnable_id: Some(id.0), + runnable_path: Some(path), + job_kind: JobKind::FlowScript, + language: Some(language), + concurrency_settings, cache_ttl, dedicated_worker, - None, - None, // custom_debounce_key removed for flow steps - None, // debounce_delay_s removed for flow steps - ), + ..Default::default() + }, JobPayload::FlowNode { id, path } => { let data = cache::flow::fetch_flow(_db, id).await?; let value = data.value(); @@ -4124,46 +4126,28 @@ pub async fn push<'c, 'd>( // `raw_flow` is fetched on pull. None }; - ( - Some(id.0), - Some(path), - None, - JobKind::FlowNode, - value_o, - status, - None, - None, - None, - None, - None, - None, - None, - None, - None, - ) + JobPayloadUntagged { + runnable_id: Some(id.0), + runnable_path: Some(path), + job_kind: JobKind::FlowNode, + raw_flow: value_o, + flow_status: status, + ..Default::default() + } } JobPayload::AppScript { id, // app_script(id). path, language, cache_ttl, - } => ( - Some(id.0), - path, - None, - JobKind::AppScript, - None, - None, - Some(language), - None, - None, - None, + } => JobPayloadUntagged { + runnable_id: Some(id.0), + runnable_path: path, + job_kind: JobKind::AppScript, + language: Some(language), cache_ttl, - None, - None, - None, - None, - ), + ..Default::default() + }, JobPayload::ScriptHub { path, apply_preprocessor } => { if path == "hub/7771/slack" || path == "hub/7836/slack" || path == "hub/9084/slack" { // these scripts send app reports to slack @@ -4180,24 +4164,12 @@ pub async fn push<'c, 'd>( get_full_hub_script_by_path(StripPath(path.clone()), &HTTP_CLIENT, Some(_db)) .await?; - ( - None, - Some(path), - None, - // Some((script.content, script.lockfile)), - JobKind::Script_Hub, - None, - None, - Some(hub_script.language), - None, - None, - None, - None, - None, - None, - None, - None, - ) + JobPayloadUntagged { + runnable_path: Some(path), + job_kind: JobKind::Script_Hub, + language: Some(hub_script.language), + ..Default::default() + } } JobPayload::Code(RawCode { content, @@ -4205,85 +4177,49 @@ pub async fn push<'c, 'd>( hash, language, lock, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, cache_ttl, dedicated_worker, - custom_debounce_key, - debounce_delay_s, - }) => ( - hash, - path, - Some((content, lock)), - JobKind::Preview, - None, - None, - Some(language), - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, + debouncing_settings, + }) => JobPayloadUntagged { + runnable_id: hash, + runnable_path: path, + raw_code_tuple: Some((content, lock)), + job_kind: JobKind::Preview, + language: Some(language), + concurrency_settings: concurrency_settings.into(), + debouncing_settings, cache_ttl, dedicated_worker, - None, - custom_debounce_key, - debounce_delay_s, - ), - JobPayload::Dependencies { hash, language, path, dedicated_worker } => ( - Some(hash.0), - Some(path.clone()), - None, - JobKind::Dependencies, - None, - None, - Some(language), - None, - None, - None, - None, + ..Default::default() + }, + JobPayload::Dependencies { hash, language, path, dedicated_worker } => JobPayloadUntagged { + runnable_id: Some(hash.0), + runnable_path: Some(path.clone()), + job_kind: JobKind::Dependencies, + language: Some(language), dedicated_worker, - None, - None, - None, - ), + ..Default::default() + }, // CLI usage, is not modifying db, no need for debouncing. - JobPayload::RawScriptDependencies { script_path, content, language } => ( - None, - Some(script_path), - Some((content, None)), - JobKind::Dependencies, - None, - None, - Some(language), - None, - None, - None, - None, - None, - None, - None, - None, - ), + JobPayload::RawScriptDependencies { script_path, content, language } => { + JobPayloadUntagged { + runnable_path: Some(script_path), + raw_code_tuple: Some((content, None)), + job_kind: JobKind::Dependencies, + language: Some(language), + ..Default::default() + } + } // CLI usage, is not modifying db, no need for debouncing. - JobPayload::RawFlowDependencies { path, flow_value } => ( - None, - Some(path), - None, - JobKind::FlowDependencies, - Some(flow_value), - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - ), + JobPayload::RawFlowDependencies { path, flow_value } => JobPayloadUntagged { + runnable_path: Some(path), + job_kind: JobKind::FlowDependencies, + raw_flow: Some(flow_value), + ..Default::default() + }, JobPayload::FlowDependencies { path, dedicated_worker, version } => { #[cfg(test)] let skip_compat = args @@ -4305,41 +4241,21 @@ pub async fn push<'c, 'd>( // `raw_flow` is fetched on pull. None }; - ( - Some(version), - Some(path.clone()), - None, - JobKind::FlowDependencies, - value_o, - None, - None, - None, - None, - None, - None, + JobPayloadUntagged { + runnable_id: Some(version), + runnable_path: Some(path.clone()), + job_kind: JobKind::FlowDependencies, + raw_flow: value_o, dedicated_worker, - None, - None, - None, - ) + ..Default::default() + } } - JobPayload::AppDependencies { path, version } => ( - Some(version), - Some(path.clone()), - None, - JobKind::AppDependencies, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - ), + JobPayload::AppDependencies { path, version } => JobPayloadUntagged { + runnable_id: Some(version), + runnable_path: Some(path.clone()), + job_kind: JobKind::AppDependencies, + ..Default::default() + }, JobPayload::RawFlow { mut value, path, restarted_from } => { add_virtual_items_if_necessary(&mut value.modules); @@ -4386,30 +4302,19 @@ pub async fn push<'c, 'd>( FlowStatus::new(&value) } // this is a new flow being pushed, flow_status is set to flow_value }; - let concurrency_key = value.concurrency_key.clone(); - let concurrent_limit = value.concurrent_limit; - let concurrency_time_window_s = value.concurrency_time_window_s; - let debounce_key = value.debounce_key.clone(); - let debounce_delay_s = value.debounce_delay_s; let cache_ttl = value.cache_ttl.map(|x| x as i32); let priority = value.priority; - ( - None, - path, - None, - JobKind::FlowPreview, - Some(value), - Some(flow_status), - None, - concurrency_key, - concurrent_limit, - concurrency_time_window_s, + JobPayloadUntagged { + runnable_path: path, + job_kind: JobKind::FlowPreview, + flow_status: Some(flow_status), cache_ttl, - None, - priority, - debounce_key, - debounce_delay_s, - ) + _low_level_priority: priority, + concurrency_settings: value.concurrency_settings.clone(), + debouncing_settings: value.debouncing_settings.clone(), + raw_flow: Some(value), + ..Default::default() + } } JobPayload::SingleStepFlow { path, @@ -4420,16 +4325,13 @@ pub async fn push<'c, 'd>( error_handler_args, skip_handler, args, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, cache_ttl, priority, tag_override, trigger_path, apply_preprocessor, - custom_debounce_key, - debounce_delay_s, + debouncing_settings, + concurrency_settings, } => { // Determine if this is a flow or a script let is_flow = flow_version.is_some(); @@ -4563,13 +4465,10 @@ pub async fn push<'c, 'd>( let flow_value = FlowValue { modules, failure_module, - concurrency_time_window_s, - concurrent_limit, - debounce_key: custom_debounce_key.clone(), - debounce_delay_s, + concurrency_settings: concurrency_settings.clone(), + debouncing_settings: debouncing_settings.clone(), priority, cache_ttl: cache_ttl.map(|val| val as u32), - concurrency_key: custom_concurrency_key.clone(), same_worker: false, early_return: None, skip_expr: None, @@ -4579,23 +4478,17 @@ pub async fn push<'c, 'd>( }; // this is a new flow being pushed, flow_status is set to flow_value: let flow_status: FlowStatus = FlowStatus::new(&flow_value); - ( - None, // No version needed - flow is stored in raw_flow like FlowPreview - Some(path), - None, - JobKind::SingleStepFlow, - Some(flow_value), - Some(flow_status), - None, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + JobPayloadUntagged { + runnable_path: Some(path), + job_kind: JobKind::SingleStepFlow, + raw_flow: Some(flow_value), + flow_status: Some(flow_status), cache_ttl, - None, - priority, - custom_debounce_key, - debounce_delay_s, - ) + _low_level_priority: priority, + concurrency_settings, + debouncing_settings, + ..Default::default() + } } JobPayload::Flow { path, dedicated_worker, apply_preprocessor, version } => { let mut ntx = tx.into_tx().await?; @@ -4618,20 +4511,18 @@ pub async fn push<'c, 'd>( 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 mut concurrent_limit = value.concurrent_limit; - - let custom_debounce_key = value.debounce_key.clone(); - let mut debounce_delay_s = value.debounce_delay_s; + let mut concurrency_settings = value.concurrency_settings.clone(); + let mut debouncing_settings = value.debouncing_settings.clone(); if !apply_preprocessor { value.preprocessor_module = None; } else { tag = None; - concurrent_limit = None; + + concurrency_settings.concurrent_limit = None; // TODO: May be re-enable? - debounce_delay_s = None; + debouncing_settings.delay_s = None; + preprocessed = Some(false); } @@ -4652,23 +4543,19 @@ pub async fn push<'c, 'd>( // by additional checks when handling the flow. None }; - ( - Some(version), // Starting from `v1.436`, the version id is used to fetch the value on pull. - Some(path), - None, - JobKind::Flow, - value_o, - Some(status), - None, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + JobPayloadUntagged { + runnable_id: Some(version), // Starting from `v1.436`, the version id is used to fetch the value on pull. + runnable_path: Some(path), + job_kind: JobKind::Flow, + raw_flow: value_o, + flow_status: Some(status), cache_ttl, dedicated_worker, - priority, - custom_debounce_key, - debounce_delay_s, - ) + _low_level_priority: priority, + concurrency_settings, + debouncing_settings, + ..Default::default() + } } JobPayload::RestartedFlow { completed_job_id, step_id, branch_or_iteration_n } => { let ( @@ -4687,6 +4574,7 @@ pub async fn push<'c, 'd>( branch_or_iteration_n, ) .await?; + let restarted_flow_status = FlowStatus { step: step_n, modules: truncated_modules, @@ -4715,11 +4603,8 @@ pub async fn push<'c, 'd>( }; let value = flow_data.value(); let priority = value.priority; - let concurrency_key = value.concurrency_key.clone(); - let concurrent_limit = value.concurrent_limit; - let concurrency_time_window_s = value.concurrency_time_window_s; - let debounce_key = value.debounce_key.clone(); - let debounce_delay_s = value.debounce_delay_s; + let concurrency_settings = value.concurrency_settings.clone(); + let debouncing_settings = value.debouncing_settings.clone(); let cache_ttl = value.cache_ttl.map(|x| x as i32); // Keep inserting `value` if not all workers are updated. // Starting at `v1.440`, the value is fetched on pull from the version id. @@ -4729,92 +4614,38 @@ pub async fn push<'c, 'd>( // `raw_flow` is fetched on pull. None }; - ( - version, - flow_path, - None, - JobKind::Flow, - value_o, - Some(restarted_flow_status), - None, - concurrency_key, - concurrent_limit, - concurrency_time_window_s, + JobPayloadUntagged { + runnable_id: version, + runnable_path: flow_path, + job_kind: JobKind::Flow, + raw_flow: value_o, + flow_status: Some(restarted_flow_status), cache_ttl, - None, - priority, - debounce_key, - debounce_delay_s, - ) + _low_level_priority: priority, + concurrency_settings, + debouncing_settings, + ..Default::default() + } } - JobPayload::DeploymentCallback { path } => ( - None, - Some(path.clone()), - None, - JobKind::DeploymentCallback, - None, - None, - None, - Some(format!("{workspace_id}:git_sync")), - Some(1), - Some(0), - None, - None, - None, - None, - None, - ), - JobPayload::Identity => ( - None, - None, - None, - JobKind::Identity, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - ), - JobPayload::Noop => ( - None, - None, - None, - JobKind::Noop, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - ), - JobPayload::AIAgent { path } => ( - None, - Some(path), - None, - JobKind::AIAgent, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - None, - ), + JobPayload::DeploymentCallback { path } => JobPayloadUntagged { + runnable_path: Some(path.clone()), + job_kind: JobKind::DeploymentCallback, + concurrency_settings: ConcurrencySettings { + concurrency_key: Some(format!("{workspace_id}:git_sync")), + concurrent_limit: Some(1), + concurrency_time_window_s: Some(0), + }, + ..Default::default() + }, + JobPayload::Identity => { + JobPayloadUntagged { job_kind: JobKind::Identity, ..Default::default() } + } + JobPayload::Noop => JobPayloadUntagged { job_kind: JobKind::Noop, ..Default::default() }, + JobPayload::AIAgent { path } => JobPayloadUntagged { + runnable_path: Some(path), + job_kind: JobKind::AIAgent, + ..Default::default() + }, }; // Enforce concurrency limit on all dependency jobs. @@ -4823,13 +4654,13 @@ pub async fn push<'c, 'd>( // // This is not the case for scripts, so we can potentially have multiple djobs for scripts at the same time. if let (Some(path), true) = ( - &script_path, + &runnable_path, cfg!(feature = "private") && job_kind.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING && *MIN_VERSION_SUPPORTS_DEBOUNCING.read().await, ) { - custom_concurrency_key = Some(format!("dependency:{workspace_id}/{path}")); + concurrency_key = Some(format!("dependency:{workspace_id}/{path}")); concurrent_limit = Some(1); } @@ -4895,7 +4726,7 @@ pub async fn push<'c, 'd>( } else { "" }, - script_path.clone().expect("dedicated script has a path") + runnable_path.clone().expect("dedicated script has a path") ) } else { if tag == Some("".to_string()) { @@ -4973,7 +4804,7 @@ pub async fn push<'c, 'd>( match ( scheduled_for_o.is_some(), job_kind.is_dependency(), - script_path.clone(), + runnable_path.clone(), *WMDEBUG_NO_DJOB_DEBOUNCING, *MIN_VERSION_SUPPORTS_DEBOUNCING.read().await, // We only do debouncing for jobs triggered by relative imports @@ -5155,10 +4986,10 @@ pub async fn push<'c, 'd>( if schedule_path.is_none() { if let Some(debounced_job_id) = crate::jobs_ee::maybe_apply_debouncing( &job_id, - debounce_delay_s, - custom_debounce_key, + debouncing_settings.delay_s, + debouncing_settings.custom_key, workspace_id, - script_path.clone(), + runnable_path.clone(), &job_kind, &args, &mut scheduled_for_o, @@ -5174,9 +5005,9 @@ pub async fn push<'c, 'd>( insert_concurrency_key( workspace_id, &args, - &script_path, + &runnable_path, job_kind, - custom_concurrency_key, + concurrency_key, &mut tx, job_id, ) @@ -5294,8 +5125,8 @@ pub async fn push<'c, 'd>( parent_job, user, permissioned_as, - script_hash, - script_path.clone(), + runnable_id, + runnable_path.clone(), Json(args) as Json, job_kind.clone() as JobKind, schedule_path, @@ -5384,7 +5215,7 @@ pub async fn push<'c, 'd>( let operation_name = match job_kind { JobKind::Preview => "jobs.run.preview", JobKind::Script => { - s = ScriptHash(script_hash.unwrap()).to_string(); + s = ScriptHash(runnable_id.unwrap()).to_string(); hm.insert("hash", s.as_str()); "jobs.run.script" } @@ -5430,7 +5261,7 @@ pub async fn push<'c, 'd>( operation_name, ActionKind::Execute, workspace_id, - script_path.as_ref().map(|x| x.as_str()), + runnable_path.as_ref().map(|x| x.as_str()), Some(hm), ) .warn_after_seconds(1) @@ -5721,8 +5552,6 @@ async fn restarted_flows_resolution( )) } - - // Wrapper struct to send both job and optional flow_runners to dedicated workers pub struct DedicatedWorkerJob { pub job: Arc, @@ -5740,7 +5569,11 @@ pub struct FlowRunners { impl Drop for FlowRunners { fn drop(&mut self) { let total_runners = self.handles.len(); - tracing::info!("dropping {} flow runners for job {}", total_runners, self.job_id); + tracing::info!( + "dropping {} flow runners for job {}", + total_runners, + self.job_id + ); // First, drop all senders to signal workers to stop gracefully self.runners.clear(); @@ -5756,12 +5589,17 @@ impl Drop for FlowRunners { // Wait up to 5 seconds for natural termination let timeout_result = tokio::time::timeout( tokio::time::Duration::from_secs(5), - futures::future::join_all(handles) - ).await; + futures::future::join_all(handles), + ) + .await; match timeout_result { Ok(_) => { - tracing::info!("all {} flow runners for job {} terminated gracefully", total_runners, job_id); + tracing::info!( + "all {} flow runners for job {} terminated gracefully", + total_runners, + job_id + ); } Err(_) => { // Timeout reached, abort only the handles that haven't finished @@ -5859,4 +5697,3 @@ pub async fn get_same_worker_job( )) }) } - diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 52a397ce33..4a74ed80d6 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -20,6 +20,8 @@ use windmill_common::flows::Retry; use windmill_common::get_flow_version_info_from_version; use windmill_common::get_latest_flow_version_id_for_path; use windmill_common::jobs::check_tag_available_for_workspace_internal; +use windmill_common::jobs::ConcurrencySettings; +use windmill_common::jobs::DebouncingSettings; use windmill_common::jobs::JobPayload; use windmill_common::schedule::schedule_to_user; use windmill_common::scripts::ScriptHash; @@ -258,16 +260,13 @@ pub async fn push_scheduled_job<'c>( stop_condition, stop_message, }), - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, cache_ttl: None, priority: None, tag_override: schedule.tag.clone(), trigger_path: None, apply_preprocessor: false, - custom_debounce_key: None, - debounce_delay_s: None, + concurrency_settings: ConcurrencySettings::default(), + debouncing_settings: DebouncingSettings::default(), }, if schedule.tag.as_ref().is_some_and(|x| x != "") { schedule.tag.clone() @@ -360,16 +359,13 @@ pub async fn push_scheduled_job<'c>( error_handler_args: None, skip_handler: None, args: static_args, - custom_concurrency_key: None, - concurrent_limit: None, - concurrency_time_window_s: None, cache_ttl, priority, tag_override: schedule.tag.clone(), trigger_path: None, apply_preprocessor: false, - custom_debounce_key: None, - debounce_delay_s: None, + concurrency_settings: ConcurrencySettings::default(), + debouncing_settings: DebouncingSettings::default(), }, if schedule.tag.as_ref().is_some_and(|x| x != "") { schedule.tag.clone() @@ -385,16 +381,21 @@ pub async fn push_scheduled_job<'c>( JobPayload::ScriptHash { hash, path: schedule.script_path.clone(), - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, cache_ttl, dedicated_worker, language, priority, apply_preprocessor: false, - custom_debounce_key, - debounce_delay_s, + debouncing_settings: DebouncingSettings { + custom_key: custom_debounce_key, + delay_s: debounce_delay_s, + ..Default::default() + }, + concurrency_settings: ConcurrencySettings { + concurrency_key: custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + }, }, if schedule.tag.as_ref().is_some_and(|x| x != "") { schedule.tag.clone() diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 84304f232c..d0f50ec750 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -366,9 +366,7 @@ async fn execute_windmill_tool( language, lock, tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, .. } => { let path = path @@ -379,32 +377,20 @@ async fn execute_windmill_tool( content, language, lock, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, tool_module, tag, tool_module.delete_after_use.unwrap_or(false), ) } - FlowModuleValue::FlowScript { - id, - language, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, - tag, - .. - } => { + FlowModuleValue::FlowScript { id, language, concurrency_settings, tag, .. } => { let path = format!("{}/tools/{}", ctx.job.runnable_path(), tool_module.id); let payload = JobPayloadWithTag { payload: JobPayload::FlowScript { id, language, - custom_concurrency_key: custom_concurrency_key.clone(), - concurrent_limit, - concurrency_time_window_s, + concurrency_settings: concurrency_settings.into(), cache_ttl: tool_module.cache_ttl.map(|x| x as i32), dedicated_worker: None, path, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index dd45d3b786..427951f18b 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -43,7 +43,8 @@ use windmill_common::flow_status::{ }; use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId, StopAfterIf}; use windmill_common::jobs::{ - script_path_to_payload, JobKind, JobPayload, OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE, + script_path_to_payload, ConcurrencySettings, ConcurrencySettingsWithCustom, DebouncingSettings, + JobKind, JobPayload, OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE, }; use windmill_common::scripts::ScriptHash; use windmill_common::users::username_to_permissioned_as; @@ -3956,9 +3957,7 @@ async fn compute_next_flow_transform( language, lock, tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, .. } => { let path = path.unwrap_or_else(|| get_path(flow_job, status, module)); @@ -3968,9 +3967,7 @@ async fn compute_next_flow_transform( content, language, lock, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, module, tag, delete_after_use, @@ -3984,9 +3981,7 @@ async fn compute_next_flow_transform( id, // flow_node(id). tag, language, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, .. } => { let path = get_path(flow_job, status, module); @@ -3995,9 +3990,7 @@ async fn compute_next_flow_transform( payload: JobPayload::FlowScript { id, language, - custom_concurrency_key: custom_concurrency_key.clone(), - concurrent_limit, - concurrency_time_window_s, + concurrency_settings: concurrency_settings.into(), cache_ttl: module.cache_ttl.map(|x| x as i32), dedicated_worker: None, path, @@ -4625,18 +4618,14 @@ async fn payload_from_simple_module( language, lock, tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, .. } => raw_script_to_payload( path.unwrap_or_else(|| inner_path), content, language, lock, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, module, tag, delete_after_use, @@ -4645,20 +4634,16 @@ async fn payload_from_simple_module( id, // flow_node(id). tag, language, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, + concurrency_settings, .. } => JobPayloadWithTag { payload: JobPayload::FlowScript { id, language, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, cache_ttl: module.cache_ttl.map(|x| x as i32), dedicated_worker: None, path: inner_path, + concurrency_settings: concurrency_settings.into(), }, tag, delete_after_use, @@ -4674,9 +4659,7 @@ pub fn raw_script_to_payload( content: String, language: windmill_common::scripts::ScriptLang, lock: Option, - custom_concurrency_key: Option, - concurrent_limit: Option, - concurrency_time_window_s: Option, + concurrency_settings: ConcurrencySettingsWithCustom, module: &FlowModule, tag: Option, delete_after_use: bool, @@ -4688,13 +4671,11 @@ pub fn raw_script_to_payload( content, language, lock, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, cache_ttl: module.cache_ttl.map(|x| x as i32), dedicated_worker: None, - custom_debounce_key: None, - debounce_delay_s: None, + concurrency_settings, + // TODO: Should this have debouncing? + debouncing_settings: DebouncingSettings::default(), }), tag, delete_after_use, @@ -4756,11 +4737,6 @@ pub async fn script_to_payload( let ScriptHashInfo { tag, - concurrency_key, - concurrent_limit, - concurrency_time_window_s, - debounce_key, - debounce_delay_s, cache_ttl, language, dedicated_worker, @@ -4769,6 +4745,11 @@ pub async fn script_to_payload( timeout, on_behalf_of_email, created_by, + concurrency_key, + concurrent_limit, + concurrency_time_window_s, + debounce_key, + debounce_delay_s, .. } = get_script_info_for_hash(None, db, &flow_job.workspace_id, hash.0).await?; let on_behalf_of = if let Some(email) = on_behalf_of_email { @@ -4784,11 +4765,16 @@ pub async fn script_to_payload( JobPayload::ScriptHash { hash, path: script_path, - custom_concurrency_key: concurrency_key, - concurrent_limit, - concurrency_time_window_s, - custom_debounce_key: debounce_key, - debounce_delay_s, + debouncing_settings: DebouncingSettings { + custom_key: debounce_key, + delay_s: debounce_delay_s, + ..Default::default() + }, + concurrency_settings: ConcurrencySettings { + concurrency_key, + concurrent_limit, + concurrency_time_window_s, + }, cache_ttl: module.cache_ttl.map(|x| x as i32).ok_or(cache_ttl).ok(), language, dedicated_worker, diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index b00d15f4eb..56a3c82fe2 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -1171,11 +1171,9 @@ async fn lock_modules<'c>( mut language, input_transforms, tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, is_trigger, assets, + concurrency_settings, } = e.get_value()? else { let mut nmodified_ids = Vec::new(); @@ -1541,11 +1539,9 @@ async fn lock_modules<'c>( content, language, tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, is_trigger, assets, + concurrency_settings, }); new_flow_modules.push(e); @@ -1737,11 +1733,9 @@ async fn reduce_flow<'c>( language, input_transforms, tag, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, is_trigger, assets, + concurrency_settings, .. } = std::mem::replace(&mut val, Identity) else { @@ -1764,11 +1758,9 @@ async fn reduce_flow<'c>( id, tag, language, - custom_concurrency_key, - concurrent_limit, - concurrency_time_window_s, is_trigger, assets, + concurrency_settings, }; } ForloopFlow { modules, modules_node, .. } diff --git a/backend/windmill-worker/src/worker_utils.rs b/backend/windmill-worker/src/worker_utils.rs index f105d8c415..e92092480b 100644 --- a/backend/windmill-worker/src/worker_utils.rs +++ b/backend/windmill-worker/src/worker_utils.rs @@ -395,15 +395,19 @@ pub async fn get_tag_and_concurrency(job_id: &Uuid, db: &DB) -> Option cache::flow::fetch_version(db, version).await, }; let flow_value = flow.map(|f| f.value().clone()).ok(); + let concurrency_key = flow_value .as_ref() - .map(|fv| fv.concurrency_key.clone()) - .flatten(); - let concurrent_limit = flow_value.as_ref().map(|fv| fv.concurrent_limit).flatten(); + .and_then(|fv| fv.concurrency_settings.concurrency_key.to_owned()); + + let concurrent_limit = flow_value + .as_ref() + .and_then(|fv| fv.concurrency_settings.concurrent_limit); + let concurrent_time_window_s = flow_value .as_ref() - .map(|fv| fv.concurrency_time_window_s) - .flatten(); + .and_then(|fv| fv.concurrency_settings.concurrency_time_window_s); + Some(TagAndConcurrencyKey { tag: tag_and_concurrency_key.tag, concurrency_key,