Fix semaphore permit drain, burst gate race, migration issues

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pg_doorman"
-version = "3.5.1"
+version = "3.5.2"
 edition = "2021"
 rust-version = "1.87.0"
 license = "MIT"

documentation/en/src/changelog.md

@@ -1,5 +1,23 @@
 # Changelog
 
+### 3.5.2 <small>Apr 20, 2026</small>
+
+#### Migration fixes
+
+- **Client ID collision after migration.** The new process started its connection counter at 0, colliding with migrated client IDs. Stats were lost for colliding clients. Now the counter advances past the highest migrated ID.
+
+- **SCRAM passthrough state preserved.** The ClientKey extracted from the first client's SCRAM handshake is now serialized in the migration payload (v2 format, backward compatible). The new process skips the `ScramPending` fallback to `server_password`.
+
+#### Session mode statistics fix
+
+Transaction time percentiles (`xact_time p50/p90/p95/p99`) in session mode previously showed the entire session duration (clamped to 10 minutes) instead of individual transaction time. Now `xact_time` is recorded per-transaction at each `ReadyForQuery(Idle)`, matching transaction mode semantics. Also fixes `query_time` accumulation in session mode — timer resets per query instead of growing monotonically.
+
+#### Adaptive anticipation budget
+
+The anticipation wait (formerly a fixed 300-500ms) now scales with real transaction latency. At cold start (no histogram data): 100ms ± 20% jitter. At steady state: `xact_p99 × 2 ± 20%`, clamped between 5ms and 500ms.
+
+Pools with coordinator previously stabilized at a fraction of `pool_size` (e.g. 12 out of 40) because anticipation waited 300-500ms for returns that came back in <1ms. With adaptive budget at xact_p99=0.7ms: wait ≈ 5ms. Pool fills to adequate size in seconds instead of minutes.
+
 ### 3.5.1 <small>Apr 20, 2026</small>
 
 #### systemd Type=notify support

documentation/en/src/tutorials/pool-pressure.md

