pylon_pgcon/

lib.rs

//
// This source file is part of the Pylon open source project.
//
// Copyright (c) 2026 Jaldis B.V.
//
// Licensed under the MIT OR Apache-2.0 license (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://opensource.org/licenses/MIT
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//

//! Postgres connection pooling and query execution, via `tokio-postgres` +
//! `deadpool-postgres`. No dependency on `pylon-core` or PyO3 — usable as a
//! plain Rust Postgres client crate on its own; `pylon-core` depends on
//! *this* crate for execution, not the other way around.
//!
//! `query_raw` returns raw `tokio_postgres::Row`s; `wire` decodes the
//! composite `result` column those rows carry into `pylon_value::DecodedValue`
//! (the shared decode target `pylon-cache` also stores).

pub mod error;
pub mod listener;
pub mod numeric;
pub mod wire;

pub use error::{Error, Result};
pub use listener::PgListener;
pub use wire::{ExtensionOids, decode_value};

use pylon_value::DecodedValue;
use std::sync::{Arc, RwLock};

/// Captures a column's raw wire bytes regardless of its declared Postgres
/// type. `tokio_postgres`'s own `&[u8]` `FromSql` impl only accepts
/// `BYTEA` — the `result` column pylon-core emits is always `record`
/// (OID 2249), so a plain `row.get::<_, &[u8]>(0)` would panic on the type
/// check. This wrapper's `accepts` is unconditionally `true`, matching the
/// well-established pattern for pulling raw bytes out of any column.
struct RawBytes<'a>(&'a [u8]);

impl<'a> postgres_types::FromSql<'a> for RawBytes<'a> {
    fn from_sql(
        _ty: &postgres_types::Type,
        raw: &'a [u8],
    ) -> std::result::Result<Self, Box<dyn std::error::Error + Sync + Send>> {
        Ok(RawBytes(raw))
    }

    fn accepts(_ty: &postgres_types::Type) -> bool {
        true
    }
}

/// Wraps a `DecodedValue` for binding as a query parameter. `accepts` is
/// unconditionally `true` (mirroring `RawBytes` above) because the target
/// `Type` isn't known until `Statement::params()` reports it — see
/// `wire::encode_value`, which does the actual type-directed encoding.
#[derive(Debug)]
struct BoundParam<'a>(&'a DecodedValue);

impl postgres_types::ToSql for BoundParam<'_> {
    fn to_sql(
        &self,
        ty: &postgres_types::Type,
        out: &mut bytes::BytesMut,
    ) -> std::result::Result<postgres_types::IsNull, Box<dyn std::error::Error + Sync + Send>> {
        Ok(wire::encode_value(self.0, ty, out)?)
    }

    fn accepts(_ty: &postgres_types::Type) -> bool {
        true
    }

    postgres_types::to_sql_checked!();
}

// `deadpool_postgres::Pool` is `Arc`-backed internally, so cloning a
// `PgPool` is cheap and shares the same underlying pool — needed at the
// pyo3 boundary, where a lock guard over the process-global pool slot
// can't be held across an `.await` (it isn't `Send`), so callers clone the
// pool out from under the lock first.
#[derive(Clone, Debug)]
pub struct PgPool {
    pool: deadpool_postgres::Pool,
    /// Shared with every clone of this pool, so a refresh reaches all of
    /// them — consumers hold the pool behind an `Arc`, with only `&self`
    /// available at the moments the registry needs replacing.
    types: Arc<RwLock<Arc<ExtensionOids>>>,
    /// Label for this pool in observability output. Set via
    /// `PgPool::set_name`; `"default"` until then.
    name: String,
}

/// The one failure a stale type registry causes, and so the only one worth
/// re-discovering types for and retrying. Any other error would survive a
/// retry, which would also run the statement twice.
pub(crate) fn is_unknown_oid(err: &Error) -> bool {
    matches!(err, Error::UnknownTypeOid { .. })
}

/// Notified with `(pool name, time spent waiting)` every time a caller
/// acquires a pooled connection.
///
/// A hook rather than a direct metrics call because the dependency only goes
/// one way: `pylon-workers` (which owns the Prometheus registry) depends on
/// this crate, so this crate can't call into it. Whoever owns the registry
/// installs an observer at startup; nobody installing one costs a single
/// `OnceLock` read per checkout.
pub type PoolWaitObserver = fn(&str, std::time::Duration);

static POOL_WAIT_OBSERVER: std::sync::OnceLock<PoolWaitObserver> = std::sync::OnceLock::new();

/// Installs the pool-wait observer. The first call wins; later ones are
/// ignored, so a process that initialises metrics twice can't double-count.
pub fn set_pool_wait_observer(observer: PoolWaitObserver) {
    let _ = POOL_WAIT_OBSERVER.set(observer);
}

/// Snapshot of a pool's connection accounting — deadpool's own `Status`,
/// re-exported under this crate's name so callers (the Prometheus gauge
/// sampler in `pylon-py`) don't need a direct `deadpool_postgres`
/// dependency just to read four numbers.
pub struct PoolStatus {
    /// Connections actually established right now (idle + checked out).
    pub size: usize,
    /// Idle connections available for immediate checkout right now.
    pub available: usize,
    /// Callers currently blocked in `.get()` waiting for a connection.
    pub waiting: usize,
    pub max_size: usize,
}

impl PgPool {
    /// Connects using a `postgresql://` DSN, matching the DSN Python's
    /// `DatabaseConfig`/`_build_dsn` already produces today. No TLS support
    /// yet — no SSL/TLS surface exists anywhere in the project currently
    /// (confirmed by a full-repo grep during planning), so this isn't a
    /// regression; it's simply not needed until it is.
    ///
    /// Unlike `deadpool_postgres::Pool::builder(..).build()` on its own —
    /// which only validates the DSN and is otherwise lazy, deferring the
    /// first real connection attempt to whenever a caller first acquires
    /// one — this eagerly acquires and immediately releases one connection
    /// before returning, so a bad host/port/database/credentials fails
    /// right here. That eager-connect
    /// behavior, which the caller (`Client.ensure_connected`) depends on to
    /// raise `ConnectionFailedError`/`ConnectionTimeoutError` immediately
    /// rather than silently deferring the failure to the first query.
    pub async fn connect(dsn: &str, max_size: usize) -> Result<Self> {
        let pg_config = session_config(dsn)?;
        let manager = deadpool_postgres::Manager::new(pg_config, tokio_postgres::NoTls);
        let pool = deadpool_postgres::Pool::builder(manager)
            .max_size(max_size)
            .create_timeout(Some(std::time::Duration::from_secs(10)))
            .runtime(deadpool_postgres::Runtime::Tokio1)
            .build()?;
        let client = pool.get().await?;
        // Same round trip that proves the pool can connect also classifies
        // this database's enum/domain/extension type OIDs, which are
        // assigned locally and so can't be known statically.
        let types = discover_types(&client).await?;
        drop(client);
        Ok(Self {
            pool,
            types: Arc::new(RwLock::new(Arc::new(types))),
            name: "default".to_string(),
        })
    }

    /// The type OIDs currently discovered for this database. Pass this to
    /// `query_typed`/`query_composite` rather than `ExtensionOids::default()`
    /// — without it, a `vector`, enum, or domain column has no decoder.
    /// Owned rather than borrowed because `refresh_types` can replace it;
    /// `&pool.types()` coerces to the `&ExtensionOids` those methods take.
    pub fn types(&self) -> Arc<ExtensionOids> {
        self.types.read().unwrap().clone()
    }

    /// Re-runs type discovery. Needed after a migration creates an enum,
    /// domain or extension type, and after a restore re-assigns every local
    /// OID underneath a running pool — the registry is a connect-time
    /// snapshot and a pool outlives both.
    ///
    /// Clears the prepared-statement caches unconditionally, since DDL also
    /// invalidates a cached plan on its own (`0A000 cached plan must not
    /// change result type`).
    pub async fn refresh_types(&self) -> Result<()> {
        let client = self.checkout().await?;
        let fresh = discover_types(&client).await?;
        *self.types.write().unwrap() = Arc::new(fresh);
        drop(client);
        self.clear_statement_caches();
        Ok(())
    }

    /// Re-runs discovery on a held connection after a decode hit an unknown
    /// OID. Clears the statement caches only when the registry really
    /// changed: a type nothing can decode reaches this on every query, and
    /// emptying every connection's cache that often is its own problem.
    async fn heal_types(&self, client: &tokio_postgres::Client) -> Result<Arc<ExtensionOids>> {
        let fresh = Arc::new(discover_types(client).await?);
        let changed = {
            let mut slot = self.types.write().unwrap();
            let changed = *slot != fresh;
            if changed {
                *slot = fresh.clone();
            }
            changed
        };
        if changed {
            self.clear_statement_caches();
        }
        Ok(fresh)
    }

    /// Drops every pooled connection's prepared-statement cache.
    ///
    /// Must be called after any DDL that could change a statement's result
    /// type or parameter types — `refresh_types` already does, so migration
    /// paths get this for free. Cheap: it only empties the cache maps, it
    /// does not touch the connections.
    pub fn clear_statement_caches(&self) {
        self.pool.manager().statement_caches.clear();
    }

