Merge pull request #15 from django-q2/error-on-timeout

Error on timeout

django_q/cluster.py

@@ -458,6 +458,29 @@ def monitor(result_queue: Queue, broker: Broker = None):
     logger.info(_("%(name)s stopped monitoring results") % {"name": proc_name})
 
 
+def _check_task_timed_out(key, task: dict):
+    result = None
+    broker = get_broker()
+    cache = broker.cache
+    working_set = cache.get(key) or set()
+    if task["id"] in working_set:
+        # the previous worker has timedout and wasn't given chance to clear
+        raise Exception(f"Task Timed-out: {task}.")
+    else:
+        working_set.add(task['id'])
+        cache.set(key, working_set, timeout=Conf.RETRY * 3)
+    return result
+
+
+def _clear_task_timeout_cache(key, task):
+    broker = get_broker()
+    cache = broker.cache
+    working_set = cache.get(key) or set()
+    if task["id"] in working_set:
+        working_set.remove(task['id'])
+        cache.set(key, working_set, timeout=Conf.RETRY * 3)
+
+
 def worker(
     task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT
 ):
@@ -479,6 +502,8 @@ def worker(
     task_count = 0
     if timeout is None:
         timeout = -1
+
+    working_tasks_key = "DJANGO-Q-WORKING-TASKS"
     # Start reading the task queue
     for task in iter(task_queue.get, "STOP"):
         result = None
@@ -518,7 +543,11 @@ def worker(
         pre_execute.send(sender="django_q", func=f, task=task)
         # execute the payload
         timer.value = timer_value  # Busy
+
+
         try:
+            if Conf.FAIL_ON_TIMEOUT:
+                _check_task_timed_out(working_tasks_key, task)
             if f is None:
                 # raise a meaningfull error if task["func"] is not a valid function
                 raise ValueError(f"Function {task['func']} is not defined")
@@ -530,6 +559,9 @@ def worker(
                 error_reporter.report()
             if task.get("sync", False):
                 raise
+        if Conf.FAIL_ON_TIMEOUT:
+            _clear_task_timeout_cache(working_tasks_key, task)
+
         with timer.get_lock():
             # Process result
             task["result"] = result[0]

django_q/conf.py

@@ -133,6 +133,9 @@ class Conf:
     # Number of seconds to wait for a worker to finish.
     TIMEOUT = conf.get("timeout", None)
 
+    # Whether to fail the task when it times-out.
+    FAIL_ON_TIMEOUT = conf.get("fail_on_timeout", False)
+
     # Whether to acknowledge unsuccessful tasks.
     # This causes failed tasks to be considered delivered, thereby removing them from
     # the task queue. Defaults to False.

docs/configure.rst

@@ -70,6 +70,11 @@ Set this to something that makes sense for your project. Can be overridden for i
 
 See :ref:`retry` for details how to set values for timeout and retry.
 
+fail_on_timeout
+~~~~~~~~~~~~~~~
+
+When set to ``True``, timeouts will result in error. Defaults to ``False``.
+
 .. _time_zone:
 
 time_zone

pyproject.toml

@@ -80,3 +80,7 @@ sentry = ["django-q-sentry"]
 [tool.isort]
 profile = "black"
 multi_line_output = 3
+
+[build-system]
+requires = ["poetry-core>=1.0.8"]
+build-backend = "poetry.core.masonry.api"