From ce757d22ea8a5695091f47029c328362c7f0ace2 Mon Sep 17 00:00:00 2001 From: Holoborodko Bohdan Date: Mon, 28 Sep 2026 11:19:15 +0000 Subject: [PATCH 1/5] [FIX] queue_job: detect a dropped job runner connection The job runner holds its lock on a connection that stays open. When the network drops it without closing it (firewall, NAT timeout, database failover), neither side notices: the job runner blocks in its next query until the kernel gives up retransmitting (about 15 minutes on Linux), and the database keeps the session, and so the lock, until its own TCP keepalive fires (2 hours by default). No other job runner can take over meanwhile. Enable TCP keepalives and a TCP user timeout on the connection, on the client side through the libpq parameters and on the server side through the session settings, configurable like the other jobrunner_db_* keys. Fixes #984 --- queue_job/jobrunner/runner.py | 41 +++++++++++++++- queue_job/readme/CONFIGURE.md | 12 +++++ queue_job/readme/newsfragments/984.bugfix | 3 ++ queue_job/tests/test_runner_runner.py | 59 +++++++++++++++++++++++ 4 files changed, 114 insertions(+), 1 deletion(-) create mode 100644 queue_job/readme/newsfragments/984.bugfix diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 5cbd52ef48..5616af5e10 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -42,6 +42,23 @@ ERROR_RECOVERY_DELAY = 5 PG_ADVISORY_LOCK_ID = 2293787760715711918 +# libpq connection parameters detecting a connection dropped by the network, +# with the server setting holding the same value, if any +KEEPALIVE_DEFAULTS = { + "keepalives": 1, + "keepalives_idle": 30, + "keepalives_interval": 10, + "keepalives_count": 3, + "tcp_user_timeout": 60000, +} +KEEPALIVE_SERVER_SETTINGS = { + "keepalives_idle": "tcp_keepalives_idle", + "keepalives_interval": "tcp_keepalives_interval", + "keepalives_count": "tcp_keepalives_count", + "tcp_user_timeout": "tcp_user_timeout", +} +LIBPQ_TCP_USER_TIMEOUT_VERSION = 120000 + _logger = logging.getLogger(__name__) select = selectors.DefaultSelector @@ -155,7 +172,7 @@ def _odoo_now(): def _connection_info_for(db_name): db_or_uri, connection_info = odoo.sql_db.connection_info_for(db_name) - for p in ("host", "port", "user", "password"): + for p in ("host", "port", "user", "password", *KEEPALIVE_DEFAULTS): cfg = os.environ.get( f"ODOO_QUEUE_JOB_JOBRUNNER_DB_{p.upper()}" ) or queue_job_config.get("jobrunner_db_" + p) @@ -163,6 +180,11 @@ def _connection_info_for(db_name): if cfg: connection_info[p] = cfg + for p, default in KEEPALIVE_DEFAULTS.items(): + connection_info.setdefault(p, default) + if psycopg2.extensions.libpq_version() < LIBPQ_TCP_USER_TIMEOUT_VERSION: + connection_info.pop("tcp_user_timeout") + return connection_info @@ -203,6 +225,7 @@ def __init__(self, db_name): self.conn = psycopg2.connect(**connection_info) try: self.conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) + self._set_server_keepalives(connection_info) self.has_queue_job = self._has_queue_job() self.has_channel_config_columns = False if self.has_queue_job: @@ -224,6 +247,22 @@ def close(self): pass self.conn = None + def _set_server_keepalives(self, connection_info): + """Apply the keepalive parameters on the server side of the connection + + The server then ends the session, releasing the master lock, when the + job runner is no longer reachable, and another one can take over. + """ + if not int(connection_info["keepalives"]): + return + with closing(self.conn.cursor()) as cr: + for param, setting in KEEPALIVE_SERVER_SETTINGS.items(): + if param in connection_info: + cr.execute( + "SELECT set_config(%s, %s, false)", + (setting, str(connection_info[param])), + ) + def _acquire_master_lock(self): """Acquire the master runner lock or raise MasterElectionLost""" with closing(self.conn.cursor()) as cr: diff --git a/queue_job/readme/CONFIGURE.md b/queue_job/readme/CONFIGURE.md index f76ddf0a66..4330a6be97 100644 --- a/queue_job/readme/CONFIGURE.md +++ b/queue_job/readme/CONFIGURE.md @@ -103,6 +103,18 @@ http_auth_user = jobrunner http_auth_password = s3cr3t ``` +- The job runner holds its lock on a database connection that stays open. To + detect this connection being dropped by the network (firewall, NAT timeout, + database failover), TCP keepalives and a TCP user timeout are enabled on it, + on the client and on the server side. They can be adjusted with the + following keys of the `[queue_job]` section, or the corresponding + `ODOO_QUEUE_JOB_JOBRUNNER_DB_*` environment variables: + - `jobrunner_db_keepalives`, default `1`, `0` disables all of them + - `jobrunner_db_keepalives_idle`, default `30` (seconds) + - `jobrunner_db_keepalives_interval`, default `10` (seconds) + - `jobrunner_db_keepalives_count`, default `3` + - `jobrunner_db_tcp_user_timeout`, default `60000` (milliseconds) + - Odoo has to be started with `queue_job` as server-wide module, either using the command line option `--load=web,queue_job`, either by setting it in the Odoo configuration file, and `--workers` greater than 1.[^1] diff --git a/queue_job/readme/newsfragments/984.bugfix b/queue_job/readme/newsfragments/984.bugfix new file mode 100644 index 0000000000..eacc8e118a --- /dev/null +++ b/queue_job/readme/newsfragments/984.bugfix @@ -0,0 +1,3 @@ +The job runner detects a database connection dropped by the network, and the +database releases the job runner lock. Before, the job runner stopped +dispatching jobs without any error until it was restarted. diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 93b5674237..fc87b66130 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -486,3 +486,62 @@ def test_next_wakeup_time_several_db(self): channel_manager_b.notify("db_b", "root", "db_b-0", 0, 0, 10, 200, "pending") channel_manager_c.notify("db_c", "root", "db_b-0", 0, 0, 10, 0, "pending") self.assertEqual(jobrunner.next_wakeup_time(), 200) + + def test_connection_info_keepalive_default(self): + with mock.patch.dict(os.environ, {}, clear=True): + connection_info = runner._connection_info_for("db_a") + for param, value in runner.KEEPALIVE_DEFAULTS.items(): + self.assertEqual(connection_info[param], value) + + def test_connection_info_keepalive_from_env(self): + with mock.patch.dict( + os.environ, + {"ODOO_QUEUE_JOB_JOBRUNNER_DB_KEEPALIVES_IDLE": "120"}, + clear=True, + ): + connection_info = runner._connection_info_for("db_a") + self.assertEqual(connection_info["keepalives_idle"], "120") + self.assertEqual(connection_info["keepalives_count"], 3) + + def test_connection_info_keepalive_from_odoo_config(self): + with ( + mock.patch.dict(os.environ, {}, clear=True), + mock.patch.object( + runner, "queue_job_config", {"jobrunner_db_tcp_user_timeout": "0"} + ), + ): + connection_info = runner._connection_info_for("db_a") + self.assertEqual(connection_info["tcp_user_timeout"], "0") + + def test_connection_info_no_tcp_user_timeout_on_old_libpq(self): + with ( + mock.patch.dict(os.environ, {}, clear=True), + mock.patch.object( + runner.psycopg2.extensions, "libpq_version", return_value=110000 + ), + ): + connection_info = runner._connection_info_for("db_a") + self.assertNotIn("tcp_user_timeout", connection_info) + + def _mock_connection_db(self): + db = runner.Database.__new__(runner.Database) + db.conn = mock.MagicMock() + return db, db.conn.cursor.return_value + + def test_set_server_keepalives(self): + db, cr = self._mock_connection_db() + db._set_server_keepalives(dict(runner.KEEPALIVE_DEFAULTS)) + self.assertEqual( + [call.args[1] for call in cr.execute.call_args_list], + [ + ("tcp_keepalives_idle", "30"), + ("tcp_keepalives_interval", "10"), + ("tcp_keepalives_count", "3"), + ("tcp_user_timeout", "60000"), + ], + ) + + def test_set_server_keepalives_disabled(self): + db, cr = self._mock_connection_db() + db._set_server_keepalives({**runner.KEEPALIVE_DEFAULTS, "keepalives": "0"}) + cr.execute.assert_not_called() From 68ed372788e599da93396a80222c1b7adbcd946e Mon Sep 17 00:00:00 2001 From: Bohdan Holoborodko Date: Wed, 14 Oct 2026 09:47:07 +0200 Subject: [PATCH 2/5] [FIX] queue_job: skip server tcp_user_timeout on PostgreSQL < 12 The tcp_user_timeout setting only exists from PostgreSQL 12: setting it on an older server raised in Database.__init__ and the job runner kept looping on errors without dispatching any job. --- queue_job/jobrunner/runner.py | 6 +++++- queue_job/readme/CONFIGURE.md | 3 ++- queue_job/tests/test_runner_runner.py | 16 ++++++++++++++-- 3 files changed, 21 insertions(+), 4 deletions(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 5616af5e10..6d13bfd5c5 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -58,6 +58,7 @@ "tcp_user_timeout": "tcp_user_timeout", } LIBPQ_TCP_USER_TIMEOUT_VERSION = 120000 +PG_TCP_USER_TIMEOUT_VERSION = 120000 _logger = logging.getLogger(__name__) @@ -255,8 +256,11 @@ def _set_server_keepalives(self, connection_info): """ if not int(connection_info["keepalives"]): return + server_settings = dict(KEEPALIVE_SERVER_SETTINGS) + if self.conn.server_version < PG_TCP_USER_TIMEOUT_VERSION: + server_settings.pop("tcp_user_timeout") with closing(self.conn.cursor()) as cr: - for param, setting in KEEPALIVE_SERVER_SETTINGS.items(): + for param, setting in server_settings.items(): if param in connection_info: cr.execute( "SELECT set_config(%s, %s, false)", diff --git a/queue_job/readme/CONFIGURE.md b/queue_job/readme/CONFIGURE.md index 4330a6be97..4f609af73c 100644 --- a/queue_job/readme/CONFIGURE.md +++ b/queue_job/readme/CONFIGURE.md @@ -113,7 +113,8 @@ http_auth_password = s3cr3t - `jobrunner_db_keepalives_idle`, default `30` (seconds) - `jobrunner_db_keepalives_interval`, default `10` (seconds) - `jobrunner_db_keepalives_count`, default `3` - - `jobrunner_db_tcp_user_timeout`, default `60000` (milliseconds) + - `jobrunner_db_tcp_user_timeout`, default `60000` (milliseconds), ignored + with libpq or PostgreSQL older than 12 - Odoo has to be started with `queue_job` as server-wide module, either using the command line option `--load=web,queue_job`, either by setting it in the diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index fc87b66130..343b35faac 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -523,9 +523,9 @@ def test_connection_info_no_tcp_user_timeout_on_old_libpq(self): connection_info = runner._connection_info_for("db_a") self.assertNotIn("tcp_user_timeout", connection_info) - def _mock_connection_db(self): + def _mock_connection_db(self, server_version=160000): db = runner.Database.__new__(runner.Database) - db.conn = mock.MagicMock() + db.conn = mock.MagicMock(server_version=server_version) return db, db.conn.cursor.return_value def test_set_server_keepalives(self): @@ -545,3 +545,15 @@ def test_set_server_keepalives_disabled(self): db, cr = self._mock_connection_db() db._set_server_keepalives({**runner.KEEPALIVE_DEFAULTS, "keepalives": "0"}) cr.execute.assert_not_called() + + def test_set_server_keepalives_no_tcp_user_timeout_on_old_server(self): + db, cr = self._mock_connection_db(server_version=110000) + db._set_server_keepalives(dict(runner.KEEPALIVE_DEFAULTS)) + self.assertEqual( + [call.args[1] for call in cr.execute.call_args_list], + [ + ("tcp_keepalives_idle", "30"), + ("tcp_keepalives_interval", "10"), + ("tcp_keepalives_count", "3"), + ], + ) From 4c0fc21c0e9534ebe3db240bbd737847b5becb32 Mon Sep 17 00:00:00 2001 From: Bohdan Holoborodko Date: Wed, 14 Oct 2026 09:47:07 +0200 Subject: [PATCH 3/5] [FIX] queue_job: disable tcp_user_timeout with keepalives = 0 The client side tcp_user_timeout was still passed to libpq when the keepalives were disabled, so 0 did not disable all of them as documented. --- queue_job/jobrunner/runner.py | 5 ++++- queue_job/tests/test_runner_runner.py | 13 +++++++++++++ 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 6d13bfd5c5..3916e9d2c7 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -183,7 +183,10 @@ def _connection_info_for(db_name): for p, default in KEEPALIVE_DEFAULTS.items(): connection_info.setdefault(p, default) - if psycopg2.extensions.libpq_version() < LIBPQ_TCP_USER_TIMEOUT_VERSION: + if ( + not int(connection_info["keepalives"]) + or psycopg2.extensions.libpq_version() < LIBPQ_TCP_USER_TIMEOUT_VERSION + ): connection_info.pop("tcp_user_timeout") return connection_info diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 343b35faac..7599b6a7bd 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -523,6 +523,19 @@ def test_connection_info_no_tcp_user_timeout_on_old_libpq(self): connection_info = runner._connection_info_for("db_a") self.assertNotIn("tcp_user_timeout", connection_info) + def test_connection_info_keepalive_disabled(self): + with mock.patch.dict( + os.environ, + { + "ODOO_QUEUE_JOB_JOBRUNNER_DB_KEEPALIVES": "0", + "ODOO_QUEUE_JOB_JOBRUNNER_DB_TCP_USER_TIMEOUT": "30000", + }, + clear=True, + ): + connection_info = runner._connection_info_for("db_a") + self.assertEqual(connection_info["keepalives"], "0") + self.assertNotIn("tcp_user_timeout", connection_info) + def _mock_connection_db(self, server_version=160000): db = runner.Database.__new__(runner.Database) db.conn = mock.MagicMock(server_version=server_version) From 4c54dfe43095731a6e0ac1102fa05f60196330ef Mon Sep 17 00:00:00 2001 From: Bohdan Holoborodko Date: Wed, 14 Oct 2026 09:47:07 +0200 Subject: [PATCH 4/5] [IMP] queue_job: log an ignored jobrunner_db_tcp_user_timeout When configured explicitly with a libpq older than 12, the parameter was dropped silently. --- queue_job/jobrunner/runner.py | 12 ++++++++---- queue_job/tests/test_runner_runner.py | 16 ++++++++++++++++ 2 files changed, 24 insertions(+), 4 deletions(-) diff --git a/queue_job/jobrunner/runner.py b/queue_job/jobrunner/runner.py index 3916e9d2c7..422356a299 100644 --- a/queue_job/jobrunner/runner.py +++ b/queue_job/jobrunner/runner.py @@ -181,12 +181,16 @@ def _connection_info_for(db_name): if cfg: connection_info[p] = cfg + tcp_user_timeout_configured = "tcp_user_timeout" in connection_info for p, default in KEEPALIVE_DEFAULTS.items(): connection_info.setdefault(p, default) - if ( - not int(connection_info["keepalives"]) - or psycopg2.extensions.libpq_version() < LIBPQ_TCP_USER_TIMEOUT_VERSION - ): + if not int(connection_info["keepalives"]): + connection_info.pop("tcp_user_timeout") + elif psycopg2.extensions.libpq_version() < LIBPQ_TCP_USER_TIMEOUT_VERSION: + if tcp_user_timeout_configured: + _logger.warning( + "jobrunner_db_tcp_user_timeout is ignored, it requires libpq >= 12" + ) connection_info.pop("tcp_user_timeout") return connection_info diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 7599b6a7bd..14f9b5d3b4 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -523,6 +523,22 @@ def test_connection_info_no_tcp_user_timeout_on_old_libpq(self): connection_info = runner._connection_info_for("db_a") self.assertNotIn("tcp_user_timeout", connection_info) + def test_connection_info_tcp_user_timeout_configured_on_old_libpq(self): + with ( + mock.patch.dict( + os.environ, + {"ODOO_QUEUE_JOB_JOBRUNNER_DB_TCP_USER_TIMEOUT": "30000"}, + clear=True, + ), + mock.patch.object( + runner.psycopg2.extensions, "libpq_version", return_value=110000 + ), + self.assertLogs(runner._logger, level="WARNING") as logs, + ): + connection_info = runner._connection_info_for("db_a") + self.assertNotIn("tcp_user_timeout", connection_info) + self.assertIn("jobrunner_db_tcp_user_timeout is ignored", logs.output[0]) + def test_connection_info_keepalive_disabled(self): with mock.patch.dict( os.environ, From 2eb2b6248918b1ea37bd9b9e21c607ddc330ace6 Mon Sep 17 00:00:00 2001 From: Bohdan Holoborodko Date: Wed, 14 Oct 2026 09:47:07 +0200 Subject: [PATCH 5/5] [DOC] queue_job: keepalive keys override the database URI --- queue_job/readme/CONFIGURE.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/queue_job/readme/CONFIGURE.md b/queue_job/readme/CONFIGURE.md index 4f609af73c..46541bb863 100644 --- a/queue_job/readme/CONFIGURE.md +++ b/queue_job/readme/CONFIGURE.md @@ -116,6 +116,9 @@ http_auth_password = s3cr3t - `jobrunner_db_tcp_user_timeout`, default `60000` (milliseconds), ignored with libpq or PostgreSQL older than 12 + These keys take precedence over the same parameters given in a + `postgresql://` URI used as database name. + - Odoo has to be started with `queue_job` as server-wide module, either using the command line option `--load=web,queue_job`, either by setting it in the Odoo configuration file, and `--workers` greater than 1.[^1]