Design a blocking connection pool

01The design, in the six-step framework

1. Requirements and use cases

A caller acquires a connection, uses it, and releases it. If none is idle and the pool is under its maximum, the pool creates one; if the pool is full, the caller blocks until a connection is returned or a timeout passes. Releasing twice is harmless. The pool reports how many connections are currently out. Leaked connections, ones never returned, are reclaimed after a lease time.

Out of scope: connection creation itself (a factory is injected), retrying failed queries, and sharding across several databases. The reported round started with acquire, release and active count and then layered on blocking, owner-only release and leak prevention.

Constraints: many threads call acquire and release concurrently; a connection must never be handed to two callers at once; idle connections are reused first-in first-out so none goes stale; the pool must never exceed max_size live connections, even transiently.

2. Core entities and relationships

• ConnectionPool: owns the idle deque, the map of active leases, the count of created connections and one condition variable guarding all three.
• Lease: a token handed to the caller, carrying the connection, a unique id and the issue time. It is the only path back to the pool, and it works as a context manager.
• Connection: a protocol with a health check and close; the real driver object is behind it.
• factory: how new connections are made; injected so the pool can be tested with fakes.
• PoolTimeout: raised when a bounded wait expires.

The pool composes leases, which are meaningless outside it, and aggregates connections, whose lifetime the factory and close control. The lease id is the identity that makes release safe: the pool releases by id, never by connection object, so a stale or copied lease cannot return a connection that is now in someone else's hands.

3. Interfaces and key methods

class Connection(Protocol):
    def is_healthy(self) -> bool: ...
    def close(self) -> None: ...

class Lease:
    """Returned by acquire(); the only way to release."""
    conn: Connection
    def release(self) -> None: ...        # idempotent

class ConnectionPool:
    def __init__(self, factory: Callable[[], Connection], max_size: int,
                 lease_ttl: float = 30.0, clock: Callable[[], float] = time.monotonic) -> None: ...
    def acquire(self, timeout: float | None = None) -> Lease:
        """Block until a connection is free; create one if under max_size."""
    def release(self, lease: Lease) -> None:
        """Return the connection; a second release or a foreign lease is a no-op."""
    def active_count(self) -> int: ...
    def reap_expired(self) -> int:
        """Reclaim leases older than lease_ttl; returns how many."""

Invariants: every connection is in exactly one of idle, active or closed; len(idle) + len(active) == _created whenever the lock is free; _created <= max_size; a lease id appears in _active at most once and never again after release or reaping.

4. Design patterns and the choices behind them

An object pool with a lease, not a bare handle

The force is ownership: who may return a connection, and what happens on a double return. Returning the raw connection lets a caller release something it no longer owns. A pool that issues a Lease with its own id turns release into a dictionary pop: a second release or a foreign lease finds nothing and is a no-op, which is the idempotent behaviour the reported prompt required.

One condition variable, blocking acquirers

Acquire-when-empty is the classic producer-consumer wait: releasers notify, acquirers wait with the remaining timeout. A semaphore of max_size permits was the alternative; it blocks correctly but does not know which connection is free, so a lock around the deque is needed anyway. One condition does both jobs with less state.

Create lazily, cap strictly

_created is incremented under the lock before the factory runs, so two acquirers cannot both decide they may create the last connection. If the factory raises, the counter must be decremented in a finally; that one line is a common interview follow-up.

Leases expire

A caller that forgets to release would exhaust the pool forever. reap_expired closes connections whose lease is older than lease_ttl rather than returning them to idle, because a connection still in use by a forgetful caller must not be reused by another (see time and expiry inside objects). The clock is injected for tests.

5. Core flow and code

1. Thread A calls acquire() on an empty pool with max_size = 2: nothing idle, _created becomes 1, the factory builds connection c1, a Lease(1) is recorded as active and returned.
2. Thread B acquires: c2 is created the same way; active_count() is 2.
3. Thread C calls acquire(timeout=1): nothing idle and the pool is full, so it waits on the condition with the remaining time.
4. Thread A exits its with block; release pops lease 1, checks c1 is healthy, appends it to the idle deque and notifies.
5. C wakes, takes c1 from the front of the deque, receives Lease(3).
6. A calls release again by mistake: lease 1 is no longer in _active, so nothing happens and c1 stays with C.
7. A background thread calls reap_expired every few seconds; any lease past its TTL is closed and the slot freed.

