Allow jobs to specify retry settings

Jobs can specify how often, and for how long, they should be
automatically retried before being considered an error.  This change
relies on the scheduler service to perform the actual retrying.

An example class:

    class ExampleJob(object):
        queue = 'basic'
        retry_every = 30    # retry every 30 seconds
        retry_timeout = 600 # give up trying after 10 minutes

        @staticmethod
        def perform():
            # do something...

Author: Kevin McConnell

pyres/__init__.py

@@ -142,8 +142,11 @@ def enqueue(self, klass, *args):
         else:
             logging.warning("unable to enqueue job with class %s" % str(klass))
 
-    def enqueue_from_string(self, klass_as_string, queue, *args):
-        self.push(queue, {'class':klass_as_string,'args':args})
+    def enqueue_from_string(self, klass_as_string, queue, *args, **kwargs):
+        payload = {'class':klass_as_string, 'queue': queue, 'args':args}
+        if 'first_attempt' in kwargs:
+            payload['first_attempt'] = kwargs['first_attempt']
+        self.push(queue, payload)
         logging.info("enqueued '%s' job" % klass_as_string)
         if args:
             logging.debug("job arguments: %s" % str(args))
@@ -201,12 +204,15 @@ def close(self):
         """
         self.redis.disconnect()
 
-    def enqueue_at(self, datetime, klass, *args):
+    def enqueue_at(self, datetime, klass, *args, **kwargs):
         class_name = '%s.%s' % (klass.__module__, klass.__name__)
         logging.info("enqueued '%s' job for execution at %s" % (class_name, datetime))
         if args:
             logging.debug("job arguments are: %s" % str(args))
-        self.delayed_push(datetime, {'class':class_name,'queue': klass.queue, 'args':args})
+        payload = {'class':class_name, 'queue': klass.queue, 'args':args}
+        if 'first_attempt' in kwargs:
+            payload['first_attempt'] = kwargs['first_attempt']
+        self.delayed_push(datetime, payload)
 
     def delayed_push(self, datetime, item):
         key = int(time.mktime(datetime.timetuple()))
@@ -227,7 +233,7 @@ def delayed_timestamp_size(self, timestamp):
         return self.redis.llen("resque:delayed:%s" % timestamp)
 
     def next_delayed_timestamp(self):
-        key = int(time.mktime(datetime.datetime.now().timetuple()))
+        key = int(time.mktime(ResQ._current_time().timetuple()))
         array = self.redis.zrangebyscore('resque:delayed_queue_schedule', '-inf', key)
         timestamp = None
         if array:
@@ -265,6 +271,10 @@ def _enqueue(cls, klass, *args):
             class_name = '%s.%s' % (klass.__module__, klass.__name__)
             _self.push(queue, {'class':class_name,'args':args})
 
+    @staticmethod
+    def _current_time():
+        return datetime.datetime.now()
+
 
 class Stat(object):
     """A Stat class which shows the current status of the queue.

pyres/job.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timedelta
 from pyres import ResQ, str_to_class, safe_str_to_class
 from pyres import failure
 
@@ -35,10 +36,11 @@ def perform(self):
         payload_class = safe_str_to_class(payload_class_str)
         payload_class.resq = self.resq
         args = self._payload.get("args", None)
-        if args:
+        try:
             return payload_class.perform(*args)
-        else:
-            return payload_class.perform()
+        except:
+            if not self.retry(payload_class, args):
+                raise
 
     def fail(self, exception):
         """This method provides a way to fail a job and will use whatever failure backend
@@ -48,7 +50,22 @@ def fail(self, exception):
         fail = failure.create(exception, self._queue, self._payload, self._worker)
         fail.save(self.resq)
         return fail
-    
+
+    def retry(self, payload_class, args):
+        retry_every = getattr(payload_class, 'retry_every', None)
+        retry_timeout = getattr(payload_class, 'retry_timeout', 0)
+
+        if retry_every:
+            now = ResQ._current_time()
+            first_attempt = self._payload.get("first_attempt", now)
+            retry_until = first_attempt + timedelta(seconds=retry_timeout)
+            retry_at = now + timedelta(seconds=retry_every)
+            if retry_at < retry_until:
+                self.resq.enqueue_at(ResQ._current_time(), payload_class, *args,
+                        **{'first_attempt':first_attempt})
+                return True
+        return False
+
     @classmethod
     def reserve(cls, queue, res, worker=None):
         """Reserve a job on the queue. This marks this job so that other workers

