diff --git a/backend/.sqlx/query-12631fecee6aa11a45cf5c8d101c0dd8de50ac9b57e68198f637038d344fdd46.json b/backend/.sqlx/query-12631fecee6aa11a45cf5c8d101c0dd8de50ac9b57e68198f637038d344fdd46.json new file mode 100644 index 0000000000..c452c33018 --- /dev/null +++ b/backend/.sqlx/query-12631fecee6aa11a45cf5c8d101c0dd8de50ac9b57e68198f637038d344fdd46.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE kafka_trigger\n SET\n kafka_resource_path = $1,\n group_id = $2,\n topics = $3,\n filters = $4,\n auto_offset_reset = $5,\n script_path = $6,\n path = $7,\n is_flow = $8,\n edited_by = $9,\n email = $10,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $13,\n error_handler_args = $14,\n retry = $15\n WHERE\n workspace_id = $11 AND path = $12\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "VarcharArray", + "JsonbArray", + "Varchar", + "Varchar", + "Varchar", + "Bool", + "Varchar", + "Varchar", + "Text", + "Text", + "Varchar", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "12631fecee6aa11a45cf5c8d101c0dd8de50ac9b57e68198f637038d344fdd46" +} diff --git a/backend/.sqlx/query-aed5439aa6dad950e505f9f8f6914fa5ca21319c501b2822c9bc751ddfc9a0a4.json b/backend/.sqlx/query-4b5a711986017654bdd495893a16ddc6ab09c98cd8723865cbd341404bc6a02f.json similarity index 59% rename from backend/.sqlx/query-aed5439aa6dad950e505f9f8f6914fa5ca21319c501b2822c9bc751ddfc9a0a4.json rename to backend/.sqlx/query-4b5a711986017654bdd495893a16ddc6ab09c98cd8723865cbd341404bc6a02f.json index 1d165a0d13..d90a467380 100644 --- a/backend/.sqlx/query-aed5439aa6dad950e505f9f8f6914fa5ca21319c501b2822c9bc751ddfc9a0a4.json +++ b/backend/.sqlx/query-4b5a711986017654bdd495893a16ddc6ab09c98cd8723865cbd341404bc6a02f.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n INSERT INTO kafka_trigger (\n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\n filters,\n script_path,\n is_flow,\n mode,\n edited_by,\n email,\n edited_at,\n error_handler_path,\n error_handler_args,\n retry\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now(), $12, $13, $14\n )\n ", + "query": "\n INSERT INTO kafka_trigger (\n workspace_id,\n path,\n kafka_resource_path,\n group_id,\n topics,\n filters,\n auto_offset_reset,\n script_path,\n is_flow,\n mode,\n edited_by,\n email,\n edited_at,\n error_handler_path,\n error_handler_args,\n retry\n ) VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now(), $13, $14, $15\n )\n ", "describe": { "columns": [], "parameters": { @@ -12,6 +12,7 @@ "VarcharArray", "JsonbArray", "Varchar", + "Varchar", "Bool", { "Custom": { @@ -34,5 +35,5 @@ }, "nullable": [] }, - "hash": "aed5439aa6dad950e505f9f8f6914fa5ca21319c501b2822c9bc751ddfc9a0a4" + "hash": "4b5a711986017654bdd495893a16ddc6ab09c98cd8723865cbd341404bc6a02f" } diff --git a/backend/.sqlx/query-b1af20cbb737a0b98621a406e38831278517b0bb3715bbc5a3c0f2a57f5ff8c2.json b/backend/.sqlx/query-b1af20cbb737a0b98621a406e38831278517b0bb3715bbc5a3c0f2a57f5ff8c2.json new file mode 100644 index 0000000000..5abc387ba7 --- /dev/null +++ b/backend/.sqlx/query-b1af20cbb737a0b98621a406e38831278517b0bb3715bbc5a3c0f2a57f5ff8c2.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH _ AS (\n DELETE FROM debounce_key\n WHERE job_id = $1\n )\n UPDATE v2_job_debounce_batch\n SET debounce_batch = nextval('debounce_batch_seq')\n WHERE id = $1\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "b1af20cbb737a0b98621a406e38831278517b0bb3715bbc5a3c0f2a57f5ff8c2" +} diff --git a/backend/.sqlx/query-c944e384c4b4c6455b431978adc54d176f294b661675392cf92561c0e6e02e6e.json b/backend/.sqlx/query-c944e384c4b4c6455b431978adc54d176f294b661675392cf92561c0e6e02e6e.json new file mode 100644 index 0000000000..45afd37651 --- /dev/null +++ b/backend/.sqlx/query-c944e384c4b4c6455b431978adc54d176f294b661675392cf92561c0e6e02e6e.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT args FROM v2_job WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "args", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + true + ] + }, + "hash": "c944e384c4b4c6455b431978adc54d176f294b661675392cf92561c0e6e02e6e" +} diff --git a/backend/.sqlx/query-c9530931f670eab1208c4a284a55afdc3fcbb0eb5f98fd63e2ec89442becbfaa.json b/backend/.sqlx/query-c9530931f670eab1208c4a284a55afdc3fcbb0eb5f98fd63e2ec89442becbfaa.json deleted file mode 100644 index 488d3c42bd..0000000000 --- a/backend/.sqlx/query-c9530931f670eab1208c4a284a55afdc3fcbb0eb5f98fd63e2ec89442becbfaa.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n WITH _ AS (\n UPDATE debounce_key\n SET debounced_times = 0,\n first_started_at = now(),\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')\n WHERE id = $1\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "c9530931f670eab1208c4a284a55afdc3fcbb0eb5f98fd63e2ec89442becbfaa" -} diff --git a/backend/.sqlx/query-e2921e44c70cf6c76c55177f2b56985e84c59ecb3e1a13fcf27d5f7ae5f8d84c.json b/backend/.sqlx/query-e2921e44c70cf6c76c55177f2b56985e84c59ecb3e1a13fcf27d5f7ae5f8d84c.json deleted file mode 100644 index ccf11a0ac1..0000000000 --- a/backend/.sqlx/query-e2921e44c70cf6c76c55177f2b56985e84c59ecb3e1a13fcf27d5f7ae5f8d84c.json +++ /dev/null @@ -1,27 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n UPDATE kafka_trigger\n SET\n kafka_resource_path = $1,\n group_id = $2,\n topics = $3,\n filters = $4,\n script_path = $5,\n path = $6,\n is_flow = $7,\n edited_by = $8,\n email = $9,\n edited_at = now(),\n server_id = NULL,\n error = NULL,\n error_handler_path = $12,\n error_handler_args = $13,\n retry = $14\n WHERE\n workspace_id = $10 AND path = $11\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "VarcharArray", - "JsonbArray", - "Varchar", - "Varchar", - "Bool", - "Varchar", - "Varchar", - "Text", - "Text", - "Varchar", - "Jsonb", - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "e2921e44c70cf6c76c55177f2b56985e84c59ecb3e1a13fcf27d5f7ae5f8d84c" -} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 30d5d3215e..0403c71e5f 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -a70d7db187aa78a7fbfd3bfaf92372160cff320a \ No newline at end of file +d1c14355026151ecdd31adda8e2c60ecd1b5ad65 diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index d8cb3f4de4..73b7907a14 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -3040,11 +3040,18 @@ impl PulledJobResult { 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!( - "Substituting `{arg_name_to_accumulate}` with: {}\n\n", + "Accumulating debounced argument `{arg_name_to_accumulate}`:\n original: {original_value}\n accumulated: {}\n\n", &new_value ), &(db.into()), @@ -3055,6 +3062,18 @@ impl PulledJobResult { .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>>, + ) + .execute(db) + .await?; + } } // Handle dependency job debouncing cleanup when a job is pulled for execution diff --git a/backend/windmill-queue/tests/debounce_test.rs b/backend/windmill-queue/tests/debounce_test.rs index 0d7f390eff..f8d0097a2f 100644 --- a/backend/windmill-queue/tests/debounce_test.rs +++ b/backend/windmill-queue/tests/debounce_test.rs @@ -975,7 +975,7 @@ mod debounce { Ok(()) } - /// Test: Post-preprocessing debounce with max count limit resets the batch. + /// Test: Post-preprocessing debounce with max count limit deletes the debounce_key entry. #[sqlx::test(migrations = "../migrations", fixtures("base"))] async fn test_post_preprocessing_debounce_max_count_resets( db: Pool, @@ -1018,13 +1018,14 @@ mod debounce { assert!(results[1].is_some(), "second job should get scheduled_for"); assert!(results[2].is_some(), "third job should get scheduled_for"); - // Job 4 (the one that exceeds the limit): when limit is exceeded, - // the batch resets and the job executes immediately (no scheduled_for delay) - // The exact behavior depends on whether the limit check happens before or after the - // new job is counted. Let's just verify the debounce_key is reset. - let dk = get_debounce_key(&db, "pp_max_count_key").await.unwrap(); - // debounced_times should have been reset at some point - assert!(dk.0 == jobs[3], "debounce_key should point to last job"); + // Job 4 exceeds the limit: the debounce_key entry should be DELETED + // (not just reset) so that a stale entry doesn't cause the next incoming + // flow to incorrectly debounce this already-executing job. + let dk = get_debounce_key(&db, "pp_max_count_key").await; + assert!( + dk.is_none(), + "debounce_key entry should be deleted when limits exceeded" + ); Ok(()) } @@ -1526,16 +1527,18 @@ mod debounce { "job 5 should return None (limit exceeded, batch reset)" ); - // After reset, debounced_times should be 0 - let dk = get_debounce_key(&db, "pp_count_boundary_key") - .await - .unwrap(); - assert_eq!(dk.2, 0, "debounced_times should be reset to 0 after limit"); + // After limit exceeded, the debounce_key entry should be deleted + let dk = get_debounce_key(&db, "pp_count_boundary_key").await; + assert!( + dk.is_none(), + "debounce_key entry should be deleted when limits exceeded" + ); Ok(()) } - /// Test: after a max_count reset, a new batch starts fresh and debouncing works again. + /// Test: after a max_count limit exceeded, a new batch starts completely fresh. + /// The debounce_key entry is deleted, so the next cycle starts with a fresh INSERT. #[sqlx::test(migrations = "../migrations", fixtures("base"))] async fn test_post_preprocessing_max_count_reset_new_batch( db: Pool, @@ -1549,8 +1552,8 @@ mod debounce { let args_hm = empty_args(); // Limit check is `debounced_times > max`, so with max=2 we need 4 jobs - // to trigger reset (debounced_times=3 on the 4th job, 3>2=true). - // Cycle 1: jobs 1-4 (job 4 exceeds limit → reset) + // to trigger limit (debounced_times=3 on the 4th job, 3>2=true). + // Cycle 1: jobs 1-4 (job 4 exceeds limit → entry deleted) let mut cycle1 = Vec::new(); for _ in 0..4 { let id = Uuid::new_v4(); @@ -1579,14 +1582,17 @@ mod debounce { ); assert!( cycle1_results[3].is_none(), - "cycle1 job4 should reset (over limit)" + "cycle1 job4 should execute immediately (over limit)" ); - // Verify debounced_times is reset to 0 - let dk = get_debounce_key(&db, "pp_reset_cycle_key").await.unwrap(); - assert_eq!(dk.2, 0, "debounced_times should be 0 after reset"); + // Verify debounce_key entry is deleted after limit exceeded + let dk = get_debounce_key(&db, "pp_reset_cycle_key").await; + assert!( + dk.is_none(), + "debounce_key entry should be deleted after limit exceeded" + ); - // Cycle 2: jobs 5-8 (new batch, should debounce independently) + // Cycle 2: jobs 5-8 (completely fresh batch since entry was deleted) let mut cycle2 = Vec::new(); for _ in 0..4 { let id = Uuid::new_v4(); @@ -1607,19 +1613,17 @@ mod debounce { .await?; cycle2_results.push(r); } - // After cycle 1 reset, debounced_times=0. Cycle 2's first job hits ON CONFLICT - // and increments to 1 (unlike cycle 1's first job which was a fresh insert at 0). - // So cycle 2 reaches the limit one job sooner: - // job5: dt=1, job6: dt=2, job7: dt=3 (>2 → reset), job8: dt=1 - assert!(cycle2_results[0].is_some(), "cycle2 job1 scheduled (dt=1)"); - assert!(cycle2_results[1].is_some(), "cycle2 job2 scheduled (dt=2)"); + // Cycle 2 starts fresh (INSERT, not CONFLICT), so it behaves identically to cycle 1: + // job5: dt=0 (INSERT), job6: dt=1, job7: dt=2, job8: dt=3 (>2 → limit) assert!( - cycle2_results[2].is_none(), - "cycle2 job3 should reset (dt=3 > 2)" + cycle2_results[0].is_some(), + "cycle2 job1 scheduled (dt=0, fresh INSERT)" ); + assert!(cycle2_results[1].is_some(), "cycle2 job2 scheduled (dt=1)"); + assert!(cycle2_results[2].is_some(), "cycle2 job3 scheduled (dt=2)"); assert!( - cycle2_results[3].is_some(), - "cycle2 job4 scheduled (fresh after reset, dt=1)" + cycle2_results[3].is_none(), + "cycle2 job4 should execute immediately (dt=3 > 2)" ); Ok(()) @@ -3074,6 +3078,29 @@ mod debounce { let other: String = serde_json::from_str(other_raw.get())?; assert_eq!(other, "x", "non-accumulated arg should be unchanged"); + // Verify accumulated args were persisted to v2_job (needed for flows + // where subsequent steps re-read args from the DB) + let db_args: Option = + sqlx::query_scalar!("SELECT args FROM v2_job WHERE id = $1", survivor_id,) + .fetch_one(&db) + .await?; + let db_args = db_args.expect("v2_job args should not be null after accumulation"); + let db_items = db_args + .get("items") + .expect("persisted args should contain 'items'"); + let db_items: Vec = + serde_json::from_value::>(db_items.clone())? + .iter() + .map(|v| v.as_i64().unwrap()) + .collect(); + let mut db_items_sorted = db_items.clone(); + db_items_sorted.sort(); + assert_eq!( + db_items_sorted, + vec![1, 2, 3, 4, 5, 6], + "persisted args in v2_job should contain all accumulated items" + ); + Ok(()) } @@ -3504,6 +3531,27 @@ mod debounce { assert_accumulated_items(&result, &[10, 20, 30, 40, 50], "items"); + // Verify accumulated args were persisted to v2_job + let db_args: Option = + sqlx::query_scalar!("SELECT args FROM v2_job WHERE id = $1", survivor_id,) + .fetch_one(&db) + .await?; + let db_items = db_args + .expect("v2_job args should not be null") + .get("items") + .expect("persisted args should contain 'items'") + .clone(); + let mut db_items: Vec = serde_json::from_value::>(db_items)? + .iter() + .map(|v| v.as_i64().unwrap()) + .collect(); + db_items.sort(); + assert_eq!( + db_items, + vec![10, 20, 30, 40, 50], + "persisted args in v2_job should contain all accumulated items" + ); + Ok(()) } @@ -3604,6 +3652,33 @@ mod debounce { let extra: String = serde_json::from_str(extra_raw.get())?; assert_eq!(extra, "v", "non-accumulated arg should be unchanged"); + // Verify accumulated args were persisted to v2_job + let db_args: Option = + sqlx::query_scalar!("SELECT args FROM v2_job WHERE id = $1", survivor_id,) + .fetch_one(&db) + .await?; + let db_args = db_args.expect("v2_job args should not be null"); + let db_items = db_args + .get("items") + .expect("persisted args should contain 'items'") + .clone(); + let mut db_items: Vec = serde_json::from_value::>(db_items)? + .iter() + .map(|v| v.as_i64().unwrap()) + .collect(); + db_items.sort(); + assert_eq!( + db_items, + vec![100, 200, 300, 400, 500, 600], + "persisted args in v2_job should contain all accumulated items" + ); + // "extra" should also be persisted unchanged + let db_extra = db_args.get("extra").unwrap().as_str().unwrap(); + assert_eq!( + db_extra, "v", + "persisted non-accumulated arg should be unchanged" + ); + Ok(()) }