Use distinct worker ID + pre-queuing validation

Author: mephenor

.devcontainer/.dev_config.yaml

@@ -8,3 +8,4 @@ kafka_enable_dlq: True
 input_config_path: input_dummy.yaml
 original_aem_pack_topic: original-aempacks
 derived_aem_pack_topic: derived-aempacks
+worker_id: test-worker

src/ets/adapters/outbound/incoming_aem_pack_queue.py

@@ -37,10 +37,10 @@ def __init__(
         self,
         *,
         collection: AsyncCollection,
-        service_instance_id: str,
+        worker_id: str,
     ):
         self._collection = collection
-        self._service_instance_id = service_instance_id
+        self._worker_id = worker_id
 
     async def queue(self, aem_pack: AEMPack) -> None:
         """Upsert an AEMPack into the queue."""
@@ -87,15 +87,15 @@ async def claim_next(self) -> IncomingAEMPack | None:
         # Check for packs abandoned by a previous crash of this instance
         doc = await self._collection.find_one(
             {
-                PROCESSOR_FIELD: self._service_instance_id,
+                PROCESSOR_FIELD: self._worker_id,
                 PROCESSED_AT_FIELD: None,
             }
         )
         if not doc:
             # No abandoned packs; try to claim a fresh one
             doc = await self._collection.find_one_and_update(
                 filter={PROCESSOR_FIELD: None, PROCESSED_AT_FIELD: None},
-                update={"$set": {PROCESSOR_FIELD: self._service_instance_id}},
+                update={"$set": {PROCESSOR_FIELD: self._worker_id}},
                 return_document=ReturnDocument.AFTER,
             )
         if not doc:
@@ -107,7 +107,7 @@ async def claim_next(self) -> IncomingAEMPack | None:
                 },
                 update={
                     "$set": {
-                        PROCESSOR_FIELD: self._service_instance_id,
+                        PROCESSOR_FIELD: self._worker_id,
                         PROCESSED_AT_FIELD: None,
                         NEEDS_REPROCESSING_FIELD: False,
                     }

src/ets/config.py

@@ -45,3 +45,7 @@ class Config(
         default=60,
         description="Seconds to sleep when no unprocessed AEMPacks are found.",
     )
+    worker_id: str = Field(
+        default=...,
+        description="Unique identifier for a service instance used specifically for the reclamation logic.",
+    )

src/ets/core/aem_pack_registry.py

@@ -24,6 +24,8 @@
 from metldata import get_transformation_registry
 from metldata.transform.handling import TransformationHandler
 from pydantic import UUID4, BaseModel, ConfigDict
+from schemapack import SchemaPackValidator
+from schemapack.exceptions import ValidationError
 from schemapack.spec.datapack import DataPack
 from schemapack.spec.schemapack import SchemaPack
 
@@ -64,6 +66,30 @@ def __init__(
 
     async def queue_unprocessed(self, aem_pack: AEMPack):
         """Fetch new AEMPacks via event subscriber and put them into the queue for processing."""
+        # For now, just load. Reloading will be dealt with in the config lock + update ticket
+        config = await self._config_loader.load_config_from_db()
+
+        # Ensure model name exists
+        matching_model = None
+        for model in config.models:
+            if model.name == aem_pack.model_name:
+                matching_model = model
+                break
+        else:
+            model_lookup_error = ValueError(
+                f"No model with name {aem_pack.model_name} registered for AEMPack with id {aem_pack.id}."
+            )
+            log.error(model_lookup_error)
+            raise model_lookup_error
+
+        # Validate DataPack against corresponding SchemaPack
+        validator = SchemaPackValidator(schemapack=matching_model.schema_)
+        try:
+            validator.validate(datapack=aem_pack.data)
+        except ValidationError as error:
+            log.error(error)
+            raise
+
         await self._incoming_aem_pack_queue.queue(aem_pack)
 
     async def process_aem_packs(self) -> None:
@@ -146,9 +172,12 @@ def _traverse_graph(
             routes_by_input.setdefault(route.input_model_name, []).append(route)
 
         if not models_by_name.get(incoming.model_name):
-            raise ValueError(
+            model_lookup_error = ValueError(
                 f"No model with name {incoming.model_name} registered for AEMPack with id {incoming.id}."
+                + "This means a previously existing model vanished, which should not happen."
             )
+            log.critical(model_lookup_error)
+            raise model_lookup_error
 
         # Relies on dict insertion-order guarantee.
         # Avoid copying or re-sorting this dict, as that would break the traversal order.

src/ets/inject.py

@@ -94,7 +94,7 @@ async def prepare_aem_pack_registry(
         )
         incoming_aem_pack_queue = IncomingAEMPackQueue(
             collection=mongo_client[config.db_name][INCOMING_AEM_PACK_COLLECTION],
-            service_instance_id=config.service_instance_id,
+            worker_id=config.worker_id,
         )
 
         yield AEMPackRegistry(

tests/fixtures/test_config.yaml

@@ -7,3 +7,4 @@ kafka_enable_dlq: True
 input_config_path: tests/fixtures/input_configs/manager/valid/basic_config.yaml
 original_aem_pack_topic: original-aempacks
 derived_aem_pack_topic: derived-aempacks
+worker_id: test-worker

tests/test_aem_pack_registry/test_queue_and_claim.py

@@ -104,7 +104,7 @@ async def test_abandoned_pack_reclaimed_by_same_instance(joint_fixture: JointFix
     # Simulate a previous crash: the doc is already claimed by this instance
     await joint_fixture.incoming_aem_pack_collection.update_one(
         {"_id": ingress.id},
-        {"$set": {PROCESSOR_FIELD: joint_fixture.config.service_instance_id}},
+        {"$set": {PROCESSOR_FIELD: joint_fixture.config.worker_id}},
     )
 
     await _assert_pack_claimed_during_processing(registry, ingress.id)
@@ -184,7 +184,7 @@ async def test_concurrent_queue_publishes_and_leaves_for_reprocessing(
     # Processor is preserved so in-flight instance can complete; needs_reprocessing signals v2 is pending
     raw = await joint_fixture.incoming_aem_pack_collection.find_one({"_id": aem_id})
     assert raw is not None
-    assert raw["processor"] == joint_fixture.config.service_instance_id
+    assert raw["processor"] == joint_fixture.config.worker_id
     assert raw["needs_reprocessing"] is True
 
     # Process v1 — should still publish