import collections, itertools, threading, time
from typing import Callable, Protocol

class Connection(Protocol):
    def is_healthy(self) -> bool: ...
    def close(self) -> None: ...

class PoolTimeout(Exception): pass

class Lease:
    def __init__(self, pool: "ConnectionPool", conn: Connection, lease_id: int, issued_at: float) -> None:
        self.conn = conn
        self.lease_id = lease_id
        self.issued_at = issued_at
        self._pool = pool

    def release(self) -> None:
        self._pool.release(self)

    def __enter__(self) -> Connection:
        return self.conn

    def __exit__(self, *exc: object) -> None:
        self.release()

class ConnectionPool:
    def __init__(self, factory: Callable[[], Connection], max_size: int,
                 lease_ttl: float = 30.0, clock: Callable[[], float] = time.monotonic) -> None:
        if max_size < 1:
            raise ValueError("max_size must be positive")
        self._factory, self._max, self._ttl, self._clock = factory, max_size, lease_ttl, clock
        self._idle: collections.deque[Connection] = collections.deque()   # FIFO
        self._active: dict[int, Lease] = {}                               # lease_id -> Lease
        self._created = 0
        self._ids = itertools.count(1)
        self._cv = threading.Condition()

    def acquire(self, timeout: float | None = None) -> Lease:
        deadline = None if timeout is None else self._clock() + timeout
        with self._cv:
            while True:
                conn = self._take_idle_locked()
                if conn is None and self._created < self._max:
                    self._created += 1
                    conn = self._factory()           # under the lock: simple, and creation is rare
                if conn is not None:
                    lease = Lease(self, conn, next(self._ids), self._clock())
                    self._active[lease.lease_id] = lease
                    return lease
                remaining = None if deadline is None else deadline - self._clock()
                if remaining is not None and remaining <= 0:
                    raise PoolTimeout("no connection free")
                self._cv.wait(remaining)

    def _take_idle_locked(self) -> Connection | None:
        while self._idle:
            conn = self._idle.popleft()
            if conn.is_healthy():
                return conn
            conn.close(); self._created -= 1      # drop a dead connection, allow a replacement
        return None

    def release(self, lease: Lease) -> None:
        with self._cv:
            if self._active.pop(lease.lease_id, None) is None:
                return                             # already released, expired, or not ours
            if lease.conn.is_healthy():
                self._idle.append(lease.conn)
            else:
                lease.conn.close(); self._created -= 1
            self._cv.notify()

    def active_count(self) -> int:
        with self._cv:
            return len(self._active)

    def reap_expired(self) -> int:
        now = self._clock(); reaped = 0
        with self._cv:
            for lease_id, lease in list(self._active.items()):
                if now - lease.issued_at > self._ttl:
                    del self._active[lease_id]
                    lease.conn.close(); self._created -= 1   # never reuse a connection in unknown hands
                    reaped += 1
            if reaped:
                self._cv.notify_all()
        return reaped

6. Extensions, concurrency and what interviewers probe

• Owner-only release (reported extension): record threading.get_ident() on the lease and refuse a release from another thread; the id-based map already makes a foreign lease object harmless.
• Health checks on idle connections: validating at acquire time (done in _take_idle_locked) catches connections the server closed; a periodic sweep adds cost but shortens the first-request penalty.
• Where the lock is essential: the take-or-create decision, because the cap is checked and incremented in one step; and the pop in release, which is what makes double release safe.
• Fairness: Condition.notify does not promise FIFO wakeups; if the interviewer insists on fair acquirers, keep an explicit deque of waiter events and hand connections to the head.
• Test first: a fake factory counting creations, two acquires then a third with a timeout that must raise; release one and assert the waiter gets exactly that connection; release the same lease twice and assert active_count is unchanged.
• Common mistake: using a LIFO stack for idle connections, which keeps one connection hot and lets the others time out server-side.

02Where this question was reported

• FastPrep · FIFO idle pool; failed/duplicate release is a no-op; extensions: semaphore permits, blocking acquirers, owner-only release, leak prevention via leases/health ch