Add on-disk topology cache (port of .NET TopologyLocalCache), enabled by default

Persist database and cluster topology to disk on each update and, on the first
topology update, seed from disk when all initial urls are unreachable - for
bootstrap/failover resilience, mirroring the .NET client (Node.js/JVM have no
equivalent). Topology/ServerNode gained to_json/from_json for round-tripping.

Enabled by default to match .NET (DocumentConventions.disable_topology_cache
defaults False). The cache location defaults to a per-user, OS-appropriate
directory (%LOCALAPPDATA%\ravendb\topology on Windows, ~/Library/Caches/ravendb
/topology on macOS, $XDG_CACHE_HOME or ~/.cache/ravendb/topology elsewhere)
rather than the working directory, so a dependent application never has topology
files written into wherever it happens to run.

The test harness disables the cache on the global test.manager store and on
every per-test store, and delete_all_topology_files scrubs the default per-user
dir as well as cwd, so the suite leaves no artifacts.

Author: poissoncorp

ravendb/documents/conventions.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import inspect
+import os
+import sys
 import threading
 from abc import abstractmethod, ABC
 from datetime import timedelta, datetime
@@ -51,6 +53,10 @@ def __init__(self):
 
         # Flags
         self.disable_topology_updates = False
+        # On-disk topology cache (mirrors the .NET client): enabled by default. Topology is persisted to, and
+        # - when the initial urls are unreachable on startup - seeded from topology_cache_location.
+        self.disable_topology_cache = False
+        self.topology_cache_location: Optional[str] = DocumentConventions._default_topology_cache_location()
         self._optimistic_concurrency_mode = None
         # Track which setter the user touched so we can reject mixing them.
         self._use_optimistic_concurrency_was_set = False
@@ -107,6 +113,18 @@ def freeze(self):
     def is_frozen(self):
         return self._frozen
 
+    @staticmethod
+    def _default_topology_cache_location() -> str:
+        # Per-user, OS-appropriate cache directory: stable across restarts and never clutters the directory
+        # the application happens to run from (unlike cwd). Mirrors the intent of .NET's AppContext.BaseDirectory.
+        if sys.platform == "win32":
+            base = os.environ.get("LOCALAPPDATA") or os.path.join(os.path.expanduser("~"), "AppData", "Local")
+        elif sys.platform == "darwin":
+            base = os.path.join(os.path.expanduser("~"), "Library", "Caches")
+        else:
+            base = os.environ.get("XDG_CACHE_HOME") or os.path.join(os.path.expanduser("~"), ".cache")
+        return os.path.join(base, "ravendb", "topology")
+
     def get_python_class_name(self, entity_type: type):
         return self._find_python_class_name(entity_type)
 

ravendb/http/request_executor.py

@@ -39,6 +39,7 @@
 from ravendb.http.raven_command import RavenCommand, RavenCommandResponseType
 from ravendb.http.server_node import ServerNode
 from ravendb.http.topology import Topology, NodeStatus, NodeSelector, CurrentIndexAndNode, UpdateTopologyParameters
+from ravendb.http import topology_local_cache
 from ravendb.serverwide.commands import GetDatabaseTopologyCommand, GetClusterTopologyCommand
 
 from http import HTTPStatus
@@ -416,6 +417,14 @@ def __supply_async():
 
                 self._topology_etag = self._node_selector.topology.etag
 
+                if not self.conventions.disable_topology_cache and self.conventions.topology_cache_location:
+                    topology_local_cache.try_save(
+                        self.conventions.topology_cache_location,
+                        topology_local_cache.server_hash(parameters.node.url, self._database_name),
+                        self._node_selector.topology,
+                        topology_local_cache.DATABASE_TOPOLOGY_EXTENSION,
+                    )
+
                 self._on_topology_updated_invoke(topology)
             except Exception as e:
                 if not self._disposed:
@@ -456,6 +465,16 @@ def __run(errors: list):
 
                     errors.append((url, e))
 
