Merge pull request #761 from gurelkaynak/gurel/issue_759

Fixes #759 Consumers down when ETA is None (or not isnumeric)

Author: synweap15

CONTRIBUTORS.md

@@ -72,3 +72,4 @@ of those changes to CLEARTYPE SRL.
 | [@LincolnPuzey](https://github.com/LincolnPuzey)       | Lincoln Puzey          |
 | [@guedesfelipe](https://github.com/guedesfelipe)       | Felipe Guedes          |
 | [@karolinepauls](https://karolinepauls.com)            | Karoline Pauls         |
+| [@gurelkaynak](https://gurel.kaynak.link)              | Gurel Kaynak           |

dramatiq/worker.py

@@ -343,6 +343,17 @@ def handle_message(self, message: MessageProxy) -> None:
         If the message has an eta, delay it.  Otherwise, put it on the
         work queue.
         """
+        if "eta" in message.options:
+            if message.options["eta"] is None:
+                del message.options["eta"]
+            elif not isinstance(message.options["eta"], (int, float)):
+                self.logger.warning(
+                    "Invalid eta value for message %r: %r. Setting eta to current_millis().",
+                    message.message_id,
+                    message.options["eta"],
+                )
+                message.options["eta"] = current_millis()
+
         try:
             if "eta" in message.options:
                 self.logger.debug("Pushing message %r onto delay queue.", message.message_id)

tests/test_actors.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import time
 from datetime import timedelta
 from unittest.mock import patch
@@ -533,3 +534,51 @@ def f2():
     # Then a ValueError should be raised
     assert exc_info.type is ValueError
     assert str(exc_info.value) == "An actor named 'foo' is already registered."
+
+
+def test_worker_handles_non_numeric_eta_and_logs_warning(stub_broker, stub_worker, caplog):
+    # Set the log level to capture warnings
+    caplog.set_level(logging.WARNING)
+
+    # Given an actor that records when it runs
+    run = []
+
+    @dramatiq.actor
+    def record():
+        run.append(True)
+
+    # If I send it a message with a non-numeric eta manually from another project or manually from rabbitmq UI (not using dramatiq)
+    message = record.message_with_options(eta="not-a-number")
+    stub_broker.queues[record.queue_name].put(message.encode())
+
+    # Then join on the queue
+    stub_broker.join(record.queue_name)
+    stub_worker.join()
+
+    # I expect the message to have been processed
+    assert run
+
+    # And a warning should have been logged about the invalid eta
+    assert any(
+        "Invalid eta value for message" in record.message for record in caplog.records if record.levelname == "WARNING"
+    )
+
+
+def test_worker_handles_none_eta(stub_broker, stub_worker):
+    # Given an actor that records when it runs
+    run = []
+
+    @dramatiq.actor
+    def record():
+        run.append(True)
+
+    # If I send it a message with a None eta
+    message = record.message_with_options(eta=None)
+    stub_broker.queues[record.queue_name].put(message.encode())
+
+    # Then join on the queue
+    stub_broker.join(record.queue_name)
+    stub_worker.join()
+
+    # I expect the message to have been processed
+    assert run