Add requeue automatically of previous enqueued jobs

Author: feketemihai

queue_job/jobrunner/runner.py

@@ -139,7 +139,7 @@
 from odoo.tools import config
 
 from . import queue_job_config
-from .channels import ENQUEUED, NOT_DONE, ChannelManager
+from .channels import ENQUEUED, NOT_DONE, PENDING, ChannelManager
 
 SELECT_TIMEOUT = 60
 ERROR_RECOVERY_DELAY = 5
@@ -296,6 +296,13 @@ def keep_alive(self):
         with closing(self.conn.cursor()) as cr:
             cr.execute(query)
 
+    def set_job_pending(self, uuid):
+        with closing(self.conn.cursor()) as cr:
+            cr.execute(
+                "UPDATE queue_job SET state=%s, " "date_enqueued=NULL " "WHERE uuid=%s",
+                (PENDING, uuid),
+            )
+
     def set_job_enqueued(self, uuid):
         with closing(self.conn.cursor()) as cr:
             cr.execute(
@@ -358,6 +365,19 @@ def _query_requeue_dead_jobs(self):
                         )
                     FOR UPDATE SKIP LOCKED
                 )
+                OR
+                id in (
+                    SELECT
+                        id
+                    FROM
+                        queue_job
+                    WHERE
+                        state = 'started' AND NOT EXISTS (
+                            SELECT 1 FROM queue_job_lock
+                            WHERE queue_job_id = queue_job.id
+                        )
+                        FOR UPDATE SKIP LOCKED
+                )
             RETURNING uuid
             """
 
@@ -467,6 +487,18 @@ def initialize_databases(self):
                 self.db_by_name[db_name] = db
                 with db.select_jobs("state in %s", (NOT_DONE,)) as cr:
                     for job_data in cr:
+                        # In case we have enqueued jobs we move them to pending,
+                        # otherwise they remain enqueued and occupy channels slots.
+                        if job_data[6] == "enqueued":
+                            try:
+                                self.db_by_name[db_name].set_job_pending(job_data[1])
+                                job_data = (*job_data[:6], "pending")
+                            except Exception:
+                                _logger.warning(
+                                    "error setting job %s to pending",
+                                    job_data[1],
+                                    exc_info=True,
+                                )
                         self.channel_manager.notify(db_name, *job_data)
                 _logger.info("queue job runner ready for db %s", db_name)
 

queue_job/tests/test_requeue_dead_job.py

@@ -131,3 +131,38 @@ def test_requeue_dead_jobs(self):
         # because we committed the cursor, the savepoint of the test method is
         # gone, and this would break TransactionCase cleanups
         self.cr.execute("SAVEPOINT test_%d" % self._savepoint_id)
+
+    def test_requeue_dead_jobs_started_before_patch(self):
+        uuid = "test_requeue_dead_jobs_before_locking"
+
+        queue_job = self.create_dummy_job(uuid)
+        job_obj = Job.load(self.env, queue_job.uuid)
+
+        job_obj.set_enqueued()
+        # simulate enqueuing was in the past
+        job_obj.date_enqueued = datetime.now() - timedelta(minutes=1)
+        job_obj.set_started()
+        # Delete the job lock to simulate job started before new implementation
+        self.env.cr.execute(
+            "DELETE FROM queue_job_lock WHERE queue_job_id=%s", (queue_job.id,)
+        )
+
+        job_obj.store()
+        self.env.cr.commit()  # pylint: disable=E8102
+
+        # requeue dead jobs using current cursor
+        query = Database(self.env.cr.dbname)._query_requeue_dead_jobs()
+        self.env.cr.execute(query)
+
+        uuids_requeued = self.env.cr.fetchall()
+
+        self.assertEqual(len(uuids_requeued), 1)
+        self.assertEqual(uuids_requeued[0][0], uuid)
+
+        # clean up
+        queue_job.unlink()
+        self.env.cr.commit()  # pylint: disable=E8102
+
+        # because we committed the cursor, the savepoint of the test method is
+        # gone, and this would break TransactionCase cleanups
+        self.cr.execute("SAVEPOINT test_%d" % self._savepoint_id)