Add operator-actionable discriminators to pool acquire timeout

The pool acquire timeout was the most-likely-to-page exception in
the driver and the least actionable. The previous message gave
``max_size`` and ``timeout`` only — an operator paged on a 3am
timeout could not tell whether the cause was a leaking application
(checked_out at max, idle 0) or a slow cluster, even though both
have very different remediations.

Add ``pool_id``, ``checked_out`` and ``idle`` to the message and
attach a one-sentence remediation hint that discriminates leak vs
slow cluster. The neighbouring ``_check_in_use`` site already proves
the codebase can write good actionable exceptions; the pool acquire
timeout now matches that bar.

Pin the new fields and the hint against regression so a future
refactor can't accidentally drop them.

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

Author: antoineleclair

src/dqliteclient/pool.py

@@ -821,9 +821,22 @@ async def acquire(self) -> AsyncIterator[DqliteConnection]:
             # re-check capacity (another coroutine may have freed a slot)
             remaining = deadline - loop.time()
             if remaining <= 0:
+                # Compute saturation snapshot once for the message —
+                # ``_size`` is checked-out-plus-reserved; ``qsize`` is
+                # idle. A pool at ``checked_out == max_size`` with
+                # ``idle == 0`` is leaking; otherwise the cluster is
+                # slow. The ``pool_id`` lets operators correlate the
+                # failed acquire with the warm-up / drain log lines
+                # that already include ``id(self)``.
+                idle = self._pool.qsize()
+                checked_out = self._size - idle
                 raise DqliteConnectionError(
                     f"Timed out waiting for a connection from the pool "
-                    f"(max_size={self._max_size}, timeout={self._timeout}s)"
+                    f"(pool_id={id(self)}, max_size={self._max_size}, "
+                    f"checked_out={checked_out}, idle={idle}, "
+                    f"timeout={self._timeout}s). "
+                    f"If checked_out is at max_size, the application is "
+                    f"leaking connections; otherwise the cluster is slow."
                 )
             # Race the queue against the state-change event so any pool
             # state change (close, size decrement, drain) wakes waiters

tests/test_pool_acquire_timeout_actionable_message.py

@@ -0,0 +1,87 @@
+"""Pin: pool acquire timeout error includes operator-actionable
+discriminators (pool_id, checked_out, idle) and a remediation hint.
+
+The message previously gave only ``max_size`` and ``timeout`` — an
+operator paged on a 3am ``DqliteConnectionError("Timed out waiting
+for a connection from the pool")`` could not tell whether the cause
+was a leaking application (checked_out at max, idle 0) or a slow
+cluster (checked_out below max, idle 0). Both have the same parameters
+in the message but very different remediations.
+
+Pin the new fields against regression so a future refactor can't
+accidentally drop them.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from dqliteclient.exceptions import DqliteConnectionError
+from dqliteclient.pool import ConnectionPool
+
+
+def _make_full_pool_at_capacity() -> ConnectionPool:
+    """Build a minimal ConnectionPool with no available connections
+    so an acquire times out immediately."""
+    import os
+
+    pool = ConnectionPool.__new__(ConnectionPool)
+    pool._closed = False
+    pool._max_size = 5
+    pool._size = 5  # checked-out + reserved == max_size
+    pool._timeout = 0.01
+    pool._pool = asyncio.Queue()  # no connections in the queue
+    pool._lock = asyncio.Lock()
+    pool._closed_event = None
+    pool._addresses = ["host:9001"]
+    pool._creator_pid = os.getpid()
+    return pool
+
+
+async def _acquire_with_timeout(pool: ConnectionPool) -> None:
+    async with pool.acquire():
+        pass
+
+
+@pytest.mark.asyncio
+async def test_acquire_timeout_message_includes_pool_id() -> None:
+    pool = _make_full_pool_at_capacity()
+    with pytest.raises(DqliteConnectionError) as exc_info:
+        await _acquire_with_timeout(pool)
+    text = str(exc_info.value)
+    assert f"pool_id={id(pool)}" in text
+
+
+@pytest.mark.asyncio
+async def test_acquire_timeout_message_includes_checked_out_and_idle() -> None:
+    pool = _make_full_pool_at_capacity()
+    with pytest.raises(DqliteConnectionError) as exc_info:
+        await _acquire_with_timeout(pool)
+    text = str(exc_info.value)
+    assert "checked_out=" in text
+    assert "idle=" in text
+
+
+@pytest.mark.asyncio
+async def test_acquire_timeout_message_includes_remediation_hint() -> None:
+    """The hint discriminates leak vs slow-cluster — pin so a future
+    edit doesn't strip it."""
+    pool = _make_full_pool_at_capacity()
+    with pytest.raises(DqliteConnectionError) as exc_info:
+        await _acquire_with_timeout(pool)
+    text = str(exc_info.value)
+    assert "leaking" in text.lower() or "leak" in text.lower()
+
+
+@pytest.mark.asyncio
+async def test_acquire_timeout_message_keeps_max_size_and_timeout() -> None:
+    """Existing fields must still be present — they are referenced
+    by operators / runbooks."""
+    pool = _make_full_pool_at_capacity()
+    with pytest.raises(DqliteConnectionError) as exc_info:
+        await _acquire_with_timeout(pool)
+    text = str(exc_info.value)
+    assert "max_size=5" in text
+    assert "timeout=" in text