fix: Move set_data_type outside transaction, to avoid locking collection row.

The logic was also incorrect, in that we should never store the data format
if the format is unsupported or the file is empty.

Author: jpmckinney

process/exceptions.py

@@ -11,8 +11,8 @@ class InvalidFormError(KingfisherProcessError, ValueError):
 
 
 class EmptyFormatError(KingfisherProcessError):
-    """Raised if a collection file's data type contains no data."""
+    """Raised if a collection file contains no data."""
 
 
 class UnsupportedFormatError(KingfisherProcessError):
-    """Raised if a collection file's data type is not supported."""
+    """Raised if a collection file's format is unsupported."""

process/management/commands/file_worker.py

@@ -32,6 +32,7 @@
     create_note,
     create_step,
     decorator,
+    delete_step,
     deleting_step,
     get_or_create,
 )
@@ -64,15 +65,9 @@ def handle(self, *args, **options):
 def finish(collection_id, collection_file_id, exception):
     # If a duplicate message is received causing an IntegrityError or similar, we still want to create the next step,
     # in case it was not created the first time. deleting_step() will delete any duplicate steps.
-    if settings.ENABLE_CHECKER and not isinstance(
-        exception,
-        # See the try/except block in the callback() function of the file_worker worker.
-        EmptyFormatError
-        | FileNotFoundError
-        | UnknownFormatError
-        | UnsupportedFormatError
-        | ijson.common.IncompleteJSONError,
-    ):
+    #
+    # See the try/except block in the callback() function of the file_worker worker.
+    if settings.ENABLE_CHECKER and not isinstance(exception, FileNotFoundError | ijson.common.IncompleteJSONError):
         create_step(ProcessingStep.Name.CHECK, collection_id, collection_file_id=collection_file_id)
 
 
@@ -87,6 +82,25 @@ def callback(client_state, channel, method, properties, input_message):
         return
 
     try:
