Merge branch 'retry_queues'

Author: Matt George

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(retry_at, 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)