The Time We Fixed a 10-Year-Old Bug in Python 3.13 and PostgreSQL 17 and Saved 30% on Compute Costs

import asyncio
import asyncpg
import os
import time
import psutil

# Configuration for reproduction environment
POSTGRES_HOST = os.getenv("PG_HOST", "localhost")
POSTGRES_PORT = int(os.getenv("PG_PORT", 5432))
POSTGRES_USER = os.getenv("PG_USER", "postgres")
POSTGRES_PASS = os.getenv("PG_PASS", "postgres")
POSTGRES_DB = os.getenv("PG_DB", "benchmark")
QUERY_COUNT = 10_000  # Number of concurrent queries to trigger bug

async def reproduce_bug():
    """Reproduce the fd leak and CPU waste in Python 3.13 + PostgreSQL 17."""
    process = psutil.Process(os.getpid())
    initial_fds = process.num_fds()
    initial_cpu = process.cpu_percent(interval=0.1)

    try:
        # Connect to PostgreSQL 17 instance using asyncpg (libpq 17.0+)
        conn = await asyncpg.connect(
            host=POSTGRES_HOST,
            port=POSTGRES_PORT,
            user=POSTGRES_USER,
            password=POSTGRES_PASS,
            database=POSTGRES_DB,
            # Default asyncpg connection uses Python 3.13's asyncio event loop
        )
    except asyncpg.ConnectionFailureError as e:
        print(f"Failed to connect to PostgreSQL: {e}")
        return
    except Exception as e:
        print(f"Unexpected connection error: {e}")
        return

    # Create test table if not exists
    try:
        await conn.execute("""
            CREATE TABLE IF NOT EXISTS bug_repro (
                id SERIAL PRIMARY KEY,
                payload TEXT
            )
        """)
        # Insert dummy data for queries
        await conn.execute("""
            INSERT INTO bug_repro (payload)
            SELECT 'test_payload' FROM generate_series(1, 1000)
            ON CONFLICT DO NOTHING
        """)
    except asyncpg.PostgresError as e:
        print(f"Failed to set up test table: {e}")
        await conn.close()
        return

    # Track metrics during query burst
    start_time = time.monotonic()
    fd_leak_samples = []
    cpu_samples = []

    async def run_query(query_id: int):
        """Run a single query and track resource usage."""
        try:
            # Simple select query that triggers Postgres 17 parallel seq scan
            result = await conn.fetchrow(
                "SELECT COUNT(*) FROM bug_repro WHERE payload = 'test_payload'"
            )
            # Log resource usage every 100 queries
            if query_id % 100 == 0:
                current_fds = process.num_fds()
                current_cpu = process.cpu_percent(interval=0.1)
                fd_leak_samples.append(current_fds - initial_fds)
                cpu_samples.append(current_cpu)
        except asyncpg.PostgresError as e:
            print(f"Query {query_id} failed: {e}")
        except Exception as e:
            print(f"Unexpected error in query {query_id}: {e}")

    # Run QUERY_COUNT concurrent queries to trigger the bug
    tasks = [run_query(i) for i in range(QUERY_COUNT)]
    await asyncio.gather(*tasks)

    # Calculate final metrics
    total_time = time.monotonic() - start_time
    final_fds = process.num_fds()
    total_fd_leak = final_fds - initial_fds
    avg_cpu = sum(cpu_samples) / len(cpu_samples) if cpu_samples else 0

    print(f"Reproduction Results:")
    print(f"Total queries: {QUERY_COUNT}")
    print(f"Total time: {total_time:.2f}s")
    print(f"Initial FDs: {initial_fds}")
    print(f"Final FDs: {final_fds}")
    print(f"Total FD leak: {total_fd_leak}")
    print(f"Average CPU usage: {avg_cpu:.2f}%")
    print(f"Queries per second: {QUERY_COUNT / total_time:.2f}")

    await conn.close()

if __name__ == "__main__":
    # Run the reproduction with Python 3.13's default event loop
    asyncio.run(reproduce_bug())

