From 0e2df88d92715b7b92bf04bf0d05c56408ddf6a7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Janne=20R=C3=B6nkk=C3=B6?= <jroo@iki.fi>
Date: Sun, 27 Jan 2019 15:47:06 +0200
Subject: [PATCH] Fix async_task timeout parameter handling when cluster
 timeout is set to None (the default)

The cluster timeout configuration default value None is documented to mean
tasks never timeout out. Also the documentation states that the timeout
can be overridden for individual tasks.

With the old implementation the timeout parameter given to async_task was
not honored if the cluster timeout was set to None.

Fixes: #335
---
 django_q/cluster.py            | 10 ++++++----
 django_q/tests/test_cluster.py |  2 ++
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/django_q/cluster.py b/django_q/cluster.py
index dc1b46c..534f887 100644
--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -172,7 +172,7 @@ class Sentinel(object):
         else:
             self.pool.remove(process)
             self.spawn_worker()
-            if self.timeout and int(process.timer.value) == 0:
+            if process.timer.value == 0:
                 # only need to terminate on timeout, otherwise we risk destabilizing the queues
                 process.terminate()
                 logger.warn(_("reincarnated worker {} after timeout").format(process.name))
@@ -210,11 +210,11 @@ class Sentinel(object):
             # Check Workers
             for p in self.pool:
                 # Are you alive?
-                if not p.is_alive() or (self.timeout and p.timer.value == 0):
+                if not p.is_alive() or p.timer.value == 0:
                     self.reincarnate(p)
                     continue
                 # Decrement timer if work is being done
-                if self.timeout and p.timer.value > 0:
+                if p.timer.value > 0:
                     p.timer.value -= cycle
             # Check Monitor
             if not self.monitor.is_alive():
@@ -347,6 +347,8 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
     name = current_process().name
     logger.info(_('{} ready for work at {}').format(name, current_process().pid))
     task_count = 0
+    if timeout is None:
+        timeout = -1
     # Start reading the task queue
     for task in iter(task_queue.get, 'STOP'):
         result = None
@@ -368,7 +370,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
         # We're still going
         if not result:
             db.close_old_connections()
-            timer_value = task['kwargs'].pop('timeout', timeout or 0)
+            timer_value = task['kwargs'].pop('timeout', timeout)
             # signal execution
             pre_execute.send(sender="django_q", func=f, task=task)
             # execute the payload
diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py
index 7dc0959..5e289d7 100644
--- a/django_q/tests/test_cluster.py
+++ b/django_q/tests/test_cluster.py
@@ -247,6 +247,7 @@ def test_enqueue(broker, admin_user):
 @pytest.mark.parametrize('cluster_config_timeout, async_task_kwargs', (
     (1, {}),
     (10, {'timeout': 1}),
+    (None, {'timeout': 1}),
 ))
 def test_timeout(broker, cluster_config_timeout, async_task_kwargs):
     # set up the Sentinel
@@ -269,6 +270,7 @@ def test_timeout(broker, cluster_config_timeout, async_task_kwargs):
     (5, {}),
     (10, {'timeout': 5}),
     (1, {'timeout': 5}),
+    (None, {'timeout': 5}),
 ))
 def test_timeout_task_finishes(broker, cluster_config_timeout, async_task_kwargs):
     # set up the Sentinel