[FIX] base_bg: do not let one job break the whole reaper

base_bg/models/bg_job.py

@@ -272,9 +272,12 @@ def fail(self, error_message: str, notify: bool = True):
         )
         if notify:
             message = _("Job %s failed: %s") % (self.name, error_message)
-            records = self._get_records().mapped(lambda r: r and r._get_html_link())
+            # exists(): a job can outlive its records, and browsing a dropped id is truthy —
+            # _get_html_link() reads display_name on it and would raise MissingError.
+            records = self._get_records().exists()
             if records:
-                message += "<br/>" + _("Related records: %s") % (", ".join(records))
+                links = [record._get_html_link() for record in records]
+                message += "<br/>" + _("Related records: %s") % (", ".join(links))
             self._notify_user(message)
 
     def cancel(self, message: str | None = None):
@@ -554,8 +557,33 @@ def _cron_check_running_jobs(self):
                 ("state", "=", "running"),
             ]
         )
+        timeout_msg = _("Job timed out")
         for job in jobs:
-            job._handle_job_error(_("Job timed out"))
-            if job.state == "failed":
-                message = _("Job %s timed out") % job._get_html_link(title=job.name)
-                job._notify_user(message)
+            job_name = job.name
+            try:
+                # Per-job savepoint: a job that raises would otherwise abort the whole reaper
+                # and leave every other timed-out job running forever.
+                with self.env.cr.savepoint():
+                    job._handle_job_error(timeout_msg)
+                    if job.state == "failed":
+                        job._notify_user(_("Job %s timed out") % job._get_html_link(title=job_name))
+            except Exception as error:
+                # No invalidation needed: the savepoint rollback already cleared the cache
+                # and the pending updates (_FlushingSavepoint.rollback -> cr.clear()).
+                if self._is_transient_error(error):
+                    _logger.warning("Job %s not timed out yet, transient error: %s", job_name, error)
+                    continue
+                _logger.exception("Could not time out job %s, giving up on it", job_name)
+                try:
+                    # Own savepoint: the recovery path writes through the registry too, so a
+                    # model override raising here would abort the whole reaper as well.
+                    with self.env.cr.savepoint():
+                        # Base implementations on purpose: the rollback restored the job to
+                        # running and whatever the model added on top is what just raised.
+                        BgJob.fail(job, timeout_msg, notify=False)
+                        job._get_next_jobs().cancel(message=_("Previous job in batch failed"))
+                except Exception as fallback_error:
+                    if self._is_transient_error(fallback_error):
+                        _logger.warning("Job %s not failed yet, transient error: %s", job_name, fallback_error)
+                    else:
+                        _logger.exception("Could not fail job %s either, leaving it for the next run", job_name)

base_bg/tests/test_bg_job.py

@@ -110,6 +110,58 @@ def test_cron_check_running_jobs(self):
         job = self.BgJob.browse(job.id)
         self.assertEqual(job.state, "failed")
 
+    def test_fail_notifies_when_records_were_deleted(self):
+        """fail() must not raise when the job outlived the records it points to."""
+        partner = self.env["res.partner"].create({"name": "Gone"})
+        job = self._create_job(name="Orphan Job", state="running", kwargs_json={"_record_ids": [partner.id]})
+        partner.unlink()  # linking it in the notification would read display_name and raise
+
+        job.fail("boom")
+
+        self.assertEqual(job.state, "failed")
+        self.assertEqual(job.error_message, "boom")
+
+    def _create_timed_out_job(self, name, **vals):
+        """Build a running job whose start_time is already past any cron timeout."""
+        old_time = fields.Datetime.now() - timedelta(hours=6)
+        return self._create_job(name=name, state="running", start_time=old_time, max_retries=1, **vals)
+
+    def test_cron_check_running_jobs_skips_poisoned_job(self):
+        """A job that raises while being timed out must not abort the reaper for the rest."""
+        poisoned = self._create_timed_out_job("Poisoned Job")
+        chained = self._create_job(name="Chained Job", batch_key=poisoned.batch_key, state="waiting")
+        poisoned.next_job_id = chained.id
+        healthy = self._create_timed_out_job("Healthy Job")
+        base_fail = type(self.BgJob).fail
+
+        def poisoned_fail(job_self, error_message, notify=True):
+            """Stand in for a model override of fail() that raises (the real-world poison)."""
+            if job_self.name == "Poisoned Job":
+                raise ValueError("boom while failing the job")
+            return base_fail(job_self, error_message, notify=notify)
+
+        self._set_cron_timeout(300)
+        with patch.object(type(self.BgJob), "fail", poisoned_fail), tools.mute_logger(
+            "odoo.addons.base_bg.models.bg_job"
+        ):
+            self.BgJob._cron_check_running_jobs()
+
+        self.assertEqual(poisoned.state, "failed", "the poisoned job is bare-failed instead of poisoning every run")
+        self.assertEqual(chained.state, "canceled", "its batch is cancelled, as on any other permanent failure")
+        self.assertEqual(healthy.state, "failed", "the remaining jobs are still timed out")
+
+    def test_cron_check_running_jobs_defers_transient_error(self):
+        """A transient PG error is not a poisoned job: the job is left for the next run."""
+        job = self._create_timed_out_job("Contended Job")
+
+        self._set_cron_timeout(300)
+        with patch.object(
+            type(self.BgJob), "_handle_job_error", side_effect=self._serialization_error()
+        ), tools.mute_logger("odoo.addons.base_bg.models.bg_job"):
+            self.BgJob._cron_check_running_jobs()
+
+        self.assertEqual(job.state, "running")
+
     def test_cron_check_running_jobs_recent(self):
         """Test that recent running jobs are not marked as timed out."""
         # Create a job that started recently