fix: prevent deadlock when batch upserting pipeline errors (#13480) (#13487)

## 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) <noreply@anthropic.com>

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<String, (String, String, String, i64, PipelineError)> =
+        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() {