From 47daba4d44560c9f1502cda17b2f58320fdfa9ff Mon Sep 17 00:00:00 2001 From: Subhra264 Date: Tue, 28 Jul 2026 08:12:25 +0530 Subject: [PATCH] fix: prevent deadlock when batch upserting pipeline errors (#13480) (#13487) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem Postgres is killing pipeline-error flushes with `deadlock detected`: ``` ERROR: deadlock detected DETAIL: Process 3407043 waits for ExclusiveLock on tuple (82,8) of relation 2739499; blocked by process 3407136. Process 3407136 waits for ShareLock on transaction 2384324529; blocked by process 3406875. Process 3406875 waits for ExclusiveLock on tuple (81,4) ... Process 3406910 waits for ShareLock on transaction 2384321712 ... Process 3406931 waits for ExclusiveLock on tuple (82,8) ... <-- cycle closes ``` All five participants run the identical `UPDATE "pipeline_last_errors" SET ...` statement. Same rows, different orders — lock-order inversion, not contention or a slow query. Two things combine to cause it: 1. **`batch_upsert` wrote rows in arbitrary order.** The batch is built straight from the queue with no sorting, and every row is updated inside one transaction, which holds each row lock until commit. Node A may update pipeline X then Y while node B does Y then X. 2. **All nodes flush on the same tick.** Every log line in the incident is within the same second, so the transactions overlap almost perfectly. ## Fix Deduplicate the batch by `pipeline_id` (keeping the newest error per pipeline) and sort by `pipeline_id` before the write loop. - **Sorting** is the actual deadlock fix: with a global lock order, a wait cycle cannot form. Nodes may still block on each other briefly, but Postgres will never find a cycle to kill. - **Dedup** shrinks each transaction. A batch usually carries many errors from the same pipeline, which previously became repeated updates of the very same row inside one transaction — more rows touched, longer lock-hold window, more chances to interleave. Everything else is unchanged: same transaction, same up-front `SELECT`, same "skip the write if the error content hasn't changed" check, same per-row update/insert loop. The diff is additions only. ## Testing - New unit test `test_dedup_and_sort` covers both the ordering and the newest-wins behavior. - `cargo clippy --lib --tests` and `cargo fmt --check` clean. Co-authored-by: Claude Opus 5 (1M context) --- src/db/src/pipeline_errors.rs | 71 +++++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/src/db/src/pipeline_errors.rs b/src/db/src/pipeline_errors.rs index 9e5243a971..69dc28770c 100644 --- a/src/db/src/pipeline_errors.rs +++ b/src/db/src/pipeline_errors.rs @@ -83,6 +83,37 @@ pub async fn upsert( Ok(()) } +/// Deduplicates a batch by pipeline_id, keeping the newest error per pipeline, and returns the +/// result sorted by pipeline_id. +/// +/// A batch usually holds many errors for the same pipeline, which would otherwise become +/// repeated updates of the very same row within one transaction. +/// +/// Sorting is what prevents deadlocks: the transaction holds every row lock it takes until it +/// commits, so if two nodes flush overlapping pipelines in different orders, each ends up +/// waiting on a row the other already holds. Writing rows in pipeline_id order everywhere means +/// that wait cycle cannot form. +fn dedup_and_sort( + errors: Vec<(String, String, String, i64, PipelineError)>, +) -> Vec<(String, String, String, i64, PipelineError)> { + let mut by_id: std::collections::HashMap = + std::collections::HashMap::with_capacity(errors.len()); + + for error in errors { + match by_id.get(&error.0) { + // already holding a newer error for this pipeline + Some(existing) if existing.3 >= error.3 => {} + _ => { + by_id.insert(error.0.clone(), error); + } + } + } + + let mut errors: Vec<_> = by_id.into_values().collect(); + errors.sort_by(|a, b| a.0.cmp(&b.0)); + errors +} + /// Batch upserts multiple pipeline error records in a single transaction. /// /// This is more efficient than individual upserts when processing multiple errors. @@ -94,6 +125,9 @@ pub async fn batch_upsert( return Ok(()); } + // Collapse repeats and fix the row order so concurrently flushing nodes can't deadlock + let errors = dedup_and_sort(errors); + let client = ORM_CLIENT.get_or_init(connect_to_orm).await; let txn = client.begin().await?; @@ -236,6 +270,43 @@ mod tests { use super::*; + #[test] + fn test_dedup_and_sort() { + let entry = |id: &str, ts: i64, msg: &str| { + ( + id.to_string(), + format!("Pipeline {id}"), + "org".to_string(), + ts, + PipelineError { + pipeline_id: id.to_string(), + pipeline_name: format!("Pipeline {id}"), + error: Some(msg.to_string()), + node_errors: HashMap::new(), + }, + ) + }; + + let deduped = dedup_and_sort(vec![ + entry("pl_b", 200, "older b"), + entry("pl_a", 100, "only a"), + entry("pl_b", 300, "newer b"), + entry("pl_c", 100, "newer c"), + entry("pl_c", 50, "older c"), + ]); + + // one row per pipeline, in pipeline_id order so every node writes rows in the same + // order + let ids: Vec<&str> = deduped.iter().map(|(id, ..)| id.as_str()).collect(); + assert_eq!(ids, vec!["pl_a", "pl_b", "pl_c"]); + + // the newest error per pipeline wins, regardless of arrival order + assert_eq!(deduped[1].3, 300); + assert_eq!(deduped[1].4.error.as_deref(), Some("newer b")); + assert_eq!(deduped[2].3, 100); + assert_eq!(deduped[2].4.error.as_deref(), Some("newer c")); + } + #[tokio::test] #[ignore] // Requires database connection async fn test_upsert_and_get() {