perf(plugin): retry batch apply on Postgres deadlock

The whole batch runs inside an outer transaction.atomic(), so all data
+ tag + audit writes for the batch become visible to readers in one
atomic transition. The trade-off is a longer lock-hold window per
batch, which can cause Postgres deadlocks (SQLSTATE 40P01) when two
concurrent batches touch the same rows in different orders.

Wrap BulkApplyView's outer transaction.atomic() in a bounded retry
loop. On a Postgres deadlock or serialization failure (40001), the
whole batch transaction is rolled back; the retry loop re-enters with
fresh deferred_changelog accumulators (no stale state from the aborted
attempt).

Behaviour:
- New plugin setting: batch_apply_deadlock_retry_max_count, default 3
  (three retries after the initial attempt = up to 4 attempts total).
- 0 disables retries entirely.
- Backoff is jittered exponential: 50ms * 2^attempt * U(0.5, 1.5).
- On success after retries, the response carries X-Diode-Batch-Retries:
  <n> so the reconciler-side log can correlate.
- On exhaustion, the original OperationalError is re-raised; caller
  handles the failure as before.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ldrozdz93

netbox_diode_plugin/__init__.py

@@ -47,6 +47,12 @@ class NetBoxDiodePluginConfig(PluginConfig):
         # Override the displayed Diode target URL without affecting internal
         # communication (e.g. to show the external ingress address).
         "diode_target_display": None,
+
+        # Max number of retries when the batch apply endpoint hits a
+        # Postgres deadlock (40P01) or serialization failure (40001).
+        # 0 disables retries; default 3 means up to three retries after
+        # the initial attempt (4 attempts total).
+        "batch_apply_deadlock_retry_max_count": 3,
     }
 
 

netbox_diode_plugin/api/views.py

@@ -2,14 +2,18 @@
 # Copyright 2025 NetBox Labs, Inc.
 """Diode NetBox Plugin - API Views."""
 import logging
+import random
 import re
+import time
 
 from django.apps import apps
 from django.db import transaction
+from django.db.utils import OperationalError
 from rest_framework import status, views
 from rest_framework.exceptions import ValidationError
 from rest_framework.response import Response
 
+from ..plugin_config import get_batch_apply_deadlock_retry_max_count
 from .applier import apply_changeset
 from .authentication import DiodeOAuth2Authentication
 from .common import (
@@ -27,6 +31,20 @@
     require_scopes,
 )
 
+# Postgres SQLSTATEs we retry on. 40P01 is deadlock_detected; 40001 is
+# serialization_failure (can show up on the SERIALIZABLE isolation
+# level path, harmless to retry the same way).
+_DEADLOCK_PGCODES = ("40P01", "40001")
+
+
+def _extract_pgcode(exc: OperationalError) -> str | None:
+    """Return the Postgres SQLSTATE on a wrapped Django OperationalError, or None."""
+    inner = exc.__cause__
+    pgcode = getattr(inner, "pgcode", None)
+    if pgcode is None:
+        pgcode = getattr(exc, "pgcode", None)
+    return pgcode
+
 logger = logging.getLogger("netbox.diode_data")
 
 
@@ -367,6 +385,46 @@ def _post(self, request, *args, **kwargs):
         if len(change_sets) == 0:
             raise ValidationError({"change_sets": ["change_sets must not be empty"]})
 
+        max_retries = int(get_batch_apply_deadlock_retry_max_count() or 0)
+        attempt = 0
+        while True:
+            try:
+                results = self._apply_batch(change_sets, request)
+                break
+            except OperationalError as exc:
+                pgcode = _extract_pgcode(exc)
+                if pgcode not in _DEADLOCK_PGCODES:
+                    raise
+                if attempt >= max_retries:
+                    logger.error(
+                        "batch apply: deadlock retries exhausted "
+                        "(attempts=%d, pgcode=%s)",
+                        attempt + 1, pgcode,
+                    )
+                    raise
+                # Jittered exponential backoff: 50ms * 2^attempt * U(0.5, 1.5)
+                sleep_s = 0.05 * (2 ** attempt) * (0.5 + random.random())
+                logger.warning(
+                    "batch apply: %s, retrying "
+                    "(attempt=%d/%d, sleep=%.3fs)",
+                    "deadlock" if pgcode == "40P01" else "serialization failure",
+                    attempt + 1, max_retries + 1, sleep_s,
+                )
+                time.sleep(sleep_s)
+                attempt += 1
+
+        http_status = (
+            status.HTTP_207_MULTI_STATUS
+            if any(r.get("errors") for r in results)
+            else status.HTTP_200_OK
+        )
+        resp = Response({"results": results}, status=http_status)
+        if attempt > 0:
+            resp["X-Diode-Batch-Retries"] = str(attempt)
+        return resp
+
+    def _apply_batch(self, change_sets, request):
+        """Apply all changesets in one outer transaction; caller wraps for retries."""
         results = []
         # Outer transaction.atomic() makes the whole batch atomic to readers:
         # data writes + tag-link writes + audit-log bulk_create commit together.
@@ -398,13 +456,7 @@ def _post(self, request, *args, **kwargs):
                     ).to_dict()
                 results.append(result)
             defc.flush()
-
-        http_status = (
-            status.HTTP_207_MULTI_STATUS
-            if any(r.get("errors") for r in results)
-            else status.HTTP_200_OK
-        )
-        return Response({"results": results}, status=http_status)
+        return results
 
 
 class GetDefaultBranchView(views.APIView):

netbox_diode_plugin/plugin_config.py

@@ -10,6 +10,7 @@
 from netbox.plugins import get_plugin_config
 
 __all__ = (
+    "get_batch_apply_deadlock_retry_max_count",
     "get_diode_auth_introspect_url",
     "get_diode_user",
 )
@@ -99,3 +100,9 @@ def get_diode_user():
 def get_required_token_audience():
     """Returns the require token audience."""
     return get_plugin_config("netbox_diode_plugin", "required_token_audience")
+
+def get_batch_apply_deadlock_retry_max_count():
+    """Max retries on Postgres deadlock for the batch apply endpoint."""
+    return get_plugin_config(
+        "netbox_diode_plugin", "batch_apply_deadlock_retry_max_count"
+    )