Switch rabbitmq message nacks for rejections (#266)

A way to resolve the rabbitmq infinite loops we have been seeing. This creates a new functions based on python-workflows methods to reject messages rather than nack them. Then replaces (almost) all instances of nack with reject.

I'm not certain I've got every requeue choice right but the general policy is:

-   requeue=False if the input message is wrong (e.g. missing parameter, malformed directory)
-   requeue=True if the failure is external (e.g. failed subprocess call, missing file)

There is a resort back to nacking if not using pika transport, mostly to avoid having to rewrite all the tests

Author: stephen-riggs

src/cryoemservices/services/bfactor_setup.py

@@ -54,7 +54,7 @@ def bfactor_setup(self, rw, header: dict, message: dict):
             self.log.info("Received a simple message")
             if not isinstance(message, dict):
                 self.log.error("Rejected invalid simple message")
-                self._transport.nack(header)
+                self._reject_message(header, requeue=False)
                 return
 
             # Create a wrapper-like object that can be passed to functions
@@ -77,7 +77,7 @@ def bfactor_setup(self, rw, header: dict, message: dict):
                 f"and recipe parameters: {rw.recipe_step.get('parameters', {})} "
                 f"with exception: {e}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
 
         self.log.info(

src/cryoemservices/services/class2d.py

@@ -33,7 +33,7 @@ def class2d(self, rw, header: dict, message: dict):
             self.log.info("Received a simple message")
             if not isinstance(message, dict):
                 self.log.error("Rejected invalid simple message")
-                self._transport.nack(header)
+                self._reject_message(header, requeue=False)
                 return
 
             # Create a wrapper-like object that can be passed to functions
@@ -57,14 +57,14 @@ def class2d(self, rw, header: dict, message: dict):
                 f"and recipe parameters: {rw.recipe_step.get('parameters', {})} "
                 f"with exception: {e}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
 
-        # In this setup we cannot nack messages on failure, so instead check here
+        # In this setup we cannot reject messages on failure, so instead check here
         if message.get("requeue", 0) >= 5:
-            self.log.warning(f"Nacking requeued file {class2d_params.particles_file}")
-            rw.transport.nack(header)
-            return False
+            self.log.warning(f"Rejecting requeued file {class2d_params.particles_file}")
+            self._reject_message(header, transport=rw.transport, requeue=False)
+            return
 
         # Acknowledge the message and disconnect from rabbitmq
         self.log.info(

src/cryoemservices/services/class3d.py

@@ -32,8 +32,8 @@ def class3d(self, rw, header: dict, message: dict):
         if not rw:
             self.log.info("Received a simple message")
             if not isinstance(message, dict):
-                self.log.error("Rejected invalid simple message")
-                self._transport.nack(header)
+                self.log.error("Rejected invaid simple message")
+                self._reject_message(header, requeue=False)
                 return
 
             # Create a wrapper-like object that can be passed to functions
@@ -57,14 +57,14 @@ def class3d(self, rw, header: dict, message: dict):
                 f"and recipe parameters: {rw.recipe_step.get('parameters', {})} "
                 f"with exception: {e}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
 
-        # In this setup we cannot nack messages on failure, so instead check here
+        # In this setup we cannot reject messages on failure, so instead check here
         if message.get("requeue", 0) >= 5:
-            self.log.warning(f"Nacking requeued file {class3d_params.particles_file}")
-            rw.transport.nack(header)
-            return False
+            self.log.warning(f"Rejecting requeued file {class3d_params.particles_file}")
+            self._reject_message(header, transport=rw.transport, requeue=False)
+            return
 
         # Acknowledge the message and disconnect from rabbitmq
         self.log.info(

src/cryoemservices/services/clem_align_and_merge.py

@@ -38,7 +38,7 @@ def call_align_and_merge(self, rw, header, message):
             self.log.info("Received a simple message")
             if not isinstance(message, dict):
                 self.log.error("Rejected invalid simple message")
-                self._transport.nack(header)
+                self._reject_message(header, requeue=False)
                 return
 
             # Create a wrapper-like object that can be passed to functions
@@ -61,7 +61,7 @@ def call_align_and_merge(self, rw, header, message):
                 f"and recipe parameters: {rw.recipe_step.get('parameters', {})} "
                 f"with exception: {e}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
 
         # Process files and collect output
@@ -75,20 +75,20 @@ def call_align_and_merge(self, rw, header, message):
                 align_across=params.align_across,
                 num_procs=params.num_procs,
             )
-        # Log error and nack message if the command fails to execute
+        # Log error and reject message if the command fails to execute
         except Exception:
             self.log.error(
                 f"Exception encountered while aligning and merging images for {params.series_name!r}: \n",
                 exc_info=True,
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
         if not result:
             self.log.error(
                 "Failed to complete the aligning and merging process for "
                 f"{params.series_name!r}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
 
         # Request for PNG image to be created

src/cryoemservices/services/clem_process_raw_lifs.py

@@ -38,7 +38,7 @@ def call_process_raw_lifs(self, rw, header, message):
             self.log.info("Received a simple message")
             if not isinstance(message, dict):
                 self.log.error("Rejected invalid simple message")
-                self._transport.nack(header)
+                self._reject_message(header, requeue=False)
                 return
 
             # Create a wrapper-like object that can be passed to functions
@@ -61,7 +61,7 @@ def call_process_raw_lifs(self, rw, header, message):
                 f"and recipe parameters: {rw.recipe_step.get('parameters', {})} "
                 f"with exception: {e}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
 
         # Process files and collect output
@@ -71,19 +71,19 @@ def call_process_raw_lifs(self, rw, header, message):
                 root_folder=params.root_folder,
                 number_of_processes=params.num_procs,
             )
-        # Log error and nack message if the command fails to execute
+        # Log error and reject message if the command fails to execute
         except Exception:
             self.log.error(
-                f"Exception encontered while processing LIF file {str(params.lif_file)!r}: \n",
+                f"Exception encountered while processing LIF file {str(params.lif_file)!r}: \n",
                 exc_info=True,
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
         if not results:
             self.log.error(
                 f"Failed to extract image stacks from {str(params.lif_file)!r}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
 
         # Send each subset of output files to Murfey for registration

src/cryoemservices/services/clem_process_raw_tiffs.py

@@ -38,7 +38,7 @@ def call_process_raw_tiffs(self, rw, header, message):
             self.log.info("Received a simple message")
             if not isinstance(message, dict):
                 self.log.error("Rejected invalid simple message")
-                self._transport.nack(header)
+                self._reject_message(header, requeue=False)
                 return
 
             # Create a wrapper-like object that can be passed to functions
@@ -61,7 +61,7 @@ def call_process_raw_tiffs(self, rw, header, message):
                 f"and recipe parameters: {rw.recipe_step.get('parameters', {})} "
                 f"with exception: {e}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
 
         # Reconstruct series name using reference file from list
@@ -74,7 +74,7 @@ def call_process_raw_tiffs(self, rw, header, message):
                 f"Subpath {params.root_folder!r} was not found in file path "
                 f"{str(ref_file.parent / ref_file.stem.split('--')[0])!r}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
         series_name = "--".join(
             [p.replace(" ", "_") if " " in p else p for p in path_parts][
@@ -90,19 +90,19 @@ def call_process_raw_tiffs(self, rw, header, message):
                 metadata_file=params.metadata,
                 number_of_processes=params.num_procs,
             )
-        # Log error and nack message if the command fails to execute
+        # Log error and reject message if the command fails to execute
         except Exception:
             self.log.error(
                 f"Exception encountered while processing TIFF files for series {series_name}: \n",
                 exc_info=True,
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
         if not result:
             self.log.error(
                 f"No processing results were returned for TIFF series {series_name!r}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
 
         # Request for PNG images to be created

src/cryoemservices/services/cluster_submission.py

@@ -46,7 +46,7 @@ def run_submit_job(self, rw, header, message):
                 Path(wrapper).parent.mkdir(parents=True, exist_ok=True)
             except OSError:
                 self.log.error(f"Cannot make directory for {wrapper}")
-                self._transport.nack(header)
+                self._reject_message(header, transport=rw.transport)
                 return
             self.log.info(f"Storing serialized recipe wrapper in {wrapper}")
             cluster_params.commands = cluster_params.commands.replace(
@@ -72,7 +72,7 @@ def run_submit_job(self, rw, header, message):
             self.log.error(
                 "No absolute working directory specified. Will not run cluster job"
             )
-            self._transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
         working_directory = Path(parameters["workingdir"])
         try:
@@ -81,7 +81,7 @@ def run_submit_job(self, rw, header, message):
             self.log.error(
                 "Could not create working directory: %s", str(e), exc_info=True
             )
-            self._transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
 
         if parameters.get("standard_output"):
@@ -105,7 +105,7 @@ def run_submit_job(self, rw, header, message):
         )
         if not jobnumber:
             self.log.error("Job was not submitted")
-            self._transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
 
         # Conditionally acknowledge receipt of the message

src/cryoemservices/services/common_service.py

@@ -2,8 +2,10 @@
 
 import logging
 import queue
+from functools import partial
 
 from workflows.transport.common_transport import CommonTransport
+from workflows.transport.pika_transport import PikaTransport
 
 
 class CommonService:
@@ -66,3 +68,32 @@ def start(self):
             self._transport.disconnect()
         except Exception as e:
             self.log.error(f"Could not disconnect transport: {e}", exc_info=True)
+
+    def _reject_message(
+        self,
+        header: dict,
+        transport: CommonTransport | None = None,
+        requeue: bool = True,
+    ):
+        """Reject failed messages back to rabbitmq"""
+        message_id = header.get("message-id")
+        subscription_id = header.get("subscription")
+        if transport is None:
+            transport = self._transport
+        if (
+            isinstance(transport, PikaTransport)
+            and message_id is not None
+            and subscription_id is not None
+        ):
+            pika_thread = transport._pika_thread
+            channel = pika_thread._pika_channels[subscription_id]
+            pika_thread._connection.add_callback_threadsafe(
+                partial(channel.basic_reject, delivery_tag=message_id, requeue=requeue)
+            )
+        else:
+            # Resort back to nacking if this isn't pika or the header is invalid
+            # Mostly just for tests compatibility
+            self.log.warning(
+                f"Message {message_id} in {subscription_id} is not valid for rabbitmq"
+            )
+            transport.nack(header, requeue=requeue)

src/cryoemservices/services/cryolo.py

@@ -112,7 +112,7 @@ def cryolo(self, rw, header: dict, message: dict):
             self.log.info("Received a simple message")
             if not isinstance(message, dict):
                 self.log.error("Rejected invalid simple message")
-                self._transport.nack(header)
+                self._reject_message(header, requeue=False)
                 return
 
             # Create a wrapper-like object that can be passed to functions
@@ -138,7 +138,7 @@ def cryolo(self, rw, header: dict, message: dict):
                 f"and recipe parameters: {rw.recipe_step.get('parameters', {})} "
                 f"with exception: {e}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
 
         # Check if this file has been run before
@@ -160,7 +160,7 @@ def cryolo(self, rw, header: dict, message: dict):
             job_number = int(job_num_search[0][4:7])
         else:
             self.log.warning(f"Invalid job directory in {cryolo_params.output_path}")
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
 
         # Check job alias
@@ -169,7 +169,7 @@ def cryolo(self, rw, header: dict, message: dict):
             job_alias.symlink_to(job_dir)
         elif not (job_alias.is_symlink() and job_alias.resolve() == job_dir.resolve()):
             self.log.error(f"Symlink {job_alias} already exists")
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
 
         Path(cryolo_params.output_path).unlink(missing_ok=True)
@@ -302,7 +302,7 @@ def cryolo(self, rw, header: dict, message: dict):
                 f"crYOLO failed with exitcode {result.returncode}:\n"
                 + result.stderr.decode("utf8", "replace")
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
 
         # If this is tomo then make an image and stop here

src/cryoemservices/services/ctffind.py

@@ -117,7 +117,7 @@ def ctf_find(self, rw, header: dict, message: dict):
             self.log.info("Received a simple message")
             if not isinstance(message, dict):
                 self.log.error("Rejected invalid simple message")
-                self._transport.nack(header)
+                self._reject_message(header, requeue=False)
                 return
 
             # Create a wrapper-like object that can be passed to functions
@@ -138,12 +138,12 @@ def ctf_find(self, rw, header: dict, message: dict):
                 f"and recipe parameters: {rw.recipe_step.get('parameters', {})} "
                 f"with exception: {e}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
 
         if ctf_params.ctffind_version not in [4, 5]:
             self.log.error(f"Cannot use CTFFind version {ctf_params.ctffind_version}")
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
         self.log.info(f"Using CTFFind version {ctf_params.ctffind_version}")
 
@@ -168,7 +168,7 @@ def ctf_find(self, rw, header: dict, message: dict):
             self.log.warning(
                 f"Could not determine job number in {ctf_params.output_image}"
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport, requeue=False)
             return
         job_alias = Path(
             re.sub(
@@ -185,7 +185,7 @@ def ctf_find(self, rw, header: dict, message: dict):
             == (job_alias.parent / f"job{ctf_job_number:03}").resolve()
         ):
             self.log.error(f"Symlink {job_alias} already exists")
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
 
         parameters_list = [
@@ -271,7 +271,7 @@ def ctf_find(self, rw, header: dict, message: dict):
                 f"CTFFind failed with exitcode {result.returncode}:\n"
                 + result.stderr.decode("utf8", "replace")
             )
-            rw.transport.nack(header)
+            self._reject_message(header, transport=rw.transport)
             return
 
         # Write stdout to logfile