fix: sort aggregation values to avoid deadlocks in process_pending (#1851)

* fix: sort aggregation values to avoid deadlocks in process_pending_aggregations and update_tree_listing

* refactor: introduce TreeListingRow and CheckoutRow

Author: gustavobtflores

backend/kernelCI_app/management/commands/helpers/aggregation_helpers.py

@@ -1,9 +1,13 @@
 import time
-from typing import Sequence
+from typing import Optional, Sequence
 
 
 from django.db import connections
 from kernelCI_app.helpers.logger import out
+from kernelCI_app.management.commands.helpers.tree_listing import (
+    CheckoutRow,
+    tree_listing_sort_key,
+)
 from kernelCI_app.models import (
     Checkouts,
     PendingTest,
@@ -14,7 +18,6 @@
     Builds,
 )
 from kernelCI_app.utils import is_boot
-from typing import Optional
 
 
 def simplify_status(status: Optional[StatusChoices]) -> SimplifiedStatusChoices:
@@ -65,20 +68,23 @@ def update_tree_listing(checkouts_instances: Sequence[Checkouts]):
         return
 
     t0 = time.time()
-    checkout_values = [
-        (
-            checkout.id,
-            checkout.origin,
-            checkout.tree_name,
-            checkout.git_repository_url,
-            checkout.git_repository_branch,
-            checkout.git_commit_hash,
-            checkout.git_commit_name,
-            checkout.git_commit_tags,
-            checkout.start_time,
-        )
-        for checkout in checkouts_instances
-    ]
+    checkout_values = sorted(
+        [
+            CheckoutRow(
+                checkout_id=checkout.id,
+                origin=checkout.origin,
+                tree_name=checkout.tree_name,
+                git_repository_url=checkout.git_repository_url,
+                git_repository_branch=checkout.git_repository_branch,
+                git_commit_hash=checkout.git_commit_hash,
+                git_commit_name=checkout.git_commit_name,
+                git_commit_tags=checkout.git_commit_tags,
+                start_time=checkout.start_time,
+            )
+            for checkout in checkouts_instances
+        ],
+        key=tree_listing_sort_key,
+    )
 
     with connections["default"].cursor() as cursor:
         # Set values as 0 when inserting a new tree

backend/kernelCI_app/management/commands/helpers/tree_listing.py

@@ -0,0 +1,52 @@
+from datetime import datetime
+from typing import Any, NamedTuple
+
+
+class CheckoutRow(NamedTuple):
+    checkout_id: str
+    origin: str
+    tree_name: str | None
+    git_repository_url: str | None
+    git_repository_branch: str | None
+    git_commit_hash: str | None
+    git_commit_name: str | None
+    git_commit_tags: list[str] | None
+    start_time: datetime | None
+
+
+class TreeListingRow(NamedTuple):
+    """Flat row passed to cursor.executemany() for UPDATE tree_listing.
+
+    Field order must match the SQL parameter placeholders in _process_tree_listing.
+    """
+
+    build_pass: int
+    build_failed: int
+    build_inc: int
+    boot_pass: int
+    boot_failed: int
+    boot_inc: int
+    test_pass: int
+    test_failed: int
+    test_inc: int
+    origin: str
+    tree_name: str | None
+    git_repository_branch: str | None
+    git_repository_url: str | None
+    git_commit_hash: str | None
+
+
+def tree_listing_sort_key(v: Any) -> tuple:
+    """Sort key for tree_listing rows by unique constraint columns.
+
+    Ensures deterministic lock acquisition order in concurrent transactions,
+    preventing deadlocks between the ingester and aggregation processor.
+    Compatible with both CheckoutRow and TreeListingRow.
+    """
+    return (
+        (v.origin is None, v.origin or ""),
+        (v.tree_name is None, v.tree_name or ""),
+        (v.git_repository_url is None, v.git_repository_url or ""),
+        (v.git_repository_branch is None, v.git_repository_branch or ""),
+        (v.git_commit_hash is None, v.git_commit_hash or ""),
+    )

backend/kernelCI_app/management/commands/process_pending_aggregations.py

@@ -14,6 +14,10 @@
 from kernelCI_app.constants.general import MAESTRO_DUMMY_BUILD_PREFIX
 from kernelCI_app.helpers.logger import out
 from kernelCI_app.management.commands.helpers.aggregation_helpers import simplify_status
+from kernelCI_app.management.commands.helpers.tree_listing import (
+    TreeListingRow,
+    tree_listing_sort_key,
+)
 from kernelCI_app.constants.ingester import PROMETHEUS_MULTIPROC_DIR
 from prometheus_client import start_http_server
 from kernelCI_app.models import (
@@ -647,25 +651,28 @@ def _process_tree_listing(
         if not tree_listing_data:
             return
 
-        values = [
-            (
-                data["build_pass"],
-                data["build_failed"],
-                data["build_inc"],
-                data["boot_pass"],
-                data["boot_failed"],
-                data["boot_inc"],
-                data["test_pass"],
-                data["test_failed"],
-                data["test_inc"],
-                data["origin"],
-                data["tree_name"],
-                data["git_repository_branch"],
-                data["git_repository_url"],
-                data["git_commit_hash"],
-            )
-            for data in tree_listing_data.values()
-        ]
+        values = sorted(
+            [
+                TreeListingRow(
+                    build_pass=data["build_pass"],
+                    build_failed=data["build_failed"],
+                    build_inc=data["build_inc"],
+                    boot_pass=data["boot_pass"],
+                    boot_failed=data["boot_failed"],
+                    boot_inc=data["boot_inc"],
+                    test_pass=data["test_pass"],
+                    test_failed=data["test_failed"],
+                    test_inc=data["test_inc"],
+                    origin=data["origin"],
+                    tree_name=data["tree_name"],
+                    git_repository_branch=data["git_repository_branch"],
+                    git_repository_url=data["git_repository_url"],
+                    git_commit_hash=data["git_commit_hash"],
+                )
+                for data in tree_listing_data.values()
+            ],
+            key=tree_listing_sort_key,
+        )
 
         t0 = time.time()
         with connection.cursor() as cursor: