Fix Default Periodic Task Unique ID (#147)

* add (failing) tests
* fix default args/kwargs
* improve error message for queue length assertions
* recalculate id between periodic executions

Author: AlecRosenbaum

tasktiger/task.py

@@ -1,3 +1,4 @@
+import copy
 import datetime
 import json
 import redis
@@ -75,6 +76,9 @@ def __init__(
         if max_queue_size is None:
             max_queue_size = getattr(func, '_task_max_queue_size', None)
 
+        # normalize falsy args/kwargs to empty structures
+        args = args or []
+        kwargs = kwargs or {}
         if unique:
             task_id = gen_unique_id(serialized_name, args, kwargs)
         else:
@@ -518,6 +522,18 @@ def delete(self):
         """
         self._move(from_state=ERROR)
 
+    def clone(self):
+        """Returns a clone of the this task"""
+        return type(self)(
+            tiger=self.tiger,
+            func=self.func,
+            queue=self.queue,
+            _state=self._state,
+            _ts=self._ts,
+            _executions=copy.copy(self._executions),
+            _data=copy.copy(self._data),
+        )
+
     def _queue_for_next_period(self):
         now = datetime.datetime.utcnow()
         schedule = self.func._task_schedule
@@ -528,5 +544,11 @@ def _queue_for_next_period(self):
             schedule_func, schedule_args = schedule
         when = schedule_func(now, *schedule_args)
         if when:
-            self.delay(when=when)
+            # recalculate the unique id so that malformed ids don't persist
+            # between executions
+            task = self.clone()
+            task._data['id'] = gen_unique_id(
+                task.serialized_func, task.args, task.kwargs
+            )
+            task.delay(when=when)
         return when

tests/test_base.py

@@ -56,10 +56,21 @@ def teardown_method(self, method):
     def _ensure_queues(
         self, queued=None, active=None, error=None, scheduled=None
     ):
+
+        expected_queues = {
+            'queued': {name for name, n in (queued or {}).items() if n},
+            'active': {name for name, n in (active or {}).items() if n},
+            'error': {name for name, n in (error or {}).items() if n},
+            'scheduled': {name for name, n in (scheduled or {}).items() if n},
+        }
+        actual_queues = {
+            i: self.conn.smembers('t:{}'.format(i))
+            for i in ('queued', 'active', 'error', 'scheduled')
+        }
+        assert expected_queues == actual_queues
+
         def _ensure_queue(typ, data):
             data = data or {}
-            data_names = set(name for name, n in data.items() if n)
-            assert self.conn.smembers('t:%s' % typ) == data_names
             ret = {}
             for name, n in data.items():
                 task_ids = self.conn.zrange('t:%s:%s' % (typ, name), 0, -1)

tests/test_periodic.py

@@ -4,8 +4,15 @@
 import time
 
 from tasktiger import Worker, periodic
-
-from .tasks_periodic import tiger
+from tasktiger import (
+    Task,
+    gen_unique_id,
+    serialize_func_name,
+    QUEUED,
+    SCHEDULED,
+)
+
+from .tasks_periodic import tiger, periodic_task
 from .test_base import BaseTestCase
 
 
@@ -107,3 +114,117 @@ def ensure_run(n):
         time.sleep(1)
 
         ensure_run(2)
+
+    def test_periodic_execution_unique_ids(self):
+        """
+        Test that periodic tasks generate the same unique ids
+
+        When a periodic task is scheduled initially as part of worker startup
+        vs re-scheduled from within python the unique id generated should be
+        the same. If they aren't it could result in duplicate tasks.
+        """
+        # Sleep until the next second
+        now = datetime.datetime.utcnow()
+        time.sleep(1 - now.microsecond / 10.0 ** 6)
+
+        # After the first worker run, the periodic task will be queued.
+        # Note that since periodic tasks register with the Tiger instance, it
+        # must be the same instance that was used to decorate the task. We
+        # therefore use `tiger` from the tasks module instead of `self.tiger`.
+        self._ensure_queues()
+        Worker(tiger).run(once=True)
+        self._ensure_queues(scheduled={'periodic': 1})
+        time.sleep(1)
+        Worker(tiger).run(once=True)
+        self._ensure_queues(queued={'periodic': 1})
+
+        # generate the expected unique id
+        expected_unique_id = gen_unique_id(
+            serialize_func_name(periodic_task), [], {}
+        )
+
+        # pull task out of the queue by id. If found, then the id is correct
+        task = Task.from_id(tiger, 'periodic', QUEUED, expected_unique_id)
+        assert task is not None
+
+        # execute and reschedule the task
+        self._ensure_queues(queued={'periodic': 1})
+        Worker(tiger).run(once=True)
+        self._ensure_queues(scheduled={'periodic': 1})
+
+        # wait for the task to need to be queued
+        time.sleep(1)
+        Worker(tiger).run(once=True)
+        self._ensure_queues(queued={'periodic': 1})
+
+        # The unique id shouldn't change between executions. Try finding the
+        # task by id again
+        task = Task.from_id(tiger, 'periodic', QUEUED, expected_unique_id)
+        assert task is not None
+
+    def test_periodic_execution_unique_ids_manual_scheduling(self):
+        """
+        Periodic tasks should have the same unique ids when manually scheduled
+
+        When a periodic task is scheduled initially as part of worker startup
+        vs ``.delay``'d manually, the unique id generated should be the same.
+        If they aren't it could result in duplicate tasks.
+        """
+        # Sleep until the next second
+        now = datetime.datetime.utcnow()
+        time.sleep(1 - now.microsecond / 10.0 ** 6)
+
+        # After the first worker run, the periodic task will be queued.
+        # Note that since periodic tasks register with the Tiger instance, it
+        # must be the same instance that was used to decorate the task. We
+        # therefore use `tiger` from the tasks module instead of `self.tiger`.
+        self._ensure_queues()
+        Worker(tiger).run(once=True)
+        self._ensure_queues(scheduled={'periodic': 1})
+        time.sleep(1)
+        Worker(tiger).run(once=True)
+        self._ensure_queues(queued={'periodic': 1})
+
+        # schedule the task manually
+        periodic_task.delay()
+
+        # make sure a duplicate wasn't scheduled
+        self._ensure_queues(queued={'periodic': 1})
+
+    def test_periodic_execution_unique_ids_self_correct(self):
+        """
+        Test that periodic tasks will self-correct unique ids
+        """
+        # Sleep until the next second
+        now = datetime.datetime.utcnow()
+        time.sleep(1 - now.microsecond / 10.0 ** 6)
+
+        # generate the ids
+        correct_unique_id = gen_unique_id(
+            serialize_func_name(periodic_task), [], {}
+        )
+        malformed_unique_id = gen_unique_id(
+            serialize_func_name(periodic_task), None, None
+        )
+
+        task = Task(tiger, func=periodic_task)
+
+        # patch the id to something slightly wrong
+        assert task.id == correct_unique_id
+        task._data['id'] = malformed_unique_id
+        assert task.id == malformed_unique_id
+
+        # schedule the task
+        task.delay()
+        self._ensure_queues(queued={'periodic': 1})
+
+        # pull task out of the queue by the malformed id
+        task = Task.from_id(tiger, 'periodic', QUEUED, malformed_unique_id)
+        assert task is not None
+
+        Worker(tiger).run(once=True)
+        self._ensure_queues(scheduled={'periodic': 1})
+
+        # pull task out of the queue by the self-corrected id
+        task = Task.from_id(tiger, 'periodic', SCHEDULED, correct_unique_id)
+        assert task is not None