+            # all initial urls unreachable - fall back to the on-disk topology cache when enabled
+            for url in initial_urls:
+                cached_topology = self._try_load_topology_from_cache(url)
+                if cached_topology is not None:
+                    self._node_selector = NodeSelector(cached_topology, self._thread_pool_executor)
+                    self._topology_etag = cached_topology.etag
+                    self.__initialize_update_topology_timer()
+                    self.__topology_taken_from_node = ServerNode(url, self._database_name)
+                    return
+
             topology = Topology(
                 self._topology_etag,
                 (
@@ -476,6 +495,13 @@ def __run(errors: list):
 
         return self._thread_pool_executor.submit(__run, errors)
 
+    def _try_load_topology_from_cache(self, url):
+        return topology_local_cache.try_load(
+            None if self.conventions.disable_topology_cache else self.conventions.topology_cache_location,
+            topology_local_cache.server_hash(url, self._database_name),
+            topology_local_cache.DATABASE_TOPOLOGY_EXTENSION,
+        )
+
     @staticmethod
     def validate_urls(initial_urls: List[str]) -> List[str]:
         # todo: implement validation
@@ -1445,6 +1471,14 @@ def __supply_async():
                 new_topology = Topology(results.etag, nodes)
                 self._topology_etag = results.etag
 
+                if not self.conventions.disable_topology_cache and self.conventions.topology_cache_location:
+                    topology_local_cache.try_save(
+                        self.conventions.topology_cache_location,
+                        topology_local_cache.server_hash(parameters.node.url),
+                        new_topology,
+                        topology_local_cache.CLUSTER_TOPOLOGY_EXTENSION,
+                    )
+
                 if self._node_selector is None:
                     self._node_selector = NodeSelector(new_topology, self._thread_pool_executor)
 
@@ -1469,5 +1503,12 @@ def __supply_async():
 
         return self._thread_pool_executor.submit(__supply_async)
 
+    def _try_load_topology_from_cache(self, url):
+        return topology_local_cache.try_load(
+            None if self.conventions.disable_topology_cache else self.conventions.topology_cache_location,
+            topology_local_cache.server_hash(url),
+            topology_local_cache.CLUSTER_TOPOLOGY_EXTENSION,
+        )
+
     def _throw_exceptions(self, details: str):
         raise RuntimeError(f"Failed to retrieve cluster topology from all known nodes {os.linesep}{details}")

ravendb/http/server_node.py

@@ -47,6 +47,24 @@ def __hash__(self) -> int:
     def last_server_version(self) -> str:
         return self.__last_server_version
 
+    def to_json(self) -> dict:
+        return {
+            "Url": self.url,
+            "Database": self.database,
+            "ClusterTag": self.cluster_tag,
+            "ServerRole": self.server_role.value if self.server_role is not None else None,
+        }
+
+    @classmethod
+    def from_json(cls, json_dict: dict) -> "ServerNode":
+        role = json_dict.get("ServerRole")
+        return cls(
+            json_dict.get("Url"),
+            json_dict.get("Database"),
+            json_dict.get("ClusterTag"),
+            cls.Role(role) if role else None,
+        )
+
     @classmethod
     def create_from(cls, topology: "ClusterTopology"):
         nodes = []

ravendb/http/topology.py

@@ -25,6 +25,16 @@ def __init__(self, etag: int, nodes: List[ServerNode]):
         self.etag = etag
         self.nodes = nodes
 
+    def to_json(self) -> Dict:
+        return {"Etag": self.etag, "Nodes": [node.to_json() for node in (self.nodes or [])]}
+
+    @classmethod
+    def from_json(cls, json_dict: Dict) -> "Topology":
+        return cls(
+            json_dict.get("Etag"),
+            [ServerNode.from_json(node_json) for node_json in (json_dict.get("Nodes") or [])],
+        )
+
 
 class ClusterTopology:
     def __init__(self):

ravendb/http/topology_local_cache.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from typing import Optional
+
+from ravendb.http.topology import Topology
+
+# On-disk topology cache (mirrors the .NET client's TopologyLocalCache). The feature is
+# opt-in: it is active only when DocumentConventions.topology_cache_location is set.
+
+DATABASE_TOPOLOGY_EXTENSION = ".raven-topology"
+CLUSTER_TOPOLOGY_EXTENSION = ".raven-cluster-topology"
+
+
+def server_hash(url: str, database: Optional[str] = None) -> str:
+    key = "{0}{1}".format(url, database if database else "")
+    return hashlib.md5(key.encode("utf-8")).hexdigest()
+
+
+def _path(location: str, topology_hash: str, extension: str) -> str:
+    return os.path.join(location, topology_hash + extension)
+
+
+def try_save(location: Optional[str], topology_hash: str, topology: Topology, extension: str) -> None:
+    # Best-effort: caching failures must never break a topology update.
+    if not location or topology is None:
+        return
+    try:
+        os.makedirs(location, exist_ok=True)
+        with open(_path(location, topology_hash, extension), "w", encoding="utf-8") as stream:
+            json.dump(topology.to_json(), stream)
+    except Exception:
+        pass
+
+
+def try_load(location: Optional[str], topology_hash: str, extension: str) -> Optional[Topology]:
+    if not location:
+        return None
+    try:
+        path = _path(location, topology_hash, extension)
+        if not os.path.isfile(path):
+            return None
+        with open(path, "r", encoding="utf-8") as stream:
+            return Topology.from_json(json.load(stream))
+    except Exception:
+        return None

ravendb/tests/driver/raven_test_driver.py

@@ -24,6 +24,7 @@ def _run_embedded_server_internal(locator: "RavenServerLocator") -> Tuple[Docume
         embedded_server = RavenServerRunner.get_embedded_server(locator)
         store = embedded_server.get_document_store("test.manager")
         store.conventions.disable_topology_updates = True
+        store.conventions.disable_topology_cache = True
 
         return store, embedded_server
 

ravendb/tests/system_tests/test_system_create_topology_files.py

@@ -1,11 +1,15 @@
+import hashlib
+import os
+import unittest
+from shutil import rmtree
+
 from ravendb.documents.store.definition import DocumentStore
+from ravendb.http import topology_local_cache
+from ravendb.http.server_node import ServerNode
+from ravendb.http.topology import Topology
 from ravendb.serverwide.database_record import DatabaseRecord
 from ravendb.serverwide.operations.common import DeleteDatabaseOperation, CreateDatabaseOperation
 from ravendb.tests.test_base import TestBase
-import hashlib
-from shutil import rmtree
-import os
-import unittest
 
 TOPOLOGY_FILES_DIR = os.path.join(os.getcwd(), "topology_files")
 DATABASE = "SystemTest"
@@ -17,17 +21,20 @@ def __init__(self, name):
 
 
 class TestSystemTopologyCreation(TestBase):
+    def _customize_store(self, store: DocumentStore) -> None:
+        # opt in to the on-disk topology cache for this test's stores
+        store.conventions.topology_cache_location = TOPOLOGY_FILES_DIR
+
     def tearDown(self):
-        self.store.maintenance.server.send(DeleteDatabaseOperation(database_name=DATABASE, hard_delete=True))
+        try:
+            self.store.maintenance.server.send(DeleteDatabaseOperation(database_name=DATABASE, hard_delete=True))
+        except Exception:
+            pass
         super(TestSystemTopologyCreation, self).tearDown()
         TestBase.delete_all_topology_files()
         if os.path.exists(TOPOLOGY_FILES_DIR):
-            try:
-                rmtree(TOPOLOGY_FILES_DIR, ignore_errors=True)
-            except OSError as ex:
-                pass
+            rmtree(TOPOLOGY_FILES_DIR, ignore_errors=True)
 
-    @unittest.skip("Topology creation")
     def test_topology_creation(self):
         created = False
         while not created:
@@ -38,26 +45,69 @@ def test_topology_creation(self):
             created = True
         TestBase.wait_for_database_topology(self.store, DATABASE)
 
-        with DocumentStore(urls=self.default_urls, database=DATABASE) as store:
+        # the embedded server uses a random port, so hash the real server url (not the placeholder default_urls)
+        base_url = self.store.urls[0]
+
+        with DocumentStore(urls=self.store.urls, database=DATABASE) as store:
+            store.conventions.topology_cache_location = TOPOLOGY_FILES_DIR
             store.initialize()
             with store.open_session() as session:
                 session.store(Author("Idan"))
                 session.save_changes()
-        topology_hash = hashlib.md5("{0}{1}".format(self.default_urls[0], DATABASE).encode("utf-8")).hexdigest()
 
-        cluster_topology_hash = hashlib.md5("{0}".format(self.default_urls[0]).encode("utf-8")).hexdigest()
+        topology_hash = hashlib.md5("{0}{1}".format(base_url, DATABASE).encode("utf-8")).hexdigest()
+        cluster_topology_hash = hashlib.md5("{0}".format(base_url).encode("utf-8")).hexdigest()
 
+        self.assertTrue(os.path.exists(os.path.join(TOPOLOGY_FILES_DIR, topology_hash + ".raven-topology")))
         self.assertTrue(
-            os.path.exists(os.path.join(TOPOLOGY_FILES_DIR, topology_hash + ".raven-topology"))
-        )  # todo: fix from here - make sure the right folder and files appear
-        self.assertTrue(
-            os.path.exists(
-                os.path.join(
-                    TOPOLOGY_FILES_DIR,
-                    cluster_topology_hash + ".raven-cluster-topology",
-                )
-            )
+            os.path.exists(os.path.join(TOPOLOGY_FILES_DIR, cluster_topology_hash + ".raven-cluster-topology"))
+        )
+
+    def test_topology_cache_round_trip(self):
+        os.makedirs(TOPOLOGY_FILES_DIR, exist_ok=True)
+        topology = Topology(3, [ServerNode("http://localhost:9999", "db1", "B", ServerNode.Role.MEMBER)])
+        topology_hash = topology_local_cache.server_hash("http://localhost:9999", "db1")
+        topology_local_cache.try_save(
+            TOPOLOGY_FILES_DIR, topology_hash, topology, topology_local_cache.DATABASE_TOPOLOGY_EXTENSION
+        )
+
+        loaded = topology_local_cache.try_load(
+            TOPOLOGY_FILES_DIR, topology_hash, topology_local_cache.DATABASE_TOPOLOGY_EXTENSION
         )
+        self.assertIsNotNone(loaded)
+        self.assertEqual(3, loaded.etag)
+        self.assertEqual(1, len(loaded.nodes))
+        node = loaded.nodes[0]
+        self.assertEqual("http://localhost:9999", node.url)
+        self.assertEqual("db1", node.database)
+        self.assertEqual("B", node.cluster_tag)
+        self.assertEqual(ServerNode.Role.MEMBER, node.server_role)
+
+    def test_topology_is_loaded_from_cache_when_urls_unreachable(self):
+        bad_url = "http://127.0.0.1:1"
+        database = "CacheSeedTest"
+        os.makedirs(TOPOLOGY_FILES_DIR, exist_ok=True)
+        cached = Topology(9, [ServerNode(bad_url, database, "A", ServerNode.Role.MEMBER)])
+        topology_local_cache.try_save(
+            TOPOLOGY_FILES_DIR,
+            topology_local_cache.server_hash(bad_url, database),
+            cached,
+            topology_local_cache.DATABASE_TOPOLOGY_EXTENSION,
+        )
+
+        # the server is unreachable, so the first topology update must fall back to the on-disk cache
+        with DocumentStore(urls=[bad_url], database=database) as store:
+            store.conventions.topology_cache_location = TOPOLOGY_FILES_DIR
+            store.initialize()
+            request_executor = store.get_request_executor()
+            request_executor._first_topology_update_task.result(30)
+
+            nodes = request_executor.topology_nodes
+            self.assertEqual(1, len(nodes))
+            self.assertEqual("A", nodes[0].cluster_tag)
+            self.assertEqual(bad_url, nodes[0].url)
+            self.assertEqual(ServerNode.Role.MEMBER, nodes[0].server_role)
+            self.assertEqual(9, request_executor.topology_etag)
 
 
 if __name__ == "__main__":

ravendb/tests/test_base.py

@@ -253,7 +253,9 @@ def _customize_db_record(self, db_record: DatabaseRecord) -> None:
         pass
 
     def _customize_store(self, store: DocumentStore) -> None:
-        pass
+        # Tests don't exercise the on-disk topology cache by default (the dedicated topology test opts in),
+        # so keep the many per-test stores from writing files into the user's cache directory.
+        store.conventions.disable_topology_cache = True
 
     @property
     def secured_document_store(self) -> DocumentStore:
@@ -336,10 +338,19 @@ def _discard_embedded_server(cls, secured: bool) -> None:
     @staticmethod
     def delete_all_topology_files():
         import os
+        from ravendb.documents.conventions import DocumentConventions
 
-        file_list = [f for f in os.listdir(".") if f.endswith("topology")]
-        for f in file_list:
-            os.remove(f)
+        # clean both the working dir and the default per-user topology-cache dir so tests leave no artifacts
+        for directory in {".", DocumentConventions._default_topology_cache_location()}:
+            try:
+                for f in os.listdir(directory):
+                    if f.endswith("topology"):
+                        try:
+                            os.remove(os.path.join(directory, f))
+                        except OSError:
+                            pass
+            except OSError:
+                pass
 
     @staticmethod
     def wait_for_database_topology(store, database_name, replication_factor=1):