[IMP] queue_job_cron_jobrunner: requeue lost started jobs

Author: Koudja

queue_job_cron_jobrunner/models/queue_job.py

@@ -7,6 +7,7 @@
 from datetime import datetime
 from io import StringIO
 
+import psutil
 from psycopg2 import OperationalError
 
 from odoo import _, api, fields, models, tools
@@ -161,6 +162,7 @@ def _process(self, commit=False):
     @api.model
     def _job_runner(self, commit=True):
         """Short-lived job runner, triggered by async crons"""
+        self._release_started_jobs(commit=commit)
         job = self._acquire_one_job(commit=commit)
         while job:
             job._process(commit=commit)
@@ -214,6 +216,24 @@ def _ensure_cron_trigger(self):
         if delayed_etas:
             self._cron_trigger(at=list(delayed_etas))
 
+    @api.model
+    def _release_started_jobs(self, commit=False):
+        pids = [x.pid for x in psutil.process_iter()]
+        for record in self.search(
+            [("state", "=", "started"), ("worker_pid", "not in", pids)]
+        ):
+            job = Job._load_from_db_record(record)
+            job.set_pending()
+            job.store()
+            _logger.info(
+                "release started job %s[channel=%s,uuid=%s]",
+                record.id,
+                record.channel,
+                record.uuid,
+            )
+        if commit:  # pragma: no cover
+            self.env.cr.commit()  # pylint: disable=invalid-commit
+
     @api.model_create_multi
     def create(self, vals_list):
         # When jobs are created, also create the cron trigger

queue_job_cron_jobrunner/tests/test_queue_job.py

@@ -174,3 +174,20 @@ def test_queue_job_creation_create_change_next_call(self):
         self.cron.nextcall = datetime(2021, 1, 21, 21, 21, 21)
         self.env["res.partner"].with_delay().create({"name": "test"})
         self.assertNotEqual(self.cron.nextcall, datetime(2022, 2, 22, 22, 22, 22))
+
+    def test_release_started_jobs(self):
+        job_known_pid = self.env["res.partner"].with_delay().create({"name": "test"})
+        job_known_pid.set_started()
+        job_known_pid.store()
+        known_pid = job_known_pid.db_record().worker_pid
+        job_unknown_pid = self.env["res.partner"].with_delay().create({"name": "test"})
+        job_unknown_pid.set_started()
+        job_unknown_pid.store()
+        job_unknown_pid.db_record().worker_pid = -1
+
+        self.env["queue.job"]._release_started_jobs(commit=False)
+
+        self.assertEqual(job_unknown_pid.db_record().state, "pending")
+        self.assertEqual(job_unknown_pid.db_record().worker_pid, 0)
+        self.assertEqual(job_known_pid.db_record().state, "started")
+        self.assertEqual(job_known_pid.db_record().worker_pid, known_pid)