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

feat(debounce): claim-based exactly-once batch consumption

Browse Source 
Eliminates the rare duplicate/loss when two survivors land on one debounce
batch (a narrow push/pull race), without locking the worker pull hot path.

- migration: v2_job_debounce_batch gains consumed_at + consumed_by.
- pull side (maybe_apply_debouncing): instead of deleting the batch on consume,
  a survivor atomically claims its own row + any unclaimed siblings (stamping
  consumed_by = itself) and accumulates exactly the rows it claimed. A second
  survivor of the same batch finds its row already consumed by another job and
  runs empty (no duplicate); a re-pulled survivor recognizes its own prior claim
  and keeps its accumulated args; a never-batched job (CE/legacy) keeps its own
  args. Non-accumulate debounce paths still hard-delete their batch rows.
- complete_debounced_job (EE companion) never completes a running predecessor,
  so its in-flight run is not killed (no loss); the claim then prevents the
  duplicate the guard would otherwise allow.
- monitor: GC sweep deletes consumed batch rows past a 1h grace.

Together with the running-survivor guard this makes debounce accumulation
exactly-once. Adds tests: batch_consumed_exactly_once, repull_keeps_accumulated.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-06-25 14:16:00 +00:00
co-authored by Claude Opus 4.8
parent f336674312
commit 320984b92e
12 changed files with 472 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0d6c3399bc637e2c599534c2cafacf623cd6e5fde2d075a4f1f0d4852efcedca.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT args->>$2 FROM v2_job WHERE id = ANY($1)",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "?column?",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"UuidArray",
"Text"
]
},
"nullable": [
null
]
},
"hash": "0d6c3399bc637e2c599534c2cafacf623cd6e5fde2d075a4f1f0d4852efcedca"
}
backend/.sqlx/query-57f375e89d63ac118c5c6767487af401e6d800488ade04facb6af14578c30a89.json
Generated
+40
View File
@@ -0,0 +1,40 @@
{
"db_name": "PostgreSQL",
"query": "WITH mine AS (\n SELECT debounce_batch, consumed_by FROM v2_job_debounce_batch WHERE id = $1\n ), claim_self AS (\n UPDATE v2_job_debounce_batch SET consumed_at = now(), consumed_by = $1\n WHERE id = $1 AND consumed_at IS NULL\n RETURNING debounce_batch\n ), claim_rest AS (\n UPDATE v2_job_debounce_batch SET consumed_at = now(), consumed_by = $1\n WHERE debounce_batch = (SELECT debounce_batch FROM claim_self)\n AND id <> $1 AND consumed_at IS NULL\n RETURNING id\n )\n SELECT\n EXISTS (SELECT 1 FROM mine) AS \"had_row!\",\n (SELECT debounce_batch FROM claim_self) AS claimed_batch,\n (SELECT consumed_by FROM mine) AS prev_consumed_by,\n ARRAY(SELECT id FROM claim_rest) AS \"claimed_ids!\"\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "had_row!",
"type_info": "Bool"
},
{
"ordinal": 1,
"name": "claimed_batch",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "prev_consumed_by",
"type_info": "Uuid"
},
{
"ordinal": 3,
"name": "claimed_ids!",
"type_info": "UuidArray"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
null,
null,
null,
null
]
},
"hash": "57f375e89d63ac118c5c6767487af401e6d800488ade04facb6af14578c30a89"
}
backend/.sqlx/query-70f01b322765442de8888b6d9b79984da751f9ae10c8b15ada925bd371501e18.json
Generated
+20
View File
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "WITH del AS (\n DELETE FROM v2_job_debounce_batch\n WHERE consumed_at IS NOT NULL AND consumed_at < now() - interval '1 hour'\n RETURNING 1\n ) SELECT count(*) FROM del",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "count",
"type_info": "Int8"
}
],
"parameters": {
"Left": []
},
"nullable": [
null
]
},
"hash": "70f01b322765442de8888b6d9b79984da751f9ae10c8b15ada925bd371501e18"
}
backend/.sqlx/query-b9ce07cda12e2540f428d4c8e5bbfa8763c03dfe3f0c74edf958667035fb5135.json
Generated
+25
View File
@@ -0,0 +1,25 @@
{
"db_name": "PostgreSQL",
"query": "\n WITH target AS (\n SELECT id FROM v2_job_queue WHERE id = $1 AND NOT running FOR UPDATE\n ), completed AS (\n INSERT INTO v2_job_completed\n (workspace_id, id, started_at, duration_ms, result,\n flow_status, workflow_as_code_status, status, worker)\n SELECT\n q.workspace_id, q.id, q.started_at,\n (EXTRACT('epoch' FROM now()) - EXTRACT('epoch' FROM COALESCE(q.started_at, now()))) * 1000,\n $3::text::jsonb,\n s.flow_status,\n s.workflow_as_code_status,\n 'skipped'::job_status,\n q.worker\n FROM v2_job_queue q\n LEFT JOIN v2_job_status s ON s.id = q.id\n WHERE q.id IN (SELECT id FROM target)\n ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = EXCLUDED.result\n RETURNING 1 AS x\n ), _deleted AS (\n DELETE FROM v2_job_queue WHERE id IN (SELECT id FROM target)\n ), _logged AS (\n INSERT INTO job_logs (logs, job_id, workspace_id)\n SELECT $4, $1, $2 WHERE EXISTS (SELECT 1 FROM target)\n ON CONFLICT (job_id) DO UPDATE SET logs = concat(job_logs.logs, EXCLUDED.logs)\n )\n SELECT x FROM completed\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "x",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Uuid",
"Varchar",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "b9ce07cda12e2540f428d4c8e5bbfa8763c03dfe3f0c74edf958667035fb5135"
}
backend/.sqlx/query-bf5db4bba06a2af085577f36446bcb900ac113c14584fa933d2a94318e47a354.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO v2_job_debounce_batch (id, debounce_batch)\n SELECT $1, debounce_batch FROM v2_job_debounce_batch WHERE id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Uuid"
]
},
"nullable": []
},
"hash": "bf5db4bba06a2af085577f36446bcb900ac113c14584fa933d2a94318e47a354"
}
backend/.sqlx/query-e050734d7642b26f8859982c55ec2c8b1fc14a8de665b15a2f2dfd6e0b5b7fdf.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM v2_job_debounce_batch WHERE debounce_batch = (\n SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1\n )",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": []
},
"hash": "e050734d7642b26f8859982c55ec2c8b1fc14a8de665b15a2f2dfd6e0b5b7fdf"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
aa82e3f706803380b36c7950ff82297a76a71e2e
30d740e619fad219108ec4b4c6a9d67c1ab42d46
backend/migrations/20260625135355_debounce_batch_consumed.down.sql
+4
View File
@@ -0,0 +1,4 @@
DROP INDEX IF EXISTS idx_v2_job_debounce_batch_consumed_at;
ALTER TABLE v2_job_debounce_batch
DROP COLUMN IF EXISTS consumed_at,
DROP COLUMN IF EXISTS consumed_by;
backend/migrations/20260625135355_debounce_batch_consumed.up.sql
+17
View File
@@ -0,0 +1,17 @@
-- Claim-based, exactly-once consumption of debounce batches.
-- A batch row is "claimed" by the survivor that accumulates its args. Stamping the
-- row consumed (instead of deleting it) lets a later-pulled survivor of the same
-- batch tell "my contribution was already processed" (consumed_at set -> no-op)
-- apart from "I was never batched" (no row at all; CE / legacy -> run my own args).
-- consumed_at doubles as the GC timestamp. NULL = not yet consumed.
-- consumed_by records which job claimed the row, so a job that is re-pulled (e.g.
-- crash recovery) can tell its own prior claim (keep its accumulated args) apart from
-- a sibling survivor having swept it in (run empty).
ALTER TABLE v2_job_debounce_batch
ADD COLUMN IF NOT EXISTS consumed_at TIMESTAMP WITH TIME ZONE,
ADD COLUMN IF NOT EXISTS consumed_by UUID;
-- Keeps the GC sweep (delete consumed rows past a grace period) cheap.
CREATE INDEX IF NOT EXISTS idx_v2_job_debounce_batch_consumed_at
ON v2_job_debounce_batch (consumed_at)
WHERE consumed_at IS NOT NULL;
backend/src/monitor.rs
+28
View File
@@ -2691,6 +2691,9 @@ pub async fn monitor_db(
if let Err(e) = cleanup_debounce_orphaned_keys(&db).await {
tracing::error!("Error cleaning up debounce keys: {:?}", e);
}
if let Err(e) = cleanup_consumed_debounce_batches(&db).await {
tracing::error!("Error cleaning up consumed debounce batches: {:?}", e);
}
}
}
};
@@ -4393,6 +4396,31 @@ RETURNING key,job_id
Ok(())
}
/// GC for claim-based debounce batches: once a batch row has been consumed (its
/// args accumulated into some survivor's run), it only lingers to let a later-pulled
/// survivor of the same batch tell "already consumed" from "never batched". A generous
/// grace period (>> any debounce window) makes that decision safe; after it, the rows
/// are dead weight. A re-pulled survivor whose row was GC'd correctly falls back to its
/// own (already-accumulated, persisted) args, so the grace period is not correctness-
/// critical.
async fn cleanup_consumed_debounce_batches(db: &DB) -> error::Result<()> {
let deleted = sqlx::query_scalar!(
"WITH del AS (
DELETE FROM v2_job_debounce_batch
WHERE consumed_at IS NOT NULL AND consumed_at < now() - interval '1 hour'
RETURNING 1
) SELECT count(*) FROM del"
)
.fetch_one(db)
.await?
.unwrap_or(0);
if deleted > 0 {
tracing::info!("Cleaned up {deleted} consumed debounce batch rows");
}
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::min_version::MIN_VERSION_SUPPORTS_RUNNABLE_SETTINGS_V0
backend/windmill-queue/src/jobs.rs
+154 -80
View File
@@ -3172,87 +3172,77 @@ impl PulledJobResult {
if let Some(args) = &mut j.args {
args.remove(field_name);
}
// No accumulation on this path: just clean up the batch rows.
sqlx::query!(
"DELETE FROM v2_job_debounce_batch WHERE debounce_batch = (
SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1
)",
j_id,
)
.execute(db)
.await?;
} 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<Box<RawValue>> = 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
)
) SELECT args->>$2 FROM ids LEFT JOIN v2_job ON v2_job.id = ids.job_id
// Claim this job's contribution to its debounce batch exactly once.
// Instead of deleting the batch rows, mark them consumed (stamping
// consumed_by = this job). A batch normally has a single survivor that
// sweeps every row; only a narrow push/pull race can leave two survivors
// on one batch. The claim lets the second survivor tell apart:
// - already swept in by the other survivor -> run empty (no duplicate),
// - its own earlier claim on re-pull -> keep its accumulated args,
// - never batched (CE / workers behind v2) -> keep its own args.
// Consumed rows are GC'd by the monitor.
let claim = sqlx::query!(
"WITH mine AS (
SELECT debounce_batch, consumed_by FROM v2_job_debounce_batch WHERE id = $1
), claim_self AS (
UPDATE v2_job_debounce_batch SET consumed_at = now(), consumed_by = $1
WHERE id = $1 AND consumed_at IS NULL
RETURNING debounce_batch
), claim_rest AS (
UPDATE v2_job_debounce_batch SET consumed_at = now(), consumed_by = $1
WHERE debounce_batch = (SELECT debounce_batch FROM claim_self)
AND id <> $1 AND consumed_at IS NULL
RETURNING id
)
SELECT
EXISTS (SELECT 1 FROM mine) AS \"had_row!\",
(SELECT debounce_batch FROM claim_self) AS claimed_batch,
(SELECT consumed_by FROM mine) AS prev_consumed_by,
ARRAY(SELECT id FROM claim_rest) AS \"claimed_ids!\"
",
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::<Vec<Box<RawValue>>>(s) {
Ok(ref mut vec) => accumulated_arg.append(vec),
Err(_) => {
// Value is not an array — wrap the scalar into a
// single-element array. This supports union types
// like T | T[] where the caller may pass a bare T.
match RawValue::from_string(s.to_string()) {
Ok(raw) => accumulated_arg.push(raw),
Err(e) => {
return Err(error::Error::ArgumentErr(format!("cannot consolidate argument `{arg_name_to_accumulate}`: value is neither a valid list nor a valid JSON value\nUnwrapped Error: {e}")));
}
}
}
}
}
}
.fetch_one(db)
.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"
);
// If the batch query returned no entries (e.g. CE where
// v2_job_debounce_batch is never populated), keep the
// original value unchanged instead of replacing it with [].
if !accumulated_arg.is_empty() {
let new_value = to_raw_value(&accumulated_arg);
let original_value = j
.args
.as_ref()
.and_then(|a| a.get(arg_name_to_accumulate))
.map(|v| v.get().to_string())
.unwrap_or_else(|| "null".to_string());
append_logs(
&j_id,
&j.workspace_id,
format!(
"Accumulating debounced argument `{arg_name_to_accumulate}`:\n original: {original_value}\n accumulated: {}\n\n",
&new_value
),
&(db.into()),
)
.await;
let consumed_by_other =
claim.claimed_batch.is_none() && claim.prev_consumed_by != Some(j_id);
if !claim.had_row {
// Never batched (CE / workers behind v2): keep the job's own args.
tracing::debug!(
job_id = %j_id,
"Debounce: no batch row, keeping original args"
);
} else if consumed_by_other {
// Another survivor of this batch already accumulated this job's
// contribution; run as a no-op so its items are not reprocessed.
tracing::info!(
job_id = %j_id,
arg_name = arg_name_to_accumulate,
"Debounce: contribution already consumed by a concurrent survivor, running empty"
);
j.args
.get_or_insert(Json(Default::default()))
.as_mut()
.insert(arg_name_to_accumulate.to_owned(), new_value);
// Persist accumulated args to v2_job so that flow steps
// re-reading from the DB (via get_mini_pulled_job) see them
.insert(
arg_name_to_accumulate.to_owned(),
to_raw_value(&Vec::<Box<RawValue>>::new()),
);
if let Some(ref args) = j.args {
sqlx::query!(
"UPDATE v2_job SET args = $2 WHERE id = $1",
@@ -3262,18 +3252,102 @@ impl PulledJobResult {
.execute(db)
.await?;
}
}
}
} else if claim.claimed_batch.is_some() {
// We claimed our own row (+ any unclaimed siblings): accumulate the
// args of exactly the rows we own.
let mut ids: Vec<Uuid> = Vec::with_capacity(claim.claimed_ids.len() + 1);
ids.push(j_id);
ids.extend(claim.claimed_ids.iter().copied());
// Clean up the debounce batch entries now that the job has been pulled
sqlx::query!(
"DELETE FROM v2_job_debounce_batch WHERE debounce_batch = (
SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1
)",
j_id,
)
.execute(db)
.await?;
tracing::debug!(
job_id = %j_id,
job_kind = ?kind,
arg_name = arg_name_to_accumulate,
claimed = ids.len(),
"Accumulating debounced arguments from claimed batch rows"
);
let mut accumulated_arg: Vec<Box<RawValue>> = vec![];
for str_o in sqlx::query_scalar!(
"SELECT args->>$2 FROM v2_job WHERE id = ANY($1)",
&ids,
arg_name_to_accumulate,
)
.fetch_all(db)
.await?
.into_iter()
{
if let Some(s) = str_o.as_ref() {
match serde_json::from_str::<Vec<Box<RawValue>>>(s) {
Ok(ref mut vec) => accumulated_arg.append(vec),
Err(_) => {
// Value is not an array — wrap the scalar into a
// single-element array. This supports union types
// like T | T[] where the caller may pass a bare T.
match RawValue::from_string(s.to_string()) {
Ok(raw) => accumulated_arg.push(raw),
Err(e) => {
return Err(error::Error::ArgumentErr(format!("cannot consolidate argument `{arg_name_to_accumulate}`: value is neither a valid list nor a valid JSON value\nUnwrapped Error: {e}")));
}
}
}
}
}
}
if !accumulated_arg.is_empty() {
let new_value = to_raw_value(&accumulated_arg);
let original_value = j
.args
.as_ref()
.and_then(|a| a.get(arg_name_to_accumulate))
.map(|v| v.get().to_string())
.unwrap_or_else(|| "null".to_string());
append_logs(
&j_id,
&j.workspace_id,
format!(
"Accumulating debounced argument `{arg_name_to_accumulate}`:\n original: {original_value}\n accumulated: {}\n\n",
&new_value
),
&(db.into()),
)
.await;
j.args
.get_or_insert(Json(Default::default()))
.as_mut()
.insert(arg_name_to_accumulate.to_owned(), new_value);
// Persist accumulated args to v2_job so that flow steps
// re-reading from the DB (via get_mini_pulled_job) see them
if let Some(ref args) = j.args {
sqlx::query!(
"UPDATE v2_job SET args = $2 WHERE id = $1",
j_id,
args as &Json<HashMap<String, Box<RawValue>>>,
)
.execute(db)
.await?;
}
}
}
// else: claimed_batch is None but it was our own prior claim (re-pull) —
// keep the args we already persisted on the first pull.
} else {
// Debounced but no args to accumulate (plain debounce / dependency job):
// consume the batch by removing this job's rows.
sqlx::query!(
"DELETE FROM v2_job_debounce_batch WHERE debounce_batch = (
SELECT debounce_batch FROM v2_job_debounce_batch WHERE id = $1
)",
j_id,
)
.execute(db)
.await?;
}
// Handle dependency job debouncing cleanup when a job is pulled for execution
if is_djob_to_debounce {
backend/windmill-queue/tests/debounce_test.rs
+131
View File
@@ -4575,6 +4575,137 @@ mod debounce {
Ok(())
}
/// Claim-based exactly-once: if two survivors end up on the same batch (only
/// possible in a narrow push/pull race), the args of each member are accumulated
/// into exactly ONE run. The survivor that claims the batch first accumulates
/// everyone; the second survivor finds its contribution already consumed and runs
/// empty — no item is dropped and none is processed twice.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_debounce_batch_consumed_exactly_once(db: Pool<Postgres>) -> anyhow::Result<()> {
let key = "exactly_once_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;
// J1 superseded, J2 the (first) survivor of batch B = {J1, J2}.
let j1 = Uuid::new_v4();
let j2 = Uuid::new_v4();
push_debounced_script(&db, j1, vec![1], &settings, rs_handle).await;
let j2_args = push_debounced_script(&db, j2, vec![2], &settings, rs_handle).await;
// Simulate the race outcome: a second survivor J3 ended up on the SAME batch B.
let j3 = Uuid::new_v4();
let j3_args = serde_json::json!({ "items": [3] });
insert_script_job_with_args(&db, j3, "test-workspace", "f/test/script", &j3_args).await;
sqlx::query!(
"UPDATE v2_job_queue SET runnable_settings_handle = $1 WHERE id = $2",
rs_handle,
j3,
)
.execute(&db)
.await?;
sqlx::query!(
"INSERT INTO v2_job_debounce_batch (id, debounce_batch)
SELECT $1, debounce_batch FROM v2_job_debounce_batch WHERE id = $2",
j3,
j2,
)
.execute(&db)
.await?;
// J2 pulled first: claims the whole batch, accumulates everyone's items.
let mut j2_res = make_pulled_job_result(
j2,
"test-workspace",
"f/test/script",
&j2_args,
JobKind::Script,
"deno",
rs_handle,
);
j2_res.maybe_apply_debouncing(&db).await?;
assert!(j2_res.job.is_some(), "J2 runs");
assert_accumulated_items(&j2_res, &[1, 2, 3], "items");
// J3 pulled next: its contribution was already consumed by J2 -> runs empty,
// so [3] is not processed a second time.
let mut j3_res = make_pulled_job_result(
j3,
"test-workspace",
"f/test/script",
&j3_args,
JobKind::Script,
"deno",
rs_handle,
);
j3_res.maybe_apply_debouncing(&db).await?;
let job = j3_res.job.as_ref().expect("J3 still runs (empty)");
let items: Vec<serde_json::Value> =
serde_json::from_str(job.job.args.as_ref().unwrap().get("items").unwrap().get())?;
assert!(
items.is_empty(),
"J3's items must be empty (already consumed by J2), got {items:?}"
);
Ok(())
}
/// A survivor re-pulled (e.g. crash recovery) must NOT mistake its own earlier
/// claim for a sibling's and wipe its accumulated args. consumed_by = self is
/// distinguished from consumed_by = another job.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_debounce_repull_keeps_accumulated(db: Pool<Postgres>) -> anyhow::Result<()> {
let key = "repull_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;
let j1 = Uuid::new_v4();
let j2 = Uuid::new_v4();
push_debounced_script(&db, j1, vec![1], &settings, rs_handle).await;
let j2_args = push_debounced_script(&db, j2, vec![2], &settings, rs_handle).await;
// First pull: J2 claims its batch and accumulates [1, 2].
let mut first = make_pulled_job_result(
j2,
"test-workspace",
"f/test/script",
&j2_args,
JobKind::Script,
"deno",
rs_handle,
);
first.maybe_apply_debouncing(&db).await?;
assert_accumulated_items(&first, &[1, 2], "items");
// Re-pull with the args persisted by the first pull: J2 sees its OWN prior claim
// (consumed_by = j2), so it keeps the accumulated args rather than running empty.
let persisted = first.job.as_ref().unwrap().job.args.as_ref().unwrap();
let persisted_json = serde_json::to_value(persisted).unwrap();
let mut second = make_pulled_job_result(
j2,
"test-workspace",
"f/test/script",
&persisted_json,
JobKind::Script,
"deno",
rs_handle,
);
second.maybe_apply_debouncing(&db).await?;
assert!(second.job.is_some(), "re-pulled J2 still runs");
assert_accumulated_items(&second, &[1, 2], "items");
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)