From 61017008686600693e75c09a6b4aea98a16f677f Mon Sep 17 00:00:00 2001 From: "Victor Piryns (pivi)" Date: Wed, 27 Mar 2024 06:49:53 +0000 Subject: [PATCH] [FIX] bus: split large `NOTIFY` payloads Description: On a database with heavy activity, it is possible for the payload size we pass for the `NOTIFY` query on the bus to be larger than what PostgreSQL allows by default, 8000 B. This limit is defined as: https://github.com/postgres/postgres/blob/6686e9676c8faff4ee04c1574e117ae38f117efa/src/backend/commands/async.c#L158-L166 and it is used to compare the size of the payload before processing: https://github.com/postgres/postgres/blob/6686e9676c8faff4ee04c1574e117ae38f117efa/src/backend/commands/async.c#L654-L657 Fix: - Binary split the payload into multiple chunks if it's exceeding the above-mentioned limit. The unit of splitting is a channel, so if the content of one channel itself is larger than said limit, it is not handled, but shouldn't occur under normal circumstances. - Introduce also a new ENV variable `ODOO_NOTIFY_PAYLOAD_MAX_LENGTH` to allow tweaking of the limit if they are running a handrolled custom PostgreSQL cluster. Reference: opw-3650618 closes odoo/odoo#159469 X-original-commit: 61c7bba44990522d853bae25aa7899cb03c72e31 Signed-off-by: Xavier Morel (xmo) Signed-off-by: Piryns Victor (pivi) --- addons/bus/models/bus.py | 43 ++++++++++++++++++++++++++++- addons/bus/tests/__init__.py | 1 + addons/bus/tests/test_notify.py | 49 +++++++++++++++++++++++++++++++++ 3 files changed, 92 insertions(+), 1 deletion(-) create mode 100644 addons/bus/tests/test_notify.py diff --git a/addons/bus/models/bus.py b/addons/bus/models/bus.py index 17e68d1c8f6..d3a3d8b2402 100644 --- a/addons/bus/models/bus.py +++ b/addons/bus/models/bus.py @@ -3,6 +3,7 @@ import contextlib import datetime import json import logging +import math import os import random import selectors @@ -24,6 +25,21 @@ TIMEOUT = 50 # custom function to call instead of default PostgreSQL's `pg_notify` ODOO_NOTIFY_FUNCTION = os.getenv('ODOO_NOTIFY_FUNCTION', 'pg_notify') + +def get_notify_payload_max_length(default=8000): + try: + length = int(os.environ.get('ODOO_NOTIFY_PAYLOAD_MAX_LENGTH', default)) + except ValueError: + _logger.warning("ODOO_NOTIFY_PAYLOAD_MAX_LENGTH has to be an integer, " + "defaulting to %d bytes", default) + length = default + return length + + +# max length in bytes for the NOTIFY query payload +NOTIFY_PAYLOAD_MAX_LENGTH = get_notify_payload_max_length() + + #---------------------------------------------------------- # Bus #---------------------------------------------------------- @@ -46,6 +62,26 @@ def channel_with_db(dbname, channel): return channel +def get_notify_payloads(channels): + """ + Generates the json payloads for the imbus NOTIFY. + Splits recursively payloads that are too large. + + :param list channels: + :return: list of payloads of json dumps + :rtype: list[str] + """ + if not channels: + return [] + payload = json_dump(channels) + if len(channels) == 1 or len(payload.encode()) < NOTIFY_PAYLOAD_MAX_LENGTH: + return [payload] + else: + pivot = math.ceil(len(channels) / 2) + return (get_notify_payloads(channels[:pivot]) + + get_notify_payloads(channels[pivot:])) + + class ImBus(models.Model): _name = 'bus.bus' @@ -84,7 +120,12 @@ class ImBus(models.Model): def notify(): with odoo.sql_db.db_connect('postgres').cursor() as cr: query = sql.SQL("SELECT {}('imbus', %s)").format(sql.Identifier(ODOO_NOTIFY_FUNCTION)) - cr.execute(query, (json_dump(list(channels)), )) + payloads = get_notify_payloads(list(channels)) + if len(payloads) > 1: + _logger.info("The imbus notification payload was too large, " + "it's been split into %d payloads.", len(payloads)) + for payload in payloads: + cr.execute(query, (payload,)) @api.model def _sendone(self, channel, notification_type, message): diff --git a/addons/bus/tests/__init__.py b/addons/bus/tests/__init__.py index f8f91a9b5ef..ad34feffec9 100644 --- a/addons/bus/tests/__init__.py +++ b/addons/bus/tests/__init__.py @@ -3,6 +3,7 @@ from . import test_assetsbundle from . import test_health from . import test_ir_model from . import test_ir_websocket +from . import test_notify from . import test_websocket_caryall from . import test_websocket_controller from . import test_websocket_rate_limiting diff --git a/addons/bus/tests/test_notify.py b/addons/bus/tests/test_notify.py new file mode 100644 index 00000000000..3fd64376bb1 --- /dev/null +++ b/addons/bus/tests/test_notify.py @@ -0,0 +1,49 @@ +# Part of Odoo. See LICENSE file for full copyright and licensing details. + +from odoo.tests import BaseCase + +from ..models.bus import json_dump, get_notify_payloads, NOTIFY_PAYLOAD_MAX_LENGTH + + +class NotifyTests(BaseCase): + + def test_get_notify_payloads(self): + """ + Asserts that the implementation of `get_notify_payloads` + actually splits correctly large payloads + """ + def check_payloads_size(payloads): + for payload in payloads: + self.assertLess(len(payload.encode()), NOTIFY_PAYLOAD_MAX_LENGTH) + + channel = ('dummy_db', 'dummy_model', 12345) + channels = [channel] + self.assertLess(len(json_dump(channels).encode()), NOTIFY_PAYLOAD_MAX_LENGTH) + payloads = get_notify_payloads(channels) + self.assertEqual(len(payloads), 1, + "The payload is less then the threshold, " + "there should be 1 payload only, as it shouldn't be split") + channels = [channel] * 100 + self.assertLess(len(json_dump(channels).encode()), NOTIFY_PAYLOAD_MAX_LENGTH) + payloads = get_notify_payloads(channels) + self.assertEqual(len(payloads), 1, + "The payload is less then the threshold, " + "there should be 1 payload only, as it shouldn't be split") + check_payloads_size(payloads) + channels = [channel] * 1000 + self.assertGreaterEqual(len(json_dump(channels).encode()), NOTIFY_PAYLOAD_MAX_LENGTH) + payloads = get_notify_payloads(channels) + self.assertGreater(len(payloads), 1, + "Payload was larger than the threshold, it should've been split") + check_payloads_size(payloads) + + fat_channel = tuple(item * 1000 for item in channel) + channels = [fat_channel] + self.assertEqual(len(channels), 1, "There should be only 1 channel") + self.assertGreaterEqual(len(json_dump(channels).encode()), NOTIFY_PAYLOAD_MAX_LENGTH) + payloads = get_notify_payloads(channels) + self.assertEqual(len(payloads), 1, + "Payload was larger than the threshold, but shouldn't be split, " + "as it contains only 1 channel") + with self.assertRaises(AssertionError): + check_payloads_size(payloads)