Put unreadable tasks on the failure queue.

rq/exceptions.py

@@ -2,5 +2,7 @@ class NoQueueError(Exception):
     pass
 
 class UnpickleError(Exception):
-    pass
+    def __init__(self, message, raw_data):
+        super(UnpickleError, self).__init__(message)
+        self.raw_data = raw_data
 

rq/job.py

@@ -15,7 +15,7 @@ class Job(object):
             assert isinstance(unpickled_obj, Job)
             return unpickled_obj
         except (AssertionError, AttributeError, IndexError, TypeError):
-            raise UnpickleError('Could not unpickle Job.')
+            raise UnpickleError('Could not unpickle Job.', pickle_data)
 
     def __init__(self, func, *args, **kwargs):
         self._id = unicode(uuid4())

rq/queue.py

@@ -3,6 +3,7 @@ from functools import total_ordering
 from pickle import loads
 from .proxy import conn
 from .job import Job
+from .exceptions import UnpickleError
 
 
 class DelayedResult(object):
@@ -103,6 +104,10 @@ class Queue(object):
         job.origin = self.name
         return job
 
+    def _push(self, pickled_job):
+        """Enqueues a pickled_job on the corresponding Redis queue."""
+        conn.rpush(self.key, pickled_job)
+
     def enqueue(self, f, *args, **kwargs):
         """Enqueues a function call for delayed execution.
 
@@ -111,7 +116,7 @@ class Queue(object):
         """
         job = self._create_job(f, *args, **kwargs)
         job.enqueued_at = datetime.utcnow()
-        conn.rpush(self.key, job.pickle())
+        self._push(job.pickle())
         return DelayedResult(job.rv_key)
 
     def requeue(self, job):
@@ -126,7 +131,13 @@ class Queue(object):
         blob = conn.lpop(self.key)
         if blob is None:
             return None
+        try:
             job = Job.unpickle(blob)
+        except UnpickleError as e:
+            # Attach queue information on the exception for improved error
+            # reporting
+            e.queue = self
+            raise e
         job.origin = self
         return job
 
@@ -162,8 +173,14 @@ class Queue(object):
                 return None
             queue_key, blob = redis_result
 
-        job = Job.unpickle(blob)
         queue = Queue.from_queue_key(queue_key)
+        try:
+            job = Job.unpickle(blob)
+        except UnpickleError as e:
+            # Attach queue information on the exception for improved error
+            # reporting
+            e.queue = queue
+            raise e
         job.origin = queue
         return job
 

rq/worker.py

@@ -14,7 +14,7 @@ except ImportError:
     from logging import Logger
 from .queue import Queue
 from .proxy import conn
-from .exceptions import NoQueueError
+from .exceptions import NoQueueError, UnpickleError
 
 def iterable(x):
     return hasattr(x, '__iter__')
@@ -34,6 +34,7 @@ def signal_name(signum):
     except KeyError:
         return 'SIG_UNKNOWN'
 
+
 class Worker(object):
     redis_worker_namespace_prefix = 'rq:worker:'
     redis_workers_keys = 'rq:workers'
@@ -249,7 +250,18 @@ class Worker(object):
                 self.procline('Listening on %s' % (','.join(qnames)))
                 self.log.info('*** Listening for work on %s...' % (', '.join(qnames)))
                 wait_for_job = not burst
+                try:
                     job = Queue.dequeue_any(self.queues, wait_for_job)
+                except UnpickleError as e:
+                    self.log.warning('*** Ignoring unpickleable data on %s.', e.queue)
+                    self.log.debug('Data follows:')
+                    self.log.debug(e.raw_data)
+                    self.log.debug('End of unreadable data.')
+
+                    q = Queue('failure')
+                    q._push(e.raw_data)
+                    continue
+
                 if job is None:
                     break
                 self.state = 'busy'

tests/__init__.py

@@ -9,6 +9,10 @@ def testjob(name=None):
         name = 'Stranger'
     return 'Hi there, %s!' % (name,)
 
+def failing_job(x):
+    # Will throw a division-by-zero error
+    return x / 0
+
 
 class RQTestCase(unittest.TestCase):
     """Base class to inherit test cases from for RQ.

tests/test_queue.py

@@ -48,6 +48,7 @@ class TestQueue(RQTestCase):
         self.assertEquals(q.empty, False)
         self.assertQueueContains(q, testjob)
 
+
     def test_dequeue(self):
         """Fetching work from specific queue."""
         q = Queue('foo')
@@ -60,7 +61,6 @@ class TestQueue(RQTestCase):
         self.assertEquals(job.args[0], 'Rick')
         self.assertEquals(job.kwargs['foo'], 'bar')
 
-
     def test_dequeue_any(self):
         """Fetching work from any given queue."""
         fooq = Queue('foo')
@@ -87,7 +87,6 @@ class TestQueue(RQTestCase):
         self.assertEquals(job.origin, barq)
         self.assertEquals(job.args[0], 'for Bar', 'Bar should be dequeued second.')
 
-
     def test_dequeue_unpicklable_data(self):
         """Error handling of invalid pickle data."""
 

tests/test_worker.py

@@ -1,6 +1,7 @@
 from tests import RQTestCase
-from tests import testjob
+from tests import testjob, failing_job
 from rq import Queue, Worker
+from rq.job import Job
 
 
 class TestWorker(RQTestCase):
@@ -19,4 +20,33 @@ class TestWorker(RQTestCase):
         fooq.enqueue(testjob, name='Frank')
         self.assertEquals(w.work(burst=True), True, 'Expected at least some work done.')
 
+    def test_work_is_unreadable(self):
+        """Worker processes unreadable job."""
+        q = Queue()
+        failure_q = Queue('failure')
+
+        self.assertEquals(failure_q.count, 0)
+        self.assertEquals(q.count, 0)
+
+        # NOTE: We have to fake this enqueueing for this test case.
+        # What we're simulating here is a call to a function that is not
+        # importable from the worker process.
+        job = Job(failing_job, 3)
+        pickled_job = job.pickle()
+        invalid_data = pickled_job.replace(
+                'failing_job', 'nonexisting_job')
+
+        # We use the low-level internal function to enqueue any data (bypassing
+        # validity checks)
+        q._push(invalid_data)
+
+        self.assertEquals(q.count, 1)
+
+        # All set, we're going to process it
+        w = Worker([q])
+        w.work(burst=True)   # should silently pass
+        self.assertEquals(q.count, 0)
+
+        self.assertEquals(failure_q.count, 1)
+