Fix maxwait metric (#183 )

Max wait was being reported as 0 after #159 This PR fixes that and adds test
Replace a few types with more developer-friendly names (#182 )
2026-03-23 09:26:30 +00:00 · 2022-10-05 21:41:09 -05:00 · 2022-10-01 10:25:59 -07:00 · 2022-09-28 08:50:14 -05:00 · 2022-09-24 09:09:15 -07:00 · 2022-09-23 12:24:30 -07:00
33 changed files with 2858 additions and 985 deletions
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -15,14 +15,34 @@ jobs:
          RUSTFLAGS: "-C instrument-coverage"
          LLVM_PROFILE_FILE: "pgcat-%m.profraw"
      - image: postgres:14
-        # auth:
-        #   username: mydockerhub-user
-        #   password: $DOCKERHUB_PASSWORD
+        command: ["postgres", "-p", "5432", "-c", "shared_preload_libraries=pg_stat_statements"]
        environment:
          POSTGRES_USER: postgres
          POSTGRES_DB: postgres
          POSTGRES_PASSWORD: postgres
          POSTGRES_HOST_AUTH_METHOD: scram-sha-256
+      - image: postgres:14
+        command: ["postgres", "-p", "7432", "-c", "shared_preload_libraries=pg_stat_statements"]
+        environment:
+          POSTGRES_USER: postgres
+          POSTGRES_DB: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_HOST_AUTH_METHOD: scram-sha-256
+      - image: postgres:14
+        command: ["postgres", "-p", "8432", "-c", "shared_preload_libraries=pg_stat_statements"]
+        environment:
+          POSTGRES_USER: postgres
+          POSTGRES_DB: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_HOST_AUTH_METHOD: scram-sha-256
+      - image: postgres:14
+        command: ["postgres", "-p", "9432", "-c", "shared_preload_libraries=pg_stat_statements"]
+        environment:
+          POSTGRES_USER: postgres
+          POSTGRES_DB: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_HOST_AUTH_METHOD: scram-sha-256
+
    # Add steps to the job
    # See: https://circleci.com/docs/2.0/configuration-reference/#steps
    steps:
--- a/.circleci/run_tests.sh
+++ b/.circleci/run_tests.sh
@@ -16,6 +16,9 @@ function start_pgcat() {

 # Setup the database with shards and user
 PGPASSWORD=postgres psql -e -h 127.0.0.1 -p 5432 -U postgres -f tests/sharding/query_routing_setup.sql
+PGPASSWORD=postgres psql -e -h 127.0.0.1 -p 7432 -U postgres -f tests/sharding/query_routing_setup.sql
+PGPASSWORD=postgres psql -e -h 127.0.0.1 -p 8432 -U postgres -f tests/sharding/query_routing_setup.sql
+PGPASSWORD=postgres psql -e -h 127.0.0.1 -p 9432 -U postgres -f tests/sharding/query_routing_setup.sql

 PGPASSWORD=sharding_user pgbench -h 127.0.0.1 -U sharding_user shard0 -i
 PGPASSWORD=sharding_user pgbench -h 127.0.0.1 -U sharding_user shard1 -i
@@ -26,7 +29,7 @@ wget -O toxiproxy-2.4.0.deb https://github.com/Shopify/toxiproxy/releases/downlo
 sudo dpkg -i toxiproxy-2.4.0.deb

 # Start Toxiproxy
-toxiproxy-server &
+LOG_LEVEL=error toxiproxy-server &
 sleep 1

 # Create a database at port 5433, forward it to Postgres
@@ -87,7 +90,8 @@ kill -SIGHUP $(pgrep pgcat) # Reload config again
 cd tests/ruby
 sudo gem install bundler
 bundle install
-ruby tests.rb
+bundle exec ruby tests.rb || exit 1
+bundle exec rspec *_spec.rb || exit 1
 cd ../..

 #
@@ -95,7 +99,7 @@ cd ../..
 # These tests will start and stop the pgcat server so it will need to be restarted after the tests
 #
 pip3 install -r tests/python/requirements.txt
-python3 tests/python/tests.py
+python3 tests/python/tests.py || exit 1

 start_pgcat "info"

@@ -105,9 +109,9 @@ psql -U admin_user -e -h 127.0.0.1 -p 6432 -d pgbouncer -c 'SHOW STATS' > /dev/n
 psql -U admin_user -h 127.0.0.1 -p 6432 -d pgbouncer -c 'RELOAD' > /dev/null
 psql -U admin_user -h 127.0.0.1 -p 6432 -d pgbouncer -c 'SHOW CONFIG' > /dev/null
 psql -U admin_user -h 127.0.0.1 -p 6432 -d pgbouncer -c 'SHOW DATABASES' > /dev/null
-psql -U admin_user -h 127.0.0.1 -p 6432 -d pgbouncer -c 'SHOW LISTS' > /dev/null
-psql -U admin_user -h 127.0.0.1 -p 6432 -d pgbouncer -c 'SHOW POOLS' > /dev/null
-psql -U admin_user -h 127.0.0.1 -p 6432 -d pgbouncer -c 'SHOW VERSION' > /dev/null
+psql -U admin_user -h 127.0.0.1 -p 6432 -d pgcat -c 'SHOW LISTS' > /dev/null
+psql -U admin_user -h 127.0.0.1 -p 6432 -d pgcat -c 'SHOW POOLS' > /dev/null
+psql -U admin_user -h 127.0.0.1 -p 6432 -d pgcat -c 'SHOW VERSION' > /dev/null
 psql -U admin_user -h 127.0.0.1 -p 6432 -d pgbouncer -c "SET client_encoding TO 'utf8'" > /dev/null # will ignore
 (! psql -U admin_user -e -h 127.0.0.1 -p 6432 -d random_db -c 'SHOW STATS' > /dev/null)
 export PGPASSWORD=sharding_user
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -53,14 +53,14 @@ checksum = "904dfeac50f3cdaba28fc6f57fdcddb75f49ed61346676a78c4ffe55877802fd"

 [[package]]
 name = "bb8"
-version = "0.7.1"
+version = "0.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2e9f4fa9768efd269499d8fba693260cfc670891cf6de3adc935588447a77cc8"
+checksum = "1627eccf3aa91405435ba240be23513eeca466b5dc33866422672264de061582"
 dependencies = [
 "async-trait",
 "futures-channel",
 "futures-util",
- "parking_lot",
+ "parking_lot 0.12.1",
 "tokio",
 ]

@@ -493,7 +493,17 @@ checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99"
 dependencies = [
 "instant",
 "lock_api",
- "parking_lot_core",
+ "parking_lot_core 0.8.5",
+]
+
+[[package]]
+name = "parking_lot"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f"
+dependencies = [
+ "lock_api",
+ "parking_lot_core 0.9.3",
 ]

 [[package]]
@@ -510,6 +520,19 @@ dependencies = [
 "winapi",
 ]

+[[package]]
+name = "parking_lot_core"
+version = "0.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09a279cbf25cb0757810394fbc1e359949b59e348145c643a939a525692e6929"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "redox_syscall",
+ "smallvec",
+ "windows-sys",
+]
+
 [[package]]
 name = "pgcat"
 version = "0.6.0-alpha1"
@@ -528,7 +551,7 @@ dependencies = [
 "md-5",
 "num_cpus",
 "once_cell",
- "parking_lot",
+ "parking_lot 0.12.1",
 "phf",
 "rand",
 "regex",
@@ -546,20 +569,19 @@ dependencies = [

 [[package]]
 name = "phf"
-version = "0.10.1"
+version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fabbf1ead8a5bcbc20f5f8b939ee3f5b0f6f281b6ad3468b84656b658b455259"
+checksum = "928c6535de93548188ef63bb7c4036bd415cd8f36ad25af44b9789b2ee72a48c"
 dependencies = [
 "phf_macros",
 "phf_shared",
- "proc-macro-hack",
 ]

 [[package]]
 name = "phf_generator"
-version = "0.10.0"
+version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d5285893bb5eb82e6aaf5d59ee909a06a16737a8970984dd7746ba9283498d6"
+checksum = "b1181c94580fa345f50f19d738aaa39c0ed30a600d95cb2d3e23f94266f14fbf"
 dependencies = [
 "phf_shared",
 "rand",
@@ -567,13 +589,12 @@ dependencies = [

 [[package]]
 name = "phf_macros"
-version = "0.10.0"
+version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "58fdf3184dd560f160dd73922bea2d5cd6e8f064bf4b13110abd81b03697b4e0"
+checksum = "92aacdc5f16768709a569e913f7451034034178b05bdc8acda226659a3dccc66"
 dependencies = [
 "phf_generator",
 "phf_shared",
- "proc-macro-hack",
 "proc-macro2",
 "quote",
 "syn",
@@ -581,9 +602,9 @@ dependencies = [

 [[package]]
 name = "phf_shared"
-version = "0.10.0"
+version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6796ad771acdc0123d2a88dc428b5e38ef24456743ddb1744ed628f9815c096"
+checksum = "e1fb5f6f826b772a8d4c0394209441e7d37cbbb967ae9c7e0e8134365c9ee676"
 dependencies = [
 "siphasher",
 ]
@@ -606,12 +627,6 @@ version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eb9f9e6e233e5c4a35559a617bf40a4ec447db2e84c20b55a6f83167b7e57872"

-[[package]]
-name = "proc-macro-hack"
-version = "0.5.19"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dbf0c48bc1d91375ae5c3cd81e3722dff1abcf81a30960240640d223f59fe0e5"
-
 [[package]]
 name = "proc-macro2"
 version = "1.0.36"
@@ -832,9 +847,9 @@ checksum = "6e63cff320ae2c57904679ba7cb63280a3dc4613885beafb148ee7bf9aa9042d"

 [[package]]
 name = "sqlparser"
-version = "0.14.0"
+version = "0.23.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b8f192f29f4aa49e57bebd0aa05858e0a1f32dd270af36efe49edb82cbfffab6"
+checksum = "0beb13adabbdda01b63d595f38c8bfd19a361e697fd94ce0098a634077bc5b25"
 dependencies = [
 "log",
 ]
@@ -913,7 +928,7 @@ dependencies = [
 "mio",
 "num_cpus",
 "once_cell",
- "parking_lot",
+ "parking_lot 0.11.2",
 "pin-project-lite",
 "signal-hook-registry",
 "tokio-macros",
@@ -1168,3 +1183,46 @@ name = "winapi-x86_64-pc-windows-gnu"
 version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
+
+[[package]]
+name = "windows-sys"
+version = "0.36.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ea04155a16a59f9eab786fe12a4a450e75cdb175f9e0d80da1e17db09f55b8d2"
+dependencies = [
+ "windows_aarch64_msvc",
+ "windows_i686_gnu",
+ "windows_i686_msvc",
+ "windows_x86_64_gnu",
+ "windows_x86_64_msvc",
+]
+
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.36.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9bb8c3fd39ade2d67e9874ac4f3db21f0d710bee00fe7cab16949ec184eeaa47"
+
+[[package]]
+name = "windows_i686_gnu"
+version = "0.36.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "180e6ccf01daf4c426b846dfc66db1fc518f074baa793aa7d9b9aaeffad6a3b6"
+
+[[package]]
+name = "windows_i686_msvc"
+version = "0.36.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e2e7917148b2812d1eeafaeb22a97e4813dfa60a3f8f78ebe204bcc88f12f024"
+
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.36.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4dcd171b8776c41b97521e5da127a2d86ad280114807d0b2ab1e462bc764d9e1"
+
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.36.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c811ca4a8c853ef420abd8592ba53ddbbac90410fab6903b3e79972a631f7680"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -9,7 +9,7 @@ edition = "2021"
 tokio = { version = "1", features = ["full"] }
 bytes = "1"
 md-5 = "0.10"
-bb8 = "0.7"
+bb8 = "0.8.0"
 async-trait = "0.1"
 rand = "0.8"
 chrono = "0.4"
@@ -20,11 +20,11 @@ serde_derive = "1"
 regex = "1"
 num_cpus = "1"
 once_cell = "1"
-sqlparser = "0.14"
+sqlparser = "0.23.0"
 log = "0.4"
 arc-swap = "1"
 env_logger = "0.9"
-parking_lot = "0.11"
+parking_lot = "0.12.1"
 hmac = "0.12"
 sha2 = "0.10"
 base64 = "0.13"
@@ -32,5 +32,5 @@ stringprep = "0.1"
 tokio-rustls = "0.23"
 rustls-pemfile = "1"
 hyper = { version = "0.14", features = ["full"] }
-phf = { version = "0.10", features = ["macros"] }
+phf = { version = "0.11.1", features = ["macros"] }
 exitcode = "1.1.2"
--- a/README.md
+++ b/README.md
@@ -1,8 +1,11 @@
-# PgCat
+![PgCat](./pgcat3.png)
+
+##### PgCat: PostgreSQL at petabyte scale

 [![CircleCI](https://circleci.com/gh/levkk/pgcat/tree/main.svg?style=svg)](https://circleci.com/gh/levkk/pgcat/tree/main)
-
-![PgCat](./pgcat3.png)
+<a href="https://discord.gg/DmyJP3qJ7U" target="_blank">
+    <img src="https://img.shields.io/discord/1013868243036930099" alt="Join our Discord!" />
+</a>

 PostgreSQL pooler (like PgBouncer) with sharding, load balancing and failover support.

@@ -57,6 +60,7 @@ psql -h 127.0.0.1 -p 6432 -c 'SELECT 1'
 | **`user`**                   |                                                                                                                                            |                                  |
 | `name`                       | The user name.                                                                                                                             | `sharding_user`                  |
 | `password`                   | The user password in plaintext.                                                                                                            | `hunter2`                        |
+| `statement_timeout` | Timeout in milliseconds for how long a query takes to execute | `0 (disabled)`
 |                              |                                                                                                                                            |                                  |
 | **`shards`**                 | Shards are numerically numbered starting from 0; the order in the config is preserved by the pooler to route queries accordingly.          | `[shards.0]`                     |
 | `servers`                    | List of servers to connect to and their roles. A server is: `[host, port, role]`, where `role` is either `primary` or `replica`.           | `["127.0.0.1", 5432, "primary"]` |
@@ -87,6 +91,14 @@ pgbench -t 1000 -p 6432 -h 127.0.0.1 --protocol extended

 See [sharding README](./tests/sharding/README.md) for sharding logic testing.

+Run `cargo test` to run Rust tests.
+
+Run the following commands to run Integration tests locally.
+```
+cd tests/docker/
+docker compose up --exit-code-from main # This will also produce coverage report under ./cov/
+```
+
 | **Feature**           | **Tested in CI**   | **Tested manually** | **Comments**                                                                                                             |
 |-----------------------|--------------------|---------------------|--------------------------------------------------------------------------------------------------------------------------|
 | Transaction pooling   | :white_check_mark: | :white_check_mark:  | Used by default for all tests.                                                                                           |
@@ -447,7 +459,7 @@ Always good to have a base line.

 ```
 $ pgbench -t 1000 -c 16 -j 2 -p 5432 -h 127.0.0.1 -S --protocol extended shard0
-Password: 
+Password:
 starting vacuum...end.
 transaction type: <builtin: select only>
 scaling factor: 1
@@ -461,7 +473,7 @@ tps = 139443.955722 (including connections establishing)
 tps = 142314.859075 (excluding connections establishing)

 $ pgbench -t 1000 -c 32 -j 2 -p 5432 -h 127.0.0.1 -S --protocol extended shard0
-Password: 
+Password:
 starting vacuum...end.
 transaction type: <builtin: select only>
 scaling factor: 1
@@ -475,7 +487,7 @@ tps = 150644.840891 (including connections establishing)
 tps = 152218.499430 (excluding connections establishing)

 $ pgbench -t 1000 -c 64 -j 2 -p 5432 -h 127.0.0.1 -S --protocol extended shard0
-Password: 
+Password:
 starting vacuum...end.
 transaction type: <builtin: select only>
 scaling factor: 1
@@ -489,7 +501,7 @@ tps = 152517.663404 (including connections establishing)
 tps = 153319.188482 (excluding connections establishing)

 $ pgbench -t 1000 -c 128 -j 2 -p 5432 -h 127.0.0.1 -S --protocol extended shard0
-Password: 
+Password:
 starting vacuum...end.
 transaction type: <builtin: select only>
 scaling factor: 1
--- a/src/admin.rs
+++ b/src/admin.rs
@@ -2,22 +2,25 @@
 use bytes::{Buf, BufMut, BytesMut};
 use log::{info, trace};
 use std::collections::HashMap;
+use tokio::time::Instant;

 use crate::config::{get_config, reload_config, VERSION};
 use crate::errors::Error;
 use crate::messages::*;
 use crate::pool::get_all_pools;
-use crate::stats::get_stats;
+use crate::stats::{
+    get_address_stats, get_client_stats, get_pool_stats, get_server_stats, ClientState, ServerState,
+};
 use crate::ClientServerMap;

 pub fn generate_server_info_for_admin() -> BytesMut {
    let mut server_info = BytesMut::new();

-    server_info.put(server_paramater_message("application_name", ""));
-    server_info.put(server_paramater_message("client_encoding", "UTF8"));
-    server_info.put(server_paramater_message("server_encoding", "UTF8"));
-    server_info.put(server_paramater_message("server_version", VERSION));
-    server_info.put(server_paramater_message("DateStyle", "ISO, MDY"));
+    server_info.put(server_parameter_message("application_name", ""));
+    server_info.put(server_parameter_message("client_encoding", "UTF8"));
+    server_info.put(server_parameter_message("server_encoding", "UTF8"));
+    server_info.put(server_parameter_message("server_version", VERSION));
+    server_info.put(server_parameter_message("DateStyle", "ISO, MDY"));

    return server_info;
 }
@@ -72,6 +75,14 @@ where
                trace!("SHOW POOLS");
                show_pools(stream).await
            }
+            "CLIENTS" => {
+                trace!("SHOW CLIENTS");
+                show_clients(stream).await
+            }
+            "SERVERS" => {
+                trace!("SHOW SERVERS");
+                show_servers(stream).await
+            }
            "STATS" => {
                trace!("SHOW STATS");
                show_stats(stream).await
@@ -91,7 +102,8 @@ async fn show_lists<T>(stream: &mut T) -> Result<(), Error>
 where
    T: tokio::io::AsyncWrite + std::marker::Unpin,
 {
-    let stats = get_stats();
+    let client_stats = get_client_stats();
+    let server_stats = get_server_stats();

    let columns = vec![("list", DataType::Text), ("items", DataType::Int4)];

@@ -111,18 +123,18 @@ where
    res.put(data_row(&vec!["pools".to_string(), databases.to_string()]));
    res.put(data_row(&vec![
        "free_clients".to_string(),
-        stats
+        client_stats
            .keys()
-            .map(|address_id| stats[&address_id]["cl_idle"])
-            .sum::<i64>()
+            .filter(|client_id| client_stats.get(client_id).unwrap().state == ClientState::Idle)
+            .count()
            .to_string(),
    ]));
    res.put(data_row(&vec![
        "used_clients".to_string(),
-        stats
+        client_stats
            .keys()
-            .map(|address_id| stats[&address_id]["cl_active"])
-            .sum::<i64>()
+            .filter(|client_id| client_stats.get(client_id).unwrap().state == ClientState::Active)
+            .count()
            .to_string(),
    ]));
    res.put(data_row(&vec![
@@ -131,18 +143,18 @@ where
    ]));
    res.put(data_row(&vec![
        "free_servers".to_string(),
-        stats
+        server_stats
            .keys()
-            .map(|address_id| stats[&address_id]["sv_idle"])
-            .sum::<i64>()
+            .filter(|server_id| server_stats.get(server_id).unwrap().state == ServerState::Idle)
+            .count()
            .to_string(),
    ]));
    res.put(data_row(&vec![
        "used_servers".to_string(),
-        stats
+        server_stats
            .keys()
-            .map(|address_id| stats[&address_id]["sv_active"])
-            .sum::<i64>()
+            .filter(|server_id| server_stats.get(server_id).unwrap().state == ServerState::Active)
+            .count()
            .to_string(),
    ]));
    res.put(data_row(&vec!["dns_names".to_string(), "0".to_string()]));
@@ -182,11 +194,12 @@ async fn show_pools<T>(stream: &mut T) -> Result<(), Error>
 where
    T: tokio::io::AsyncWrite + std::marker::Unpin,
 {
-    let stats = get_stats();
+    let all_pool_stats = get_pool_stats();

    let columns = vec![
        ("database", DataType::Text),
        ("user", DataType::Text),
+        ("pool_mode", DataType::Text),
        ("cl_idle", DataType::Numeric),
        ("cl_active", DataType::Numeric),
        ("cl_waiting", DataType::Numeric),
@@ -198,32 +211,33 @@ where
        ("sv_login", DataType::Numeric),
        ("maxwait", DataType::Numeric),
        ("maxwait_us", DataType::Numeric),
-        ("pool_mode", DataType::Text),
    ];

    let mut res = BytesMut::new();
    res.put(row_description(&columns));
-    for (_, pool) in get_all_pools() {
+    for (user_pool, pool) in get_all_pools() {
+        let def = HashMap::default();
+        let pool_stats = all_pool_stats
+            .get(&(user_pool.db.clone(), user_pool.user.clone()))
+            .unwrap_or(&def);
+
        let pool_config = &pool.settings;
-        for shard in 0..pool.shards() {
-            for server in 0..pool.servers(shard) {
-                let address = pool.address(shard, server);
-                let stats = match stats.get(&address.id) {
-                    Some(stats) => stats.clone(),
-                    None => HashMap::new(),
-                };
-
-                let mut row = vec![address.name(), pool_config.user.username.clone()];
-
-                for column in &columns[2..columns.len() - 1] {
-                    let value = stats.get(column.0).unwrap_or(&0).to_string();
-                    row.push(value);
+        let mut row = vec![
+            user_pool.db.clone(),
+            user_pool.user.clone(),
+            pool_config.pool_mode.to_string(),
+        ];
+        for column in &columns[3..columns.len()] {
+            let value = match column.0 {
+                "maxwait" => (pool_stats.get("maxwait_us").unwrap_or(&0) / 1_000_000).to_string(),
+                "maxwait_us" => {
+                    (pool_stats.get("maxwait_us").unwrap_or(&0) % 1_000_000).to_string()
                }
-
-                row.push(pool_config.pool_mode.to_string());
-                res.put(data_row(&row));
-            }
+                _other_values => pool_stats.get(column.0).unwrap_or(&0).to_string(),
+            };
+            row.push(value);
        }
+        res.put(data_row(&row));
    }

    res.put(command_complete("SHOW"));
@@ -387,6 +401,7 @@ where
    T: tokio::io::AsyncWrite + std::marker::Unpin,
 {
    let columns = vec![
+        ("instance", DataType::Text),
        ("database", DataType::Text),
        ("user", DataType::Text),
        ("total_xact_count", DataType::Numeric),
@@ -396,32 +411,32 @@ where
        ("total_xact_time", DataType::Numeric),
        ("total_query_time", DataType::Numeric),
        ("total_wait_time", DataType::Numeric),
+        ("total_errors", DataType::Numeric),
        ("avg_xact_count", DataType::Numeric),
        ("avg_query_count", DataType::Numeric),
        ("avg_recv", DataType::Numeric),
        ("avg_sent", DataType::Numeric),
+        ("avg_errors", DataType::Numeric),
        ("avg_xact_time", DataType::Numeric),
        ("avg_query_time", DataType::Numeric),
        ("avg_wait_time", DataType::Numeric),
    ];

-    let stats = get_stats();
+    let all_stats = get_address_stats();
    let mut res = BytesMut::new();
    res.put(row_description(&columns));

-    for ((_db_name, username), pool) in get_all_pools() {
+    for (user_pool, pool) in get_all_pools() {
        for shard in 0..pool.shards() {
            for server in 0..pool.servers(shard) {
                let address = pool.address(shard, server);
-                let stats = match stats.get(&address.id) {
+                let stats = match all_stats.get(&address.id) {
                    Some(stats) => stats.clone(),
                    None => HashMap::new(),
                };

-                let mut row = vec![address.name()];
-                row.push(username.clone());
-
-                for column in &columns[2..] {
+                let mut row = vec![address.name(), user_pool.db.clone(), user_pool.user.clone()];
+                for column in &columns[3..] {
                    row.push(stats.get(column.0).unwrap_or(&0).to_string());
                }

@@ -439,3 +454,107 @@ where

    write_all_half(stream, res).await
 }
+
+/// Show currently connected clients
+async fn show_clients<T>(stream: &mut T) -> Result<(), Error>
+where
+    T: tokio::io::AsyncWrite + std::marker::Unpin,
+{
+    let columns = vec![
+        ("client_id", DataType::Text),
+        ("database", DataType::Text),
+        ("user", DataType::Text),
+        ("application_name", DataType::Text),
+        ("state", DataType::Text),
+        ("transaction_count", DataType::Numeric),
+        ("query_count", DataType::Numeric),
+        ("error_count", DataType::Numeric),
+        ("age_seconds", DataType::Numeric),
+    ];
+
+    let new_map = get_client_stats();
+    let mut res = BytesMut::new();
+    res.put(row_description(&columns));
+
+    for (_, client) in new_map {
+        let row = vec![
+            format!("{:#010X}", client.client_id),
+            client.pool_name,
+            client.username,
+            client.application_name.clone(),
+            client.state.to_string(),
+            client.transaction_count.to_string(),
+            client.query_count.to_string(),
+            client.error_count.to_string(),
+            Instant::now()
+                .duration_since(client.connect_time)
+                .as_secs()
+                .to_string(),
+        ];
+
+        res.put(data_row(&row));
+    }
+
+    res.put(command_complete("SHOW"));
+
+    // ReadyForQuery
+    res.put_u8(b'Z');
+    res.put_i32(5);
+    res.put_u8(b'I');
+
+    write_all_half(stream, res).await
+}
+
+/// Show currently connected servers
+async fn show_servers<T>(stream: &mut T) -> Result<(), Error>
+where
+    T: tokio::io::AsyncWrite + std::marker::Unpin,
+{
+    let columns = vec![
+        ("server_id", DataType::Text),
+        ("database_name", DataType::Text),
+        ("user", DataType::Text),
+        ("address_id", DataType::Text),
+        ("application_name", DataType::Text),
+        ("state", DataType::Text),
+        ("transaction_count", DataType::Numeric),
+        ("query_count", DataType::Numeric),
+        ("bytes_sent", DataType::Numeric),
+        ("bytes_received", DataType::Numeric),
+        ("age_seconds", DataType::Numeric),
+    ];
+
+    let new_map = get_server_stats();
+    let mut res = BytesMut::new();
+    res.put(row_description(&columns));
+
+    for (_, server) in new_map {
+        let row = vec![
+            format!("{:#010X}", server.server_id),
+            server.pool_name,
+            server.username,
+            server.address_name,
+            server.application_name,
+            server.state.to_string(),
+            server.transaction_count.to_string(),
+            server.query_count.to_string(),
+            server.bytes_sent.to_string(),
+            server.bytes_received.to_string(),
+            Instant::now()
+                .duration_since(server.connect_time)
+                .as_secs()
+                .to_string(),
+        ];
+
+        res.put(data_row(&row));
+    }
+
+    res.put(command_complete("SHOW"));
+
+    // ReadyForQuery
+    res.put_u8(b'Z');
+    res.put_i32(5);
+    res.put_u8(b'I');
+
+    write_all_half(stream, res).await
+}
--- a/src/client.rs
+++ b/src/client.rs
@@ -1,18 +1,19 @@
 /// Handle clients by pretending to be a PostgreSQL server.
 use bytes::{Buf, BufMut, BytesMut};
-use log::{debug, error, info, trace};
+use log::{debug, error, info, trace, warn};
 use std::collections::HashMap;
+use std::time::Instant;
 use tokio::io::{split, AsyncReadExt, BufReader, ReadHalf, WriteHalf};
 use tokio::net::TcpStream;
 use tokio::sync::broadcast::Receiver;
 use tokio::sync::mpsc::Sender;

 use crate::admin::{generate_server_info_for_admin, handle_admin};
-use crate::config::{get_config, Address};
+use crate::config::{get_config, Address, PoolMode};
 use crate::constants::*;
 use crate::errors::Error;
 use crate::messages::*;
-use crate::pool::{get_pool, ClientServerMap, ConnectionPool, PoolMode};
+use crate::pool::{get_pool, ClientServerMap, ConnectionPool};
 use crate::query_router::{Command, QueryRouter};
 use crate::server::Server;
 use crate::stats::{get_reporter, Reporter};
@@ -59,6 +60,7 @@ pub struct Client<S, T> {
    client_server_map: ClientServerMap,

    /// Client parameters, e.g. user, client_encoding, etc.
+    #[allow(dead_code)]
    parameters: HashMap<String, String>,

    /// Statistics
@@ -82,6 +84,9 @@ pub struct Client<S, T> {
    /// Postgres user for this client (This comes from the user in the connection string)
    username: String,

+    /// Application name for this client (defaults to pgcat)
+    application_name: String,
+
    /// Used to notify clients about an impending shutdown
    shutdown: Receiver<()>,
 }
@@ -91,7 +96,7 @@ pub async fn client_entrypoint(
    mut stream: TcpStream,
    client_server_map: ClientServerMap,
    shutdown: Receiver<()>,
-    drain: Sender<i8>,
+    drain: Sender<i32>,
    admin_only: bool,
 ) -> Result<(), Error> {
    // Figure out if the client wants TLS or not.
@@ -200,7 +205,7 @@ pub async fn client_entrypoint(
                Ok(mut client) => {
                    info!("Client {:?} connected (plain)", addr);

-                    if client.is_admin() {
+                    if !client.is_admin() {
                        let _ = drain.send(1).await;
                    }

@@ -225,7 +230,7 @@ pub async fn client_entrypoint(
                Ok(mut client) => {
                    info!("Client {:?} issued a cancel query request", addr);

-                    if client.is_admin() {
+                    if !client.is_admin() {
                        let _ = drain.send(1).await;
                    }

@@ -365,6 +370,11 @@ where
            None => return Err(Error::ClientError),
        };

+        let application_name = match parameters.get("application_name") {
+            Some(application_name) => application_name,
+            None => "pgcat",
+        };
+
        let admin = ["pgcat", "pgbouncer"]
            .iter()
            .filter(|db| *db == &pool_name)
@@ -426,7 +436,7 @@ where
            );

            if password_hash != password_response {
-                debug!("Password authentication failed");
+                warn!("Invalid password {{ username: {:?}, pool_name: {:?}, application_name: {:?} }}", pool_name, username, application_name);
                wrong_password(&mut write, username).await?;

                return Err(Error::ClientError);
@@ -436,7 +446,7 @@ where
        }
        // Authenticate normal user.
        else {
-            let pool = match get_pool(pool_name.clone(), username.clone()) {
+            let pool = match get_pool(&pool_name, &username) {
                Some(pool) => pool,
                None => {
                    error_response(
@@ -448,6 +458,7 @@ where
                    )
                    .await?;

+                    warn!("Invalid pool name {{ username: {:?}, pool_name: {:?}, application_name: {:?} }}", pool_name, username, application_name);
                    return Err(Error::ClientError);
                }
            };
@@ -456,7 +467,7 @@ where
            let password_hash = md5_hash_password(&username, &pool.settings.user.password, &salt);

            if password_hash != password_response {
-                debug!("Password authentication failed");
+                warn!("Invalid password {{ username: {:?}, pool_name: {:?}, application_name: {:?} }}", pool_name, username, application_name);
                wrong_password(&mut write, username).await?;

                return Err(Error::ClientError);
@@ -493,6 +504,7 @@ where
            last_server_id: None,
            pool_name: pool_name.clone(),
            username: username.clone(),
+            application_name: application_name.to_string(),
            shutdown,
            connected_to_server: false,
        });
@@ -526,6 +538,7 @@ where
            last_server_id: None,
            pool_name: String::from("undefined"),
            username: String::from("undefined"),
+            application_name: String::from("undefined"),
            shutdown,
            connected_to_server: false,
        });
@@ -566,6 +579,12 @@ where
        // The query router determines where the query is going to go,
        // e.g. primary, replica, which shard.
        let mut query_router = QueryRouter::new();
+        self.stats.client_register(
+            self.process_id,
+            self.pool_name.clone(),
+            self.username.clone(),
+            self.application_name.clone(),
+        );

        // Our custom protocol loop.
        // We expect the client to either start a transaction with regular queries
@@ -600,10 +619,23 @@ where
                message_result = read_message(&mut self.read) => message_result?
            };

-            // Avoid taking a server if the client just wants to disconnect.
-            if message[0] as char == 'X' {
-                debug!("Client disconnecting");
-                return Ok(());
+            match message[0] as char {
+                // Buffer extended protocol messages even if we do not have
+                // a server connection yet. Hopefully, when we get the S message
+                // we'll be able to allocate a connection. Also, clients do not expect
+                // the server to respond to these messages so even if we were not able to
+                // allocate a connection, we wouldn't be able to send back an error message
+                // to the client so we buffer them and defer the decision to error out or not
+                // to when we get the S message
+                'P' | 'B' | 'D' | 'E' => {
+                    self.buffer.put(&message[..]);
+                    continue;
+                }
+                'X' => {
+                    debug!("Client disconnecting");
+                    return Ok(());
+                }
+                _ => (),
            }

            // Handle admin database queries.
@@ -616,7 +648,7 @@ where
            // Get a pool instance referenced by the most up-to-date
            // pointer. This ensures we always read the latest config
            // when starting a query.
-            let pool = match get_pool(self.pool_name.clone(), self.username.clone()) {
+            let pool = match get_pool(&self.pool_name, &self.username) {
                Some(pool) => pool,
                None => {
                    error_response(
@@ -627,6 +659,8 @@ where
                        ),
                    )
                    .await?;
+
+                    warn!("Invalid pool name {{ username: {:?}, pool_name: {:?}, application_name: {:?} }}", self.pool_name, self.username, self.application_name);
                    return Err(Error::ClientError);
                }
            };
@@ -714,22 +748,19 @@ where
                    conn
                }
                Err(err) => {
-                    // Clients do not expect to get SystemError followed by ReadyForQuery in the middle
-                    // of extended protocol submission. So we will hold off on sending the actual error
-                    // message to the client until we get 'S' message
-                    match message[0] as char {
-                        'P' | 'B' | 'E' | 'D' => (),
-                        _ => {
-                            error_response(
-                                &mut self.write,
-                                "could not get connection from the pool",
-                            )
-                            .await?;
-                        }
-                    };
-
-                    error!("Could not get connection from pool: {:?}", err);
+                    // Client is attempting to get results from the server,
+                    // but we were unable to grab a connection from the pool
+                    // We'll send back an error message and clean the extended
+                    // protocol buffer
+                    if message[0] as char == 'S' {
+                        error!("Got Sync message but failed to get a connection from the pool");
+                        self.buffer.clear();
+                    }
+                    error_response(&mut self.write, "could not get connection from the pool")
+                        .await?;

+                    error!("Could not get connection from pool: {{ pool_name: {:?}, username: {:?}, shard: {:?}, role: \"{:?}\", error: \"{:?}\" }}",
+                    self.pool_name.clone(), self.username.clone(), query_router.shard(), query_router.role(), err);
                    continue;
                }
            };
@@ -743,15 +774,12 @@ where
            server.claim(self.process_id, self.secret_key);
            self.connected_to_server = true;

-            // Update statistics.
-            if let Some(last_address_id) = self.last_address_id {
-                self.stats
-                    .client_disconnecting(self.process_id, last_address_id);
-            }
-            self.stats.client_active(self.process_id, address.id);
+            // Update statistics
+            self.stats
+                .client_active(self.process_id, server.server_id());

            self.last_address_id = Some(address.id);
-            self.last_server_id = Some(server.process_id());
+            self.last_server_id = Some(server.server_id());

            debug!(
                "Client {:?} talking to server {:?}",
@@ -759,13 +787,10 @@ where
                server.address()
            );

-            // Set application_name if any.
            // TODO: investigate other parameters and set them too.
-            if self.parameters.contains_key("application_name") {
-                server
-                    .set_name(&self.parameters["application_name"])
-                    .await?;
-            }
+
+            // Set application_name.
+            server.set_name(&self.application_name).await?;

            // Transaction loop. Multiple queries can be issued by the client here.
            // The connection belongs to the client until the transaction is over,
@@ -782,12 +807,7 @@ where
                        Err(err) => {
                            // Client disconnected inside a transaction.
                            // Clean up the server and re-use it.
-                            // This prevents connection thrashing by bad clients.
-                            if server.in_transaction() {
-                                server.query("ROLLBACK").await?;
-                                server.query("DISCARD ALL").await?;
-                                server.set_name("pgcat").await?;
-                            }
+                            server.checkin_cleanup().await?;

                            return Err(err);
                        }
@@ -817,7 +837,7 @@ where

                        if !server.in_transaction() {
                            // Report transaction executed statistics.
-                            self.stats.transaction(self.process_id, address.id);
+                            self.stats.transaction(self.process_id, server.server_id());

                            // Release server back to the pool if we are in transaction mode.
                            // If we are in session mode, we keep the server until the client disconnects.
@@ -829,16 +849,7 @@ where

                    // Terminate
                    'X' => {
-                        // Client closing. Rollback and clean up
-                        // connection before releasing into the pool.
-                        // Pgbouncer closes the connection which leads to
-                        // connection thrashing when clients misbehave.
-                        if server.in_transaction() {
-                            server.query("ROLLBACK").await?;
-                            server.query("DISCARD ALL").await?;
-                            server.set_name("pgcat").await?;
-                        }
-
+                        server.checkin_cleanup().await?;
                        self.release();

                        return Ok(());
@@ -875,6 +886,23 @@ where

                        self.buffer.put(&original[..]);

+                        // Clone after freeze does not allocate
+                        let first_message_code = (*self.buffer.get(0).unwrap_or(&0)) as char;
+
+                        // Almost certainly true
+                        if first_message_code == 'P' {
+                            // Message layout
+                            // P followed by 32 int followed by null-terminated statement name
+                            // So message code should be in offset 0 of the buffer, first character
+                            // in prepared statement name would be index 5
+                            let first_char_in_name = *self.buffer.get(5).unwrap_or(&0);
+                            if first_char_in_name != 0 {
+                                // This is a named prepared statement
+                                // Server connection state will need to be cleared at checkin
+                                server.mark_dirty();
+                            }
+                        }
+
                        self.send_and_receive_loop(
                            code,
                            self.buffer.clone(),
@@ -887,7 +915,7 @@ where
                        self.buffer.clear();

                        if !server.in_transaction() {
-                            self.stats.transaction(self.process_id, address.id);
+                            self.stats.transaction(self.process_id, server.server_id());

                            // Release server back to the pool if we are in transaction mode.
                            // If we are in session mode, we keep the server until the client disconnects.
@@ -922,7 +950,7 @@ where
                        };

                        if !server.in_transaction() {
-                            self.stats.transaction(self.process_id, address.id);
+                            self.stats.transaction(self.process_id, server.server_id());

                            // Release server back to the pool if we are in transaction mode.
                            // If we are in session mode, we keep the server until the client disconnects.
@@ -942,10 +970,12 @@ where

            // The server is no longer bound to us, we can't cancel it's queries anymore.
            debug!("Releasing server back into the pool");
-            self.stats.server_idle(server.process_id(), address.id);
+            server.checkin_cleanup().await?;
+            self.stats.server_idle(server.server_id());
            self.connected_to_server = false;
+
            self.release();
-            self.stats.client_idle(self.process_id, address.id);
+            self.stats.client_idle(self.process_id);
        }
    }

@@ -968,6 +998,7 @@ where
        self.send_server_message(server, message, &address, &pool)
            .await?;

+        let query_start = Instant::now();
        // Read all data the server has to offer, which can be multiple messages
        // buffered in 8196 bytes chunks.
        loop {
@@ -987,7 +1018,11 @@ where
        }

        // Report query executed statistics.
-        self.stats.query(self.process_id, address.id);
+        self.stats.query(
+            self.process_id,
+            server.server_id(),
+            Instant::now().duration_since(query_start).as_millis(),
+        );

        Ok(())
    }
@@ -1068,14 +1103,9 @@ impl<S, T> Drop for Client<S, T> {

        // Dirty shutdown
        // TODO: refactor, this is not the best way to handle state management.
-        if let Some(address_id) = self.last_address_id {
-            self.stats.client_disconnecting(self.process_id, address_id);
-
-            if self.connected_to_server {
-                if let Some(process_id) = self.last_server_id {
-                    self.stats.server_idle(process_id, address_id);
-                }
-            }
+        self.stats.client_disconnecting(self.process_id);
+        if self.connected_to_server && self.last_server_id.is_some() {
+            self.stats.server_idle(self.last_server_id.unwrap());
        }
    }
 }
--- a/src/config.rs
+++ b/src/config.rs
@@ -3,7 +3,7 @@ use arc_swap::ArcSwap;
 use log::{error, info};
 use once_cell::sync::Lazy;
 use serde_derive::{Deserialize, Serialize};
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::hash::Hash;
 use std::path::Path;
 use std::sync::Arc;
@@ -12,8 +12,9 @@ use tokio::io::AsyncReadExt;
 use toml;

 use crate::errors::Error;
+use crate::pool::{ClientServerMap, ConnectionPool};
+use crate::sharding::ShardingFunction;
 use crate::tls::{load_certs, load_keys};
-use crate::{ClientServerMap, ConnectionPool};

 pub const VERSION: &str = env!("CARGO_PKG_VERSION");

@@ -23,7 +24,9 @@ static CONFIG: Lazy<ArcSwap<Config>> = Lazy::new(|| ArcSwap::from_pointee(Config
 /// Server role: primary or replica.
 #[derive(Clone, PartialEq, Serialize, Deserialize, Hash, std::cmp::Eq, Debug, Copy)]
 pub enum Role {
+    #[serde(alias = "primary", alias = "Primary")]
    Primary,
+    #[serde(alias = "replica", alias = "Replica")]
    Replica,
 }

@@ -120,11 +123,12 @@ impl Address {
 }

 /// PostgreSQL user.
-#[derive(Clone, PartialEq, Hash, std::cmp::Eq, Serialize, Deserialize, Debug)]
+#[derive(Clone, PartialEq, Hash, Eq, Serialize, Deserialize, Debug)]
 pub struct User {
    pub username: String,
    pub password: String,
    pub pool_size: u32,
+    #[serde(default)] // 0
    pub statement_timeout: u64,
 }

@@ -142,34 +146,81 @@ impl Default for User {
 /// General configuration.
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub struct General {
+    #[serde(default = "General::default_host")]
    pub host: String,
+
+    #[serde(default = "General::default_port")]
    pub port: i16,
+
    pub enable_prometheus_exporter: Option<bool>,
    pub prometheus_exporter_port: i16,
+
+    #[serde(default = "General::default_connect_timeout")]
    pub connect_timeout: u64,
-    pub healthcheck_timeout: u64,
+
+    #[serde(default = "General::default_shutdown_timeout")]
    pub shutdown_timeout: u64,
+
+    #[serde(default = "General::default_healthcheck_timeout")]
+    pub healthcheck_timeout: u64,
+
+    #[serde(default = "General::default_healthcheck_delay")]
    pub healthcheck_delay: u64,
+
+    #[serde(default = "General::default_ban_time")]
    pub ban_time: i64,
+
+    #[serde(default)] // False
    pub autoreload: bool,
+
    pub tls_certificate: Option<String>,
    pub tls_private_key: Option<String>,
    pub admin_username: String,
    pub admin_password: String,
 }

+impl General {
+    pub fn default_host() -> String {
+        "0.0.0.0".into()
+    }
+
+    pub fn default_port() -> i16 {
+        5432
+    }
+
+    pub fn default_connect_timeout() -> u64 {
+        1000
+    }
+
+    pub fn default_shutdown_timeout() -> u64 {
+        60000
+    }
+
+    pub fn default_healthcheck_timeout() -> u64 {
+        1000
+    }
+
+    pub fn default_healthcheck_delay() -> u64 {
+        30000
+    }
+
+    pub fn default_ban_time() -> i64 {
+        60
+    }
+}
+
 impl Default for General {
    fn default() -> General {
        General {
-            host: String::from("localhost"),
-            port: 5432,
+            host: Self::default_host(),
+            port: Self::default_port(),
            enable_prometheus_exporter: Some(false),
            prometheus_exporter_port: 9930,
-            connect_timeout: 5000,
-            healthcheck_timeout: 1000,
-            shutdown_timeout: 60000,
-            healthcheck_delay: 30000,
-            ban_time: 60,
+            connect_timeout: General::default_connect_timeout(),
+            shutdown_timeout: Self::default_shutdown_timeout(),
+            healthcheck_timeout: Self::default_healthcheck_timeout(),
+            healthcheck_delay: Self::default_healthcheck_delay(),
+            ban_time: Self::default_ban_time(),
            autoreload: false,
            tls_certificate: None,
            tls_private_key: None,
@@ -178,50 +229,166 @@ impl Default for General {
        }
    }
 }
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
-pub struct Pool {
-    pub pool_mode: String,
-    pub default_role: String,
-    pub query_parser_enabled: bool,
-    pub primary_reads_enabled: bool,
-    pub sharding_function: String,
-    pub shards: HashMap<String, Shard>,
-    pub users: HashMap<String, User>,
+
+/// Pool mode:
+/// - transaction: server serves one transaction,
+/// - session: server is attached to the client.
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Copy, Hash)]
+pub enum PoolMode {
+    #[serde(alias = "transaction", alias = "Transaction")]
+    Transaction,
+
+    #[serde(alias = "session", alias = "Session")]
+    Session,
 }
-impl Default for Pool {
-    fn default() -> Pool {
-        Pool {
-            pool_mode: String::from("transaction"),
-            shards: HashMap::from([(String::from("1"), Shard::default())]),
-            users: HashMap::default(),
-            default_role: String::from("any"),
-            query_parser_enabled: false,
-            primary_reads_enabled: true,
-            sharding_function: "pg_bigint_hash".to_string(),
+
+impl ToString for PoolMode {
+    fn to_string(&self) -> String {
+        match *self {
+            PoolMode::Transaction => "transaction".to_string(),
+            PoolMode::Session => "session".to_string(),
        }
    }
 }

+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)]
+pub struct Pool {
+    #[serde(default = "Pool::default_pool_mode")]
+    pub pool_mode: PoolMode,
+
+    pub default_role: String,
+
+    #[serde(default)] // False
+    pub query_parser_enabled: bool,
+
+    #[serde(default)] // False
+    pub primary_reads_enabled: bool,
+
+    pub connect_timeout: Option<u64>,
+
+    pub sharding_function: ShardingFunction,
+    pub shards: BTreeMap<String, Shard>,
+    pub users: BTreeMap<String, User>,
+}
+
+impl Pool {
+    pub fn default_pool_mode() -> PoolMode {
+        PoolMode::Transaction
+    }
+
+    pub fn validate(&self) -> Result<(), Error> {
+        match self.default_role.as_ref() {
+            "any" => (),
+            "primary" => (),
+            "replica" => (),
+            other => {
+                error!(
+                    "Query router default_role must be 'primary', 'replica', or 'any', got: '{}'",
+                    other
+                );
+                return Err(Error::BadConfig);
+            }
+        };
+
+        for (shard_idx, shard) in &self.shards {
+            match shard_idx.parse::<usize>() {
+                Ok(_) => (),
+                Err(_) => {
+                    error!(
+                        "Shard '{}' is not a valid number, shards must be numbered starting at 0",
+                        shard_idx
+                    );
+                    return Err(Error::BadConfig);
+                }
+            };
+            shard.validate()?;
+        }
+
+        Ok(())
+    }
+}
+
+impl Default for Pool {
+    fn default() -> Pool {
+        Pool {
+            pool_mode: Self::default_pool_mode(),
+            shards: BTreeMap::from([(String::from("1"), Shard::default())]),
+            users: BTreeMap::default(),
+            default_role: String::from("any"),
+            query_parser_enabled: false,
+            primary_reads_enabled: false,
+            sharding_function: ShardingFunction::PgBigintHash,
+            connect_timeout: None,
+        }
+    }
+}
+
+#[derive(Clone, PartialEq, Serialize, Deserialize, Debug, Hash, Eq)]
+pub struct ServerConfig {
+    pub host: String,
+    pub port: u16,
+    pub role: Role,
+}
+
 /// Shard configuration.
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Hash, Eq)]
 pub struct Shard {
    pub database: String,
-    pub servers: Vec<(String, u16, String)>,
+    pub servers: Vec<ServerConfig>,
+}
+
+impl Shard {
+    pub fn validate(&self) -> Result<(), Error> {
+        // We use addresses as unique identifiers,
+        // let's make sure they are unique in the config as well.
+        let mut dup_check = HashSet::new();
+        let mut primary_count = 0;
+
+        if self.servers.len() == 0 {
+            error!("Shard {} has no servers configured", self.database);
+            return Err(Error::BadConfig);
+        }
+
+        for server in &self.servers {
+            dup_check.insert(server);
+
+            // Check that we define only zero or one primary.
+            match server.role {
+                Role::Primary => primary_count += 1,
+                _ => (),
+            };
+        }
+
+        if primary_count > 1 {
+            error!(
+                "Shard {} has more than on primary configured",
+                self.database
+            );
+            return Err(Error::BadConfig);
+        }
+
+        if dup_check.len() != self.servers.len() {
+            error!("Shard {} contains duplicate server configs", self.database);
+            return Err(Error::BadConfig);
+        }
+
+        Ok(())
+    }
 }

 impl Default for Shard {
    fn default() -> Shard {
        Shard {
-            servers: vec![(String::from("localhost"), 5432, String::from("primary"))],
+            servers: vec![ServerConfig {
+                host: String::from("localhost"),
+                port: 5432,
+                role: Role::Primary,
+            }],
            database: String::from("postgres"),
        }
    }
 }

-fn default_path() -> String {
-    String::from("pgcat.toml")
-}
-
 /// Configuration wrapper.
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub struct Config {
@@ -229,24 +396,30 @@ pub struct Config {
    // so we should always put simple fields before nested fields
    // in all serializable structs to avoid ValueAfterTable errors
    // These errors occur when the toml serializer is about to produce
-    // ambigous toml structure like the one below
+    // ambiguous toml structure like the one below
    // [main]
    // field1_under_main = 1
    // field2_under_main = 2
    // [main.subconf]
    // field1_under_subconf = 1
    // field3_under_main = 3 # This field will be interpreted as being under subconf and not under main
-    #[serde(default = "default_path")]
+    #[serde(default = "Config::default_path")]
    pub path: String,

    pub general: General,
    pub pools: HashMap<String, Pool>,
 }

+impl Config {
+    pub fn default_path() -> String {
+        String::from("pgcat.toml")
+    }
+}
+
 impl Default for Config {
    fn default() -> Config {
        Config {
-            path: String::from("pgcat.toml"),
+            path: Self::default_path(),
            general: General::default(),
            pools: HashMap::default(),
        }
@@ -262,7 +435,7 @@ impl From<&Config> for std::collections::HashMap<String, String> {
                [
                    (
                        format!("pools.{}.pool_mode", pool_name),
-                        pool.pool_mode.clone(),
+                        pool.pool_mode.to_string(),
                    ),
                    (
                        format!("pools.{}.primary_reads_enabled", pool_name),
@@ -278,7 +451,7 @@ impl From<&Config> for std::collections::HashMap<String, String> {
                    ),
                    (
                        format!("pools.{}.sharding_function", pool_name),
-                        pool.sharding_function.clone(),
+                        pool.sharding_function.to_string(),
                    ),
                    (
                        format!("pools.{:?}.shard_count", pool_name),
@@ -370,10 +543,22 @@ impl Config {
                    .sum::<u32>()
                    .to_string()
            );
-            info!("[pool: {}] Pool mode: {}", pool_name, pool_config.pool_mode);
+            info!(
+                "[pool: {}] Pool mode: {:?}",
+                pool_name, pool_config.pool_mode
+            );
+            let connect_timeout = match pool_config.connect_timeout {
+                Some(connect_timeout) => connect_timeout,
+                None => self.general.connect_timeout,
+            };
+            info!(
+                "[pool: {}] Connection timeout: {}ms",
+                pool_name, connect_timeout
+            );
            info!(
                "[pool: {}] Sharding function: {}",
-                pool_name, pool_config.sharding_function
+                pool_name,
+                pool_config.sharding_function.to_string()
            );
            info!(
                "[pool: {}] Primary reads: {}",
@@ -406,6 +591,50 @@ impl Config {
            }
        }
    }
+
+    pub fn validate(&mut self) -> Result<(), Error> {
+        // Validate TLS!
+        match self.general.tls_certificate.clone() {
+            Some(tls_certificate) => {
+                match load_certs(&Path::new(&tls_certificate)) {
+                    Ok(_) => {
+                        // Cert is okay, but what about the private key?
+                        match self.general.tls_private_key.clone() {
+                            Some(tls_private_key) => {
+                                match load_keys(&Path::new(&tls_private_key)) {
+                                    Ok(_) => (),
+                                    Err(err) => {
+                                        error!(
+                                            "tls_private_key is incorrectly configured: {:?}",
+                                            err
+                                        );
+                                        return Err(Error::BadConfig);
+                                    }
+                                }
+                            }
+
+                            None => {
+                                error!("tls_certificate is set, but the tls_private_key is not");
+                                return Err(Error::BadConfig);
+                            }
+                        };
+                    }
+
+                    Err(err) => {
+                        error!("tls_certificate is incorrectly configured: {:?}", err);
+                        return Err(Error::BadConfig);
+                    }
+                }
+            }
+            None => (),
+        };
+
+        for (_, pool) in &mut self.pools {
+            pool.validate()?;
+        }
+
+        Ok(())
+    }
 }

 /// Get a read-only instance of the configuration
@@ -442,132 +671,7 @@ pub async fn parse(path: &str) -> Result<(), Error> {
        }
    };

-    // Validate TLS!
-    match config.general.tls_certificate.clone() {
-        Some(tls_certificate) => {
-            match load_certs(&Path::new(&tls_certificate)) {
-                Ok(_) => {
-                    // Cert is okay, but what about the private key?
-                    match config.general.tls_private_key.clone() {
-                        Some(tls_private_key) => match load_keys(&Path::new(&tls_private_key)) {
-                            Ok(_) => (),
-                            Err(err) => {
-                                error!("tls_private_key is incorrectly configured: {:?}", err);
-                                return Err(Error::BadConfig);
-                            }
-                        },
-
-                        None => {
-                            error!("tls_certificate is set, but the tls_private_key is not");
-                            return Err(Error::BadConfig);
-                        }
-                    };
-                }
-
-                Err(err) => {
-                    error!("tls_certificate is incorrectly configured: {:?}", err);
-                    return Err(Error::BadConfig);
-                }
-            }
-        }
-        None => (),
-    };
-
-    for (pool_name, pool) in &config.pools {
-        match pool.sharding_function.as_ref() {
-            "pg_bigint_hash" => (),
-            "sha1" => (),
-            _ => {
-                error!(
-                    "Supported sharding functions are: 'pg_bigint_hash', 'sha1', got: '{}' in pool {} settings",
-                    pool.sharding_function,
-                    pool_name
-                );
-                return Err(Error::BadConfig);
-            }
-        };
-
-        match pool.default_role.as_ref() {
-            "any" => (),
-            "primary" => (),
-            "replica" => (),
-            other => {
-                error!(
-                    "Query router default_role must be 'primary', 'replica', or 'any', got: '{}'",
-                    other
-                );
-                return Err(Error::BadConfig);
-            }
-        };
-
-        match pool.pool_mode.as_ref() {
-            "transaction" => (),
-            "session" => (),
-            other => {
-                error!(
-                    "pool_mode can be 'session' or 'transaction', got: '{}'",
-                    other
-                );
-                return Err(Error::BadConfig);
-            }
-        };
-
-        for shard in &pool.shards {
-            // We use addresses as unique identifiers,
-            // let's make sure they are unique in the config as well.
-            let mut dup_check = HashSet::new();
-            let mut primary_count = 0;
-
-            match shard.0.parse::<usize>() {
-                Ok(_) => (),
-                Err(_) => {
-                    error!(
-                        "Shard '{}' is not a valid number, shards must be numbered starting at 0",
-                        shard.0
-                    );
-                    return Err(Error::BadConfig);
-                }
-            };
-
-            if shard.1.servers.len() == 0 {
-                error!("Shard {} has no servers configured", shard.0);
-                return Err(Error::BadConfig);
-            }
-
-            for server in &shard.1.servers {
-                dup_check.insert(server);
-
-                // Check that we define only zero or one primary.
-                match server.2.as_ref() {
-                    "primary" => primary_count += 1,
-                    _ => (),
-                };
-
-                // Check role spelling.
-                match server.2.as_ref() {
-                    "primary" => (),
-                    "replica" => (),
-                    _ => {
-                        error!(
-                            "Shard {} server role must be either 'primary' or 'replica', got: '{}'",
-                            shard.0, server.2
-                        );
-                        return Err(Error::BadConfig);
-                    }
-                };
-            }
-
-            if primary_count > 1 {
-                error!("Shard {} has more than on primary configured", &shard.0);
-                return Err(Error::BadConfig);
-            }
-
-            if dup_check.len() != shard.1.servers.len() {
-                error!("Shard {} contains duplicate server configs", &shard.0);
-                return Err(Error::BadConfig);
-            }
-        }
-    }
+    config.validate()?;

    config.path = path.to_string();

@@ -589,7 +693,7 @@ pub async fn reload_config(client_server_map: ClientServerMap) -> Result<bool, E
    let new_config = get_config();

    if old_config.pools != new_config.pools {
-        info!("Pool configuration changed, re-creating server pools");
+        info!("Pool configuration changed");
        ConnectionPool::from_config(client_server_map).await?;
        Ok(true)
    } else if old_config != new_config {
@@ -617,12 +721,12 @@ mod test {
        assert_eq!(get_config().pools["simple_db"].users.len(), 1);

        assert_eq!(
-            get_config().pools["sharded_db"].shards["0"].servers[0].0,
+            get_config().pools["sharded_db"].shards["0"].servers[0].host,
            "127.0.0.1"
        );
        assert_eq!(
-            get_config().pools["sharded_db"].shards["1"].servers[0].2,
-            "primary"
+            get_config().pools["sharded_db"].shards["1"].servers[0].role,
+            Role::Primary
        );
        assert_eq!(
            get_config().pools["sharded_db"].shards["1"].database,
@@ -640,11 +744,11 @@ mod test {
        assert_eq!(get_config().pools["sharded_db"].default_role, "any");

        assert_eq!(
-            get_config().pools["simple_db"].shards["0"].servers[0].0,
+            get_config().pools["simple_db"].shards["0"].servers[0].host,
            "127.0.0.1"
        );
        assert_eq!(
-            get_config().pools["simple_db"].shards["0"].servers[0].1,
+            get_config().pools["simple_db"].shards["0"].servers[0].port,
            5432
        );
        assert_eq!(
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -0,0 +1,32 @@
+pub mod config;
+pub mod constants;
+pub mod errors;
+pub mod messages;
+pub mod pool;
+pub mod scram;
+pub mod server;
+pub mod sharding;
+pub mod stats;
+pub mod tls;
+
+/// Format chrono::Duration to be more human-friendly.
+///
+/// # Arguments
+///
+/// * `duration` - A duration of time
+pub fn format_duration(duration: &chrono::Duration) -> String {
+    let milliseconds = format!("{:0>3}", duration.num_milliseconds() % 1000);
+
+    let seconds = format!("{:0>2}", duration.num_seconds() % 60);
+
+    let minutes = format!("{:0>2}", duration.num_minutes() % 60);
+
+    let hours = format!("{:0>2}", duration.num_hours() % 24);
+
+    let days = duration.num_days().to_string();
+
+    format!(
+        "{}d {}:{}:{}.{}",
+        days, hours, minutes, seconds, milliseconds
+    )
+}
--- a/src/main.rs
+++ b/src/main.rs
@@ -37,8 +37,9 @@ extern crate tokio;
 extern crate tokio_rustls;
 extern crate toml;

-use log::{debug, error, info};
+use log::{error, info, warn};
 use parking_lot::Mutex;
+use pgcat::format_duration;
 use tokio::net::TcpListener;
 use tokio::{
    signal::unix::{signal as unix_signal, SignalKind},
@@ -74,7 +75,8 @@ use crate::stats::{Collector, Reporter, REPORTER};

 #[tokio::main(worker_threads = 4)]
 async fn main() {
-    env_logger::init();
+    env_logger::builder().format_timestamp_micros().init();
+
    info!("Welcome to PgCat! Meow. (Version {})", VERSION);

    if !query_router::QueryRouter::setup() {
@@ -154,12 +156,31 @@ async fn main() {

    info!("Config autoreloader: {}", config.general.autoreload);

+    let mut autoreload_interval = tokio::time::interval(tokio::time::Duration::from_millis(15_000));
+    let autoreload_client_server_map = client_server_map.clone();
+    tokio::task::spawn(async move {
+        loop {
+            autoreload_interval.tick().await;
+            if config.general.autoreload {
+                info!("Automatically reloading config");
+
+                match reload_config(autoreload_client_server_map.clone()).await {
+                    Ok(changed) => {
+                        if changed {
+                            get_config().show()
+                        }
+                    }
+                    Err(_) => (),
+                };
+            }
+        }
+    });
+
    let mut term_signal = unix_signal(SignalKind::terminate()).unwrap();
    let mut interrupt_signal = unix_signal(SignalKind::interrupt()).unwrap();
    let mut sighup_signal = unix_signal(SignalKind::hangup()).unwrap();
-    let mut autoreload_interval = tokio::time::interval(tokio::time::Duration::from_millis(15_000));
    let (shutdown_tx, _) = broadcast::channel::<()>(1);
-    let (drain_tx, mut drain_rx) = mpsc::channel::<i8>(2048);
+    let (drain_tx, mut drain_rx) = mpsc::channel::<i32>(2048);
    let (exit_tx, mut exit_rx) = mpsc::channel::<()>(1);

    info!("Waiting for clients");
@@ -182,21 +203,6 @@ async fn main() {
                get_config().show();
            },

-            _ = autoreload_interval.tick() => {
-                if config.general.autoreload {
-                    info!("Automatically reloading config");
-
-                    match reload_config(client_server_map.clone()).await {
-                        Ok(changed) => {
-                            if changed {
-                                get_config().show()
-                            }
-                        }
-                        Err(_) => (),
-                    };
-                }
-            },
-
            // Initiate graceful shutdown sequence on sig int
            _ = interrupt_signal.recv() => {
                info!("Got SIGINT, waiting for client connection drain now");
@@ -217,13 +223,16 @@ async fn main() {
                    interval.tick().await;

                    // We're done waiting.
-                    error!("Timed out waiting for clients");
+                    error!("Graceful shutdown timed out. {} active clients being closed", total_clients);

                    let _ = exit_tx.send(()).await;
                });
            },

-            _ = term_signal.recv() => break,
+            _ = term_signal.recv() => {
+                info!("Got SIGTERM, closing with {} clients active", total_clients);
+                break;
+            },

            new_client = listener.accept() => {
                let (socket, addr) = match new_client {
@@ -270,7 +279,7 @@ async fn main() {
                                }
                            }

-                            debug!("Client disconnected with error {:?}", err);
+                            warn!("Client disconnected with error {:?}", err);
                        }
                    };
                });
@@ -293,41 +302,3 @@ async fn main() {

    info!("Shutting down...");
 }
-
-/// Format chrono::Duration to be more human-friendly.
-///
-/// # Arguments
-///
-/// * `duration` - A duration of time
-fn format_duration(duration: &chrono::Duration) -> String {
-    let seconds = {
-        let seconds = duration.num_seconds() % 60;
-        if seconds < 10 {
-            format!("0{}", seconds)
-        } else {
-            format!("{}", seconds)
-        }
-    };
-
-    let minutes = {
-        let minutes = duration.num_minutes() % 60;
-        if minutes < 10 {
-            format!("0{}", minutes)
-        } else {
-            format!("{}", minutes)
-        }
-    };
-
-    let hours = {
-        let hours = duration.num_hours() % 24;
-        if hours < 10 {
-            format!("0{}", hours)
-        } else {
-            format!("{}", hours)
-        }
-    };
-
-    let days = duration.num_days().to_string();
-
-    format!("{}d {}:{}:{}", days, hours, minutes, seconds)
-}
--- a/src/messages.rs
+++ b/src/messages.rs
@@ -496,7 +496,7 @@ where
    Ok(bytes)
 }

-pub fn server_paramater_message(key: &str, value: &str) -> BytesMut {
+pub fn server_parameter_message(key: &str, value: &str) -> BytesMut {
    let mut server_info = BytesMut::new();

    let null_byte_size = 1;
--- a/src/pool.rs
+++ b/src/pool.rs
@@ -8,39 +8,50 @@ use once_cell::sync::Lazy;
 use parking_lot::{Mutex, RwLock};
 use rand::seq::SliceRandom;
 use rand::thread_rng;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use std::time::Instant;

-use crate::config::{get_config, Address, Role, User};
+use crate::config::{get_config, Address, PoolMode, Role, User};
 use crate::errors::Error;

 use crate::server::Server;
 use crate::sharding::ShardingFunction;
 use crate::stats::{get_reporter, Reporter};

+pub type ProcessId = i32;
+pub type SecretKey = i32;
+pub type ServerHost = String;
+pub type ServerPort = u16;
+
 pub type BanList = Arc<RwLock<Vec<HashMap<Address, NaiveDateTime>>>>;
-pub type ClientServerMap = Arc<Mutex<HashMap<(i32, i32), (i32, i32, String, u16)>>>;
-pub type PoolMap = HashMap<(String, String), ConnectionPool>;
+pub type ClientServerMap =
+    Arc<Mutex<HashMap<(ProcessId, SecretKey), (ProcessId, SecretKey, ServerHost, ServerPort)>>>;
+pub type PoolMap = HashMap<PoolIdentifier, ConnectionPool>;
 /// The connection pool, globally available.
 /// This is atomic and safe and read-optimized.
 /// The pool is recreated dynamically when the config is reloaded.
 pub static POOLS: Lazy<ArcSwap<PoolMap>> = Lazy::new(|| ArcSwap::from_pointee(HashMap::default()));
+static POOLS_HASH: Lazy<ArcSwap<HashSet<crate::config::Pool>>> =
+    Lazy::new(|| ArcSwap::from_pointee(HashSet::default()));

-/// Pool mode:
-/// - transaction: server serves one transaction,
-/// - session: server is attached to the client.
-#[derive(Debug, Clone, Copy, PartialEq)]
-pub enum PoolMode {
-    Session,
-    Transaction,
+/// An identifier for a PgCat pool,
+/// a database visible to clients.
+#[derive(Hash, Debug, Clone, PartialEq, Eq)]
+pub struct PoolIdentifier {
+    // The name of the database clients want to connect to.
+    pub db: String,
+
+    /// The username the client connects with. Each user gets its own pool.
+    pub user: String,
 }

-impl std::fmt::Display for PoolMode {
-    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        match *self {
-            PoolMode::Session => write!(f, "session"),
-            PoolMode::Transaction => write!(f, "transaction"),
+impl PoolIdentifier {
+    /// Create a new user/pool identifier.
+    pub fn new(db: &str, user: &str) -> PoolIdentifier {
+        PoolIdentifier {
+            db: db.to_string(),
+            user: user.to_string(),
        }
    }
 }
@@ -119,9 +130,37 @@ impl ConnectionPool {
        let mut new_pools = HashMap::new();
        let mut address_id = 0;

+        let mut pools_hash = (*(*POOLS_HASH.load())).clone();
+
        for (pool_name, pool_config) in &config.pools {
+            let changed = pools_hash.insert(pool_config.clone());
+
            // There is one pool per database/user pair.
            for (_, user) in &pool_config.users {
+                // If the pool hasn't changed, get existing reference and insert it into the new_pools.
+                // We replace all pools at the end, but if the reference is kept, the pool won't get re-created (bb8).
+                if !changed {
+                    match get_pool(&pool_name, &user.username) {
+                        Some(pool) => {
+                            info!(
+                                "[pool: {}][user: {}] has not changed",
+                                pool_name, user.username
+                            );
+                            new_pools.insert(
+                                PoolIdentifier::new(&pool_name, &user.username),
+                                pool.clone(),
+                            );
+                            continue;
+                        }
+                        None => (),
+                    }
+                }
+
+                info!(
+                    "[pool: {}][user: {}] creating new pool",
+                    pool_name, user.username
+                );
+
                let mut shards = Vec::new();
                let mut addresses = Vec::new();
                let mut banlist = Vec::new();
@@ -143,21 +182,12 @@ impl ConnectionPool {
                    let mut replica_number = 0;

                    for server in shard.servers.iter() {
-                        let role = match server.2.as_ref() {
-                            "primary" => Role::Primary,
-                            "replica" => Role::Replica,
-                            _ => {
-                                error!("Config error: server role can be 'primary' or 'replica', have: '{}'. Defaulting to 'replica'.", server.2);
-                                Role::Replica
-                            }
-                        };
-
                        let address = Address {
                            id: address_id,
                            database: shard.database.clone(),
-                            host: server.0.clone(),
-                            port: server.1 as u16,
-                            role: role,
+                            host: server.host.clone(),
+                            port: server.port,
+                            role: server.role,
                            address_index,
                            replica_number,
                            shard: shard_idx.parse::<usize>().unwrap(),
@@ -168,7 +198,7 @@ impl ConnectionPool {
                        address_id += 1;
                        address_index += 1;

-                        if role == Role::Replica {
+                        if server.role == Role::Replica {
                            replica_number += 1;
                        }

@@ -180,11 +210,14 @@ impl ConnectionPool {
                            get_reporter(),
                        );

+                        let connect_timeout = match pool_config.connect_timeout {
+                            Some(connect_timeout) => connect_timeout,
+                            None => config.general.connect_timeout,
+                        };
+
                        let pool = Pool::builder()
                            .max_size(user.pool_size)
-                            .connection_timeout(std::time::Duration::from_millis(
-                                config.general.connect_timeout,
-                            ))
+                            .connection_timeout(std::time::Duration::from_millis(connect_timeout))
                            .test_on_check_out(false)
                            .build(manager)
                            .await
@@ -208,11 +241,7 @@ impl ConnectionPool {
                    stats: get_reporter(),
                    server_info: BytesMut::new(),
                    settings: PoolSettings {
-                        pool_mode: match pool_config.pool_mode.as_str() {
-                            "transaction" => PoolMode::Transaction,
-                            "session" => PoolMode::Session,
-                            _ => unreachable!(),
-                        },
+                        pool_mode: pool_config.pool_mode,
                        // shards: pool_config.shards.clone(),
                        shards: shard_ids.len(),
                        user: user.clone(),
@@ -224,11 +253,7 @@ impl ConnectionPool {
                        },
                        query_parser_enabled: pool_config.query_parser_enabled.clone(),
                        primary_reads_enabled: pool_config.primary_reads_enabled,
-                        sharding_function: match pool_config.sharding_function.as_str() {
-                            "pg_bigint_hash" => ShardingFunction::PgBigintHash,
-                            "sha1" => ShardingFunction::Sha1,
-                            _ => unreachable!(),
-                        },
+                        sharding_function: pool_config.sharding_function,
                    },
                };

@@ -243,11 +268,12 @@ impl ConnectionPool {
                };

                // There is one pool per database/user pair.
-                new_pools.insert((pool_name.clone(), user.username.clone()), pool);
+                new_pools.insert(PoolIdentifier::new(&pool_name, &user.username), pool);
            }
        }

        POOLS.store(Arc::new(new_pools.clone()));
+        POOLS_HASH.store(Arc::new(pools_hash.clone()));

        Ok(())
    }
@@ -307,7 +333,6 @@ impl ConnectionPool {
        role: Option<Role>, // primary or replica
        process_id: i32,    // client id
    ) -> Result<(PooledConnection<'_, ServerPool>, Address), Error> {
-        let now = Instant::now();
        let mut candidates: Vec<&Address> = self.addresses[shard]
            .iter()
            .filter(|address| address.role == role)
@@ -332,7 +357,8 @@ impl ConnectionPool {
            }

            // Indicate we're waiting on a server connection from a pool.
-            self.stats.client_waiting(process_id, address.id);
+            let now = Instant::now();
+            self.stats.client_waiting(process_id);

            // Check if we can connect
            let mut conn = match self.databases[address.shard][address.address_index]
@@ -343,8 +369,7 @@ impl ConnectionPool {
                Err(err) => {
                    error!("Banning instance {:?}, error: {:?}", address, err);
                    self.ban(&address, process_id);
-                    self.stats
-                        .checkout_time(now.elapsed().as_micros(), process_id, address.id);
+                    self.stats.client_checkout_error(process_id, address.id);
                    continue;
                }
            };
@@ -361,27 +386,30 @@ impl ConnectionPool {
            // Health checks are pretty expensive.
            if !require_healthcheck {
                self.stats
-                    .checkout_time(now.elapsed().as_micros(), process_id, address.id);
-                self.stats.server_active(conn.process_id(), address.id);
+                    .checkout_time(now.elapsed().as_micros(), process_id, server.server_id());
+                self.stats.server_active(process_id, server.server_id());
                return Ok((conn, address.clone()));
            }

            debug!("Running health check on server {:?}", address);

-            self.stats.server_tested(server.process_id(), address.id);
+            self.stats.server_tested(server.server_id());

            match tokio::time::timeout(
                tokio::time::Duration::from_millis(healthcheck_timeout),
-                server.query(";"), // Cheap query (query parser not used in PG)
+                server.query(";"), // Cheap query as it skips the query planner
            )
            .await
            {
                // Check if health check succeeded.
                Ok(res) => match res {
                    Ok(_) => {
-                        self.stats
-                            .checkout_time(now.elapsed().as_micros(), process_id, address.id);
-                        self.stats.server_active(conn.process_id(), address.id);
+                        self.stats.checkout_time(
+                            now.elapsed().as_micros(),
+                            process_id,
+                            conn.server_id(),
+                        );
+                        self.stats.server_active(process_id, conn.server_id());
                        return Ok((conn, address.clone()));
                    }

@@ -421,10 +449,9 @@ impl ConnectionPool {
    /// Ban an address (i.e. replica). It no longer will serve
    /// traffic for any new transactions. Existing transactions on that replica
    /// will finish successfully or error out to the clients.
-    pub fn ban(&self, address: &Address, process_id: i32) {
-        self.stats.client_disconnecting(process_id, address.id);
-
+    pub fn ban(&self, address: &Address, client_id: i32) {
        error!("Banning {:?}", address);
+        self.stats.client_ban_error(client_id, address.id);

        let now = chrono::offset::Utc::now().naive_utc();
        let mut guard = self.banlist.write();
@@ -559,19 +586,21 @@ impl ManageConnection for ServerPool {

    /// Attempts to create a new connection.
    async fn connect(&self) -> Result<Self::Connection, Self::Error> {
-        info!(
-            "Creating a new connection to {:?} using user {:?}",
-            self.address.name(),
-            self.user.username
-        );
+        info!("Creating a new server connection {:?}", self.address);
+        let server_id = rand::random::<i32>();

-        // Put a temporary process_id into the stats
-        // for server login.
-        let process_id = rand::random::<i32>();
-        self.stats.server_login(process_id, self.address.id);
+        self.stats.server_register(
+            server_id,
+            self.address.id,
+            self.address.name(),
+            self.address.pool_name.clone(),
+            self.address.username.clone(),
+        );
+        self.stats.server_login(server_id);

        // Connect to the PostgreSQL server.
        match Server::startup(
+            server_id,
            &self.address,
            &self.user,
            &self.database,
@@ -581,20 +610,18 @@ impl ManageConnection for ServerPool {
        .await
        {
            Ok(conn) => {
-                // Remove the temporary process_id from the stats.
-                self.stats.server_disconnecting(process_id, self.address.id);
+                self.stats.server_idle(server_id);
                Ok(conn)
            }
            Err(err) => {
-                // Remove the temporary process_id from the stats.
-                self.stats.server_disconnecting(process_id, self.address.id);
+                self.stats.server_disconnecting(server_id);
                Err(err)
            }
        }
    }

    /// Determines if the connection is still connected to the database.
-    async fn is_valid(&self, _conn: &mut PooledConnection<'_, Self>) -> Result<(), Self::Error> {
+    async fn is_valid(&self, _conn: &mut Self::Connection) -> Result<(), Self::Error> {
        Ok(())
    }

@@ -605,13 +632,18 @@ impl ManageConnection for ServerPool {
 }

 /// Get the connection pool
-pub fn get_pool(db: String, user: String) -> Option<ConnectionPool> {
-    match get_all_pools().get(&(db, user)) {
+pub fn get_pool(db: &str, user: &str) -> Option<ConnectionPool> {
+    match get_all_pools().get(&PoolIdentifier::new(&db, &user)) {
        Some(pool) => Some(pool.clone()),
        None => None,
    }
 }

+/// Get a pointer to all configured pools.
+pub fn get_all_pools() -> HashMap<PoolIdentifier, ConnectionPool> {
+    return (*(*POOLS.load())).clone();
+}
+
 /// How many total servers we have in the config.
 pub fn get_number_of_addresses() -> usize {
    get_all_pools()
@@ -619,8 +651,3 @@ pub fn get_number_of_addresses() -> usize {
        .map(|(_, pool)| pool.databases())
        .sum()
 }
-
-/// Get a pointer to all configured pools.
-pub fn get_all_pools() -> HashMap<(String, String), ConnectionPool> {
-    return (*(*POOLS.load())).clone();
-}
--- a/src/prometheus.rs
+++ b/src/prometheus.rs
@@ -8,7 +8,7 @@ use std::net::SocketAddr;

 use crate::config::Address;
 use crate::pool::get_all_pools;
-use crate::stats::get_stats;
+use crate::stats::get_address_stats;

 struct MetricHelpType {
    help: &'static str,
@@ -164,7 +164,7 @@ impl PrometheusMetric {
 async fn prometheus_stats(request: Request<Body>) -> Result<Response<Body>, hyper::http::Error> {
    match (request.method(), request.uri().path()) {
        (&Method::GET, "/metrics") => {
-            let stats = get_stats();
+            let stats: HashMap<usize, HashMap<String, i64>> = get_address_stats();

            let mut lines = Vec::new();
            for (_, pool) in get_all_pools() {
--- a/src/query_router.rs
+++ b/src/query_router.rs
@@ -169,8 +169,8 @@ impl QueryRouter {

            Command::ShowShard => self.shard().to_string(),
            Command::ShowServerRole => match self.active_role {
-                Some(Role::Primary) => String::from("primary"),
-                Some(Role::Replica) => String::from("replica"),
+                Some(Role::Primary) => Role::Primary.to_string(),
+                Some(Role::Replica) => Role::Replica.to_string(),
                None => {
                    if self.query_parser_enabled {
                        String::from("auto")
@@ -359,8 +359,8 @@ impl QueryRouter {
 #[cfg(test)]
 mod test {
    use super::*;
+    use crate::config::PoolMode;
    use crate::messages::simple_query;
-    use crate::pool::PoolMode;
    use crate::sharding::ShardingFunction;
    use bytes::BufMut;

--- a/src/server.rs
+++ b/src/server.rs
@@ -1,7 +1,8 @@
 /// Implementation of the PostgreSQL server (database) protocol.
 /// Here we are pretending to the a Postgres client.
 use bytes::{Buf, BufMut, BytesMut};
-use log::{debug, error, info, trace};
+use log::{debug, error, info, trace, warn};
+use std::io::Read;
 use std::time::SystemTime;
 use tokio::io::{AsyncReadExt, BufReader};
 use tokio::net::{
@@ -13,12 +14,14 @@ use crate::config::{Address, User};
 use crate::constants::*;
 use crate::errors::Error;
 use crate::messages::*;
+use crate::pool::ClientServerMap;
 use crate::scram::ScramSha256;
 use crate::stats::Reporter;
-use crate::ClientServerMap;

 /// Server state.
 pub struct Server {
+    server_id: i32,
+
    /// Server host, e.g. localhost,
    /// port, e.g. 5432, and role, e.g. primary or replica.
    address: Address,
@@ -48,6 +51,9 @@ pub struct Server {
    /// Is the server broken? We'll remote it from the pool if so.
    bad: bool,

+    /// If server connection requires a DISCARD ALL before checkin
+    needs_cleanup: bool,
+
    /// Mapping of clients and servers used for query cancellation.
    client_server_map: ClientServerMap,

@@ -68,6 +74,7 @@ impl Server {
    /// Pretend to be the Postgres client and connect to the server given host, port and credentials.
    /// Perform the authentication and return the server in a ready for query state.
    pub async fn startup(
+        server_id: i32,
        address: &Address,
        user: &User,
        database: &str,
@@ -311,11 +318,13 @@ impl Server {
                        write: write,
                        buffer: BytesMut::with_capacity(8196),
                        server_info: server_info,
+                        server_id: server_id,
                        process_id: process_id,
                        secret_key: secret_key,
                        in_transaction: false,
                        data_available: false,
                        bad: false,
+                        needs_cleanup: false,
                        client_server_map: client_server_map,
                        connected_at: chrono::offset::Utc::now().naive_utc(),
                        stats: stats,
@@ -367,8 +376,7 @@ impl Server {

    /// Send messages to the server from the client.
    pub async fn send(&mut self, messages: BytesMut) -> Result<(), Error> {
-        self.stats
-            .data_sent(messages.len(), self.process_id, self.address.id);
+        self.stats.data_sent(messages.len(), self.server_id);

        match write_all_half(&mut self.write, messages).await {
            Ok(_) => {
@@ -440,6 +448,29 @@ impl Server {
                    break;
                }

+                // CommandComplete
+                'C' => {
+                    let mut command_tag = String::new();
+                    match message.reader().read_to_string(&mut command_tag) {
+                        Ok(_) => {
+                            // Non-exhaustive list of commands that are likely to change session variables/resources
+                            // which can leak between clients. This is a best effort to block bad clients
+                            // from poisoning a transaction-mode pool by setting inappropriate session variables
+                            match command_tag.as_str() {
+                                "SET\0" | "PREPARE\0" => {
+                                    debug!("Server connection marked for clean up");
+                                    self.needs_cleanup = true;
+                                }
+                                _ => (),
+                            }
+                        }
+
+                        Err(err) => {
+                            warn!("Encountered an error while parsing CommandTag {}", err);
+                        }
+                    }
+                }
+
                // DataRow
                'D' => {
                    // More data is available after this message, this is not the end of the reply.
@@ -477,8 +508,7 @@ impl Server {
        let bytes = self.buffer.clone();

        // Keep track of how much data we got from the server for stats.
-        self.stats
-            .data_received(bytes.len(), self.process_id, self.address.id);
+        self.stats.data_received(bytes.len(), self.server_id);

        // Clear the buffer for next query.
        self.buffer.clear();
@@ -553,14 +583,45 @@ impl Server {
        Ok(())
    }

+    /// Perform any necessary cleanup before putting the server
+    /// connection back in the pool
+    pub async fn checkin_cleanup(&mut self) -> Result<(), Error> {
+        // Client disconnected with an open transaction on the server connection.
+        // Pgbouncer behavior is to close the server connection but that can cause
+        // server connection thrashing if clients repeatedly do this.
+        // Instead, we ROLLBACK that transaction before putting the connection back in the pool
+        if self.in_transaction() {
+            warn!("Server returned while still in transaction, rolling back transaction");
+            self.query("ROLLBACK").await?;
+        }
+
+        // Client disconnected but it perfromed session-altering operations such as
+        // SET statement_timeout to 1 or create a prepared statement. We clear that
+        // to avoid leaking state between clients. For performance reasons we only
+        // send `DISCARD ALL` if we think the session is altered instead of just sending
+        // it before each checkin.
+        if self.needs_cleanup {
+            warn!("Server returned with session state altered, discarding state");
+            self.query("DISCARD ALL").await?;
+            self.needs_cleanup = false;
+        }
+
+        return Ok(());
+    }
+
    /// A shorthand for `SET application_name = $1`.
-    #[allow(dead_code)]
    pub async fn set_name(&mut self, name: &str) -> Result<(), Error> {
        if self.application_name != name {
            self.application_name = name.to_string();
-            Ok(self
+            // We don't want `SET application_name` to mark the server connection
+            // as needing cleanup
+            let needs_cleanup_before = self.needs_cleanup;
+
+            let result = Ok(self
                .query(&format!("SET application_name = '{}'", name))
-                .await?)
+                .await?);
+            self.needs_cleanup = needs_cleanup_before;
+            return result;
        } else {
            Ok(())
        }
@@ -572,15 +633,21 @@ impl Server {
        self.address.clone()
    }

-    /// Get the server's unique identifier.
-    pub fn process_id(&self) -> i32 {
-        self.process_id
+    /// Get the server connection identifier
+    /// Used to uniquely identify connection in statistics
+    pub fn server_id(&self) -> i32 {
+        self.server_id
    }

    // Get server's latest response timestamp
    pub fn last_activity(&self) -> SystemTime {
        self.last_activity
    }
+
+    // Marks a connection as needing DISCARD ALL at checkin
+    pub fn mark_dirty(&mut self) {
+        self.needs_cleanup = true;
+    }
 }

 impl Drop for Server {
@@ -588,8 +655,7 @@ impl Drop for Server {
    /// the socket is in non-blocking mode, so it may not be ready
    /// for a write.
    fn drop(&mut self) {
-        self.stats
-            .server_disconnecting(self.process_id(), self.address.id);
+        self.stats.server_disconnecting(self.server_id);

        let mut bytes = BytesMut::with_capacity(4);
        bytes.put_u8(b'X');
@@ -607,7 +673,8 @@ impl Drop for Server {
        let duration = now - self.connected_at;

        info!(
-            "Server connection closed, session duration: {}",
+            "Server connection closed {:?}, session duration: {}",
+            self.address,
            crate::format_duration(&duration)
        );
    }
--- a/src/sharding.rs
+++ b/src/sharding.rs
@@ -1,3 +1,4 @@
+use serde_derive::{Deserialize, Serialize};
 /// Implements various sharding functions.
 use sha1::{Digest, Sha1};

@@ -5,12 +6,23 @@ use sha1::{Digest, Sha1};
 const PARTITION_HASH_SEED: u64 = 0x7A5B22367996DCFD;

 /// The sharding functions we support.
-#[derive(Debug, PartialEq, Copy, Clone)]
+#[derive(Debug, PartialEq, Copy, Clone, Serialize, Deserialize, Hash, std::cmp::Eq)]
 pub enum ShardingFunction {
+    #[serde(alias = "pg_bigint_hash", alias = "PgBigintHash")]
    PgBigintHash,
+    #[serde(alias = "sha1", alias = "Sha1")]
    Sha1,
 }

+impl ToString for ShardingFunction {
+    fn to_string(&self) -> String {
+        match *self {
+            ShardingFunction::PgBigintHash => "pg_bigint_hash".to_string(),
+            ShardingFunction::Sha1 => "sha1".to_string(),
+        }
+    }
+}
+
 /// The sharder.
 pub struct Sharder {
    /// Number of shards in the cluster.
--- a/src/stats.rs
+++ b/src/stats.rs
--- a/tests/docker/Dockerfile
+++ b/tests/docker/Dockerfile
@@ -0,0 +1,5 @@
+FROM rust:bullseye
+
+RUN apt-get update && apt-get install llvm-11 psmisc postgresql-contrib postgresql-client ruby ruby-dev libpq-dev python3 python3-pip lcov sudo curl -y
+RUN cargo install cargo-binutils rustfilt
+RUN rustup component add llvm-tools-preview
--- a/tests/docker/docker-compose.yml
+++ b/tests/docker/docker-compose.yml
@@ -0,0 +1,47 @@
+version: "3"
+services:
+  pg1:
+    image: postgres:14
+    network_mode: "service:main"
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_DB: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_HOST_AUTH_METHOD: scram-sha-256
+    command: ["postgres", "-c", "shared_preload_libraries=pg_stat_statements", "-c", "pg_stat_statements.track=all", "-p", "5432"]
+  pg2:
+    image: postgres:14
+    network_mode: "service:main"
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_DB: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_HOST_AUTH_METHOD: scram-sha-256
+    command: ["postgres", "-c", "shared_preload_libraries=pg_stat_statements", "-c", "pg_stat_statements.track=all", "-p", "7432"]
+  pg3:
+    image: postgres:14
+    network_mode: "service:main"
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_DB: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_HOST_AUTH_METHOD: scram-sha-256
+    command: ["postgres", "-c", "shared_preload_libraries=pg_stat_statements", "-c", "pg_stat_statements.track=all", "-p", "8432"]
+  pg4:
+    image: postgres:14
+    network_mode: "service:main"
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_DB: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_HOST_AUTH_METHOD: scram-sha-256
+    command: ["postgres", "-c", "shared_preload_libraries=pg_stat_statements", "-c", "pg_stat_statements.track=all", "-p", "9432"]
+  main:
+    build: .
+    command: ["bash", "/app/tests/docker/run.sh"]
+    environment:
+      RUSTFLAGS: "-C instrument-coverage"
+      LLVM_PROFILE_FILE: "pgcat-%m.profraw"
+    volumes:
+      - ../../:/app/
+      - /app/target/
--- a/tests/docker/run.sh
+++ b/tests/docker/run.sh
@@ -0,0 +1,21 @@
+#!/bin/bash
+
+rm /app/*.profraw || true
+rm /app/pgcat.profdata || true
+rm -rf /app/cov || true
+
+cd /app/
+
+cargo build
+cargo test --tests
+
+bash .circleci/run_tests.sh
+
+rust-profdata merge -sparse pgcat-*.profraw -o pgcat.profdata
+
+rust-cov export -ignore-filename-regex="rustc|registry" -Xdemangler=rustfilt -instr-profile=pgcat.profdata --object ./target/debug/pgcat --format lcov > ./lcov.info
+
+genhtml lcov.info --output-directory cov --prefix $(pwd)
+
+rm /app/*.profraw
+rm /app/pgcat.profdata
--- a/tests/python/tests.py
+++ b/tests/python/tests.py
@@ -18,9 +18,14 @@ def pgcat_start():


 def pg_cat_send_signal(signal: signal.Signals):
-    for proc in psutil.process_iter(["pid", "name"]):
-        if "pgcat" == proc.name():
-            os.kill(proc.pid, signal)
+    try:
+        for proc in psutil.process_iter(["pid", "name"]):
+            if "pgcat" == proc.name():
+                os.kill(proc.pid, signal)
+    except Exception as e:
+        # The process can be gone when we send this signal
+        print(e)
+
    if signal == signal.SIGTERM:
        # Returns 0 if pgcat process exists
        time.sleep(2)
--- a/tests/ruby/Gemfile
+++ b/tests/ruby/Gemfile
@@ -1,6 +1,8 @@
 source "https://rubygems.org"

 gem "pg"
-gem "activerecord"
+gem "toml"
+gem "rspec"
 gem "rubocop"
-gem "toml", "~> 0.3.0"
+gem "toxiproxy"
+gem "activerecord"
--- a/tests/ruby/Gemfile.lock
+++ b/tests/ruby/Gemfile.lock
@@ -13,6 +13,7 @@ GEM
      tzinfo (~> 2.0)
    ast (2.4.2)
    concurrent-ruby (1.1.10)
+    diff-lcs (1.5.0)
    i18n (1.11.0)
      concurrent-ruby (~> 1.0)
    minitest (5.16.2)
@@ -24,6 +25,19 @@ GEM
    rainbow (3.1.1)
    regexp_parser (2.3.1)
    rexml (3.2.5)
+    rspec (3.11.0)
+      rspec-core (~> 3.11.0)
+      rspec-expectations (~> 3.11.0)
+      rspec-mocks (~> 3.11.0)
+    rspec-core (3.11.0)
+      rspec-support (~> 3.11.0)
+    rspec-expectations (3.11.0)
+      diff-lcs (>= 1.2.0, < 2.0)
+      rspec-support (~> 3.11.0)
+    rspec-mocks (3.11.1)
+      diff-lcs (>= 1.2.0, < 2.0)
+      rspec-support (~> 3.11.0)
+    rspec-support (3.11.0)
    rubocop (1.29.0)
      parallel (~> 1.10)
      parser (>= 3.1.0.0)
@@ -38,19 +52,23 @@ GEM
    ruby-progressbar (1.11.0)
    toml (0.3.0)
      parslet (>= 1.8.0, < 3.0.0)
+    toxiproxy (2.0.1)
    tzinfo (2.0.4)
      concurrent-ruby (~> 1.0)
    unicode-display_width (2.1.0)

 PLATFORMS
+  aarch64-linux
  arm64-darwin-21
  x86_64-linux

 DEPENDENCIES
  activerecord
  pg
+  rspec
  rubocop
-  toml (~> 0.3.0)
+  toml
+  toxiproxy

 BUNDLED WITH
-   2.3.7
+   2.3.21
--- a/tests/ruby/admin_spec.rb
+++ b/tests/ruby/admin_spec.rb
@@ -0,0 +1,289 @@
+# frozen_string_literal: true
+require 'uri'
+require_relative 'spec_helper'
+
+describe "Admin" do
+  let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 10) }
+  let(:pgcat_conn_str) { processes.pgcat.connection_string("sharded_db", "sharding_user") }
+
+  after do
+    processes.all_databases.map(&:reset)
+    processes.pgcat.shutdown
+  end
+
+  describe "SHOW STATS" do
+    context "clients connect and make one query" do
+      it "updates *_query_time and *_wait_time" do
+        connection = PG::connect("#{pgcat_conn_str}?application_name=one_query")
+        connection.async_exec("SELECT pg_sleep(0.25)")
+        connection.async_exec("SELECT pg_sleep(0.25)")
+        connection.async_exec("SELECT pg_sleep(0.25)")
+        connection.close
+
+        # wait for averages to be calculated, we shouldn't do this too often
+        sleep(15.5)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW STATS")[0]
+        admin_conn.close
+        expect(results["total_query_time"].to_i).to be_within(200).of(750)
+        expect(results["avg_query_time"].to_i).to_not eq(0)
+
+        expect(results["total_wait_time"].to_i).to_not eq(0)
+        expect(results["avg_wait_time"].to_i).to_not eq(0)
+      end
+    end
+  end
+
+  describe "SHOW POOLS" do
+    context "bad credentials" do
+      it "does not change any stats" do
+        bad_passsword_url = URI(pgcat_conn_str)
+        bad_passsword_url.password = "wrong"
+        expect { PG::connect("#{bad_passsword_url.to_s}?application_name=bad_password") }.to raise_error(PG::ConnectionBad)
+
+        sleep(1)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+
+        expect(results["sv_idle"]).to eq("1")
+      end
+    end
+
+    context "bad database name" do
+      it "does not change any stats" do
+        bad_db_url = URI(pgcat_conn_str)
+        bad_db_url.path = "/wrong_db"
+        expect { PG::connect("#{bad_db_url.to_s}?application_name=bad_db") }.to raise_error(PG::ConnectionBad)
+
+        sleep(1)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+
+        expect(results["sv_idle"]).to eq("1")
+      end
+    end
+
+    context "client connects but issues no queries" do
+      it "only affects cl_idle stats" do
+        connections = Array.new(20) { PG::connect(pgcat_conn_str) }
+        sleep(1)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_idle"]).to eq("20")
+        expect(results["sv_idle"]).to eq("1")
+
+        connections.map(&:close)
+        sleep(1.1)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_idle cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["sv_idle"]).to eq("1")
+      end
+    end
+
+    context "clients connect and make one query" do
+      it "only affects cl_idle, sv_idle stats" do
+        connections = Array.new(5) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          Thread.new { c.async_exec("SELECT pg_sleep(2.5)") }
+        end
+
+        sleep(1.1)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_waiting cl_cancel_req sv_idle sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_active"]).to eq("5")
+        expect(results["sv_active"]).to eq("5")
+
+        sleep(3)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_idle"]).to eq("5")
+        expect(results["sv_idle"]).to eq("5")
+
+        connections.map(&:close)
+        sleep(1)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["sv_idle"]).to eq("5")
+      end
+    end
+
+    context "client connects and opens a transaction and closes connection uncleanly" do
+      it "produces correct statistics" do
+        connections = Array.new(5) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          Thread.new do
+            c.async_exec("BEGIN")
+            c.async_exec("SELECT pg_sleep(0.01)")
+            c.close
+          end
+        end
+
+        sleep(1.1)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["sv_idle"]).to eq("5")
+      end
+    end
+
+    context "client fail to checkout connection from the pool" do
+      it "counts clients as idle" do
+        new_configs = processes.pgcat.current_config
+        new_configs["general"]["connect_timeout"] = 500
+        new_configs["general"]["ban_time"] = 1
+        new_configs["general"]["shutdown_timeout"] = 1
+        new_configs["pools"]["sharded_db"]["users"]["0"]["pool_size"] = 1
+        processes.pgcat.update_config(new_configs)
+        processes.pgcat.reload_config
+
+        threads = []
+        connections = Array.new(5) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          threads << Thread.new { c.async_exec("SELECT pg_sleep(1)") rescue PG::SystemError }
+        end
+
+        sleep(2)
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_idle"]).to eq("5")
+        expect(results["sv_idle"]).to eq("1")
+
+        threads.map(&:join)
+        connections.map(&:close)
+      end
+    end
+
+    context "clients overwhelm server pools" do
+      let(:processes) { Helpers::Pgcat.single_instance_setup("sharded_db", 2) }
+
+      it "cl_waiting is updated to show it" do
+        threads = []
+        connections = Array.new(4) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          threads << Thread.new { c.async_exec("SELECT pg_sleep(1.5)") }
+        end
+
+        sleep(1.1) # Allow time for stats to update
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_idle cl_cancel_req sv_idle sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+
+        expect(results["cl_waiting"]).to eq("2")
+        expect(results["cl_active"]).to eq("2")
+        expect(results["sv_active"]).to eq("2")
+
+        sleep(2.5) # Allow time for stats to update
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        %w[cl_active cl_waiting cl_cancel_req sv_active sv_used sv_tested sv_login maxwait].each do |s|
+          raise StandardError, "Field #{s} was expected to be 0 but found to be #{results[s]}" if results[s] != "0"
+        end
+        expect(results["cl_idle"]).to eq("4")
+        expect(results["sv_idle"]).to eq("2")
+
+        threads.map(&:join)
+        connections.map(&:close)
+      end
+
+      it "show correct max_wait" do
+        threads = []
+        connections = Array.new(4) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          threads << Thread.new { c.async_exec("SELECT pg_sleep(1.5)") }
+        end
+
+        sleep(2.5) # Allow time for stats to update
+        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+
+        expect(results["maxwait"]).to eq("1")
+        expect(results["maxwait_us"].to_i).to be_within(100_000).of(500_000)
+
+        sleep(4.5) # Allow time for stats to update
+        results = admin_conn.async_exec("SHOW POOLS")[0]
+        expect(results["maxwait"]).to eq("0")
+
+        threads.map(&:join)
+        connections.map(&:close)
+      end
+    end
+  end
+
+  describe "SHOW CLIENTS" do
+    it "reports correct number and application names" do
+      conn_str = processes.pgcat.connection_string("sharded_db", "sharding_user")
+      connections = Array.new(20) { |i| PG::connect("#{conn_str}?application_name=app#{i % 5}") }
+
+      admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+      sleep(1) # Wait for stats to be updated
+
+      results = admin_conn.async_exec("SHOW CLIENTS")
+      expect(results.count).to eq(21) # count admin clients
+      expect(results.select { |c| c["application_name"] == "app3" ||  c["application_name"] == "app4" }.count).to eq(8)
+      expect(results.select { |c| c["database"] == "pgcat" }.count).to eq(1)
+
+      connections[0..5].map(&:close)
+      sleep(1) # Wait for stats to be updated
+      results = admin_conn.async_exec("SHOW CLIENTS")
+      expect(results.count).to eq(15)
+
+      connections[6..].map(&:close)
+      sleep(1) # Wait for stats to be updated
+      expect(admin_conn.async_exec("SHOW CLIENTS").count).to eq(1)
+      admin_conn.close
+    end
+
+    it "reports correct number of queries and transactions" do
+      conn_str = processes.pgcat.connection_string("sharded_db", "sharding_user")
+
+      connections = Array.new(2) { |i| PG::connect("#{conn_str}?application_name=app#{i}") }
+      connections.each do |c|
+        c.async_exec("SELECT 1")
+        c.async_exec("SELECT 2")
+        c.async_exec("SELECT 3")
+        c.async_exec("BEGIN")
+        c.async_exec("SELECT 4")
+        c.async_exec("SELECT 5")
+        c.async_exec("COMMIT")
+      end
+
+      admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+      sleep(1) # Wait for stats to be updated
+
+      results = admin_conn.async_exec("SHOW CLIENTS")
+      expect(results.count).to eq(3)
+      normal_client_results = results.reject { |r| r["database"] == "pgcat" }
+      expect(normal_client_results[0]["transaction_count"]).to eq("4")
+      expect(normal_client_results[1]["transaction_count"]).to eq("4")
+      expect(normal_client_results[0]["query_count"]).to eq("7")
+      expect(normal_client_results[1]["query_count"]).to eq("7")
+
+      admin_conn.close
+      connections.map(&:close)
+    end
+  end
+end
--- a/tests/ruby/helpers/pg_instance.rb
+++ b/tests/ruby/helpers/pg_instance.rb
@@ -0,0 +1,82 @@
+require 'pg'
+require 'toxiproxy'
+
+class PgInstance
+  attr_reader :port
+  attr_reader :username
+  attr_reader :password
+  attr_reader :database_name
+
+  def initialize(port, username, password, database_name)
+    @original_port = port
+    @toxiproxy_port = 10000 + port.to_i
+    @port = @toxiproxy_port
+
+    @username = username
+    @password = password
+    @database_name = database_name
+    @toxiproxy_name = "database_#{@original_port}"
+    Toxiproxy.populate([{
+      name: @toxiproxy_name,
+      listen: "0.0.0.0:#{@toxiproxy_port}",
+      upstream: "localhost:#{@original_port}",
+    }])
+
+    # Toxiproxy server will outlive our PgInstance objects
+    # so we want to destroy our proxies before exiting
+    # Ruby finalizer is ideal for doing this
+    ObjectSpace.define_finalizer(@toxiproxy_name, proc { Toxiproxy[@toxiproxy_name].destroy })
+  end
+
+  def with_connection
+    conn = PG.connect("postgres://#{@username}:#{@password}@localhost:#{port}/#{database_name}")
+    yield conn
+  ensure
+    conn&.close
+  end
+
+  def reset
+    reset_toxics
+    reset_stats
+  end
+
+  def toxiproxy
+    Toxiproxy[@toxiproxy_name]
+  end
+
+  def take_down
+    if block_given?
+      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 5).apply { yield }
+    else
+      Toxiproxy[@toxiproxy_name].toxic(:limit_data, bytes: 5).toxics.each(&:save)
+    end
+  end
+
+  def add_latency(latency)
+    if block_given?
+      Toxiproxy[@toxiproxy_name].toxic(:latency, latency: latency).apply { yield }
+    else
+      Toxiproxy[@toxiproxy_name].toxic(:latency, latency: latency).toxics.each(&:save)
+    end
+  end
+
+  def delete_proxy
+    Toxiproxy[@toxiproxy_name].delete
+  end
+
+  def reset_toxics
+    Toxiproxy[@toxiproxy_name].toxics.each(&:destroy)
+  end
+
+  def reset_stats
+    with_connection { |c| c.async_exec("SELECT pg_stat_statements_reset()") }
+  end
+
+  def count_query(query)
+    with_connection { |c| c.async_exec("SELECT SUM(calls) FROM pg_stat_statements WHERE query = '#{query}'")[0]["sum"].to_i }
+  end
+
+  def count_select_1_plus_2
+    with_connection { |c| c.async_exec("SELECT SUM(calls) FROM pg_stat_statements WHERE query = 'SELECT $1 + $2'")[0]["sum"].to_i }
+  end
+end
--- a/tests/ruby/helpers/pgcat_helper.rb
+++ b/tests/ruby/helpers/pgcat_helper.rb
@@ -0,0 +1,144 @@
+require 'json'
+require 'ostruct'
+require_relative 'pgcat_process'
+require_relative 'pg_instance'
+
+module Helpers
+  module Pgcat
+    def self.three_shard_setup(pool_name, pool_size, pool_mode="transaction")
+      user = {
+        "password" => "sharding_user",
+        "pool_size" => pool_size,
+        "statement_timeout" => 0,
+        "username" => "sharding_user"
+      }
+
+      pgcat    = PgcatProcess.new("info")
+      primary0 = PgInstance.new(5432, user["username"], user["password"], "shard0")
+      primary1 = PgInstance.new(7432, user["username"], user["password"], "shard1")
+      primary2 = PgInstance.new(8432, user["username"], user["password"], "shard2")
+
+      pgcat_cfg = pgcat.current_config
+      pgcat_cfg["pools"] = {
+        "#{pool_name}" => {
+          "default_role" => "any",
+          "pool_mode" => pool_mode,
+          "primary_reads_enabled" => false,
+          "query_parser_enabled" => false,
+          "sharding_function" => "pg_bigint_hash",
+          "shards" => {
+            "0" => { "database" => "shard0", "servers" => [["localhost", primary0.port.to_s, "primary"]] },
+            "1" => { "database" => "shard1", "servers" => [["localhost", primary1.port.to_s, "primary"]] },
+            "2" => { "database" => "shard2", "servers" => [["localhost", primary2.port.to_s, "primary"]] },
+          },
+          "users" => { "0" => user }
+        }
+      }
+      pgcat.update_config(pgcat_cfg)
+
+      pgcat.start
+      pgcat.wait_until_ready
+
+      OpenStruct.new.tap do |struct|
+        struct.pgcat = pgcat
+        struct.shards = [primary0, primary1, primary2]
+        struct.all_databases = [primary0, primary1, primary2]
+      end
+    end
+
+    def self.single_instance_setup(pool_name, pool_size, pool_mode="transaction")
+      user = {
+        "password" => "sharding_user",
+        "pool_size" => pool_size,
+        "statement_timeout" => 0,
+        "username" => "sharding_user"
+      }
+
+      pgcat = PgcatProcess.new("trace")
+      pgcat_cfg = pgcat.current_config
+
+      primary  = PgInstance.new(5432, user["username"], user["password"], "shard0")
+
+      # Main proxy configs
+      pgcat_cfg["pools"] = {
+        "#{pool_name}" => {
+          "default_role" => "primary",
+          "pool_mode" => pool_mode,
+          "primary_reads_enabled" => false,
+          "query_parser_enabled" => false,
+          "sharding_function" => "pg_bigint_hash",
+          "shards" => {
+            "0" => {
+              "database" => "shard0",
+              "servers" => [
+                ["localhost", primary.port.to_s, "primary"]
+              ]
+            },
+          },
+          "users" => { "0" => user }
+        }
+      }
+      pgcat_cfg["general"]["port"] = pgcat.port
+      pgcat.update_config(pgcat_cfg)
+      pgcat.start
+      pgcat.wait_until_ready
+
+      OpenStruct.new.tap do |struct|
+        struct.pgcat = pgcat
+        struct.primary = primary
+        struct.all_databases = [primary]
+      end
+    end
+
+    def self.single_shard_setup(pool_name, pool_size, pool_mode="transaction")
+      user = {
+        "password" => "sharding_user",
+        "pool_size" => pool_size,
+        "statement_timeout" => 0,
+        "username" => "sharding_user"
+      }
+
+      pgcat = PgcatProcess.new("info")
+      pgcat_cfg = pgcat.current_config
+
+      primary  = PgInstance.new(5432, user["username"], user["password"], "shard0")
+      replica0 = PgInstance.new(7432, user["username"], user["password"], "shard0")
+      replica1 = PgInstance.new(8432, user["username"], user["password"], "shard0")
+      replica2 = PgInstance.new(9432, user["username"], user["password"], "shard0")
+
+      # Main proxy configs
+      pgcat_cfg["pools"] = {
+        "#{pool_name}" => {
+          "default_role" => "any",
+          "pool_mode" => pool_mode,
+          "primary_reads_enabled" => false,
+          "query_parser_enabled" => false,
+          "sharding_function" => "pg_bigint_hash",
+          "shards" => {
+            "0" => {
+              "database" => "shard0",
+              "servers" => [
+                ["localhost", primary.port.to_s, "primary"],
+                ["localhost", replica0.port.to_s, "replica"],
+                ["localhost", replica1.port.to_s, "replica"],
+                ["localhost", replica2.port.to_s, "replica"]
+              ]
+            },
+          },
+          "users" => { "0" => user }
+        }
+      }
+      pgcat_cfg["general"]["port"] = pgcat.port
+      pgcat.update_config(pgcat_cfg)
+      pgcat.start
+      pgcat.wait_until_ready
+
+      OpenStruct.new.tap do |struct|
+        struct.pgcat = pgcat
+        struct.primary = primary
+        struct.replicas = [replica0, replica1, replica2]
+        struct.all_databases = [primary, replica0, replica1, replica2]
+      end
+    end
+  end
+end
--- a/tests/ruby/helpers/pgcat_process.rb
+++ b/tests/ruby/helpers/pgcat_process.rb
@@ -0,0 +1,116 @@
+require 'pg'
+require 'toml'
+require 'fileutils'
+require 'securerandom'
+
+class PgcatProcess
+  attr_reader :port
+  attr_reader :pid
+
+  def self.finalize(pid, log_filename, config_filename)
+    `kill #{pid}`
+    File.delete(config_filename) if File.exist?(config_filename)
+    File.delete(log_filename) if File.exist?(log_filename)
+  end
+
+  def initialize(log_level)
+    @env = {"RUST_LOG" => log_level}
+    @port = rand(20000..32760)
+    @log_level = log_level
+    @log_filename = "/tmp/pgcat_log_#{SecureRandom.urlsafe_base64}.log"
+    @config_filename = "/tmp/pgcat_cfg_#{SecureRandom.urlsafe_base64}.toml"
+
+    @command = "../../target/debug/pgcat #{@config_filename}"
+
+    FileUtils.cp("../../pgcat.toml", @config_filename)
+    cfg = current_config
+    cfg["general"]["port"] = @port.to_i
+    cfg["general"]["enable_prometheus_exporter"] = false
+
+    update_config(cfg)
+  end
+
+  def logs
+    File.read(@log_filename)
+  end
+
+  def update_config(config_hash)
+    @original_config = current_config
+    output_to_write = TOML::Generator.new(config_hash).body
+    output_to_write = output_to_write.gsub(/,\s*["|'](\d+)["|']\s*,/, ',\1,')
+    File.write(@config_filename, output_to_write)
+  end
+
+  def current_config
+    old_cfg = File.read(@config_filename)
+    loadable_string = old_cfg.gsub(/,\s*(\d+)\s*,/, ', "\1",')
+    TOML.load(loadable_string)
+  end
+
+  def reload_config
+    `kill -s HUP #{@pid}`
+    sleep 0.1
+  end
+
+  def start
+    raise StandardError, "Process is already started" unless @pid.nil?
+    @pid = Process.spawn(@env, @command, err: @log_filename, out: @log_filename)
+    ObjectSpace.define_finalizer(@log_filename, proc { PgcatProcess.finalize(@pid, @log_filename, @config_filename) })
+
+    return self
+  end
+
+  def wait_until_ready
+    exc = nil
+    10.times do
+      PG::connect(example_connection_string).close
+
+      return self
+    rescue => e
+      exc = e
+      sleep(0.5)
+    end
+    puts exc
+    raise StandardError, "Process #{@pid} never became ready. Logs #{logs}"
+  end
+
+  def stop
+    `kill #{@pid}`
+    sleep 0.1
+  end
+
+  def shutdown
+    stop
+    File.delete(@config_filename) if File.exist?(@config_filename)
+    File.delete(@log_filename) if File.exist?(@log_filename)
+  end
+
+  def admin_connection_string
+    cfg = current_config
+    username = cfg["general"]["admin_username"]
+    password = cfg["general"]["admin_password"]
+
+    "postgresql://#{username}:#{password}@0.0.0.0:#{@port}/pgcat"
+  end
+
+  def connection_string(pool_name, username)
+    cfg = current_config
+
+    user_idx, user_obj = cfg["pools"][pool_name]["users"].detect { |k, user| user["username"] == username }
+    password = user_obj["password"]
+
+    "postgresql://#{username}:#{password}@0.0.0.0:#{@port}/#{pool_name}"
+  end
+
+  def example_connection_string
+    cfg = current_config
+    first_pool_name = cfg["pools"].keys[0]
+
+    db_name = first_pool_name
+
+    username = cfg["pools"][first_pool_name]["users"]["0"]["username"]
+    password = cfg["pools"][first_pool_name]["users"]["0"]["password"]
+
+    "postgresql://#{username}:#{password}@0.0.0.0:#{@port}/#{db_name}?application_name=example_app"
+  end
+end
--- a/tests/ruby/load_balancing_spec.rb
+++ b/tests/ruby/load_balancing_spec.rb
@@ -0,0 +1,61 @@
+# frozen_string_literal: true
+require_relative 'spec_helper'
+
+describe "Load Balancing" do
+  let(:processes) { Helpers::Pgcat.single_shard_setup("sharded_db", 5) }
+  after do
+    processes.all_databases.map(&:reset)
+    processes.pgcat.shutdown
+  end
+
+  context "under regular circumstances" do
+    it "balances query volume between all instances" do
+      conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+
+      query_count = QUERY_COUNT
+      expected_share = query_count / processes.all_databases.count
+      failed_count = 0
+
+      query_count.times do
+        conn.async_exec("SELECT 1 + 2")
+      rescue
+        failed_count += 1
+      end
+
+      expect(failed_count).to eq(0)
+      processes.all_databases.map(&:count_select_1_plus_2).each do |instance_share|
+        expect(instance_share).to be_within(expected_share * MARGIN_OF_ERROR).of(expected_share)
+      end
+    end
+  end
+
+  context "when some replicas are down" do
+    it "balances query volume between working instances" do
+      conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+      expected_share = QUERY_COUNT / (processes.all_databases.count - 2)
+      failed_count = 0
+
+      processes[:replicas][0].take_down do
+        processes[:replicas][1].take_down do
+          QUERY_COUNT.times do
+            conn.async_exec("SELECT 1 + 2")
+          rescue
+            conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+            failed_count += 1
+          end
+        end
+      end
+
+      expect(failed_count).to eq(2)
+      processes.all_databases.each do |instance|
+        queries_routed = instance.count_select_1_plus_2
+        if processes.replicas[0..1].include?(instance)
+          expect(queries_routed).to eq(0)
+        else
+          expect(queries_routed).to be_within(expected_share * MARGIN_OF_ERROR).of(expected_share)
+        end
+      end
+    end
+  end
+end
+
--- a/tests/ruby/misc_spec.rb
+++ b/tests/ruby/misc_spec.rb
@@ -0,0 +1,193 @@
+# frozen_string_literal: true
+require_relative 'spec_helper'
+
+describe "Miscellaneous" do
+  let(:processes) { Helpers::Pgcat.single_shard_setup("sharded_db", 5) }
+  after do
+    processes.all_databases.map(&:reset)
+    processes.pgcat.shutdown
+  end
+
+  describe "Extended Protocol handling" do
+    it "does not send packets that client does not expect during extended protocol sequence" do
+      new_configs = processes.pgcat.current_config
+
+      new_configs["general"]["connect_timeout"] = 500
+      new_configs["general"]["ban_time"] = 1
+      new_configs["general"]["shutdown_timeout"] = 1
+      new_configs["pools"]["sharded_db"]["users"]["0"]["pool_size"] = 1
+
+      processes.pgcat.update_config(new_configs)
+      processes.pgcat.reload_config
+
+      25.times do
+        Thread.new do
+          conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+          conn.async_exec("SELECT pg_sleep(5)") rescue PG::SystemError
+        ensure
+          conn&.close
+        end
+      end
+
+      sleep(0.5)
+      conn_under_test = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+      stdout, stderr = with_captured_stdout_stderr do
+        15.times do |i|
+          conn_under_test.async_exec("SELECT 1") rescue PG::SystemError
+          conn_under_test.exec_params("SELECT #{i} + $1", [i]) rescue PG::SystemError
+          sleep 1
+        end
+      end
+
+      raise StandardError, "Libpq got unexpected messages while idle" if stderr.include?("arrived from server while idle")
+    end
+  end
+
+  describe "Pool recycling after config reload" do
+    let(:processes) { Helpers::Pgcat.three_shard_setup("sharded_db", 5) }
+
+    it "should update pools for new clients and clients that are no longer in transaction" do
+      server_conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+      server_conn.async_exec("BEGIN")
+
+      # No config change yet, client should set old configs
+      current_datebase_from_pg = server_conn.async_exec("SELECT current_database();")[0]["current_database"]
+      expect(current_datebase_from_pg).to eq('shard0')
+
+      # Swap shards
+      new_config = processes.pgcat.current_config
+      shard0 = new_config["pools"]["sharded_db"]["shards"]["0"]
+      shard1 = new_config["pools"]["sharded_db"]["shards"]["1"]
+      new_config["pools"]["sharded_db"]["shards"]["0"] = shard1
+      new_config["pools"]["sharded_db"]["shards"]["1"] = shard0
+
+      # Reload config
+      processes.pgcat.update_config(new_config)
+      processes.pgcat.reload_config
+      sleep 0.5
+
+      # Config changed but transaction is in progress, client should set old configs
+      current_datebase_from_pg = server_conn.async_exec("SELECT current_database();")[0]["current_database"]
+      expect(current_datebase_from_pg).to eq('shard0')
+      server_conn.async_exec("COMMIT")
+
+      # Transaction finished, client should get new configs
+      current_datebase_from_pg = server_conn.async_exec("SELECT current_database();")[0]["current_database"]
+      expect(current_datebase_from_pg).to eq('shard1')
+
+      # New connection should get new configs
+      server_conn.close()
+      server_conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+      current_datebase_from_pg = server_conn.async_exec("SELECT current_database();")[0]["current_database"]
+      expect(current_datebase_from_pg).to eq('shard1')
+    end
+  end
+
+  describe "Clients closing connection in the middle of transaction" do
+    it "sends a rollback to the server" do
+      conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+      conn.async_exec("SET SERVER ROLE to 'primary'")
+      conn.async_exec("BEGIN")
+      conn.close
+
+      expect(processes.primary.count_query("ROLLBACK")).to eq(1)
+    end
+  end
+
+  describe "Server version reporting" do
+    it "reports correct version for normal and admin databases" do
+      server_conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+      expect(server_conn.server_version).not_to eq(0)
+      server_conn.close
+
+      admin_conn = PG::connect(processes.pgcat.admin_connection_string)
+      expect(admin_conn.server_version).not_to eq(0)
+      admin_conn.close
+    end
+  end
+
+  describe "State clearance" do
+    context "session mode" do
+      let(:processes) { Helpers::Pgcat.single_shard_setup("sharded_db", 5, "session") }
+
+      it "Clears state before connection checkin" do
+        # Both modes of operation should not raise
+        # ERROR:  prepared statement "prepared_q" already exists
+        15.times do
+          conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+          conn.async_exec("PREPARE prepared_q (int) AS SELECT $1")
+          conn.close
+        end
+
+        conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        initial_value = conn.async_exec("SHOW statement_timeout")[0]["statement_timeout"]
+        conn.async_exec("SET statement_timeout to 1000")
+        current_value = conn.async_exec("SHOW statement_timeout")[0]["statement_timeout"]
+        expect(conn.async_exec("SHOW statement_timeout")[0]["statement_timeout"]).to eq("1s")
+        conn.close
+      end
+
+      it "Does not send DISCARD ALL unless necessary" do
+        10.times do
+          conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+          conn.async_exec("SET SERVER ROLE to 'primary'")
+          conn.async_exec("SELECT 1")
+          conn.close
+        end
+
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
+
+        10.times do
+          conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+          conn.async_exec("SET SERVER ROLE to 'primary'")
+          conn.async_exec("SELECT 1")
+          conn.async_exec("SET statement_timeout to 5000")
+          conn.close
+        end
+
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(10)
+      end
+    end
+
+    context "transaction mode" do
+      let(:processes) { Helpers::Pgcat.single_shard_setup("sharded_db", 5, "transaction") }
+      it "Clears state before connection checkin" do
+        # Both modes of operation should not raise
+        # ERROR:  prepared statement "prepared_q" already exists
+        15.times do
+          conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+          conn.async_exec("PREPARE prepared_q (int) AS SELECT $1")
+          conn.close
+        end
+
+        15.times do
+          conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+          conn.prepare("prepared_q", "SELECT $1")
+          conn.close
+        end
+      end
+
+      it "Does not send DISCARD ALL unless necessary" do
+        10.times do
+          conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+          conn.async_exec("SET SERVER ROLE to 'primary'")
+          conn.async_exec("SELECT 1")
+          conn.exec_params("SELECT $1", [1])
+          conn.close
+        end
+
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
+
+        10.times do
+          conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+          conn.async_exec("SET SERVER ROLE to 'primary'")
+          conn.async_exec("SELECT 1")
+          conn.async_exec("SET statement_timeout to 5000")
+          conn.close
+        end
+
+        expect(processes.primary.count_query("DISCARD ALL")).to eq(10)
+      end
+    end
+  end
+end
--- a/tests/ruby/routing_spec.rb
+++ b/tests/ruby/routing_spec.rb
@@ -0,0 +1,81 @@
+# frozen_string_literal: true
+require_relative 'spec_helper'
+
+
+describe "Routing" do
+  let(:processes) { Helpers::Pgcat.single_shard_setup("sharded_db", 5) }
+  after do
+    processes.all_databases.map(&:reset)
+    processes.pgcat.shutdown
+  end
+
+  describe "SET ROLE" do
+    context "primary" do
+      it "routes queries only to primary" do
+        conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        conn.async_exec("SET SERVER ROLE to 'primary'")
+
+        query_count = 30
+        failed_count = 0
+
+        query_count.times do
+          conn.async_exec("SELECT 1 + 2")
+        rescue
+          failed_count += 1
+        end
+
+        expect(failed_count).to eq(0)
+        processes.replicas.map(&:count_select_1_plus_2).each do |instance_share|
+          expect(instance_share).to eq(0)
+        end
+
+        expect(processes.primary.count_select_1_plus_2).to eq(query_count)
+      end
+    end
+    context "replica" do
+      it "routes queries only to replicas" do
+        conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        conn.async_exec("SET SERVER ROLE to 'replica'")
+
+        expected_share = QUERY_COUNT / processes.replicas.count
+        failed_count = 0
+
+        QUERY_COUNT.times do
+          conn.async_exec("SELECT 1 + 2")
+        rescue
+          failed_count += 1
+        end
+
+        expect(failed_count).to eq(0)
+
+        processes.replicas.map(&:count_select_1_plus_2).each do |instance_share|
+          expect(instance_share).to be_within(expected_share * MARGIN_OF_ERROR).of(expected_share)
+        end
+
+        expect(processes.primary.count_select_1_plus_2).to eq(0)
+      end
+    end
+
+    context "any" do
+      it "routes queries to all instances" do
+        conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+        conn.async_exec("SET SERVER ROLE to 'any'")
+
+        expected_share = QUERY_COUNT / processes.all_databases.count
+        failed_count = 0
+
+        QUERY_COUNT.times do
+          conn.async_exec("SELECT 1 + 2")
+        rescue
+          failed_count += 1
+        end
+
+        expect(failed_count).to eq(0)
+
+        processes.all_databases.map(&:count_select_1_plus_2).each do |instance_share|
+          expect(instance_share).to be_within(expected_share * MARGIN_OF_ERROR).of(expected_share)
+        end
+      end
+    end
+  end
+end
--- a/tests/ruby/spec_helper.rb
+++ b/tests/ruby/spec_helper.rb
@@ -0,0 +1,21 @@
+# frozen_string_literal: true
+
+require 'pg'
+require_relative 'helpers/pgcat_helper'
+
+QUERY_COUNT = 300
+MARGIN_OF_ERROR = 0.30
+
+def with_captured_stdout_stderr
+  sout = STDOUT.clone
+  serr = STDERR.clone
+  STDOUT.reopen("/tmp/out.txt", "w+")
+  STDERR.reopen("/tmp/err.txt", "w+")
+  STDOUT.sync = true
+  STDERR.sync = true
+  yield
+  return File.read('/tmp/out.txt'), File.read('/tmp/err.txt')
+ensure
+  STDOUT.reopen(sout)
+  STDERR.reopen(serr)
+end
--- a/tests/ruby/tests.rb
+++ b/tests/ruby/tests.rb
@@ -1,93 +1,6 @@
 # frozen_string_literal: true
-
-require 'active_record'
 require 'pg'
-require 'toml'
-
-$stdout.sync = true
-$stderr.sync = true
-
-class ConfigEditor
-  def initialize
-    @original_config_text = File.read('../../.circleci/pgcat.toml')
-    text_to_load = @original_config_text.gsub("5432", "\"5432\"")
-
-    @original_configs = TOML.load(text_to_load)
-  end
-
-  def original_configs
-    TOML.load(TOML::Generator.new(@original_configs).body)
-  end
-
-  def with_modified_configs(new_configs)
-    text_to_write = TOML::Generator.new(new_configs).body
-    text_to_write = text_to_write.gsub("\"5432\"", "5432")
-    File.write('../../.circleci/pgcat.toml', text_to_write)
-    yield
-  ensure
-    File.write('../../.circleci/pgcat.toml', @original_config_text)
-  end
-end
-
-def with_captured_stdout_stderr
-  sout = STDOUT.clone
-  serr = STDERR.clone
-  STDOUT.reopen("/tmp/out.txt", "w+")
-  STDERR.reopen("/tmp/err.txt", "w+")
-  STDOUT.sync = true
-  STDERR.sync = true
-  yield
-  return File.read('/tmp/out.txt'), File.read('/tmp/err.txt')
-ensure
-  STDOUT.reopen(sout)
-  STDERR.reopen(serr)
-end
-
-
-def test_extended_protocol_pooler_errors
-  admin_conn = PG::connect("postgres://admin_user:admin_pass@127.0.0.1:6432/pgcat")
-
-  conf_editor = ConfigEditor.new
-  new_configs = conf_editor.original_configs
-
-  # shorter timeouts
-  new_configs["general"]["connect_timeout"] = 500
-  new_configs["general"]["ban_time"] = 1
-  new_configs["general"]["shutdown_timeout"] = 1
-  new_configs["pools"]["sharded_db"]["users"]["0"]["pool_size"] = 1
-  new_configs["pools"]["sharded_db"]["users"]["1"]["pool_size"] = 1
-
-  conf_editor.with_modified_configs(new_configs) { admin_conn.async_exec("RELOAD") }
-
-  conn_str = "postgres://sharding_user:sharding_user@127.0.0.1:6432/sharded_db"
-  10.times do
-    Thread.new do
-      conn = PG::connect(conn_str)
-      conn.async_exec("SELECT pg_sleep(5)") rescue PG::SystemError
-    ensure
-      conn&.close
-    end
-  end
-
-  sleep(0.5)
-  conn_under_test = PG::connect(conn_str)
-  stdout, stderr = with_captured_stdout_stderr do
-    5.times do |i|
-      conn_under_test.async_exec("SELECT 1") rescue PG::SystemError
-      conn_under_test.exec_params("SELECT #{i} + $1", [i]) rescue PG::SystemError
-      sleep 1
-    end
-  end
-
-  raise StandardError, "Libpq got unexpected messages while idle" if stderr.include?("arrived from server while idle")
-  puts "Pool checkout errors not breaking clients passed"
-ensure
-  sleep 1
-  admin_conn.async_exec("RELOAD") # Reset state
-  conn_under_test&.close
-end
-
-test_extended_protocol_pooler_errors
+require 'active_record'

 # Uncomment these two to see all queries.
 # ActiveRecord.verbose_query_logs = true
@@ -198,68 +111,3 @@ begin
 rescue ActiveRecord::StatementInvalid
  puts 'OK'
 end
-
-# Test evil clients
-def poorly_behaved_client
-  conn = PG::connect("postgres://sharding_user:sharding_user@127.0.0.1:6432/sharded_db?application_name=testing_pgcat")
-  conn.async_exec 'BEGIN'
-  conn.async_exec 'SELECT 1'
-
-  conn.close
-  puts 'Bad client ok'
-end
-
-25.times do
-  poorly_behaved_client
-end
-
-
-def test_server_parameters
-  server_conn = PG::connect("postgres://sharding_user:sharding_user@127.0.0.1:6432/sharded_db?application_name=testing_pgcat")
-  raise StandardError, "Bad server version" if server_conn.server_version == 0
-  server_conn.close
-
-  admin_conn = PG::connect("postgres://admin_user:admin_pass@127.0.0.1:6432/pgcat")
-  raise StandardError, "Bad server version" if admin_conn.server_version == 0
-  admin_conn.close
-
-  puts 'Server parameters ok'
-end
-
-
-def test_reload_pool_recycling
-  admin_conn = PG::connect("postgres://admin_user:admin_pass@127.0.0.1:6432/pgcat")
-  server_conn = PG::connect("postgres://sharding_user:sharding_user@127.0.0.1:6432/sharded_db?application_name=testing_pgcat")
-
-  server_conn.async_exec("BEGIN")
-  conf_editor = ConfigEditor.new
-  new_configs = conf_editor.original_configs
-
-  # swap shards
-  new_configs["pools"]["sharded_db"]["shards"]["0"]["database"] = "shard1"
-  new_configs["pools"]["sharded_db"]["shards"]["1"]["database"] = "shard0"
-
-  raise StandardError if server_conn.async_exec("SELECT current_database();")[0]["current_database"] != 'shard0'
-  conf_editor.with_modified_configs(new_configs) { admin_conn.async_exec("RELOAD") }
-  raise StandardError if server_conn.async_exec("SELECT current_database();")[0]["current_database"] != 'shard0'
-  server_conn.async_exec("COMMIT;")
-
-  # Transaction finished, client should get new configs
-  raise StandardError if server_conn.async_exec("SELECT current_database();")[0]["current_database"] != 'shard1'
-  server_conn.close()
-
-  # New connection should get new configs
-  server_conn = PG::connect("postgres://sharding_user:sharding_user@127.0.0.1:6432/sharded_db?application_name=testing_pgcat")
-  raise StandardError if server_conn.async_exec("SELECT current_database();")[0]["current_database"] != 'shard1'
-
-ensure
-  admin_conn.async_exec("RELOAD") # Go back to old state
-  admin_conn.close
-  server_conn.close
-  puts "Pool Recycling okay!"
-end
-
-test_reload_pool_recycling
-
-
-
--- a/tests/sharding/query_routing_setup.sql
+++ b/tests/sharding/query_routing_setup.sql
@@ -70,23 +70,35 @@ GRANT CONNECT ON DATABASE shard2  TO other_user;
 GRANT CONNECT ON DATABASE some_db TO simple_user;

 \c shard0
+CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
+GRANT EXECUTE ON FUNCTION pg_stat_statements_reset TO sharding_user;
 GRANT ALL ON SCHEMA public TO sharding_user;
 GRANT ALL ON TABLE data TO sharding_user;
 GRANT ALL ON SCHEMA public TO other_user;
 GRANT ALL ON TABLE data TO other_user;
+GRANT EXECUTE ON FUNCTION pg_stat_statements_reset TO other_user;

 \c shard1
+CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
+GRANT EXECUTE ON FUNCTION pg_stat_statements_reset TO sharding_user;
 GRANT ALL ON SCHEMA public TO sharding_user;
 GRANT ALL ON TABLE data TO sharding_user;
 GRANT ALL ON SCHEMA public TO other_user;
 GRANT ALL ON TABLE data TO other_user;
+GRANT EXECUTE ON FUNCTION pg_stat_statements_reset TO other_user;
+

 \c shard2
+CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
+GRANT EXECUTE ON FUNCTION pg_stat_statements_reset TO sharding_user;
 GRANT ALL ON SCHEMA public TO sharding_user;
 GRANT ALL ON TABLE data TO sharding_user;
 GRANT ALL ON SCHEMA public TO other_user;
 GRANT ALL ON TABLE data TO other_user;
+GRANT EXECUTE ON FUNCTION pg_stat_statements_reset TO other_user;

 \c some_db
+CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
+GRANT EXECUTE ON FUNCTION pg_stat_statements_reset TO simple_user;
 GRANT ALL ON SCHEMA public TO simple_user;
 GRANT ALL ON TABLE data TO simple_user;
Author	SHA1	Message	Date
Mostafa Abdelraouf	3d33ccf4b0	Fix maxwait metric (#183 ) Max wait was being reported as 0 after #159 This PR fixes that and adds test	2022-10-05 21:41:09 -05:00
Lev Kokotov	7987c5ffad	Replace a few types with more developer-friendly names (#182 ) * Replace a few types with more developer-friendly names * UserPool -> PoolIdentifier	2022-10-01 10:25:59 -07:00
zainkabani	24f5eec3ea	Change sharding config to enum and move validation of configs into public functions (#178 ) Moves config validation to own functions to enable tools to use them Moves sharding config to enum Makes defaults public Make connect_timeout on pool and option which is overwritten by general connect_timeout	2022-09-28 08:50:14 -05:00
Mostafa Abdelraouf	af064ef447	Set client state to idle after error (#179 ) * Set client state to idle after error * fmt * spelling * clean up	2022-09-24 09:09:15 -07:00
Lev Kokotov	e84a6f834c	Update README.md	2022-09-23 12:24:30 -07:00
Lev Kokotov	19fd677891	Fix the pool fix (#176 ) * Always listen to the compiler * Its fine	2022-09-23 12:06:07 -07:00
Lev Kokotov	964a5e1708	Don't drop connections if DB hasn't changed (#175 ) * Don't drop connections if DB hasn't changed * Incoporate connect_timeout into the pool config * use the field	2022-09-23 11:32:05 -07:00
Mostafa Abdelraouf	d126c7424d	Log failed client logins (#173 ) * Log failed client logins * more logging * remove clones * remove	2022-09-23 09:08:38 -07:00
zainkabani	f72dac420b	Add defaults for configs (#174 ) * add statement timeout to readme * Add defaults to various configs * primary read enabled default to false	2022-09-22 23:00:46 -07:00
zainkabani	3a729bb75b	Minor refactor for configs (#172 ) * Changes shard struct to use vector of ServerConfig * Adds to query router * Change client disconnect with error message to warn instead of debug * Add warning logs for clean up actions	2022-09-22 10:07:02 -07:00
zainkabani	85cc2f4147	Update to latest library versions (#170 )	2022-09-21 13:48:33 -07:00
zainkabani	8c09ab6c20	Export pgcat objects in lib (#169 ) * Export pgcat objects in lib * fmt	2022-09-20 18:47:32 -07:00
Mostafa Abdelraouf	f7a951745c	Report Query times (#166 ) * Report avg and total query timing * Report query times * fmt	2022-09-15 02:21:45 -04:00
Mostafa Abdelraouf	4ae1bc8d32	Add SHOW CLIENTS / SHOW SERVERS + Stats refactor and tests (#159 ) * wip * Main Thread Panic when swarmed with clients * fix * fix * 1024 * fix * remove test * Add SHOW CLIENTS * revert * fmt * Refactor + tests * fmt * add test * Add SHOW SERVERS + Make PR unreviewable * prometheus * add state to clients and servers * fmt * Add application_name to server stats * Add tests for waiting clients * Docs * remove comment * comments * typo * cleanup * CI	2022-09-14 11:20:41 -04:00
Lev Kokotov	075167431d	Add Discord link (#164 ) * Add Discord link * move it up * :) * hmm * hmm * :O	2022-09-08 08:12:37 -07:00
Mostafa Abdelraouf	9514b3b2d1	Clean connection state up after protocol named prepared statement (#163 ) * Clean connection state up after protocol named prepared statement * Avoid cloning + add test * fmt	2022-09-07 20:37:17 -07:00
Lev Kokotov	6d41640ea9	Send signal even if process is gone (#162 ) * Send signal even if process is gone * hmm * hmm	2022-09-07 09:22:52 -07:00
Mostafa Abdelraouf	744ceada86	Better logging for failure to get connection from pool (#161 )	2022-09-07 08:24:07 -07:00
Mostafa Abdelraouf	a5c8dd69b2	Avoid reporting ProtocolSyncError when admin session disconnects (#160 ) * Avoid reporting ProtocolSyncError when admin session disconnects * rebuild * rebuild	2022-09-06 22:22:31 -07:00
zainkabani	6a9a4db648	Adds microsecond logging and also reformats duration to include milliseconds (#156 ) * Adds microsecond logging and also reformats duration to include milliseconds * fmt * attempt to fix cd * revert	2022-09-05 01:21:27 -07:00
Mostafa Abdelraouf	976b406468	Main Thread Panic when swarmed with clients (#158 ) * Main Thread Panic when swarmed with clients * fix * fix * 1024 * fix * remove test * Update src/client.rs * Update src/main.rs * Update src/client.rs * Update src/main.rs Co-authored-by: Lev Kokotov <levkk@users.noreply.github.com>	2022-09-05 01:21:06 -07:00
zain-kabani	417358c35d	Patch graceful shutdown bug (#157 ) * Fixes non-admin client counting error * Add log when sigterm received and log number of active clients when shutdown timeout is reached	2022-09-05 01:02:49 -07:00
Mostafa Abdelraouf	23a642f4a4	Send DISCARD ALL even if client is not in transaction (#152 ) * Send DISCARD ALL even if client is not in transaction * fmt * Added tests + avoided sending extra discard all * Adds set name logic to beginning of handle client * fmt * refactor dead code handling * Refactor reading command tag * remove unnecessary trim * Removing debugging statement * typo * typo{ * documentation * edit text * un-unwrap * run ci * run ci Co-authored-by: Zain Kabani <zain.kabani@instacart.com>	2022-09-01 20:06:55 -07:00
Mostafa Abdelraouf	7f20dc3054	Better handling extended protocol messages in the event of busy pool (#155 ) * Better handling for checkout errors during extended protocol messages * Fix specs * comment	2022-09-01 15:02:39 -07:00
Mostafa Abdelraouf	36339bd96f	Log Address information in connection create/drop (#154 ) * Log Address information in connection create/drop * run ci	2022-09-01 11:16:22 -07:00
Mostafa Abdelraouf	65b69b46d2	Allow running integration tests with coverage locally (#151 )	2022-08-30 10:43:45 -07:00
Mostafa Abdelraouf	d48c04a7fb	Ruby integration tests (#147 ) * Ruby integration tests * forgot a file * refactor * refactoring * more refactoring * remove config helper * try multiple databases * fix * more databases * Use pg stats * ports * speed * Fix tests * preload library * comment	2022-08-30 09:14:53 -07:00
zainkabani	2628dec42e	Move autoreloader to own tokio task (#148 )	2022-08-29 00:08:44 -07:00