From ceab96a8cec87a9b21707950e658698983d4b17a Mon Sep 17 00:00:00 2001 From: Nilirad Date: Wed, 1 Jul 2026 14:18:22 +0200 Subject: [PATCH] Fix trigger queue coalescing --- ...3ac49077fb81e012c1d3172bcb24941835bb.json} | 4 +- .../20260701120612_fix_trigger_coalescing.sql | 2 + src/polling/mod.rs | 76 +++++++++++++++++++ src/repository/sqlite.rs | 4 +- 4 files changed, 82 insertions(+), 4 deletions(-) rename .sqlx/{query-8fd1e14a687da7642c1505001190a13b2e71555d51113601377cba0164a95623.json => query-30b46e820bb013286bf5a18db3513ac49077fb81e012c1d3172bcb24941835bb.json} (57%) create mode 100644 migrations/20260701120612_fix_trigger_coalescing.sql diff --git a/.sqlx/query-8fd1e14a687da7642c1505001190a13b2e71555d51113601377cba0164a95623.json b/.sqlx/query-30b46e820bb013286bf5a18db3513ac49077fb81e012c1d3172bcb24941835bb.json similarity index 57% rename from .sqlx/query-8fd1e14a687da7642c1505001190a13b2e71555d51113601377cba0164a95623.json rename to .sqlx/query-30b46e820bb013286bf5a18db3513ac49077fb81e012c1d3172bcb24941835bb.json index fb47829..2217d50 100644 --- a/.sqlx/query-8fd1e14a687da7642c1505001190a13b2e71555d51113601377cba0164a95623.json +++ b/.sqlx/query-30b46e820bb013286bf5a18db3513ac49077fb81e012c1d3172bcb24941835bb.json @@ -1,6 +1,6 @@ { "db_name": "SQLite", - "query": "INSERT INTO trigger_queue (branch_id, new_hash, target_repo, event_type, gh_app_installation_id)\n SELECT ?, ?, s.target_repo, s.event_type, s.gh_app_installation_id\n FROM subscriptions s\n WHERE s.branch_id = ?\n ON CONFLICT(branch_id, target_repo, event_type) WHERE status = 'PENDING'\n DO UPDATE SET new_hash = excluded.new_hash, status_updated_at = CURRENT_TIMESTAMP", + "query": "INSERT INTO trigger_queue (branch_id, new_hash, target_repo, event_type, gh_app_installation_id)\n SELECT ?, ?, s.target_repo, s.event_type, s.gh_app_installation_id\n FROM subscriptions s\n WHERE s.branch_id = ?\n ON CONFLICT(target_repo, event_type) WHERE status = 'PENDING'\n DO UPDATE SET branch_id = excluded.branch_id, new_hash = excluded.new_hash, status_updated_at = CURRENT_TIMESTAMP", "describe": { "columns": [], "parameters": { @@ -8,5 +8,5 @@ }, "nullable": [] }, - "hash": "8fd1e14a687da7642c1505001190a13b2e71555d51113601377cba0164a95623" + "hash": "30b46e820bb013286bf5a18db3513ac49077fb81e012c1d3172bcb24941835bb" } diff --git a/migrations/20260701120612_fix_trigger_coalescing.sql b/migrations/20260701120612_fix_trigger_coalescing.sql new file mode 100644 index 0000000..4a5140f --- /dev/null +++ b/migrations/20260701120612_fix_trigger_coalescing.sql @@ -0,0 +1,2 @@ +DROP INDEX IF EXISTS idx_trigger_queue_pending_target; +CREATE UNIQUE INDEX idx_trigger_queue_pending_target ON trigger_queue (target_repo, event_type) WHERE status = 'PENDING'; diff --git a/src/polling/mod.rs b/src/polling/mod.rs index 869db4f..bbcc32b 100644 --- a/src/polling/mod.rs +++ b/src/polling/mod.rs @@ -273,4 +273,80 @@ mod tests { assert_eq!(queued_events[0].branch_id, Some(1)); assert_eq!(queued_events[0].new_hash, Some("c".repeat(40))); } + + #[tokio::test] + async fn test_coalescing_of_trigger_events_multiple_branches() { + let pool = crate::test_utils::create_test_db().await; + + // Insert first branch + let hash1 = "a".repeat(40); + sqlx::query!( + "INSERT INTO branches (repo_url, name, last_commit_hash) VALUES (?, ?, ?)", + "https://github.com/owner/repo1", + "main", + hash1 + ) + .execute(&pool) + .await + .unwrap(); + + // Insert first subscription + sqlx::query!("INSERT INTO subscriptions (branch_id, target_repo, event_type, gh_app_installation_id) VALUES (?, ?, ?, ?)", + 1, + "org/target", + "dispatch", + 1 + ) + .execute(&pool) + .await + .unwrap(); + + // Insert second branch + let hash2 = "b".repeat(40); + sqlx::query!( + "INSERT INTO branches (repo_url, name, last_commit_hash) VALUES (?, ?, ?)", + "https://github.com/owner/repo2", + "main", + hash2 + ) + .execute(&pool) + .await + .unwrap(); + + // Insert second subscription with the same target repo and event type + sqlx::query!("INSERT INTO subscriptions (branch_id, target_repo, event_type, gh_app_installation_id) VALUES (?, ?, ?, ?)", + 2, + "org/target", + "dispatch", + 1 + ) + .execute(&pool) + .await + .unwrap(); + + let ctx = SharedContext { + config: crate::test_utils::create_test_config(), + repository: std::sync::Arc::new(crate::repository::SqliteRepository::new(pool.clone())), + db_pool: pool.clone(), + git_fetcher: std::sync::Arc::new(crate::test_utils::MockGitFetcher { + hash: CommitHash::new("c".repeat(40)).unwrap(), + }), + token: tokio_util::sync::CancellationToken::new(), + }; + + // Poll for both branches. The first branch updates to 'c' + // The second branch updates to 'c' and coalesces with the first one + poll_branches(&ctx).await.unwrap(); + + // Verify only one entry in queue + let queued_events = sqlx::query!("SELECT branch_id, new_hash FROM trigger_queue") + .fetch_all(&pool) + .await + .unwrap(); + + assert_eq!(queued_events.len(), 1); + // It should contain the last updated branch + assert_eq!(queued_events[0].branch_id, Some(2)); + assert_eq!(queued_events[0].new_hash, Some("c".repeat(40))); + } } diff --git a/src/repository/sqlite.rs b/src/repository/sqlite.rs index 03cfd25..55f7d33 100644 --- a/src/repository/sqlite.rs +++ b/src/repository/sqlite.rs @@ -508,8 +508,8 @@ impl TriggerRepository for SqliteRepository { SELECT ?, ?, s.target_repo, s.event_type, s.gh_app_installation_id FROM subscriptions s WHERE s.branch_id = ? - ON CONFLICT(branch_id, target_repo, event_type) WHERE status = 'PENDING' - DO UPDATE SET new_hash = excluded.new_hash, status_updated_at = CURRENT_TIMESTAMP", + ON CONFLICT(target_repo, event_type) WHERE status = 'PENDING' + DO UPDATE SET branch_id = excluded.branch_id, new_hash = excluded.new_hash, status_updated_at = CURRENT_TIMESTAMP", branch_id, new_hash, branch_id