starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(debouncing): fix perf issues and re-enable debouncing (#6932)

Browse Source 
* fix(debouncing): fix perf issues and re-enable debouncing

Signed-off-by: pyranota <pyra@duck.com>

* add debounce data in clone_script

Signed-off-by: pyranota <pyra@duck.com>

* update sqlx cache

Signed-off-by: pyranota <pyra@duck.com>

---------

Signed-off-by: pyranota <pyra@duck.com>
This commit is contained in:
Pyra
2025-10-24 17:24:46 +02:00
committed by GitHub
parent 583b870983
commit 999aaaacd8
8 changed files with 72 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-53dee7c119d724624b9973ee981576154cec84a09069286d2d7144dbad54f4d6.json
Generated
+16
View File
@@ -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"
}
backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json
Generated
+1 -1
View File
@@ -15,7 +15,7 @@
]
},
"nullable": [
null
true
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
backend/migrations/20251024141453_job_debouncing_index_by_job_id.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add down migration script here
DROP INDEX IF EXISTS idx_debounce_key_job_id;
backend/migrations/20251024141453_job_debouncing_index_by_job_id.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
CREATE INDEX IF NOT EXISTS idx_debounce_key_job_id ON debounce_key (job_id);
backend/tests/relative_imports.rs
+43
View File
@@ -577,6 +577,12 @@ def main():
async fn test_1(db: sqlx::Pool<sqlx::Postgres>) -> 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<sqlx::Postgres>) -> 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<sqlx::Postgres>) -> 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<sqlx::Postgres>) -> 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<sqlx::Postgres>) -> 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<sqlx::Postgres>) -> 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<sqlx::Postgres>) -> 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)));
backend/windmill-common/src/scripts.rs
+2 -2
View File
@@ -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?;
backend/windmill-common/src/worker.rs
+4
View File
@@ -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,
backend/windmill-queue/src/jobs.rs
+2 -10
View File
@@ -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.