Safer registration of data collections

Author: stephen-riggs

src/murfey/server/feedback.py

@@ -2226,7 +2226,7 @@ def feedback_callback(header: dict, message: dict, _db=murfey_db) -> None:
                 murfey_db=_db,
             )
             if murfey.server._transport_object:
-                if result.get("success", False):
+                if result.get("success"):
                     murfey.server._transport_object.transport.ack(header)
                 else:
                     # Send it directly to DLQ without trying to rerun it

src/murfey/workflows/register_data_collection.py

@@ -1,4 +1,5 @@
 import logging
+import time
 
 import ispyb.sqlalchemy._auto_db_schema as ISPyBDB
 from sqlmodel import select
@@ -39,6 +40,7 @@ def run(
         dcgid = dcg[0].id
         # flush_data_collections(message["source"], murfey_db)
     else:
+        time.sleep(2)
         logger.warning(
             "No data collection group ID was found for image directory "
             f"{sanitise(message['image_directory'])} and source "
@@ -84,21 +86,20 @@ def run(
                     else ""
                 ),
             ).get("return_value", None)
+            if dcid is None:
+                time.sleep(2)
+                logger.error(
+                    "Failed to register the following data collection: \n"
+                    f"{message} \n"
+                    "Requeueing message"
+                )
+                return {"success": False, "requeue": True}
             murfey_dc = MurfeyDB.DataCollection(
                 id=dcid,
                 tag=message.get("tag"),
                 dcg_id=dcgid,
             )
         murfey_db.add(murfey_dc)
         murfey_db.commit()
-        dcid = murfey_dc.id
         murfey_db.close()
-
-    if dcid is None:
-        logger.error(
-            "Failed to register the following data collection: \n"
-            f"{message} \n"
-            "Requeueing message"
-        )
-        return {"success": False, "requeue": True}
     return {"success": True}

src/murfey/workflows/register_data_collection_group.py

@@ -54,6 +54,15 @@ def run(
                 "return_value", None
             )
 
+            if dcgid is None:
+                time.sleep(2)
+                logger.error(
+                    "Failed to register the following data collection group: \n"
+                    f"{message} \n"
+                    "Requeuing message"
+                )
+                return {"success": False, "requeue": True}
+
             atlas_record = ISPyBDB.Atlas(
                 dataCollectionGroupId=dcgid,
                 atlasImage=message.get("atlas", ""),
@@ -77,15 +86,6 @@ def run(
         murfey_db.commit()
         murfey_db.close()
 
-    if dcgid is None:
-        time.sleep(2)
-        logger.error(
-            "Failed to register the following data collection group: \n"
-            f"{message} \n"
-            "Requeuing message"
-        )
-        return {"success": False, "requeue": True}
-
     if dcg_hooks := entry_points(group="murfey.hooks", name="data_collection_group"):
         try:
             for hook in dcg_hooks:

src/murfey/workflows/register_processing_job.py

@@ -63,6 +63,8 @@ def run(message: dict, murfey_db: SQLModelSession, demo: bool = False):
                 pid = _transport_object.do_create_ispyb_job(record).get(
                     "return_value", None
                 )
+            if pid is None:
+                return {"success": False, "requeue": True}
             murfey_pj = MurfeyDB.ProcessingJob(
                 id=pid, recipe=message["recipe"], dc_id=_dcid
             )
@@ -71,9 +73,6 @@ def run(message: dict, murfey_db: SQLModelSession, demo: bool = False):
         pid = murfey_pj.id
         murfey_db.close()
 
-    if pid is None:
-        return {"success": False, "requeue": True}
-
     # Update Prometheus counter for preprocessed movies
     prom.preprocessed_movies.labels(processing_job=pid)