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

fix(debounce): atomic upsert robust to concurrent pull-time key deletion

Browse Source 
Companion to windmill-ee-private: keep upsert_debounce_key a single atomic
INSERT ... ON CONFLICT DO UPDATE so a chaining push cannot fail when the
worker pull path concurrently deletes the holder's debounce_key (the prior
read+UPDATE split could hit "no row updated"). Adds
test_debounce_push_races_key_deletion_by_pull (races a chaining push against
the key deletion 50x, asserts the push never errors) and refreshes the SQLx
cache.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-06-25 13:13:12 +00:00
co-authored by Claude Opus 4.8
parent 1c3b2041dd
commit f336674312
9 changed files with 172 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-76774e6f72c8c8b7473487e4176dc17b17372b7292e39d3888a93ff4fe49e4f5.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM debounce_key WHERE key = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "76774e6f72c8c8b7473487e4176dc17b17372b7292e39d3888a93ff4fe49e4f5"
}
backend/.sqlx/query-830be634b82fe4c839df8015ca05a94ff3ba5204fa8bdbe0fcbfd230e6eaf72a.json
Generated
-15
View File
@@ -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"
}
backend/.sqlx/query-a938f959c63a4f8bf5b355f06e2c1b66824819d97d9bcbe3ab0b9b9fc41c4b1a.json
Generated
-29
View File
@@ -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"
}
backend/.sqlx/query-c91cb3e73d160b65bc2a055d279d17d07a743969a3a0e47abed3539e10c35bf5.json
Generated
-28
View File
@@ -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"
}
backend/.sqlx/query-ca5d392ab0684f2622fbcd514d64bc10f72fd9c2078c4ae5e398ad29320bf06a.json
Generated
-30
View File
@@ -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"
}
backend/.sqlx/query-f04f84ac63e98566311245a40a8dd83166a32ae4370a4f1ca622b8535c930ea2.json
Generated
+35
View File
@@ -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"
}
backend/.sqlx/query-f44595adab2d127c83f9829b74429dcb7e5f2513bc59c2bdf7dd0be1864cd2c5.json
Generated
+35
View File
@@ -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"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
8968cca45c56e605dc68dea4d93d53c648ec3b5f
aa82e3f706803380b36c7950ff82297a76a71e2e
backend/windmill-queue/tests/debounce_test.rs
+87
View File
@@ -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<Postgres>,
) -> 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<String, Box<RawValue>> =
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)