[PERF] bus: do not build registry to check session

When there are more databases than what the LRU cache can hold,
the gevent worker struggles to keep notifications flowing. Each
outgoing message requires a registry, which is highly inefficient.

This commit is a first step towards removing the need for registry
when dispatching outgoing message. From now on, checking the session
won't require a registry. The result of `_get_session_token_query_params`
is cached, allowing to compute the session token with a direct query.

Due to this change, dispatching notifications won't upgrade the
`res.device` anymore which is considered acceptable.

Part-of: odoo#235746
Signed-off-by: Julien Castiaux (juc) <[email protected]>

Author: tsm-odoo

addons/bus/session_helpers.py

@@ -0,0 +1,70 @@
+import hashlib
+import hmac
+import time
+
+from odoo.api import Environment
+from odoo.modules.registry import Registry
+from odoo.sql_db import SQL
+from odoo.tools.lru import LRU
+from odoo.tools.misc import consteq
+
+_query_params_by_user = LRU(8192)
+
+
+def _get_session_token_query_params(cr, session):
+    """
+    Retrieve the session token query parameters like
+    `res.users@_get_session_token_query_params`, but with caching to avoid building the
+    full registry. The cache is invalidated when `registry.registry_sequence` has changed.
+    """
+    cache_key = (cr.dbname, session.uid)
+    if cached_value := _query_params_by_user.get(cache_key):
+        cr.execute('SELECT MAX(id) FROM orm_signaling_registry')
+        if cached_value['registry_sequence'] == cr.fetchone()[0]:
+            return cached_value['query_params']
+    Registry(cr.dbname).check_signaling()
+    env = new_env(cr, session)
+    params = env.user._get_session_token_query_params()
+    _query_params_by_user[cache_key] = {
+        'registry_sequence': env.registry.registry_sequence,
+        'query_params': params,
+    }
+    return params
+
+
+def check_session(cr, session):
+    session._delete_old_sessions()
+    if 'deletion_time' in session and session['deletion_time'] <= time.time():
+        return False
+    query_params = _get_session_token_query_params(cr, session)
+    cr.execute(
+        SQL(
+            'SELECT %(select)s FROM %(from)s %(joins)s WHERE %(where)s GROUP BY %(group_by)s',
+            **query_params,
+        ),
+    )
+    if cr.rowcount != 1:
+        return False
+    row = cr.fetchone()
+    key_tuple = tuple(
+        (col.name, row[i]) for i, col in enumerate(cr.description) if row[i] is not None
+    )
+    key = str(key_tuple).encode()
+    token = hmac.new(key, session.sid.encode(), hashlib.sha256).hexdigest()
+    return consteq(token, session.session_token)
+
+
+def new_env(cr, session, *, set_lang=False):
+    """
+    Create a new environment. Make sure the transaction has a `default_env` and
+    if requested, set the language of the user in the context.
+    """
+    uid = session.uid
+    ctx = dict(session.context, lang=None)  # lang is not guaranteed to be correct
+    env = Environment(cr, uid, ctx)
+    if set_lang:
+        lang = env['res.lang']._get_code(ctx['lang'])
+        env = env(context=dict(ctx, lang=lang))
+    if not env.transaction.default_env:
+        env.transaction.default_env = env
+    return env

addons/bus/tests/__init__.py

@@ -6,6 +6,7 @@
 from . import test_ir_websocket
 from . import test_notify
 from . import test_websocket_caryall
+from . import test_websocket_check_session
 from . import test_close_websocket_after_tour
 from . import test_websocket_controller
 from . import test_websocket_rate_limiting

addons/bus/tests/test_websocket_check_session.py