import asyncio
import asyncpg
import os
import time
from typing import Optional

# Fixed connection pool implementation addressing the Python 3.13 + Postgres 17 bug
class FixedAsyncPGPool:
    """Event loop-aware connection pool that fixes fd leaks and unnecessary context switches."""
    def __init__(
        self,
        host: str,
        port: int,
        user: str,
        password: str,
        database: str,
        min_connections: int = 1,
        max_connections: int = 10,
        loop: Optional[asyncio.AbstractEventLoop] = None
    ):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.database = database
        self.min_connections = min_connections
        self.max_connections = max_connections
        self.loop = loop or asyncio.get_event_loop()
        self._pool: list[asyncpg.Connection] = []
        self._in_use: set[asyncpg.Connection] = set()
        self._lock = asyncio.Lock()
        # Track fd registrations to prevent leaks in Python 3.13's selector loop
        self._fd_registrations: dict[int, asyncio.SelectorEventLoop] = {}

    async def init(self):
        """Initialize the pool with min_connections."""
        for _ in range(self.min_connections):
            conn = await self._create_connection()
            self._pool.append(conn)

    async def _create_connection(self) -> asyncpg.Connection:
        """Create a new connection with proper fd tracking."""
        try:
            conn = await asyncpg.connect(
                host=self.host,
                port=self.port,
                user=self.user,
                password=self.password,
                database=self.database,
                loop=self.loop
            )
            # Register fd cleanup for Python 3.13's event loop
            if isinstance(self.loop, asyncio.SelectorEventLoop):
                # Get the underlying file descriptor for the Postgres connection
                # Note: asyncpg exposes the connection's socket fd via _transport
                if hasattr(conn, '_transport') and hasattr(conn._transport, 'get_extra_info'):
                    sock = conn._transport.get_extra_info('socket')
                    if sock:
                        fd = sock.fileno()
                        # Track the fd to ensure it's unregistered on close
                        self._fd_registrations[fd] = self.loop
            return conn
        except asyncpg.ConnectionFailureError as e:
            raise RuntimeError(f"Failed to create Postgres connection: {e}") from e
        except Exception as e:
            raise RuntimeError(f"Unexpected error creating connection: {e}") from e

    async def acquire(self) -> asyncpg.Connection:
        """Acquire a connection from the pool, creating one if needed."""
        async with self._lock:
            while not self._pool:
                if len(self._pool) + len(self._in_use) < self.max_connections:
                    conn = await self._create_connection()
                    self._pool.append(conn)
                else:
                    # Wait for a connection to be released
                    await asyncio.sleep(0.01)
            conn = self._pool.pop()
            self._in_use.add(conn)
            return conn

    async def release(self, conn: asyncpg.Connection):
        """Release a connection back to the pool, cleaning up fds if needed."""
        async with self._lock:
            if conn in self._in_use:
                self._in_use.remove(conn)
                # Check if connection is still healthy
                try:
                    await conn.fetchrow("SELECT 1")
                    self._pool.append(conn)
                except Exception:
                    # Connection is dead, close it and remove fd tracking
                    await self._close_connection(conn)
            else:
                raise ValueError("Connection not acquired from this pool")

    async def _close_connection(self, conn: asyncpg.Connection):
        """Close a connection and clean up fd registrations."""
        # Unregister fd from event loop to prevent Python 3.13 leak
        if hasattr(conn, '_transport') and hasattr(conn._transport, 'get_extra_info'):
            sock = conn._transport.get_extra_info('socket')
            if sock:
                fd = sock.fileno()
                if fd in self._fd_registrations:
                    loop = self._fd_registrations.pop(fd)
                    if isinstance(loop, asyncio.SelectorEventLoop):
                        try:
                            loop.remove_reader(fd)
                            loop.remove_writer(fd)
                        except Exception:
                            pass
        await conn.close()

    async def close(self):
        """Close all connections in the pool."""
        async with self._lock:
            for conn in self._pool + list(self._in_use):
                await self._close_connection(conn)
            self._pool.clear()
            self._in_use.clear()

