diff --git a/rq/job.py b/rq/job.py
index 84fa3f87..a8c79378 100644
--- a/rq/job.py
+++ b/rq/job.py
@@ -87,11 +87,16 @@ class Job(object):
     def func_name(self):
         return self._func_name
 
-    @property
-    def status(self):
+    def _get_status(self):
         self._status = self.connection.hget(self.key, 'status')
         return self._status
 
+    def _set_status(self, status):
+        self._status = status
+        self.connection.hset(self.key, 'status', self._status)
+
+    status = property(_get_status, _set_status)
+
     @property
     def is_finished(self):
         return self.status == Status.FINISHED
diff --git a/rq/worker.py b/rq/worker.py
index 23786142..4d5c0b4e 100644
--- a/rq/worker.py
+++ b/rq/worker.py
@@ -395,7 +395,8 @@ class Worker(object):
             pickled_rv = dumps(rv)
             job._status = Status.FINISHED
         except:
-            job._status = Status.FAILED
+            # Use the public setter here, to immediately update Redis
+            job.status = Status.FAILED
             self.handle_exception(job, *sys.exc_info())
             return False
 
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 67cf6e3d..e3e706e4 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -96,6 +96,34 @@ class TestWorker(RQTestCase):
         self.assertEquals(job.enqueued_at, enqueued_at_date)
         self.assertIsNotNone(job.exc_info)  # should contain exc_info
 
+    def test_custom_exc_handling(self):
+        """Custom exception handling."""
+        def black_hole(job, *exc_info):
+            # Don't fall through to default behaviour (moving to failed queue)
+            return False
+
+        q = Queue()
+        failed_q = get_failed_queue()
+
+        # Preconditions
+        self.assertEquals(failed_q.count, 0)
+        self.assertEquals(q.count, 0)
+
+        # Action
+        job = q.enqueue(div_by_zero)
+        self.assertEquals(q.count, 1)
+
+        w = Worker([q], exc_handler=black_hole)
+        w.work(burst=True)  # should silently pass
+
+        # Postconditions
+        self.assertEquals(q.count, 0)
+        self.assertEquals(failed_q.count, 0)
+
+        # Check the job
+        job = Job.fetch(job.id)
+        self.assertEquals(job.is_failed, True)
+
     def test_cancelled_jobs_arent_executed(self):  # noqa
         """Cancelling jobs."""