diff --git a/rq/job.py b/rq/job.py index 61bc9aa..5094b2d 100644 --- a/rq/job.py +++ b/rq/job.py @@ -912,7 +912,10 @@ class Job: self.allow_dependency_failures = bool(int(allow_failures)) if allow_failures else None self.enqueue_at_front = bool(int(obj['enqueue_at_front'])) if 'enqueue_at_front' in obj else None self.ttl = int(obj.get('ttl')) if obj.get('ttl') else None - self.meta = self.serializer.loads(obj.get('meta')) if obj.get('meta') else {} + try: + self.meta = self.serializer.loads(obj.get('meta')) if obj.get('meta') else {} + except Exception: # depends on the serializer + self.meta = {'unserialized': obj.get('meta', {})} self.retries_left = int(obj.get('retries_left')) if obj.get('retries_left') else None if obj.get('retry_intervals'): diff --git a/tests/test_worker.py b/tests/test_worker.py index d9b6121..5410180 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -225,6 +225,23 @@ class TestWorker(RQTestCase): failed_job_registry = FailedJobRegistry(queue=q) self.assertTrue(job in failed_job_registry) + def test_meta_is_unserializable(self): + """Unserializable jobs are put on the failed job registry.""" + q = Queue() + self.assertEqual(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.create(func=do_nothing, origin=q.name, meta={'key': 'value'}) + job.save() + + invalid_meta = '{{{{{{{{INVALID_JSON' + self.testconn.hset(job.key, 'meta', invalid_meta) + job.refresh() + self.assertIsInstance(job.meta, dict) + self.assertTrue('unserialized' in job.meta.keys()) + @mock.patch('rq.worker.logger.error') def test_deserializing_failure_is_handled(self, mock_logger_error): """