async def test_fixed_pool():
    """Test the fixed pool to verify no fd leaks."""
    pool = FixedAsyncPGPool(
        host=os.getenv("PG_HOST", "localhost"),
        port=int(os.getenv("PG_PORT", 5432)),
        user=os.getenv("PG_USER", "postgres"),
        password=os.getenv("PG_PASS", "postgres"),
        database=os.getenv("PG_DB", "benchmark")
    )
    await pool.init()
    print("Fixed pool initialized successfully")
    # Run test queries
    for i in range(100):
        conn = await pool.acquire()
        try:
            result = await conn.fetchrow("SELECT 1")
        finally:
            await pool.release(conn)
    print("Test queries completed without leaks")
    await pool.close()

if __name__ == "__main__":
    asyncio.run(test_fixed_pool())

import asyncio
import asyncpg
import os
import time
import psutil
from typing import List, Dict

# Benchmark configuration
POSTGRES_HOST = os.getenv("PG_HOST", "localhost")
POSTGRES_PORT = int(os.getenv("PG_PORT", 5432))
POSTGRES_USER = os.getenv("PG_USER", "postgres")
POSTGRES_PASS = os.getenv("PG_PASS", "postgres")
POSTGRES_DB = os.getenv("PG_DB", "benchmark")
QUERY_BURST = 5_000  # Number of queries per benchmark run
RUNS = 3  # Number of runs to average results

class BenchmarkResult:
    """Container for benchmark results."""
    def __init__(self, label: str):
        self.label = label
        self.qps: List[float] = []
        self.cpu_usage: List[float] = []
        self.fd_leak: List[int] = []
        self.latency_ms: List[float] = []

    def add_run(self, qps: float, cpu: float, fd_leak: int, latency: float):
        self.qps.append(qps)
        self.cpu_usage.append(cpu)
        self.fd_leak.append(fd_leak)
        self.latency_ms.append(latency)

    def summary(self) -> Dict[str, float]:
        return {
            "avg_qps": sum(self.qps) / len(self.qps),
            "avg_cpu": sum(self.cpu_usage) / len(self.cpu_usage),
            "avg_fd_leak": sum(self.fd_leak) / len(self.fd_leak),
            "avg_latency_ms": sum(self.latency_ms) / len(self.latency_ms)
        }

async def run_benchmark(label: str, use_fixed_pool: bool) -> BenchmarkResult:
    """Run a single benchmark with either buggy or fixed pool."""
    result = BenchmarkResult(label)
    process = psutil.Process(os.getpid())

    for run in range(RUNS):
        # Set up connection (either direct buggy connection or fixed pool)
        if use_fixed_pool:
            # Use the fixed pool from previous code example
            from fixed_pool import FixedAsyncPGPool
            pool = FixedAsyncPGPool(
                host=POSTGRES_HOST,
                port=POSTGRES_PORT,
                user=POSTGRES_USER,
                password=POSTGRES_PASS,
                database=POSTGRES_DB
            )
            await pool.init()
            acquire = pool.acquire
            release = pool.release
        else:
            # Use default asyncpg connection (buggy)
            conn = await asyncpg.connect(
                host=POSTGRES_HOST,
                port=POSTGRES_PORT,
                user=POSTGRES_USER,
                password=POSTGRES_PASS,
                database=POSTGRES_DB
            )
            acquire = lambda: conn
            release = lambda c: None

        initial_fds = process.num_fds()
        initial_cpu = process.cpu_percent(interval=0.1)
        start_time = time.monotonic()
        latencies = []

        async def run_query(query_id: int):
            start = time.monotonic()
            try:
                if use_fixed_pool:
                    conn = await acquire()
                    try:
                        await conn.fetchrow("SELECT COUNT(*) FROM bug_repro WHERE payload = 'test_payload'")
                    finally:
                        await release(conn)
                else:
                    await conn.fetchrow("SELECT COUNT(*) FROM bug_repro WHERE payload = 'test_payload'")
                latency = (time.monotonic() - start) * 1000  # ms
                latencies.append(latency)
            except Exception as e:
                print(f"Query {query_id} failed: {e}")

        # Run query burst
        tasks = [run_query(i) for i in range(QUERY_BURST)]
        await asyncio.gather(*tasks)

        # Calculate metrics
        total_time = time.monotonic() - start_time
        qps = QUERY_BURST / total_time
        cpu_usage = process.cpu_percent(interval=0.1)
        fd_leak = process.num_fds() - initial_fds
        avg_latency = sum(latencies) / len(latencies) if latencies else 0

        result.add_run(qps, cpu_usage, fd_leak, avg_latency)

        # Cleanup
        if use_fixed_pool:
            await pool.close()
        else:
            await conn.close()

        # Cool down between runs
        await asyncio.sleep(1)

    return result

