diff --git a/queue_job/README.rst b/queue_job/README.rst index a37109a861..3ba1564fab 100644 --- a/queue_job/README.rst +++ b/queue_job/README.rst @@ -143,6 +143,22 @@ Configuration * Jobs that remain in `enqueued` or `started` state (because, for instance, their worker has been killed) will be automatically re-queued. +* By default, a worker will defer (not execute) a job if its own installed + module code no longer matches what the database expects (e.g. it hasn't yet + picked up a rolling deployment's update). Disable with the + ``queue_job.check_modules_up_to_date`` system parameter set to ``False``. + +* By default, a worker will also defer any job while modules are being + installed/upgraded/removed anywhere in the cluster (the same condition core + Odoo already uses to skip scheduled actions, see ``ir_cron``), since a job + committed early in an install (e.g. from a ``post_init_hook``) may reference + code that doesn't match the database until the install fully completes. + Pending states older than ``queue_job.install_in_progress_timeout_minutes`` + (default ``60``) are treated as abandoned and ignored, with a warning + logged, rather than blocking jobs forever. Disable the check entirely with + the ``queue_job.check_install_in_progress`` system parameter set to + ``False``. + Usage ===== diff --git a/queue_job/controllers/main.py b/queue_job/controllers/main.py index 02e05a6a96..f12b19b2e8 100644 --- a/queue_job/controllers/main.py +++ b/queue_job/controllers/main.py @@ -14,6 +14,7 @@ from werkzeug.exceptions import BadRequest, Forbidden from odoo import SUPERUSER_ID, _, api, http, tools +from odoo.modules.module import get_manifest from odoo.service.model import PG_CONCURRENCY_ERRORS_TO_RETRY from odoo.tools import config @@ -27,6 +28,15 @@ DEPENDS_MAX_TRIES_ON_CONCURRENCY_FAILURE = 5 +MODULES_OUTDATED_POSTPONE_SECONDS = 15 +INSTALL_IN_PROGRESS_POSTPONE_SECONDS = 15 + +# How long a "is this worker's module code up to date" verdict is cached before +# being recomputed, per database (a single process can serve several). Keeps a +# busy jobrunner from re-querying ir.module.module on every single job dispatch. +MODULES_UP_TO_DATE_CACHE_SECONDS = 5 +_modules_up_to_date_cache = {} # dbname -> (checked_at, outdated: bool) + @contextmanager def _prevent_commit(cr): @@ -110,6 +120,133 @@ def _try_perform_job(cls, env, job): env.cr.commit() _logger.debug("%s done", job) + @classmethod + def _install_in_progress(cls, env): + """True while any module is marked to be installed/upgraded/removed. + + Same guard core already applies to cron jobs + (ir_cron._check_modules_state): while an install/upgrade is running + anywhere in the cluster, module states sit at 'to install'/'to + upgrade'/'to remove' in the database, and only flip back once the + operation completes (or core's reset_modules_state() clears them on + failure). Jobs executed during that window may run with code that + does not yet match what the database will end up recording -- a + module's own 'installed' state and latest_version are only committed + near the end of its install, but data/hooks that ran earlier in the + same install (e.g. a post_init_hook enqueuing a job) may already be + committed and dispatchable. _worker_is_outdated() has nothing to + compare against in that window (the module isn't even marked + installed yet); this check covers exactly that gap. + + Deliberately not cached like _worker_is_outdated(): the whole point + is that a job committed by a post_init_hook becomes visible in the + same commit as the pending module state, so a fresh read makes this + check exact. A TTL would reopen the race for its duration. The query + is one cheap aggregate over ir_module_module. + + Pending states older than + queue_job.install_in_progress_timeout_minutes (default 60) are + treated as abandoned (e.g. modules left marked by a crashed/killed + process) and ignored, with a warning logged, rather than freezing + job processing forever. Disable the whole check with the + queue_job.check_install_in_progress system parameter. + """ + icp = env["ir.config_parameter"].sudo() + if icp.get_param("queue_job.check_install_in_progress", "True") == "False": + return False + + timeout_minutes = int( + icp.get_param("queue_job.install_in_progress_timeout_minutes", "60") + ) + env.cr.execute( + """ + SELECT COUNT(*), + MAX(COALESCE(write_date, create_date)) + >= (now() AT TIME ZONE 'UTC') - %s * interval '1 minute' + FROM ir_module_module + WHERE state IN ('to install', 'to upgrade', 'to remove') + """, + (timeout_minutes,), + ) + pending, fresh = env.cr.fetchone() + if not pending: + return False + if not fresh: + _logger.warning( + "%d module(s) have been marked to install/upgrade/remove for " + "more than %d minutes; assuming an abandoned operation and " + "resuming job processing", + pending, + timeout_minutes, + ) + return False + # Only reached while an install is ACTIVELY in progress (pending & + # fresh) -- the common no-install case already returned False above. + # Bust the version-check cache so the first job dispatched after the + # install completes recomputes _worker_is_outdated() freshly instead + # of reusing a pre-install verdict for up to + # MODULES_UP_TO_DATE_CACHE_SECONDS. + _modules_up_to_date_cache.pop(env.cr.dbname, None) + return True + + @classmethod + def _worker_is_outdated(cls, env): + """True if this worker's own on-disk module code no longer matches + what the database considers the currently installed version, for at + least one installed module. + + ir.module.module.installed_version is a non-stored compute that reads + the manifest fresh from *this process's* addons_path (get_manifest() + is lru_cached per process, so this is cheap after the first call). + ir.module.module.latest_version is what the module loader persisted to + the database the last time this module was actually + installed/upgraded (see odoo/modules/loading.py). A mismatch means + some other process has since upgraded this module past what this + worker is currently running -- most commonly, one worker in a + multi-process or multi-host deployment has picked up new code and run + an upgrade, while this one is still serving requests with the old + code, in the window before it gets restarted (a rolling-deployment + race). Running a job in that state risks acting on stale files/logic + that no longer match the database. + + The result is cached briefly, per database, so a busy queue doesn't + re-run this on every single job (see MODULES_UP_TO_DATE_CACHE_SECONDS). + + Disable via the ``queue_job.check_modules_up_to_date`` system + parameter (set to ``"False"``); enabled by default. + """ + icp = env["ir.config_parameter"].sudo() + if icp.get_param("queue_job.check_modules_up_to_date", "True") == "False": + return False + + dbname = env.cr.dbname + now = time.monotonic() + cached = _modules_up_to_date_cache.get(dbname) + if cached and now - cached[0] < MODULES_UP_TO_DATE_CACHE_SECONDS: + return cached[1] + + outdated = False + modules = env["ir.module.module"].sudo().search([("state", "=", "installed")]) + for module in modules: + manifest = get_manifest(module.name) + if not manifest: + # Can't read this module's manifest from this process's + # addons_path (e.g. a module removed from disk but still + # marked installed in the database) -- skip it rather than + # treat an unreadable manifest as evidence of staleness. + continue + on_disk_version = manifest.get("version") + if ( + on_disk_version + and module.latest_version + and on_disk_version != module.latest_version + ): + outdated = True + break + + _modules_up_to_date_cache[dbname] = (now, outdated) + return outdated + @classmethod def _enqueue_dependent_jobs(cls, env, job): tries = 0 @@ -150,6 +287,37 @@ def retry_postpone(job, message, seconds=None): job.set_pending(reset_retry=False) job.store() + if cls._install_in_progress(env): + _logger.info( + "%s: modules are being installed/upgraded somewhere in the " + "cluster; postponing %ss", + job, + INSTALL_IN_PROGRESS_POSTPONE_SECONDS, + ) + retry_postpone( + job, + "modules install in progress", + seconds=INSTALL_IN_PROGRESS_POSTPONE_SECONDS, + ) + if not config["test_enable"]: + env.cr.rollback() + return + + if cls._worker_is_outdated(env): + _logger.info( + "%s: this worker's module code is outdated relative to the " + "database (likely a rolling deployment in progress); " + "postponing %ss", + job, + MODULES_OUTDATED_POSTPONE_SECONDS, + ) + retry_postpone( + job, "worker outdated", seconds=MODULES_OUTDATED_POSTPONE_SECONDS + ) + if not config["test_enable"]: + env.cr.rollback() + return + try: try: cls._try_perform_job(env, job) diff --git a/queue_job/readme/CONFIGURE.rst b/queue_job/readme/CONFIGURE.rst index 0d24075c0a..0194487000 100644 --- a/queue_job/readme/CONFIGURE.rst +++ b/queue_job/readme/CONFIGURE.rst @@ -48,3 +48,19 @@ of running Odoo is obviously not for production purposes. * Jobs that remain in `enqueued` or `started` state (because, for instance, their worker has been killed) will be automatically re-queued. + +* By default, a worker will defer (not execute) a job if its own installed + module code no longer matches what the database expects (e.g. it hasn't yet + picked up a rolling deployment's update). Disable with the + ``queue_job.check_modules_up_to_date`` system parameter set to ``False``. + +* By default, a worker will also defer any job while modules are being + installed/upgraded/removed anywhere in the cluster (the same condition core + Odoo already uses to skip scheduled actions, see ``ir_cron``), since a job + committed early in an install (e.g. from a ``post_init_hook``) may reference + code that doesn't match the database until the install fully completes. + Pending states older than ``queue_job.install_in_progress_timeout_minutes`` + (default ``60``) are treated as abandoned and ignored, with a warning + logged, rather than blocking jobs forever. Disable the check entirely with + the ``queue_job.check_install_in_progress`` system parameter set to + ``False``. diff --git a/queue_job/static/description/index.html b/queue_job/static/description/index.html index 4d9d62b9c8..4f3e63af9c 100644 --- a/queue_job/static/description/index.html +++ b/queue_job/static/description/index.html @@ -503,6 +503,20 @@

Configuration

diff --git a/queue_job/tests/test_run_rob_controller.py b/queue_job/tests/test_run_rob_controller.py index 1a15f4363a..0be0e21e8e 100644 --- a/queue_job/tests/test_run_rob_controller.py +++ b/queue_job/tests/test_run_rob_controller.py @@ -1,12 +1,34 @@ # License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). +from unittest.mock import patch + +from odoo.modules.module import get_manifest as real_get_manifest from odoo.tests.common import TransactionCase +from ..controllers import main as main_module from ..controllers.main import RunJobController from ..job import Job +_CTRL_MOD = "odoo.addons.queue_job.controllers.main" + class TestRunJobController(TransactionCase): + def setUp(self): + super().setUp() + # _worker_is_outdated() caches its verdict in a process-level global, + # not per-transaction -- TransactionCase's rollback between tests has + # no way to clear it, so each test must start from a clean cache. + main_module._modules_up_to_date_cache.clear() + # When this suite runs as part of installing several modules at once + # (e.g. -i queue_job,test_queue_job), a module later in the batch can + # genuinely still be 'to install' while queue_job's own tests run -- + # that's real, unrelated to anything under test here. Normalize to a + # clean baseline; individual tests set up their own pending state. + self.env.cr.execute( + "UPDATE ir_module_module SET state = 'installed' " + "WHERE state IN ('to install', 'to upgrade', 'to remove')" + ) + def test_get_failure_values(self): method = self.env["res.users"].mapped job = Job(method) @@ -21,3 +43,146 @@ def test_runjob_success(self): RunJobController._runjob(self.env, job) self.assertEqual(job.state, "done") self.assertEqual(job.db_record().state, "done") + + def test_runjob_postpones_when_worker_outdated(self): + """When the worker's module code is outdated, the job is postponed + instead of executed, and _try_perform_job is never reached.""" + # job.store() here runs inside job.in_temporary_env(), which commits + # on its own separate cursor (job.py) so a postponed job survives + # even if its own transaction rolls back -- meaning this row also + # survives TransactionCase's rollback. Clean it up the same way. + job = self.env["queue.job"].with_delay()._test_job() + self.addCleanup(self._delete_committed_job, job.uuid) + with patch.object(RunJobController, "_worker_is_outdated", return_value=True): + with patch.object(RunJobController, "_try_perform_job") as mock_perform: + RunJobController._runjob(self.env, job) + mock_perform.assert_not_called() + self.assertEqual(job.state, "pending") + self.assertEqual(job.db_record().state, "pending") + + def test_runjob_postpones_when_install_in_progress(self): + """When modules are being installed/upgraded, the job is postponed + instead of executed, and _try_perform_job is never reached.""" + job = self.env["queue.job"].with_delay()._test_job() + self.addCleanup(self._delete_committed_job, job.uuid) + with patch.object(RunJobController, "_install_in_progress", return_value=True): + with patch.object(RunJobController, "_try_perform_job") as mock_perform: + RunJobController._runjob(self.env, job) + mock_perform.assert_not_called() + self.assertEqual(job.state, "pending") + self.assertEqual(job.db_record().state, "pending") + + def _delete_committed_job(self, uuid): + with self.env.registry.cursor() as cr: + cr.execute("DELETE FROM queue_job WHERE uuid = %s", (uuid,)) + + def test_worker_is_outdated_false_by_default(self): + """A freshly installed test DB is never outdated: on-disk versions + already match what was recorded in the database.""" + # A fresh test DB has every installed module's on-disk version + # matching what was recorded at install time -- no setup needed. + self.assertFalse(RunJobController._worker_is_outdated(self.env)) + + def test_worker_is_outdated_true_on_mismatch(self): + """A module whose recorded latest_version differs from its on-disk + manifest version is detected as outdated.""" + base = ( + self.env["ir.module.module"].sudo().search([("name", "=", "base")], limit=1) + ) + base.write({"latest_version": "0.0.0"}) + self.assertTrue(RunJobController._worker_is_outdated(self.env)) + + def test_worker_is_outdated_ignores_unreadable_manifest(self): + """A module whose manifest can't be read from disk is skipped, not + treated as evidence of staleness.""" + + def fake_get_manifest(module_name, mod_path=None): + if module_name == "base": + return {} + return real_get_manifest(module_name, mod_path) + + with patch(_CTRL_MOD + ".get_manifest", side_effect=fake_get_manifest): + # "base"'s manifest is unreadable here, but nothing else is + # actually mismatched -- must not be treated as outdated. + self.assertFalse(RunJobController._worker_is_outdated(self.env)) + + def test_worker_is_outdated_disabled_via_config_param(self): + """Setting queue_job.check_modules_up_to_date to False skips the + check entirely, even with a real version mismatch present.""" + base = ( + self.env["ir.module.module"].sudo().search([("name", "=", "base")], limit=1) + ) + base.write({"latest_version": "0.0.0"}) + self.env["ir.config_parameter"].sudo().set_param( + "queue_job.check_modules_up_to_date", "False" + ) + self.assertFalse(RunJobController._worker_is_outdated(self.env)) + + def test_worker_is_outdated_result_is_cached(self): + """The outdated verdict is cached per database for a few seconds, + and only recomputed once that cache entry is expired.""" + base = ( + self.env["ir.module.module"].sudo().search([("name", "=", "base")], limit=1) + ) + original_latest_version = base.latest_version + base.write({"latest_version": "0.0.0"}) + self.assertTrue(RunJobController._worker_is_outdated(self.env)) + + # Fix the mismatch -- the cached verdict should still say True. + base.write({"latest_version": original_latest_version}) + self.assertTrue(RunJobController._worker_is_outdated(self.env)) + + # Force the cache entry to look expired -> recompute -> now False. + dbname = self.env.cr.dbname + main_module._modules_up_to_date_cache[dbname] = (0.0, True) + self.assertFalse(RunJobController._worker_is_outdated(self.env)) + + def test_install_in_progress_false_by_default(self): + """A freshly installed test DB has no module pending install/upgrade/ + removal -- no setup needed.""" + self.assertFalse(RunJobController._install_in_progress(self.env)) + + def test_install_in_progress_true_when_module_pending(self): + """A module marked 'to upgrade' (or 'to install'/'to remove'), with a + recent write_date, is detected as an install in progress.""" + self.env.cr.execute( + "UPDATE ir_module_module SET state = 'to upgrade', " + "write_date = (now() AT TIME ZONE 'UTC') WHERE name = 'base'" + ) + self.assertTrue(RunJobController._install_in_progress(self.env)) + + def test_install_in_progress_ignores_stale_pending_states(self): + """A module that has been pending for longer than the configured + timeout is treated as an abandoned operation, not an active one.""" + self.env.cr.execute( + "UPDATE ir_module_module SET state = 'to upgrade', " + "write_date = (now() AT TIME ZONE 'UTC') - interval '61 minutes' " + "WHERE name = 'base'" + ) + with self.assertLogs("odoo.addons.queue_job.controllers.main", level="WARNING"): + self.assertFalse(RunJobController._install_in_progress(self.env)) + + def test_install_in_progress_disabled_via_config_param(self): + """Setting queue_job.check_install_in_progress to False skips the + check entirely, even with a module genuinely pending.""" + self.env.cr.execute( + "UPDATE ir_module_module SET state = 'to upgrade', " + "write_date = (now() AT TIME ZONE 'UTC') WHERE name = 'base'" + ) + self.env["ir.config_parameter"].sudo().set_param( + "queue_job.check_install_in_progress", "False" + ) + self.assertFalse(RunJobController._install_in_progress(self.env)) + + def test_install_in_progress_busts_outdated_cache(self): + """Detecting an active install clears any cached _worker_is_outdated + verdict, so the first job dispatched once the install completes + recomputes it freshly instead of reusing a pre-install answer.""" + dbname = self.env.cr.dbname + main_module._modules_up_to_date_cache[dbname] = (0.0, False) + self.env.cr.execute( + "UPDATE ir_module_module SET state = 'to upgrade', " + "write_date = (now() AT TIME ZONE 'UTC') WHERE name = 'base'" + ) + self.assertTrue(RunJobController._install_in_progress(self.env)) + self.assertNotIn(dbname, main_module._modules_up_to_date_cache)