    /// Labels this pool in observability output — see `PoolWaitObserver`.
    pub fn set_name(&mut self, name: impl Into<String>) {
        self.name = name.into();
    }

    /// Acquires a pooled connection, reporting how long that took to the
    /// pool-wait observer if one is installed.
    ///
    /// Every query path goes through this rather than `self.pool.get()`
    /// directly, so "time spent waiting for a connection" is measured in one
    /// place and can't drift between call sites.
    async fn checkout(&self) -> Result<deadpool_postgres::Object> {
        let started = std::time::Instant::now();
        let client = self.pool.get().await?;
        if let Some(observe) = POOL_WAIT_OBSERVER.get() {
            observe(&self.name, started.elapsed());
        }
        Ok(client)
    }

    /// Current connection accounting — for the Prometheus gauge sampler
    /// (`pylon-py`'s `record_pool_metrics`), not used on any query path.
    pub fn status(&self) -> PoolStatus {
        let s = self.pool.status();
        PoolStatus {
            size: s.size,
            available: s.available,
            waiting: s.waiting,
            max_size: s.max_size,
        }
    }

    /// Executes `sql` with no parameters and returns the raw rows.
    /// Composite/record decoding is a later phase — this only proves the
    /// pool can connect and round-trip a query end to end.
    pub async fn query_raw(&self, sql: &str) -> Result<Vec<tokio_postgres::Row>> {
        let client = self.checkout().await?;
        let rows = client.query(sql, &[]).await?;
        Ok(rows)
    }

    /// Runs `sql` (expected to produce exactly one column, matching
    /// pylon-core's `SELECT (...) AS result` emission) and decodes that
    /// column of every row via `wire::decode_value`, using its actual
    /// declared Postgres type (not assumed to be `record` — a bare scalar
    /// `result` column decodes just as well through the same path).
    pub async fn query_composite(&self, sql: &str, ext: &ExtensionOids) -> Result<Vec<DecodedValue>> {
        let client = self.checkout().await?;
        let rows = client.query(sql, &[]).await?;
        let decode = |ext: &ExtensionOids| -> Result<Vec<DecodedValue>> {
            rows.iter().map(|r| decode_result_column(r, ext)).collect()
        };
        match decode(ext) {
            Err(e) if is_unknown_oid(&e) => {
                let healed = self.heal_types(&client).await?;
                decode(&healed)
            }
            other => other,
        }
    }

    /// Runs `sql` with bound `params`, matched positionally to `$1, $2, ...`
    /// — the same convention `pylon-core`'s `param_names` already assumes.
    /// No caller-supplied parameter types: `prepare` asks Postgres itself
    /// to analyze the SQL and report each placeholder's expected `Type`
    /// (`Statement::params()`), which drives `wire::encode_value`'s
    /// encoding directly. Decodes the single result column exactly like
    /// `query_composite`.
    pub async fn query_typed(
        &self,
        sql: &str,
        params: &[DecodedValue],
        ext: &ExtensionOids,
    ) -> Result<Vec<DecodedValue>> {
        let client = self.checkout().await?;
        match query_typed_on(&client, sql, params, ext).await {
            Err(e) if is_unknown_oid(&e) => {
                let healed = self.heal_types(&client).await?;
                query_typed_on(&client, sql, params, &healed).await
            }
            other => other,
        }
    }

    /// Like `query_typed`, but decodes every column of every row by name
    /// instead of assuming a single `(...) AS result` column — for
    /// hand-written admin SQL (CLI commands, not `pylon-core`-emitted
    /// query bodies) that reads named columns directly.
    pub async fn query_typed_named(
        &self,
        sql: &str,
        params: &[DecodedValue],
        ext: &ExtensionOids,
    ) -> Result<Vec<DecodedValue>> {
        let client = self.checkout().await?;
        match query_typed_named_on(&client, sql, params, ext).await {
            Err(e) if is_unknown_oid(&e) => {
                let healed = self.heal_types(&client).await?;
                query_typed_named_on(&client, sql, params, &healed).await
            }
            other => other,
        }
    }

    /// `query_typed`, with the session globals database triggers read
    /// (`pylon.globals`) set first on the same connection. Each mutating
    /// statement sets them anew, so a pooled connection never hands one
    /// statement's globals to the next.
    pub async fn query_typed_with_globals(
        &self,
        sql: &str,
        params: &[DecodedValue],
        ext: &ExtensionOids,
        globals: &str,
    ) -> Result<Vec<DecodedValue>> {
        let client = self.checkout().await?;
        set_globals_on(&client, globals, false).await?;
        match query_typed_on(&client, sql, params, ext).await {
            Err(e) if is_unknown_oid(&e) => {
                let healed = self.heal_types(&client).await?;
                query_typed_on(&client, sql, params, &healed).await
            }
            other => other,
        }
    }

    /// `execute_typed`, with `pylon.globals` set — see `query_typed_with_globals`.
    pub async fn execute_typed_with_globals(&self, sql: &str, params: &[DecodedValue], globals: &str) -> Result<u64> {
        let client = self.checkout().await?;
        set_globals_on(&client, globals, false).await?;
        execute_typed_on(&client, sql, params).await
    }

    /// Runs `sql` with bound `params` (same convention as `query_typed`)
    /// and discards the result, returning the number of rows affected —
    /// for `INSERT`/`UPDATE`/`DELETE` where the caller has no `RETURNING`
    /// clause to decode.
    pub async fn execute_typed(&self, sql: &str, params: &[DecodedValue]) -> Result<u64> {
        let client = self.checkout().await?;
        execute_typed_on(&client, sql, params).await
    }

    /// Runs `EXPLAIN (ANALYZE, FORMAT JSON, VERBOSE) sql` with bound
    /// `params` and returns the raw JSON output text verbatim — for
    /// `analyze <query>` (see `pylon_core::analyze`), which parses this text
    /// itself to correlate plan nodes back to the query's own shape.
    /// Actually *runs* the query (`ANALYZE`), same as Postgres's own
    /// `EXPLAIN ANALYZE`, not just its planner estimate.
    pub async fn query_explain(&self, sql: &str, params: &[DecodedValue]) -> Result<String> {
        let client = self.checkout().await?;
        query_explain_on(&client, sql, params).await
    }

    /// Acquires one pooled connection and starts an explicit transaction at
    /// the given isolation level (`"read_uncommitted"`, `"read_committed"`,
    /// `"repeatable_read"`, or `"serializable"` — matching
    /// `AsyncTransaction`'s existing accepted values in `client.py`, itself
    /// the four standard SQL isolation levels). The returned
    /// `PgTransaction` owns the connection until `commit`/`rollback`
    /// consumes it.
    pub async fn begin(&self, isolation: &str) -> Result<PgTransaction> {
        let client = self.checkout().await?;
        let level = match isolation {
            "read_uncommitted" => "READ UNCOMMITTED",
            "read_committed" => "READ COMMITTED",
            "repeatable_read" => "REPEATABLE READ",
            "serializable" => "SERIALIZABLE",
            other => return Err(Error::message(format!("unknown isolation level: {other:?}"))),
        };
        client.batch_execute(&format!("BEGIN ISOLATION LEVEL {level}")).await?;
        Ok(PgTransaction {
            client,
            types: self.types(),
        })
    }

    /// Starts a transaction with no explicit isolation level — whatever
    /// Postgres's own session/database default is applies. Used by
    /// migration execution, which (unlike `begin`) never needs a specific
    /// isolation level — this matches a plain `conn.transaction()`
    /// (no `isolation=` kwarg) that the old Python migration executor used.
    pub async fn begin_default(&self) -> Result<PgTransaction> {
        let client = self.checkout().await?;
        client.batch_execute("BEGIN").await?;
        Ok(PgTransaction {
            client,
            types: self.types(),
        })
    }

    /// Runs `sql` via the simple query protocol — no bind parameters, but
    /// (unlike `query_typed`/`execute_typed`, which prepare via the extended
    /// protocol and so accept exactly one statement) able to run several
    /// `;`-separated statements in one call. Matches
    /// `Connection.execute(sql)` called with no arguments, which migration
    /// DDL steps rely on (a step's body is whatever raw SQL text sits
    /// between `-- pylon:step` markers, often more than one statement).
    pub async fn batch_execute(&self, sql: &str) -> Result<()> {
        let client = self.checkout().await?;
        client.batch_execute(sql).await?;
        Ok(())
    }

    /// Checks out one pooled connection and hands back a handle the caller
    /// holds across several calls, with no transaction started — for
    /// state that's scoped to a single session rather than a single
    /// statement or transaction, the way Postgres advisory locks
    /// (`pg_advisory_lock`/`pg_advisory_unlock`) are: they're released by
    /// an explicit unlock (or the session ending), not by a transaction
    /// boundary, so acquiring and releasing one has to happen on the same
    /// held connection — calling `PgPool::batch_execute` twice wouldn't
    /// work, since each call may checkout a different pooled connection.
    pub async fn connection(&self) -> Result<PgConnection> {
        let client = self.checkout().await?;
        Ok(PgConnection {
            client,
            types: self.types(),
        })
    }
}

/// One connection checked out of the pool and held by the caller, with no
/// transaction open — see `PgPool::connection`.
#[derive(Debug)]
pub struct PgConnection {
    client: deadpool_postgres::Object,
    types: Arc<ExtensionOids>,
}