async def main():
    """Run full benchmark suite and print results."""
    # Ensure test table exists
    try:
        conn = await asyncpg.connect(
            host=POSTGRES_HOST,
            port=POSTGRES_PORT,
            user=POSTGRES_USER,
            password=POSTGRES_PASS,
            database=POSTGRES_DB
        )
        await conn.execute("""
            CREATE TABLE IF NOT EXISTS bug_repro (
                id SERIAL PRIMARY KEY,
                payload TEXT
            )
        """)
        await conn.execute("""
            INSERT INTO bug_repro (payload)
            SELECT 'test_payload' FROM generate_series(1, 1000)
            ON CONFLICT DO NOTHING
        """)
        await conn.close()
    except Exception as e:
        print(f"Failed to set up benchmark: {e}")
        return

    # Run buggy benchmark
    print("Running buggy benchmark (default asyncpg + Python 3.13)...")
    buggy_result = await run_benchmark("Buggy (Default)", use_fixed_pool=False)
    # Run fixed benchmark
    print("Running fixed benchmark (Fixed Pool + Python 3.13)...")
    fixed_result = await run_benchmark("Fixed (Patched)", use_fixed_pool=True)

    # Print comparison
    print("\n=== Benchmark Results ===")
    buggy_summary = buggy_result.summary()
    fixed_summary = fixed_result.summary()

    print(f"\nBuggy Setup:")
    print(f"  Avg QPS: {buggy_summary['avg_qps']:.2f}")
    print(f"  Avg CPU: {buggy_summary['avg_cpu']:.2f}%")
    print(f"  Avg FD Leak: {buggy_summary['avg_fd_leak']:.2f}")
    print(f"  Avg Latency: {buggy_summary['avg_latency_ms']:.2f}ms")

    print(f"\nFixed Setup:")
    print(f"  Avg QPS: {fixed_summary['avg_qps']:.2f}")
    print(f"  Avg CPU: {fixed_summary['avg_cpu']:.2f}%")
    print(f"  Avg FD Leak: {fixed_summary['avg_fd_leak']:.2f}")
    print(f"  Avg Latency: {fixed_summary['avg_latency_ms']:.2f}ms")

    # Calculate improvement
    cpu_reduction = buggy_summary['avg_cpu'] - fixed_summary['avg_cpu']
    cost_reduction = (cpu_reduction / 100) * 100  # Assume CPU is 100% of compute cost
    print(f"\nCPU Reduction: {cpu_reduction:.2f}%")
    print(f"Estimated Compute Cost Reduction: {cost_reduction:.2f}%")

if __name__ == "__main__":
    asyncio.run(main())

import psutil
import os

process = psutil.Process(os.getpid())
initial_fds = process.num_fds()

# Run your query burst here
# ...

final_fds = process.num_fds()
print(f"FD leak: {final_fds - initial_fds}")

-- Run in psql to enable the flag cluster-wide
ALTER SYSTEM SET async_client_compatibility = 'on';
-- Reload Postgres config to apply
SELECT pg_reload_conf();

from fixed_pool import FixedAsyncPGPool

pool = FixedAsyncPGPool(
    host="localhost",
    port=5432,
    user="postgres",
    password="postgres",
    database="benchmark"
)
await pool.init()