Transaction pooling can route a stale `CancelRequest` to the next frontend

[apkipa]

PgDog version

v0.1.57 (main, commit 7594279, transaction pool mode)

Description

In transaction pooling, a CancelRequest captured for frontend A can cancel a later query from frontend B after A's physical backend is returned to the pool and assigned to B. Direct PostgreSQL does not reproduce this cross-frontend cancellation.

PgDog snapshots A's backend key, releases the pool lock, and only then opens a new TCP connection to send the cancel pool_impl.rs server.rs. During that window, transaction pooling can give the returned backend to a waiting frontend inner.rs, so the in-flight key can cancel B instead of A taken.rs.

Reproduce

1. Start PostgreSQL 16 directly and PgDog in transaction pool mode with pool size 8. Set DIRECT_DSN and PGDOG_DSN to the two endpoints.
2. Install psycopg[binary] and run:

#!/usr/bin/env python3
from contextlib import suppress
import threading
import time

import psycopg


RUNS = 5


def suppress_query(conn, sql: str) -> None:
    with suppress(BaseException):
        conn.execute(sql)


def one(dsn: str) -> bool:
    a, *holders, b = [psycopg.connect(dsn, autocommit=True) for _ in range(9)]
    workers = []
    for conn in holders:
        thread = threading.Thread(target=suppress_query, args=(conn, "select pg_sleep(1)"), daemon=True)
        thread.start()
        workers.append(thread)
    thread = threading.Thread(target=suppress_query, args=(a, "select pg_sleep(.5)"), daemon=True)
    thread.start()
    workers.append(thread)
    time.sleep(.03)

    error = []

    def wait() -> None:
        try:
            b.execute("select pg_sleep(.5)")
        except BaseException as exc:
            error.append(exc)

    thread = threading.Thread(target=wait, daemon=True)
    thread.start()
    workers.append(thread)
    cancels = [threading.Thread(target=a.cancel_safe, kwargs={"timeout": 1}, daemon=True) for _ in range(2)]
    for thread in cancels:
        thread.start()
    for thread in cancels:
        thread.join(1)
    for thread in workers:
        thread.join(2)
    for conn in [a, b, *holders]:
        conn.close()
    return bool(error and getattr(error[0], "sqlstate", None) == "57014")


for label, dsn in (("direct", DIRECT_DSN), ("pgdog", PGDOG_DSN)):
    print(f"{label}: cross_client_57014={sum(one(dsn) for _ in range(RUNS))}/{RUNS}")

Expected behavior

The query submitted by B must not be canceled by A's CancelRequest. Both endpoints should report zero cross-client 57014 results.

Actual behavior

The second cancel races backend reassignment, so the count is timing-dependent. A complete run on 7594279 produced:

direct: cross_client_57014=0/5
pgdog: cross_client_57014=5/5