impl PgConnection {
    /// The registry the owning pool held at checkout — a snapshot, since a
    /// held connection lives for one unit of work.
    pub fn types(&self) -> &ExtensionOids {
        &self.types
    }

    pub async fn query_typed(
        &self,
        sql: &str,
        params: &[DecodedValue],
        ext: &ExtensionOids,
    ) -> Result<Vec<DecodedValue>> {
        query_typed_on(&self.client, sql, params, ext).await
    }

    pub async fn execute_typed(&self, sql: &str, params: &[DecodedValue]) -> Result<u64> {
        execute_typed_on(&self.client, sql, params).await
    }

    pub async fn batch_execute(&self, sql: &str) -> Result<()> {
        self.client.batch_execute(sql).await?;
        Ok(())
    }
}

/// Binds `params` to an already-prepared `stmt` and decodes the single
/// `result` column of every row.
///
/// Split out from the `*_on` entry points below so the two kinds of
/// connection this crate holds can each prepare the way that suits them —
/// a pooled `deadpool_postgres::Object` through its per-connection
/// statement cache, a bare `tokio_postgres::Client` (the LISTEN/NOTIFY
/// connection, which is not pooled) through a plain `prepare` — without
/// either duplicating the binding and decoding below.
async fn query_with_stmt(
    client: &tokio_postgres::Client,
    stmt: &tokio_postgres::Statement,
    params: &[DecodedValue],
    ext: &ExtensionOids,
) -> Result<Vec<DecodedValue>> {
    let bound: Vec<BoundParam<'_>> = params.iter().map(BoundParam).collect();
    let param_refs: Vec<&(dyn postgres_types::ToSql + Sync)> =
        bound.iter().map(|p| p as &(dyn postgres_types::ToSql + Sync)).collect();
    let rows = client.query(stmt, &param_refs).await?;
    rows.iter().map(|row| decode_result_column(row, ext)).collect()
}

/// Like `query_with_stmt`, but decodes every column of every row by name
/// (`decode_row_named`) instead of assuming column 0 is the whole result —
/// see `decode_row_named`'s doc comment.
async fn query_named_with_stmt(
    client: &tokio_postgres::Client,
    stmt: &tokio_postgres::Statement,
    params: &[DecodedValue],
    ext: &ExtensionOids,
) -> Result<Vec<DecodedValue>> {
    let bound: Vec<BoundParam<'_>> = params.iter().map(BoundParam).collect();
    let param_refs: Vec<&(dyn postgres_types::ToSql + Sync)> =
        bound.iter().map(|p| p as &(dyn postgres_types::ToSql + Sync)).collect();
    let rows = client.query(stmt, &param_refs).await?;
    rows.iter().map(|row| decode_row_named(row, ext)).collect()
}

async fn execute_with_stmt(
    client: &tokio_postgres::Client,
    stmt: &tokio_postgres::Statement,
    params: &[DecodedValue],
) -> Result<u64> {
    let bound: Vec<BoundParam<'_>> = params.iter().map(BoundParam).collect();
    let param_refs: Vec<&(dyn postgres_types::ToSql + Sync)> =
        bound.iter().map(|p| p as &(dyn postgres_types::ToSql + Sync)).collect();
    Ok(client.execute(stmt, &param_refs).await?)
}

/// Prepares through the pooled connection's own statement cache, so a query
/// run twice on the same connection costs one round trip rather than two.
///
/// `tokio_postgres::Client::prepare` always issues a PARSE/DESCRIBE round
/// trip; on a query whose execution is itself a single round trip that
/// doubles the wire cost. `deadpool`'s cache is keyed on the SQL text and
/// lives as long as the connection, which is exactly the lifetime a
/// prepared statement has server-side. Any statement it hands back can go
/// stale if the schema changes underneath it — see
/// `PgPool::clear_statement_caches`, which every migration path calls.
pub(crate) async fn query_typed_on(
    client: &deadpool_postgres::Object,
    sql: &str,
    params: &[DecodedValue],
    ext: &ExtensionOids,
) -> Result<Vec<DecodedValue>> {
    let stmt = client.prepare_cached(sql).await?;
    query_with_stmt(client, &stmt, params, ext).await
}

/// See `query_typed_on` — same caching, `decode_row_named` decoding.
pub(crate) async fn query_typed_named_on(
    client: &deadpool_postgres::Object,
    sql: &str,
    params: &[DecodedValue],
    ext: &ExtensionOids,
) -> Result<Vec<DecodedValue>> {
    let stmt = client.prepare_cached(sql).await?;
    query_named_with_stmt(client, &stmt, params, ext).await
}

/// Sets `pylon.globals`, which a database trigger reads its session globals
/// from; `local` scopes it to the current transaction.
async fn set_globals_on(client: &deadpool_postgres::Object, globals: &str, local: bool) -> Result<()> {
    let sql = if local {
        "SELECT set_config('pylon.globals', $1, true)"
    } else {
        "SELECT set_config('pylon.globals', $1, false)"
    };
    execute_typed_on(client, sql, &[DecodedValue::Str(globals.to_string())]).await?;
    Ok(())
}

/// See `query_typed_on` — same caching, result discarded.
pub(crate) async fn execute_typed_on(
    client: &deadpool_postgres::Object,
    sql: &str,
    params: &[DecodedValue],
) -> Result<u64> {
    let stmt = client.prepare_cached(sql).await?;
    execute_with_stmt(client, &stmt, params).await
}

/// The unpooled counterparts, for the LISTEN/NOTIFY connection — a bare
/// `tokio_postgres::Client` with no `deadpool` wrapper around it and so no
/// statement cache to reach. Its query volume is a handful of worker
/// statements, not the query hot path, so a plain `prepare` is fine here.
pub(crate) async fn query_typed_on_raw(
    client: &tokio_postgres::Client,
    sql: &str,
    params: &[DecodedValue],
    ext: &ExtensionOids,
) -> Result<Vec<DecodedValue>> {
    let stmt = client.prepare(sql).await?;
    query_with_stmt(client, &stmt, params, ext).await
}

pub(crate) async fn query_typed_named_on_raw(
    client: &tokio_postgres::Client,
    sql: &str,
    params: &[DecodedValue],
    ext: &ExtensionOids,
) -> Result<Vec<DecodedValue>> {
    let stmt = client.prepare(sql).await?;
    query_named_with_stmt(client, &stmt, params, ext).await
}

pub(crate) async fn execute_typed_on_raw(
    client: &tokio_postgres::Client,
    sql: &str,
    params: &[DecodedValue],
) -> Result<u64> {
    let stmt = client.prepare(sql).await?;
    execute_with_stmt(client, &stmt, params).await
}

/// `EXPLAIN (FORMAT JSON)` returns exactly one row with one column (named
/// `QUERY PLAN`, typed `json`) holding the entire plan as JSON text — reused
/// via `RawBytes` (see its own doc comment) rather than depending on the
/// `with-serde_json-1` feature, since `json`'s wire format is just its plain
/// UTF-8 text (unlike `jsonb`, which prefixes a version byte).
pub(crate) async fn query_explain_on(
    client: &deadpool_postgres::Object,
    sql: &str,
    params: &[DecodedValue],
) -> Result<String> {
    let wrapped = format!("EXPLAIN (ANALYZE, FORMAT JSON, VERBOSE) {sql}");
    let stmt = client.prepare_cached(&wrapped).await?;
    let bound: Vec<BoundParam<'_>> = params.iter().map(BoundParam).collect();
    let param_refs: Vec<&(dyn postgres_types::ToSql + Sync)> =
        bound.iter().map(|p| p as &(dyn postgres_types::ToSql + Sync)).collect();
    let rows = client.query(&stmt, &param_refs).await?;
    let row = rows
        .into_iter()
        .next()
        .ok_or_else(|| Error::message("EXPLAIN produced no output row".to_string()))?;
    let RawBytes(bytes) = row.try_get::<_, RawBytes<'_>>(0)?;
    Ok(String::from_utf8_lossy(bytes).into_owned())
}

/// An explicit transaction on a single connection checked out of the pool.
/// `deadpool-postgres`'s default recycling method (`Fast`) does *not* run
/// any reset query when a connection is returned to the pool — unlike
/// pool release, which always issues `ROLLBACK` itself if the
/// released connection still has an open transaction. That safety net has
/// to be reproduced here explicitly, or a connection released mid- or
/// aborted-transaction would silently corrupt the next borrower's session.
/// Hence `commit` rolls back on its own failure before returning the error,
/// and both `commit`/`rollback` consume `self` so the connection is only
/// ever returned to the pool (via `Drop`) once it is guaranteed to be back
/// in a clean, non-transactional state.
#[derive(Debug)]
pub struct PgTransaction {
    client: deadpool_postgres::Object,
    types: Arc<ExtensionOids>,
}

impl PgTransaction {
    /// The registry the owning pool held when this transaction began — a
    /// snapshot, as in `PgConnection::types`.
    pub fn types(&self) -> &ExtensionOids {
        &self.types
    }

