From 999aaaacd8727f2a8afdac2007e6536c66f1c2f7 Mon Sep 17 00:00:00 2001 From: Pyra <92104930+pyranota@users.noreply.github.com> Date: Fri, 24 Oct 2025 17:24:46 +0200 Subject: [PATCH] fix(debouncing): fix perf issues and re-enable debouncing (#6932) * fix(debouncing): fix perf issues and re-enable debouncing Signed-off-by: pyranota * add debounce data in clone_script Signed-off-by: pyranota * update sqlx cache Signed-off-by: pyranota --------- Signed-off-by: pyranota --- ...576154cec84a09069286d2d7144dbad54f4d6.json | 16 +++++++ ...153c43903f929ae5d62fbba12610f89c36d55.json | 2 +- ...53_job_debouncing_index_by_job_id.down.sql | 2 + ...1453_job_debouncing_index_by_job_id.up.sql | 2 + backend/tests/relative_imports.rs | 43 +++++++++++++++++++ backend/windmill-common/src/scripts.rs | 4 +- backend/windmill-common/src/worker.rs | 4 ++ backend/windmill-queue/src/jobs.rs | 12 +----- 8 files changed, 72 insertions(+), 13 deletions(-) create mode 100644 backend/.sqlx/query-53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6.json create mode 100644 backend/migrations/20251024141453_job_debouncing_index_by_job_id.down.sql create mode 100644 backend/migrations/20251024141453_job_debouncing_index_by_job_id.up.sql diff --git a/backend/.sqlx/query-53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6.json b/backend/.sqlx/query-53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6.json new file mode 100644 index 0000000000..902ed458c2 --- /dev/null +++ b/backend/.sqlx/query-53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO script\n (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s)\n\n SELECT workspace_id, $1, path, array_prepend($2::bigint, COALESCE(parent_hashes, '{}'::bigint[])), summary, description, content, created_by, schema, is_template, extra_perms, NULL, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s\n\n FROM script WHERE hash = $2 AND workspace_id = $3;\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int8", + "Int8", + "Text" + ] + }, + "nullable": [] + }, + "hash": "53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6" +} diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/migrations/20251024141453_job_debouncing_index_by_job_id.down.sql b/backend/migrations/20251024141453_job_debouncing_index_by_job_id.down.sql new file mode 100644 index 0000000000..719fdc0390 --- /dev/null +++ b/backend/migrations/20251024141453_job_debouncing_index_by_job_id.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +DROP INDEX IF EXISTS idx_debounce_key_job_id; diff --git a/backend/migrations/20251024141453_job_debouncing_index_by_job_id.up.sql b/backend/migrations/20251024141453_job_debouncing_index_by_job_id.up.sql new file mode 100644 index 0000000000..baaa49400e --- /dev/null +++ b/backend/migrations/20251024141453_job_debouncing_index_by_job_id.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +CREATE INDEX IF NOT EXISTS idx_debounce_key_job_id ON debounce_key (job_id); diff --git a/backend/tests/relative_imports.rs b/backend/tests/relative_imports.rs index 78484d0879..0403365850 100644 --- a/backend/tests/relative_imports.rs +++ b/backend/tests/relative_imports.rs @@ -577,6 +577,12 @@ def main(): async fn test_1(db: sqlx::Pool) -> anyhow::Result<()> { // This tests if debouncing and consolidation works. // Also makes sures that dependency job does not create new flow version + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } let (client, port, _s) = init_client(db.clone()).await; let mut completed = listen_for_completed_jobs(&db).await; @@ -792,6 +798,12 @@ def main(): async fn test_left(db: sqlx::Pool) -> anyhow::Result<()> { use crate::common::RunJob; + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } // TODO: We don't care about timer. If there is no timer, it will be set automatically for djobs?? let (_client, port, _s) = init_client(db.clone()).await; let mut completed = listen_for_completed_jobs(&db).await; @@ -1114,6 +1126,12 @@ def main(): // #[windmill::all_min_versions] async fn test_3(db: sqlx::Pool) -> anyhow::Result<()> { // This tests checks if concurrency limit works correcly and there is no race conditions. + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } let (_client, port, _s) = init_client(db.clone()).await; let mut completed = listen_for_completed_jobs(&db).await; @@ -1441,6 +1459,12 @@ WHERE #[cfg(feature = "python")] #[sqlx::test(fixtures("base", "djob_debouncing"))] async fn test_1(db: sqlx::Pool) -> anyhow::Result<()> { + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } // This tests if debouncing and consolidation works. // Also makes sures that dependency job does not create new flow version @@ -1620,6 +1644,12 @@ WHERE #[sqlx::test(fixtures("base", "djob_debouncing"))] async fn test_left(db: sqlx::Pool) -> anyhow::Result<()> { use crate::common::RunJob; + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } // TODO: We don't care about timer. If there is no timer, it will be set automatically for djobs?? let (_client, port, _s) = init_client(db.clone()).await; @@ -1929,6 +1959,12 @@ WHERE #[sqlx::test(fixtures("base", "djob_debouncing"))] // TODO: Same test_but script fails. async fn test_1(db: sqlx::Pool) -> anyhow::Result<()> { + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } // This tests if debouncing and consolidation works. // Also makes sures that dependency job does not create new flow version let (client, port, _s) = init_client(db.clone()).await; @@ -2172,6 +2208,12 @@ WHERE #[sqlx::test(fixtures("base", "djob_debouncing"))] async fn test_left(db: sqlx::Pool) -> anyhow::Result<()> { use crate::common::RunJob; + { + let mut mvsd = windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING + .write() + .await; + *mvsd = true; + } // TODO: We don't care about timer. If there is no timer, it will be set automatically for djobs?? let (_client, port, _s) = init_client(db.clone()).await; @@ -2281,6 +2323,7 @@ WHERE .await .unwrap(); + dbg!(&r); assert_eq!(r.len(), 4); assert!(r.contains(&Some(-221349019907577876))); assert!(r.contains(&Some(533400))); diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index b0e771d138..b6a4fd3391 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -799,14 +799,14 @@ pub async fn clone_script<'c>( draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \ dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \ delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, \ - codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets) + codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s) SELECT workspace_id, $1, path, array_prepend($2::bigint, COALESCE(parent_hashes, '{}'::bigint[])), summary, description, \ content, created_by, schema, is_template, extra_perms, NULL, language, kind, tag, \ draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \ dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \ delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, \ - codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets + codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s FROM script WHERE hash = $2 AND workspace_id = $3; ", new_hash, base_hash.0, w_id).execute(&mut **tx).await?; diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 2a346ea2b3..94ef70fed9 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -409,6 +409,10 @@ fn format_pull_query(peek: String) -> String { raw_flow, script_entrypoint_override, preprocessed FROM v2_job WHERE id = (SELECT id FROM peek) + ), delete_debounce AS NOT MATERIALIZED ( + DELETE FROM debounce_key + USING j + WHERE j.kind::text != 'flowdependencies' AND j.kind::text != 'appdependencies' AND j.kind::text != 'dependencies' AND debounce_key.job_id = j.id ) SELECT j.id, j.workspace_id, j.parent_job, j.created_by, started_at, scheduled_for, j.runnable_id, j.runnable_path, j.args, canceled_by, canceled_reason, j.kind, j.trigger, j.trigger_kind, j.permissioned_as, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 6112045b73..acf0b5df76 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3622,8 +3622,8 @@ pub async fn push<'c, 'd>( cache_ttl, dedicated_worker, _low_level_priority, - mut custom_debounce_key, - mut debounce_delay_s, + custom_debounce_key, + debounce_delay_s, ) = match job_payload { JobPayload::ScriptHash { hash, @@ -4391,14 +4391,6 @@ pub async fn push<'c, 'd>( ), }; - if custom_debounce_key.is_some() { - tracing::warn!("debouncing has been disabled temporarily, ignoring debounce_key"); - custom_debounce_key = None; - } - if debounce_delay_s.is_some() { - tracing::warn!("debouncing has been disabled temporarily, ignoring debounce_delay_s"); - debounce_delay_s = None; - } // Enforce concurrency limit on all dependency jobs. // TODO: We can ignore this for scripts djobs. The main reason we need all djobs to be sequential is because we have // nodes_to_relock and we need all locks whose corresponding steps aren't in nodes_to_relock be already present.