Implement a Database Connection Pool

Problem Statement

Design and implement a thread-safe ConnectionPool that manages a fixed maximum number of database connections. In high-concurrency environments, creating connections on demand is expensive, while unbounded connections can overwhelm the database. Your pool must manage connection reuse, validate connection health, and handle blocking callers gracefully when the pool is exhausted.

Assume a Connection object provides the following interface:

class Connection:
    def is_valid(self) -> bool:
        """Returns True if the connection is alive, False otherwise."""
        pass

    def close(self) -> None:
        """Closes the underlying connection resource."""
        pass

Implement the ConnectionPool class with the following signature:

from typing import Callable, Optional

class ConnectionPool:
    def __init__(self, max_connections: int, factory: Callable[[], Connection]):
        """
        Initialize the pool with a maximum capacity and a factory callable 
        that creates a new Connection instance when needed.
        """
        pass

    def acquire(self, timeout: Optional[float] = None) -> Connection:
        """
        Retrieves a valid connection from the pool. If no connection is 
        available and max_connections has not been reached, create a new one.
        If max_connections has been reached, block until a connection is returned 
        or the specified timeout (in seconds) expires. If timeout expires, 
        raise a TimeoutError.
        If a retrieved connection is not valid, discard it and obtain/create 
        another one.
        """
        pass

    def release(self, conn: Connection) -> None:
        """
        Returns a connection back to the pool. If the connection is no longer 
        valid, close it and allow a new connection to be created in the future.
        """
        pass

    def close_all(self) -> None:
        """
        Closes all idle connections and shuts down the pool. Any subsequent 
        call to acquire() should raise a RuntimeError.
        """
        pass

Example Usage

pool = ConnectionPool(max_connections=2, factory=create_db_connection)

# Thread 1
conn1 = pool.acquire(timeout=1.0)

# Thread 2
conn2 = pool.acquire(timeout=1.0)

# Thread 3 attempts to acquire; blocks until conn1 or conn2 is released
# If no connection is released within 0.5s, raises TimeoutError
try:
    conn3 = pool.acquire(timeout=0.5)
except TimeoutError:
    print("Timed out waiting for connection")

pool.release(conn1)
conn3 = pool.acquire(timeout=0.5)  # Now succeeds, receiving conn1

Constraints & Edge Cases

• 1 <= max_connections <= 1000
• Multiple threads may call acquire, release, and close_all concurrently.
• Do not create all max_connections upfront; instantiate them lazily as needed.
• A thread releasing an already-invalid connection should not cause the pool to permanently lose capacity.
• Calling release with a connection not managed by the pool or returning the same connection twice should be guarded against.