    /// `query_typed`, with `pylon.globals` set for the rest of this
    /// transaction — see `PgPool::query_typed_with_globals`.
    pub async fn query_typed_with_globals(
        &self,
        sql: &str,
        params: &[DecodedValue],
        ext: &ExtensionOids,
        globals: &str,
    ) -> Result<Vec<DecodedValue>> {
        set_globals_on(&self.client, globals, true).await?;
        query_typed_on(&self.client, sql, params, ext).await
    }

    /// `execute_typed`, with `pylon.globals` set — see `query_typed_with_globals`.
    pub async fn execute_typed_with_globals(&self, sql: &str, params: &[DecodedValue], globals: &str) -> Result<u64> {
        set_globals_on(&self.client, globals, true).await?;
        execute_typed_on(&self.client, sql, params).await
    }

    pub async fn query_typed(
        &self,
        sql: &str,
        params: &[DecodedValue],
        ext: &ExtensionOids,
    ) -> Result<Vec<DecodedValue>> {
        query_typed_on(&self.client, sql, params, ext).await
    }

    pub async fn execute_typed(&self, sql: &str, params: &[DecodedValue]) -> Result<u64> {
        execute_typed_on(&self.client, sql, params).await
    }

    /// Runs `sql` via the simple query protocol — see `PgPool::batch_execute`
    /// for why migration DDL steps need this instead of `execute_typed`.
    pub async fn batch_execute(&self, sql: &str) -> Result<()> {
        self.client.batch_execute(sql).await?;
        Ok(())
    }

    /// Establishes a named savepoint inside this transaction. Used by
    /// migration execution's dev-mode rebase: a step runs inside a
    /// savepoint so an "already exists" error (from `watch` having applied
    /// the same DDL earlier) can be rolled back to just that step instead
    /// of aborting the whole outer transaction.
    pub async fn savepoint(&self, name: &str) -> Result<()> {
        self.client
            .batch_execute(&format!("SAVEPOINT {}", listener::quote_ident(name)))
            .await?;
        Ok(())
    }

    pub async fn release_savepoint(&self, name: &str) -> Result<()> {
        self.client
            .batch_execute(&format!("RELEASE SAVEPOINT {}", listener::quote_ident(name)))
            .await?;
        Ok(())
    }

    pub async fn rollback_to_savepoint(&self, name: &str) -> Result<()> {
        self.client
            .batch_execute(&format!("ROLLBACK TO SAVEPOINT {}", listener::quote_ident(name)))
            .await?;
        Ok(())
    }

    /// Commits the transaction. On failure (e.g. a serialization failure or
    /// deadlock detected at COMMIT time), best-effort rolls back first so
    /// the connection isn't returned to the pool still aborted — the
    /// original commit error is what's returned either way.
    pub async fn commit(self) -> Result<()> {
        match self.client.batch_execute("COMMIT").await {
            Ok(()) => Ok(()),
            Err(e) => {
                let _ = self.client.batch_execute("ROLLBACK").await;
                Err(e.into())
            }
        }
    }

    pub async fn rollback(self) -> Result<()> {
        self.client.batch_execute("ROLLBACK").await?;
        Ok(())
    }
}

/// Decodes a row's column 0 (the `result` column pylon-core's SQL always
/// projects) into a `DecodedValue`, using its actual declared Postgres type.
/// Column 0 itself can be SQL NULL at the top level (not just a NULL
/// *field within* a composite, which `decode_record`/`decode_array`
/// already handle) — `RawBytes` has no `from_sql_null` override, so a
/// direct `row.try_get::<_, RawBytes>(0)` errors on a null column; going
/// through `Option<RawBytes>` (which `postgres_types` implements generically
/// for any `T: FromSql`, yielding `None` for SQL NULL) avoids that.
/// Parses `dsn` and pins the session settings every Pylon connection needs.
///
/// `timezone` decides what a `timestamptz` renders as and which day
/// `date_part('day', …)` reports; `intervalstyle` decides whether an interval
/// renders as `01:30:00` or `PT1H30M`. Left to the server's configuration,
/// the same query would give different answers against two databases, and
/// `<str>` of a datetime or a duration would stop matching the ISO 8601 the
/// rest of the stack assumes. Sent as startup-packet options rather than a
/// `SET` after connecting, so a pooled connection cannot be handed out
/// before they apply.
fn session_config(dsn: &str) -> Result<tokio_postgres::Config> {
    let mut config: tokio_postgres::Config = dsn.parse()?;
    let pinned = "-c timezone=UTC -c intervalstyle=iso_8601";
    let options = match config.get_options() {
        Some(existing) => format!("{existing} {pinned}"),
        None => pinned.to_string(),
    };
    config.options(options);
    Ok(config)
}

/// Runs `wire::TYPE_DISCOVERY_SQL` on `client` and builds the registry from
/// it. Shared by `PgPool::connect` and `PgListener::connect` so a pooled
/// connection and a dedicated listener classify types identically.
pub(crate) async fn discover_types(client: &tokio_postgres::Client) -> Result<ExtensionOids> {
    let rows = client.query(wire::TYPE_DISCOVERY_SQL, &[]).await?;
    Ok(ExtensionOids::from_discovery_rows(rows.iter().map(|r| {
        let oid: i64 = r.get(0);
        let typtype: String = r.get(1);
        let base_oid: i64 = r.get(2);
        let typname: String = r.get(3);
        (oid as u32, typtype, base_oid as u32, typname)
    })))
}

fn decode_result_column(row: &tokio_postgres::Row, ext: &ExtensionOids) -> Result<DecodedValue> {
    let oid = row.columns()[0].type_().oid();
    match row.try_get::<_, Option<RawBytes>>(0)? {
        None => Ok(DecodedValue::Null),
        Some(RawBytes(bytes)) => wire::decode_value(oid, bytes, ext),
    }
}