pyres/scheduler.py

@@ -67,12 +67,15 @@ def handle_delayed_items(self):
                 klass = item['class']
                 queue = item['queue']
                 args = item['args']
-                self.resq.enqueue_from_string(klass, queue, *args)
+                kwargs = {}
+                if 'first_attempt' in item:
+                    kwargs['first_attempt'] = item['first_attempt']
+                self.resq.enqueue_from_string(klass, queue, *args, **kwargs)
 
 
     @classmethod
     def run(cls, server, password=None):
         sched = cls(server=server, password=password)
         sched()
 
-    
+    

pyres/worker.py

@@ -154,7 +154,7 @@ def process(self, job=None):
             job = self.reserve()
         try:
             self.working_on(job)
-            job.perform()
+            return job.perform()
         except Exception, e:
             exceptionType, exceptionValue, exceptionTraceback = sys.exc_info()
             logging.error("%s failed: %s" % (job, e))

tests/__init__.py

@@ -25,6 +25,20 @@ class ReturnAllArgsJob(object):
     def perform(*args):
         return args
 
+
+class RetryOnExceptionJob(object):
+    queue = 'basic'
+    retry_every = 5
+    retry_timeout = 15
+
+    @staticmethod
+    def perform(fail_until):
+        if ResQ._current_time() < fail_until:
+            raise Exception("Don't blame me!  I'm supposed to fail!")
+        else:
+            return True
+
+
 class TestProcess(object):
     queue = 'high'
 

tests/test_worker.py

@@ -1,9 +1,13 @@
-from tests import PyResTests, Basic, TestProcess, ErrorObject
+from tests import PyResTests, Basic, TestProcess, ErrorObject, RetryOnExceptionJob
 from pyres import ResQ
 from pyres.job import Job
+from pyres.scheduler import Scheduler
 from pyres.worker import Worker
 import os
 import time
+import datetime
+
+
 class WorkerTests(PyResTests):
     def test_worker_init(self):
         from pyres.exceptions import NoQueueError
@@ -173,3 +177,54 @@ def test_prune_dead_workers(self):
         # the assertion below should hold, because the workers we registered above are on a
         # different host, and thus should not be pruned by this process
         assert self.redis.scard('resque:workers') == 3
+
+    def test_retry_on_exception(self):
+        now = datetime.datetime.now()
+        self.set_current_time(now)
+        worker = Worker(['basic'])
+        scheduler = Scheduler()
+
+        # queue up a job that will fail for 30 seconds
+        self.resq.enqueue(RetryOnExceptionJob,
+                now + datetime.timedelta(seconds=30))
+        worker.process()
+        assert worker.get_failed() == 0
+
+        # check it retries the first time
+        self.set_current_time(now + datetime.timedelta(seconds=5))
+        scheduler.handle_delayed_items()
+        assert None == worker.process()
+        assert worker.get_failed() == 0
+
+        # check it runs fine when it's stopped crashing
+        self.set_current_time(now + datetime.timedelta(seconds=60))
+        scheduler.handle_delayed_items()
+        assert True == worker.process()
+        assert worker.get_failed() == 0
+
+    def test_retries_give_up_eventually(self):
+        now = datetime.datetime.now()
+        self.set_current_time(now)
+        worker = Worker(['basic'])
+        scheduler = Scheduler()
+
+        # queue up a job that will fail for 60 seconds
+        self.resq.enqueue(RetryOnExceptionJob,
+                now + datetime.timedelta(seconds=60))
+        worker.process()
+        assert worker.get_failed() == 0
+
+        # check it retries the first time
+        self.set_current_time(now + datetime.timedelta(seconds=5))
+        scheduler.handle_delayed_items()
+        assert None == worker.process()
+        assert worker.get_failed() == 0
+
+        # check it fails when we've been trying too long
+        self.set_current_time(now + datetime.timedelta(seconds=20))
+        scheduler.handle_delayed_items()
+        assert None == worker.process()
+        assert worker.get_failed() == 1
+
+    def set_current_time(self, time):
+        ResQ._current_time = staticmethod(lambda: time)