Forward max_attempts from ConnectionPool / create_pool to cluster.connect

ClusterClient.connect already accepts max_attempts (default 3), but
the pool surface didn't expose it — pool users were stuck on the
default unless they constructed a custom ClusterClient and passed it
via cluster=. Operators with known-flaky networks who want to absorb
more transient leader-flips inside a single acquire() call had no
clean way to do it.

Add max_attempts as an additive keyword-only parameter on
ConnectionPool.__init__ and create_pool, validate (>= 1 if not None),
and forward to the underlying cluster.connect call.

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

Author: antoineleclair

src/dqliteclient/__init__.py

@@ -118,6 +118,7 @@ async def create_pool(
     max_continuation_frames: int | None = _DEFAULT_MAX_CONTINUATION_FRAMES,
     trust_server_heartbeat: bool = False,
     close_timeout: float = 0.5,
+    max_attempts: int | None = None,
 ) -> ConnectionPool:
     """Create a connection pool with automatic leader detection.
 
@@ -148,6 +149,10 @@ async def create_pool(
             deployments where FIN/ACK round-trip is slower, or
             decrease to tighten SIGTERM-shutdown budgets. See
             ``DqliteConnection.__init__`` for full rationale.
+        max_attempts: Maximum leader-discovery attempts per pool
+            connect (forwarded to ``ClusterClient.connect``). ``None``
+            (default) uses the cluster client's default of 3. Must be
+            ``>= 1`` if not ``None``.
 
     Returns:
         An initialized ConnectionPool
@@ -164,6 +169,7 @@ async def create_pool(
         max_continuation_frames=max_continuation_frames,
         trust_server_heartbeat=trust_server_heartbeat,
         close_timeout=close_timeout,
+        max_attempts=max_attempts,
     )
     await pool.initialize()
     return pool

src/dqliteclient/pool.py

@@ -131,6 +131,7 @@ def __init__(
         max_continuation_frames: int | None = _DEFAULT_MAX_CONTINUATION_FRAMES,
         trust_server_heartbeat: bool = False,
         close_timeout: float = 0.5,
+        max_attempts: int | None = None,
     ) -> None:
         """Initialize connection pool.
 
@@ -192,13 +193,23 @@ def __init__(
                 round-trip is slower, or decrease to tighten
                 SIGTERM-shutdown budgets. See
                 ``DqliteConnection.__init__`` for full rationale.
+            max_attempts: Maximum leader-discovery attempts per
+                ``_create_connection`` (forwarded to
+                ``ClusterClient.connect``). ``None`` (default) uses the
+                cluster client's default of 3 — covers one leader
+                change plus one transport hiccup. Increase only with
+                eyes open: hiding genuine cluster instability behind
+                a long retry loop just delays the diagnosis. Must be
+                ``>= 1`` if not ``None``.
         """
         if min_size < 0:
             raise ValueError(f"min_size must be non-negative, got {min_size}")
         if max_size < 1:
             raise ValueError(f"max_size must be at least 1, got {max_size}")
         if min_size > max_size:
             raise ValueError(f"min_size ({min_size}) must not exceed max_size ({max_size})")
+        if max_attempts is not None and max_attempts < 1:
+            raise ValueError(f"max_attempts must be at least 1 if provided, got {max_attempts}")
         _validate_timeout(timeout)
         _validate_timeout(close_timeout, name="close_timeout")
         if cluster is not None and node_store is not None:
@@ -217,6 +228,7 @@ def __init__(
         )
         self._trust_server_heartbeat = trust_server_heartbeat
         self._close_timeout = close_timeout
+        self._max_attempts = max_attempts
 
         if cluster is not None:
             self._cluster = cluster
@@ -404,6 +416,7 @@ async def _create_connection(self) -> DqliteConnection:
             max_continuation_frames=self._max_continuation_frames,
             trust_server_heartbeat=self._trust_server_heartbeat,
             close_timeout=self._close_timeout,
+            max_attempts=self._max_attempts,
         )
 
     async def _release_reservation(self) -> None:

tests/test_pool_max_attempts_forwarding.py

@@ -0,0 +1,89 @@
+"""``ConnectionPool`` and ``create_pool`` must forward ``max_attempts``
+to the underlying ``ClusterClient.connect()`` call so operators can
+tune connect-retry behavior without having to construct a custom
+``ClusterClient`` and pass it via ``cluster=``.
+
+Today the parameter exists at the cluster layer
+(``ClusterClient.connect(max_attempts=...)``) but is invisible from
+the pool surface, leaving pool users with the hard-coded default of 3.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from dqliteclient import create_pool
+from dqliteclient.pool import ConnectionPool
+
+
+@pytest.mark.asyncio
+async def test_connection_pool_forwards_max_attempts() -> None:
+    """``ConnectionPool(..., max_attempts=N)`` forwards N to
+    ``cluster.connect``."""
+    pool = ConnectionPool(["a:9001", "b:9002"], max_attempts=7)
+    fake_connect = AsyncMock(return_value=AsyncMock())
+    pool._cluster.connect = fake_connect
+
+    await pool._create_connection()
+
+    fake_connect.assert_awaited_once()
+    assert fake_connect.await_args is not None
+    call_kwargs = fake_connect.await_args.kwargs
+    assert call_kwargs.get("max_attempts") == 7
+
+
+@pytest.mark.asyncio
+async def test_connection_pool_default_max_attempts_is_none() -> None:
+    """When ``max_attempts`` is not passed, the pool forwards ``None``
+    so ``ClusterClient.connect`` uses its built-in default."""
+    pool = ConnectionPool(["a:9001"])
+    fake_connect = AsyncMock(return_value=AsyncMock())
+    pool._cluster.connect = fake_connect
+
+    await pool._create_connection()
+
+    assert fake_connect.await_args is not None
+    call_kwargs = fake_connect.await_args.kwargs
+    assert call_kwargs.get("max_attempts") is None
+
+
+@pytest.mark.asyncio
+async def test_create_pool_forwards_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
+    """``create_pool(..., max_attempts=N)`` reaches the underlying
+    cluster.connect call too. Bypass real network connect with a
+    mock cluster on the pool itself; we just need to verify the
+    parameter flows through."""
+    captured: dict[str, object] = {}
+
+    async def fake_initialize(self: object) -> None:
+        # Skip the bootstrap connect; we are only verifying parameter
+        # forwarding from create_pool → ConnectionPool, not real
+        # cluster reachability.
+        return
+
+    monkeypatch.setattr("dqliteclient.pool.ConnectionPool.initialize", fake_initialize)
+    pool = await create_pool(["a:9001"], max_attempts=5)
+    try:
+
+        async def fake_connect(**kwargs: object) -> AsyncMock:
+            captured.update(kwargs)
+            return AsyncMock()
+
+        pool._cluster.connect = fake_connect  # type: ignore[assignment]
+        await pool._create_connection()
+        assert captured.get("max_attempts") == 5
+    finally:
+        # ``close()`` is safe to call on a pool that did not initialize.
+        await pool.close()
+
+
+def test_connection_pool_rejects_zero_max_attempts() -> None:
+    with pytest.raises(ValueError, match="max_attempts"):
+        ConnectionPool(["a:9001"], max_attempts=0)
+
+
+def test_connection_pool_rejects_negative_max_attempts() -> None:
+    with pytest.raises(ValueError, match="max_attempts"):
+        ConnectionPool(["a:9001"], max_attempts=-1)