HomeDB cache: mixed cluster (#648)

* Mixed cluster tests:
  * old & new protocol version mixed
  * with & without connection hint mixed
* tests driver maintains cache when ssr is disabled
* Feature flag for home database cache

Co-authored-by: MaxAake <[email protected]>
Co-authored-by: Dmitriy Tverdiakov <[email protected]>
Co-authored-by: Dmitriy Tverdiakov <[email protected]>

Author: 3 people

nutkit/frontend/driver.py

@@ -15,6 +15,7 @@ def __init__(self, backend, uri, auth_token, user_agent=None,
                  connection_timeout_ms=None, fetch_size=None,
                  max_tx_retry_time_ms=None, encrypted=None,
                  trusted_certificates=None, liveness_check_timeout_ms=None,
+                 max_connection_lifetime_ms=None,
                  max_connection_pool_size=None,
                  connection_acquisition_timeout_ms=None,
                  notifications_min_severity=None,
@@ -58,6 +59,7 @@ def __init__(self, backend, uri, auth_token, user_agent=None,
             fetchSize=fetch_size, maxTxRetryTimeMs=max_tx_retry_time_ms,
             encrypted=encrypted, trustedCertificates=trusted_certificates,
             liveness_check_timeout_ms=liveness_check_timeout_ms,
+            max_connection_lifetime_ms=max_connection_lifetime_ms,
             max_connection_pool_size=max_connection_pool_size,
             connection_acquisition_timeout_ms=connection_acquisition_timeout_ms,  # noqa: E501
             notifications_min_severity=notifications_min_severity,

nutkit/protocol/feature.py

@@ -27,6 +27,9 @@ class Feature(Enum):
     # The driver offers a method for driver objects to report if they were
     # configured with a or without encryption.
     API_DRIVER_IS_ENCRYPTED = "Feature:API:Driver.IsEncrypted"
+    # The driver supports setting a custom max connection lifetime
+    API_DRIVER_MAX_CONNECTION_LIFETIME = \
+        "Feature:API:Driver:MaxConnectionLifetime"
     # The driver supports notification filters configuration.
     API_DRIVER_NOTIFICATIONS_CONFIG = "Feature:API:Driver:NotificationsConfig"
     # The driver offers a method for checking if the provided authentication
@@ -169,6 +172,10 @@ class Feature(Enum):
     # sending BEGIN but pipelines the RUN and PULL right afterwards and
     # consumes three messages after that. This saves 2 full round-trips.
     OPT_EXECUTE_QUERY_PIPELINING = "Optimization:ExecuteQueryPipelining"
+    # The driver implements a cache to match users to their most recently
+    # resolved home database, routing requests with no set database to this
+    # cached database if all open connections have an SSR connection hint.
+    OPT_HOME_DB_CACHE = "Optimization:HomeDatabaseCache"
     # The home db cache for optimistic home db resolution treats the principal
     # in basic auth the exact same way it treats impersonated users.
     OPT_HOME_DB_CACHE_BASIC_PRINCIPAL_IS_IMP_USER = \

nutkit/protocol/requests.py

@@ -71,7 +71,8 @@ def __init__(
         domainNameResolverRegistered=False, connectionTimeoutMs=None,
         fetchSize=None, maxTxRetryTimeMs=None,
         encrypted=None, trustedCertificates=None,
-        liveness_check_timeout_ms=None, max_connection_pool_size=None,
+        liveness_check_timeout_ms=None, max_connection_lifetime_ms=None,
+        max_connection_pool_size=None,
         connection_acquisition_timeout_ms=None,
         notifications_min_severity=None,
         notifications_disabled_categories=None,
@@ -92,6 +93,8 @@ def __init__(
         self.fetchSize = fetchSize
         self.maxTxRetryTimeMs = maxTxRetryTimeMs
         self.livenessCheckTimeoutMs = liveness_check_timeout_ms
+        if max_connection_lifetime_ms is not None:
+            self.maxConnectionLifetimeMs = max_connection_lifetime_ms
         self.maxConnectionPoolSize = max_connection_pool_size
         self.connectionAcquisitionTimeoutMs = connection_acquisition_timeout_ms
         assert (client_certificate is None

tests/shared.py

@@ -16,6 +16,7 @@
 import os
 import re
 import socket
+import time
 import unittest
 import warnings
 from contextlib import contextmanager
@@ -24,6 +25,7 @@
 
 from nutkit import protocol
 from nutkit.backend import Backend
+from nutkit.frontend import FakeTime
 
 
 def get_backend_host_and_port():
@@ -255,3 +257,35 @@ class Potential(enum.Enum):
     NO = 0.0
     MAYBE = 0.5
     # CAN_YOU_REPEAT_THE_QUESTION = "?"
+
+
+class TimeoutManager:
+    def __init__(
+        self, test_case: TestkitTestCase, timeout_ms: int,
+        use_real_timers: bool = False
+    ):
+        self._timeout_ms = timeout_ms
+        self._fake_time = None
+        if test_case.driver_supports_features(
+            protocol.Feature.BACKEND_MOCK_TIME
+        ) and not use_real_timers:
+            self._fake_time = FakeTime(test_case._backend)
+
+    def __enter__(self):
+        if self._fake_time:
+            self._fake_time.__enter__()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self._fake_time:
+            self._fake_time.__exit__(exc_type, exc_val, exc_tb)
+
+    def tick_to_before_timeout(self):
+        if self._fake_time:
+            self._fake_time.tick(self._timeout_ms - 1)
+
+    def tick_to_after_timeout(self):
+        if self._fake_time:
+            self._fake_time.tick(self._timeout_ms + 1)
+        else:
+            time.sleep(self._timeout_ms / 1000)

tests/stub/driver_parameters/test_liveness_check.py

@@ -1,4 +1,3 @@
-import time
 from contextlib import contextmanager
 
 import nutkit.protocol as types
@@ -9,42 +8,11 @@
 from tests.shared import (
     driver_feature,
     TestkitTestCase,
+    TimeoutManager,
 )
 from tests.stub.shared import StubServer
 
 
-class TimeoutManager:
-    def __init__(
-        self, test_case: TestkitTestCase, timeout_ms: int,
-        use_real_timers: bool = False
-    ):
-        self._timeout_ms = timeout_ms
-        self._fake_time = None
-        if test_case.driver_supports_features(
-            types.Feature.BACKEND_MOCK_TIME
-        ) and not use_real_timers:
-            self._fake_time = FakeTime(test_case._backend)
-
-    def __enter__(self):
-        if self._fake_time:
-            self._fake_time.__enter__()
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        if self._fake_time:
-            self._fake_time.__exit__(exc_type, exc_val, exc_tb)
-
-    def tick_to_before_timeout(self):
-        if self._fake_time:
-            self._fake_time.tick(self._timeout_ms - 1)
-
-    def tick_to_after_timeout(self):
-        if self._fake_time:
-            self._fake_time.tick(self._timeout_ms + 1)
-        else:
-            time.sleep(self._timeout_ms / 1000)
-
-
 class TestLivenessCheck(TestkitTestCase):
     required_features = (
         types.Feature.BOLT_5_4,

tests/stub/homedb/scripts/mixed/reader_5x7_keep_warm.script

@@ -0,0 +1,19 @@
+!: BOLT 5.7
+
+A: HELLO {"{}": "*"}
+A: LOGON {"{}": "*"}
+*: RESET
+
+C: RUN {"U": "*"} {} {"db": "homedb1", "[bookmarks]": {"[]": "*"}, "mode": "r"}
+S: SUCCESS {"fields": ["n"]}
+{{
+    C: PULL {"n": {"Z": "*"}, "[qid]": -1}
+    S: RECORD [1]
+----
+    C: DISCARD {"n": {"Z": "*"}, "[qid]": -1}
+}}
+S: SUCCESS {}
+*: RESET
+
+
+?: GOODBYE

tests/stub/homedb/scripts/mixed/reader_5x8_keep_warm.script

@@ -0,0 +1,29 @@
+!: BOLT 5.8
+
+A: HELLO {"{}": "*"}
+A: LOGON {"{}": "*"}
+*: RESET
+
+C: RUN {"U": "*"} {} {"db": "homedb1", "[bookmarks]": {"[]": "*"}, "mode": "r", "imp_user": "user2"}
+S: SUCCESS {"fields": ["n"]}
+{{
+    C: PULL {"n": {"Z": "*"}, "[qid]": -1}
+    S: RECORD [1]
+----
+    C: DISCARD {"n": {"Z": "*"}, "[qid]": -1}
+}}
+S: SUCCESS {}
+*: RESET
+
+C: RUN {"U": "*"} {} {"[bookmarks]": {"[]": "*"}, "mode": "r"}
+S: SUCCESS {"fields": ["n"]}
+{{
+    C: PULL {"n": {"Z": "*"}, "[qid]": -1}
+    S: RECORD [1]
+----
+    C: DISCARD {"n": {"Z": "*"}, "[qid]": -1}
+}}
+S: SUCCESS {}
+*: RESET
+
+?: GOODBYE

tests/stub/homedb/scripts/mixed/reader_5x8_ssr.script

@@ -0,0 +1,20 @@
+!: BOLT 5.8
+
+A: HELLO {"{}": "*"}
+A: LOGON {"{}": "*"}
+*: RESET
+
+{+
+    C: RUN {"U": "*"} {} {"db": "homedb1", "[bookmarks]": {"[]": "*"}, "mode": "r"}
+    S: SUCCESS {"fields": ["n"]}
+    {{
+        C: PULL {"n": {"Z": "*"}, "[qid]": -1}
+        S: RECORD [1]
+    ----
+        C: DISCARD {"n": {"Z": "*"}, "[qid]": -1}
+    }}
+    S: SUCCESS {}
+    *: RESET
++}
+
+?: GOODBYE

tests/stub/homedb/scripts/mixed/router_5x8.script

@@ -0,0 +1,19 @@
+!: BOLT 5.8
+!: ALLOW CONCURRENT
+
+A: HELLO {"{}": "*"}
+A: LOGON {"{}": "*"}
+*: RESET
+
+
+# explicit resolution (1st connection)
+C: ROUTE "*" "*" "*"
+S: SUCCESS { "rt": { "ttl": 1000000, "db": "homedb1", "servers": [{"addresses": ["#HOST#:9000"], "role": "ROUTE"}, {"addresses": ["#HOST#:9010"], "role": "READ"}, {"addresses": ["#HOST#:9020"], "role": "WRITE"}]}}
+*: RESET
+
+# explicit resolution (fallback connection)
+C: ROUTE "*" "*" "*"
+S: SUCCESS { "rt": { "ttl": 1000000, "db": "homedb2", "servers": [{"addresses": ["#HOST#:9000"], "role": "ROUTE"}, {"addresses": ["#HOST#:9010"], "role": "READ"}, {"addresses": ["#HOST#:#WRITER_2#"], "role": "WRITE"}]}}
+*: RESET
+
+?: GOODBYE

tests/stub/homedb/scripts/mixed/router_keep_warm.script

@@ -0,0 +1,10 @@
+!: BOLT #BOLT_VERSION#
+!: ALLOW CONCURRENT
+
+A: HELLO {"{}": "*"}
+A: LOGON {"{}": "*"}
+*: RESET
+
+C: ROUTE {"{}": "*"} [] {"[db]": null#IMPERSONATED_USER#}
+S: SUCCESS { "rt": { "ttl": 1000000, "db": "homedb1", "servers": [{"addresses": ["#HOST#:9000"], "role": "ROUTE"}, {"addresses": ["#HOST#:9010"], "role": "READ"}, {"addresses": [], "role": "WRITE"}]}}
+*: RESET