Accept injected ClusterClient or NodeStore on ConnectionPool

ConnectionPool constructed its own ClusterClient from raw addresses
with no way to share one across pools. Multi-database apps, persistent
node stores, and testability all suffered.

Add optional cluster= and node_store= kwargs; keep addresses= the
default path. When cluster= is provided, the pool reuses its node
store and (future) leader cache. Validate that callers pass exactly
one source.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqliteclient/__init__.py

@@ -63,21 +63,27 @@ async def connect(
 
 
 async def create_pool(
-    addresses: list[str],
+    addresses: list[str] | None = None,
     *,
     database: str = "default",
     min_size: int = 1,
     max_size: int = 10,
     timeout: float = 10.0,
+    cluster: ClusterClient | None = None,
+    node_store: NodeStore | None = None,
 ) -> ConnectionPool:
     """Create a connection pool with automatic leader detection.
 
     Args:
-        addresses: List of node addresses in "host:port" format
+        addresses: List of node addresses in "host:port" format. Ignored if
+            ``cluster`` or ``node_store`` is provided.
         database: Database name to open
         min_size: Minimum number of connections to maintain
         max_size: Maximum number of connections
         timeout: Connection timeout in seconds
+        cluster: Externally-owned ClusterClient shared across pools.
+        node_store: Externally-owned NodeStore used to build a new
+            ClusterClient. Mutually exclusive with ``cluster``.
 
     Returns:
         An initialized ConnectionPool
@@ -88,6 +94,8 @@ async def create_pool(
         min_size=min_size,
         max_size=max_size,
         timeout=timeout,
+        cluster=cluster,
+        node_store=node_store,
     )
     await pool.initialize()
     return pool

src/dqliteclient/pool.py

@@ -9,6 +9,7 @@
 from dqliteclient.cluster import ClusterClient
 from dqliteclient.connection import DqliteConnection
 from dqliteclient.exceptions import DqliteConnectionError
+from dqliteclient.node_store import NodeStore
 
 
 def _socket_looks_dead(conn: DqliteConnection) -> bool:
@@ -48,21 +49,29 @@ class ConnectionPool:
 
     def __init__(
         self,
-        addresses: list[str],
+        addresses: list[str] | None = None,
         *,
         database: str = "default",
         min_size: int = 1,
         max_size: int = 10,
         timeout: float = 10.0,
+        cluster: ClusterClient | None = None,
+        node_store: NodeStore | None = None,
     ) -> None:
         """Initialize connection pool.
 
         Args:
-            addresses: List of node addresses
+            addresses: List of node addresses. Ignored if ``cluster`` or
+                ``node_store`` is provided; required otherwise.
             database: Database name
             min_size: Minimum connections to maintain
             max_size: Maximum connections allowed
             timeout: Connection timeout
+            cluster: Externally-owned ClusterClient. Lets multiple pools
+                share one ClusterClient (and thus its node store, leader
+                cache, etc.) across databases or tenants.
+            node_store: Externally-owned NodeStore used to build a new
+                ClusterClient. Mutually exclusive with ``cluster``.
         """
         if min_size < 0:
             raise ValueError(f"min_size must be non-negative, got {min_size}")
@@ -72,14 +81,23 @@ def __init__(
             raise ValueError(f"min_size ({min_size}) must not exceed max_size ({max_size})")
         if timeout <= 0:
             raise ValueError(f"timeout must be positive, got {timeout}")
+        if cluster is not None and node_store is not None:
+            raise ValueError("pass only one of cluster= or node_store=")
+        if cluster is None and node_store is None and not addresses:
+            raise ValueError("pass one of addresses, cluster, or node_store")
 
-        self._addresses = addresses
+        self._addresses = addresses or []
         self._database = database
         self._min_size = min_size
         self._max_size = max_size
         self._timeout = timeout
 
-        self._cluster = ClusterClient.from_addresses(addresses, timeout=timeout)
+        if cluster is not None:
+            self._cluster = cluster
+        elif node_store is not None:
+            self._cluster = ClusterClient(node_store, timeout=timeout)
+        else:
+            self._cluster = ClusterClient.from_addresses(self._addresses, timeout=timeout)
         self._pool: asyncio.Queue[DqliteConnection] = asyncio.Queue(maxsize=max_size)
         self._size = 0
         self._lock = asyncio.Lock()

tests/test_pool.py

@@ -873,6 +873,39 @@ async def slow_connect(**kwargs):
 
         await pool.close()
 
+    async def test_pool_accepts_injected_cluster_client(self) -> None:
+        """Callers must be able to share a ClusterClient across multiple pools."""
+        from dqliteclient.cluster import ClusterClient
+
+        shared_cluster = ClusterClient.from_addresses(["localhost:9001"])
+        pool_a = ConnectionPool(cluster=shared_cluster, min_size=0, max_size=3)
+        pool_b = ConnectionPool(cluster=shared_cluster, min_size=0, max_size=3)
+        assert pool_a._cluster is shared_cluster
+        assert pool_b._cluster is shared_cluster
+
+    async def test_pool_accepts_injected_node_store(self) -> None:
+        """Callers with a persistent NodeStore must be able to thread it in."""
+        from dqliteclient.node_store import MemoryNodeStore
+
+        store = MemoryNodeStore(["localhost:9001", "localhost:9002"])
+        pool = ConnectionPool(node_store=store, min_size=0, max_size=1)
+        nodes = await pool._cluster._node_store.get_nodes()
+        assert [n.address for n in nodes] == ["localhost:9001", "localhost:9002"]
+
+    async def test_pool_requires_some_cluster_source(self) -> None:
+        """Constructing with neither addresses nor cluster/node_store must raise."""
+        with pytest.raises(ValueError, match="addresses.*cluster.*node_store"):
+            ConnectionPool()
+
+    async def test_pool_rejects_cluster_and_node_store_together(self) -> None:
+        """Passing both cluster= and node_store= must raise — pick one."""
+        from dqliteclient.cluster import ClusterClient
+        from dqliteclient.node_store import MemoryNodeStore
+
+        cluster = ClusterClient.from_addresses(["localhost:9001"])
+        store = MemoryNodeStore(["localhost:9001"])
+        with pytest.raises(ValueError, match="only one"):
+            ConnectionPool(cluster=cluster, node_store=store)
 
     async def test_reset_connection_returns_false_on_cancelled_error(self) -> None:
         """_reset_connection must return False (not raise) when ROLLBACK is cancelled."""