@@ -0,0 +1,44 @@
+import time
+from datetime import timedelta
+from unittest.mock import patch
+
+from freezegun import freeze_time
+
+from odoo.tests import HttpCase, new_test_user
+
+from ..session_helpers import _get_session_token_query_params, check_session
+
+
+class TestWebsocketCheckSession(HttpCase):
+    def test_check_session_deletion_time(self):
+        bob = new_test_user(self.env, "bob", groups="base.group_user")
+        self.authenticate(bob.login, bob.password)
+        with freeze_time() as frozen_time:
+            self.session["deletion_time"] = time.time() + 3600
+            self.assertTrue(check_session(self.env.cr, self.session))
+            frozen_time.tick(delta=timedelta(hours=2))
+            self.assertFalse(check_session(self.env.cr, self.session))
+
+    def test_check_session_token_field_changes(self):
+        bob = new_test_user(self.env, "bob", groups="base.group_user")
+        self.authenticate(bob.login, bob.password)
+        self.assertTrue(check_session(self.env.cr, self.session))
+        bob.password = "bob_new_password"
+        self.assertFalse(check_session(self.env.cr, self.session))
+
+    def test_update_cache_when_registry_changes(self):
+        bob = new_test_user(self.env, "bob", groups="base.group_user")
+        self.authenticate(bob.login, bob.password)
+        bob_query_params = _get_session_token_query_params(self.env.cr, self.session)
+        self.assertIs(
+            bob_query_params, _get_session_token_query_params(self.env.cr, self.session)
+        )
+        jane = new_test_user(self.env, "jane", groups="base.group_user")
+        self.authenticate(jane.login, jane.password)
+        current_registry_sequence = self.env.registry.registry_sequence
+        # Signaling is patched during test, simulate first entry coming from an old registry.
+        with patch.object(self.env.registry, "registry_sequence", current_registry_sequence - 1):
+            jane_query_params = _get_session_token_query_params(self.env.cr, self.session)
+        next_jane_query_params = _get_session_token_query_params(self.env.cr, self.session)
+        self.assertIsNot(jane_query_params, next_jane_query_params)
+        self.assertIs(next_jane_query_params, _get_session_token_query_params(self.env.cr, self.session))

addons/bus/websocket.py

@@ -26,13 +26,13 @@
 from werkzeug.exceptions import BadRequest, HTTPException, ServiceUnavailable
 
 import odoo
-from odoo import api, modules
+from odoo import modules
 from .models.bus import dispatch
+from .session_helpers import check_session, new_env
 from odoo.http import root, Request, Response, SessionExpiredException, get_default_session
 from odoo.modules.registry import Registry
 from odoo.service import model as service_model
 from odoo.service.server import CommonServer
-from odoo.service.security import check_session
 from odoo.tools import config
 
 _logger = logging.getLogger(__name__)
@@ -620,7 +620,7 @@ def _terminate(self):
         dispatch.unsubscribe(self)
         self._trigger_lifecycle_event(LifecycleEvent.CLOSE)
         with acquire_cursor(self._db) as cr:
-            env = self.new_env(cr, self._session)
+            env = new_env(cr, self._session)
             env["ir.websocket"]._on_websocket_closed(self._cookies)
 
     def _handle_control_frame(self, frame):
@@ -697,7 +697,7 @@ def _trigger_lifecycle_event(self, event_type):
         if not self.__event_callbacks[event_type]:
             return
         with closing(acquire_cursor(self._db)) as cr:
-            env = self.new_env(cr, self._session, set_lang=True)
+            env = new_env(cr, self._session, set_lang=True)
             for callback in self.__event_callbacks[event_type]:
                 try:
                     service_model.retrying(functools.partial(callback, env, self), env)
@@ -726,8 +726,7 @@ def _assert_session_validity(self):
         if session.uid is None:
             return
         with acquire_cursor(session.db) as cr:
-            env = self.new_env(cr, session)
-            if not check_session(session, env):
+            if not check_session(cr, session):
                 raise SessionExpiredException()
 
     def _send_control_command(self, command, data=None):
@@ -754,7 +753,7 @@ def _process_control_command(self, command, data):
     def _dispatch_bus_notifications(self):
         self._waiting_for_dispatch = False
         with acquire_cursor(self._session.db) as cr:
-            env = self.new_env(cr, self._session)
+            env = new_env(cr, self._session)
             notifications = env['bus.bus']._poll(
                 self._channels, self._last_notif_sent_id, [n[0] for n in self._notif_history]
             )
@@ -784,23 +783,6 @@ def _dispatch_bus_notifications(self):
                 self._notif_history = self._notif_history[last_index + 1 :]
             self._send(notifications)
 
-    def new_env(self, cr, session, *, set_lang=False):
-        """
-        Create a new environment.
-        Make sure the transaction has a `default_env` and if requested, set the
-        language of the user in the context.
-        """
-        uid = session.uid
-        # lang is not guaranteed to be correct, set None
-        ctx = dict(session.context, lang=None)
-        env = api.Environment(cr, uid, ctx)
-        if set_lang:
-            lang = env['res.lang']._get_code(ctx['lang'])
-            env = env(context=dict(ctx, lang=lang))
-        if not env.transaction.default_env:
-            env.transaction.default_env = env
-        return env
-
 
 class TimeoutManager:
     """
@@ -913,7 +895,7 @@ def serve_websocket_message(self, message):
             raise InvalidDatabaseException() from exc
 
         with closing(acquire_cursor(self.db)) as cr:
-            self.env = self.ws.new_env(cr, self.session, set_lang=True)
+            self.env = new_env(cr, self.session, set_lang=True)
             service_model.retrying(
                 functools.partial(self._serve_ir_websocket, event_name, data),
                 self.env,