From 943d92a5eea56d8de95e00e22a2d7557509e36b5 Mon Sep 17 00:00:00 2001 From: Manh Nguyen Date: Fri, 30 Oct 2026 18:09:34 +0700 Subject: [PATCH] [FIX] queue_job: wake the job runner up on stop A runner that lost the master election (or hit an error) waited ERROR_RECOVERY_DELAY with time.sleep, and the runner thread and worker waited START_DELAY the same way before starting. stop() could not interrupt them, and the server joins the runner thread on shutdown, so stopping the server waited up to 5 seconds. Both delays now wait on an event that stop() sets. Port of #1011 (19.0). --- queue_job/jobrunner/__init__.py | 5 ++-- queue_job/jobrunner/runner.py | 14 +++++++++-- queue_job/tests/test_runner_runner.py | 35 ++++++++++++++++++++++++++- 3 files changed, 48 insertions(+), 6 deletions(-) diff --git a/queue_job/jobrunner/__init__.py b/queue_job/jobrunner/__init__.py index 66575184c8..2e824a20bd 100644 --- a/queue_job/jobrunner/__init__.py +++ b/queue_job/jobrunner/__init__.py @@ -4,7 +4,6 @@ import logging from threading import Thread -import time from odoo.service import server from odoo.tools import config @@ -40,7 +39,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): @@ -70,7 +69,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): diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 5cbd52ef48..8428b9d88d 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -498,6 +498,7 @@ def __init__( self.db_by_name = {} self._stop = False + self._stop_event = threading.Event() self._stop_pipe = os.pipe() def __del__(self): @@ -811,9 +812,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".") @@ -845,12 +855,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") diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 93b5674237..fe1663ebdc 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -2,12 +2,13 @@ # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) import doctest import os +import threading from unittest import mock from odoo.tests import BaseCase, tagged from odoo.tools import mute_logger -from odoo.addons.queue_job.jobrunner import runner +from odoo.addons.queue_job.jobrunner import START_DELAY, QueueJobRunnerThread, runner # pylint: disable=odoo-addons-relative-import # we are testing, we want to test as we were an external consumer of the API @@ -112,6 +113,38 @@ 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 mock.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()) + def test_max_capacity_from_env(self): with ( mock.patch.dict(