+        # Detect and save the data_type before the transaction, to avoid locking collection rows during process_file().
+        #
+        # NOTE: Kingfisher Process assumes a single format for all collection files. Unknown, unsupported, or empty
+        # formats are detected only for the first collection files processed.
+        try:
+            set_data_type(collection, collection_file)
+        except (UnknownFormatError, UnsupportedFormatError):  # UnknownFormatError is raised by detect_format()
+            logger.exception("Source %s yields an unknown or unsupported format, skipping", collection.source_id)
+            delete_step(ProcessingStep.Name.LOAD, collection_file_id=collection_file_id)
+            create_note(collection, ERROR, f"Source {collection.source_id} yields unknown format", data=input_message)
+            nack(client_state, channel, method.delivery_tag, requeue=False)
+            return
+        except EmptyFormatError as e:
+            # Don't log a message, since sources with empty packages also have non-empty packages.
+            delete_step(ProcessingStep.Name.LOAD, collection_file_id=collection_file_id)
+            create_note(collection, CollectionNote.Level.WARNING, str(e), data=input_message)
+            nack(client_state, channel, method.delivery_tag, requeue=False)
+            return
+
         for attempt in range(1, MAX_ATTEMPTS + 1):
             try:
                 with (
@@ -113,26 +127,18 @@ def callback(client_state, channel, method, properties, input_message):
         publish(client_state, channel, message, routing_key)
 
         if upgraded_collection_file_id:
-            # The deleting_step() context manager sets upgraded_collection_file_id only if successful, so we don't need
-            # to create this step in the finish() function.
+            # The deleting_step() context manager sets upgraded_collection_file_id only if successful, so we can create
+            # this step here instead of in the finish() function.
             if settings.ENABLE_CHECKER:
                 create_step(ProcessingStep.Name.CHECK, collection_id, collection_file_id=upgraded_collection_file_id)
 
             message = {"collection_id": collection_id, "collection_file_id": upgraded_collection_file_id}
             publish(client_state, channel, message, routing_key)
-    # "Expected" errors.
-    except EmptyFormatError as e:  # raised by process_file()
-        create_note(collection, CollectionNote.Level.WARNING, str(e), data=input_message)
-        nack(client_state, channel, method.delivery_tag, requeue=False)
     # Irrecoverable errors. Discard the message to allow other messages to be processed.
     except FileNotFoundError:  # raised by detect_format() or open()
         logger.exception("%s has disappeared, skipping", collection_file.filename)
         create_note(collection, ERROR, f"{collection_file.filename} has disappeared", data=input_message)
         nack(client_state, channel, method.delivery_tag, requeue=False)
-    except (UnknownFormatError, UnsupportedFormatError):  # raised by detect_format() or process_file()
-        logger.exception("Source %s yields an unknown or unsupported format, skipping", collection.source_id)
-        create_note(collection, ERROR, f"Source {collection.source_id} yields unknown format", data=input_message)
-        nack(client_state, channel, method.delivery_tag, requeue=False)
     except ijson.common.IncompleteJSONError:  # raised by ijson.parse()
         logger.exception("Source %s yields invalid JSON, skipping", collection.source_id)
         create_note(collection, ERROR, f"Source {collection.source_id} yields invalid JSON", data=input_message)
@@ -151,20 +157,9 @@ def process_file(collection_file) -> int | None:
     :param collection_file: collection file for which should be releases checked
     :returns: upgraded collection file id or None (if there is no upgrade planned)
     """
-    data_type = _get_data_type(collection_file)
-
-    data_format = data_type["format"]
-
-    # https://github.com/open-contracting/kingfisher-collect/issues/1012
-    if data_format == Format.empty_package:
-        raise EmptyFormatError(f"Empty format '{data_format}' for file {collection_file}.")
-    if data_format not in SUPPORTED_FORMATS:
-        raise UnsupportedFormatError(
-            f"Unsupported format '{data_format}' for file {collection_file}. "
-            f"Must be one of: {', '.join(sorted(SUPPORTED_FORMATS))}."
-        )
+    data_type = collection_file.collection.data_type
 
-    if data_format == Format.compiled_release:
+    if data_type["format"] == Format.compiled_release:
         package = None
     else:
         package = _read_package_data_from_file(collection_file.filename, data_type)
@@ -188,12 +183,21 @@ def process_file(collection_file) -> int | None:
     return None
 
 
-def _get_data_type(collection_file):
-    collection = collection_file.collection
+def set_data_type(collection, collection_file):
     if not collection.data_type:
         detected_format, is_concatenated, is_array = detect_format(
             collection_file.filename, additional_prefixes=("extensions",)
         )
+
+        # https://github.com/open-contracting/kingfisher-collect/issues/1012
+        if detected_format == Format.empty_package:
+            raise EmptyFormatError(f"Empty format '{detected_format}' for file {collection_file}.")
+        if detected_format not in SUPPORTED_FORMATS:
+            raise UnsupportedFormatError(
+                f"Unsupported format '{detected_format}' for file {collection_file}. "
+                f"Must be one of: {', '.join(sorted(SUPPORTED_FORMATS))}."
+            )
+
         data_type = {
             "format": detected_format,
             "concatenated": is_concatenated,
@@ -207,8 +211,6 @@ def _get_data_type(collection_file):
             upgraded_collection.data_type = data_type
             upgraded_collection.save(update_fields=["data_type"])
 
-    return collection.data_type
-
 
 class ControlCodesFilter:
     def __init__(self, file):

process/management/commands/finisher.py

@@ -49,8 +49,9 @@ def callback(client_state, channel, method, properties, input_message):
             return
 
         if completable(collection):
-            # COUNT first, before any UPDATE, to avoid locking rows in between UPDATEs.
+            # COUNT first, before any UPDATE, to avoid locking the collection row in between UPDATEs.
             counts = _count_releases_and_records(collection)
+            # If not for the lock, we'd COUNT only if `updated`, to save time on simultaneous messages.
             if upgraded_collection := collection.get_upgraded_collection():
                 upgraded_collection_counts = _count_releases_and_records(upgraded_collection)
             else:

process/util.py

@@ -10,12 +10,11 @@
 import simplejson as json
 from django.conf import settings
 from django.db import IntegrityError, connections, transaction
-from ocdskit.exceptions import UnknownFormatError
 from yapw.clients import AsyncConsumer, Blocking
 from yapw.decorators import decorate
 from yapw.methods import add_callback_threadsafe, nack
 
-from process.exceptions import AlreadyExists, EmptyFormatError, InvalidFormError, UnsupportedFormatError
+from process.exceptions import AlreadyExists, InvalidFormError
 from process.models import Collection, CollectionFile, CollectionNote, ProcessingStep, Record
 
 logger = logging.getLogger(__name__)
@@ -131,10 +130,7 @@ def deleting_step(*args, **kwargs):
         InvalidFormError,
         IntegrityError,
         # See the try/except block in the callback() function of the file_worker worker.
-        EmptyFormatError,
         FileNotFoundError,
-        UnknownFormatError,
-        UnsupportedFormatError,
         ijson.common.IncompleteJSONError,
     ) as exception:
         delete_step(*args, **kwargs, exception=exception)

tests/processors/test_process_file.py

@@ -2,7 +2,7 @@
 from ocdskit.exceptions import UnknownFormatError
 
 from process.exceptions import EmptyFormatError, UnsupportedFormatError
-from process.management.commands.file_worker import process_file
+from process.management.commands.file_worker import process_file, set_data_type
 from process.models import CollectionFile, PackageData, Release
 from tests.fixtures import collection
 
@@ -12,45 +12,58 @@ def test_empty_format(self):
         source = collection()
         source.save()
 
+        collection_file = CollectionFile(collection=source, filename="tests/fixtures/detect-format_empty.json")
+
         with self.assertRaisesRegex(
             EmptyFormatError,
             r"^Empty format 'empty package' for file tests/fixtures/detect-format_empty\.json \(id: {id}\)\.$",
         ):
-            process_file(CollectionFile(collection=source, filename="tests/fixtures/detect-format_empty.json"))
+            set_data_type(source, collection_file)
 
-        self.assertEqual(source.data_type, {"array": False, "format": "empty package", "concatenated": False})
+        self.assertEqual(source.data_type, {})
 
     def test_unsupported_format(self):
         source = collection()
         source.save()
 
+        collection_file = CollectionFile(collection=source, filename="tests/fixtures/detect-format_versioned.json")
+
         with self.assertRaisesRegex(
             UnsupportedFormatError,
             r"^Unsupported format 'versioned release' for file tests/fixtures/detect-format_versioned.json "
             r"\(id: {id}\). Must be one of: compiled release, record package, release package\.$",
         ):
-            process_file(CollectionFile(collection=source, filename="tests/fixtures/detect-format_versioned.json"))
+            set_data_type(source, collection_file)
 
-        self.assertEqual(source.data_type, {"array": False, "format": "versioned release", "concatenated": False})
+        self.assertEqual(source.data_type, {})
 
     def test_unknown_format(self):
         source = collection()
         source.save()
 
+        collection_file = CollectionFile(collection=source, filename="tests/fixtures/detect-format_object.json")
+
         with self.assertRaisesRegex(UnknownFormatError, r"^top-level JSON value is a non-OCDS object$"):
-            process_file(CollectionFile(collection=source, filename="tests/fixtures/detect-format_object.json"))
+            set_data_type(source, collection_file)
 
         self.assertEqual(source.data_type, {})
 
+    def test_file_not_found(self):
+        source = collection()
+        source.save()
 
-class ProcessFileTests(TransactionTestCase):
-    fixtures = ["tests/fixtures/complete_db.json"]
+        collection_file = CollectionFile(collection=source, filename="tests/fixtures/detect-format_nonexistent.json")
 
-    def test_file_not_found(self):
         with self.assertRaises(FileNotFoundError) as e:
-            collection_file = CollectionFile.objects.select_related("collection").get(pk=5)
-            process_file(collection_file)
-        self.assertEqual(str(e.exception), "[Errno 2] No such file or directory: 'ocds-px0z7d-10094-10001-1'")
+            set_data_type(collection_file.collection, collection_file)
+
+        self.assertEqual(
+            str(e.exception), "[Errno 2] No such file or directory: 'tests/fixtures/detect-format_nonexistent.json'"
+        )
+
+
+class ProcessFileTests(TransactionTestCase):
+    fixtures = ["tests/fixtures/complete_db.json"]
 
     def test_happy_day(self):
         collection_file = CollectionFile.objects.get(pk=1)