diff --git a/backend/.sqlx/query-76774e6f72c8c8b7473487e4176dc17b17372b7292e39d3888a93ff4fe49e4f5.json b/backend/.sqlx/query-76774e6f72c8c8b7473487e4176dc17b17372b7292e39d3888a93ff4fe49e4f5.json new file mode 100644 index 0000000000..694ed1887f --- /dev/null +++ b/backend/.sqlx/query-76774e6f72c8c8b7473487e4176dc17b17372b7292e39d3888a93ff4fe49e4f5.json @@ -0,0 +1,14 @@ +{ + "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-830be634b82fe4c839df8015ca05a94ff3ba5204fa8bdbe0fcbfd230e6eaf72a.json b/backend/.sqlx/query-830be634b82fe4c839df8015ca05a94ff3ba5204fa8bdbe0fcbfd230e6eaf72a.json deleted file mode 100644 index a9eba04b45..0000000000 --- a/backend/.sqlx/query-830be634b82fe4c839df8015ca05a94ff3ba5204fa8bdbe0fcbfd230e6eaf72a.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO v2_job_debounce_batch (id, debounce_batch)\n SELECT $1, COALESCE(\n (SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $2 LIMIT 1),\n nextval('debounce_batch_seq'))", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "830be634b82fe4c839df8015ca05a94ff3ba5204fa8bdbe0fcbfd230e6eaf72a" -} diff --git a/backend/.sqlx/query-a938f959c63a4f8bf5b355f06e2c1b66824819d97d9bcbe3ab0b9b9fc41c4b1a.json b/backend/.sqlx/query-a938f959c63a4f8bf5b355f06e2c1b66824819d97d9bcbe3ab0b9b9fc41c4b1a.json deleted file mode 100644 index 666fc190dc..0000000000 --- a/backend/.sqlx/query-a938f959c63a4f8bf5b355f06e2c1b66824819d97d9bcbe3ab0b9b9fc41c4b1a.json +++ /dev/null @@ -1,29 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO debounce_key (job_id, key)\n VALUES ($1, $2)\n ON CONFLICT (key) DO UPDATE SET\n job_id = EXCLUDED.job_id,\n previous_job_id = NULL,\n debounced_times = 0,\n first_started_at = now()\n RETURNING debounced_times, first_started_at", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "debounced_times", - "type_info": "Int4" - }, - { - "ordinal": 1, - "name": "first_started_at", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Varchar" - ] - }, - "nullable": [ - false, - false - ] - }, - "hash": "a938f959c63a4f8bf5b355f06e2c1b66824819d97d9bcbe3ab0b9b9fc41c4b1a" -} diff --git a/backend/.sqlx/query-c91cb3e73d160b65bc2a055d279d17d07a743969a3a0e47abed3539e10c35bf5.json b/backend/.sqlx/query-c91cb3e73d160b65bc2a055d279d17d07a743969a3a0e47abed3539e10c35bf5.json deleted file mode 100644 index cd0a9e1f47..0000000000 --- a/backend/.sqlx/query-c91cb3e73d160b65bc2a055d279d17d07a743969a3a0e47abed3539e10c35bf5.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT dk.job_id, COALESCE(q.running, false) AS \"running!\"\n FROM debounce_key dk\n LEFT JOIN v2_job_queue q ON q.id = dk.job_id\n WHERE dk.key = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "job_id", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "running!", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Text" - ] - }, - "nullable": [ - false, - null - ] - }, - "hash": "c91cb3e73d160b65bc2a055d279d17d07a743969a3a0e47abed3539e10c35bf5" -} diff --git a/backend/.sqlx/query-ca5d392ab0684f2622fbcd514d64bc10f72fd9c2078c4ae5e398ad29320bf06a.json b/backend/.sqlx/query-ca5d392ab0684f2622fbcd514d64bc10f72fd9c2078c4ae5e398ad29320bf06a.json deleted file mode 100644 index 60bd4084b3..0000000000 --- a/backend/.sqlx/query-ca5d392ab0684f2622fbcd514d64bc10f72fd9c2078c4ae5e398ad29320bf06a.json +++ /dev/null @@ -1,30 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE debounce_key\n SET job_id = $1, previous_job_id = $3, debounced_times = debounced_times + 1\n WHERE key = $2\n RETURNING debounced_times, first_started_at", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "debounced_times", - "type_info": "Int4" - }, - { - "ordinal": 1, - "name": "first_started_at", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text", - "Uuid" - ] - }, - "nullable": [ - false, - false - ] - }, - "hash": "ca5d392ab0684f2622fbcd514d64bc10f72fd9c2078c4ae5e398ad29320bf06a" -} diff --git a/backend/.sqlx/query-f04f84ac63e98566311245a40a8dd83166a32ae4370a4f1ca622b8535c930ea2.json b/backend/.sqlx/query-f04f84ac63e98566311245a40a8dd83166a32ae4370a4f1ca622b8535c930ea2.json new file mode 100644 index 0000000000..1039dd3950 --- /dev/null +++ b/backend/.sqlx/query-f04f84ac63e98566311245a40a8dd83166a32ae4370a4f1ca622b8535c930ea2.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH dk AS (\n INSERT INTO debounce_key (job_id, key)\n VALUES ($1, $2)\n ON CONFLICT (key)\n DO UPDATE SET\n previous_job_id = CASE WHEN EXISTS\n (SELECT 1 FROM v2_job_queue q WHERE q.id = debounce_key.job_id AND q.running)\n THEN NULL ELSE debounce_key.job_id END,\n job_id = EXCLUDED.job_id,\n debounced_times = CASE WHEN EXISTS\n (SELECT 1 FROM v2_job_queue q WHERE q.id = debounce_key.job_id AND q.running)\n THEN 0 ELSE debounce_key.debounced_times + 1 END,\n first_started_at = CASE WHEN EXISTS\n (SELECT 1 FROM v2_job_queue q WHERE q.id = debounce_key.job_id AND q.running)\n THEN now() ELSE debounce_key.first_started_at END\n RETURNING debounced_times, first_started_at, previous_job_id AS job_id_to_debounce\n ), _batch AS (\n INSERT INTO v2_job_debounce_batch (id, debounce_batch)\n SELECT $1, COALESCE(\n (SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = dk.job_id_to_debounce LIMIT 1),\n nextval('debounce_batch_seq'))\n FROM dk\n )\n SELECT debounced_times, first_started_at, job_id_to_debounce FROM dk\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": "f04f84ac63e98566311245a40a8dd83166a32ae4370a4f1ca622b8535c930ea2" +} diff --git a/backend/.sqlx/query-f44595adab2d127c83f9829b74429dcb7e5f2513bc59c2bdf7dd0be1864cd2c5.json b/backend/.sqlx/query-f44595adab2d127c83f9829b74429dcb7e5f2513bc59c2bdf7dd0be1864cd2c5.json new file mode 100644 index 0000000000..81b34464f4 --- /dev/null +++ b/backend/.sqlx/query-f44595adab2d127c83f9829b74429dcb7e5f2513bc59c2bdf7dd0be1864cd2c5.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 = CASE WHEN EXISTS\n (SELECT 1 FROM v2_job_queue q WHERE q.id = debounce_key.job_id AND q.running)\n THEN NULL ELSE debounce_key.job_id END,\n job_id = EXCLUDED.job_id,\n debounced_times = CASE WHEN EXISTS\n (SELECT 1 FROM v2_job_queue q WHERE q.id = debounce_key.job_id AND q.running)\n THEN 0 ELSE debounce_key.debounced_times + 1 END,\n first_started_at = CASE WHEN EXISTS\n (SELECT 1 FROM v2_job_queue q WHERE q.id = debounce_key.job_id AND q.running)\n THEN now() ELSE debounce_key.first_started_at END\n RETURNING debounced_times, first_started_at, 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": "f44595adab2d127c83f9829b74429dcb7e5f2513bc59c2bdf7dd0be1864cd2c5" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 9372eb37d7..793a6f9440 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -8968cca45c56e605dc68dea4d93d53c648ec3b5f \ No newline at end of file +aa82e3f706803380b36c7950ff82297a76a71e2e \ No newline at end of file diff --git a/backend/windmill-queue/tests/debounce_test.rs b/backend/windmill-queue/tests/debounce_test.rs index ed9b1e778b..23b1e9e98b 100644 --- a/backend/windmill-queue/tests/debounce_test.rs +++ b/backend/windmill-queue/tests/debounce_test.rs @@ -4488,6 +4488,93 @@ mod debounce { Ok(()) } + /// Regression: a push that would chain onto a queued holder must not error if the + /// worker pull path concurrently deletes that holder's debounce_key + /// (`DELETE ... WHERE job_id = ...`, which does NOT take the push advisory lock). + /// The upsert is a single atomic `INSERT ... ON CONFLICT`, so a deleted holder simply + /// yields a fresh window rather than a "no row updated" failure. Races the two and + /// asserts the push always succeeds and leaves a consistent key. + #[sqlx::test(migrations = "../migrations", fixtures("base"))] + async fn test_debounce_push_races_key_deletion_by_pull( + db: Pool, + ) -> anyhow::Result<()> { + let key = "push_vs_pull_key"; + let settings = DebouncingSettings { + debounce_delay_s: Some(5), + debounce_key: Some(key.to_string()), + debounce_args_to_accumulate: Some(vec!["items".to_string()]), + ..Default::default() + }; + let rs_handle = setup_debouncing_settings(&db, &settings).await; + + // 50 rounds to give the interleaving a chance to land in the read/write window + // that the old split read+UPDATE path would have failed on. + for round in 0..50 { + sqlx::query!("DELETE FROM debounce_key WHERE key = $1", key) + .execute(&db) + .await?; + let holder = Uuid::new_v4(); + push_debounced_script(&db, holder, vec![round], &settings, rs_handle).await; + + let late = Uuid::new_v4(); + let late_args = serde_json::json!({ "items": [round * 1000] }); + insert_script_job_with_args(&db, late, "test-workspace", "f/test/script", &late_args) + .await; + sqlx::query!( + "UPDATE v2_job_queue SET runnable_settings_handle = $1 WHERE id = $2", + rs_handle, + late, + ) + .execute(&db) + .await?; + + // Race: push the late arrival (chains onto `holder`) against the worker pull + // cleanup deleting `holder`'s key. + let push = { + let db = db.clone(); + let settings = settings.clone(); + async move { + let args_hm: HashMap> = + serde_json::from_value(serde_json::json!({ "items": [round * 1000] })) + .unwrap(); + let push_args = PushArgs::from(&args_hm); + let mut scheduled_for = None; + let mut tx = db.begin().await.unwrap(); + let res = windmill_queue::jobs_ee::maybe_debounce( + &settings, + &mut scheduled_for, + &Some("f/test/script".to_string()), + "test-workspace", + JobKind::Script, + late, + &push_args, + &mut tx, + ) + .await; + if res.is_ok() { + tx.commit().await.unwrap(); + } + res + } + }; + let delete = { + let db = db.clone(); + async move { + sqlx::query!("DELETE FROM debounce_key WHERE job_id = $1", holder) + .execute(&db) + .await + } + }; + let (push_res, _) = tokio::join!(push, delete); + assert!( + push_res.is_ok(), + "round {round}: push must not error when the holder key is concurrently deleted: {push_res:?}" + ); + } + + Ok(()) + } + /// Test: Push-time (script) debounce with max_total_debounces_amount=2. /// 5 calls, each sending {x: [i]}. Expected: /// Call 1: debounced (scheduled_for set)