fix: signal state change on capacity decrements; remove 500ms poll

acquire() used asyncio.wait_for(pool.get(), timeout=0.5) as a polling
loop. With K waiters, that's 2K timer events/s and a 250 ms median
wake latency even when state has changed. Extended the closed-event
to a general state-change signal (same primitive, cleared-and-set per
wait cycle) and fire it on every size decrement (release, full queue,
drain, user-exception cleanup). Wake latency is now bounded by the
actual state change.

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

Author: antoineleclair

src/dqliteclient/pool.py

@@ -111,6 +111,17 @@ def _get_closed_event(self) -> asyncio.Event:
                 self._closed_event.set()
         return self._closed_event
 
+    def _signal_state_change(self) -> None:
+        """Wake any acquire() waiters to re-check pool state.
+
+        Reuses the closed-event path so we don't add a second primitive.
+        acquire() clears the event right before it parks each iteration,
+        so set()s that come through this helper reliably wake the current
+        wait; waiters always re-check _closed at the loop top.
+        """
+        if self._closed_event is not None:
+            self._closed_event.set()
+
     async def _drain_idle(self) -> None:
         """Close all idle connections in the pool.
 
@@ -129,6 +140,7 @@ async def _drain_idle(self) -> None:
                 pass
             finally:
                 self._size -= 1
+                self._signal_state_change()
 
     @asynccontextmanager
     async def acquire(self) -> AsyncIterator[DqliteConnection]:
@@ -167,15 +179,19 @@ async def acquire(self) -> AsyncIterator[DqliteConnection]:
                     f"Timed out waiting for a connection from the pool "
                     f"(max_size={self._max_size}, timeout={self._timeout}s)"
                 )
-            # Race the queue against the closed event so close() wakes
-            # waiters promptly instead of leaving them on the polling loop.
+            # Race the queue against the state-change event so any pool
+            # state change (close, size decrement, drain) wakes waiters
+            # promptly. Clear right before parking so we only wake on
+            # subsequent signals; the loop top re-checks _closed.
             closed_event = self._get_closed_event()
+            if not self._closed:
+                closed_event.clear()
             get_task: asyncio.Task[DqliteConnection] = asyncio.create_task(self._pool.get())
             closed_task = asyncio.create_task(closed_event.wait())
             try:
                 done, _pending = await asyncio.wait(
                     {get_task, closed_task},
-                    timeout=min(remaining, 0.5),
+                    timeout=remaining,
                     return_when=asyncio.FIRST_COMPLETED,
                 )
             finally:
@@ -211,13 +227,15 @@ async def acquire(self) -> AsyncIterator[DqliteConnection]:
                     with contextlib.suppress(Exception):
                         await conn.close()
                     self._size -= 1
+                    self._signal_state_change()
                     raise
                 conn._pool_released = True
                 try:
                     self._pool.put_nowait(conn)
                 except asyncio.QueueFull:
                     await conn.close()
                     self._size -= 1
+                    self._signal_state_change()
             else:
                 # Connection is broken (invalidated by execute/fetch error handlers).
                 # Drain other idle connections — they likely point to the same dead server.
@@ -226,6 +244,7 @@ async def acquire(self) -> AsyncIterator[DqliteConnection]:
                 with contextlib.suppress(BaseException):
                     await conn.close()
                 self._size -= 1
+                self._signal_state_change()
             raise
         else:
             await self._release(conn)
@@ -256,13 +275,15 @@ async def _release(self, conn: DqliteConnection) -> None:
             conn._pool_released = True
             await conn.close()
             self._size -= 1
+            self._signal_state_change()
             return
 
         if not await self._reset_connection(conn):
             conn._pool_released = True
             with contextlib.suppress(Exception):
                 await conn.close()
             self._size -= 1
+            self._signal_state_change()
             return
 
         conn._pool_released = True
@@ -271,6 +292,7 @@ async def _release(self, conn: DqliteConnection) -> None:
         except asyncio.QueueFull:
             await conn.close()
             self._size -= 1
+            self._signal_state_change()
 
     async def execute(self, sql: str, params: Sequence[Any] | None = None) -> tuple[int, int]:
         """Execute a SQL statement using a pooled connection."""

tests/test_pool.py

@@ -736,6 +736,55 @@ async def track_exec(*args, **kwargs):
         assert result is False
         assert not exec_called, "ROLLBACK must not be sent on a dead socket"
 
+    async def test_acquire_wakes_on_release_without_polling_delay(self) -> None:
+        """When a held connection is released, a waiter must wake up
+        promptly — not sit on the 500 ms polling cadence.
+        """
+        import asyncio
+        import time
+
+        from dqliteclient.connection import DqliteConnection
+
+        pool = ConnectionPool(["localhost:9001"], max_size=1, timeout=5.0)
+
+        mock_conn = MagicMock(spec=DqliteConnection)
+        mock_conn.is_connected = True
+        mock_conn.close = AsyncMock()
+        mock_conn._in_transaction = False
+        mock_conn._in_use = False
+        mock_conn._bound_loop = None
+        mock_conn._pool_released = False
+        mock_conn._check_in_use = MagicMock()
+
+        with patch.object(pool._cluster, "connect", return_value=mock_conn):
+            await pool.initialize()
+
+            holder_acquired = asyncio.Event()
+            release_holder = asyncio.Event()
+
+            async def hold_connection() -> None:
+                async with pool.acquire():
+                    holder_acquired.set()
+                    await release_holder.wait()
+
+            holder = asyncio.create_task(hold_connection())
+            await holder_acquired.wait()
+
+            async def grab() -> float:
+                t0 = time.monotonic()
+                async with pool.acquire():
+                    return time.monotonic() - t0
+
+            waiter = asyncio.create_task(grab())
+            await asyncio.sleep(0.05)
+            release_holder.set()
+            elapsed = await asyncio.wait_for(waiter, timeout=1.0)
+            await holder
+
+            await pool.close()
+
+        assert elapsed < 0.2, f"waiter should wake immediately on release; took {elapsed:.3f}s"
+
     async def test_close_wakes_waiter_promptly(self) -> None:
         """A task blocked in acquire() waiting for a connection must be
         woken quickly when pool.close() is called — not sit on the queue