/// Decodes *every* column of `row`, keyed by name, as a `DecodedValue::Object`
/// — the general named-column decode path, unlike
/// `decode_result_column` (which only ever decodes column 0, matching
/// pylon-core's own single-composite-column SQL emission convention). For
/// hand-written queries with several named columns a caller accesses by
/// name (`row["col"]`) — `pylon.worker`/`pylon.vector`/`pylon.search`'s
/// index-outbox queries, not PyQL-compiled SQL.
fn decode_row_named(row: &tokio_postgres::Row, ext: &ExtensionOids) -> Result<DecodedValue> {
    let mut fields = Vec::with_capacity(row.columns().len());
    for (i, col) in row.columns().iter().enumerate() {
        let oid = col.type_().oid();
        let value = match row.try_get::<_, Option<RawBytes>>(i)? {
            None => DecodedValue::Null,
            Some(RawBytes(bytes)) => wire::decode_value(oid, bytes, ext)?,
        };
        fields.push((col.name().to_string(), value));
    }
    Ok(DecodedValue::Object(fields))
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Real Postgres required — `PYLON_PGCON_TEST_DSN` must be set (no
    /// hardcoded fallback; point it at a disposable database, never a real
    /// one). Not run by default (`cargo test -- --ignored` to opt in) so
    /// the default test run stays hermetic.
    fn test_dsn() -> String {
        std::env::var("PYLON_PGCON_TEST_DSN").expect("PYLON_PGCON_TEST_DSN must be set to run live-Postgres tests")
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn connects_and_round_trips_a_scalar_query() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let rows = pool.query_raw("SELECT 1 + 1").await.unwrap();
        assert_eq!(rows.len(), 1);
        let value: i32 = rows[0].get(0);
        assert_eq!(value, 2);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn pool_is_reused_across_multiple_queries() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        for i in 0..5 {
            let rows = pool.query_raw(&format!("SELECT {i}")).await.unwrap();
            let value: i32 = rows[0].get(0);
            assert_eq!(value, i);
        }
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn invalid_dsn_fails_to_connect() {
        let result = PgPool::connect("not-a-valid-dsn", 5).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn connect_fails_eagerly_against_a_nonexistent_database() {
        // A well-formed DSN pointing at a database that doesn't exist must
        // fail right here, not lazily on the first query — matching
        // the eager-connect behavior, which
        // `Client.ensure_connected()` depends on to map this straight to
        // `ConnectionFailedError` instead of silently deferring the
        // failure past `ensure_connected()` returning successfully.
        // Swap out whatever the real database name is (not a hardcoded
        // substring — that silently no-ops and leaves this pointed at the
        // real, existing test database if the DSN's db name ever changes)
        // for one guaranteed not to exist.
        let dsn = test_dsn();
        let (prefix, _db) = dsn.rsplit_once('/').expect("DSN must have a database path segment");
        let bad_dsn = format!("{prefix}/pgcon_definitely_does_not_exist");
        let result = PgPool::connect(&bad_dsn, 5).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn bad_sql_returns_an_error_not_a_panic() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let result = pool.query_raw("SELECT this is not valid sql").await;
        assert!(result.is_err());
    }

    // ── query_composite against real Postgres wire data ─────────────────
    //
    // The wire::tests module already checks decode_value against
    // hand-crafted byte buffers — these tests instead run real SQL through
    // a real connection, so any mismatch between my assumptions about
    // Postgres's binary format and what Postgres actually sends shows up
    // here, not just in self-consistent hand-rolled fixtures.

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_a_bare_scalar_result_column() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let rows = pool
            .query_composite("SELECT 42::int8 AS result", &ExtensionOids::default())
            .await
            .unwrap();
        assert_eq!(rows, vec![DecodedValue::I64(42)]);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_a_composite_matching_pylon_cores_own_emission_shape() {
        // Mirrors exactly what `sql/mod.rs::emit_bound_select` emits:
        // `SELECT (type_disc, col1, col2, ...) AS result`.
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let sql = "SELECT ('Person'::text, 'Alice'::text, 30::int8, NULL::text) AS result";
        let rows = pool.query_composite(sql, &ExtensionOids::default()).await.unwrap();
        assert_eq!(
            rows,
            vec![DecodedValue::Composite(vec![
                DecodedValue::Str("Person".into()),
                DecodedValue::Str("Alice".into()),
                DecodedValue::I64(30),
                DecodedValue::Null,
            ])]
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_nested_composite_and_array_of_composite_for_real() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let sql = "SELECT (\
            'Product'::text, \
            ROW('Tag'::text, 'sale'::text), \
            ARRAY[ROW(1::int8), ROW(2::int8)]::record[]\
        ) AS result";
        let rows = pool.query_composite(sql, &ExtensionOids::default()).await.unwrap();
        assert_eq!(
            rows,
            vec![DecodedValue::Composite(vec![
                DecodedValue::Str("Product".into()),
                DecodedValue::Composite(vec![DecodedValue::Str("Tag".into()), DecodedValue::Str("sale".into())]),
                DecodedValue::Array(vec![
                    DecodedValue::Composite(vec![DecodedValue::I64(1)]),
                    DecodedValue::Composite(vec![DecodedValue::I64(2)]),
                ]),
            ])]
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_array_of_text_for_real() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let sql = "SELECT (ARRAY['a', 'b', NULL]::text[]) AS result";
        let rows = pool.query_composite(sql, &ExtensionOids::default()).await.unwrap();
        assert_eq!(
            rows,
            vec![DecodedValue::Array(vec![
                DecodedValue::Str("a".into()),
                DecodedValue::Str("b".into()),
                DecodedValue::Null,
            ])]
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_numeric_and_jsonb_and_uuid_for_real() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let sql = "SELECT (\
            12.50::numeric, \
            '{\"a\": 1, \"b\": [1,2]}'::jsonb, \
            '11111111-1111-1111-1111-111111111111'::uuid\
        ) AS result";
        let rows = pool.query_composite(sql, &ExtensionOids::default()).await.unwrap();
        let DecodedValue::Composite(fields) = &rows[0] else {
            panic!("expected Composite")
        };
        assert_eq!(fields[0], DecodedValue::Decimal("12.50".to_string()));
        assert_eq!(
            fields[1],
            DecodedValue::Object(vec![
                ("a".into(), DecodedValue::I64(1)),
                (
                    "b".into(),
                    DecodedValue::Array(vec![DecodedValue::I64(1), DecodedValue::I64(2)])
                ),
            ])
        );
        assert_eq!(fields[2], DecodedValue::Uuid([0x11; 16]));
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_bytea_for_real() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let rows = pool
            .query_composite("SELECT '\\xdeadbeef'::bytea AS result", &ExtensionOids::default())
            .await
            .unwrap();
        assert_eq!(rows, vec![DecodedValue::Bytes(vec![0xde, 0xad, 0xbe, 0xef])]);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_enum_cast_to_text_for_real() {
        // pylon-core always ::text-casts enum-typed columns (sql/mod.rs's
        // emit_scalar) specifically so the runtime-assigned enum OID never
        // needs to be known statically — this is the actually-exercised path.
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let sql = "DO $$ BEGIN CREATE TYPE pgcon_test_enum AS ENUM ('a', 'b'); \
                   EXCEPTION WHEN duplicate_object THEN NULL; END $$;";
        pool.query_raw(sql).await.ok();
        let rows = pool
            .query_composite(
                "SELECT ('a'::pgcon_test_enum::text) AS result",
                &ExtensionOids::default(),
            )
            .await
            .unwrap();
        assert_eq!(rows, vec![DecodedValue::Str("a".to_string())]);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_a_real_pgvector_value_through_connect_time_discovery() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        if pool.query_raw("CREATE EXTENSION IF NOT EXISTS vector").await.is_err() {
            eprintln!("skipping: pgvector not installable on this server");
            return;
        }
        // The extension may have been created after this pool connected.
        pool.refresh_types().await.unwrap();
        assert!(
            pool.types().vector.is_some(),
            "discovery should have found the vector OID"
        );

        // Nested in a record, exactly as pylon-core emits every result.
        let rows = pool
            .query_composite("SELECT ('doc', '[1.5,2.5]'::vector) AS result", &pool.types())
            .await
            .unwrap();

        let DecodedValue::Composite(fields) = &rows[0] else {
            panic!("expected Composite, got {:?}", rows[0])
        };
        assert_eq!(fields[0], DecodedValue::Str("doc".to_string()));
        assert_eq!(
            fields[1],
            DecodedValue::Array(vec![DecodedValue::F64(1.5), DecodedValue::F64(2.5)]),
            "a vector must decode to its floats, not to mojibake or an error"
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_a_real_enum_and_domain_through_connect_time_discovery() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw(
            "DO $$ BEGIN CREATE TYPE pgcon_disc_enum AS ENUM ('x', 'y'); \
             EXCEPTION WHEN duplicate_object THEN NULL; END $$;",
        )
        .await
        .ok();
        pool.query_raw(
            "DO $$ BEGIN CREATE DOMAIN pgcon_disc_domain AS int8 CHECK (VALUE > 0); \
             EXCEPTION WHEN duplicate_object THEN NULL; END $$;",
        )
        .await
        .ok();

        pool.refresh_types().await.unwrap();

        // Neither is ::text-cast here, so both arrive with their real
        // database-assigned OID and can only decode via the registry.
        let rows = pool
            .query_composite(
                "SELECT ('x'::pgcon_disc_enum, 42::pgcon_disc_domain) AS result",
                &pool.types(),
            )
            .await
            .unwrap();

        let DecodedValue::Composite(fields) = &rows[0] else {
            panic!("expected Composite, got {:?}", rows[0])
        };
        assert_eq!(fields[0], DecodedValue::Str("x".to_string()));
        assert_eq!(
            fields[1],
            DecodedValue::I64(42),
            "a domain must decode as its base type, not as text"
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn decodes_a_real_named_composite_through_connect_time_discovery() {
        // A tuple's column type: the type is the database's own, so its OID
        // is database-assigned and only the registry can say it is readable.
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw(
            "DO $$ BEGIN CREATE TYPE pgcon_disc_tuple AS (\"amount\" numeric, \"note\" text); \
             EXCEPTION WHEN duplicate_object THEN NULL; END $$;",
        )
        .await
        .ok();

        pool.refresh_types().await.unwrap();

        let rows = pool
            .query_composite(
                "SELECT (
                     ROW(12.3400, 'x')::pgcon_disc_tuple,
                     ARRAY[ROW(1.500, 'y')::pgcon_disc_tuple]
                 ) AS result",
                &pool.types(),
            )
            .await
            .unwrap();

        let DecodedValue::Composite(fields) = &rows[0] else {
            panic!("expected Composite, got {:?}", rows[0])
        };
        assert_eq!(
            fields[0],
            DecodedValue::Composite(vec![
                DecodedValue::Decimal("12.3400".to_string()),
                DecodedValue::Str("x".to_string()),
            ]),
            "each member keeps its own type, scale included"
        );
        assert_eq!(
            fields[1],
            DecodedValue::Array(vec![DecodedValue::Composite(vec![
                DecodedValue::Decimal("1.500".to_string()),
                DecodedValue::Str("y".to_string()),
            ])]),
            "an array of them reads element-wise"
        );
    }

    // ── a registry that goes stale underneath a running pool ────────────

    /// Read as `int8`, so this needs no registry of its own.
    async fn live_type_oid(pool: &PgPool, typname: &str) -> u32 {
        let rows = pool
            .query_typed(
                "SELECT (oid::int8) AS result FROM pg_type WHERE typname = $1",
                &[DecodedValue::Str(typname.to_string())],
                &ExtensionOids::default(),
            )
            .await
            .unwrap();
        match rows.first() {
            Some(DecodedValue::I64(oid)) => *oid as u32,
            other => panic!("expected one int8 oid for {typname}, got {other:?}"),
        }
    }

    /// Recreating a type reproduces what a restore does to every OID at
    /// once: Postgres never reissues the one it just freed.
    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn a_type_whose_oid_moved_under_a_live_pool_still_decodes() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.batch_execute(
            "DROP TYPE IF EXISTS pgcon_moved_enum; \
             CREATE TYPE pgcon_moved_enum AS ENUM ('a', 'b')",
        )
        .await
        .unwrap();
        // Stand in for a pool that connected while this type existed.
        pool.refresh_types().await.unwrap();
        let stale = pool.types();
        let before = live_type_oid(&pool, "pgcon_moved_enum").await;
        assert!(
            stale.enums.contains(&before),
            "the pool must start out knowing the pre-move OID, or this test proves nothing"
        );

        pool.batch_execute(
            "DROP TYPE pgcon_moved_enum; \
             CREATE TYPE pgcon_moved_enum AS ENUM ('a', 'b')",
        )
        .await
        .unwrap();
        let after = live_type_oid(&pool, "pgcon_moved_enum").await;
        assert_ne!(before, after, "recreating a type must assign it a fresh OID");

        // Before: the registry the pool is holding can't decode the new OID.
        let err = decode_value(after, b"a", &stale).unwrap_err();
        assert!(
            matches!(err, Error::UnknownTypeOid { oid } if oid == after),
            "expected UnknownTypeOid for the moved OID, got {err}"
        );

        // After: `pool.types()` is still that stale registry when it's
        // passed, and the query succeeds anyway.
        let rows = pool
            .query_composite("SELECT ('a'::pgcon_moved_enum) AS result", &pool.types())
            .await
            .unwrap();
        assert_eq!(rows, vec![DecodedValue::Str("a".to_string())]);
        assert!(
            pool.types().enums.contains(&after),
            "the heal must leave the new OID in the pool's registry, not just decode one query"
        );
    }

    /// Consumers hold clones, so a refresh that reached only the handle it
    /// was called on would leave the clone serving queries stale.
    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn a_refresh_through_one_handle_is_visible_through_every_clone() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.batch_execute("DROP TYPE IF EXISTS pgcon_clone_enum")
            .await
            .unwrap();
        pool.refresh_types().await.unwrap();

        let clone = pool.clone();
        pool.batch_execute("CREATE TYPE pgcon_clone_enum AS ENUM ('a')")
            .await
            .unwrap();
        let oid = live_type_oid(&pool, "pgcon_clone_enum").await;
        assert!(!clone.types().enums.contains(&oid), "the clone must start out stale");

        clone.refresh_types().await.unwrap();

        assert!(pool.types().enums.contains(&oid), "the original must see the refresh");
        let tx = pool.begin_default().await.unwrap();
        assert!(
            tx.types().enums.contains(&oid),
            "a transaction opened after the refresh must inherit the refreshed registry"
        );
        tx.rollback().await.unwrap();
        pool.batch_execute("DROP TYPE pgcon_clone_enum").await.unwrap();
    }

    // ── query_typed: bound-parameter round trips against real Postgres ──
    //
    // Each test binds a DecodedValue as $1, has Postgres echo it straight
    // back out (so both encode_value AND decode_value are exercised in one
    // pass — a mismatch in either direction fails the assertion), matching
    // exactly how a real PyQL query binds a param and gets a result back.

    async fn round_trip(pool: &PgPool, pg_type: &str, param: DecodedValue) -> DecodedValue {
        let sql = format!("SELECT ($1::{pg_type}) AS result");
        let rows = pool
            .query_typed(&sql, &[param], &ExtensionOids::default())
            .await
            .unwrap();
        rows.into_iter().next().unwrap()
    }

    // ── query_explain against real Postgres ─────────────────────────────

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn query_explain_returns_parseable_json_with_a_plan_node() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let raw = pool.query_explain("SELECT 1 + 1", &[]).await.unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&raw).unwrap();
        assert!(parsed[0]["Plan"]["Node Type"].is_string());
        // ANALYZE was requested, so the plan must carry real execution stats.
        assert!(parsed[0]["Plan"]["Actual Total Time"].is_number());
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn query_explain_binds_params_the_same_way_query_typed_does() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let raw = pool
            .query_explain("SELECT $1::int8 + 1", &[DecodedValue::I64(41)])
            .await
            .unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&raw).unwrap();
        assert!(parsed[0]["Plan"]["Node Type"].is_string());
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_bool_param() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        assert_eq!(
            round_trip(&pool, "bool", DecodedValue::Bool(true)).await,
            DecodedValue::Bool(true)
        );
        assert_eq!(
            round_trip(&pool, "bool", DecodedValue::Bool(false)).await,
            DecodedValue::Bool(false)
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_integer_params_at_every_width() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        assert_eq!(
            round_trip(&pool, "int2", DecodedValue::I64(30)).await,
            DecodedValue::I64(30)
        );
        assert_eq!(
            round_trip(&pool, "int4", DecodedValue::I64(70_000)).await,
            DecodedValue::I64(70_000)
        );
        assert_eq!(
            round_trip(&pool, "int8", DecodedValue::I64(9_223_372_036_854_775_807)).await,
            DecodedValue::I64(9_223_372_036_854_775_807)
        );
        assert_eq!(
            round_trip(&pool, "int8", DecodedValue::I64(-1)).await,
            DecodedValue::I64(-1)
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_float_params() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        assert_eq!(
            round_trip(&pool, "float4", DecodedValue::F64(1.5)).await,
            DecodedValue::F64(1.5)
        );
        assert_eq!(
            round_trip(&pool, "float8", DecodedValue::F64(2.25)).await,
            DecodedValue::F64(2.25)
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_text_param() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        assert_eq!(
            round_trip(&pool, "text", DecodedValue::Str("héllo 🎉".to_string())).await,
            DecodedValue::Str("héllo 🎉".to_string())
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_bytea_param() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        assert_eq!(
            round_trip(&pool, "bytea", DecodedValue::Bytes(vec![1, 2, 3, 255])).await,
            DecodedValue::Bytes(vec![1, 2, 3, 255])
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_uuid_param() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let bytes = [0x11u8; 16];
        assert_eq!(
            round_trip(&pool, "uuid", DecodedValue::Uuid(bytes)).await,
            DecodedValue::Uuid(bytes)
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn binds_a_plain_string_as_a_uuid_param() {
        // A JSON API request body (see `pylon-server`'s /api/<connection>/query
        // handler) necessarily carries a UUID query parameter as plain text —
        // there's no JSON "uuid" type — so it arrives as `DecodedValue::Str`,
        // not `::Uuid`. Regression test for a real bug: binding that Str
        // directly against a `uuid`-typed parameter used to send raw UTF-8
        // text bytes for a binary-format column, which Postgres rejected
        // with "incorrect binary data format in bind parameter 1".
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let as_string = DecodedValue::Str("11111111-1111-1111-1111-111111111111".to_string());
        assert_eq!(
            round_trip(&pool, "uuid", as_string).await,
            DecodedValue::Uuid([0x11; 16])
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn binds_a_plain_string_as_a_jsonb_param() {
        // Regression test for a real bug: `migration apply`'s db_state
        // snapshot update binds already-serialized JSON text (a plain Rust
        // `String`, not a `DecodedValue::Object`) as `$1::jsonb` — the same
        // class of bug as the UUID case above (a Str value bound against a
        // non-text binary-format target type).
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let as_string = DecodedValue::Str(r#"{"a":1,"b":[1,2]}"#.to_string());
        assert_eq!(
            round_trip(&pool, "jsonb", as_string).await,
            DecodedValue::Object(vec![
                ("a".into(), DecodedValue::I64(1)),
                (
                    "b".into(),
                    DecodedValue::Array(vec![DecodedValue::I64(1), DecodedValue::I64(2)])
                ),
            ])
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_numeric_param() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        assert_eq!(
            round_trip(&pool, "numeric", DecodedValue::Decimal("12.50".to_string())).await,
            DecodedValue::Decimal("12.50".to_string())
        );
        assert_eq!(
            round_trip(&pool, "numeric", DecodedValue::Decimal("-9999.001".to_string())).await,
            DecodedValue::Decimal("-9999.001".to_string())
        );

        // Values a fixed 96-bit/28-place carrier cannot hold. Each of these
        // used to come back scaled by a power of ten, or refuse to serialize,
        // with `numeric` itself perfectly able to store them. The first is
        // what Python's `Decimal(0.0000004)` stringifies to.
        for text in [
            "0.00000039999999999999998189924473035450347424557548947632312774658203125",
            "0.00000012222222222222222222222222222222222222222",
            "12222222222222222222222222222222000000000",
            "100000000000000000000000000000",
            "123456789012345678901234567890.123456789012345678901234567890",
        ] {
            assert_eq!(
                round_trip(&pool, "numeric", DecodedValue::Decimal(text.to_string())).await,
                DecodedValue::Decimal(text.to_string()),
                "{text}"
            );
        }

        // A JSON request body has no decimal type, so a numeric parameter can
        // arrive as a number or as text — both reach the same encoder.
        assert_eq!(
            round_trip(&pool, "numeric", DecodedValue::Str("0.0000004".to_string())).await,
            DecodedValue::Decimal("0.0000004".to_string())
        );
        assert_eq!(
            round_trip(&pool, "numeric", DecodedValue::I64(-12345)).await,
            DecodedValue::Decimal("-12345".to_string())
        );
        assert_eq!(
            round_trip(&pool, "numeric", DecodedValue::F64(0.1)).await,
            DecodedValue::Decimal("0.1".to_string())
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_null_param() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        assert_eq!(round_trip(&pool, "int8", DecodedValue::Null).await, DecodedValue::Null);
        assert_eq!(round_trip(&pool, "text", DecodedValue::Null).await, DecodedValue::Null);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_array_param() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let param = DecodedValue::Array(vec![
            DecodedValue::Str("a".into()),
            DecodedValue::Str("b".into()),
            DecodedValue::Null,
        ]);
        assert_eq!(round_trip(&pool, "text[]", param.clone()).await, param);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_int_array_param() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let param = DecodedValue::Array(vec![DecodedValue::I64(1), DecodedValue::I64(2), DecodedValue::I64(3)]);
        assert_eq!(round_trip(&pool, "int8[]", param.clone()).await, param);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn round_trips_jsonb_object_param() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let param = DecodedValue::Object(vec![
            ("a".into(), DecodedValue::I64(1)),
            ("b".into(), DecodedValue::Str("two".into())),
            (
                "c".into(),
                DecodedValue::Array(vec![DecodedValue::I64(1), DecodedValue::I64(2)]),
            ),
        ]);
        assert_eq!(round_trip(&pool, "jsonb", param.clone()).await, param);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn query_typed_matches_pylon_cores_own_param_binding_convention() {
        // $1, $2, ... positional, matching multiple params in one query —
        // the same shape a real PyQL query with several kwargs produces.
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let sql = "SELECT ($1::text, $2::int8, $3::bool) AS result";
        let params = vec![
            DecodedValue::Str("Alice".into()),
            DecodedValue::I64(30),
            DecodedValue::Bool(true),
        ];
        let rows = pool.query_typed(sql, &params, &ExtensionOids::default()).await.unwrap();
        assert_eq!(
            rows,
            vec![DecodedValue::Composite(vec![
                DecodedValue::Str("Alice".into()),
                DecodedValue::I64(30),
                DecodedValue::Bool(true),
            ])]
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn wrong_param_count_returns_an_error_not_a_panic() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let result = pool
            .query_typed(
                "SELECT $1::int8, $2::int8",
                &[DecodedValue::I64(1)],
                &ExtensionOids::default(),
            )
            .await;
        assert!(result.is_err());
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn execute_typed_runs_a_mutation_and_reports_affected_rows() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw("CREATE TEMP TABLE IF NOT EXISTS pgcon_execute_test (id int8, name text)")
            .await
            .unwrap();

        let inserted = pool
            .execute_typed(
                "INSERT INTO pgcon_execute_test (id, name) VALUES ($1::int8, $2::text)",
                &[DecodedValue::I64(1), DecodedValue::Str("alice".into())],
            )
            .await
            .unwrap();
        assert_eq!(inserted, 1);

        let updated = pool
            .execute_typed(
                "UPDATE pgcon_execute_test SET name = $1::text WHERE id = $2::int8",
                &[DecodedValue::Str("bob".into()), DecodedValue::I64(1)],
            )
            .await
            .unwrap();
        assert_eq!(updated, 1);

        let rows = pool
            .query_composite(
                "SELECT (name) AS result FROM pgcon_execute_test",
                &ExtensionOids::default(),
            )
            .await
            .unwrap();
        assert_eq!(rows, vec![DecodedValue::Str("bob".to_string())]);
    }

    // ── Error::sqlstate() against real Postgres constraint violations ──
    //
    // Error mapping to Pylon's Python exception hierarchy (a later phase)
    // classifies on these codes (`"23505"` unique violation, etc.)
    // already do today — verified against a real server response, not
    // assumed from the SQLSTATE spec alone.

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn unique_violation_reports_23505() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw("CREATE TEMP TABLE pgcon_unique_test (id int8 PRIMARY KEY)")
            .await
            .unwrap();
        pool.execute_typed(
            "INSERT INTO pgcon_unique_test (id) VALUES ($1::int8)",
            &[DecodedValue::I64(1)],
        )
        .await
        .unwrap();

        let err = pool
            .execute_typed(
                "INSERT INTO pgcon_unique_test (id) VALUES ($1::int8)",
                &[DecodedValue::I64(1)],
            )
            .await
            .unwrap_err();
        assert_eq!(err.sqlstate(), Some(&tokio_postgres::error::SqlState::UNIQUE_VIOLATION));
        assert_eq!(err.sqlstate().unwrap().code(), "23505");
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn foreign_key_violation_reports_23503() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw("CREATE TEMP TABLE pgcon_fk_parent (id int8 PRIMARY KEY)")
            .await
            .unwrap();
        pool.query_raw("CREATE TEMP TABLE pgcon_fk_child (parent_id int8 REFERENCES pgcon_fk_parent(id))")
            .await
            .unwrap();

        let err = pool
            .execute_typed(
                "INSERT INTO pgcon_fk_child (parent_id) VALUES ($1::int8)",
                &[DecodedValue::I64(999)],
            )
            .await
            .unwrap_err();
        assert_eq!(
            err.sqlstate(),
            Some(&tokio_postgres::error::SqlState::FOREIGN_KEY_VIOLATION)
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn check_violation_reports_23514() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw("CREATE TEMP TABLE pgcon_check_test (age int8 CHECK (age >= 0))")
            .await
            .unwrap();

        let err = pool
            .execute_typed(
                "INSERT INTO pgcon_check_test (age) VALUES ($1::int8)",
                &[DecodedValue::I64(-1)],
            )
            .await
            .unwrap_err();
        assert_eq!(err.sqlstate(), Some(&tokio_postgres::error::SqlState::CHECK_VIOLATION));
        // Temp tables live in a session-specific `pg_temp_N` schema, so only
        // the table name (not the exact schema) is asserted here.
        assert_eq!(err.violated_table().map(|(_, table)| table), Some("pgcon_check_test"));
        assert_eq!(err.violated_scalar(), None);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn domain_check_violation_reports_the_domain_name_not_the_constraint_name() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw(
            "DO $$ BEGIN CREATE DOMAIN pgcon_rating AS int8 CHECK (VALUE BETWEEN 1 AND 5); \
             EXCEPTION WHEN duplicate_object THEN NULL; END $$",
        )
        .await
        .unwrap();
        pool.query_raw("CREATE TEMP TABLE pgcon_domain_check_test (rating pgcon_rating)")
            .await
            .unwrap();

        let err = pool
            .execute_typed(
                "INSERT INTO pgcon_domain_check_test (rating) VALUES ($1::pgcon_rating)",
                &[DecodedValue::I64(99)],
            )
            .await
            .unwrap_err();
        assert_eq!(err.sqlstate(), Some(&tokio_postgres::error::SqlState::CHECK_VIOLATION));
        assert_eq!(err.violated_scalar(), Some(("public", "pgcon_rating")));
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn syntax_error_has_no_sqlstate_matching_constraint_codes() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let err = pool.query_raw("SELECT this is not valid sql").await.unwrap_err();
        assert_ne!(err.sqlstate(), Some(&tokio_postgres::error::SqlState::UNIQUE_VIOLATION));
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn connection_pool_error_has_no_sqlstate() {
        // A bad DSN never reaches Postgres at all — no SQLSTATE to report,
        // unlike a real server-side rejection.
        let result = PgPool::connect("not-a-valid-dsn", 5).await;
        let err = result.unwrap_err();
        assert_eq!(err.sqlstate(), None);
    }

    // ── PgTransaction: begin/commit/rollback on real Postgres ───────────

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn committed_transaction_persists_its_writes() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw("CREATE TEMP TABLE pgcon_tx_commit_test (id int8 PRIMARY KEY)")
            .await
            .unwrap();

        let tx = pool.begin("serializable").await.unwrap();
        tx.execute_typed(
            "INSERT INTO pgcon_tx_commit_test (id) VALUES ($1::int8)",
            &[DecodedValue::I64(1)],
        )
        .await
        .unwrap();
        tx.commit().await.unwrap();

        let rows = pool
            .query_composite(
                "SELECT (id) AS result FROM pgcon_tx_commit_test",
                &ExtensionOids::default(),
            )
            .await
            .unwrap();
        assert_eq!(rows, vec![DecodedValue::I64(1)]);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn rolled_back_transaction_discards_its_writes() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw("CREATE TEMP TABLE pgcon_tx_rollback_test (id int8 PRIMARY KEY)")
            .await
            .unwrap();

        let tx = pool.begin("serializable").await.unwrap();
        tx.execute_typed(
            "INSERT INTO pgcon_tx_rollback_test (id) VALUES ($1::int8)",
            &[DecodedValue::I64(1)],
        )
        .await
        .unwrap();
        tx.rollback().await.unwrap();

        let rows = pool
            .query_composite(
                "SELECT (id) AS result FROM pgcon_tx_rollback_test",
                &ExtensionOids::default(),
            )
            .await
            .unwrap();
        assert!(rows.is_empty());
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn begin_actually_sets_the_requested_isolation_level() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        for (level, expected) in [
            ("read_committed", "read committed"),
            ("repeatable_read", "repeatable read"),
            ("serializable", "serializable"),
        ] {
            let tx = pool.begin(level).await.unwrap();
            let rows = tx
                .query_typed(
                    "SELECT (current_setting('transaction_isolation')) AS result",
                    &[],
                    &ExtensionOids::default(),
                )
                .await
                .unwrap();
            assert_eq!(rows, vec![DecodedValue::Str(expected.to_string())]);
            tx.rollback().await.unwrap();
        }
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn begin_rejects_an_unknown_isolation_level() {
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        let result = pool.begin("not_a_real_level").await;
        assert!(result.is_err());
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn a_pooled_connection_is_reusable_after_commit_and_after_rollback() {
        // Guards the exact hazard begin()/PgTransaction's doc comment
        // describes: deadpool's default Fast recycling does nothing to a
        // connection returned mid-transaction, so if commit/rollback ever
        // failed to leave the session clean, this small pool (max_size 1)
        // would hang forever on the second `begin()` waiting for a
        // connection that never becomes usable again.
        let pool = PgPool::connect(&test_dsn(), 1).await.unwrap();

        let tx = pool.begin("serializable").await.unwrap();
        tx.commit().await.unwrap();

        let tx = pool.begin("serializable").await.unwrap();
        tx.rollback().await.unwrap();

        let rows = pool.query_raw("SELECT 1").await.unwrap();
        assert_eq!(rows.len(), 1);
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn failed_commit_leaves_the_connection_reusable() {
        // Forces a real 40001 serialization failure at COMMIT time (the
        // same interleaving as `serializable_transactions_conflict_with_40001`
        // below), on a pool sized to exactly the two connections both
        // transactions occupy, then drains the pool with fresh queries to
        // prove every connection — including the one that failed COMMIT —
        // comes back healthy. Without `commit`'s best-effort
        // ROLLBACK-on-failure, the failed connection would still be
        // aborted server-side and the next query to land on it would
        // immediately fail with "current transaction is aborted".
        let pool = PgPool::connect(&test_dsn(), 2).await.unwrap();
        pool.query_raw("DROP TABLE IF EXISTS pgcon_tx_failed_commit_test")
            .await
            .unwrap();
        pool.query_raw("CREATE TABLE pgcon_tx_failed_commit_test (class int8, value int8)")
            .await
            .unwrap();
        pool.execute_typed(
            "INSERT INTO pgcon_tx_failed_commit_test (class, value) VALUES ($1::int8, $2::int8), ($3::int8, $4::int8)",
            &[
                DecodedValue::I64(1),
                DecodedValue::I64(10),
                DecodedValue::I64(2),
                DecodedValue::I64(20),
            ],
        )
        .await
        .unwrap();

        let tx1 = pool.begin("serializable").await.unwrap();
        let tx2 = pool.begin("serializable").await.unwrap();

        tx1.query_typed(
            "SELECT (sum(value)) AS result FROM pgcon_tx_failed_commit_test WHERE class = 1::int8",
            &[],
            &ExtensionOids::default(),
        )
        .await
        .unwrap();
        tx2.query_typed(
            "SELECT (sum(value)) AS result FROM pgcon_tx_failed_commit_test WHERE class = 2::int8",
            &[],
            &ExtensionOids::default(),
        )
        .await
        .unwrap();
        tx1.execute_typed(
            "INSERT INTO pgcon_tx_failed_commit_test (class, value) VALUES (2::int8, $1::int8)",
            &[DecodedValue::I64(10)],
        )
        .await
        .unwrap();
        tx2.execute_typed(
            "INSERT INTO pgcon_tx_failed_commit_test (class, value) VALUES (1::int8, $1::int8)",
            &[DecodedValue::I64(20)],
        )
        .await
        .unwrap();

        tx1.commit().await.unwrap();
        let commit_result = tx2.commit().await;
        assert!(commit_result.is_err());

        // Both pooled connections are back now (tx1 released on success,
        // tx2 released on Drop after the failed commit) — round-trip each.
        for _ in 0..2 {
            let rows = pool.query_raw("SELECT 1").await.unwrap();
            assert_eq!(rows.len(), 1);
        }
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn serializable_transactions_conflict_with_40001() {
        // The canonical serialization-anomaly example from the Postgres
        // docs (13.2.3): two SERIALIZABLE transactions each read one
        // class's total, then insert a row into the *other* class based on
        // what they read. Run concurrently with each SELECT completing
        // before either INSERT, this is guaranteed to leave one commit
        // rejected with 40001 — this is the exact SQLSTATE `pgcon_err`
        // (pylon-py/src/pgcon.rs) maps to `TransactionSerializationError`.
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw("DROP TABLE IF EXISTS pgcon_serialization_test")
            .await
            .unwrap();
        pool.query_raw("CREATE TABLE pgcon_serialization_test (class int8, value int8)")
            .await
            .unwrap();
        pool.execute_typed(
            "INSERT INTO pgcon_serialization_test (class, value) VALUES ($1::int8, $2::int8), ($3::int8, $4::int8)",
            &[
                DecodedValue::I64(1),
                DecodedValue::I64(10),
                DecodedValue::I64(2),
                DecodedValue::I64(20),
            ],
        )
        .await
        .unwrap();

        let tx1 = pool.begin("serializable").await.unwrap();
        let tx2 = pool.begin("serializable").await.unwrap();

        tx1.query_typed(
            "SELECT (sum(value)) AS result FROM pgcon_serialization_test WHERE class = 1::int8",
            &[],
            &ExtensionOids::default(),
        )
        .await
        .unwrap();
        tx2.query_typed(
            "SELECT (sum(value)) AS result FROM pgcon_serialization_test WHERE class = 2::int8",
            &[],
            &ExtensionOids::default(),
        )
        .await
        .unwrap();

        tx1.execute_typed(
            "INSERT INTO pgcon_serialization_test (class, value) VALUES (2::int8, $1::int8)",
            &[DecodedValue::I64(10)],
        )
        .await
        .unwrap();
        tx2.execute_typed(
            "INSERT INTO pgcon_serialization_test (class, value) VALUES (1::int8, $1::int8)",
            &[DecodedValue::I64(20)],
        )
        .await
        .unwrap();

        tx1.commit().await.unwrap();
        let err = tx2.commit().await.unwrap_err();
        assert_eq!(
            err.sqlstate(),
            Some(&tokio_postgres::error::SqlState::T_R_SERIALIZATION_FAILURE)
        );
    }

    #[tokio::test]
    #[ignore = "requires a live Postgres via PYLON_PGCON_TEST_DSN"]
    async fn concurrent_transactions_deadlock_with_40p01() {
        // Classic reproducible deadlock: two transactions lock two rows in
        // opposite order. tx1 locks row 1 then blocks on row 2; tx2 locks
        // row 2 then blocks on row 1 — Postgres's deadlock detector aborts
        // one of them with 40P01, `pgcon_err`'s other mapped SQLSTATE
        // (-> `TransactionDeadlockError`).
        let pool = PgPool::connect(&test_dsn(), 5).await.unwrap();
        pool.query_raw("DROP TABLE IF EXISTS pgcon_deadlock_test")
            .await
            .unwrap();
        pool.query_raw("CREATE TABLE pgcon_deadlock_test (id int8 PRIMARY KEY, value int8)")
            .await
            .unwrap();
        pool.execute_typed(
            "INSERT INTO pgcon_deadlock_test (id, value) VALUES ($1::int8, $2::int8), ($3::int8, $4::int8)",
            &[
                DecodedValue::I64(1),
                DecodedValue::I64(0),
                DecodedValue::I64(2),
                DecodedValue::I64(0),
            ],
        )
        .await
        .unwrap();

        let tx1 = pool.begin("read_committed").await.unwrap();
        let tx2 = pool.begin("read_committed").await.unwrap();

        tx1.execute_typed("UPDATE pgcon_deadlock_test SET value = 1::int8 WHERE id = 1::int8", &[])
            .await
            .unwrap();
        tx2.execute_typed("UPDATE pgcon_deadlock_test SET value = 2::int8 WHERE id = 2::int8", &[])
            .await
            .unwrap();

        // Now each blocks on the row the other is holding — issue both
        // concurrently and let Postgres's deadlock detector break the tie.
        let (r1, r2) = tokio::join!(
            tx1.execute_typed("UPDATE pgcon_deadlock_test SET value = 3::int8 WHERE id = 2::int8", &[]),
            tx2.execute_typed("UPDATE pgcon_deadlock_test SET value = 4::int8 WHERE id = 1::int8", &[]),
        );

        let results = [r1, r2];
        let deadlock_errors: Vec<_> = results
            .iter()
            .filter(|r| matches!(r, Err(e) if e.sqlstate() == Some(&tokio_postgres::error::SqlState::T_R_DEADLOCK_DETECTED)))
            .collect();
        assert_eq!(
            deadlock_errors.len(),
            1,
            "expected exactly one side to be aborted with 40P01, got {results:?}"
        );
    }
}

#[cfg(test)]
mod error_message_tests {
    use super::*;

    /// A server-side failure must render what the server said, not
    /// `tokio_postgres::Error`'s generic `"db error"` — the regression that
    /// made `IndexWorker(Vector): claim_batch failed: db error` impossible
    /// to act on.
    #[tokio::test]
    #[ignore = "requires a live database"]
    async fn a_server_error_renders_the_server_message() {
        let pool = PgPool::connect(&std::env::var("PYLON_PGCON_TEST_DSN").unwrap(), 2)
            .await
            .unwrap();
        // Deliberately needs no fixture: a missing relation is a
        // server-side error on any database, so this cannot rot when some
        // other test's table comes or goes.
        let err = pool
            .query_typed("SELECT 1 FROM pylon_no_such_table", &[], &pool.types())
            .await
            .unwrap_err();
        let rendered = err.to_string();
        assert!(
            rendered.contains("pylon_no_such_table"),
            "expected the server's own message, got {rendered:?}"
        );
        assert_ne!(rendered, "db error");
    }
}