fix: prevent double pool initialization from concurrent callers

Wrap initialize() body in async with self._lock so that two
concurrent callers (e.g., via asyncio.gather on __aenter__) cannot
both pass the _initialized check and create 2*min_size connections.

Without the lock, the await points in _create_connection() and
_pool.put() allowed interleaving, causing the pool to exceed
max_size or even deadlock when min_size == max_size (the second
initializer blocks trying to put into a full queue).

Closes #082

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

Author: antoineleclair

src/dqliteclient/pool.py

@@ -59,12 +59,13 @@ def __init__(
 
     async def initialize(self) -> None:
         """Initialize the pool with minimum connections."""
-        if self._initialized:
-            return
-        for _ in range(self._min_size):
-            conn = await self._create_connection()
-            await self._pool.put(conn)
-        self._initialized = True
+        async with self._lock:
+            if self._initialized:
+                return
+            for _ in range(self._min_size):
+                conn = await self._create_connection()
+                await self._pool.put(conn)
+            self._initialized = True
 
     async def _create_connection(self) -> DqliteConnection:
         """Create a new connection to the leader."""

tests/test_pool.py

@@ -73,6 +73,35 @@ async def mock_connect(**kwargs):
 
         await pool.close()
 
+    async def test_concurrent_initialize_creates_only_min_size_connections(self) -> None:
+        """Concurrent initialize() calls must not create 2*min_size connections."""
+        import asyncio
+
+        pool = ConnectionPool(["localhost:9001"], min_size=2, max_size=5)
+
+        create_count = 0
+
+        async def mock_connect(**kwargs):
+            nonlocal create_count
+            create_count += 1
+            await asyncio.sleep(0)  # Yield to allow interleaving
+            mock_conn = MagicMock()
+            mock_conn.is_connected = True
+            mock_conn.connect = AsyncMock()
+            mock_conn.close = AsyncMock()
+            return mock_conn
+
+        with patch.object(pool._cluster, "connect", side_effect=mock_connect):
+            await asyncio.gather(pool.initialize(), pool.initialize())
+
+        assert create_count == 2, (
+            f"Expected 2 connections (min_size), got {create_count}. "
+            f"Concurrent initialize() created duplicate connections."
+        )
+        assert pool._size == 2
+
+        await pool.close()
+
     async def test_initialize_retryable_after_failure(self) -> None:
         """If initialize() fails, it should be retryable (not permanently stuck)."""
         pool = ConnectionPool(["localhost:9001"], min_size=1, max_size=5)