[19.0][FIX] queue_job: wake the job runner up on stop

queue_job/jobrunner/__init__.py

@@ -4,7 +4,6 @@
 
 import logging
 from threading import Thread
-import time
 from configparser import ConfigParser
 
 from odoo.service import server
@@ -50,7 +49,7 @@ def __init__(self):
 
     def run(self):
         # sleep a bit to let the workers start at ease
-        time.sleep(START_DELAY)
+        self.runner.sleep(START_DELAY)
         self.runner.run()
 
     def stop(self):
@@ -80,7 +79,7 @@ def process_work(self):
             self.runner = QueueJobRunner.from_environ_or_config()
             self._recover = False
         _logger.debug("WorkerJobRunner (%s) starting up", self.pid)
-        time.sleep(START_DELAY)
+        self.runner.sleep(START_DELAY)
         self.runner.run()
 
     def signal_time_expired_handler(self, n, stack):

queue_job/jobrunner/runner.py

@@ -335,6 +335,7 @@ def __init__(
         self.channel_manager.simple_configure(channel_config_string)
         self.db_by_name = {}
         self._stop = False
+        self._stop_event = threading.Event()
         self._stop_pipe = os.pipe()
 
     def __del__(self):
@@ -486,9 +487,18 @@ def wait_notification(self):
                             continue
                         key.fileobj.poll()
 
+    def sleep(self, seconds):
+        """Wait for ``seconds``, or less if :meth:`stop` is called meanwhile.
+
+        A plain ``time.sleep`` would hold the thread, and so the server
+        shutdown that joins it, for up to the whole delay.
+        """
+        self._stop_event.wait(seconds)
+
     def stop(self):
         _logger.info("graceful stop requested")
         self._stop = True
+        self._stop_event.set()
         # wakeup the select() in wait_notification
         os.write(self._stop_pipe[1], b".")
 
@@ -520,12 +530,12 @@ def run(self):
                     ERROR_RECOVERY_DELAY,
                 )
                 self.close_databases()
-                time.sleep(ERROR_RECOVERY_DELAY)
+                self.sleep(ERROR_RECOVERY_DELAY)
             except Exception:
                 _logger.exception(
                     "exception: sleeping %ds and retrying", ERROR_RECOVERY_DELAY
                 )
                 self.close_databases()
-                time.sleep(ERROR_RECOVERY_DELAY)
+                self.sleep(ERROR_RECOVERY_DELAY)
         self.close_databases(remove_jobs=False)
         _logger.info("stopped")

queue_job/tests/test_runner_runner.py

@@ -4,10 +4,12 @@
 # pylint: disable=odoo-addons-relative-import
 # we are testing, we want to test as we were an external consumer of the API
 import os
+import threading
+from unittest.mock import patch
 
 from odoo.tests import BaseCase, tagged
 
-from odoo.addons.queue_job.jobrunner import runner
+from odoo.addons.queue_job.jobrunner import START_DELAY, QueueJobRunnerThread, runner
 
 from .common import load_doctests
 
@@ -57,3 +59,35 @@ def test_runner_file_closed_write_descriptor(self):
 
         self.assertFalse(self._is_open_file_descriptor(read_fd))
         self.assertFalse(self._is_open_file_descriptor(write_fd))
+
+    def test_runner_stop_interrupts_the_recovery_delay(self):
+        """A runner that lost the master election stops without sleeping it out.
+
+        The server joins the runner thread on shutdown, so a stop waiting for
+        ERROR_RECOVERY_DELAY held the whole process for up to 5 seconds.
+        """
+        a_runner = runner.QueueJobRunner.from_environ_or_config()
+        election_lost = threading.Event()
+
+        def initialize_databases():
+            election_lost.set()
+            raise runner.MasterElectionLost("held by another runner")
+
+        with patch.object(
+            a_runner, "initialize_databases", side_effect=initialize_databases
+        ):
+            thread = threading.Thread(target=a_runner.run, daemon=True)
+            thread.start()
+            self.addCleanup(thread.join)
+            self.assertTrue(election_lost.wait(timeout=5))
+            a_runner.stop()
+            thread.join(timeout=runner.ERROR_RECOVERY_DELAY / 2)
+            self.assertFalse(thread.is_alive())
+
+    def test_runner_thread_stop_interrupts_the_start_delay(self):
+        thread = QueueJobRunnerThread()
+        thread.start()
+        self.addCleanup(thread.join)
+        thread.stop()
+        thread.join(timeout=START_DELAY / 2)
+        self.assertFalse(thread.is_alive())