@@ -163,14 +163,21 @@ dropped. `return_object` detects the dropped receiver (`send` returns
 This way timed-out waiters are cleaned up lazily without a separate
 garbage-collection pass.
 
-The deadline is `min(query_wait_timeout - 500 ms, PHASE_4_HARD_CAP)`
-where `PHASE_4_HARD_CAP` is randomly chosen between **300 ms and
-500 ms** (uniform jitter) for each checkout attempt, measured against
-a timestamp captured at the top of `timeout_get`. Phase 1/2 semaphore
-wait consumes from the same budget, so the cumulative wait across
-phases cannot exceed the caller's `query_wait_timeout`.
-
-The jitter prevents a **timeout cliff**: without it, N clients that
+The deadline is adaptive: `min(query_wait_timeout - 500 ms, adaptive_cap)`
+where `adaptive_cap` is derived from real transaction latency:
+
+| Pool state | Budget | Example |
+|------------|--------|---------|
+| Cold start (no stats) | 100ms ± 20% jitter | 80-120ms |
+| Steady state | xact_p99 × 2 ± 20% jitter | p99=0.7ms → 5ms (min); p99=50ms → 100ms |
+| High latency | Capped at 500ms | p99=300ms → 500ms |
+
+The budget is measured against a timestamp captured at the top of
+`timeout_get`. Phase 1/2 semaphore wait consumes from the same budget,
+so the cumulative wait across phases cannot exceed the caller's
+`query_wait_timeout`.
+
+The ±20% jitter prevents a **timeout cliff**: without it, N clients that
 entered Phase 4 at the same instant all exit simultaneously and
 stampede into the burst gate, creating N new backend connections for a
 pool that needs far fewer. With jitter, clients exit in staggered

documentation/ru/pool-pressure.md

@@ -185,16 +185,22 @@ recycle 10 раз в плотном цикле `yield_now` (контролиру
 Таким образом протухшие waiter'ы вычищаются лениво, без отдельного
 прохода сборки мусора.
 
-Дедлайн равен `min(query_wait_timeout - 500 ms, PHASE_4_HARD_CAP)`,
-где `PHASE_4_HARD_CAP` выбирается случайно в диапазоне **300–500 ms**
-(равномерный jitter) для каждой попытки checkout'а. Фаза 1/2 ожидания
-семафора тратит из того же бюджета, поэтому суммарное ожидание не
-может увести клиента за его `query_wait_timeout`.
-
-Jitter предотвращает **timeout cliff**: без него N клиентов, вошедших
-в Phase 4 одновременно, выходят одновременно и лавиной входят в burst
-gate, создавая N новых backend-соединений для пула, которому нужно
-значительно меньше. С jitter'ом клиенты выходят порциями — первые
+Дедлайн адаптивный: `min(query_wait_timeout - 500 ms, adaptive_cap)`,
+где `adaptive_cap` вычисляется из реальной latency транзакций:
+
+| Состояние пула | Бюджет | Пример |
+|---------------|--------|--------|
+| Холодный старт (нет данных) | 100ms ± 20% jitter | 80-120ms |
+| Steady state | xact_p99 × 2 ± 20% jitter | p99=0.7ms → 5ms (min); p99=50ms → 100ms |
+| Высокая latency | Ограничено 500ms | p99=300ms → 500ms |
+
+Фаза 1/2 ожидания семафора тратит из того же бюджета, поэтому
+суммарное ожидание не может увести клиента за его `query_wait_timeout`.
+
+Jitter ±20% предотвращает **timeout cliff**: без него N клиентов,
+вошедших в Phase 4 одновременно, выходят одновременно и лавиной входят
+в burst gate, создавая N новых backend-соединений для пула, которому
+нужно значительно меньше. С jitter'ом клиенты выходят порциями — первые
 создают соединения, и к моменту выхода последних эти соединения уже
 использованы и возвращены в idle queue для переиспользования.
 

src/client/core.rs

@@ -358,6 +358,11 @@ pub struct Client<S, T> {
     /// Connected to server
     pub(crate) connected_to_server: bool,
 
+    /// Session mode: transaction start timestamp for per-transaction xact_time.
+    /// Set when server transitions into a transaction (ReadyForQuery 'T'/'E').
+    /// Consumed when transaction ends (ReadyForQuery 'I').
+    pub(crate) session_xact_start: Option<quanta::Instant>,
+
     /// Name of the server pool for this client (This comes from the database name in the connection string)
     pub(crate) pool_name: String,
 

src/client/migration.rs

@@ -23,7 +23,7 @@
 use super::core::PreparedStatementState;
 
 const MIGRATION_MAGIC: u32 = 0x50474D47; // "PGMG"
-const MIGRATION_VERSION: u16 = 1;
+const MIGRATION_VERSION: u16 = 2;
 /// Fixed-size header: magic(4) + version(2) + connection_id(8) + secret_key(4) + transaction_mode(1)
 const HEADER_SIZE: usize = 4 + 2 + 8 + 4 + 1;
 const MAX_PREPARED_ENTRIES: usize = 100_000;
@@ -193,6 +193,31 @@
 
         buf.put_u8(use_tls as u8);
 
+        // Backend auth state (v2): allows new process to skip ScramPending
+        // fallback by receiving the ClientKey from the old process.
+        if let Some(pool) = get_pool(&self.pool_name, &self.username) {
+            if let Some(ref ba_lock) = pool.address.backend_auth {
+                match &*ba_lock.read() {
+                    crate::config::BackendAuthMethod::Md5PassTheHash(hash) => {
+                        buf.put_u8(1);
+                        put_str(&mut buf, hash);
+                    }
+                    crate::config::BackendAuthMethod::ScramPassthrough(client_key) => {
+                        buf.put_u8(2);
+                        buf.put_u16(client_key.len() as u16);
+                        buf.put_slice(client_key);
+                    }
+                    crate::config::BackendAuthMethod::ScramPending => {
+                        buf.put_u8(3);
+                    }
+                }
+            } else {
+                buf.put_u8(0); // no backend auth
+            }
+        } else {
+            buf.put_u8(0);
+        }
+
         buf
     }
 }
@@ -243,6 +268,7 @@
     prepared_entries: Vec<PreparedEntry>,
     #[allow(dead_code)]
     use_tls: bool,
+    backend_auth: Option<crate::config::BackendAuthMethod>,
 }
 
 struct PreparedEntry {
@@ -262,7 +288,7 @@
         )));
     }
     let version = buf.get_u16();
