chore: track client idle-in-transaction disconnects as a pool stat

pgdog-stats/src/pool.rs

@@ -63,6 +63,9 @@ pub struct Counts {
     pub writes: usize,
     /// Password attempts.
     pub auth_attempts: usize,
+    /// Number of clients disconnected because they idled in a transaction
+    /// past `client_idle_in_transaction_timeout` while holding a server connection.
+    pub client_idle_xact_timeouts: usize,
 }
 
 impl Sub for Counts {
@@ -95,6 +98,9 @@ impl Sub for Counts {
             reads: self.reads.saturating_sub(rhs.reads),
             writes: self.writes.saturating_sub(rhs.writes),
             auth_attempts: self.auth_attempts.saturating_sub(rhs.auth_attempts),
+            client_idle_xact_timeouts: self
+                .client_idle_xact_timeouts
+                .saturating_sub(rhs.client_idle_xact_timeouts),
         }
     }
 }
@@ -129,6 +135,9 @@ impl Add for Counts {
             reads: self.reads.saturating_add(rhs.reads),
             writes: self.writes.saturating_add(rhs.writes),
             auth_attempts: self.auth_attempts.saturating_add(rhs.auth_attempts),
+            client_idle_xact_timeouts: self
+                .client_idle_xact_timeouts
+                .saturating_add(rhs.client_idle_xact_timeouts),
         }
     }
 }
@@ -167,6 +176,7 @@ impl Div<usize> for Counts {
             reads: self.reads.checked_div(rhs).unwrap_or(0),
             writes: self.writes.checked_div(rhs).unwrap_or(0),
             auth_attempts: self.auth_attempts.checked_div(rhs).unwrap_or(0),
+            client_idle_xact_timeouts: self.client_idle_xact_timeouts.checked_div(rhs).unwrap_or(0),
         }
     }
 }

pgdog-stats/src/server.rs

@@ -58,6 +58,7 @@ impl Add<Counts> for PoolCounts {
             writes: self.writes,
             reads: self.reads,
             auth_attempts: self.auth_attempts,
+            client_idle_xact_timeouts: self.client_idle_xact_timeouts,
         }
     }
 }

pgdog/src/admin/show_stats.rs

@@ -52,6 +52,7 @@ impl Command for ShowStats {
                         Field::numeric(&format!("{}_reads", prefix)),
                         Field::numeric(&format!("{}_writes", prefix)),
                         Field::numeric(&format!("{}_auth_attempts", prefix)),
+                        Field::numeric(&format!("{}_client_idle_xact_timeouts", prefix)),
                     ]
                 })
                 .collect::<Vec<Field>>(),
@@ -101,7 +102,8 @@ impl Command for ShowStats {
                             .add(stat.connect_count)
                             .add(stat.reads)
                             .add(stat.writes)
-                            .add(stat.auth_attempts);
+                            .add(stat.auth_attempts)
+                            .add(stat.client_idle_xact_timeouts);
                     }
 
                     messages.push(dr.message()?);

pgdog/src/backend/pool/connection/binding.rs

@@ -54,6 +54,20 @@ impl Binding {
         self.disconnect();
     }
 
