diff --git a/base_bg/models/bg_job.py b/base_bg/models/bg_job.py index eb54521f..a0900427 100644 --- a/base_bg/models/bg_job.py +++ b/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 += "
" + _("Related records: %s") % (", ".join(records)) + links = [record._get_html_link() for record in records] + message += "
" + _("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) diff --git a/base_bg/tests/test_bg_job.py b/base_bg/tests/test_bg_job.py index a210942a..f27e1ba7 100644 --- a/base_bg/tests/test_bg_job.py +++ b/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