refactor: Move deadlock backoff into errback, drop file_worker retry loop

The file_worker retry loop re-ran process_file() on every attempt, so it
re-read and re-parsed the file and re-inserted the data each time; its only
saving over a requeue was the CollectionFile fetch. Its real value was the
randomized backoff that desynchronizes the competing transactions.

Centralize that: errback() now sleeps a random 1-5s before requeuing a
deadlocked message, and file_worker just lets the OperationalError
propagate. This removes the bespoke retry loop and gives every worker
(file_worker, wiper, compilers) the same staggered backoff. errback() runs
in the consumer thread pool, off the pika I/O loop, so the sleep does not
block heartbeats.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CP8dBe7ptjmf25KCzJyHtQ

Author: claude

process/management/commands/file_worker.py

@@ -1,6 +1,4 @@
 import logging
-import random
-import time
 from collections import OrderedDict
 from itertools import islice
 
@@ -46,7 +44,6 @@
 Level = CollectionNote.Level
 
 SUPPORTED_FORMATS = {Format.release_package, Format.record_package, Format.compiled_release}
-MAX_ATTEMPTS = 5
 
 
 class Command(BaseCommand):
@@ -116,43 +113,35 @@ def callback(client_state, channel, method, properties, input_message):
             nack(client_state, channel, method.delivery_tag, requeue=False)
             return
 
-        for attempt in range(1, MAX_ATTEMPTS + 1):
-            try:
-                with (
-                    deleting_step(
-                        ProcessingStep.Name.LOAD,
-                        collection_file_id=collection_file_id,
-                        finish=finish,
-                        finish_args=(collection_id, collection_file_id),
-                    ),
-                    transaction.atomic(),
-                ):
-                    upgraded_collection_file_id = process_file(collection_file)
-            except OperationalError as e:
-                # Data that exceeds a PostgreSQL size limit can never be stored, so skip the file.
-                if isinstance(e.__cause__, ProgramLimitExceeded):
-                    logger.exception("%s is too large to store, skipping", collection_file.filename)
-                    delete_step(ProcessingStep.Name.LOAD, collection_file_id=collection_file_id)
-                    create_note(
-                        collection,
-                        Level.ERROR,
-                        f"{collection_file.filename} is too large to store",
-                        data={"type": type(e).__name__, "message": str(e), **input_message},
-                    )
-                    nack(client_state, channel, method.delivery_tag, requeue=False)
-                    return
-
-                # If another transaction in another thread INSERTs the same data, concurrently.
-                logger.warning("Deadlock on %s %s (%d/%d)\n%s", collection, collection_file, attempt, MAX_ATTEMPTS, e)
-                if attempt == MAX_ATTEMPTS:
-                    # The transaction is rolled back and the LOAD step preserved, so the errback() function in the
-                    # decorator() function can requeue the message to retry it later, instead of shutting down.
-                    raise
-
-                # Make the threads retry at different times, to avoid repeating the deadlock.
-                time.sleep(random.randint(1, 5))  # noqa: S311 # non-cryptographic
-            else:
-                break
+        try:
+            with (
+                deleting_step(
+                    ProcessingStep.Name.LOAD,
+                    collection_file_id=collection_file_id,
+                    finish=finish,
+                    finish_args=(collection_id, collection_file_id),
+                ),
+                transaction.atomic(),
+            ):
+                upgraded_collection_file_id = process_file(collection_file)
+        except OperationalError as e:
+            # Data that exceeds a PostgreSQL size limit can never be stored, so skip the file.
+            if isinstance(e.__cause__, ProgramLimitExceeded):
+                logger.exception("%s is too large to store, skipping", collection_file.filename)
+                delete_step(ProcessingStep.Name.LOAD, collection_file_id=collection_file_id)
+                create_note(
+                    collection,
+                    Level.ERROR,
+                    f"{collection_file.filename} is too large to store",
+                    data={"type": type(e).__name__, "message": str(e), **input_message},
+                )
+                nack(client_state, channel, method.delivery_tag, requeue=False)
+                return
+
+            # A deadlock can occur when another thread concurrently INSERTs the same deduplicated data. The transaction
+            # is rolled back and the LOAD step is preserved, so the errback() function in the decorator() function can
+            # requeue the message to retry it later, instead of shutting down.
+            raise
 
         message = {"collection_id": collection_id, "collection_file_id": collection_file_id}
         publish(client_state, channel, message, routing_key)

process/util.py

@@ -2,6 +2,8 @@
 import io
 import logging
 import os
+import random
+import time
 from contextlib import contextmanager
 from textwrap import fill
 
@@ -76,16 +78,17 @@ def errback(exception):
         # A foreign-key violation is not a duplicate-message symptom: a package_data or data row is still referenced
         # (e.g. by another collection's release), which indicates an error in logic rather than a redelivered message.
         #
-        # A deadlock is transient: it occurs when concurrent transactions INSERT the same deduplicated data into the
-        # unique index on the data or package_data table in a different order. Requeue the message to retry it later,
-        # rather than shutting down the worker. The file_worker worker retries in-process first (see its MAX_ATTEMPTS
-        # loop) and reaches here only if every attempt deadlocks; the compiler workers have no in-process retry and
-        # rely on this requeue.
+        # A deadlock is transient: it occurs when concurrent transactions write the same rows in a different order
+        # (for example, when INSERTing the same deduplicated data, or DELETEing the same shared data). Requeue the
+        # message to retry it, rather than shutting down the worker. The callback must leave no partial state on a
+        # rolled-back transaction (see file_worker and wiper). Sleep first, so that the transaction that won the
+        # deadlock can commit, and so that concurrent threads retry at different times, to avoid repeating it.
         if isinstance(exception, IntegrityError) and isinstance(exception.__cause__, errors.ForeignKeyViolation):
             logger.exception("Unhandled exception when consuming %r, shutting down gracefully", body)
             add_callback_threadsafe(state.connection, state.interrupt)
         elif isinstance(exception, OperationalError) and isinstance(exception.__cause__, errors.DeadlockDetected):
             logger.exception("Deadlock when consuming %r, requeuing", body)
+            time.sleep(random.randint(1, 5))  # noqa: S311 # non-cryptographic
             nack(state, channel, method.delivery_tag, requeue=True)
         elif isinstance(exception, AlreadyExists | InvalidFormError | IntegrityError | Collection.DoesNotExist):
             logger.exception("%s maybe caused by duplicate message %r, skipping", type(exception).__name__, body)

tests/test_util.py

@@ -60,15 +60,17 @@ def test_foreign_key_violation_shuts_down(self, add_callback_threadsafe, nack):
         add_callback_threadsafe.assert_called_once_with(state.connection, state.interrupt)
         nack.assert_not_called()
 
+    @patch("process.util.time.sleep")
     @patch("process.util.nack")
     @patch("process.util.add_callback_threadsafe")
-    def test_deadlock_requeues(self, add_callback_threadsafe, nack):
+    def test_deadlock_requeues(self, add_callback_threadsafe, nack, sleep):
         exception = OperationalError("deadlock detected")
         exception.__cause__ = errors.DeadlockDetected("deadlock detected")
 
         state, channel, method = self.run_errback(exception)
 
         add_callback_threadsafe.assert_not_called()
+        sleep.assert_called_once()
         nack.assert_called_once_with(state, channel, method.delivery_tag, requeue=True)
 
     @patch("process.util.nack")