-    if version != MIGRATION_VERSION {
+    if version != 1 && version != MIGRATION_VERSION {
         return Err(Error::ClientError(format!(
             "migration: unsupported version {version}"
         )));
@@ -349,6 +375,31 @@
     require(&buf, 1)?;
     let use_tls = buf.get_u8() != 0;
 
+    // v2: backend auth state (ScramPassthrough ClientKey, Md5 hash)
+    let backend_auth = if version >= 2 && buf.remaining() > 0 {
+        let tag = buf.get_u8();
+        match tag {
+            1 => {
+                // Md5PassTheHash
+                let hash = get_str(&mut buf)?;
+                Some(crate::config::BackendAuthMethod::Md5PassTheHash(hash))
+            }
+            2 => {
+                // ScramPassthrough(ClientKey)
+                require(&buf, 2)?;
+                let key_len = buf.get_u16() as usize;
+                require(&buf, key_len)?;
+                let mut key = vec![0u8; key_len];
+                buf.copy_to_slice(&mut key);
+                Some(crate::config::BackendAuthMethod::ScramPassthrough(key))
+            }
+            3 => Some(crate::config::BackendAuthMethod::ScramPending),
+            _ => None,
+        }
+    } else {
+        None // v1 format: no auth state
+    };
+
     Ok(DeserializedState {
         connection_id,
         secret_key,
@@ -361,6 +412,7 @@
         async_client,
         prepared_entries,
         use_tls,
+        backend_auth,
     })
 }
 
@@ -389,6 +441,23 @@
 
     // Reconstruct prepared statement cache
     let pool = get_pool(&state.pool_name, &state.username);
+
+    // Restore backend auth state (e.g. SCRAM ClientKey) so the new
+    // process can authenticate to PostgreSQL via passthrough without
+    // waiting for a fresh client SCRAM handshake.
+    if let (Some(ref p), Some(ref migrated_auth)) = (&pool, &state.backend_auth) {
+        if let Some(ref ba_lock) = p.address.backend_auth {
+            let needs_update = matches!(*ba_lock.read(), crate::config::BackendAuthMethod::ScramPending);
+            if needs_update {
+                *ba_lock.write() = migrated_auth.clone();
+                info!(
+                    "[{}@{}] restored backend auth from migrated client",
+                    state.username, state.pool_name
+                );
+            }
+        }
+    }
+
     let prepared = reconstruct_prepared_state(
         state.prepared_enabled,
         state.async_client,
@@ -430,6 +499,7 @@
         admin: false,
         last_server_stats: None,
         connected_to_server: false,
+        session_xact_start: None,
         pool_name: state.pool_name,
         username: state.username,
         server_parameters: state.server_parameters,
@@ -485,6 +555,20 @@
 
     let config = get_config();
     let pool = get_pool(&state.pool_name, &state.username);
+
+    if let (Some(ref p), Some(ref migrated_auth)) = (&pool, &state.backend_auth) {
+        if let Some(ref ba_lock) = p.address.backend_auth {
+            let needs_update = matches!(*ba_lock.read(), crate::config::BackendAuthMethod::ScramPending);
+            if needs_update {
+                *ba_lock.write() = migrated_auth.clone();
+                info!(
+                    "[{}@{}] restored backend auth from migrated TLS client",
+                    state.username, state.pool_name
+                );
+            }
+        }
+    }
+
     let prepared = reconstruct_prepared_state(
         state.prepared_enabled,
         state.async_client,
@@ -526,6 +610,7 @@
         admin: false,
         last_server_stats: None,
         connected_to_server: false,
+        session_xact_start: None,
         pool_name: state.pool_name,
         username: state.username,
         server_parameters: state.server_parameters,
@@ -861,6 +946,7 @@
     #[cfg(all(target_os = "linux", feature = "tls-migration"))]
     let tls_acceptor = _tls_acceptor;
     use crate::app::server::CURRENT_CLIENT_COUNT;
+    use crate::stats::TOTAL_CONNECTION_COUNTER;
     use std::sync::atomic::Ordering;
 
     info!("migration receiver: listening for migrated clients");
@@ -882,6 +968,10 @@
                             {
                                 Ok(mut client) => {
                                     CURRENT_CLIENT_COUNT.fetch_add(1, Ordering::SeqCst);
+                                    TOTAL_CONNECTION_COUNTER.fetch_max(
+                                        client.connection_id as usize,
+                                        Ordering::Relaxed,
+                                    );
                                     info!(
                                         "[{}@{} #c{}] migrated TLS client from {}",
                                         client.username,
@@ -915,6 +1005,12 @@
                     match reconstruct_client(fd, state_buf, csm).await {
                         Ok(mut client) => {
                             CURRENT_CLIENT_COUNT.fetch_add(1, Ordering::SeqCst);
+                            // Advance the global counter past the migrated id so new
+                            // connections don't collide with migrated client ids.
+                            TOTAL_CONNECTION_COUNTER.fetch_max(
+                                client.connection_id as usize,
+                                Ordering::Relaxed,
+                            );
                             info!(
                                 "[{}@{} #c{}] migrated client accepted from {}",
                                 client.username,

src/client/startup.rs

@@ -411,6 +411,7 @@ where
             admin,
             last_server_stats: None,
             connected_to_server: false,
+            session_xact_start: None,
             pool_name,
             username: std::mem::take(&mut client_identifier.username),
             server_parameters,
@@ -460,6 +461,7 @@ where
             server_parameters: ServerParameters::new(),
             prepared: PreparedStatementState::default(),
             connected_to_server: false,
+            session_xact_start: None,
             client_last_messages_in_tx: PooledBuffer::new(),
             max_memory_usage: 128 * 1024 * 1024,
             pooler_check_query_request_vec: Vec::new(),