From 539884617a36ea0cc1520e84f1057cf8283e3033 Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Mon, 17 Aug 2026 17:59:36 +0200 Subject: [PATCH 1/6] [ADD] queue_job: add jobrunner config on channels Not used at this point by the jobrunner, but the changes on channels trigger a notify to the jobrunner. --- queue_job/jobrunner/channels.py | 1 + queue_job/models/queue_job_channel.py | 54 +++++++++++++- queue_job/tests/test_model_job_channel.py | 83 +++++++++++++++++++++ queue_job/views/queue_job_channel_views.xml | 13 ++++ 4 files changed, 149 insertions(+), 2 deletions(-) diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index e3cb480420..a40b182ee6 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -10,6 +10,7 @@ from ..exception import ChannelNotFound from ..job import CANCELLED, DONE, ENQUEUED, FAILED, PENDING, STARTED, WAIT_DEPENDENCIES +RELOAD_PAYLOAD = "reload" NOT_DONE = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, STARTED, FAILED) JobSortingKey = namedtuple("SortingKey", "eta priority date_created seq") diff --git a/queue_job/models/queue_job_channel.py b/queue_job/models/queue_job_channel.py index 4aabb0188c..47e5bd62a8 100644 --- a/queue_job/models/queue_job_channel.py +++ b/queue_job/models/queue_job_channel.py @@ -4,12 +4,19 @@ from odoo import _, api, exceptions, fields, models +from ..jobrunner.channels import RELOAD_PAYLOAD + class QueueJobChannel(models.Model): _name = "queue.job.channel" _description = "Job Channels" _rec_name = "complete_name" + # fields that trigger a reload of the jobrunner for this database when changed + _JOBRUNNER_CONFIG_FIELDS = frozenset( + ("capacity", "sequential", "throttle", "paused", "name", "parent_id") + ) + name = fields.Char() complete_name = fields.Char( compute="_compute_complete_name", store=True, readonly=True, recursive=True @@ -25,11 +32,44 @@ class QueueJobChannel(models.Model): removal_interval = fields.Integer( default=lambda self: self.env["queue.job"]._removal_interval, required=True ) + capacity = fields.Integer( + help="Maximum number of jobs running at the same time in this channel. " + "0 means no limit, but they are still limited by the capacity of the parent " + "channel. On the root channel, 0 is limited by the global server-side " + "configuration." + ) + sequential = fields.Boolean( + help="Jobs are executed one after the other and failed jobs block the channel. " + "Requires a capacity of 1." + ) + throttle = fields.Integer( + help="Minimum delay in seconds between the start of two jobs in this channel." + ) + paused = fields.Boolean( + help="A paused channel (an its sub-channels) do not execute any jobs until " + "resumed." + ) _sql_constraints = [ ("name_uniq", "unique(complete_name)", "Channel complete name must be unique") ] + @api.constrains("capacity", "sequential", "throttle") + def _check_jobrunner_configuration(self): + for record in self: + if record.capacity < 0: + raise exceptions.ValidationError( + self.env._("The capacity of a channel cannot be negative.") + ) + if record.throttle < 0: + raise exceptions.ValidationError( + self.env._("The throttle of a channel cannot be negative.") + ) + if record.sequential and record.capacity != 1: + raise exceptions.ValidationError( + self.env._("A sequential channel must have a capacity of 1.") + ) + @api.depends("name", "parent_id.complete_name") def _compute_complete_name(self): for record in self: @@ -70,6 +110,7 @@ def create(self, vals_list): new_vals_list.append(vals) vals_list = new_vals_list records |= super().create(vals_list) + records._notify_channel_config_changed() return records def write(self, values): @@ -80,10 +121,19 @@ def write(self, values): and ("name" in values or "parent_id" in values) ): raise exceptions.UserError(_("Cannot change the root channel")) - return super().write(values) + res = super().write(values) + if self._JOBRUNNER_CONFIG_FIELDS.intersection(values): + self._notify_channel_config_changed() + return res def unlink(self): for channel in self: if channel.name == "root": raise exceptions.UserError(_("Cannot remove the root channel")) - return super().unlink() + res = super().unlink() + self._notify_channel_config_changed() + return res + + def _notify_channel_config_changed(self): + """Notify the jobrunner to reload its configuration""" + self.env.cr.execute("SELECT pg_notify('queue_job', %s)", (RELOAD_PAYLOAD,)) diff --git a/queue_job/tests/test_model_job_channel.py b/queue_job/tests/test_model_job_channel.py index 20ebbc0bfe..1277071ef4 100644 --- a/queue_job/tests/test_model_job_channel.py +++ b/queue_job/tests/test_model_job_channel.py @@ -1,9 +1,12 @@ # copyright 2018 Camptocamp # license lgpl-3.0 or later (http://www.gnu.org/licenses/lgpl.html) +from unittest import mock + from psycopg2 import IntegrityError import odoo +from odoo import exceptions from odoo.tests import common @@ -57,3 +60,83 @@ def test_channel_display_name(self): {"name": "test", "parent_id": self.root_channel.id} ) self.assertEqual(channel.display_name, channel.complete_name) + + def test_capacity_should_not_be_negative(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "The capacity of a channel cannot be negative.", + ): + self.Channel.create( + { + "name": "test_capacity", + "parent_id": self.root_channel.id, + "capacity": -1, + } + ) + + def test_throttle_should_not_be_negative(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "The throttle of a channel cannot be negative.", + ): + self.Channel.create( + { + "name": "test_throttle", + "parent_id": self.root_channel.id, + "throttle": -1, + } + ) + + def test_sequential_should_have_capacity_one(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "A sequential channel must have a capacity of 1.", + ): + self.Channel.create( + { + "name": "test_sequential", + "parent_id": self.root_channel.id, + "sequential": True, + "capacity": 2, + } + ) + + def _patch_notify(self): + return mock.patch.object( + type(self.Channel), "_notify_channel_config_changed", autospec=True + ) + + def test_notify_create_channel(self): + with self._patch_notify() as notify: + self.Channel.create( + { + "name": "create_notify", + "parent_id": self.root_channel.id, + "capacity": 2, + } + ) + notify.assert_called_once() + + def test_notify_write_jobrunner_config(self): + channel = self.Channel.create( + {"name": "write_notify", "parent_id": self.root_channel.id} + ) + with self._patch_notify() as notify: + channel.capacity = 3 + notify.assert_called_once() + + with self._patch_notify() as notify: + channel.paused = True + notify.assert_called_once() + + with self._patch_notify() as notify: + channel.removal_interval = 60 + notify.assert_not_called() + + def test_notify_unlink_channel(self): + channel = self.Channel.create( + {"name": "unlink_notify", "parent_id": self.root_channel.id} + ) + with self._patch_notify() as notify: + channel.unlink() + notify.assert_called_once() diff --git a/queue_job/views/queue_job_channel_views.xml b/queue_job/views/queue_job_channel_views.xml index 50c245716b..b6c189b427 100644 --- a/queue_job/views/queue_job_channel_views.xml +++ b/queue_job/views/queue_job_channel_views.xml @@ -19,6 +19,12 @@ + + + + + + @@ -32,6 +38,8 @@ + + @@ -44,6 +52,11 @@ + From 934ccb4fa276a10f26162943347827e17984a3e7 Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Tue, 18 Aug 2026 11:45:10 +0200 Subject: [PATCH 2/6] queue_job: wip hot reload of channel managers by DB --- queue_job/jobrunner/__init__.py | 9 +- queue_job/jobrunner/channels.py | 17 +++ queue_job/jobrunner/runner.py | 202 ++++++++++++++++++++++++++++---- 3 files changed, 200 insertions(+), 28 deletions(-) diff --git a/queue_job/jobrunner/__init__.py b/queue_job/jobrunner/__init__.py index e2561b0e74..267cac4e14 100644 --- a/queue_job/jobrunner/__init__.py +++ b/queue_job/jobrunner/__init__.py @@ -20,7 +20,7 @@ queue_job_config = config.misc.get("queue_job", {}) -from .runner import QueueJobRunner, _channels +from .runner import QueueJobRunner, _channels, _max_capacity _logger = logging.getLogger(__name__) @@ -87,7 +87,9 @@ def signal_time_expired_handler(self, n, stack): def _is_runner_enabled(): - return not _channels().strip().startswith("root:0") + if _channels().strip().startswith("root:0"): + return False + return _max_capacity() != 0 def _start_runner_thread(server_type): @@ -100,7 +102,8 @@ def _start_runner_thread(server_type): else: _logger.info( "jobrunner thread (in %s) NOT started, " - "because the root channel's capacity is set to 0", + "because the root channel's capacity or the max capacity " + "is set to 0", server_type, ) diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index a40b182ee6..b88590b55e 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -3,6 +3,7 @@ # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) import logging from collections import namedtuple +from dataclasses import asdict, dataclass from functools import total_ordering from heapq import heappop, heappush from weakref import WeakValueDictionary @@ -17,6 +18,17 @@ _logger = logging.getLogger(__name__) +@dataclass +class ChannelConfig: + """Configuration of a channel""" + + name: str + capacity: int = 0 + sequential: bool = False + throttle: int = 0 + paused: bool = False + + class PriorityQueue: """A priority queue that supports removing arbitrary objects. @@ -966,6 +978,11 @@ def simple_configure(self, config_string): for config in ChannelManager.parse_simple_config(config_string): self.get_channel_from_config(config) + def configure(self, configs): + """Configure the channel manager from list of :class:`ChannelConfig`""" + for config in configs: + self.get_channel_from_config(asdict(config)) + def get_channel_from_config(self, config): """Return a Channel object from a parsed configuration. diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 95e134ba44..10b4070c56 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -34,7 +34,7 @@ from odoo.tools import config from . import queue_job_config -from .channels import ENQUEUED, NOT_DONE, ChannelManager +from .channels import ENQUEUED, NOT_DONE, RELOAD_PAYLOAD, ChannelConfig, ChannelManager SELECT_TIMEOUT = 60 ERROR_RECOVERY_DELAY = 5 @@ -57,6 +57,52 @@ class MasterElectionLost(Exception): # so we check it in addition to the environment variables. +def _root_capacity_from_channels_config(config_string): + """Capacity of the root channel from the channels string + + >>> _root_capacity_from_channels_config('root:4,sub:2') + 4 + >>> _root_capacity_from_channels_config('sub:2') + 1 + >>> _root_capacity_from_channels_config('root:0') + 0 + """ + for channel_config in ChannelManager.parse_simple_config(config_string): + if channel_config["name"] == "root": + return channel_config.get("capacity", 1) + return 1 + + +def _max_capacity(channel_config_string=None): + """Maximum number of jobs running at the same time across all databases + + When not configured, fallbacks on the channels server-side configuration + string. + """ + value = os.environ.get("ODOO_QUEUE_JOB_MAX_CAPACITY") or queue_job_config.get( + "max_capacity" + ) + if value: + return int(value) + if channel_config_string is None: + channel_config_string = _channels() + return _root_capacity_from_channels_config(channel_config_string) + + +def _db_max_capacity(): + return int( + os.environ.get("ODOO_QUEUE_JOB_DB_MAX_CAPACITY") + or queue_job_config.get("db_max_capacity") + or _max_capacity() + ) + + +def _server_side_channels_configured(): + return bool( + os.environ.get("ODOO_QUEUE_JOB_CHANNELS") or queue_job_config.get("channels") + ) + + def _channels(): return ( os.environ.get("ODOO_QUEUE_JOB_CHANNELS") @@ -182,6 +228,30 @@ def _initialize(self): with closing(self.conn.cursor()) as cr: cr.execute("LISTEN queue_job") + def load_channels_config(self): + """Return the channels configuration stored in the database""" + with closing(self.conn.cursor()) as cr: + cr.execute( + "SELECT complete_name, " + "COALESCE(capacity, 0), " + "COALESCE(sequential, false), " + "COALESCE(throttle, 0), " + "COALESCE(paused, false) " + "FROM queue_job_channel " + ) + rows = cr.fetchall() + configs = [ + ChannelConfig( + name=name, + capacity=capacity, + sequential=sequential, + throttle=throttle, + paused=paused, + ) + for name, capacity, sequential, throttle, paused in rows + ] + return configs + @contextmanager def select_jobs(self, where, args): # pylint: disable=sql-injection @@ -327,10 +397,20 @@ def __init__( self.port = port self.user = user self.password = password - self.channel_manager = ChannelManager() + if channel_config_string is None: channel_config_string = _channels() - self.channel_manager.simple_configure(channel_config_string) + + self._server_side_channel_manager = None + if _server_side_channels_configured(): + channel_manager = ChannelManager() + channel_manager.simple_configure(channel_config_string) + self._server_side_channel_manager = channel_manager + + self.max_capacity = _max_capacity() + + self.channel_manager_by_db = {} + self.db_by_name = {} self._stop = False self._stop_pipe = os.pipe() @@ -387,21 +467,59 @@ def close_databases(self, remove_jobs=True): for db_name, db in self.db_by_name.items(): try: if remove_jobs: - self.channel_manager.remove_db(db_name) + self.channel_manager_by_db[db_name].remove_db(db_name) db.close() except Exception: _logger.warning("error closing database %s", db_name, exc_info=True) self.db_by_name = {} + def _build_channel_manager(self, db): + """Build and configure the channel manager of a database""" + # TODO: parse string "capacity per database with pattern" + db_max = _db_max_capacity() + channels_config = db.load_channels_config() + channel_manager = ChannelManager() + + root_config = next( + (config for config in channels_config if config.name == "root"), None + ) + if root_config is None: + root_config = ChannelConfig("root") + channels_config.insert(0, root_config) + + if not db_max: + # if a database is set at 0, it does not run any jobs, pause it + root_config.paused = True + elif not root_config.capacity: + root_config.capacity = db_max + else: + root_config.capacity = min(root_config.capacity, db_max) + channel_manager.configure(channels_config) + return channel_manager + + def _reconfigure_db(self, db_name): + """Rebuild the channel manager for a database and reload its jobs""" + db = self.db_by_name.get(db_name) + if db is None: + return + if self._server_side_channel_manager: + # fallback on server-side configuration with a unique channel manager + channel_manager = self._server_side_channel_manager + else: + channel_manager = self._build_channel_manager(db) + with db.select_jobs("state in %s", (NOT_DONE,)) as cr: + for job_data in cr: + channel_manager.notify(db_name, *job_data) + self.channel_manager_by_db[db_name] = channel_manager + _logger.info("channels configuration loaded for db %s", db_name) + def initialize_databases(self): for db_name in sorted(self.get_db_names()): # sorting is important to avoid deadlocks in acquiring the master lock db = Database(db_name) if db.has_queue_job: self.db_by_name[db_name] = db - with db.select_jobs("state in %s", (NOT_DONE,)) as cr: - for job_data in cr: - self.channel_manager.notify(db_name, *job_data) + self._reconfigure_db(db_name) _logger.info("queue job runner ready for db %s", db_name) else: db.close() @@ -411,24 +529,41 @@ def requeue_dead_jobs(self): if db.has_queue_job: db.requeue_dead_jobs() + def _dispatch_job(self, job): + _logger.info("asking Odoo to run job %s on db %s", job.uuid, job.db_name) + self.db_by_name[job.db_name].set_job_enqueued(job.uuid) + _async_http_get( + self.scheme, + self.host, + self.port, + self.user, + self.password, + job.db_name, + job.uuid, + ) + def run_jobs(self): + db_names = list(self.channel_manager_by_db) + if not db_names: + return + now = _odoo_now() - for job in self.channel_manager.get_jobs_to_run(now): - if self._stop: - break - _logger.info("asking Odoo to run job %s on db %s", job.uuid, job.db_name) - self.db_by_name[job.db_name].set_job_enqueued(job.uuid) - _async_http_get( - self.scheme, - self.host, - self.port, - self.user, - self.password, - job.db_name, - job.uuid, - ) + + # TODO: round robin + for db_name in db_names: + jobs = self.channel_manager_by_db[db_name].get_jobs_to_run(now) + while True: + if self._stop: + break + # TODO: check if max capacity for db reached so another db + # can dispatch jobs + job = next(jobs, None) + if job is None: + break + self._dispatch_job(job) def process_notifications(self): + reload_db_names = set() for db in self.db_by_name.values(): if not db.conn.notifies: # If there are no activity in the queue_job table it seems that @@ -440,13 +575,29 @@ def process_notifications(self): if self._stop: break notification = db.conn.notifies.pop() - uuid = notification.payload + payload = notification.payload + if payload == RELOAD_PAYLOAD and not self._server_side_channel_manager: + reload_db_names.add(db.db_name) + continue + + uuid = payload + channel_manager = self.channel_manager_by_db[db.db_name] with db.select_jobs("uuid = %s", (uuid,)) as cr: job_datas = cr.fetchone() if job_datas: - self.channel_manager.notify(db.db_name, *job_datas) + channel_manager.notify(db.db_name, *job_datas) else: - self.channel_manager.remove_job(uuid) + channel_manager.remove_job(uuid) + + for db_name in reload_db_names: + self._reconfigure_db(db_name) + + def next_wakeup_time(self): + wakeup_times = [ + channel_manager.get_wakeup_time() + for channel_manager in self.channel_manager_by_db.values() + ] + return min(wakeup_times, default=0) def wait_notification(self): for db in self.db_by_name.values(): @@ -458,7 +609,8 @@ def wait_notification(self): conns = [db.conn for db in self.db_by_name.values()] conns.append(self._stop_pipe[0]) # look if the channels specify a wakeup time - wakeup_time = self.channel_manager.get_wakeup_time() + # TODO: get min wakeup time? + wakeup_time = self.next_wakeup_time() if not wakeup_time: # this could very well be no timeout at all, because # any activity in the job queue will wake us up, but From 3d8346eb2246531a1db712c4092236235d7b87fd Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Wed, 19 Aug 2026 10:18:18 +0200 Subject: [PATCH 3/6] fixup! queue_job: wip hot reload of channel managers by DB --- queue_job/jobrunner/channels.py | 5 ++ queue_job/jobrunner/runner.py | 130 +++++++++++++++++++++++++++----- 2 files changed, 116 insertions(+), 19 deletions(-) diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index b88590b55e..4d7b441e26 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -1133,3 +1133,8 @@ def get_jobs_to_run(self, now): def get_wakeup_time(self): return self._root_channel.get_wakeup_time() + + @property + def running_count(self) -> int: + """Number of jobs currently running""" + return len(self._root_channel._running) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 10b4070c56..d6bee40ae5 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -19,6 +19,7 @@ anonymous ``/queue_job/runjob`` HTTP request. """ +import fnmatch import logging import os import selectors @@ -57,6 +58,12 @@ class MasterElectionLost(Exception): # so we check it in addition to the environment variables. +def _server_side_channels_configured(): + return bool( + os.environ.get("ODOO_QUEUE_JOB_CHANNELS") or queue_job_config.get("channels") + ) + + def _root_capacity_from_channels_config(config_string): """Capacity of the root channel from the channels string @@ -73,34 +80,88 @@ def _root_capacity_from_channels_config(config_string): return 1 -def _max_capacity(channel_config_string=None): +def _max_capacity(channel_config_string: str | None = None) -> int: """Maximum number of jobs running at the same time across all databases When not configured, fallbacks on the channels server-side configuration string. """ + if _server_side_channels_configured(): + if channel_config_string is None: + channel_config_string = _channels() + return _root_capacity_from_channels_config(channel_config_string) + value = os.environ.get("ODOO_QUEUE_JOB_MAX_CAPACITY") or queue_job_config.get( "max_capacity" ) if value: return int(value) - if channel_config_string is None: - channel_config_string = _channels() - return _root_capacity_from_channels_config(channel_config_string) + return 0 -def _db_max_capacity(): - return int( +def _db_max_capacity() -> str: + return ( os.environ.get("ODOO_QUEUE_JOB_DB_MAX_CAPACITY") or queue_job_config.get("db_max_capacity") - or _max_capacity() + or "" ) -def _server_side_channels_configured(): - return bool( - os.environ.get("ODOO_QUEUE_JOB_CHANNELS") or queue_job_config.get("channels") - ) +def parse_db_max_capacity(spec): + """Parse a per-database max capacity configuration string + + The string is a comma-separated list of ``pattern:capacity`` items, where + ``pattern`` matches database names with fnmatch wildcards. + + The first matching pattern wins, so specific patterns must be first in the + string. + + A single integer is applied to all databases, as a shorthand for + ``*:capacity``. + + >>> parse_db_max_capacity('prod_*:20,staging:2,*:5') + [('prod_*', 20), ('staging', 2), ('*', 5)] + >>> parse_db_max_capacity('8') + [('*', 8)] + >>> parse_db_max_capacity('') + [] + >>> parse_db_max_capacity(None) + [] + """ + rules = [] + if not spec: + return rules + for item in spec.replace("\n", ",").split(","): + item = item.strip() + if not item: + continue + pattern, sep, capacity = item.rpartition(":") + if not sep: + pattern = "*" + try: + rules.append((pattern.strip(), int(capacity))) + except ValueError as ex: + raise ValueError(f"Invalid db max capacity {spec}: {capacity}") from ex + return rules + + +def db_max_capacity_for(db_name, rules, default=None): + """Max capacity of a database, first match wins + + >>> rules = parse_db_max_capacity('prod_*:20,staging:2,*:5') + >>> db_max_capacity_for('prod_foo', rules) + 20 + >>> db_max_capacity_for('staging', rules) + 2 + >>> db_max_capacity_for('dev', rules) + 5 + >>> db_max_capacity_for('dev', [], default=7) + 7 + """ + for pattern, capacity in rules: + if fnmatch.fnmatch(db_name, pattern): + return capacity + return default def _channels(): @@ -391,6 +452,8 @@ def __init__( user=None, password=None, channel_config_string=None, + max_capacity=None, + db_max_capacity=None, ): self.scheme = scheme self.host = host @@ -407,10 +470,18 @@ def __init__( channel_manager.simple_configure(channel_config_string) self._server_side_channel_manager = channel_manager - self.max_capacity = _max_capacity() + if max_capacity is None: + max_capacity = _max_capacity() + self.max_capacity = max_capacity + + if db_max_capacity is None: + db_max_capacity = _db_max_capacity() + self.db_max_capacity_rules = parse_db_max_capacity(db_max_capacity) self.channel_manager_by_db = {} + self._round_robin_offset = 0 + self.db_by_name = {} self._stop = False self._stop_pipe = os.pipe() @@ -475,8 +546,9 @@ def close_databases(self, remove_jobs=True): def _build_channel_manager(self, db): """Build and configure the channel manager of a database""" - # TODO: parse string "capacity per database with pattern" - db_max = _db_max_capacity() + db_max = db_max_capacity_for( + db.db_name, self.db_max_capacity_rules, default=self.max_capacity + ) channels_config = db.load_channels_config() channel_manager = ChannelManager() @@ -529,6 +601,14 @@ def requeue_dead_jobs(self): if db.has_queue_job: db.requeue_dead_jobs() + def _all_running_count(self) -> int: + if self._server_side_channel_manager: + return self._server_side_channel_manager.running_count + return sum( + channel_manager.running_count + for channel_manager in self.channel_manager_by_db.values() + ) + def _dispatch_job(self, job): _logger.info("asking Odoo to run job %s on db %s", job.uuid, job.db_name) self.db_by_name[job.db_name].set_job_enqueued(job.uuid) @@ -549,14 +629,27 @@ def run_jobs(self): now = _odoo_now() - # TODO: round robin - for db_name in db_names: + round_robin_offset = self._round_robin_offset % len(db_names) + self._round_robin_offset += 1 + + # rotate the databases order so that each database gets a chance to enqueue jobs + # under contention + for db_name in db_names[round_robin_offset:] + db_names[:round_robin_offset]: jobs = self.channel_manager_by_db[db_name].get_jobs_to_run(now) while True: if self._stop: break - # TODO: check if max capacity for db reached so another db - # can dispatch jobs + # TODO: think about server-side vs per multi channel manager behaviors + if self.max_capacity and self._all_running_count() >= self.max_capacity: + # we trigger a round-robin only when the global max + # capacity is reached, before that, databases already can + # enqueue jobs + _logger.debug( + "max capacity of %s reached for db %s, round-robin to next db", + self.max_capacity, + db_name, + ) + return job = next(jobs, None) if job is None: break @@ -609,7 +702,6 @@ def wait_notification(self): conns = [db.conn for db in self.db_by_name.values()] conns.append(self._stop_pipe[0]) # look if the channels specify a wakeup time - # TODO: get min wakeup time? wakeup_time = self.next_wakeup_time() if not wakeup_time: # this could very well be no timeout at all, because From 31f1eaaf20586a7776595d6c18e38c2f04baf3e9 Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Thu, 20 Aug 2026 11:25:04 +0200 Subject: [PATCH 4/6] fixup! fixup! queue_job: wip hot reload of channel managers by DB --- queue_job/jobrunner/runner.py | 58 ++++++++++++++++++++++------------- 1 file changed, 36 insertions(+), 22 deletions(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index d6bee40ae5..565aaf0acc 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -470,6 +470,9 @@ def __init__( channel_manager.simple_configure(channel_config_string) self._server_side_channel_manager = channel_manager + self._channel_manager_by_db = {} + self._channel_managers = [] + if max_capacity is None: max_capacity = _max_capacity() self.max_capacity = max_capacity @@ -478,8 +481,6 @@ def __init__( db_max_capacity = _db_max_capacity() self.db_max_capacity_rules = parse_db_max_capacity(db_max_capacity) - self.channel_manager_by_db = {} - self._round_robin_offset = 0 self.db_by_name = {} @@ -538,11 +539,23 @@ def close_databases(self, remove_jobs=True): for db_name, db in self.db_by_name.items(): try: if remove_jobs: - self.channel_manager_by_db[db_name].remove_db(db_name) + self._channel_manager_by_db[db_name].remove_db(db_name) db.close() except Exception: _logger.warning("error closing database %s", db_name, exc_info=True) self.db_by_name = {} + self._channel_manager_by_db = {} + self._channel_managers = [] + + @staticmethod + def _unique_channel_managers(channel_managers): + seen = set() + result = [] + for channel_manager in channel_managers: + if id(channel_manager) not in seen: + seen.add(id(channel_manager)) + result.append(channel_manager) + return result def _build_channel_manager(self, db): """Build and configure the channel manager of a database""" @@ -582,7 +595,10 @@ def _reconfigure_db(self, db_name): with db.select_jobs("state in %s", (NOT_DONE,)) as cr: for job_data in cr: channel_manager.notify(db_name, *job_data) - self.channel_manager_by_db[db_name] = channel_manager + self._channel_manager_by_db[db_name] = channel_manager + self._channel_managers = self._unique_channel_managers( + self._channel_manager_by_db.values() + ) _logger.info("channels configuration loaded for db %s", db_name) def initialize_databases(self): @@ -602,11 +618,8 @@ def requeue_dead_jobs(self): db.requeue_dead_jobs() def _all_running_count(self) -> int: - if self._server_side_channel_manager: - return self._server_side_channel_manager.running_count return sum( - channel_manager.running_count - for channel_manager in self.channel_manager_by_db.values() + channel_manager.running_count for channel_manager in self._channel_managers ) def _dispatch_job(self, job): @@ -623,31 +636,32 @@ def _dispatch_job(self, job): ) def run_jobs(self): - db_names = list(self.channel_manager_by_db) - if not db_names: + channel_managers = self._channel_managers + if not channel_managers: return now = _odoo_now() - round_robin_offset = self._round_robin_offset % len(db_names) + round_robin_offset = self._round_robin_offset % len(channel_managers) self._round_robin_offset += 1 - # rotate the databases order so that each database gets a chance to enqueue jobs - # under contention - for db_name in db_names[round_robin_offset:] + db_names[:round_robin_offset]: - jobs = self.channel_manager_by_db[db_name].get_jobs_to_run(now) + # rotate the channel managers order so that each database gets a chance + # to enqueue jobs under contention + for channel_manager in ( + channel_managers[round_robin_offset:] + + channel_managers[:round_robin_offset] + ): + jobs = channel_manager.get_jobs_to_run(now) while True: if self._stop: break - # TODO: think about server-side vs per multi channel manager behaviors if self.max_capacity and self._all_running_count() >= self.max_capacity: # we trigger a round-robin only when the global max - # capacity is reached, before that, databases already can - # enqueue jobs + # capacity is reached, before that, all channel managers can + # already enqueue jobs _logger.debug( - "max capacity of %s reached for db %s, round-robin to next db", + "max capacity of %s reached, round-robin to next db", self.max_capacity, - db_name, ) return job = next(jobs, None) @@ -674,7 +688,7 @@ def process_notifications(self): continue uuid = payload - channel_manager = self.channel_manager_by_db[db.db_name] + channel_manager = self._channel_manager_by_db[db.db_name] with db.select_jobs("uuid = %s", (uuid,)) as cr: job_datas = cr.fetchone() if job_datas: @@ -688,7 +702,7 @@ def process_notifications(self): def next_wakeup_time(self): wakeup_times = [ channel_manager.get_wakeup_time() - for channel_manager in self.channel_manager_by_db.values() + for channel_manager in self._channel_managers ] return min(wakeup_times, default=0) From 8f9f24d8fb8478d2dd5c37a219131634b2a33907 Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Thu, 20 Aug 2026 11:34:54 +0200 Subject: [PATCH 5/6] fixup! fixup! fixup! queue_job: wip hot reload of channel managers by DB --- queue_job/jobrunner/runner.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 565aaf0acc..2adb125086 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -83,8 +83,14 @@ def _root_capacity_from_channels_config(config_string): def _max_capacity(channel_config_string: str | None = None) -> int: """Maximum number of jobs running at the same time across all databases - When not configured, fallbacks on the channels server-side configuration - string. + If a channels server-side configuration exists, it is equivalent to the + capacity of the root channel. + + Otherwise, it comes from the ``ODOO_QUEUE_JOB_MAX_CAPACITY`` environment + variable, then ``max_capacity`` in the ``[queue_job]`` section of the + configuration file. + + If none is configured, the max capacity is 0. """ if _server_side_channels_configured(): if channel_config_string is None: @@ -588,7 +594,6 @@ def _reconfigure_db(self, db_name): if db is None: return if self._server_side_channel_manager: - # fallback on server-side configuration with a unique channel manager channel_manager = self._server_side_channel_manager else: channel_manager = self._build_channel_manager(db) From 3407603730dabb390cda796a9036c9d829f50a78 Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Fri, 21 Aug 2026 09:21:33 +0200 Subject: [PATCH 6/6] fixup! fixup! fixup! fixup! queue_job: wip hot reload of channel managers by DB --- queue_job/views/queue_job_channel_views.xml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/queue_job/views/queue_job_channel_views.xml b/queue_job/views/queue_job_channel_views.xml index b6c189b427..a679596438 100644 --- a/queue_job/views/queue_job_channel_views.xml +++ b/queue_job/views/queue_job_channel_views.xml @@ -20,6 +20,10 @@ +

+ These properties will be used only if the server-side + channels configuration is not used. +