fix: mirror prepared statement mode in session pooling (#827)

This might be a two bug layer. First, it's might be an issue with
DEALLOCATE sent as a simple query that doesn't get rewritten, which
causes a transaction failure and messes up the whole mirroring.

But I'm using pgdog in session mode with mirroring to test performance
to see if I can switch the database to a cheaper one so I'm not affected
by the DEALLOCATE bug on the source. What I noticed is that the mirror
traffic gets this error:

```
ERROR:  prepared statement "pdo_stmt_00000001" does not exist
STATEMENT:  DEALLOCATE pdo_stmt_00000001
```

I also included a fix to speed up the dockerfile build with caches since
I iterated on the dockerfile with the fix.
I also have a repro with a docker-compose and a simple php app here:
https://github.com/costi/pgdog-pdo-repro

The flow is like this:

1. In session mode, the client path uses the **effective**
prepared-statements setting from:

```rust
config.prepared_statements()
```

2. `ConfigAndUsers::prepared_statements()` disables prepared statements
in session mode:

```rust
pub fn prepared_statements(&self) -> PreparedStatements {
    if self.config.general.pooler_mode == PoolerMode::Session {
        PreparedStatements::Disabled
    } else {
        self.config.general.prepared_statements
    }
}
```

3. The mirror path was not using that effective value. It constructed
its prepared-statement state with the raw default/configured level
instead.

4. So with:

```toml
pooler_mode = "session"
prepared_statements = "extended"
```

the source/client path effectively ran with `Disabled`, while the mirror
path still ran with `Extended`.

5. That caused the mirror path to rewrite extended-protocol prepared
statement names to `__pgdog_*`, while the source path kept the original
PDO names like `pdo_stmt_00000001`.

6. Later, PDO sent simple SQL cleanup:

```sql
DEALLOCATE pdo_stmt_00000001
```

7. The mirror backend only knew the prepared statement under the
rewritten `__pgdog_*` name, so PostgreSQL correctly errored:

```text
ERROR:  prepared statement "pdo_stmt_00000001" does not exist
STATEMENT:  DEALLOCATE pdo_stmt_00000001
```

Author: costi

.github/workflows/ci.yml

@@ -117,7 +117,7 @@ jobs:
           sudo -u postgres psql -c 'ALTER SYSTEM SET wal_level TO logical;'
           sudo service postgresql restart
           bash integration/setup.sh
-          sudo apt update && sudo apt install -y python3-virtualenv mold
+          sudo apt update && sudo apt install -y python3-virtualenv mold php-cli php-pgsql
           sudo gem install bundler
           sudo apt remove -y cmake
           sudo pip3 install cmake==3.31.6
@@ -157,6 +157,8 @@ jobs:
         run: bash integration/ruby/run.sh
       - name: Java
         run: bash integration/java/run.sh
+      - name: Mirror
+        run: bash integration/mirror/run.sh
       - name: SQL
         run: bash integration/sql/run.sh
       - name: Toxi

integration/mirror/dev.sh

@@ -1,5 +1,15 @@
 #!/bin/bash
+set -euo pipefail
+
 SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+
 pushd ${SCRIPT_DIR}/ruby
-bundle exec *_spec.rb
+export GEM_HOME=~/.gem
+mkdir -p ${GEM_HOME}
+bundle install
+bundle exec rspec *_spec.rb
+popd
+
+pushd ${SCRIPT_DIR}/php
+bash run.sh
 popd

integration/mirror/php/pdo_mirror.php

@@ -0,0 +1,81 @@
+<?php
+
+declare(strict_types=1);
+
+function connectDb(string $dbName): PDO
+{
+    $pdo = new PDO(
+        "pgsql:host=127.0.0.1;port=6432;dbname={$dbName}",
+        "pgdog",
+        "pgdog",
+        [
+            PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
+            PDO::ATTR_EMULATE_PREPARES => false,
+        ]
+    );
+
+    return $pdo;
+}
+
+function sleepForMirror(): void
+{
+    // Mirror delivery is asynchronous; a short delay keeps the test stable.
+    usleep(500_000);
+}
+
+$source = connectDb("pgdog");
+$mirror = connectDb("pgdog_mirror");
+
+$source->exec("DROP TABLE IF EXISTS public.mirror_php_test");
+$source->exec("CREATE TABLE public.mirror_php_test (id BIGINT PRIMARY KEY, value TEXT NOT NULL)");
+sleepForMirror();
+
+$source->beginTransaction();
+
+$insert = $source->prepare("INSERT INTO public.mirror_php_test (id, value) VALUES (?, ?)");
+$insert->execute([1, "one"]);
+$insert->execute([2, "two"]);
+
+$source->commit();
+
+sleepForMirror();
+
+$sourceCount = (int) $source->query("SELECT COUNT(*) FROM public.mirror_php_test")->fetchColumn();
+$mirrorCount = (int) $mirror->query("SELECT COUNT(*) FROM public.mirror_php_test")->fetchColumn();
+
+if ($sourceCount !== 2) {
+    fwrite(STDERR, "expected 2 rows on source, got {$sourceCount}\n");
+    exit(1);
+}
+
+if ($mirrorCount !== 2) {
+    fwrite(STDERR, "expected 2 rows on mirror, got {$mirrorCount}\n");
+    exit(1);
+}
+
+$mirrorRows = $mirror
+    ->query("SELECT id, value FROM public.mirror_php_test ORDER BY id")
+    ->fetchAll(PDO::FETCH_ASSOC);
+
+$mirrorRows = array_map(
+    static fn(array $row): array => [
+        "id" => (int) $row["id"],
+        "value" => (string) $row["value"],
+    ],
+    $mirrorRows
+);
+
+$expected = [
+    ["id" => 1, "value" => "one"],
+    ["id" => 2, "value" => "two"],
+];
+
+if ($mirrorRows !== $expected) {
+    fwrite(STDERR, "unexpected mirror rows: " . json_encode($mirrorRows) . "\n");
+    exit(1);
+}
+
+$source->exec("DROP TABLE IF EXISTS public.mirror_php_test");
+sleepForMirror();
+
+echo "php mirror test passed\n";

integration/mirror/php/run.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+set -euo pipefail
+
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+
+php "${SCRIPT_DIR}/pdo_mirror.php"

integration/mirror/ruby/Gemfile.lock

@@ -51,6 +51,7 @@ GEM
 PLATFORMS
   arm64-darwin-24
   ruby
+  x86_64-linux
 
 DEPENDENCIES
   csv

pgdog/src/backend/pool/connection/mirror/handler.rs

@@ -5,6 +5,7 @@
 
 use super::*;
 use crate::backend::pool::MirrorStats;
+use crate::frontend::ClientRequest;
 use parking_lot::Mutex;
 use std::sync::Arc;
 

pgdog/src/backend/pool/connection/mirror/mod.rs

@@ -16,8 +16,6 @@ use crate::frontend::client::TransactionType;
 use crate::frontend::{ClientComms, PreparedStatements};
 use crate::net::{BackendKeyData, Parameter, Parameters, Stream};
 
-use crate::frontend::ClientRequest;
-
 use super::Error;
 
 pub mod buffer_with_delay;
@@ -51,9 +49,12 @@ pub struct Mirror {
 
 impl Mirror {
     fn new(params: &Parameters, config: &ConfigAndUsers) -> Self {
+        let mut prepared_statements = PreparedStatements::new();
+        prepared_statements.set_level(config.prepared_statements());
+
         Self {
             id: BackendKeyData::new(),
-            prepared_statements: PreparedStatements::new(),
+            prepared_statements,
             params: params.clone(),
             timeouts: Timeouts::from_config(&config.config.general),
             stream: Stream::dev_null(),
@@ -173,7 +174,11 @@ impl Mirror {
 
 #[cfg(test)]
 mod test {
-    use crate::{backend::pool::Request, config, net::Query};
+    use crate::{
+        backend::pool::Request,
+        config::{self, PoolerMode, PreparedStatements as PreparedStatementsLevel},
+        net::{Parameter, Parameters, Query},
+    };
 
     use super::*;
 
@@ -319,4 +324,40 @@ mod test {
 
         cluster.shutdown();
     }
+
+    #[tokio::test]
+    async fn test_mirror_uses_effective_prepared_statements_level() {
+        config::load_test();
+        let mut test_config = (*config::config()).clone();
+        test_config.config.general.pooler_mode = PoolerMode::Session;
+        test_config.config.general.prepared_statements = PreparedStatementsLevel::Extended;
+        let test_config = config::set(test_config).unwrap();
+        let config = std::sync::Arc::new(test_config);
+
+        assert_eq!(
+            config.config.general.prepared_statements,
+            PreparedStatementsLevel::Extended
+        );
+        assert_eq!(
+            config.prepared_statements(),
+            PreparedStatementsLevel::Disabled
+        );
+
+        let params = Parameters::from(vec![
+            Parameter {
+                name: "user".into(),
+                value: "pgdog".into(),
+            },
+            Parameter {
+                name: "database".into(),
+                value: "pgdog".into(),
+            },
+        ]);
+
+        let mirror = Mirror::new(&params, &config);
+        assert_eq!(
+            mirror.prepared_statements.level(),
+            PreparedStatementsLevel::Disabled
+        );
+    }
 }

pgdog/src/frontend/prepared_statements/mod.rs

@@ -122,6 +122,12 @@ impl PreparedStatements {
         self.local.len()
     }
 
+    /// Current prepared statements compatibility level.
+    #[cfg(test)]
+    pub fn level(&self) -> PreparedStatementsLevel {
+        self.level
+    }
+
     /// Is the local cache empty?
     pub fn is_empty(&self) -> bool {
         self.len_local() == 0
@@ -157,7 +163,6 @@ impl PreparedStatements {
     }
 
     /// Set the prepared statements level.
-    #[cfg(test)]
     pub fn set_level(&mut self, level: PreparedStatementsLevel) {
         self.level = level;
     }