@@ -216,6 +216,8 @@ def __init__(self, id: str | None = None, connection: Redis | None = None, seria
         self._status: JobStatus = JobStatus.CREATED
         self._dependency_ids: list[str] = []
         self.meta: dict[str, Any] = {}
+        # Stored meta that could not be deserialized here, kept so a save writes it back unchanged.
+        self._unreadable_meta: bytes | None = None
         self.serializer = resolve_serializer(serializer)
         # Tracks remaining retries for exception-based retry. Set to retry.max when a job
         # is enqueued with retry=Retry(...) via queue.enqueue(). Decremented each time
@@ -1039,10 +1041,12 @@ def restore(self, raw_data) -> Any:
         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['ttl']) if obj.get('ttl') else None
+        self._unreadable_meta = None
         try:
             self.meta = self.serializer.loads(obj['meta']) if obj.get('meta') else {}
         except Exception:  # depends on the serializer
             self.meta = {'unserialized': obj.get('meta', {})}
+            self._unreadable_meta = obj['meta']
 
         self.number_of_retries = int(obj['number_of_retries']) if obj.get('number_of_retries') else None
         self.retries_left = int(obj['retries_left']) if obj.get('retries_left') else None
@@ -1158,7 +1162,7 @@ def to_dict(self, include_meta: bool = True, include_result: bool = True) -> dic
             obj['dependency_id'] = self._dependency_ids[0]  # for backwards compatibility
             obj['dependency_ids'] = json.dumps(self._dependency_ids)
         if self.meta and include_meta:
-            obj['meta'] = self.serializer.dumps(self.meta)
+            obj['meta'] = self._serialize_meta()
         if self.ttl:
             obj['ttl'] = self.ttl
 
@@ -1195,9 +1199,17 @@ def save(self, pipeline: Pipeline | None = None, include_meta: bool = True, incl
 
     def save_meta(self):
         """Stores job meta from the job instance to the corresponding Redis key."""
-        meta = self.serializer.dumps(self.meta)
+        meta = self._serialize_meta()
         self.connection.hset(self.key, 'meta', meta)
 
+    def _serialize_meta(self) -> bytes:
+        # If this process could not deserialize the stored meta and it is still the
+        # placeholder holding those bytes, write the bytes back instead of the placeholder.
+        is_placeholder = len(self.meta) == 1 and self.meta.get('unserialized') is self._unreadable_meta
+        if is_placeholder and self._unreadable_meta is not None:
+            return self._unreadable_meta
+        return self.serializer.dumps(self.meta)
+
     def cancel(
         self,
         pipeline: Pipeline | None = None,
 
@@ -471,6 +471,31 @@ def test_custom_meta_is_rewriten_by_save_meta(self):
         serialized2.pop('meta')
         self.assertDictEqual(serialized, serialized2)
 
+    def test_unreadable_meta(self):
+        """Meta this process could not deserialize is written back by save() and save_meta() until it is changed."""
+        job = Job.create(func=fixtures.say_hello, args=('Lionel',), connection=self.connection)
+        job.save()
+        # Pickled meta naming a module this process cannot import (e.g. a type only the enqueuing app has)
+        unreadable_meta = b'cnot_an_importable_module\nOwner\n.'
+        self.connection.hset(job.key, 'meta', unreadable_meta)
+        job.refresh()
+        self.assertEqual(job.meta, {'unserialized': unreadable_meta})
+
+        job.save()
+        self.assertEqual(self.connection.hget(job.key, 'meta'), unreadable_meta)
+
+        job.save_meta()
+        self.assertEqual(self.connection.hget(job.key, 'meta'), unreadable_meta)
+
+        # Changed meta is saved as usual
+        job.meta['foo'] = 'bar'
+        job.save_meta()
+        self.assertEqual(loads(self.connection.hget(job.key, 'meta')), {'unserialized': unreadable_meta, 'foo': 'bar'})
+
+        job.meta = {'foo': 'baz'}
+        job.save()
+        self.assertEqual(loads(self.connection.hget(job.key, 'meta')), {'foo': 'baz'})
+
     def test_unpickleable_result(self):
         """Unpickleable job result doesn't crash job.save() and job.refresh()"""
         job = Job.create(func=fixtures.say_hello, args=('Lionel',), connection=self.connection)
 
@@ -332,6 +332,22 @@ def test_meta_is_unserializable(self):
         self.assertIsInstance(job.meta, dict)
         self.assertIn('unserialized', job.meta.keys())
 
+    def test_retry_keeps_unreadable_meta(self):
+        """A worker that cannot deserialize meta leaves it unchanged when it retries the job."""
+        q = Queue(connection=self.connection)
+        job = q.enqueue(div_by_zero, 1, retry=Retry(max=1), meta={'key': 'value'})
+        # Pickled meta naming a module the worker cannot import
+        unreadable_meta = b'cnot_an_importable_module\nOwner\n.'
+        self.connection.hset(job.key, 'meta', unreadable_meta)
+
+        w = Worker([q], connection=self.connection)
+        w.work(burst=True)
+
+        job.refresh()
+        self.assertEqual(job.retries_left, 0)
+        self.assertIn(job, FailedJobRegistry(queue=q))
+        self.assertEqual(self.connection.hget(job.key, 'meta'), unreadable_meta)
+
     @mock.patch('rq.worker.logger.error')
     def test_deserializing_failure_is_handled(self, mock_logger_error):
         """