Don't enqueue waitlisted jobs on failed execution.

rq/queue.py

@@ -123,6 +123,7 @@ class Queue(object):
         contain options for RQ itself.
         """
         timeout = timeout or self._default_timeout
+        # TODO: job with dependency shouldn't have "queued" as status
         job = Job.create(func, args, kwargs, connection=self.connection,
                          result_ttl=result_ttl, status=Status.QUEUED,
                          parent=after)

rq/worker.py

@@ -328,7 +328,8 @@ class Worker(object):
                 self.connection.expire(self.key, (job.timeout or 180) + 60)
                 self.fork_and_perform_job(job)
                 self.connection.expire(self.key, self.default_worker_ttl)
-                queue.enqueue_waitlist(job)
+                if job.status == 'finished':
+                    queue.enqueue_waitlist(job)
 
                 did_perform_work = True
         finally:

tests/test_worker.py

@@ -236,6 +236,7 @@ class TestWorker(RQTestCase):
         self.assertEqual(job.is_failed, True)
 
     def test_job_dependency(self):
+        """Waitlisted jobs are enqueued only if their parents don't fail"""
         q = Queue()
         w = Worker([q])
         parent_job = q.enqueue(say_hello)
@@ -243,3 +244,9 @@ class TestWorker(RQTestCase):
         w.work(burst=True)
         job = Job.fetch(job.id)
         self.assertEqual(job.status, 'finished')
+
+        parent_job = q.enqueue(div_by_zero)
+        job = q.enqueue_call(say_hello, after=parent_job)
+        w.work(burst=True)
+        job = Job.fetch(job.id)
+        self.assertNotEqual(job.status, 'finished')