Clear retried executions from successful periodic tasks (#188)

Previously, if you had a periodic task that retries up to N times for valid reasons, we would never clear the executions, so N would accumulate over the lifetime of a periodic task (forever). This commit doesn't completely fix the issue of tracking executions for previously failed periods, but at least it prevents executions from accumulating if the periodic task completes successfully after retries.

Author: thomasst

tasktiger/task.py

@@ -274,16 +274,23 @@ def _move(self, from_state=None, to_state=None, when=None, mode=None):
 
         if not to_state:  # Remove the task if necessary
             if self.unique:
-                # Only delete if it's not in any other queue
-                check_states = {ACTIVE, QUEUED, ERROR, SCHEDULED}
-                check_states.remove(from_state)
                 # TODO: Do the following two in one call.
+
+                # Delete executions if there were no errors.
+                if from_state == ERROR:
+                    check_states = {}
+                else:
+                    check_states = {ERROR}
                 scripts.delete_if_not_in_zsets(
                     _key('task', self.id, 'executions'),
                     self.id,
                     [_key(state, queue) for state in check_states],
                     client=pipeline,
                 )
+
+                # Only delete task if it's not in any other queue
+                check_states = {ACTIVE, QUEUED, ERROR, SCHEDULED}
+                check_states.remove(from_state)
                 scripts.delete_if_not_in_zsets(
                     _key('task', self.id),
                     self.id,

tests/tasks_periodic.py

@@ -10,11 +10,18 @@
 tiger = get_tiger()
 
 
-@tiger.task(schedule=periodic(seconds=1), queue='periodic')
+@tiger.task(
+    schedule=periodic(seconds=1), queue='periodic', retry_on=(ValueError,)
+)
 def periodic_task():
     """Periodic task."""
     conn = redis.Redis(host=REDIS_HOST, db=TEST_DB, decode_responses=True)
     conn.incr('period_count', 1)
+    fail = conn.get('fail-periodic-task')
+    if fail == 'retriable':
+        raise ValueError('retriable failure')
+    elif fail == 'permanent':
+        raise Exception('permanent failure')
 
 
 @tiger.task(schedule=periodic(seconds=1), queue='periodic_ignore')

tests/test_periodic.py

@@ -229,3 +229,97 @@ def test_periodic_execution_unique_ids_self_correct(self):
         # 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
+
+    def test_successful_execution_clears_executions_from_retries(self):
+        """
+        Ensure previous executions from retries are cleared after a successful
+        execution.
+        """
+        sleep_until_next_second()
+
+        # Queue the periodic task.
+        self._ensure_queues()
+        Worker(tiger).run(once=True)
+
+        # Prepare to execute the periodic task (as retriable failure).
+        tiger.connection.set('fail-periodic-task', 'retriable')
+        n_total, tasks = Task.tasks_from_queue(tiger, 'periodic', SCHEDULED)
+        task_id = tasks[0].id
+        time.sleep(1)
+
+        # Queue the periodic task.
+        self._ensure_queues(scheduled={'periodic': 1})
+        Worker(tiger).run(once=True)
+
+        # Run the failing periodic task.
+        self._ensure_queues(queued={'periodic': 1})
+        Worker(tiger).run(once=True)
+
+        task = Task.from_id(
+            tiger, 'periodic', SCHEDULED, task_id, load_executions=10
+        )
+        assert len(task.executions) == 1
+
+        tiger.connection.delete('fail-periodic-task')
+        time.sleep(1)
+
+        # Queue the periodic task.
+        self._ensure_queues(scheduled={'periodic': 1})
+        Worker(tiger).run(once=True)
+
+        # Run the successful periodic task.
+        self._ensure_queues(queued={'periodic': 1})
+        Worker(tiger).run(once=True)
+
+        # Ensure we cleared any previous executions.
+        task = Task.from_id(
+            tiger, 'periodic', SCHEDULED, task_id, load_executions=10
+        )
+        assert len(task.executions) == 0
+
+    def test_successful_execution_doesnt_clear_previous_errors(self):
+        """
+        Ensure previous executions are not cleared if we have had non-retriable
+        errors.
+        """
+        sleep_until_next_second()
+
+        # Queue the periodic task.
+        self._ensure_queues()
+        Worker(tiger).run(once=True)
+
+        # Prepare to execute the periodic task (as permanent failure).
+        tiger.connection.set('fail-periodic-task', 'permanent')
+        n_total, tasks = Task.tasks_from_queue(tiger, 'periodic', SCHEDULED)
+        task_id = tasks[0].id
+        time.sleep(1)
+
+        # Queue the periodic task.
+        self._ensure_queues(scheduled={'periodic': 1})
+        Worker(tiger).run(once=True)
+
+        # Run the failing periodic task.
+        self._ensure_queues(queued={'periodic': 1})
+        Worker(tiger).run(once=True)
+
+        task = Task.from_id(
+            tiger, 'periodic', SCHEDULED, task_id, load_executions=10
+        )
+        assert len(task.executions) == 1
+
+        tiger.connection.delete('fail-periodic-task')
+        time.sleep(1)
+
+        # Queue the periodic task.
+        self._ensure_queues(scheduled={'periodic': 1}, error={'periodic': 1})
+        Worker(tiger).run(once=True)
+
+        # Run the successful periodic task.
+        self._ensure_queues(queued={'periodic': 1}, error={'periodic': 1})
+        Worker(tiger).run(once=True)
+
+        # Ensure we didn't clear previous executions.
+        task = Task.from_id(
+            tiger, 'periodic', SCHEDULED, task_id, load_executions=10
+        )
+        assert len(task.executions) == 1