+    /// Record an idle-in-transaction timeout on every pool
+    /// we're currently holding a connection from.
+    pub fn record_client_idle_xact_timeout(&self) {
+        match self {
+            Binding::Direct(guard, _) => guard.pool.record_client_idle_xact_timeout(),
+            Binding::MultiShard(guards, _) => {
+                for guard in guards {
+                    guard.pool.record_client_idle_xact_timeout();
+                }
+            }
+            _ => (),
+        }
+    }
+
     /// Are we connected to a backend?
     pub fn connected(&self) -> bool {
         match self {

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

@@ -428,6 +428,12 @@ impl Connection {
         self.binding.done() && !self.locked
     }
 
+    /// Record that the client holding this connection was disconnected
+    /// because it idled inside a transaction for too long.
+    pub(crate) fn record_client_idle_xact_timeout(&self) {
+        self.binding.record_client_idle_xact_timeout();
+    }
+
     /// Lock this connection to the client, preventing it's
     /// release back into the pool.
     pub(crate) fn lock(&mut self, lock: bool) {

pgdog/src/backend/pool/pool_impl.rs

@@ -400,6 +400,12 @@ impl Pool {
         &self.inner.comms
     }
 
+    /// Record a client disconnected for idling in a transaction
+    /// while holding a connection from this pool.
+    pub fn record_client_idle_xact_timeout(&self) {
+        self.lock().stats.counts.client_idle_xact_timeouts += 1;
+    }
+
     /// Pool address.
     #[inline]
     pub fn addr(&self) -> &Address {

pgdog/src/backend/pool/stats.rs

@@ -180,6 +180,7 @@ mod tests {
             reads: 25,
             writes: 50,
             auth_attempts: 30,
+            client_idle_xact_timeouts: 0,
         }
         .into();
 
@@ -207,6 +208,7 @@ mod tests {
             reads: 10,
             writes: 20,
             auth_attempts: 20,
+            client_idle_xact_timeouts: 0,
         }
         .into();
 
@@ -263,6 +265,7 @@ mod tests {
             reads: 25,
             writes: 50,
             auth_attempts: 50,
+            client_idle_xact_timeouts: 0,
         }
         .into();
 
@@ -290,6 +293,7 @@ mod tests {
             reads: 10,
             writes: 20,
             auth_attempts: 30,
+            client_idle_xact_timeouts: 0,
         }
         .into();
 
@@ -368,6 +372,7 @@ mod tests {
             reads: 10,
             writes: 20,
             auth_attempts: 10,
+            client_idle_xact_timeouts: 0,
         }
         .into();
 
@@ -439,6 +444,7 @@ mod tests {
             reads: 10,
             writes: 25,
             auth_attempts: 100,
+            client_idle_xact_timeouts: 0,
         }
         .into();
 

pgdog/src/frontend/client/mod.rs

@@ -506,6 +506,12 @@ impl Client {
                     match event {
                         // Client disconnected, we're done.
                         BufferEvent::DisconnectAbrupt | BufferEvent::DisconnectGraceful => break,
+                        // Client sat idle in a transaction for too long
+                        // while hogging a server connection. Count it, then disconnect.
+                        BufferEvent::IdleXactTimeout => {
+                            query_engine.record_client_idle_xact_timeout();
+                            break;
+                        }
                         BufferEvent::HaveRequest => (),
                     }
                 }
@@ -613,7 +619,11 @@ impl Client {
                         self.stream
                             .fatal(ErrorResponse::client_idle_timeout(idle_timeout, &state))
                             .await?;
-                        return Ok(BufferEvent::DisconnectAbrupt);
+                        return Ok(if state == State::IdleInTransaction {
+                            BufferEvent::IdleXactTimeout
+                        } else {
+                            BufferEvent::DisconnectAbrupt
+                        });
                     }
 
                     Ok(Ok(message)) => message.stream(self.streaming).frontend(),
@@ -712,5 +722,7 @@ pub mod test;
 enum BufferEvent {
     DisconnectGraceful,
     DisconnectAbrupt,
+    /// Client disconnected by the idle-in-transaction timeout.
+    IdleXactTimeout,
     HaveRequest,
 }

pgdog/src/frontend/client/query_engine/mod.rs

@@ -106,6 +106,10 @@ impl QueryEngine {
         self.begin_stmt.is_none() && self.backend.done()
     }
 
+    pub fn record_client_idle_xact_timeout(&self) {
+        self.backend.record_client_idle_xact_timeout();
+    }
+
     /// Current state.
     pub fn client_state(&self) -> State {
         self.stats.state

pgdog/src/frontend/client/test/mod.rs

@@ -305,6 +305,45 @@ async fn test_client_idle_timeout() {
     );
 }
 
+#[tokio::test]
+async fn test_client_idle_xact_timeout_stat() {
+    let (mut conn, mut client, _) = new_client!(false);
+
+    let mut config = (*config()).clone();
+    config.config.general.client_idle_in_transaction_timeout = 100;
+    set(config).unwrap();
+
+    let handle = tokio::spawn(async move {
+        client.run().await.unwrap();
+    });
+
+    // Open a transaction and run a query so a server connection is checked out.
+    conn.write_all(&Query::new("BEGIN").to_bytes())
+        .await
+        .unwrap();
+    let _ = read!(conn, ['C', 'Z']);
+    conn.write_all(&Query::new("SELECT 1").to_bytes())
+        .await
+        .unwrap();
+    let _ = read!(conn, ['T', 'D', 'C', 'Z']);
+
+    // Go idle inside the transaction; the timeout disconnects us.
+    let start = Instant::now();
+    let err = read_one!(conn);
+    assert!(start.elapsed() >= Duration::from_millis(100));
+    let err = ErrorResponse::from_bytes(err.freeze()).unwrap();
+    assert_eq!(err.code, "57P05");
+    assert_eq!(err.message, "disconnecting idle in transaction client");
+
+    handle.await.unwrap();
+
+    // The pool recorded the event.
+    let dbs = databases();
+    let cluster = dbs.cluster(("pgdog", "pgdog")).unwrap();
+    let state = cluster.shards()[0].pools()[0].state();
+    assert_eq!(state.stats.counts.client_idle_xact_timeouts, 1);
+}
+
 #[tokio::test]
 async fn test_parse_describe_flush_bind_execute_close_sync() {
     let (mut conn, mut client, _) = new_client!(false);

pgdog/src/stats/pools.rs

@@ -86,6 +86,8 @@ impl Pools {
         let mut total_sv_xact_idle = vec![];
         let mut total_auth_attempts = vec![];
         let mut avg_auth_attempts = vec![];
+        let mut total_client_idle_xact_timeouts = vec![];
+        let mut avg_client_idle_xact_timeouts = vec![];
 
         let general = &crate::config::config().config.general;
 
@@ -315,6 +317,16 @@ impl Pools {
                         labels: labels.clone(),
                         measurement: averages.auth_attempts.into(),
                     });
+
+                    total_client_idle_xact_timeouts.push(Measurement {
+                        labels: labels.clone(),
+                        measurement: totals.client_idle_xact_timeouts.into(),
+                    });
+
+                    avg_client_idle_xact_timeouts.push(Measurement {
+                        labels: labels.clone(),
+                        measurement: averages.client_idle_xact_timeouts.into(),
+                    });
                 }
             }
         }
@@ -684,6 +696,22 @@ impl Pools {
             metric_type: None,
         }));
 
+        metrics.push(Metric::new(PoolMetric {
+            name: "total_client_idle_xact_timeouts".into(),
+            measurements: total_client_idle_xact_timeouts,
+            help: "Total number of clients disconnected for idling in a transaction.".into(),
+            unit: None,
+            metric_type: Some("counter".into()),
+        }));
+
+        metrics.push(Metric::new(PoolMetric {
+            name: "avg_client_idle_xact_timeouts".into(),
+            measurements: avg_client_idle_xact_timeouts,
+            help: "Average number of clients disconnected for idling in a transaction per statistics period.".into(),
+            unit: None,
+            metric_type: None,
+        }));
+
         Pools { metrics }
     }