Add opt-in on-disk topology cache (port of .NET TopologyLocalCache)

The topology-file cache is absent from the nodejs/jvm clients; ported from the
.NET client's TopologyLocalCache design. Opt-in via the new
DocumentConventions.topology_cache_location (default None = disabled), so
existing behavior is unchanged for all other stores/tests.

- New ravendb/http/topology_local_cache.py: server_hash + try_save/try_load for
  DB (.raven-topology) and cluster (.raven-cluster-topology) topology JSON.
- RequestExecutor: persist topology after each update (DB + cluster) and seed
  from cache before the "!" placeholder when all initial urls are unreachable;
  ClusterRequestExecutor overrides the loader to use the cluster file.
- Unskip test_system_create_topology_files; enable the cache via _customize_store
  and hash the real (random-port) server url rather than the 8080 placeholder.

Author: poissoncorp

ravendb/documents/conventions.py

@@ -51,6 +51,8 @@ def __init__(self):
 
         # Flags
         self.disable_topology_updates = False
+        # Opt-in on-disk topology cache: directory to persist/read topology files, or None to disable.
+        self.topology_cache_location: Optional[str] = None
         self._optimistic_concurrency_mode = None
         # Track which setter the user touched so we can reject mixing them.
         self._use_optimistic_concurrency_was_set = False

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 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,15 @@ 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.__initialize_update_topology_timer()
+                    self.__topology_taken_from_node = ServerNode(url, self._database_name)
+                    return
+
             topology = Topology(
                 self._topology_etag,
                 (
@@ -476,6 +494,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(
+            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 +1470,14 @@ def __supply_async():
                 new_topology = Topology(results.etag, nodes)
                 self._topology_etag = results.etag
 
+                if 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 +1502,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(
+            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/topology_local_cache.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from typing import Optional
+
+from ravendb.http.server_node import ServerNode
+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 _topology_to_json(topology: Topology) -> dict:
+    return {
+        "Etag": topology.etag,
+        "Nodes": [
+            {
+                "Url": node.url,
+                "Database": node.database,
+                "ClusterTag": node.cluster_tag,
+                "ServerRole": node.server_role.value if node.server_role is not None else None,
+            }
+            for node in (topology.nodes or [])
+        ],
+    }
+
+
+def _topology_from_json(data: dict) -> Topology:
+    nodes = []
+    for node_json in data.get("Nodes") or []:
+        role = node_json.get("ServerRole")
+        nodes.append(
+            ServerNode(
+                node_json.get("Url"),
+                node_json.get("Database"),
+                node_json.get("ClusterTag"),
+                ServerNode.Role(role) if role else None,
+            )
+        )
+    return Topology(data.get("Etag"), nodes)
+
+
+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(topology), 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/system_tests/test_system_create_topology_files.py

@@ -1,11 +1,12 @@
+import hashlib
+import os
+import unittest
+from shutil import rmtree
+
 from ravendb.documents.store.definition import DocumentStore
 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 +18,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,25 +42,22 @@ 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"))
         )