diff --git a/.github/workflows/backend-test.yml b/.github/workflows/backend-test.yml index 029f4c114e..c4f0576313 100644 --- a/.github/workflows/backend-test.yml +++ b/.github/workflows/backend-test.yml @@ -85,6 +85,7 @@ jobs: CARGO_NET_GIT_FETCH_WITH_CLI: true WMDEBUG_FORCE_V0_WORKSPACE_DEPENDENCIES: 1 WMDEBUG_FORCE_RUNNABLE_SETTINGS_V0: 1 + WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT: 1 run: | deno --version && bun -v && go version && python3 --version cd windmill-duckdb-ffi-internal && ./build_dev.sh && cd .. diff --git a/backend/.sqlx/query-16c96166ffa6b9aec65c6072b204b52b87e3c2f3d76e47eb173fc78721355066.json b/backend/.sqlx/query-16c96166ffa6b9aec65c6072b204b52b87e3c2f3d76e47eb173fc78721355066.json new file mode 100644 index 0000000000..7fd747c125 --- /dev/null +++ b/backend/.sqlx/query-16c96166ffa6b9aec65c6072b204b52b87e3c2f3d76e47eb173fc78721355066.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH _ AS (\n UPDATE debounce_key\n SET debounced_times = 0, -- reset debounced_times\n first_started_at = now(), -- rest\n previous_job_id = NULL\n WHERE job_id = $1\n )\n UPDATE v2_job_debounce_batch \n SET debounce_batch = nextval('debounce_batch_seq') -- move to new batch\n WHERE id = $1\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "16c96166ffa6b9aec65c6072b204b52b87e3c2f3d76e47eb173fc78721355066" +} diff --git a/backend/.sqlx/query-3d38720e807b379645d8f3ab61c6a968143d42c3014152608f7d1b252cd8085c.json b/backend/.sqlx/query-3d38720e807b379645d8f3ab61c6a968143d42c3014152608f7d1b252cd8085c.json new file mode 100644 index 0000000000..a97d16b89c --- /dev/null +++ b/backend/.sqlx/query-3d38720e807b379645d8f3ab61c6a968143d42c3014152608f7d1b252cd8085c.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO app_version\n (app_id, value, created_by, raw_app)\n SELECT app_id, value, created_by, raw_app\n FROM app_version WHERE id = $1\n RETURNING id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "3d38720e807b379645d8f3ab61c6a968143d42c3014152608f7d1b252cd8085c" +} diff --git a/backend/.sqlx/query-44bf04fb504cd1b708657f01c8fb26e81e78807315226557f08449bf43982abb.json b/backend/.sqlx/query-44bf04fb504cd1b708657f01c8fb26e81e78807315226557f08449bf43982abb.json deleted file mode 100644 index 9f0e810e97..0000000000 --- a/backend/.sqlx/query-44bf04fb504cd1b708657f01c8fb26e81e78807315226557f08449bf43982abb.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO debounce_key (key, job_id) VALUES ($1, $2) ON CONFLICT (key) DO UPDATE SET job_id = EXCLUDED.job_id", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "44bf04fb504cd1b708657f01c8fb26e81e78807315226557f08449bf43982abb" -} diff --git a/backend/.sqlx/query-454ace9ce391725ef4f4c129cd66e4c12a5c40f512b70551958178c8b4d6c183.json b/backend/.sqlx/query-454ace9ce391725ef4f4c129cd66e4c12a5c40f512b70551958178c8b4d6c183.json new file mode 100644 index 0000000000..14a6ab4a40 --- /dev/null +++ b/backend/.sqlx/query-454ace9ce391725ef4f4c129cd66e4c12a5c40f512b70551958178c8b4d6c183.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO debounce_key (job_id, key)\n VALUES ($1, $2)\n ON CONFLICT (key)\n DO UPDATE SET\n previous_job_id = debounce_key.job_id,\n job_id = EXCLUDED.job_id, -- replace current job with new one \n debounced_times = debounce_key.debounced_times + 1 -- evaluated only if conflict,\n -- conflict means there is already existing value,\n -- which means overriding it will also imply adding new entry to v2_job_debounce_batch and thus debouncing the job\n -- so the counter should be incremented\n RETURNING\n debounced_times,\n first_started_at,\n previous_job_id AS job_id_to_debounce\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "debounced_times", + "type_info": "Int4" + }, + { + "ordinal": 1, + "name": "first_started_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 2, + "name": "job_id_to_debounce", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Varchar" + ] + }, + "nullable": [ + false, + false, + true + ] + }, + "hash": "454ace9ce391725ef4f4c129cd66e4c12a5c40f512b70551958178c8b4d6c183" +} diff --git a/backend/.sqlx/query-4f11760f5d283728ded5533e6a0b51f49fdb7c11bf7d47b8536d607e646bd265.json b/backend/.sqlx/query-4f11760f5d283728ded5533e6a0b51f49fdb7c11bf7d47b8536d607e646bd265.json deleted file mode 100644 index 40c82a50aa..0000000000 --- a/backend/.sqlx/query-4f11760f5d283728ded5533e6a0b51f49fdb7c11bf7d47b8536d607e646bd265.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT job_id FROM debounce_key WHERE key = $1 AND job_id IN (SELECT id FROM v2_job_queue) FOR UPDATE", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "job_id", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "4f11760f5d283728ded5533e6a0b51f49fdb7c11bf7d47b8536d607e646bd265" -} diff --git a/backend/.sqlx/query-61b37cb4db6e60c2d35f7d23db5afbe04e040a8dcd1d93afaaaa320665c8779a.json b/backend/.sqlx/query-61b37cb4db6e60c2d35f7d23db5afbe04e040a8dcd1d93afaaaa320665c8779a.json new file mode 100644 index 0000000000..64fb5a6673 --- /dev/null +++ b/backend/.sqlx/query-61b37cb4db6e60c2d35f7d23db5afbe04e040a8dcd1d93afaaaa320665c8779a.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH prev_sd AS (\n DELETE FROM debounce_stale_data WHERE job_id = $1 RETURNING to_relock\n ) INSERT INTO debounce_stale_data (job_id, to_relock)\n VALUES ($2, array_cat((SELECT to_relock FROM prev_sd), $3))\n ON CONFLICT (job_id) DO UPDATE SET to_relock = EXCLUDED.to_relock\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Uuid", + "TextArray" + ] + }, + "nullable": [] + }, + "hash": "61b37cb4db6e60c2d35f7d23db5afbe04e040a8dcd1d93afaaaa320665c8779a" +} diff --git a/backend/.sqlx/query-76774e6f72c8c8b7473487e4176dc17b17372b7292e39d3888a93ff4fe49e4f5.json b/backend/.sqlx/query-76774e6f72c8c8b7473487e4176dc17b17372b7292e39d3888a93ff4fe49e4f5.json deleted file mode 100644 index 694ed1887f..0000000000 --- a/backend/.sqlx/query-76774e6f72c8c8b7473487e4176dc17b17372b7292e39d3888a93ff4fe49e4f5.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM debounce_key WHERE key = $1", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [] - }, - "hash": "76774e6f72c8c8b7473487e4176dc17b17372b7292e39d3888a93ff4fe49e4f5" -} diff --git a/backend/.sqlx/query-7ec724b84479c2f737637e91b8cbed6cae29f361167deee879b8b683ad1bf684.json b/backend/.sqlx/query-7ec724b84479c2f737637e91b8cbed6cae29f361167deee879b8b683ad1bf684.json deleted file mode 100644 index a979af6114..0000000000 --- a/backend/.sqlx/query-7ec724b84479c2f737637e91b8cbed6cae29f361167deee879b8b683ad1bf684.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO debounce_stale_data (job_id, to_relock)\n VALUES ($1, $2)\n ON CONFLICT (job_id)\n DO UPDATE SET to_relock = (\n SELECT array_agg(DISTINCT x)\n FROM unnest(\n -- Combine existing array with new values, removing duplicates\n array_cat(debounce_stale_data.to_relock, EXCLUDED.to_relock)\n ) AS x\n )\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid", - "TextArray" - ] - }, - "nullable": [] - }, - "hash": "7ec724b84479c2f737637e91b8cbed6cae29f361167deee879b8b683ad1bf684" -} diff --git a/backend/.sqlx/query-83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604.json b/backend/.sqlx/query-83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604.json deleted file mode 100644 index 27a5df6de9..0000000000 --- a/backend/.sqlx/query-83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO app_version\n (app_id, value, created_by, raw_app)\n SELECT app_id, value, created_by, raw_app\n FROM app_version WHERE id = $1\n RETURNING id", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [ - false - ] - }, - "hash": "83232f2db5eb1b6fef744998e60420ef920d472286cf4c1f78452446a4bcb604" -} diff --git a/backend/.sqlx/query-8360ab72d60f07dde6ecae599e6531b5b86862029ab51fdbdd44ec16239108e2.json b/backend/.sqlx/query-8360ab72d60f07dde6ecae599e6531b5b86862029ab51fdbdd44ec16239108e2.json new file mode 100644 index 0000000000..70a904cfc3 --- /dev/null +++ b/backend/.sqlx/query-8360ab72d60f07dde6ecae599e6531b5b86862029ab51fdbdd44ec16239108e2.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO v2_job_debounce_batch (id, debounce_batch)\n -- if it the first one, nextval will be evaluated, otherwise take from the job we will debounce\n SELECT\n $2,\n COALESCE(\n (\n SELECT debounce_batch\n FROM v2_job_debounce_batch\n WHERE id = $1\n LIMIT 1\n ), -- maybe use current batch\n nextval('debounce_batch_seq')\n )\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "8360ab72d60f07dde6ecae599e6531b5b86862029ab51fdbdd44ec16239108e2" +} diff --git a/backend/.sqlx/query-a6a973dcd92d2e40fd9a1c1be42052fcd350bd47ee4f63832448b6e6f0f472f0.json b/backend/.sqlx/query-a6a973dcd92d2e40fd9a1c1be42052fcd350bd47ee4f63832448b6e6f0f472f0.json new file mode 100644 index 0000000000..1c396d38fd --- /dev/null +++ b/backend/.sqlx/query-a6a973dcd92d2e40fd9a1c1be42052fcd350bd47ee4f63832448b6e6f0f472f0.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO flow_version\n (workspace_id, path, value, schema, created_by)\n\n SELECT workspace_id, path, value, schema, created_by\n FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3\n\n RETURNING id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "a6a973dcd92d2e40fd9a1c1be42052fcd350bd47ee4f63832448b6e6f0f472f0" +} diff --git a/backend/.sqlx/query-cc3aadd61539cfa349e65f37d04c3754d88fa8d651d8cbbd95aadfaced0c0a22.json b/backend/.sqlx/query-cc3aadd61539cfa349e65f37d04c3754d88fa8d651d8cbbd95aadfaced0c0a22.json new file mode 100644 index 0000000000..f05650f0b1 --- /dev/null +++ b/backend/.sqlx/query-cc3aadd61539cfa349e65f37d04c3754d88fa8d651d8cbbd95aadfaced0c0a22.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "\nDELETE FROM debounce_key\nWHERE job_id IN (SELECT id FROM v2_job_completed)\nRETURNING key,job_id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "key", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "job_id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + false + ] + }, + "hash": "cc3aadd61539cfa349e65f37d04c3754d88fa8d651d8cbbd95aadfaced0c0a22" +} diff --git a/backend/.sqlx/query-cfe06702916362aaf5122bb95593eff389e0d44b7a58b69fd5c79629599902fc.json b/backend/.sqlx/query-cfe06702916362aaf5122bb95593eff389e0d44b7a58b69fd5c79629599902fc.json deleted file mode 100644 index e9cedd467d..0000000000 --- a/backend/.sqlx/query-cfe06702916362aaf5122bb95593eff389e0d44b7a58b69fd5c79629599902fc.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM debounce_stale_data WHERE job_id = $1 RETURNING to_relock", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "to_relock", - "type_info": "TextArray" - } - ], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [ - true - ] - }, - "hash": "cfe06702916362aaf5122bb95593eff389e0d44b7a58b69fd5c79629599902fc" -} diff --git a/backend/.sqlx/query-d0e826043e5a129ae6768c274c67b6254ff6c5fd450ecdab886a3183a894d266.json b/backend/.sqlx/query-d0e826043e5a129ae6768c274c67b6254ff6c5fd450ecdab886a3183a894d266.json new file mode 100644 index 0000000000..e900b0f9e0 --- /dev/null +++ b/backend/.sqlx/query-d0e826043e5a129ae6768c274c67b6254ff6c5fd450ecdab886a3183a894d266.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH ids AS (\n SELECT id as job_id FROM v2_job_debounce_batch WHERE debounce_batch = (\n SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1\n )\n ) SELECT args->>$2 FROM ids LEFT JOIN v2_job ON v2_job.id = ids.job_id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "d0e826043e5a129ae6768c274c67b6254ff6c5fd450ecdab886a3183a894d266" +} diff --git a/backend/.sqlx/query-e65c79d792f0e8285ea9acac54bc569f22ca7c28b205533e8ba73722bf438c94.json b/backend/.sqlx/query-e65c79d792f0e8285ea9acac54bc569f22ca7c28b205533e8ba73722bf438c94.json new file mode 100644 index 0000000000..940d95133e --- /dev/null +++ b/backend/.sqlx/query-e65c79d792f0e8285ea9acac54bc569f22ca7c28b205533e8ba73722bf438c94.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH _ AS (\n DELETE FROM debounce_key WHERE job_id = $1\n ) SELECT status = 'skipped' FROM v2_job_completed WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "e65c79d792f0e8285ea9acac54bc569f22ca7c28b205533e8ba73722bf438c94" +} diff --git a/backend/.sqlx/query-f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2.json b/backend/.sqlx/query-f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2.json deleted file mode 100644 index 32616f298f..0000000000 --- a/backend/.sqlx/query-f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO flow_version\n (workspace_id, path, value, schema, created_by)\n\n SELECT workspace_id, path, value, schema, created_by\n FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3\n\n RETURNING id\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Int8" - ] - }, - "nullable": [ - false - ] - }, - "hash": "f0efa383f2025158de160577ad839ae72faf0c8fe097e6ad6d309aee9a8aede2" -} diff --git a/backend/.sqlx/query-f1fe8508f6cd29d4c0d354473529fc46f506ddf98d45fc870e2855c4a4b424ea.json b/backend/.sqlx/query-f1fe8508f6cd29d4c0d354473529fc46f506ddf98d45fc870e2855c4a4b424ea.json deleted file mode 100644 index effa800640..0000000000 --- a/backend/.sqlx/query-f1fe8508f6cd29d4c0d354473529fc46f506ddf98d45fc870e2855c4a4b424ea.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO debounce_key (key, job_id)\n VALUES ($1, $2)\n ON CONFLICT (key)\n DO UPDATE SET job_id = debounce_key.job_id -- No actual change, just to trigger UPDATE\n RETURNING CASE WHEN xmax != 0 THEN job_id ELSE NULL END AS job_id", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "job_id", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Varchar", - "Uuid" - ] - }, - "nullable": [ - null - ] - }, - "hash": "f1fe8508f6cd29d4c0d354473529fc46f506ddf98d45fc870e2855c4a4b424ea" -} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 9d92c60fc2..490ee34e88 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -576abf6519d1aa12a2b989a58a123501206284fb +71d872d27de44227b82f75a9c0879d8428684880 diff --git a/backend/migrations/20251216114001_v2_job_debounce_data.down.sql b/backend/migrations/20251216114001_v2_job_debounce_data.down.sql new file mode 100644 index 0000000000..800dbd6f2d --- /dev/null +++ b/backend/migrations/20251216114001_v2_job_debounce_data.down.sql @@ -0,0 +1,6 @@ +ALTER TABLE debounce_key DROP COLUMN IF EXISTS debounced_times; +ALTER TABLE debounce_key DROP COLUMN IF EXISTS first_started_at; +ALTER TABLE debounce_key DROP COLUMN IF EXISTS previous_job_id; +DROP INDEX IF EXISTS idx_v2_job_debounce_batch_debounce_batch; +DROP TABLE IF EXISTS v2_job_debounce_batch; +DROP SEQUENCE IF EXISTS debounce_batch_seq; diff --git a/backend/migrations/20251216114001_v2_job_debounce_data.up.sql b/backend/migrations/20251216114001_v2_job_debounce_data.up.sql new file mode 100644 index 0000000000..d7600b1e1b --- /dev/null +++ b/backend/migrations/20251216114001_v2_job_debounce_data.up.sql @@ -0,0 +1,13 @@ +CREATE SEQUENCE debounce_batch_seq START 1; + +CREATE TABLE IF NOT EXISTS v2_job_debounce_batch( + id UUID PRIMARY KEY, + debounce_batch BIGINT NOT NULL DEFAULT nextval('debounce_batch_seq') +); + +CREATE INDEX IF NOT EXISTS idx_v2_job_debounce_batch_debounce_batch ON v2_job_debounce_batch(debounce_batch); + +ALTER TABLE debounce_key ADD COLUMN IF NOT EXISTS previous_job_id UUID; +ALTER TABLE debounce_key ADD COLUMN IF NOT EXISTS first_started_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(); +ALTER TABLE debounce_key ADD COLUMN IF NOT EXISTS debounced_times INTEGER NOT NULL DEFAULT 0; + diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index facdacf849..8da90fdbde 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1606,6 +1606,17 @@ pub async fn monitor_db( } }; + // run every 30s (every iteration) + let cleanup_debounce_keys_completed_f = async { + if server_mode && !initial_load { + if let Some(db) = conn.as_sql() { + if let Err(e) = cleanup_debounce_keys_for_completed_jobs(&db).await { + tracing::error!("Error cleaning up debounce keys for completed jobs: {:?}", e); + } + } + } + }; + // run every hour (60 minutes / 30 seconds = 120) let cleanup_worker_group_stats_f = async { if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(120) { @@ -1726,6 +1737,7 @@ pub async fn monitor_db( cleanup_concurrency_counters_f, cleanup_concurrency_counters_empty_keys_f, cleanup_debounce_keys_f, + cleanup_debounce_keys_completed_f, cleanup_worker_group_stats_f, ); } @@ -2791,3 +2803,33 @@ RETURNING key,job_id } Ok(()) } + +async fn cleanup_debounce_keys_for_completed_jobs(db: &DB) -> error::Result<()> { + // If min version doesn't support runnable settings, clean up debounce keys for completed jobs + if !*windmill_common::worker::MIN_VERSION_SUPPORTS_RUNNABLE_SETTINGS_V0.read().await { + let result = sqlx::query!( + " +DELETE FROM debounce_key +WHERE job_id IN (SELECT id FROM v2_job_completed) +RETURNING key,job_id + ", + ) + .fetch_all(db) + .await?; + + if result.len() > 0 { + tracing::warn!( + "Cleaned up {} debounce keys for completed jobs (runnable settings v0 not supported by all workers)", + result.len() + ); + for row in result { + tracing::debug!( + "Debounce key for completed job cleaned up: key: {}, job_id: {:?}", + row.key, + row.job_id + ); + } + } + } + Ok(()) +} diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 8966c95f7f..f6e2557a7c 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -143,13 +143,12 @@ impl ApiServer { pub struct RunJob { pub payload: JobPayload, pub args: serde_json::Map, - pub debounce_job_id_o: Option, pub scheduled_for_o: Option>, } impl From for RunJob { fn from(payload: JobPayload) -> Self { - Self { payload, args: Default::default(), debounce_job_id_o: None, scheduled_for_o: None } + Self { payload, args: Default::default(), scheduled_for_o: None } } } @@ -159,11 +158,6 @@ impl RunJob { self } - pub fn push_arg_debounce_job_id_o(mut self, job_id: Option) -> Self { - self.debounce_job_id_o = job_id; - self - } - pub fn push_arg_scheduled_for_o( mut self, scheduled_for_o: Option>, @@ -173,7 +167,7 @@ impl RunJob { } pub async fn push(self, db: &Pool) -> Uuid { - let RunJob { payload, args, debounce_job_id_o, scheduled_for_o } = self; + let RunJob { payload, args, scheduled_for_o } = self; let mut hm_args = std::collections::HashMap::new(); for (k, v) in args { hm_args.insert(k, windmill_common::worker::to_raw_value(&v)); @@ -207,7 +201,6 @@ impl RunJob { None, false, None, - debounce_job_id_o, None, None, ) @@ -591,6 +584,7 @@ pub async fn assert_lockfile( hash: ScriptHash(script.hash), dedicated_worker: None, language, + debouncing_settings: Default::default(), }) .push(&db2) .await; diff --git a/backend/tests/job_payload.rs b/backend/tests/job_payload.rs index ed1413439c..ae492363af 100644 --- a/backend/tests/job_payload.rs +++ b/backend/tests/job_payload.rs @@ -131,6 +131,7 @@ mod job_payload { path: "f/system/hello_with_nodes_flow".to_string(), dedicated_worker: None, version: 1443253234253454, + debouncing_settings: Default::default(), }) .run_until_complete(&db, false, port) .await @@ -221,6 +222,7 @@ mod job_payload { path: "f/system/hello_with_nodes_flow".to_string(), dedicated_worker: None, version: 1443253234253454, + debouncing_settings: Default::default(), }) .run_until_complete(&db, false, port) .await @@ -263,6 +265,7 @@ mod job_payload { path: "f/system/hello".to_string(), hash: ScriptHash(123412), language: ScriptLang::Deno, + debouncing_settings: Default::default(), dedicated_worker: None, }) .run_until_complete(&db, false, port) @@ -308,6 +311,7 @@ mod job_payload { path: "f/system/hello_with_nodes_flow".to_string(), dedicated_worker: None, version: 1443253234253454, + debouncing_settings: Default::default(), }) .run_until_complete(&db, false, port) .await @@ -449,6 +453,7 @@ mod job_payload { path: "f/system/hello_with_nodes_flow".to_string(), dedicated_worker: None, version: 1443253234253454, + debouncing_settings: Default::default(), }) .run_until_complete(&db, false, port) .await @@ -521,6 +526,7 @@ mod job_payload { path: "f/system/hello_with_preprocessor".to_string(), dedicated_worker: None, version: 1443253234253456, + debouncing_settings: Default::default(), }) .run_until_complete(db, false, port) .await @@ -577,6 +583,7 @@ mod job_payload { path: "f/system/hello_with_nodes_flow".to_string(), dedicated_worker: None, version: 1443253234253454, + debouncing_settings: Default::default(), }) .run_until_complete(&db, false, port) .await diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 3bdcef1b0f..9eeaa28827 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -16555,6 +16555,14 @@ components: type: string debounce_delay_s: type: integer + debounce_args_to_accumulate: + type: array + items: + type: string + max_total_debouncing_time: + type: integer + max_total_debounces_amount: + type: integer cache_ttl: type: number dedicated_worker: @@ -16658,6 +16666,14 @@ components: type: string debounce_delay_s: type: integer + debounce_args_to_accumulate: + type: array + items: + type: string + max_total_debouncing_time: + type: integer + max_total_debounces_amount: + type: integer visible_to_runner_only: type: boolean no_main_func: diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 77e7763b01..785b7e9962 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -61,7 +61,7 @@ use windmill_common::{ users::username_to_permissioned_as, utils::{ http_get_from_hub, not_found_if_none, paginate, query_elems_from_hub, require_admin, - Pagination, RunnableKind, StripPath, WarnAfterExt, + Pagination, RunnableKind, StripPath, }, variables::{build_crypt, build_crypt_with_key_suffix, encrypt}, worker::{to_raw_value, CLOUD_HOSTED}, @@ -1222,7 +1222,11 @@ async fn create_app_internal<'a>( &db, tx, w_id, - JobPayload::AppDependencies { path: app.path.clone(), version: v_id }, + JobPayload::AppDependencies { + path: app.path.clone(), + version: v_id, + debouncing_settings: Default::default(), + }, PushArgs { args: &args, extra: None }, &authed.username, &authed.email, @@ -1247,7 +1251,6 @@ async fn create_app_internal<'a>( None, None, None, - None, ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -1531,14 +1534,6 @@ async fn update_app_internal<'a>( path.to_owned() }; let v_id = if let Some(nvalue) = &ns.value { - // Row lock debounce key for path. We need this to make all updates of runnables sequential and predictable. - tokio::time::timeout( - core::time::Duration::from_secs(60), - windmill_common::jobs::lock_debounce_key(&w_id, &npath, &mut tx), - ) - .warn_after_seconds(10) - .await??; - let app_id = sqlx::query_scalar!( "SELECT id FROM app WHERE path = $1 AND workspace_id = $2", npath, @@ -1613,7 +1608,11 @@ async fn update_app_internal<'a>( &db, tx, w_id, - JobPayload::AppDependencies { path: npath.clone(), version: v_id }, + JobPayload::AppDependencies { + path: npath.clone(), + version: v_id, + debouncing_settings: Default::default(), + }, PushArgs { args: &args, extra: None }, &authed.username, &authed.email, @@ -1638,7 +1637,6 @@ async fn update_app_internal<'a>( None, None, None, - None, ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); @@ -1971,7 +1969,6 @@ async fn execute_component( end_user_email, None, None, - None, ) .await?; diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index c68d21a6bf..c4cb613992 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -33,8 +33,10 @@ use sqlx::{FromRow, Postgres, Transaction}; use windmill_audit::audit_oss::audit_log; use windmill_audit::ActionKind; use windmill_common::runnable_settings::RunnableSettingsTrait; -use windmill_common::utils::{query_elems_from_hub, WarnAfterExt}; -use windmill_common::worker::{to_raw_value, CLOUD_HOSTED, MIN_VERSION_SUPPORTS_DEBOUNCING}; +use windmill_common::utils::query_elems_from_hub; +use windmill_common::worker::{ + to_raw_value, CLOUD_HOSTED, MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2, +}; use windmill_common::HUB_BASE_URL; use windmill_common::{ db::UserDB, @@ -46,6 +48,7 @@ use windmill_common::{ utils::{http_get_from_hub, not_found_if_none, paginate, Pagination, RunnableKind, StripPath}, }; use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; +use windmill_queue::WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT; use windmill_queue::{push, schedule::push_scheduled_job, PushIsolationLevel}; use windmill_worker::scoped_dependency_map::ScopedDependencyMap; @@ -541,6 +544,7 @@ async fn create_flow( path: nf.path.clone(), dedicated_worker: nf.dedicated_worker, version: version, + debouncing_settings: Default::default(), }, windmill_queue::PushArgs { args: &args, extra: None }, &authed.username, @@ -566,7 +570,6 @@ async fn create_flow( None, None, None, - None, ) .await?; @@ -968,14 +971,6 @@ async fn update_flow( .await?; } - // Row lock debounce key for path. We need this to make all updates of runnables sequential and predictable. - tokio::time::timeout( - core::time::Duration::from_secs(60), - windmill_common::jobs::lock_debounce_key(&w_id, &nf.path, &mut tx), - ) - .warn_after_seconds(10) - .await??; - // tracing::error!("Updating flow: {:?}", nf.value.get()); // This will lock anyone who is trying to iterate on flow_versions with given path and parameters. @@ -1087,6 +1082,7 @@ async fn update_flow( path: nf.path.clone(), dedicated_worker: nf.dedicated_worker, version, + debouncing_settings: Default::default(), }, windmill_queue::PushArgs { args: &args, extra: None }, &authed.username, @@ -1112,7 +1108,6 @@ async fn update_flow( None, None, None, - None, ) .await?; @@ -1488,6 +1483,20 @@ async fn guard_flow_from_debounce_data(nf: &NewFlow) -> Result<()> { "Flow debouncing configuration rejected: workers are behind minimum required version for debouncing feature" ); Err(Error::WorkersAreBehind { feature: "Debouncing".into(), min_version: "1.566.0".into() }) + } else if !*MIN_VERSION_SUPPORTS_DEBOUNCING_V2.read().await + && !nf + .parse_flow_value()? + .debouncing_settings + .is_legacy_compatible() + && !*WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT + { + tracing::warn!( + "Flow debouncing configuration rejected: workers are behind minimum required version for debouncing feature" + ); + Err(Error::WorkersAreBehind { + feature: "V2 Debouncing".into(), + min_version: "1.597.0".into(), + }) } else { Ok(()) } diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index e6160d5f0b..c9950c3c89 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -4199,7 +4199,6 @@ pub async fn run_flow<'c>( push_authed.as_ref(), false, None, - None, trigger, run_query.suspended_mode, ) @@ -4481,7 +4480,6 @@ pub async fn restart_flow( false, None, None, - None, run_query.suspended_mode, ) .await?; @@ -4615,7 +4613,6 @@ pub async fn push_script_job_by_path_into_queue<'c>( push_authed.as_ref(), false, None, - None, trigger, run_query.suspended_mode, ) @@ -4792,7 +4789,6 @@ pub async fn run_workflow_as_code( None, None, None, - None, ) .await?; @@ -5330,7 +5326,6 @@ pub async fn run_wait_result_job_by_path_get( false, None, None, - None, run_query.suspended_mode, ) .await?; @@ -5476,7 +5471,6 @@ pub async fn run_wait_result_script_by_path_internal( false, None, None, - None, run_query.suspended_mode, ) .await?; @@ -5601,7 +5595,6 @@ pub async fn run_wait_result_script_by_hash( false, None, None, - None, run_query.suspended_mode, ) .await?; @@ -6077,7 +6070,6 @@ async fn run_preview_script( None, None, None, - None, ) .await?; tx.commit().await?; @@ -6230,7 +6222,6 @@ async fn run_bundle_preview_script( None, None, None, - None, ) .await?; job_id = Some(uuid); @@ -6382,7 +6373,6 @@ async fn run_dependencies_job( None, None, None, - None, ) .await?; tx.commit().await?; @@ -6471,7 +6461,6 @@ async fn run_flow_dependencies_job( None, None, None, - None, ) .await?; tx.commit().await?; @@ -6828,7 +6817,6 @@ async fn run_preview_flow_job( None, None, None, - None, ) .await?; @@ -7026,7 +7014,6 @@ async fn run_dynamic_select( None, None, None, - None, ) .await?; tx.commit().await?; @@ -7169,7 +7156,6 @@ pub async fn run_job_by_hash_inner( push_authed.as_ref(), false, None, - None, trigger, run_query.suspended_mode, ) diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index aa9470366f..c8bf79edda 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -49,7 +49,7 @@ use windmill_common::{ s3_helpers::upload_artifact_to_store, scripts::{hash_script, ScriptRunnableSettingsHandle, ScriptRunnableSettingsInline}, utils::{paginate_without_limits, WarnAfterExt}, - worker::{CLOUD_HOSTED, MIN_VERSION_SUPPORTS_DEBOUNCING}, + worker::{CLOUD_HOSTED, MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2}, }; use windmill_common::{ @@ -69,7 +69,9 @@ use windmill_common::{ }; use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; use windmill_parser_ts::remove_pinned_imports; -use windmill_queue::{schedule::push_scheduled_job, PushIsolationLevel}; +use windmill_queue::{ + schedule::push_scheduled_job, PushIsolationLevel, WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT, +}; const MAX_HASH_HISTORY_LENGTH_STORED: usize = 20; @@ -866,14 +868,6 @@ async fn create_script_internal<'c>( ) }; - // Row lock debounce key for path. We need this to make all updates of runnables sequential and predictable. - tokio::time::timeout( - core::time::Duration::from_secs(60), - windmill_common::jobs::lock_debounce_key(&w_id, &ns.path, &mut tx), - ) - .warn_after_seconds(10) - .await??; - sqlx::query!( "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \ content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \ @@ -1087,6 +1081,7 @@ async fn create_script_internal<'c>( language: ns.language, path: ns.path.clone(), dedicated_worker: ns.dedicated_worker, + debouncing_settings: Default::default(), }, windmill_queue::PushArgs::from(&args), &authed.username, @@ -1112,7 +1107,6 @@ async fn create_script_internal<'c>( None, None, None, - None, ) .await?; @@ -2311,6 +2305,17 @@ async fn guard_script_from_debounce_data(ns: &NewScript) -> Result<()> { "Script debouncing configuration rejected: workers are behind minimum required version for debouncing feature" ); Err(Error::WorkersAreBehind { feature: "Debouncing".into(), min_version: "1.566.0".into() }) + } else if !*MIN_VERSION_SUPPORTS_DEBOUNCING_V2.read().await + && !ns.debouncing_settings.is_legacy_compatible() + && !*WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT + { + tracing::warn!( + "Script debouncing configuration rejected: workers are behind minimum required version for debouncing feature" + ); + Err(Error::WorkersAreBehind { + feature: "V2 Debouncing".into(), + min_version: "1.597.0".into(), + }) } else { Ok(()) } diff --git a/backend/windmill-api/src/triggers/trigger_helpers.rs b/backend/windmill-api/src/triggers/trigger_helpers.rs index 800113f868..c533dcaaeb 100644 --- a/backend/windmill-api/src/triggers/trigger_helpers.rs +++ b/backend/windmill-api/src/triggers/trigger_helpers.rs @@ -944,7 +944,6 @@ async fn trigger_script_with_retry_and_error_handler<'c>( push_authed.as_ref(), false, None, - None, Some(trigger), suspended_mode, ) diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 534db20bf0..ebdb16bcdc 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -407,6 +407,7 @@ pub enum JobPayload { hash: ScriptHash, language: ScriptLang, dedicated_worker: Option, + debouncing_settings: DebouncingSettings, }, /// Flow Dependency Job @@ -414,12 +415,14 @@ pub enum JobPayload { path: String, dedicated_worker: Option, version: i64, + debouncing_settings: DebouncingSettings, }, /// App Dependency Job AppDependencies { path: String, version: i64, + debouncing_settings: DebouncingSettings, }, /// Flow Dependency Job, exposed with API. Requirements can be partially or fully predefined @@ -874,34 +877,6 @@ pub async fn check_tag_available_for_workspace_internal( return Ok(()); } -pub async fn lock_debounce_key<'c>( - w_id: &str, - runnable_path: &str, - tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, -) -> error::Result> { - if !*crate::worker::MIN_VERSION_SUPPORTS_DEBOUNCING.read().await { - tracing::warn!("Debouncing is not supported on this version of Windmill. Minimum version required for debouncing support."); - return Ok(None); - } - - let key = format!("{w_id}:{runnable_path}:dependency"); - - tracing::debug!( - workspace_id = %w_id, - runnable_path = %runnable_path, - debounce_key = %key, - "Locking debounce_key for dependency job scheduling" - ); - - sqlx::query_scalar!( - "SELECT job_id FROM debounce_key WHERE key = $1 AND job_id IN (SELECT id FROM v2_job_queue) FOR UPDATE", - &key - ) - .fetch_optional(&mut **tx) - .await - .map_err(error::Error::from) -} - pub struct RunInlinePreviewScriptFnParams { pub workspace_id: String, pub content: String, diff --git a/backend/windmill-common/src/runnable_settings/settings.rs b/backend/windmill-common/src/runnable_settings/settings.rs index 7819ed2247..6dc4eb38ca 100644 --- a/backend/windmill-common/src/runnable_settings/settings.rs +++ b/backend/windmill-common/src/runnable_settings/settings.rs @@ -179,6 +179,12 @@ impl DebouncingSettings { ..self } } + + pub fn is_legacy_compatible(&self) -> bool { + self.max_total_debouncing_time.is_none() + && self.max_total_debounces_amount.is_none() + && self.debounce_args_to_accumulate.is_none() + } } impl ConcurrencySettings { diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index c01d1d2fb1..13ab002f83 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -977,9 +977,9 @@ pub async fn clone_script<'c>( w_id: &str, deployment_message: Option, db: &DB, - tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, ) -> crate::error::Result { - let s = if let Some(s) = fetch_script_for_update(base_hash, w_id, &mut **tx).await? { + let mut tx = db.begin().await?; + let s = if let Some(s) = fetch_script_for_update(base_hash, w_id, &mut *tx).await? { s } else { return Err(crate::error::Error::NotFound(format!( @@ -1065,7 +1065,7 @@ pub async fn clone_script<'c>( codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets, debounce_key, debounce_delay_s, runnable_settings_handle FROM script WHERE hash = $2 AND workspace_id = $3; - ", new_hash, base_hash.0, w_id).execute(&mut **tx).await?; + ", new_hash, base_hash.0, w_id).execute(&mut *tx).await?; // Archive base. sqlx::query!( @@ -1073,8 +1073,9 @@ pub async fn clone_script<'c>( *base_hash, w_id ) - .execute(&mut **tx) + .execute(&mut *tx) .await?; + tx.commit().await?; Ok(ClonedScript { old_script: ns, new_hash }) } diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index c526b8a9be..71196a77ce 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -265,6 +265,7 @@ lazy_static::lazy_static! { .unwrap_or(false); pub static ref MIN_VERSION: Arc> = Arc::new(RwLock::new(Version::new(0, 0, 0))); + pub static ref MIN_VERSION_SUPPORTS_DEBOUNCING_V2: Arc> = Arc::new(RwLock::new(false)); pub static ref MIN_VERSION_SUPPORTS_RUNNABLE_SETTINGS_V0: Arc> = Arc::new(RwLock::new(false)); /// Global flag indicating if all workers support workspace dependencies feature (>= 1.583.0) /// This flag is updated during worker initialization by checking the minimum version across all workers @@ -429,10 +430,6 @@ 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, @@ -1298,6 +1295,7 @@ pub async fn update_min_version(conn: &Connection) -> bool { tracing::info!("Minimal worker version: {min_version}"); } + *MIN_VERSION_SUPPORTS_DEBOUNCING_V2.write().await = min_version >= Version::new(1, 597, 0); *MIN_VERSION_SUPPORTS_RUNNABLE_SETTINGS_V0.write().await = min_version >= *crate::runnable_settings::MIN_VERSION_RUNNABLE_SETTINGS_V0; diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index f0bd2b4c3f..7ea48fca7d 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -46,8 +46,11 @@ use windmill_common::runnable_settings::{ RunnableSettingsTrait, }; use windmill_common::triggers::TriggerMetadata; -use windmill_common::utils::{configure_client, now_from_db}; -use windmill_common::worker::{Connection, MIN_VERSION_SUPPORTS_DEBOUNCING, SCRIPT_TOKEN_EXPIRY}; +use windmill_common::utils::{calculate_hash, configure_client, now_from_db}; +use windmill_common::worker::{ + Connection, MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2, + SCRIPT_TOKEN_EXPIRY, +}; use windmill_common::{ auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username}, @@ -117,14 +120,14 @@ lazy_static::lazy_static! { .connect_timeout(std::time::Duration::from_secs(10))) .build().unwrap(); + pub static ref WMDEBUG_NO_DEBOUNCING: bool = std::env::var("WMDEBUG_NO_DEBOUNCING").is_ok(); + pub static ref WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT: bool = std::env::var("WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT").is_ok(); static ref JOB_ARGS_AUDIT_LOGS: bool = std::env::var("JOB_ARGS_AUDIT_LOGS") .ok() .and_then(|x| x.parse().ok()) .unwrap_or(false); - // TODO: Remove - static ref WMDEBUG_NO_DJOB_DEBOUNCING: bool = std::env::var("WMDEBUG_NO_DJOB_DEBOUNCING").is_ok(); } #[cfg(feature = "cloud")] @@ -464,7 +467,6 @@ pub async fn push_init_job<'c>( None, None, None, - None, ) .await?; inner_tx.commit().await?; @@ -522,7 +524,6 @@ pub async fn push_periodic_bash_job<'c>( None, None, None, - None, ) .await?; inner_tx.commit().await?; @@ -863,7 +864,7 @@ pub async fn add_completed_job( async fn commit_completed_job( db: &Pool, - queued_job: &MiniCompletedJob, + completed_job: &MiniCompletedJob, success: bool, skipped: bool, result: Json<&T>, @@ -879,7 +880,7 @@ async fn commit_completed_job( let mut tx = db.begin().warn_after_seconds(10).await?; - let job_id = queued_job.id; + let job_id = completed_job.id; // tracing::error!("1 {:?}", start.elapsed()); // tracing::debug!( @@ -891,7 +892,7 @@ async fn commit_completed_job( let mem_peak = mem_peak; // add_time!(bench, "add_completed_job query START"); - if let Some(value) = check_result_size(db, queued_job, result).await { + if let Some(value) = check_result_size(db, completed_job, result).await { return value; } @@ -920,7 +921,7 @@ async fn commit_completed_job( q.worker FROM v2_job_queue q LEFT JOIN v2_job_status USING (id) WHERE q.id = $1 ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3 RETURNING duration_ms AS \"duration_ms!\"", - /* $1 */ queued_job.id, + /* $1 */ completed_job.id, /* $2 */ success, /* $3 */ result as Json<&T>, /* $4 */ canceled_by.is_some(), @@ -975,8 +976,8 @@ async fn commit_completed_job( .map_err(|e| Error::InternalErr(format!("Could not update job labels: {e:#}")))?; } - if !queued_job.is_flow_step() { - if let Some(parent_job) = queued_job.parent_job { + if !completed_job.is_flow_step() { + if let Some(parent_job) = completed_job.parent_job { let _ = sqlx::query_scalar!( "UPDATE v2_job_status SET workflow_as_code_status = jsonb_set( @@ -989,7 +990,7 @@ async fn commit_completed_job( to_jsonb($2::bigint) ) WHERE id = $3", - &queued_job.id.to_string(), + &completed_job.id.to_string(), duration, parent_job ) @@ -1010,8 +1011,8 @@ async fn commit_completed_job( tx = delete_job(tx, &job_id).warn_after_seconds(10).await?; // tracing::error!("3 {:?}", start.elapsed()); - if queued_job.is_flow_step() { - if let Some(parent_job) = queued_job.parent_job { + if completed_job.is_flow_step() { + if let Some(parent_job) = completed_job.parent_job { // persist the flow last progress timestamp to avoid zombie flow jobs tracing::debug!( "Persisting flow last progress timestamp to flow job: {:?}", @@ -1025,7 +1026,7 @@ async fn commit_completed_job( AND q.workspace_id = $2 AND canceled_by IS NULL", parent_job, - &queued_job.workspace_id + &completed_job.workspace_id ) .execute(&mut *tx) .warn_after_seconds(10) @@ -1034,22 +1035,22 @@ async fn commit_completed_job( let r = sqlx::query_scalar!( "UPDATE parallel_monitor_lock SET last_ping = now() WHERE parent_flow_id = $1 and job_id = $2 RETURNING 1", parent_job, - &queued_job.id + &completed_job.id ).fetch_optional(&mut *tx).warn_after_seconds(10).await?; if r.is_some() { tracing::info!( "parallel flow iteration is done, setting parallel monitor last ping lock for job {}", - &queued_job.id + &completed_job.id ); } } } } else { - if queued_job.schedule_path().is_some() && queued_job.runnable_path.is_some() { - let schedule_path = queued_job.schedule_path().unwrap(); - let script_path = queued_job.runnable_path.as_ref().unwrap(); + if completed_job.schedule_path().is_some() && completed_job.runnable_path.is_some() { + let schedule_path = completed_job.schedule_path().unwrap(); + let script_path = completed_job.runnable_path.as_ref().unwrap(); - let schedule = get_schedule_opt(&mut *tx, &queued_job.workspace_id, &schedule_path) + let schedule = get_schedule_opt(&mut *tx, &completed_job.workspace_id, &schedule_path) .warn_after_seconds(10) .await?; @@ -1063,7 +1064,7 @@ async fn commit_completed_job( // for flows, only try to schedule next tick here if flow failed and because first handle_flow failed (step = 0, modules[0] = {type: 'Failure', 'job': uuid::nil()}) // or job was cancelled before first handle_flow was called (step = 0, modules = [] OR modules[0].type == 'WaitingForPriorSteps') // otherwise flow rescheduling is done inside handle_flow - let schedule_next_tick = !queued_job.is_flow() + let schedule_next_tick = !completed_job.is_flow() || from_cache || !success && sqlx::query_scalar!( @@ -1079,8 +1080,8 @@ async fn commit_completed_job( ) FROM v2_job_completed WHERE id = $2 AND workspace_id = $3", Uuid::nil().to_string(), - &queued_job.id, - &queued_job.workspace_id + &completed_job.id, + &completed_job.workspace_id ) .fetch_optional(&mut *tx) .warn_after_seconds(10) @@ -1091,10 +1092,10 @@ async fn commit_completed_job( if schedule_next_tick { if let Err(err) = Box::pin(handle_maybe_scheduled_job( db, - queued_job, + completed_job, &schedule, &script_path, - &queued_job.workspace_id, + &completed_job.workspace_id, )) .warn_after_seconds(10) .await @@ -1112,12 +1113,12 @@ async fn commit_completed_job( db, &schedule, &script_path, - &queued_job.workspace_id, + &completed_job.workspace_id, success, result, job_id, - queued_job.started_at.unwrap_or(chrono::Utc::now()), - queued_job.priority, + completed_job.started_at.unwrap_or(chrono::Utc::now()), + completed_job.priority, ) .warn_after_seconds(10) .await @@ -1125,14 +1126,14 @@ async fn commit_completed_job( if !success { tracing::error!("Could not apply schedule error handler: {}", err); let base_url = windmill_common::BASE_URL.read().await; - let w_id: &String = &queued_job.workspace_id; + let w_id: &String = &completed_job.workspace_id; if !matches!(err, Error::QuotaExceeded(_)) { report_error_to_workspace_handler_or_critical_side_channel( - &queued_job, + &completed_job, db, format!( "Failed to push schedule error handler job to handle failed job ({base_url}/run/{}?workspace={w_id}): {}", - queued_job.id, + completed_job.id, err ), ) @@ -1146,20 +1147,20 @@ async fn commit_completed_job( } else { tracing::error!( "Schedule {schedule_path} in {} not found. Impossible to schedule again and apply schedule handlers", - &queued_job.workspace_id + &completed_job.workspace_id ); } } } - if queued_job.concurrent_limit.is_some() - || RunnableSettings::prefetch_cached_from_handle(queued_job.runnable_settings_handle, db) + if completed_job.concurrent_limit.is_some() + || RunnableSettings::prefetch_cached_from_handle(completed_job.runnable_settings_handle, db) .await? .1 .concurrent_limit .is_some() { - let concurrency_key = concurrency_key(db, &queued_job.id).await?; + let concurrency_key = concurrency_key(db, &completed_job.id).await?; if *DISABLE_CONCURRENCY_LIMIT || concurrency_key.is_none() { tracing::warn!("Concurrency limit is disabled, skipping"); } else { @@ -1167,7 +1168,7 @@ async fn commit_completed_job( sqlx::query_scalar!( "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1", concurrency_key, - queued_job.id.hyphenated().to_string(), + completed_job.id.hyphenated().to_string(), ) .execute(&mut *tx) .warn_after_seconds(10) @@ -1175,14 +1176,14 @@ async fn commit_completed_job( .map_err(|e| { Error::internal_err(format!( "Could not decrement concurrency counter for job_id={}: {e:#}", - queued_job.id + completed_job.id )) })?; } if let Err(e) = sqlx::query_scalar!( "UPDATE concurrency_key SET ended_at = now() WHERE job_id = $1", - queued_job.id, + completed_job.id, ) .execute(&mut *tx) .warn_after_seconds(10) @@ -1190,7 +1191,7 @@ async fn commit_completed_job( { tracing::error!( "Could not update concurrency_key ended_at for job_id={}: {e:#}", - queued_job.id, + completed_job.id, ); } tracing::debug!("decremented concurrency counter"); @@ -1212,21 +1213,21 @@ async fn commit_completed_job( tracing::info!( %job_id, - root_job = ?queued_job.flow_innermost_root_job.map(|x| x.to_string()).unwrap_or_else(|| String::new()), - path = &queued_job.runnable_path, - job_kind = ?queued_job.kind, - started_at = ?queued_job.started_at.map(|x| x.to_string()).unwrap_or_else(|| String::new()), + root_job = ?completed_job.flow_innermost_root_job.map(|x| x.to_string()).unwrap_or_else(|| String::new()), + path = &completed_job.runnable_path, + job_kind = ?completed_job.kind, + started_at = ?completed_job.started_at.map(|x| x.to_string()).unwrap_or_else(|| String::new()), duration = ?duration, - permissioned_as = ?queued_job.permissioned_as, - email = ?queued_job.permissioned_as_email, - created_by = queued_job.created_by, - is_flow_step = queued_job.is_flow_step(), - language = ?queued_job.script_lang, - scheduled_for = ?queued_job.scheduled_for, - workspace_id = ?queued_job.workspace_id, + permissioned_as = ?completed_job.permissioned_as, + email = ?completed_job.permissioned_as_email, + created_by = completed_job.created_by, + is_flow_step = completed_job.is_flow_step(), + language = ?completed_job.script_lang, + scheduled_for = ?completed_job.scheduled_for, + workspace_id = ?completed_job.workspace_id, success, "inserted completed job: {} (success: {success})", - queued_job.id + completed_job.id ); // tracing::info!("completed job: {:?}", start.elapsed().as_micros()); Ok((None, duration, _skip_downstream_error_handlers)) @@ -1393,7 +1394,6 @@ async fn restart_job_if_perpetual_inner( None, None, None, - None, ) .await?; tx.commit().await?; @@ -1960,7 +1960,6 @@ pub async fn push_error_handler<'a, 'c, T: Serialize + Send + Sync>( None, None, None, - None, ) .await?; tx.commit().await?; @@ -2605,248 +2604,211 @@ impl PulledJobResult { /// Generic preprocess function /// Can be used for any kind of preprocessing - pub async fn preprocess(&mut self, db: &DB) -> error::Result<()> { - let PulledJobResult { job: Some(ref mut pulled_job), .. } = self else { + pub async fn maybe_apply_debouncing(&mut self, db: &DB) -> error::Result<()> { + let PulledJobResult { job: Some(ref mut j), .. } = self else { return Ok(()); }; - let kind = pulled_job.kind; - // Handle dependency job debouncing cleanup when a job is pulled for execution - if kind.is_dependency() - && pulled_job - .args + let DebouncingSettings { debounce_delay_s, debounce_args_to_accumulate, .. } = + RunnableSettings::prefetch_cached_from_handle(j.runnable_settings_handle, db) + .await? + .0; + + let (kind, j_id) = (j.kind, j.id); + let is_djob_to_debounce = kind.is_dependency() + && j.args .as_ref() - .map(|x| x.get("triggered_by_relative_import").is_some()) - .unwrap_or_default() - && !*WMDEBUG_NO_DJOB_DEBOUNCING + .and_then(|x| x.get("triggered_by_relative_import")) + .is_some(); + + if (is_djob_to_debounce || debounce_delay_s.filter(|x| *x > 0).is_some()) + && *MIN_VERSION_SUPPORTS_DEBOUNCING.read().await + && !*WMDEBUG_NO_DEBOUNCING { - return Box::pin(async move { - // Only used for testing in tests/relative_imports.rs - // Give us some space to work with. - #[cfg(debug_assertions)] - if let Some(duration) = pulled_job - .args - .as_ref() - .map(|x| { - x.get("dbg_sleep_between_pull_and_debounce_key_removal") - .map(|v| serde_json::from_str::(v.get()).ok()) - .flatten() - }) - .flatten() - { - tracing::debug!("going to sleep",); - sleep(std::time::Duration::from_secs(duration as u64)).await; - } + let needs_debounce = sqlx::query_scalar!( + "WITH _ AS ( + DELETE FROM debounce_key WHERE job_id = $1 + ) SELECT status = 'skipped' FROM v2_job_completed WHERE id = $1", + j_id, + ) + .fetch_optional(db) + .await? + .flatten() + .unwrap_or_default(); - tracing::debug!( - "Processing debounce cleanup for dependency job {} at path {:?}", - &pulled_job.id, - &pulled_job.runnable_path + if needs_debounce { + tracing::info!( + job_id = %j_id, + "Late debounce: job was already debounced by another job, skipping execution" ); + self.job = None; + return Ok(()); + } - let key = format!("{}:{}:dependency", &pulled_job.workspace_id, pulled_job.runnable_path()); - let mut tx = db.begin().await?; - - // === DEBOUNCE CLEANUP === - // - // Clean up the debounce_key entry for this job (if it exists). - // - // IMPORTANT: We delete by key (not job_id) to avoid race conditions: - // If pusher has locked this row then this call will be blocked until all txs are commited. - // - // The idea is that the worker_lockfiles::trigger_dependents_to_recompute_locks will fetch the latest version of the obj. - // This object needs to be created before the djob is executed and it happens right here. - // - // This way the next pusher can fetch the latest version of object and base their djob payload on newest version. - // The concurrency limit on djobs will make sure that by the time next djob is started executing the base version it is referencing - // has already calculated all locks. This way even next djob will always use the fully finalized version of object. - // - // - // - // Note: We don't use a transaction here for performance (it's called during job pull). - // This means there's a tiny window where the job is running but key isn't deleted yet, - // which is acceptable because new requests will just accumulate data to this job. - tracing::debug!( - job_id = %pulled_job.id, - "Cleaning up debounce_key entry for completed/pulled job" - ); - - // This will either: - // 1. Block until pusher pushed. Which gives us: - // - If there was any stale data in pusher, then we will read it here (couple of lines below) - // 2. Block pusher until we are done here. This gives us: - // - We will clone objects and retrieve the latest version. So when we are done the pusher can read latest version. - sqlx::query!("DELETE FROM debounce_key WHERE key = $1", &key) - .execute(&mut *tx) - .await - .map_err(|e| { - tracing::error!( - error = %e, - job_id = %pulled_job.id, - "Failed to delete debounce_key" - ); - e - })?; - - let Some(base_hash) = pulled_job.runnable_id else { - return Err(Error::InternalErr( - "Missing runnable_id for dependency job triggered by relative import" - .to_string(), - )); + if matches!(kind, JobKind::FlowDependencies | JobKind::AppDependencies) + && !*MIN_VERSION_SUPPORTS_DEBOUNCING_V2.read().await + && !*WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT + { + // Simply disable optimization for apps and flows if min version doesn't support debouncing v2 + let field_name = match kind { + JobKind::FlowDependencies => "nodes_to_relock", + JobKind::AppDependencies => "components_to_relock", + _ => unreachable!(), }; tracing::debug!( - job_id = %pulled_job.id, - base_hash = %base_hash, + job_id = %j_id, job_kind = ?kind, - "Creating new version for dependency job triggered by relative import" + field_removed = field_name, + "Removing optimization field: workers behind v2 debouncing version, disabling relock optimization" ); - let new_id = match kind { - JobKind::Dependencies => { - let deployment_message = pulled_job - .args - .clone() - .map(|hashmap| { - hashmap - .get("deployment_message") - .map(|map_value| { - serde_json::from_str::(map_value.get()).ok() - }) - .flatten() - }) - .flatten(); - - // This way we tell downstream which script we should archive when the resolution is finished. - // (not used at the moment) - pulled_job.args.as_mut().map(|args| { - args.insert("base_hash".to_owned(), to_raw_value(&*base_hash)) - }); - - let cloned_script = windmill_common::scripts::clone_script( - base_hash, - &pulled_job.workspace_id, - deployment_message, - db, - &mut tx, + if let Some(args) = &mut j.args { + args.remove(field_name); + } + } else if let Some(arg_name_to_accumulate) = + // TODO: Maybe support multiple arguments in future + debounce_args_to_accumulate.as_ref().and_then(|v| v.get(0)) + { + tracing::debug!( + job_id = %j_id, + job_kind = ?kind, + arg_name = arg_name_to_accumulate, + "Accumulating debounced arguments from batch" + ); + let mut accumulated_arg: Vec> = vec![]; + for str_o in sqlx::query_scalar!( + "WITH ids AS ( + SELECT id as job_id FROM v2_job_debounce_batch WHERE debounce_batch = ( + SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1 ) - .await?; - - cloned_script.new_hash - } - JobKind::FlowDependencies => { - sqlx::query_scalar!( - "INSERT INTO flow_version - (workspace_id, path, value, schema, created_by) - - SELECT workspace_id, path, value, schema, created_by - FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3 - - RETURNING id - ", - pulled_job.runnable_path(), - pulled_job.workspace_id, - *base_hash, - ) - .fetch_one(&mut *tx) - .await? - } - JobKind::AppDependencies => { - sqlx::query_scalar!( - "INSERT INTO app_version - (app_id, value, created_by, raw_app) - SELECT app_id, value, created_by, raw_app - FROM app_version WHERE id = $1 - RETURNING id", - *base_hash - ) - .fetch_one(&mut *tx) - .await? - } - _ => { - return Err(Error::InternalErr(format!( - "Matched unexpected JobKind ({:?}). This is a bug!", - kind - ))) - } - }; - - pulled_job.runnable_id.replace(new_id.into()); - - if *windmill_common::worker::MIN_VERSION_SUPPORTS_DEBOUNCING.read().await { - // === RETRIEVE ACCUMULATED DEBOUNCE DATA === - // - // For flows and apps, retrieve all nodes/components that were accumulated - // during the debounce window. This data comes from requests that were merged - // into this job instead of creating their own jobs. - // - // Scripts don't need this because they don't have nodes/components to relock. - if let Some(to_relock_field) = match &pulled_job.kind { - JobKind::FlowDependencies => Some("nodes_to_relock"), - JobKind::AppDependencies => Some("components_to_relock"), - _ => None, // Scripts don't use accumulated stale data - } { - tracing::debug!( - job_id = %pulled_job.id, - job_kind = ?pulled_job.kind, - field = %to_relock_field, - "Retrieving accumulated stale data from debounced requests" - ); - - if let Some(stale_data) = sqlx::query_scalar!( - "DELETE FROM debounce_stale_data WHERE job_id = $1 RETURNING to_relock", - &pulled_job.id - ) - .fetch_optional(&mut *tx) - .await - .map_err(|e| { - tracing::error!( - error = %e, - job_id = %pulled_job.id, - "Failed to retrieve debounce_stale_data" - ); - e - })? - .flatten() - { - tracing::debug!( - job_id = %pulled_job.id, - node_count = stale_data.len(), - nodes = ?stale_data, - "Retrieved accumulated nodes/components from {} debounced requests", - stale_data.len() - ); - - // Replace the job's relock list with the accumulated data - // This ensures all nodes from all debounced requests are processed - if let Some(args) = pulled_job.args.as_mut() { - args.insert(to_relock_field.to_owned(), to_raw_value(&stale_data)); - tracing::debug!( - field = %to_relock_field, - "Updated job args with accumulated debounce data" - ); + ) SELECT args->>$2 FROM ids LEFT JOIN v2_job ON v2_job.id = ids.job_id + ", + j_id, + arg_name_to_accumulate, + ) + .fetch_all(db) + .await? + .into_iter() + { + if let Some(s) = str_o.as_ref() { + match serde_json::from_str::>>(s) { + Ok(ref mut vec) => accumulated_arg.append(vec), + Err(e) => { + return Err(error::Error::ArgumentErr(format!("cannot consolidate arguments of non-list type. Type provided for argument `{arg_name_to_accumulate}` is not a list\nUnwrapped Error: {e}"))); } - } else { - tracing::trace!( - job_id = %pulled_job.id, - "No accumulated stale data found (no debounced requests or already cleaned up)" - ); } } - } 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?; - Ok(()) - }).await; + tracing::debug!( + job_id = %j_id, + arg_name = arg_name_to_accumulate, + accumulated_count = accumulated_arg.len(), + "Accumulated arguments from debounced jobs in batch" + ); + + j.args + .get_or_insert(Json(Default::default())) + .as_mut() + .insert( + arg_name_to_accumulate.to_owned(), + to_raw_value(&accumulated_arg), + ); + } + + // Handle dependency job debouncing cleanup when a job is pulled for execution + if is_djob_to_debounce { + clone_runnable(j, db).await?; + } } Ok(()) } } +async fn clone_runnable(j: &mut PulledJob, db: &DB) -> error::Result<()> { + let Some(base_hash) = j.runnable_id else { + return Err(Error::InternalErr( + "Missing runnable_id for dependency job triggered by relative import".to_string(), + )); + }; + + tracing::debug!( + job_id = %j.id, + base_hash = %base_hash, + job_kind = ?j.kind, + runnable_path = ?j.runnable_path, + "Creating new version for dependency job triggered by relative import" + ); + + { + let maybe_new_id = match j.kind { + JobKind::Dependencies => { + let deployment_message = j + .args + .clone() + .map(|hashmap| { + hashmap + .get("deployment_message") + .map(|map_value| serde_json::from_str::(map_value.get()).ok()) + .flatten() + }) + .flatten(); + + // This way we tell downstream which script we should archive when the resolution is finished. + // (not used at the moment) + j.args + .as_mut() + .map(|args| args.insert("base_hash".to_owned(), to_raw_value(&*base_hash))); + + windmill_common::scripts::clone_script( + base_hash, + &j.workspace_id, + deployment_message, + db, + ) + .await? + .new_hash + } + JobKind::FlowDependencies => { + sqlx::query_scalar!( + "INSERT INTO flow_version + (workspace_id, path, value, schema, created_by) + + SELECT workspace_id, path, value, schema, created_by + FROM flow_version WHERE path = $1 AND workspace_id = $2 AND id = $3 + + RETURNING id + ", + j.runnable_path(), + j.workspace_id, + *base_hash, + ) + .fetch_one(db) + .await? + } + JobKind::AppDependencies => { + sqlx::query_scalar!( + "INSERT INTO app_version + (app_id, value, created_by, raw_app) + SELECT app_id, value, created_by, raw_app + FROM app_version WHERE id = $1 + RETURNING id", + *base_hash + ) + .fetch_one(db) + .await? + } + _ => *base_hash, + }; + + j.runnable_id.replace(maybe_new_id.into()); + } + + Ok(()) +} + // TODO: Factorize /// Pull the job from queue pub async fn pull( @@ -2944,7 +2906,7 @@ pub async fn pull( Some(job) if concurrency_settings.concurrent_limit.is_some() // Concurrency limit is available for either enterprise job or dependency job - && (cfg!(feature = "enterprise") || (job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING)) => + && (cfg!(feature = "enterprise") || (job.is_dependency() && !*WMDEBUG_NO_DEBOUNCING)) => { timeout( Duration::from_secs(15), @@ -3017,7 +2979,7 @@ pub async fn pull( let has_concurent_limit = job.is_dependency() && job.concurrent_limit.is_some() && cfg!(feature = "private") - && !*WMDEBUG_NO_DJOB_DEBOUNCING; + && !*WMDEBUG_NO_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 @@ -3040,7 +3002,7 @@ pub async fn pull( #[cfg(feature = "private")] if cfg!(feature = "enterprise") - || (pulled_job.is_dependency() && !*WMDEBUG_NO_DJOB_DEBOUNCING) + || (pulled_job.is_dependency() && !*WMDEBUG_NO_DEBOUNCING) { if let Some(pulled_job_res) = timeout( Duration::from_secs(15), @@ -3201,44 +3163,139 @@ pub async fn custom_debounce_key( .await } -/// Helper function to extract nodes/components to relock from job arguments -/// Returns the list of nodes to relock if present in either nodes_to_relock (flows) or components_to_relock (apps) -fn extract_to_relock_from_args(args: &HashMap>) -> Option> { - args.get("nodes_to_relock") // For flows - .or(args.get("components_to_relock")) // For apps - .and_then(|rv| { - serde_json::from_str::>(&rv.to_string()) - .map_err(|e| tracing::warn!("Failed to deserialize relock data: {}", e)) - .ok() - }) +pub fn resolve_debounce_key<'b>( + unresolved_debounce_key: Option, + runnable_path: &Option, + workspace_id: &str, + job_kind: JobKind, + args: &PushArgs<'b>, + args_to_ignore_if_default: Option<&String>, +) -> String { + let original_debounce_key = unresolved_debounce_key + .map(|x| crate::interpolate_args(x, &args, workspace_id)) + .unwrap_or(format!( + "{}#args:{}", + crate::fullpath_with_workspace(workspace_id, runnable_path.as_ref(), &job_kind), + args.args + .iter() + .filter_map(|(k, v)| { + if args_to_ignore_if_default + .map(|name| name == k) + .unwrap_or_default() + { + None + } else { + Some(v.to_string()) + } + }) + // TODO: disable sorted? + .sorted() + .collect_vec() + .join(":"), + )); + + tracing::debug!("Original debounce key: {}", original_debounce_key); + + // If debounce_key is not too long (< 255 chars), keep it as is, otherwise hash it + const MAX_DEBOUNCE_KEY_LENGTH: usize = 255; + let resolved = if original_debounce_key.len() <= MAX_DEBOUNCE_KEY_LENGTH { + original_debounce_key + } else { + let hash = calculate_hash(&original_debounce_key); + tracing::debug!( + "Debounce key too long ({}), using hash: {}", + original_debounce_key.len(), + hash + ); + hash + }; + + #[cfg(feature = "cloud")] + let resolved = format!("{workspace_id}:{resolved}"); + + tracing::debug!("Final debounce key: {}", resolved); + resolved } -/// Helper function to accumulate nodes/components to relock for a debounced job -/// This merges new items with existing ones, removing duplicates -async fn accumulate_debounce_stale_data( - tx: &mut Transaction<'_, Postgres>, +#[derive(Debug)] +pub enum DebouncingLimitsReport { + MaxCountExceeded, + TimeExceeded, + CanDebounce, +} + +pub async fn check_debouncing_within_limits( + (current_time, current_amount): (DateTime, i32), + (allowed_time, allowed_amount): (Option, Option), job_id: &Uuid, - to_relock: &[String], -) -> Result<(), Error> { - sqlx::query!( - " - INSERT INTO debounce_stale_data (job_id, to_relock) - VALUES ($1, $2) - ON CONFLICT (job_id) - DO UPDATE SET to_relock = ( - SELECT array_agg(DISTINCT x) - FROM unnest( - -- Combine existing array with new values, removing duplicates - array_cat(debounce_stale_data.to_relock, EXCLUDED.to_relock) - ) AS x - ) - ", - job_id, - to_relock - ) - .execute(&mut **tx) - .await?; - Ok(()) + runnable_path: &Option, +) -> DebouncingLimitsReport { + use DebouncingLimitsReport::*; + + let no_legacy_compat = *MIN_VERSION_SUPPORTS_DEBOUNCING_V2.read().await + || *WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT; + + if !no_legacy_compat { + tracing::warn!( + job_id = %job_id, + runnable_path = ?runnable_path, + allowed_time = ?allowed_time, + allowed_amount = ?allowed_amount, + "Debouncing limits not enforced: workers are behind minimum version for v2 debouncing (require >= 1.597.0)" + ); + } + + let elapsed_seconds = (Utc::now() - current_time).num_seconds() as i32; + + tracing::debug!( + job_id = %job_id, + runnable_path = ?runnable_path, + current_amount = current_amount, + allowed_amount = ?allowed_amount, + elapsed_seconds = elapsed_seconds, + allowed_time = ?allowed_time, + first_started_at = %current_time, + no_legacy_compat = no_legacy_compat, + "Checking debouncing limits" + ); + + if allowed_amount + .map(|allowed_amount| current_amount > allowed_amount) + .unwrap_or_default() + && no_legacy_compat + { + tracing::info!( + job_id = %job_id, + runnable_path = ?runnable_path, + current_amount = current_amount, + allowed_amount = allowed_amount, + "Debouncing limit exceeded: max debounce count reached" + ); + MaxCountExceeded + } else if allowed_time + .map(|allowed_time| elapsed_seconds > allowed_time) + .unwrap_or_default() + && no_legacy_compat + { + tracing::info!( + job_id = %job_id, + runnable_path = ?runnable_path, + elapsed_seconds = elapsed_seconds, + allowed_time = allowed_time, + first_started_at = %current_time, + "Debouncing limit exceeded: max debounce time window reached" + ); + TimeExceeded + } else { + tracing::debug!( + job_id = %job_id, + runnable_path = ?runnable_path, + current_amount = current_amount, + elapsed_seconds = elapsed_seconds, + "Debouncing within limits: can continue debouncing" + ); + CanDebounce + } } pub fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> String { @@ -3973,9 +4030,6 @@ pub async fn push<'c, 'd>( authed: Option<&Authed>, running: bool, // whether the job is already running: only set this to true if you don't want the job to be picked up by a worker from the queue. It will also set started_at to now. end_user_email: Option, - // If we know there is already a debounce job, we can use this for debouncing. - // NOTE: Only works with dependency jobs triggered by relative imports - debounce_job_id_o: Option, trigger: Option, suspended_mode: Option, ) -> Result<(Uuid, Transaction<'c, Postgres>), Error> { @@ -4322,12 +4376,19 @@ pub async fn push<'c, 'd>( dedicated_worker, ..Default::default() }, - JobPayload::Dependencies { hash, language, path, dedicated_worker } => JobPayloadUntagged { + JobPayload::Dependencies { + hash, + language, + path, + dedicated_worker, + debouncing_settings, + } => JobPayloadUntagged { runnable_id: Some(hash.0), - runnable_path: Some(path.clone()), + runnable_path: Some(path), job_kind: JobKind::Dependencies, language: Some(language), dedicated_worker, + debouncing_settings, ..Default::default() }, @@ -4349,7 +4410,7 @@ pub async fn push<'c, 'd>( raw_flow: Some(flow_value), ..Default::default() }, - JobPayload::FlowDependencies { path, dedicated_worker, version } => { + JobPayload::FlowDependencies { path, dedicated_worker, version, debouncing_settings } => { #[cfg(test)] let skip_compat = args .args @@ -4376,13 +4437,15 @@ pub async fn push<'c, 'd>( job_kind: JobKind::FlowDependencies, raw_flow: value_o, dedicated_worker, + debouncing_settings, ..Default::default() } } - JobPayload::AppDependencies { path, version } => JobPayloadUntagged { + JobPayload::AppDependencies { path, version, debouncing_settings } => JobPayloadUntagged { runnable_id: Some(version), runnable_path: Some(path.clone()), job_kind: JobKind::AppDependencies, + debouncing_settings, ..Default::default() }, JobPayload::RawFlow { mut value, path, restarted_from } => { @@ -4802,7 +4865,7 @@ pub async fn push<'c, 'd>( &runnable_path, cfg!(feature = "private") && job_kind.is_dependency() - && !*WMDEBUG_NO_DJOB_DEBOUNCING + && !*WMDEBUG_NO_DEBOUNCING && *MIN_VERSION_SUPPORTS_DEBOUNCING.read().await, ) { concurrency_settings.concurrency_key = Some(format!("dependency:{workspace_id}/{path}")); @@ -4941,209 +5004,20 @@ pub async fn push<'c, 'd>( Ulid::new().into() }; - // Dependency job debouncing: When multiple dependency jobs are scheduled for the same script/flow/app, - // we want to deduplicate them to avoid redundant work. The debouncing mechanism works by: - // 1. Creating a unique debounce key for each dependency target (dependency:workspace/type/path) - // 2. Reusing existing jobs when possible, or creating new ones when the existing job is already running - // 3. Accumulating the nodes/components that need relocking across all debounced requests - match ( - scheduled_for_o.is_some(), - job_kind.is_dependency(), - runnable_path.clone(), - *WMDEBUG_NO_DJOB_DEBOUNCING, - *MIN_VERSION_SUPPORTS_DEBOUNCING.read().await, - // We only do debouncing for jobs triggered by relative imports - // We do not want this be the case for normal djobs, since they will always be sequential. - args.args.contains_key("triggered_by_relative_import"), - ) { - (_, _, _, _, false, _) => { - tracing::warn!( - "Debouncing is disabled because workers are behind the minimum required version 1.566.0. \ - Please update workers to enable debouncing feature." - ); - } - // === DEPENDENCY JOB DEBOUNCING === - // - // Debouncing consolidates multiple dependency job requests into a single execution, - // reducing redundant work when many scripts/flows/apps are updated simultaneously. - // - // Prerequisites for debouncing (all must be true): - // 1. Job is scheduled in the future (debounce_delay is not None) - provides consolidation window - // 2. Job is a dependency job - // 3. Object path is provided (script/flow/app path) - // 4. Fallback mode is disabled (normal operation) - // 5. min version supports debouncing - // 6. Job was created by relative imports (triggered by dependency chain) - // - // How it works: - // - // PHASE 1 - PUSH (in jobs.rs::push): - // When a dependency job is scheduled with delay, check debounce_key table - // - If key exists: Merge request into existing job, accumulate nodes/components - // - If key doesn't exist: Create new entry and store initial nodes/components - // - // PHASE 2 - ACCUMULATION: - // During the debounce window (typically 5-15 seconds), multiple requests merge - // - Each request adds nodes/components to debounce_stale_data table - // - SQL DISTINCT automatically removes duplicates during merge - // - // PHASE 3 - PULL (in jobs.rs::pull): - // When the delayed job finally executes: - // - Lock debounce_key FOR UPDATE to prevent races - // - Retrieve all accumulated nodes/components from debounce_stale_data - // - Process all collected dependencies in single execution - // - Clean up both debounce_key and debounce_stale_data entries - (true, true, Some(obj_path), false, true, true) => { - // Generate unique debounce key: "workspace_id:object_path:dependency" - // This ensures each workspace+path combination has independent debounce window - let debounce_key = format!("{workspace_id}:{obj_path}:dependency"); - - tracing::debug!( - workspace_id = %workspace_id, - object_path = %obj_path, - debounce_key = %debounce_key, - "Checking for existing debounced dependency job" - ); - - // Check if there's already a pending job registered for this debounce key - // The debounce_job_id_o is passed in by the caller after locking the key FOR UPDATE - // IMPORTANT: This is assumed that the caller will lock debounce_key row in this transaction. - // We do this to block puller from further actions until we are done with consolidation and stuff that we do here in push. - if let Some(debounce_job_id) = debounce_job_id_o { - tracing::debug!( - existing_job_id = %debounce_job_id, - new_job_id = %job_id, - "Found existing debounced job, merging this request" - ); - - // NOTE: Race condition handling: - // In rare cases, the debounce_key entry may still exist even though the job - // has been pulled and is running. This can happen because: - // - Job pull marks job as running first - // - Then debounce_key cleanup happens (without transaction for performance) - // - Between these steps, new requests might see the old debounce_key - // - // This is acceptable because the puller will be blocked and cannot proceed until this transaction finishes. - // This will give us some space to add consolidated data (if such) and debounce the request. - // Once tx is commited, the puller will be unblocked and continue execution. - // Accumulate the nodes/components that need relocking from this request - - // This ensures all dependency updates are handled even if jobs are debounced - if let Some(to_relock) = extract_to_relock_from_args(&args.args) { - tracing::debug!( - job_id = %debounce_job_id, - node_count = to_relock.len(), - nodes = ?to_relock, - "Accumulating nodes/components to existing debounced job" - ); - - accumulate_debounce_stale_data(&mut tx, &debounce_job_id, &to_relock) - .await - .map_err(|e| { - tracing::error!( - error = %e, - job_id = %debounce_job_id, - debounce_key = %debounce_key, - "Failed to accumulate stale data for debounced job" - ); - e - })?; - } else { - tracing::trace!( - job_id = %debounce_job_id, - "No nodes to relock in this request, skipping accumulation" - ); - } - - // Return the existing job ID, effectively debouncing this request - // The new job_id we generated won't be used - tracing::debug!( - returned_job_id = %debounce_job_id, - skipped_job_id = %job_id, - "Debounced: returning existing job ID instead of creating new job" - ); - - // We will skip some of the work downstream and just debounce the job. - return Ok((debounce_job_id, tx)); - } else { - // No existing debounce entry - this is the first request in the debounce window - tracing::debug!( - job_id = %job_id, - debounce_key = %debounce_key, - "Creating new debounce entry (first request in window)" - ); - - sqlx::query!( - "INSERT INTO debounce_key (key, job_id) VALUES ($1, $2) ON CONFLICT (key) DO UPDATE SET job_id = EXCLUDED.job_id", - &debounce_key, - job_id, - ) - .execute(&mut *tx) - .await - .map_err(|e| { - tracing::error!( - error = %e, - debounce_key = %debounce_key, - job_id = %job_id, - "Failed to insert debounce_key entry" - ); - Error::InternalErr(format!("Failed to create debounce entry: {}", e)) - })?; - - // Store initial nodes/components to relock if provided - if let Some(to_relock) = extract_to_relock_from_args(&args.args) { - tracing::debug!( - job_id = %job_id, - node_count = to_relock.len(), - nodes = ?to_relock, - "Storing initial nodes/components for new debounced job" - ); - - accumulate_debounce_stale_data(&mut tx, &job_id, &to_relock) - .await - .map_err(|e| { - tracing::error!( - error = %e, - job_id = %job_id, - "Failed to store initial stale data for debounced job" - ); - e - })?; - } else { - tracing::trace!( - job_id = %job_id, - "No initial nodes to relock, debounce entry created without stale data" - ); - } - } - } - _ => { - // Debouncing not applicable - proceed with normal job creation - tracing::trace!( - job_id = %job_id, - job_kind = ?job_kind, - "Debouncing conditions not met, proceeding with normal job creation" - ); - } - }; - - #[cfg(all(feature = "enterprise", feature = "private"))] - if schedule_path.is_none() { - if let Some(debounced_job_id) = crate::jobs_ee::maybe_apply_debouncing( - &job_id, - debouncing_settings.debounce_delay_s, - debouncing_settings.debounce_key.clone(), - workspace_id, - runnable_path.clone(), - &job_kind, - &args, + #[cfg(feature = "private")] + if schedule_path.is_none() && flow_step_id.is_none() { + crate::jobs_ee::maybe_debounce( + &debouncing_settings, &mut scheduled_for_o, + &runnable_path, + workspace_id, + job_kind, + job_id, + &args, &mut tx, + _db, ) .await? - { - return Ok((debounced_job_id, tx)); - } } let (trigger_path, trigger_kind) = trigger diff --git a/backend/windmill-queue/src/schedule.rs b/backend/windmill-queue/src/schedule.rs index 31ef18546b..6a9ea1be82 100644 --- a/backend/windmill-queue/src/schedule.rs +++ b/backend/windmill-queue/src/schedule.rs @@ -513,7 +513,6 @@ pub async fn push_scheduled_job<'c>( push_authed, false, None, - None, Some(TriggerMetadata::new( Some(schedule.path.clone()), JobTriggerKind::Schedule, diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index a0755aeee7..638129e34d 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -459,7 +459,6 @@ async fn execute_windmill_tool( None, None, None, - None, ) .await?; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 73347eb2bc..36ec01b2fb 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -986,6 +986,7 @@ pub fn start_interactive_worker_shell( let query = ("".to_string(), make_pull_query(&[common_worker_prefix])); #[cfg(feature = "benchmark")] let mut bench = windmill_common::bench::BenchmarkIter::new(); + let job = pull( &db, false, @@ -996,22 +997,23 @@ pub fn start_interactive_worker_shell( ) .await; + use PulledJobResultToJobErr::*; match job { - Ok(j) => { - match j.to_pulled_job() { - Ok(j) => Ok(j - .clone() - .map(|job| NextJob::Sql { flow_runners: None, job })), - Err(PulledJobResultToJobErr::MissingConcurrencyKey(jc)) - | Err(PulledJobResultToJobErr::ErrorWhilePreprocessing(jc)) => { - if let Err(err) = job_completed_tx.send_job(jc, true).await - { - tracing::error!("An error occurred while sending job completed: {:#?}", err) - } - Ok(None) + Ok(j) => match j.to_pulled_job() { + Ok(j) => Ok(j + .clone() + .map(|job| NextJob::Sql { flow_runners: None, job })), + Err(MissingConcurrencyKey(jc)) + | Err(ErrorWhilePreprocessing(jc)) => { + if let Err(err) = job_completed_tx.send_job(jc, true).await { + tracing::error!( + "An error occurred while sending job completed: {:#?}", + err + ) } + Ok(None) } - } + }, Err(err) => Err(err), } } @@ -1792,7 +1794,7 @@ pub async fn run_worker( if let Err(e) = timeout( // Will fail if longer than 10 seconds core::time::Duration::from_secs(10), - pulled_job_res.preprocess(db), + pulled_job_res.maybe_apply_debouncing(db), ) .warn_after_seconds(2) .await diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 2a2d789c41..0a4cb01339 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -3412,7 +3412,6 @@ async fn push_next_flow_job( None, None, None, - None, ) .warn_after_seconds(2) .await?; diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 0e6e19bd6b..b7b11cf6af 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -21,11 +21,14 @@ use windmill_common::error::Error; use windmill_common::error::Result; use windmill_common::flows::{FlowModule, FlowModuleValue, FlowNodeId}; use windmill_common::jobs::JobPayload; +use windmill_common::runnable_settings::DebouncingSettings; use windmill_common::scripts::ScriptHash; use windmill_common::utils::WarnAfterExt; #[cfg(feature = "python")] use windmill_common::worker::PythonAnnotations; -use windmill_common::worker::{to_raw_value, to_raw_value_owned, write_file, Connection}; +use windmill_common::worker::{ + to_raw_value, to_raw_value_owned, write_file, Connection, MIN_VERSION_SUPPORTS_DEBOUNCING_V2, +}; use windmill_common::workspace_dependencies::{ RawWorkspaceDependencies, WorkspaceDependencies, WorkspaceDependenciesPrefetched, }; @@ -44,7 +47,10 @@ use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; #[cfg(feature = "python")] use windmill_parser_py_imports::parse_relative_imports; use windmill_parser_ts::parse_expr_for_imports; -use windmill_queue::{append_logs, CanceledBy, MiniPulledJob, PushIsolationLevel}; +use windmill_queue::{ + append_logs, CanceledBy, MiniPulledJob, PushIsolationLevel, + WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT, +}; // TODO: To be removed in future versions lazy_static::lazy_static! { @@ -497,27 +503,14 @@ pub async fn trigger_dependents_to_recompute_dependencies( args.insert( "triggered_by_relative_import".to_string(), - to_raw_value(&()), + to_raw_value(&true), ); - // Lock the debounce_key entry FOR UPDATE to coordinate with the push side. - // This prevents concurrent modifications during dependency job scheduling. - // - // The lock serves two purposes: - // 1. Ensures we get the current debounce_job_id atomically - // 2. Blocks new push requests from modifying this key until we commit - // 3. Blocks puller from actually starting the job and gives us a chance to still squeeze the debounce in. - // - // After our transaction commits, any pending push/pull requests can proceed with - // their debounce logic. - let debounce_job_id_o = - windmill_common::jobs::lock_debounce_key(w_id, &importer_path, &mut tx).await?; - - tracing::debug!( - debounce_job_id = ?debounce_job_id_o, - importer_path = %importer_path, - "Retrieved debounce job ID (if exists)" - ); + let mut debouncing_settings = DebouncingSettings { + debounce_key: Some(format!("{w_id}:{importer_path}:dependency")), + debounce_delay_s: Some(5), + ..Default::default() + }; let job_payload = match importer_kind.as_str() { // TODO: Make it query only non-archived @@ -541,6 +534,7 @@ pub async fn trigger_dependents_to_recompute_dependencies( hash: ScriptHash(hash), language: info.language, dedicated_worker: info.dedicated_worker, + debouncing_settings, } } None => { @@ -575,10 +569,13 @@ pub async fn trigger_dependents_to_recompute_dependencies( to_raw_value(&importer_node_ids), ); + debouncing_settings.debounce_args_to_accumulate = Some(vec!["nodes_to_relock".into()]); + JobPayload::FlowDependencies { path: importer_path.clone(), version, dedicated_worker: None, + debouncing_settings, } } None => { @@ -608,7 +605,9 @@ pub async fn trigger_dependents_to_recompute_dependencies( to_raw_value(importer_node_ids), ); - JobPayload::AppDependencies { path: importer_path.clone(), version } + debouncing_settings.debounce_args_to_accumulate = Some(vec!["components_to_relock".into()]); + + JobPayload::AppDependencies { path: importer_path.clone(), version, debouncing_settings } } None => { ScopedDependencyMap::clear_map_for_item(importer_path, w_id, "app", tx, &None) @@ -658,7 +657,6 @@ pub async fn trigger_dependents_to_recompute_dependencies( None, false, None, - debounce_job_id_o, None, None, ) @@ -1457,8 +1455,10 @@ async fn lock_modules<'c>( } } else { if lock.as_ref().is_some_and(|x| !x.trim().is_empty()) { - let skip_creating_new_lock = skip_creating_new_lock(&language, &content); - if skip_creating_new_lock { + if skip_creating_new_lock(&language, &content) + && (*MIN_VERSION_SUPPORTS_DEBOUNCING_V2.read().await + || *WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT) + { tx = dependency_map .patch(get_references(), e.id.clone(), tx) .await?; @@ -1978,7 +1978,10 @@ async fn lock_modules_app( .get("lock") .is_some_and(|x| !x.as_str().unwrap().trim().is_empty()) { - if skip_creating_new_lock(&language, &content) { + if skip_creating_new_lock(&language, &content) + && (*MIN_VERSION_SUPPORTS_DEBOUNCING_V2.read().await + || *WMDEBUG_FORCE_NO_LEGACY_DEBOUNCING_COMPAT) + { dependency_map .patch( referenced_paths.clone(), diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index e4f8655157..40240199bc 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -522,6 +522,12 @@ concurrency_time_window_s: script.concurrency_time_window_s, debounce_key: emptyString(script.debounce_key) ? undefined : script.debounce_key, debounce_delay_s: script.debounce_delay_s, + debounce_args_to_accumulate: + script.debounce_args_to_accumulate && script.debounce_args_to_accumulate.length > 0 + ? script.debounce_args_to_accumulate + : undefined, + max_total_debouncing_time: script.max_total_debouncing_time, + max_total_debounces_amount: script.max_total_debounces_amount, cache_ttl: script.cache_ttl, cache_ignore_s3_path: script.cache_ignore_s3_path, ws_error_handler_muted: script.ws_error_handler_muted, @@ -666,6 +672,12 @@ concurrency_time_window_s: script.concurrency_time_window_s, debounce_key: emptyString(script.debounce_key) ? undefined : script.debounce_key, debounce_delay_s: script.debounce_delay_s, + debounce_args_to_accumulate: + script.debounce_args_to_accumulate && script.debounce_args_to_accumulate.length > 0 + ? script.debounce_args_to_accumulate + : undefined, + max_total_debouncing_time: script.max_total_debouncing_time, + max_total_debounces_amount: script.max_total_debounces_amount, cache_ttl: script.cache_ttl, cache_ignore_s3_path: script.cache_ignore_s3_path, ws_error_handler_muted: script.ws_error_handler_muted, @@ -1369,6 +1381,10 @@ size="sm" bind:debounce_delay_s={script.debounce_delay_s} bind:debounce_key={script.debounce_key} + bind:debounce_args_to_accumulate={script.debounce_args_to_accumulate} + bind:max_total_debouncing_time={script.max_total_debouncing_time} + bind:max_total_debounces_amount={script.max_total_debounces_amount} + schema={script.schema} placeholder={`$workspace/script/${script.path}-$args[foo]`} /> diff --git a/frontend/src/lib/components/flows/DebounceLimit.svelte b/frontend/src/lib/components/flows/DebounceLimit.svelte index b7467663cb..8c8937016d 100644 --- a/frontend/src/lib/components/flows/DebounceLimit.svelte +++ b/frontend/src/lib/components/flows/DebounceLimit.svelte @@ -4,10 +4,15 @@ import { SecondsInput } from '$lib/components/common' import { enterpriseLicense } from '$lib/stores' import Label from '../Label.svelte' + import type { Schema } from '$lib/common' let { debounce_delay_s = $bindable(), debounce_key = $bindable(), + debounce_args_to_accumulate = $bindable(), + max_total_debouncing_time = $bindable(), + max_total_debounces_amount = $bindable(), + schema = undefined, placeholder, size = 'xs', color = undefined, @@ -15,11 +20,45 @@ }: { debounce_delay_s: number | undefined debounce_key: string | undefined + debounce_args_to_accumulate: string[] | undefined + max_total_debouncing_time: number | undefined + max_total_debounces_amount: number | undefined + schema?: Schema placeholder: string size: 'xs' | 'sm' color?: 'nord' | undefined fontClass?: string } = $props() + + // Get list of array-type arguments from schema + let arrayArgs = $derived( + schema?.properties + ? Object.entries(schema.properties) + .filter(([_, prop]) => prop.type === 'array') + .map(([key, _]) => key) + : [] + ) + + // Single selected arg for UI (convert between single selection and array) + let selectedArg = $state(debounce_args_to_accumulate?.[0] ?? '') + + $effect(() => { + if (selectedArg && selectedArg !== '') { + debounce_args_to_accumulate = [selectedArg] + } else { + debounce_args_to_accumulate = undefined + } + }) + + // Convert 0 to undefined for max values + $effect(() => { + if (max_total_debouncing_time === 0) { + max_total_debouncing_time = undefined + } + if (max_total_debounces_amount === 0) { + max_total_debounces_amount = undefined + } + })
@@ -69,6 +108,44 @@ {placeholder} /> + + +
{/if} diff --git a/frontend/src/lib/components/flows/content/FlowSettings.svelte b/frontend/src/lib/components/flows/content/FlowSettings.svelte index c79f9427e4..68774d936f 100644 --- a/frontend/src/lib/components/flows/content/FlowSettings.svelte +++ b/frontend/src/lib/components/flows/content/FlowSettings.svelte @@ -477,6 +477,10 @@ fontClass="font-medium" bind:debounce_delay_s={flowStore.val.value.debounce_delay_s} bind:debounce_key={flowStore.val.value.debounce_key} + bind:debounce_args_to_accumulate={flowStore.val.value.debounce_args_to_accumulate} + bind:max_total_debouncing_time={flowStore.val.value.max_total_debouncing_time} + bind:max_total_debounces_amount={flowStore.val.value.max_total_debounces_amount} + schema={flowStore.val.schema} placeholder={`$workspace/flow/${$pathStore}-$args[foo]`} /> {/if} diff --git a/frontend/src/lib/components/runs/RunRow.svelte b/frontend/src/lib/components/runs/RunRow.svelte index 470ff0f56a..f088d26e09 100644 --- a/frontend/src/lib/components/runs/RunRow.svelte +++ b/frontend/src/lib/components/runs/RunRow.svelte @@ -26,7 +26,8 @@ ListFilterPlus, Play, ShieldQuestion, - X + X, + Layers } from 'lucide-svelte' import { createEventDispatcher } from 'svelte' import TimeAgo from '../TimeAgo.svelte' diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index e1c3af382f..cdf87ae187 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -72,6 +72,17 @@ components: debounce_key: type: string description: Expression to group debounced executions + debounce_args_to_accumulate: + type: array + description: Arguments to accumulate across debounced executions + items: + type: string + max_total_debouncing_time: + type: number + description: Maximum total time in seconds that a job can be debounced + max_total_debounces_amount: + type: number + description: Maximum number of times a job can be debounced skip_expr: type: string description: JavaScript expression to conditionally skip the entire flow