mm

remove debug msg
reqs
2026-03-23 09:26:30 +00:00 · 2023-04-22 08:02:20 -07:00 · 2023-04-22 07:47:19 -07:00 · 2023-04-22 07:40:21 -07:00
37 changed files with 465 additions and 2717 deletions
--- a/.circleci/run_tests.sh
+++ b/.circleci/run_tests.sh
@@ -110,6 +110,10 @@ python3 tests/python/tests.py || exit 1

 start_pgcat "info"

+python3 tests/python/async_test.py
+
+start_pgcat "info"
+
 # Admin tests
 export PGPASSWORD=admin_pass
 psql -U admin_user -e -h 127.0.0.1 -p 6432 -d pgbouncer -c 'SHOW STATS' > /dev/null
--- a/CONFIG.md
+++ b/CONFIG.md
@@ -49,14 +49,6 @@ default: 30000 # milliseconds

 How long an idle connection with a server is left open (ms).

-### server_lifetime
-```
-path: general.server_lifetime
-default: 86400000 # 24 hours
-```
-
-Max connection lifetime before it's closed, even if actively used.
-
 ### idle_client_in_transaction_timeout
 ```
 path: general.idle_client_in_transaction_timeout
@@ -188,22 +180,6 @@ default: "admin_pass"

 Password to access the virtual administrative database

-### dns_cache_enabled
-```
-path: general.dns_cache_enabled
-default: false
-```
-When enabled, ip resolutions for server connections specified using hostnames will be cached
-and checked for changes every `dns_max_ttl` seconds. If a change in the host resolution is found
-old ip connections are closed (gracefully) and new connections will start using new ip.
-
-### dns_max_ttl
-```
-path: general.dns_max_ttl
-default: 30
-```
-Specifies how often (in seconds) cached ip addresses for servers are rechecked (see `dns_cache_enabled`).
-
 ## `pools.<pool_name>` Section

 ### pool_mode
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -26,27 +26,6 @@ version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bddcadddf5e9015d310179a59bb28c4d4b9920ad0f11e8e14dbadf654890c9a6"

-[[package]]
-name = "async-stream"
-version = "0.3.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dad5c83079eae9969be7fadefe640a1c566901f05ff91ab221de4b6f68d9507e"
-dependencies = [
- "async-stream-impl",
- "futures-core",
-]
-
-[[package]]
-name = "async-stream-impl"
-version = "0.3.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "10f203db73a71dfa2fb6dd22763990fa26f3d2625a6da2da900d23b87d26be27"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 1.0.109",
-]
-
 [[package]]
 name = "async-trait"
 version = "0.1.68"
@@ -233,12 +212,6 @@ dependencies = [
 "syn 1.0.109",
 ]

-[[package]]
-name = "data-encoding"
-version = "2.3.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3ee2393c4a91429dffb4bedf19f4d6abf27d8a732c8ce4980305d782e5426d57"
-
 [[package]]
 name = "digest"
 version = "0.10.6"
@@ -250,24 +223,6 @@ dependencies = [
 "subtle",
 ]

-[[package]]
-name = "either"
-version = "1.8.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7fcaabb2fef8c910e7f4c7ce9f67a1283a1715879a7c230ca9d6d1ae31f16d91"
-
-[[package]]
-name = "enum-as-inner"
-version = "0.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c9720bba047d567ffc8a3cba48bf19126600e249ab7f128e9233e6376976a116"
-dependencies = [
- "heck",
- "proc-macro2",
- "quote",
- "syn 1.0.109",
-]
-
 [[package]]
 name = "env_logger"
 version = "0.10.0"
@@ -314,27 +269,12 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7"

-[[package]]
-name = "fallible-iterator"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
-
 [[package]]
 name = "fnv"
 version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"

-[[package]]
-name = "form_urlencoded"
-version = "1.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9c384f161156f5260c24a097c56119f9be8c798586aecc13afbcbe7b7e26bf8"
-dependencies = [
- "percent-encoding",
-]
-
 [[package]]
 name = "futures"
 version = "0.3.28"
@@ -470,12 +410,6 @@ version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888"

-[[package]]
-name = "heck"
-version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2540771e65fc8cb83cd6e8a237f70c319bd5c29f78ed1084ba5d50eeac86f7f9"
-
 [[package]]
 name = "hermit-abi"
 version = "0.2.6"
@@ -500,17 +434,6 @@ dependencies = [
 "digest",
 ]

-[[package]]
-name = "hostname"
-version = "0.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c731c3e10504cc8ed35cfe2f1db4c9274c3d35fa486e3b31df46f068ef3e867"
-dependencies = [
- "libc",
- "match_cfg",
- "winapi",
-]
-
 [[package]]
 name = "http"
 version = "0.2.9"
@@ -599,27 +522,6 @@ dependencies = [
 "cxx-build",
 ]

-[[package]]
-name = "idna"
-version = "0.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "418a0a6fab821475f634efe3ccc45c013f742efe03d853e8d3355d5cb850ecf8"
-dependencies = [
- "matches",
- "unicode-bidi",
- "unicode-normalization",
-]
-
-[[package]]
-name = "idna"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e14ddfc70884202db2244c223200c204c2bda1bc6e0998d11b5e024d657209e6"
-dependencies = [
- "unicode-bidi",
- "unicode-normalization",
-]
-
 [[package]]
 name = "indexmap"
 version = "1.9.2"
@@ -640,24 +542,6 @@ dependencies = [
 "windows-sys",
 ]

-[[package]]
-name = "ipconfig"
-version = "0.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bd302af1b90f2463a98fa5ad469fc212c8e3175a41c3068601bfa2727591c5be"
-dependencies = [
- "socket2",
- "widestring",
- "winapi",
- "winreg",
-]
-
-[[package]]
-name = "ipnet"
-version = "2.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f88c5561171189e69df9d98bcf18fd5f9558300f7ea7b801eb8a0fd748bd8745"
-
 [[package]]
 name = "is-terminal"
 version = "0.4.4"
@@ -670,15 +554,6 @@ dependencies = [
 "windows-sys",
 ]

-[[package]]
-name = "itertools"
-version = "0.10.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473"
-dependencies = [
- "either",
-]
-
 [[package]]
 name = "itoa"
 version = "1.0.5"
@@ -714,12 +589,6 @@ dependencies = [
 "wasm-bindgen",
 ]

-[[package]]
-name = "lazy_static"
-version = "1.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
-
 [[package]]
 name = "libc"
 version = "0.2.139"
@@ -735,12 +604,6 @@ dependencies = [
 "cc",
 ]

-[[package]]
-name = "linked-hash-map"
-version = "0.5.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0717cef1bc8b636c6e1c1bbdefc09e6322da8a9321966e8928ef80d20f7f770f"
-
 [[package]]
 name = "linux-raw-sys"
 version = "0.1.4"
@@ -766,27 +629,6 @@ dependencies = [
 "cfg-if",
 ]

-[[package]]
-name = "lru-cache"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "31e24f1ad8321ca0e8a1e0ac13f23cb668e6f5466c2c57319f6a5cf1cc8e3b1c"
-dependencies = [
- "linked-hash-map",
-]
-
-[[package]]
-name = "match_cfg"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ffbee8634e0d45d258acb448e7eaab3fce7a0a467395d4d9f228e3c1f01fb2e4"
-
-[[package]]
-name = "matches"
-version = "0.1.9"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a3e378b66a060d48947b590737b30a1be76706c8dd7b8ba0f2fe3989c68a853f"
-
 [[package]]
 name = "md-5"
 version = "0.10.5"
@@ -895,15 +737,9 @@ dependencies = [
 "windows-sys",
 ]

-[[package]]
-name = "percent-encoding"
-version = "2.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "478c572c3d73181ff3c2539045f6eb99e5491218eae919370993b890cdbdd98e"
-
 [[package]]
 name = "pgcat"
-version = "1.0.2-alpha3"
+version = "1.0.1"
 dependencies = [
 "arc-swap",
 "async-trait",
@@ -914,11 +750,10 @@ dependencies = [
 "chrono",
 "env_logger",
 "exitcode",
- "fallible-iterator 0.3.0",
+ "fallible-iterator",
 "futures",
 "hmac",
 "hyper",
- "itertools",
 "jemallocator",
 "log",
 "md-5",
@@ -927,15 +762,12 @@ dependencies = [
 "once_cell",
 "parking_lot",
 "phf",
- "pin-project",
 "postgres-protocol",
 "rand",
 "regex",
- "rustls",
 "rustls-pemfile",
 "serde",
 "serde_derive",
- "serde_json",
 "sha-1",
 "sha2",
 "socket2",
@@ -943,10 +775,7 @@ dependencies = [
 "stringprep",
 "tokio",
 "tokio-rustls",
- "tokio-test",
 "toml",
- "trust-dns-resolver",
- "webpki-roots",
 ]

 [[package]]
@@ -991,26 +820,6 @@ dependencies = [
 "siphasher",
 ]

-[[package]]
-name = "pin-project"
-version = "1.0.12"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ad29a609b6bcd67fee905812e544992d216af9d755757c05ed2d0e15a74c6ecc"
-dependencies = [
- "pin-project-internal",
-]
-
-[[package]]
-name = "pin-project-internal"
-version = "1.0.12"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "069bdb1e05adc7a8990dce9cc75370895fbe4e3d58b9b73bf1aee56359344a55"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 1.0.109",
-]
-
 [[package]]
 name = "pin-project-lite"
 version = "0.2.9"
@@ -1032,7 +841,7 @@ dependencies = [
 "base64",
 "byteorder",
 "bytes",
- "fallible-iterator 0.2.0",
+ "fallible-iterator",
 "hmac",
 "md-5",
 "memchr",
@@ -1056,12 +865,6 @@ dependencies = [
 "unicode-ident",
 ]

-[[package]]
-name = "quick-error"
-version = "1.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0"
-
 [[package]]
 name = "quote"
 version = "1.0.26"
@@ -1112,9 +915,9 @@ dependencies = [

 [[package]]
 name = "regex"
-version = "1.8.1"
+version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "af83e617f331cc6ae2da5443c602dfa5af81e517212d9d611a5b3ba1777b5370"
+checksum = "ac6cf59af1067a3fb53fbe5c88c053764e930f932be1d71d3ffe032cbe147f59"
 dependencies = [
 "aho-corasick",
 "memchr",
@@ -1123,19 +926,9 @@ dependencies = [

 [[package]]
 name = "regex-syntax"
-version = "0.7.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a5996294f19bd3aae0453a862ad728f60e6600695733dd5df01da90c54363a3c"
-
-[[package]]
-name = "resolv-conf"
 version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "52e44394d2086d010551b14b53b1f24e31647570cd1deb0379e2c21b329aba00"
-dependencies = [
- "hostname",
- "quick-error",
-]
+checksum = "b6868896879ba532248f33598de5181522d8b3d9d724dfd230911e1a7d4822f5"

 [[package]]
 name = "ring"
@@ -1168,9 +961,9 @@ dependencies = [

 [[package]]
 name = "rustls"
-version = "0.21.1"
+version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c911ba11bc8433e811ce56fde130ccf32f5127cab0e0194e9c68c5a5b671791e"
+checksum = "07180898a28ed6a7f7ba2311594308f595e3dd2e3c3812fa0a80a47b45f17e5d"
 dependencies = [
 "log",
 "ring",
@@ -1197,12 +990,6 @@ dependencies = [
 "untrusted",
 ]

-[[package]]
-name = "ryu"
-version = "1.0.13"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f91339c0467de62360649f8d3e185ca8de4224ff281f66000de5eb2a77a79041"
-
 [[package]]
 name = "scopeguard"
 version = "1.1.0"
@@ -1230,9 +1017,6 @@ name = "serde"
 version = "1.0.160"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bb2f3770c8bce3bcda7e149193a069a0f4365bda1fa5cd88e03bca26afc1216c"
-dependencies = [
- "serde_derive",
-]

 [[package]]
 name = "serde_derive"
@@ -1245,22 +1029,11 @@ dependencies = [
 "syn 2.0.9",
 ]

-[[package]]
-name = "serde_json"
-version = "1.0.96"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "057d394a50403bcac12672b2b18fb387ab6d289d957dab67dd201875391e52f1"
-dependencies = [
- "itoa",
- "ryu",
- "serde",
-]
-
 [[package]]
 name = "serde_spanned"
-version = "0.6.2"
+version = "0.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "93107647184f6027e3b7dcb2e11034cf95ffa1e3a682c67951963ac69c1c007d"
+checksum = "0efd8caf556a6cebd3b285caf480045fcc1ac04f6bd786b09a6f11af30c4fcf4"
 dependencies = [
 "serde",
 ]
@@ -1340,18 +1113,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "355dc4d4b6207ca8a3434fc587db0a8016130a574dbcdbfb93d7f7b5bc5b211a"
 dependencies = [
 "log",
- "sqlparser_derive",
-]
-
-[[package]]
-name = "sqlparser_derive"
-version = "0.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "55fe75cb4a364c7f7ae06c7dbbc8d84bddd85d6cdf9975963c3935bc1991761e"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 1.0.109",
 ]

 [[package]]
@@ -1407,26 +1168,6 @@ dependencies = [
 "winapi-util",
 ]

-[[package]]
-name = "thiserror"
-version = "1.0.37"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "10deb33631e3c9018b9baf9dcbbc4f737320d2b576bac10f6aefa048fa407e3e"
-dependencies = [
- "thiserror-impl",
-]
-
-[[package]]
-name = "thiserror-impl"
-version = "1.0.37"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "982d17546b47146b28f7c22e3d08465f6b8903d0ea13c1660d9d84a6e7adcdbb"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 1.0.109",
-]
-
 [[package]]
 name = "time"
 version = "0.1.45"
@@ -1494,30 +1235,6 @@ dependencies = [
 "tokio",
 ]

-[[package]]
-name = "tokio-stream"
-version = "0.1.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d660770404473ccd7bc9f8b28494a811bc18542b915c0855c51e8f419d5223ce"
-dependencies = [
- "futures-core",
- "pin-project-lite",
- "tokio",
-]
-
-[[package]]
-name = "tokio-test"
-version = "0.4.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "53474327ae5e166530d17f2d956afcb4f8a004de581b3cae10f12006bc8163e3"
-dependencies = [
- "async-stream",
- "bytes",
- "futures-core",
- "tokio",
- "tokio-stream",
-]
-
 [[package]]
 name = "tokio-util"
 version = "0.7.7"
@@ -1534,9 +1251,9 @@ dependencies = [

 [[package]]
 name = "toml"
-version = "0.7.4"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6135d499e69981f9ff0ef2167955a5333c35e36f6937d382974566b3d5b94ec"
+checksum = "b403acf6f2bb0859c93c7f0d967cb4a75a7ac552100f9322faf64dc047669b21"
 dependencies = [
 "serde",
 "serde_spanned",
@@ -1546,18 +1263,18 @@ dependencies = [

 [[package]]
 name = "toml_datetime"
-version = "0.6.2"
+version = "0.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5a76a9312f5ba4c2dec6b9161fdf25d87ad8a09256ccea5a556fef03c706a10f"
+checksum = "3ab8ed2edee10b50132aed5f331333428b011c99402b5a534154ed15746f9622"
 dependencies = [
 "serde",
 ]

 [[package]]
 name = "toml_edit"
-version = "0.19.9"
+version = "0.19.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92d964908cec0d030b812013af25a0e57fddfadb1e066ecc6681d86253129d4f"
+checksum = "08de71aa0d6e348f070457f85af8bd566e2bc452156a423ddf22861b3a953fae"
 dependencies = [
 "indexmap",
 "serde",
@@ -1580,21 +1297,9 @@ checksum = "8ce8c33a8d48bd45d624a6e523445fd21ec13d3653cd51f681abf67418f54eb8"
 dependencies = [
 "cfg-if",
 "pin-project-lite",
- "tracing-attributes",
 "tracing-core",
 ]

-[[package]]
-name = "tracing-attributes"
-version = "0.1.23"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4017f8f45139870ca7e672686113917c71c7a6e02d4924eda67186083c03081a"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 1.0.109",
-]
-
 [[package]]
 name = "tracing-core"
 version = "0.1.30"
@@ -1604,51 +1309,6 @@ dependencies = [
 "once_cell",
 ]

-[[package]]
-name = "trust-dns-proto"
-version = "0.22.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4f7f83d1e4a0e4358ac54c5c3681e5d7da5efc5a7a632c90bb6d6669ddd9bc26"
-dependencies = [
- "async-trait",
- "cfg-if",
- "data-encoding",
- "enum-as-inner",
- "futures-channel",
- "futures-io",
- "futures-util",
- "idna 0.2.3",
- "ipnet",
- "lazy_static",
- "rand",
- "smallvec",
- "thiserror",
- "tinyvec",
- "tokio",
- "tracing",
- "url",
-]
-
-[[package]]
-name = "trust-dns-resolver"
-version = "0.22.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aff21aa4dcefb0a1afbfac26deb0adc93888c7d295fb63ab273ef276ba2b7cfe"
-dependencies = [
- "cfg-if",
- "futures-util",
- "ipconfig",
- "lazy_static",
- "lru-cache",
- "parking_lot",
- "resolv-conf",
- "smallvec",
- "thiserror",
- "tokio",
- "tracing",
- "trust-dns-proto",
-]
-
 [[package]]
 name = "try-lock"
 version = "0.2.4"
@@ -1694,17 +1354,6 @@ version = "0.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a"

-[[package]]
-name = "url"
-version = "2.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0d68c799ae75762b8c3fe375feb6600ef5602c883c5d21eb51c09f22b83c4643"
-dependencies = [
- "form_urlencoded",
- "idna 0.3.0",
- "percent-encoding",
-]
-
 [[package]]
 name = "version_check"
 version = "0.9.4"
@@ -1797,21 +1446,6 @@ dependencies = [
 "wasm-bindgen",
 ]

-[[package]]
-name = "webpki-roots"
-version = "0.23.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aa54963694b65584e170cf5dc46aeb4dcaa5584e652ff5f3952e56d66aff0125"
-dependencies = [
- "rustls-webpki",
-]
-
-[[package]]
-name = "widestring"
-version = "0.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "17882f045410753661207383517a6f62ec3dbeb6a4ed2acce01f0728238d1983"
-
 [[package]]
 name = "winapi"
 version = "0.3.9"
@@ -1911,18 +1545,9 @@ checksum = "447660ad36a13288b1db4d4248e857b510e8c3a225c822ba4fb748c0aafecffd"

 [[package]]
 name = "winnow"
-version = "0.4.6"
+version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "61de7bac303dc551fe038e2b3cef0f571087a47571ea6e79a87692ac99b99699"
+checksum = "faf09497b8f8b5ac5d3bb4d05c0a99be20f26fd3d5f2db7b0716e946d5103658"
 dependencies = [
 "memchr",
 ]
-
-[[package]]
-name = "winreg"
-version = "0.10.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "80d0f4e272c85def139476380b12f9ac60926689dd2e01d4923222f40580869d"
-dependencies = [
- "winapi",
-]
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "pgcat"
-version = "1.0.2-alpha3"
+version = "1.0.1"
 edition = "2021"

 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -14,12 +14,12 @@ rand = "0.8"
 chrono = "0.4"
 sha-1 = "0.10"
 toml = "0.7"
-serde = { version = "1", features = ["derive"] }
+serde = "1"
 serde_derive = "1"
 regex = "1"
 num_cpus = "1"
 once_cell = "1"
-sqlparser = {version = "0.33", features = ["visitor"] }
+sqlparser = "0.33.0"
 log = "0.4"
 arc-swap = "1"
 env_logger = "0.10"
@@ -38,15 +38,7 @@ socket2 = { version = "0.4.7", features = ["all"] }
 nix = "0.26.2"
 atomic_enum = "0.2.0"
 postgres-protocol = "0.6.5"
-fallible-iterator = "0.3"
-pin-project = "1"
-webpki-roots = "0.23"
-rustls = { version = "0.21", features = ["dangerous_configuration"] }
-trust-dns-resolver = "0.22.0"
-tokio-test = "0.4.2"
-serde_json = "1"
-itertools = "0.10"
+fallible-iterator = "0.2"

 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 jemallocator = "0.5.0"
-
--- a/README.md
+++ b/README.md
@@ -18,7 +18,7 @@ PostgreSQL pooler and proxy (like PgBouncer) with support for sharding, load bal
 | Failover | **Stable** | Queries are automatically rerouted around broken replicas, validated by regular health checks. |
 | Admin database statistics | **Stable** | Pooler statistics and administration via the `pgbouncer` and `pgcat` databases. |
 | Prometheus statistics | **Stable** | Statistics are reported via a HTTP endpoint for Prometheus. |
-| SSL/TLS | **Stable** | Clients can connect to the pooler using TLS. Pooler can connect to Postgres servers using TLS. |
+| Client TLS | **Stable** | Clients can connect to the pooler using TLS/SSL. |
 | Client/Server authentication | **Stable** | Clients can connect using MD5 authentication, supported by `libpq` and all Postgres client drivers. PgCat can connect to Postgres using MD5 and SCRAM-SHA-256. |
 | Live configuration reloading | **Stable** | Identical to PgBouncer; all settings can be reloaded dynamically (except `host` and `port`). |
 | Auth passthrough | **Stable** | MD5 password authentication can be configured to use an `auth_query` so no cleartext passwords are needed in the config file.|
--- a/dev/docker-compose.yaml
+++ b/dev/docker-compose.yaml
@@ -25,7 +25,7 @@ x-common-env-pg:

 services:
  main:
-    image: gcr.io/google_containers/pause:3.2
+    image: kubernetes/pause
    ports:
      - 6432

@@ -64,7 +64,7 @@ services:
      <<: *common-env-pg
      POSTGRES_INITDB_ARGS: --auth-local=md5 --auth-host=md5 --auth=md5
      PGPORT: 10432
-    command: ["postgres", "-p", "10432", "-c", "shared_preload_libraries=pg_stat_statements", "-c", "pg_stat_statements.track=all", "-c", "pg_stat_statements.max=100000"]
+    command: ["postgres", "-p", "5432", "-c", "shared_preload_libraries=pg_stat_statements", "-c", "pg_stat_statements.track=all", "-c", "pg_stat_statements.max=100000"]

  toxiproxy:
    build: .
--- a/pgcat.minimal.toml
+++ b/pgcat.minimal.toml
@@ -1,22 +0,0 @@
-# This is an example of the most basic config
-# that will mimic what PgBouncer does in transaction mode with one server.
-
-[general]
-
-host = "0.0.0.0"
-port = 6433
-admin_username = "pgcat"
-admin_password = "pgcat"
-
-[pools.pgml.users.0]
-username = "postgres"
-password = "postgres"
-pool_size = 10
-min_pool_size = 1
-pool_mode = "transaction"
-
-[pools.pgml.shards.0]
-servers = [
-  ["127.0.0.1", 28815, "primary"]
-]
-database = "postgres"
--- a/pgcat.toml
+++ b/pgcat.toml
@@ -23,9 +23,6 @@ connect_timeout = 5000 # milliseconds
 # How long an idle connection with a server is left open (ms).
 idle_timeout = 30000 # milliseconds

-# Max connection lifetime before it's closed, even if actively used.
-server_lifetime = 86400000 # 24 hours
-
 # How long a client is allowed to be idle while in a transaction (ms).
 idle_client_in_transaction_timeout = 0 # milliseconds

@@ -61,15 +58,9 @@ tcp_keepalives_count = 5
 tcp_keepalives_interval = 5

 # Path to TLS Certificate file to use for TLS connections
-# tls_certificate = ".circleci/server.cert"
+# tls_certificate = "server.cert"
 # Path to TLS private key file to use for TLS connections
-# tls_private_key = ".circleci/server.key"
-
-# Enable/disable server TLS
-server_tls = false
-
-# Verify server certificate is completely authentic.
-verify_server_certificate = false
+# tls_private_key = "server.key"

 # User name to access the virtual administrative database (pgbouncer or pgcat)
 # Connecting to that database allows running commands like `SHOW POOLS`, `SHOW DATABASES`, etc..
@@ -77,58 +68,6 @@ admin_username = "admin_user"
 # Password to access the virtual administrative database
 admin_password = "admin_pass"

-# Default plugins that are configured on all pools.
-[plugins]
-
-# Prewarmer plugin that runs queries on server startup, before giving the connection
-# to the client.
-[plugins.prewarmer]
-enabled = false
-queries = [
-  "SELECT pg_prewarm('pgbench_accounts')",
-]
-
-# Log all queries to stdout.
-[plugins.query_logger]
-enabled = false
-
-# Block access to tables that Postgres does not allow us to control.
-[plugins.table_access]
-enabled = false
-tables = [
-  "pg_user",
-  "pg_roles",
-  "pg_database",
-]
-
-# Intercept user queries and give a fake reply.
-[plugins.intercept]
-enabled = true
-
-[plugins.intercept.queries.0]
-
-query = "select current_database() as a, current_schemas(false) as b"
-schema = [
-  ["a", "text"],
-  ["b", "text"],
-]
-result = [
-  ["${DATABASE}", "{public}"],
-]
-
-[plugins.intercept.queries.1]
-
-query = "select current_database(), current_schema(), current_user"
-schema = [
-  ["current_database", "text"],
-  ["current_schema", "text"],
-  ["current_user", "text"],
-]
-result = [
-  ["${DATABASE}", "public", "${USER}"],
-]
-
-
 # pool configs are structured as pool.<pool_name>
 # the pool_name is what clients use as database name when connecting.
 # For a pool named `sharded_db`, clients access that pool using connection string like
@@ -198,61 +137,6 @@ idle_timeout = 40000
 # Connect timeout can be overwritten in the pool
 connect_timeout = 3000

-# When enabled, ip resolutions for server connections specified using hostnames will be cached
-# and checked for changes every `dns_max_ttl` seconds. If a change in the host resolution is found
-# old ip connections are closed (gracefully) and new connections will start using new ip.
-# dns_cache_enabled = false
-
-# Specifies how often (in seconds) cached ip addresses for servers are rechecked (see `dns_cache_enabled`).
-# dns_max_ttl = 30
-
-# Plugins can be configured on a pool-per-pool basis. This overrides the global plugins setting,
-# so all plugins have to be configured here again.
-[pool.sharded_db.plugins]
-
-[pools.sharded_db.plugins.prewarmer]
-enabled = true
-queries = [
-  "SELECT pg_prewarm('pgbench_accounts')",
-]
-
-[pools.sharded_db.plugins.query_logger]
-enabled = false
-
-[pools.sharded_db.plugins.table_access]
-enabled = false
-tables = [
-  "pg_user",
-  "pg_roles",
-  "pg_database",
-]
-
-[pools.sharded_db.plugins.intercept]
-enabled = true
-
-[pools.sharded_db.plugins.intercept.queries.0]
-
-query = "select current_database() as a, current_schemas(false) as b"
-schema = [
-  ["a", "text"],
-  ["b", "text"],
-]
-result = [
-  ["${DATABASE}", "{public}"],
-]
-
-[pools.sharded_db.plugins.intercept.queries.1]
-
-query = "select current_database(), current_schema(), current_user"
-schema = [
-  ["current_database", "text"],
-  ["current_schema", "text"],
-  ["current_user", "text"],
-]
-result = [
-  ["${DATABASE}", "public", "${USER}"],
-]
-
 # User configs are structured as pool.<pool_name>.users.<user_index>
 # This section holds the credentials for users that may connect to this cluster
 [pools.sharded_db.users.0]
@@ -322,8 +206,6 @@ sharding_function = "pg_bigint_hash"
 username = "simple_user"
 password = "simple_user"
 pool_size = 5
-min_pool_size = 3
-server_lifetime = 60000
 statement_timeout = 0

 [pools.simple_db.shards.0]
--- a/src/admin.rs
+++ b/src/admin.rs
@@ -12,9 +12,9 @@ use tokio::time::Instant;
 use crate::config::{get_config, reload_config, VERSION};
 use crate::errors::Error;
 use crate::messages::*;
-use crate::pool::ClientServerMap;
 use crate::pool::{get_all_pools, get_pool};
 use crate::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();
--- a/src/auth_passthrough.rs
+++ b/src/auth_passthrough.rs
@@ -77,8 +77,6 @@ impl AuthPassthrough {
            pool_size: 1,
            statement_timeout: 0,
            pool_mode: None,
-            server_lifetime: None,
-            min_pool_size: None,
        };

        let user = &address.username;
--- a/src/client.rs
+++ b/src/client.rs
@@ -16,7 +16,6 @@ use crate::auth_passthrough::refetch_auth_hash;
 use crate::config::{get_config, get_idle_client_in_transaction_timeout, Address, PoolMode};
 use crate::constants::*;
 use crate::messages::*;
-use crate::plugins::PluginOutput;
 use crate::pool::{get_pool, ClientServerMap, ConnectionPool};
 use crate::query_router::{Command, QueryRouter};
 use crate::server::Server;
@@ -540,7 +539,6 @@ where
                Some(md5_hash_password(username, password, &salt))
            } else {
                if !get_config().is_auth_query_configured() {
-                    wrong_password(&mut write, username).await?;
                    return Err(Error::ClientAuthImpossible(username.into()));
                }

@@ -567,8 +565,6 @@ where
                        }

                        Err(err) => {
-                            wrong_password(&mut write, username).await?;
-
                            return Err(Error::ClientAuthPassthroughError(
                                err.to_string(),
                                client_identifier,
@@ -591,15 +587,7 @@ where
                    client_identifier
                );

-                let fetched_hash = match refetch_auth_hash(&pool).await {
-                    Ok(fetched_hash) => fetched_hash,
-                    Err(err) => {
-                        wrong_password(&mut write, username).await?;
-
-                        return Err(err);
-                    }
-                };
-
+                let fetched_hash = refetch_auth_hash(&pool).await?;
                let new_password_hash = md5_hash_second_pass(&fetched_hash, &salt);

                // Ok password changed in server an auth is possible.
@@ -766,9 +754,6 @@ where

        self.stats.register(self.stats.clone());

-        // Result returned by one of the plugins.
-        let mut plugin_output = None;
-
        // Our custom protocol loop.
        // We expect the client to either start a transaction with regular queries
        // or issue commands for our sharding and server selection protocol.
@@ -819,25 +804,7 @@ where

                'Q' => {
                    if query_router.query_parser_enabled() {
-                        if let Ok(ast) = QueryRouter::parse(&message) {
-                            let plugin_result = query_router.execute_plugins(&ast).await;
-
-                            match plugin_result {
-                                Ok(PluginOutput::Deny(error)) => {
-                                    error_response(&mut self.write, &error).await?;
-                                    continue;
-                                }
-
-                                Ok(PluginOutput::Intercept(result)) => {
-                                    write_all(&mut self.write, result).await?;
-                                    continue;
-                                }
-
-                                _ => (),
-                            };
-
-                            let _ = query_router.infer(&ast);
-                        }
+                        query_router.infer(&message);
                    }
                }

@@ -845,13 +812,7 @@ where
                    self.buffer.put(&message[..]);

                    if query_router.query_parser_enabled() {
-                        if let Ok(ast) = QueryRouter::parse(&message) {
-                            if let Ok(output) = query_router.execute_plugins(&ast).await {
-                                plugin_output = Some(output);
-                            }
-
-                            let _ = query_router.infer(&ast);
-                        }
+                        query_router.infer(&message);
                    }

                    continue;
@@ -885,18 +846,6 @@ where
                continue;
            }

-            // Check on plugin results.
-            match plugin_output {
-                Some(PluginOutput::Deny(error)) => {
-                    self.buffer.clear();
-                    error_response(&mut self.write, &error).await?;
-                    plugin_output = None;
-                    continue;
-                }
-
-                _ => (),
-            };
-
            // Get a pool instance referenced by the most up-to-date
            // pointer. This ensures we always read the latest config
            // when starting a query.
@@ -983,7 +932,7 @@ where
            }

            // Grab a server from the pool.
-            let connection = match pool
+            let mut connection = match pool
                .get(query_router.shard(), query_router.role(), &self.stats)
                .await
            {
@@ -1026,9 +975,8 @@ where
                }
            };

-            let mut reference = connection.0;
+            let server = &mut *connection.0;
            let address = connection.1;
-            let server = &mut *reference;

            // Server is assigned to the client in case the client wants to
            // cancel a query later.
@@ -1051,6 +999,7 @@ where

            // Set application_name.
            server.set_name(&self.application_name).await?;
+            server.switch_async(false);

            let mut initial_message = Some(message);

@@ -1070,12 +1019,37 @@ where
                    None => {
                        trace!("Waiting for message inside transaction or in session mode");

-                        match tokio::time::timeout(
-                            idle_client_timeout_duration,
-                            read_message(&mut self.read),
-                        )
-                        .await
-                        {
+                        let message = tokio::select! {
+                            message = tokio::time::timeout(
+                                idle_client_timeout_duration,
+                                read_message(&mut self.read),
+                            ) => message,
+
+                            server_message = server.recv() => {
+                                debug!("Got async message");
+
+                                let server_message = match server_message {
+                                    Ok(message) => message,
+                                    Err(err) => {
+                                        pool.ban(&address, BanReason::MessageReceiveFailed, Some(&self.stats));
+                                        server.mark_bad();
+                                        return Err(err);
+                                    }
+                                };
+
+                                match write_all_half(&mut self.write, &server_message).await {
+                                    Ok(_) => (),
+                                    Err(err) => {
+                                        server.mark_bad();
+                                        return Err(err);
+                                    }
+                                };
+
+                                continue;
+                            }
+                        };
+
+                        match message {
                            Ok(Ok(message)) => message,
                            Ok(Err(err)) => {
                                // Client disconnected inside a transaction.
@@ -1125,27 +1099,6 @@ where
                match code {
                    // Query
                    'Q' => {
-                        if query_router.query_parser_enabled() {
-                            if let Ok(ast) = QueryRouter::parse(&message) {
-                                let plugin_result = query_router.execute_plugins(&ast).await;
-
-                                match plugin_result {
-                                    Ok(PluginOutput::Deny(error)) => {
-                                        error_response(&mut self.write, &error).await?;
-                                        continue;
-                                    }
-
-                                    Ok(PluginOutput::Intercept(result)) => {
-                                        write_all(&mut self.write, result).await?;
-                                        continue;
-                                    }
-
-                                    _ => (),
-                                };
-
-                                let _ = query_router.infer(&ast);
-                            }
-                        }
                        debug!("Sending query to server");

                        self.send_and_receive_loop(
@@ -1185,14 +1138,6 @@ where
                    // Parse
                    // The query with placeholders is here, e.g. `SELECT * FROM users WHERE email = $1 AND active = $2`.
                    'P' => {
-                        if query_router.query_parser_enabled() {
-                            if let Ok(ast) = QueryRouter::parse(&message) {
-                                if let Ok(output) = query_router.execute_plugins(&ast).await {
-                                    plugin_output = Some(output);
-                                }
-                            }
-                        }
-
                        self.buffer.put(&message[..]);
                    }

@@ -1221,26 +1166,13 @@ where

                    // Sync
                    // Frontend (client) is asking for the query result now.
-                    'S' => {
+                    'S' | 'H' => {
                        debug!("Sending query to server");

-                        match plugin_output {
-                            Some(PluginOutput::Deny(error)) => {
-                                error_response(&mut self.write, &error).await?;
-                                plugin_output = None;
-                                self.buffer.clear();
-                                continue;
-                            }
-
-                            Some(PluginOutput::Intercept(result)) => {
-                                write_all(&mut self.write, result).await?;
-                                plugin_output = None;
-                                self.buffer.clear();
-                                continue;
-                            }
-
-                            _ => (),
-                        };
+                        if code == 'H' {
+                            server.switch_async(true);
+                            debug!("Client requested flush, going async");
+                        }

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

@@ -1313,7 +1245,7 @@ where
                            .receive_server_message(server, &address, &pool, &self.stats.clone())
                            .await?;

-                        match write_all_flush(&mut self.write, &response).await {
+                        match write_all_half(&mut self.write, &response).await {
                            Ok(_) => (),
                            Err(err) => {
                                server.mark_bad();
@@ -1408,7 +1340,7 @@ where
                .receive_server_message(server, address, pool, client_stats)
                .await?;

-            match write_all_flush(&mut self.write, &response).await {
+            match write_all_half(&mut self.write, &response).await {
                Ok(_) => (),
                Err(err) => {
                    server.mark_bad();
--- a/src/config.rs
+++ b/src/config.rs
@@ -12,7 +12,6 @@ use std::sync::Arc;
 use tokio::fs::File;
 use tokio::io::AsyncReadExt;

-use crate::dns_cache::CachedResolver;
 use crate::errors::Error;
 use crate::pool::{ClientServerMap, ConnectionPool};
 use crate::sharding::ShardingFunction;
@@ -122,16 +121,6 @@ impl Default for Address {
    }
 }

-impl std::fmt::Display for Address {
-    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        write!(
-            f,
-            "[address: {}:{}][database: {}][user: {}]",
-            self.host, self.port, self.database, self.username
-        )
-    }
-}
-
 // We need to implement PartialEq by ourselves so we skip stats in the comparison
 impl PartialEq for Address {
    fn eq(&self, other: &Self) -> bool {
@@ -192,9 +181,7 @@ pub struct User {
    pub server_username: Option<String>,
    pub server_password: Option<String>,
    pub pool_size: u32,
-    pub min_pool_size: Option<u32>,
    pub pool_mode: Option<PoolMode>,
-    pub server_lifetime: Option<u64>,
    #[serde(default)] // 0
    pub statement_timeout: u64,
 }
@@ -207,34 +194,12 @@ impl Default for User {
            server_username: None,
            server_password: None,
            pool_size: 15,
-            min_pool_size: None,
            statement_timeout: 0,
            pool_mode: None,
-            server_lifetime: None,
        }
    }
 }

-impl User {
-    fn validate(&self) -> Result<(), Error> {
-        match self.min_pool_size {
-            Some(min_pool_size) => {
-                if min_pool_size > self.pool_size {
-                    error!(
-                        "min_pool_size of {} cannot be larger than pool_size of {}",
-                        min_pool_size, self.pool_size
-                    );
-                    return Err(Error::BadConfig);
-                }
-            }
-
-            None => (),
-        };
-
-        Ok(())
-    }
-}
-
 /// General configuration.
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub struct General {
@@ -245,8 +210,6 @@ pub struct General {
    pub port: u16,

    pub enable_prometheus_exporter: Option<bool>,
-
-    #[serde(default = "General::default_prometheus_exporter_port")]
    pub prometheus_exporter_port: i16,

    #[serde(default = "General::default_connect_timeout")]
@@ -268,12 +231,6 @@ pub struct General {
    #[serde(default)] // False
    pub log_client_disconnections: bool,

-    #[serde(default)] // False
-    pub dns_cache_enabled: bool,
-
-    #[serde(default = "General::default_dns_max_ttl")]
-    pub dns_max_ttl: u64,
-
    #[serde(default = "General::default_shutdown_timeout")]
    pub shutdown_timeout: u64,

@@ -289,9 +246,6 @@ pub struct General {
    #[serde(default = "General::default_idle_client_in_transaction_timeout")]
    pub idle_client_in_transaction_timeout: u64,

-    #[serde(default = "General::default_server_lifetime")]
-    pub server_lifetime: u64,
-
    #[serde(default = "General::default_worker_threads")]
    pub worker_threads: usize,

@@ -300,20 +254,9 @@ pub struct General {

    pub tls_certificate: Option<String>,
    pub tls_private_key: Option<String>,
-
-    #[serde(default)] // false
-    pub server_tls: bool,
-
-    #[serde(default)] // false
-    pub verify_server_certificate: bool,
-
    pub admin_username: String,
    pub admin_password: String,

-    #[serde(default = "General::default_validate_config")]
-    pub validate_config: bool,
-
-    // Support for auth query
    pub auth_query: Option<String>,
    pub auth_query_user: Option<String>,
    pub auth_query_password: Option<String>,
@@ -328,10 +271,6 @@ impl General {
        5432
    }

-    pub fn default_server_lifetime() -> u64 {
-        1000 * 60 * 60 * 24 // 24 hours
-    }
-
    pub fn default_connect_timeout() -> u64 {
        1000
    }
@@ -359,10 +298,6 @@ impl General {
        60000
    }

-    pub fn default_dns_max_ttl() -> u64 {
-        30
-    }
-
    pub fn default_healthcheck_timeout() -> u64 {
        1000
    }
@@ -382,14 +317,6 @@ impl General {
    pub fn default_idle_client_in_transaction_timeout() -> u64 {
        0
    }
-
-    pub fn default_validate_config() -> bool {
-        true
-    }
-
-    pub fn default_prometheus_exporter_port() -> i16 {
-        9930
-    }
 }

 impl Default for General {
@@ -413,19 +340,13 @@ impl Default for General {
            log_client_connections: false,
            log_client_disconnections: false,
            autoreload: None,
-            dns_cache_enabled: false,
-            dns_max_ttl: Self::default_dns_max_ttl(),
            tls_certificate: None,
            tls_private_key: None,
-            server_tls: false,
-            verify_server_certificate: false,
            admin_username: String::from("admin"),
            admin_password: String::from("admin"),
            auth_query: None,
            auth_query_user: None,
            auth_query_password: None,
-            server_lifetime: 1000 * 3600 * 24, // 24 hours,
-            validate_config: true,
        }
    }
 }
@@ -478,7 +399,6 @@ pub struct Pool {
    #[serde(default = "Pool::default_load_balancing_mode")]
    pub load_balancing_mode: LoadBalancingMode,

-    #[serde(default = "Pool::default_default_role")]
    pub default_role: String,

    #[serde(default)] // False
@@ -487,18 +407,10 @@ pub struct Pool {
    #[serde(default)] // False
    pub primary_reads_enabled: bool,

-    /// Maximum time to allow for establishing a new server connection.
    pub connect_timeout: Option<u64>,

-    /// Close idle connections that have been opened for longer than this.
    pub idle_timeout: Option<u64>,

-    /// Close server connections that have been opened for longer than this.
-    /// Only applied to idle connections. If the connection is actively used for
-    /// longer than this period, the pool will not interrupt it.
-    pub server_lifetime: Option<u64>,
-
-    #[serde(default = "Pool::default_sharding_function")]
    pub sharding_function: ShardingFunction,

    #[serde(default = "Pool::default_automatic_sharding_key")]
@@ -512,10 +424,6 @@ pub struct Pool {
    pub auth_query_user: Option<String>,
    pub auth_query_password: Option<String>,

-    #[serde(default = "Pool::default_cleanup_server_connections")]
-    pub cleanup_server_connections: bool,
-
-    pub plugins: Option<Plugins>,
    pub shards: BTreeMap<String, Shard>,
    pub users: BTreeMap<String, User>,
    // Note, don't put simple fields below these configs. There's a compatibility issue with TOML that makes it
@@ -548,18 +456,6 @@ impl Pool {
        None
    }

-    pub fn default_default_role() -> String {
-        "any".into()
-    }
-
-    pub fn default_sharding_function() -> ShardingFunction {
-        ShardingFunction::PgBigintHash
-    }
-
-    pub fn default_cleanup_server_connections() -> bool {
-        true
-    }
-
    pub fn validate(&mut self) -> Result<(), Error> {
        match self.default_role.as_ref() {
            "any" => (),
@@ -619,10 +515,6 @@ impl Pool {
            None => None,
        };

-        for (_, user) in &self.users {
-            user.validate()?;
-        }
-
        Ok(())
    }
 }
@@ -647,9 +539,6 @@ impl Default for Pool {
            auth_query: None,
            auth_query_user: None,
            auth_query_password: None,
-            server_lifetime: None,
-            plugins: None,
-            cleanup_server_connections: true,
        }
    }
 }
@@ -699,7 +588,7 @@ impl Shard {

        if primary_count > 1 {
            error!(
-                "Shard {} has more than one primary configured",
+                "Shard {} has more than on primary configured",
                self.database
            );
            return Err(Error::BadConfig);
@@ -728,76 +617,6 @@ impl Default for Shard {
    }
 }

-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, Hash, Eq)]
-pub struct Plugins {
-    pub intercept: Option<Intercept>,
-    pub table_access: Option<TableAccess>,
-    pub query_logger: Option<QueryLogger>,
-    pub prewarmer: Option<Prewarmer>,
-}
-
-impl std::fmt::Display for Plugins {
-    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        write!(
-            f,
-            "interceptor: {}, table_access: {}, query_logger: {}, prewarmer: {}",
-            self.intercept.is_some(),
-            self.table_access.is_some(),
-            self.query_logger.is_some(),
-            self.prewarmer.is_some(),
-        )
-    }
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, Hash, Eq)]
-pub struct Intercept {
-    pub enabled: bool,
-    pub queries: BTreeMap<String, Query>,
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, Hash, Eq)]
-pub struct TableAccess {
-    pub enabled: bool,
-    pub tables: Vec<String>,
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, Hash, Eq)]
-pub struct QueryLogger {
-    pub enabled: bool,
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, Hash, Eq)]
-pub struct Prewarmer {
-    pub enabled: bool,
-    pub queries: Vec<String>,
-}
-
-impl Intercept {
-    pub fn substitute(&mut self, db: &str, user: &str) {
-        for (_, query) in self.queries.iter_mut() {
-            query.substitute(db, user);
-            query.query = query.query.to_ascii_lowercase();
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, Hash, Eq)]
-pub struct Query {
-    pub query: String,
-    pub schema: Vec<Vec<String>>,
-    pub result: Vec<Vec<String>>,
-}
-
-impl Query {
-    pub fn substitute(&mut self, db: &str, user: &str) {
-        for col in self.result.iter_mut() {
-            for i in 0..col.len() {
-                col[i] = col[i].replace("${USER}", user).replace("${DATABASE}", db);
-            }
-        }
-    }
-}
-
 /// Configuration wrapper.
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub struct Config {
@@ -815,13 +634,7 @@ pub struct Config {
    #[serde(default = "Config::default_path")]
    pub path: String,

-    // General and global settings.
    pub general: General,
-
-    // Plugins that should run in all pools.
-    pub plugins: Option<Plugins>,
-
-    // Connection pools.
    pub pools: HashMap<String, Pool>,
 }

@@ -859,7 +672,6 @@ impl Default for Config {
            path: Self::default_path(),
            general: General::default(),
            pools: HashMap::default(),
-            plugins: None,
        }
    }
 }
@@ -979,10 +791,6 @@ impl Config {
        );
        info!("Shutdown timeout: {}ms", self.general.shutdown_timeout);
        info!("Healthcheck delay: {}ms", self.general.healthcheck_delay);
-        info!(
-            "Default max server lifetime: {}ms",
-            self.general.server_lifetime
-        );
        match self.general.tls_certificate.clone() {
            Some(tls_certificate) => {
                info!("TLS certificate: {}", tls_certificate);
@@ -1001,18 +809,6 @@ impl Config {
                info!("TLS support is disabled");
            }
        };
-        info!("Server TLS enabled: {}", self.general.server_tls);
-        info!(
-            "Server TLS certificate verification: {}",
-            self.general.verify_server_certificate
-        );
-        info!(
-            "Plugins: {}",
-            match self.plugins {
-                Some(ref plugins) => plugins.to_string(),
-                None => "not configured".into(),
-            }
-        );

        for (pool_name, pool_config) in &self.pools {
            // TODO: Make this output prettier (maybe a table?)
@@ -1071,38 +867,12 @@ impl Config {
                pool_name,
                pool_config.users.len()
            );
-            info!(
-                "[pool: {}] Max server lifetime: {}",
-                pool_name,
-                match pool_config.server_lifetime {
-                    Some(server_lifetime) => format!("{}ms", server_lifetime),
-                    None => "default".to_string(),
-                }
-            );
-            info!(
-                "[pool: {}] Cleanup server connections: {}",
-                pool_name, pool_config.cleanup_server_connections
-            );
-            info!(
-                "[pool: {}] Plugins: {}",
-                pool_name,
-                match pool_config.plugins {
-                    Some(ref plugins) => plugins.to_string(),
-                    None => "not configured".into(),
-                }
-            );

            for user in &pool_config.users {
                info!(
                    "[pool: {}][user: {}] Pool size: {}",
                    pool_name, user.1.username, user.1.pool_size,
                );
-                info!(
-                    "[pool: {}][user: {}] Minimum pool size: {}",
-                    pool_name,
-                    user.1.username,
-                    user.1.min_pool_size.unwrap_or(0)
-                );
                info!(
                    "[pool: {}][user: {}] Statement timeout: {}",
                    pool_name, user.1.username, user.1.statement_timeout
@@ -1116,15 +886,6 @@ impl Config {
                        None => pool_config.pool_mode.to_string(),
                    }
                );
-                info!(
-                    "[pool: {}][user: {}] Max server lifetime: {}",
-                    pool_name,
-                    user.1.username,
-                    match user.1.server_lifetime {
-                        Some(server_lifetime) => format!("{}ms", server_lifetime),
-                        None => "default".to_string(),
-                    }
-                );
            }
        }
    }
@@ -1135,13 +896,7 @@ impl Config {
            && (self.general.auth_query_user.is_none()
                || self.general.auth_query_password.is_none())
        {
-            error!(
-                "If auth_query is specified, \
-                you need to provide a value \
-                for `auth_query_user`, \
-                `auth_query_password`"
-            );
-
+            error!("If auth_query is specified, you need to provide a value for `auth_query_user`, `auth_query_password`");
            return Err(Error::BadConfig);
        }

@@ -1149,14 +904,7 @@ impl Config {
            if pool.auth_query.is_some()
                && (pool.auth_query_user.is_none() || pool.auth_query_password.is_none())
            {
-                error!(
-                    "Error in pool {{ {} }}. \
-                    If auth_query is specified, you need \
-                    to provide a value for `auth_query_user`, \
-                    `auth_query_password`",
-                    name
-                );
-
+                error!("Error in pool {{ {} }}. If auth_query is specified, you need to provide a value for `auth_query_user`, `auth_query_password`", name);
                return Err(Error::BadConfig);
            }

@@ -1166,13 +914,7 @@ impl Config {
                    || pool.auth_query_user.is_none())
                    && user_data.password.is_none()
                {
-                    error!(
-                        "Error in pool {{ {} }}. \
-                        You have to specify a user password \
-                        for every pool if auth_query is not specified",
-                        name
-                    );
-
+                    error!("Error in pool {{ {} }}. You have to specify a user password for every pool if auth_query is not specified", name);
                    return Err(Error::BadConfig);
                }
            }
@@ -1270,7 +1012,6 @@ pub async fn parse(path: &str) -> Result<(), Error> {

 pub async fn reload_config(client_server_map: ClientServerMap) -> Result<bool, Error> {
    let old_config = get_config();
-
    match parse(&old_config.path).await {
        Ok(()) => (),
        Err(err) => {
@@ -1278,18 +1019,14 @@ pub async fn reload_config(client_server_map: ClientServerMap) -> Result<bool, E
            return Err(Error::BadConfig);
        }
    };
-
    let new_config = get_config();

-    match CachedResolver::from_config().await {
-        Ok(_) => (),
-        Err(err) => error!("DNS cache reinitialization error: {:?}", err),
-    };
-
-    if old_config != new_config {
-        info!("Config changed, reloading");
+    if old_config.pools != new_config.pools {
+        info!("Pool configuration changed");
        ConnectionPool::from_config(client_server_map).await?;
        Ok(true)
+    } else if old_config != new_config {
+        Ok(true)
    } else {
        Ok(false)
    }
--- a/src/dns_cache.rs
+++ b/src/dns_cache.rs
@@ -1,410 +0,0 @@
-use crate::config::get_config;
-use crate::errors::Error;
-use arc_swap::ArcSwap;
-use log::{debug, error, info, warn};
-use once_cell::sync::Lazy;
-use std::collections::{HashMap, HashSet};
-use std::io;
-use std::net::IpAddr;
-use std::sync::Arc;
-use std::sync::RwLock;
-use tokio::time::{sleep, Duration};
-use trust_dns_resolver::error::{ResolveError, ResolveResult};
-use trust_dns_resolver::lookup_ip::LookupIp;
-use trust_dns_resolver::TokioAsyncResolver;
-
-/// Cached Resolver Globally available
-pub static CACHED_RESOLVER: Lazy<ArcSwap<CachedResolver>> =
-    Lazy::new(|| ArcSwap::from_pointee(CachedResolver::default()));
-
-// Ip addressed are returned as a set of addresses
-// so we can compare.
-#[derive(Clone, PartialEq, Debug)]
-pub struct AddrSet {
-    set: HashSet<IpAddr>,
-}
-
-impl AddrSet {
-    fn new() -> AddrSet {
-        AddrSet {
-            set: HashSet::new(),
-        }
-    }
-}
-
-impl From<LookupIp> for AddrSet {
-    fn from(lookup_ip: LookupIp) -> Self {
-        let mut addr_set = AddrSet::new();
-        for address in lookup_ip.iter() {
-            addr_set.set.insert(address);
-        }
-        addr_set
-    }
-}
-
-///
-/// A CachedResolver is a DNS resolution cache mechanism with customizable expiration time.
-///
-/// The system works as follows:
-///
-/// When a host is to be resolved, if we have not resolved it before, a new resolution is
-/// executed and stored in the internal cache. Concurrently, every `dns_max_ttl` time, the
-/// cache is refreshed.
-///
-/// # Example:
-///
-/// ```
-/// use pgcat::dns_cache::{CachedResolverConfig, CachedResolver};
-///
-/// # tokio_test::block_on(async {
-/// let config = CachedResolverConfig::default();
-/// let resolver = CachedResolver::new(config, None).await.unwrap();
-/// let addrset = resolver.lookup_ip("www.example.com.").await.unwrap();
-/// # })
-/// ```
-///
-/// // Now the ip resolution is stored in local cache and subsequent
-/// // calls will be returned from cache. Also, the cache is refreshed
-/// // and updated every 10 seconds.
-///
-/// // You can now check if an 'old' lookup differs from what it's currently
-/// // store in cache by using `has_changed`.
-/// resolver.has_changed("www.example.com.", addrset)
-#[derive(Default)]
-pub struct CachedResolver {
-    // The configuration of the cached_resolver.
-    config: CachedResolverConfig,
-
-    // This is the hash that contains the hash.
-    data: Option<RwLock<HashMap<String, AddrSet>>>,
-
-    // The resolver to be used for DNS queries.
-    resolver: Option<TokioAsyncResolver>,
-
-    // The RefreshLoop
-    refresh_loop: RwLock<Option<tokio::task::JoinHandle<()>>>,
-}
-
-///
-/// Configuration
-#[derive(Clone, Debug, Default, PartialEq)]
-pub struct CachedResolverConfig {
-    /// Amount of time in secods that a resolved dns address is considered stale.
-    dns_max_ttl: u64,
-
-    /// Enabled or disabled? (this is so we can reload config)
-    enabled: bool,
-}
-
-impl CachedResolverConfig {
-    fn new(dns_max_ttl: u64, enabled: bool) -> Self {
-        CachedResolverConfig {
-            dns_max_ttl,
-            enabled,
-        }
-    }
-}
-
-impl From<crate::config::Config> for CachedResolverConfig {
-    fn from(config: crate::config::Config) -> Self {
-        CachedResolverConfig::new(config.general.dns_max_ttl, config.general.dns_cache_enabled)
-    }
-}
-
-impl CachedResolver {
-    ///
-    /// Returns a new Arc<CachedResolver> based on passed configuration.
-    /// It also starts the loop that will refresh cache entries.
-    ///
-    /// # Arguments:
-    ///
-    /// * `config` - The `CachedResolverConfig` to be used to create the resolver.
-    ///
-    /// # Example:
-    ///
-    /// ```
-    /// use pgcat::dns_cache::{CachedResolverConfig, CachedResolver};
-    ///
-    /// # tokio_test::block_on(async {
-    /// let config = CachedResolverConfig::default();
-    /// let resolver = CachedResolver::new(config, None).await.unwrap();
-    /// # })
-    /// ```
-    ///
-    pub async fn new(
-        config: CachedResolverConfig,
-        data: Option<HashMap<String, AddrSet>>,
-    ) -> Result<Arc<Self>, io::Error> {
-        // Construct a new Resolver with default configuration options
-        let resolver = Some(TokioAsyncResolver::tokio_from_system_conf()?);
-
-        let data = if let Some(hash) = data {
-            Some(RwLock::new(hash))
-        } else {
-            Some(RwLock::new(HashMap::new()))
-        };
-
-        let instance = Arc::new(Self {
-            config,
-            resolver,
-            data,
-            refresh_loop: RwLock::new(None),
-        });
-
-        if instance.enabled() {
-            info!("Scheduling DNS refresh loop");
-            let refresh_loop = tokio::task::spawn({
-                let instance = instance.clone();
-                async move {
-                    instance.refresh_dns_entries_loop().await;
-                }
-            });
-            *(instance.refresh_loop.write().unwrap()) = Some(refresh_loop);
-        }
-
-        Ok(instance)
-    }
-
-    pub fn enabled(&self) -> bool {
-        self.config.enabled
-    }
-
-    // Schedules the refresher
-    async fn refresh_dns_entries_loop(&self) {
-        let resolver = TokioAsyncResolver::tokio_from_system_conf().unwrap();
-        let interval = Duration::from_secs(self.config.dns_max_ttl);
-        loop {
-            debug!("Begin refreshing cached DNS addresses.");
-            // To minimize the time we hold the lock, we first create
-            // an array with keys.
-            let mut hostnames: Vec<String> = Vec::new();
-            {
-                if let Some(ref data) = self.data {
-                    for hostname in data.read().unwrap().keys() {
-                        hostnames.push(hostname.clone());
-                    }
-                }
-            }
-
-            for hostname in hostnames.iter() {
-                let addrset = self
-                    .fetch_from_cache(hostname.as_str())
-                    .expect("Could not obtain expected address from cache, this should not happen");
-
-                match resolver.lookup_ip(hostname).await {
-                    Ok(lookup_ip) => {
-                        let new_addrset = AddrSet::from(lookup_ip);
-                        debug!(
-                            "Obtained address for host ({}) -> ({:?})",
-                            hostname, new_addrset
-                        );
-
-                        if addrset != new_addrset {
-                            debug!(
-                                "Addr changed from {:?} to {:?} updating cache.",
-                                addrset, new_addrset
-                            );
-                            self.store_in_cache(hostname, new_addrset);
-                        }
-                    }
-                    Err(err) => {
-                        error!(
-                            "There was an error trying to resolv {}: ({}).",
-                            hostname, err
-                        );
-                    }
-                }
-            }
-            debug!("Finished refreshing cached DNS addresses.");
-            sleep(interval).await;
-        }
-    }
-
-    /// Returns a `AddrSet` given the specified hostname.
-    ///
-    /// This method first tries to fetch the value from the cache, if it misses
-    /// then it is resolved and stored in the cache. TTL from records is ignored.
-    ///
-    /// # Arguments
-    ///
-    /// * `host`      - A string slice referencing the hostname to be resolved.
-    ///
-    /// # Example:
-    ///
-    /// ```
-    /// use pgcat::dns_cache::{CachedResolverConfig, CachedResolver};
-    ///
-    /// # tokio_test::block_on(async {
-    /// let config = CachedResolverConfig::default();
-    /// let resolver = CachedResolver::new(config, None).await.unwrap();
-    /// let response = resolver.lookup_ip("www.google.com.");
-    /// # })
-    /// ```
-    ///
-    pub async fn lookup_ip(&self, host: &str) -> ResolveResult<AddrSet> {
-        debug!("Lookup up {} in cache", host);
-        match self.fetch_from_cache(host) {
-            Some(addr_set) => {
-                debug!("Cache hit!");
-                Ok(addr_set)
-            }
-            None => {
-                debug!("Not found, executing a dns query!");
-                if let Some(ref resolver) = self.resolver {
-                    let addr_set = AddrSet::from(resolver.lookup_ip(host).await?);
-                    debug!("Obtained: {:?}", addr_set);
-                    self.store_in_cache(host, addr_set.clone());
-                    Ok(addr_set)
-                } else {
-                    Err(ResolveError::from("No resolver available"))
-                }
-            }
-        }
-    }
-
-    //
-    // Returns true if the stored host resolution differs from the AddrSet passed.
-    pub fn has_changed(&self, host: &str, addr_set: &AddrSet) -> bool {
-        if let Some(fetched_addr_set) = self.fetch_from_cache(host) {
-            return fetched_addr_set != *addr_set;
-        }
-        false
-    }
-
-    // Fetches an AddrSet from the inner cache adquiring the read lock.
-    fn fetch_from_cache(&self, key: &str) -> Option<AddrSet> {
-        if let Some(ref hash) = self.data {
-            if let Some(addr_set) = hash.read().unwrap().get(key) {
-                return Some(addr_set.clone());
-            }
-        }
-        None
-    }
-
-    // Sets up the global CACHED_RESOLVER static variable so we can globally use DNS
-    // cache.
-    pub async fn from_config() -> Result<(), Error> {
-        let cached_resolver = CACHED_RESOLVER.load();
-        let desired_config = CachedResolverConfig::from(get_config());
-
-        if cached_resolver.config != desired_config {
-            if let Some(ref refresh_loop) = *(cached_resolver.refresh_loop.write().unwrap()) {
-                warn!("Killing Dnscache refresh loop as its configuration is being reloaded");
-                refresh_loop.abort()
-            }
-            let new_resolver = if let Some(ref data) = cached_resolver.data {
-                let data = Some(data.read().unwrap().clone());
-                CachedResolver::new(desired_config, data).await
-            } else {
-                CachedResolver::new(desired_config, None).await
-            };
-
-            match new_resolver {
-                Ok(ok) => {
-                    CACHED_RESOLVER.store(ok);
-                    Ok(())
-                }
-                Err(err) => {
-                    let message = format!("Error setting up cached_resolver. Error: {:?}, will continue without this feature.", err);
-                    Err(Error::DNSCachedError(message))
-                }
-            }
-        } else {
-            Ok(())
-        }
-    }
-
-    // Stores the AddrSet in cache adquiring the write lock.
-    fn store_in_cache(&self, host: &str, addr_set: AddrSet) {
-        if let Some(ref data) = self.data {
-            data.write().unwrap().insert(host.to_string(), addr_set);
-        } else {
-            error!("Could not insert, Hash not initialized");
-        }
-    }
-}
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use trust_dns_resolver::error::ResolveError;
-
-    #[tokio::test]
-    async fn new() {
-        let config = CachedResolverConfig {
-            dns_max_ttl: 10,
-            enabled: true,
-        };
-        let resolver = CachedResolver::new(config, None).await;
-        assert!(resolver.is_ok());
-    }
-
-    #[tokio::test]
-    async fn lookup_ip() {
-        let config = CachedResolverConfig {
-            dns_max_ttl: 10,
-            enabled: true,
-        };
-        let resolver = CachedResolver::new(config, None).await.unwrap();
-        let response = resolver.lookup_ip("www.google.com.").await;
-        assert!(response.is_ok());
-    }
-
-    #[tokio::test]
-    async fn has_changed() {
-        let config = CachedResolverConfig {
-            dns_max_ttl: 10,
-            enabled: true,
-        };
-        let resolver = CachedResolver::new(config, None).await.unwrap();
-        let hostname = "www.google.com.";
-        let response = resolver.lookup_ip(hostname).await;
-        let addr_set = response.unwrap();
-        assert!(!resolver.has_changed(hostname, &addr_set));
-    }
-
-    #[tokio::test]
-    async fn unknown_host() {
-        let config = CachedResolverConfig {
-            dns_max_ttl: 10,
-            enabled: true,
-        };
-        let resolver = CachedResolver::new(config, None).await.unwrap();
-        let hostname = "www.idontexists.";
-        let response = resolver.lookup_ip(hostname).await;
-        assert!(matches!(response, Err(ResolveError { .. })));
-    }
-
-    #[tokio::test]
-    async fn incorrect_address() {
-        let config = CachedResolverConfig {
-            dns_max_ttl: 10,
-            enabled: true,
-        };
-        let resolver = CachedResolver::new(config, None).await.unwrap();
-        let hostname = "w  ww.idontexists.";
-        let response = resolver.lookup_ip(hostname).await;
-        assert!(matches!(response, Err(ResolveError { .. })));
-        assert!(!resolver.has_changed(hostname, &AddrSet::new()));
-    }
-
-    #[tokio::test]
-    // Ok, this test is based on the fact that google does DNS RR
-    // and does not responds with every available ip everytime, so
-    // if I cache here, it will miss after one cache iteration or two.
-    async fn thread() {
-        let config = CachedResolverConfig {
-            dns_max_ttl: 10,
-            enabled: true,
-        };
-        let resolver = CachedResolver::new(config, None).await.unwrap();
-        let hostname = "www.google.com.";
-        let response = resolver.lookup_ip(hostname).await;
-        let addr_set = response.unwrap();
-        assert!(!resolver.has_changed(hostname, &addr_set));
-        let resolver_for_refresher = resolver.clone();
-        let _thread_handle = tokio::task::spawn(async move {
-            resolver_for_refresher.refresh_dns_entries_loop().await;
-        });
-        assert!(!resolver.has_changed(hostname, &addr_set));
-    }
-}
--- a/src/errors.rs
+++ b/src/errors.rs
@@ -1,7 +1,7 @@
 //! Errors.

 /// Various errors.
-#[derive(Debug, PartialEq, Clone)]
+#[derive(Debug, PartialEq)]
 pub enum Error {
    SocketError(String),
    ClientSocketError(String, ClientIdentifier),
@@ -19,13 +19,10 @@ pub enum Error {
    ClientError(String),
    TlsError,
    StatementTimeout,
-    DNSCachedError(String),
    ShuttingDown,
    ParseBytesError(String),
    AuthError(String),
    AuthPassthroughError(String),
-    UnsupportedStatement,
-    QueryRouterParserError(String),
 }

 #[derive(Clone, PartialEq, Debug)]
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,17 +1,11 @@
-pub mod admin;
 pub mod auth_passthrough;
-pub mod client;
 pub mod config;
 pub mod constants;
-pub mod dns_cache;
 pub mod errors;
 pub mod messages;
 pub mod mirrors;
 pub mod multi_logger;
-pub mod plugins;
 pub mod pool;
-pub mod prometheus;
-pub mod query_router;
 pub mod scram;
 pub mod server;
 pub mod sharding;
--- a/src/main.rs
+++ b/src/main.rs
@@ -36,7 +36,6 @@ extern crate sqlparser;
 extern crate tokio;
 extern crate tokio_rustls;
 extern crate toml;
-extern crate trust_dns_resolver;

 #[cfg(not(target_env = "msvc"))]
 use jemallocator::Jemalloc;
@@ -61,19 +60,36 @@ use std::str::FromStr;
 use std::sync::Arc;
 use tokio::sync::broadcast;

-use pgcat::config::{get_config, reload_config, VERSION};
-use pgcat::dns_cache;
-use pgcat::messages::configure_socket;
-use pgcat::pool::{ClientServerMap, ConnectionPool};
-use pgcat::prometheus::start_metric_server;
-use pgcat::stats::{Collector, Reporter, REPORTER};
+mod admin;
+mod auth_passthrough;
+mod client;
+mod config;
+mod constants;
+mod errors;
+mod messages;
+mod mirrors;
+mod multi_logger;
+mod pool;
+mod prometheus;
+mod query_router;
+mod scram;
+mod server;
+mod sharding;
+mod stats;
+mod tls;
+
+use crate::config::{get_config, reload_config, VERSION};
+use crate::messages::configure_socket;
+use crate::pool::{ClientServerMap, ConnectionPool};
+use crate::prometheus::start_metric_server;
+use crate::stats::{Collector, Reporter, REPORTER};

 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    pgcat::multi_logger::MultiLogger::init().unwrap();
+    multi_logger::MultiLogger::init().unwrap();

    info!("Welcome to PgCat! Meow. (Version {})", VERSION);

-    if !pgcat::query_router::QueryRouter::setup() {
+    if !query_router::QueryRouter::setup() {
        error!("Could not setup query router");
        std::process::exit(exitcode::CONFIG);
    }
@@ -91,7 +107,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
        let runtime = Builder::new_multi_thread().worker_threads(1).build()?;

        runtime.block_on(async {
-            match pgcat::config::parse(&config_file).await {
+            match config::parse(&config_file).await {
                Ok(_) => (),
                Err(err) => {
                    error!("Config parse error: {:?}", err);
@@ -150,12 +166,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
        // Statistics reporting.
        REPORTER.store(Arc::new(Reporter::default()));

-        // Starts (if enabled) dns cache before pools initialization
-        match dns_cache::CachedResolver::from_config().await {
-                Ok(_) => (),
-                Err(err) => error!("DNS cache initialization error: {:?}", err),
-        };
-
        // Connection pool that allows to query all shards and replicas.
        match ConnectionPool::from_config(client_server_map.clone()).await {
            Ok(_) => (),
@@ -285,7 +295,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
                    tokio::task::spawn(async move {
                        let start = chrono::offset::Utc::now().naive_utc();

-                        match pgcat::client::client_entrypoint(
+                        match client::client_entrypoint(
                            socket,
                            client_server_map,
                            shutdown_rx,
@@ -316,7 +326,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {

                            Err(err) => {
                                match err {
-                                    pgcat::errors::Error::ClientBadStartup => debug!("Client disconnected with error {:?}", err),
+                                    errors::Error::ClientBadStartup => debug!("Client disconnected with error {:?}", err),
                                    _ => warn!("Client disconnected with error {:?}", err),
                                }

--- a/src/messages.rs
+++ b/src/messages.rs
@@ -20,10 +20,6 @@ pub enum DataType {
    Text,
    Int4,
    Numeric,
-    Bool,
-    Oid,
-    AnyArray,
-    Any,
 }

 impl From<&DataType> for i32 {
@@ -32,10 +28,6 @@ impl From<&DataType> for i32 {
            DataType::Text => 25,
            DataType::Int4 => 23,
            DataType::Numeric => 1700,
-            DataType::Bool => 16,
-            DataType::Oid => 26,
-            DataType::AnyArray => 2277,
-            DataType::Any => 2276,
        }
    }
 }
@@ -124,10 +116,7 @@ where

 /// Send the startup packet the server. We're pretending we're a Pg client.
 /// This tells the server which user we are and what database we want.
-pub async fn startup<S>(stream: &mut S, user: &str, database: &str) -> Result<(), Error>
-where
-    S: tokio::io::AsyncWrite + std::marker::Unpin,
-{
+pub async fn startup(stream: &mut TcpStream, user: &str, database: &str) -> Result<(), Error> {
    let mut bytes = BytesMut::with_capacity(25);

    bytes.put_i32(196608); // Protocol number
@@ -161,21 +150,6 @@ where
    }
 }

-pub async fn ssl_request(stream: &mut TcpStream) -> Result<(), Error> {
-    let mut bytes = BytesMut::with_capacity(12);
-
-    bytes.put_i32(8);
-    bytes.put_i32(80877103);
-
-    match stream.write_all(&bytes).await {
-        Ok(_) => Ok(()),
-        Err(err) => Err(Error::SocketError(format!(
-            "Error writing SSLRequest to server socket - Error: {:?}",
-            err
-        ))),
-    }
-}
-
 /// Parse the params the server sends as a key/value format.
 pub fn parse_params(mut bytes: BytesMut) -> Result<HashMap<String, String>, Error> {
    let mut result = HashMap::new();
@@ -451,10 +425,6 @@ pub fn row_description(columns: &Vec<(&str, DataType)>) -> BytesMut {
            DataType::Text => -1,
            DataType::Int4 => 4,
            DataType::Numeric => -1,
-            DataType::Bool => 1,
-            DataType::Oid => 4,
-            DataType::AnyArray => -1,
-            DataType::Any => -1,
        };

        row_desc.put_i16(type_size);
@@ -493,29 +463,6 @@ pub fn data_row(row: &Vec<String>) -> BytesMut {
    res
 }

-pub fn data_row_nullable(row: &Vec<Option<String>>) -> BytesMut {
-    let mut res = BytesMut::new();
-    let mut data_row = BytesMut::new();
-
-    data_row.put_i16(row.len() as i16);
-
-    for column in row {
-        if let Some(column) = column {
-            let column = column.as_bytes();
-            data_row.put_i32(column.len() as i32);
-            data_row.put_slice(column);
-        } else {
-            data_row.put_i32(-1 as i32);
-        }
-    }
-
-    res.put_u8(b'D');
-    res.put_i32(data_row.len() as i32 + 4);
-    res.put(data_row);
-
-    res
-}
-
 /// Create a CommandComplete message.
 pub fn command_complete(command: &str) -> BytesMut {
    let cmd = BytesMut::from(format!("{}\0", command).as_bytes());
@@ -558,29 +505,6 @@ where
    }
 }

-pub async fn write_all_flush<S>(stream: &mut S, buf: &[u8]) -> Result<(), Error>
-where
-    S: tokio::io::AsyncWrite + std::marker::Unpin,
-{
-    match stream.write_all(buf).await {
-        Ok(_) => match stream.flush().await {
-            Ok(_) => Ok(()),
-            Err(err) => {
-                return Err(Error::SocketError(format!(
-                    "Error flushing socket - Error: {:?}",
-                    err
-                )))
-            }
-        },
-        Err(err) => {
-            return Err(Error::SocketError(format!(
-                "Error writing to socket - Error: {:?}",
-                err
-            )))
-        }
-    }
-}
-
 /// Read a complete message from the socket.
 pub async fn read_message<S>(stream: &mut S) -> Result<BytesMut, Error>
 where
--- a/src/mirrors.rs
+++ b/src/mirrors.rs
@@ -43,8 +43,6 @@ impl MirroredClient {
            ClientServerMap::default(),
            Arc::new(PoolStats::new(identifier, cfg.clone())),
            Arc::new(RwLock::new(None)),
-            None,
-            true,
        );

        Pool::builder()
--- a/src/plugins/intercept.rs
+++ b/src/plugins/intercept.rs
@@ -1,120 +0,0 @@
-//! The intercept plugin.
-//!
-//! It intercepts queries and returns fake results.
-
-use async_trait::async_trait;
-use bytes::{BufMut, BytesMut};
-use serde::{Deserialize, Serialize};
-use sqlparser::ast::Statement;
-
-use log::debug;
-
-use crate::{
-    config::Intercept as InterceptConfig,
-    errors::Error,
-    messages::{command_complete, data_row_nullable, row_description, DataType},
-    plugins::{Plugin, PluginOutput},
-    query_router::QueryRouter,
-};
-
-// TODO: use these structs for deserialization
-#[derive(Serialize, Deserialize)]
-pub struct Rule {
-    query: String,
-    schema: Vec<Column>,
-    result: Vec<Vec<String>>,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct Column {
-    name: String,
-    data_type: String,
-}
-
-/// The intercept plugin.
-pub struct Intercept<'a> {
-    pub enabled: bool,
-    pub config: &'a InterceptConfig,
-}
-
-#[async_trait]
-impl<'a> Plugin for Intercept<'a> {
-    async fn run(
-        &mut self,
-        query_router: &QueryRouter,
-        ast: &Vec<Statement>,
-    ) -> Result<PluginOutput, Error> {
-        if !self.enabled || ast.is_empty() {
-            return Ok(PluginOutput::Allow);
-        }
-
-        let mut config = self.config.clone();
-        config.substitute(
-            &query_router.pool_settings().db,
-            &query_router.pool_settings().user.username,
-        );
-
-        let mut result = BytesMut::new();
-
-        for q in ast {
-            // Normalization
-            let q = q.to_string().to_ascii_lowercase();
-
-            for (_, target) in config.queries.iter() {
-                if target.query.as_str() == q {
-                    debug!("Intercepting query: {}", q);
-
-                    let rd = target
-                        .schema
-                        .iter()
-                        .map(|row| {
-                            let name = &row[0];
-                            let data_type = &row[1];
-                            (
-                                name.as_str(),
-                                match data_type.as_str() {
-                                    "text" => DataType::Text,
-                                    "anyarray" => DataType::AnyArray,
-                                    "oid" => DataType::Oid,
-                                    "bool" => DataType::Bool,
-                                    "int4" => DataType::Int4,
-                                    _ => DataType::Any,
-                                },
-                            )
-                        })
-                        .collect::<Vec<(&str, DataType)>>();
-
-                    result.put(row_description(&rd));
-
-                    target.result.iter().for_each(|row| {
-                        let row = row
-                            .iter()
-                            .map(|s| {
-                                let s = s.as_str().to_string();
-
-                                if s == "" {
-                                    None
-                                } else {
-                                    Some(s)
-                                }
-                            })
-                            .collect::<Vec<Option<String>>>();
-                        result.put(data_row_nullable(&row));
-                    });
-
-                    result.put(command_complete("SELECT"));
-                }
-            }
-        }
-
-        if !result.is_empty() {
-            result.put_u8(b'Z');
-            result.put_i32(5);
-            result.put_u8(b'I');
-
-            return Ok(PluginOutput::Intercept(result));
-        } else {
-            Ok(PluginOutput::Allow)
-        }
-    }
-}
--- a/src/plugins/mod.rs
+++ b/src/plugins/mod.rs
@@ -1,44 +0,0 @@
-//! The plugin ecosystem.
-//!
-//! Currently plugins only grant access or deny access to the database for a particual query.
-//! Example use cases:
-//!   - block known bad queries
-//!   - block access to system catalogs
-//!   - block dangerous modifications like `DROP TABLE`
-//!   - etc
-//!
-
-pub mod intercept;
-pub mod prewarmer;
-pub mod query_logger;
-pub mod table_access;
-
-use crate::{errors::Error, query_router::QueryRouter};
-use async_trait::async_trait;
-use bytes::BytesMut;
-use sqlparser::ast::Statement;
-
-pub use intercept::Intercept;
-pub use query_logger::QueryLogger;
-pub use table_access::TableAccess;
-
-#[derive(Clone, Debug, PartialEq)]
-pub enum PluginOutput {
-    Allow,
-    Deny(String),
-    Overwrite(Vec<Statement>),
-    Intercept(BytesMut),
-}
-
-#[async_trait]
-pub trait Plugin {
-    // Run before the query is sent to the server.
-    async fn run(
-        &mut self,
-        query_router: &QueryRouter,
-        ast: &Vec<Statement>,
-    ) -> Result<PluginOutput, Error>;
-
-    // TODO: run after the result is returned
-    // async fn callback(&mut self, query_router: &QueryRouter);
-}
--- a/src/plugins/prewarmer.rs
+++ b/src/plugins/prewarmer.rs
@@ -1,28 +0,0 @@
-//! Prewarm new connections before giving them to the client.
-use crate::{errors::Error, server::Server};
-use log::info;
-
-pub struct Prewarmer<'a> {
-    pub enabled: bool,
-    pub server: &'a mut Server,
-    pub queries: &'a Vec<String>,
-}
-
-impl<'a> Prewarmer<'a> {
-    pub async fn run(&mut self) -> Result<(), Error> {
-        if !self.enabled {
-            return Ok(());
-        }
-
-        for query in self.queries {
-            info!(
-                "{} Prewarning with query: `{}`",
-                self.server.address(),
-                query
-            );
-            self.server.query(&query).await?;
-        }
-
-        Ok(())
-    }
-}
--- a/src/plugins/query_logger.rs
+++ b/src/plugins/query_logger.rs
@@ -1,38 +0,0 @@
-//! Log all queries to stdout (or somewhere else, why not).
-
-use crate::{
-    errors::Error,
-    plugins::{Plugin, PluginOutput},
-    query_router::QueryRouter,
-};
-use async_trait::async_trait;
-use log::info;
-use sqlparser::ast::Statement;
-
-pub struct QueryLogger<'a> {
-    pub enabled: bool,
-    pub user: &'a str,
-    pub db: &'a str,
-}
-
-#[async_trait]
-impl<'a> Plugin for QueryLogger<'a> {
-    async fn run(
-        &mut self,
-        _query_router: &QueryRouter,
-        ast: &Vec<Statement>,
-    ) -> Result<PluginOutput, Error> {
-        if !self.enabled {
-            return Ok(PluginOutput::Allow);
-        }
-
-        let query = ast
-            .iter()
-            .map(|q| q.to_string())
-            .collect::<Vec<String>>()
-            .join("; ");
-        info!("[pool: {}][user: {}] {}", self.user, self.db, query);
-
-        Ok(PluginOutput::Allow)
-    }
-}
--- a/src/plugins/table_access.rs
+++ b/src/plugins/table_access.rs
@@ -1,59 +0,0 @@
-//! This query router plugin will check if the user can access a particular
-//! table as part of their query. If they can't, the query will not be routed.
-
-use async_trait::async_trait;
-use sqlparser::ast::{visit_relations, Statement};
-
-use crate::{
-    errors::Error,
-    plugins::{Plugin, PluginOutput},
-    query_router::QueryRouter,
-};
-
-use log::debug;
-
-use core::ops::ControlFlow;
-
-pub struct TableAccess<'a> {
-    pub enabled: bool,
-    pub tables: &'a Vec<String>,
-}
-
-#[async_trait]
-impl<'a> Plugin for TableAccess<'a> {
-    async fn run(
-        &mut self,
-        _query_router: &QueryRouter,
-        ast: &Vec<Statement>,
-    ) -> Result<PluginOutput, Error> {
-        if !self.enabled {
-            return Ok(PluginOutput::Allow);
-        }
-
-        let mut found = None;
-
-        visit_relations(ast, |relation| {
-            let relation = relation.to_string();
-            let parts = relation.split(".").collect::<Vec<&str>>();
-            let table_name = parts.last().unwrap();
-
-            if self.tables.contains(&table_name.to_string()) {
-                found = Some(table_name.to_string());
-                ControlFlow::<()>::Break(())
-            } else {
-                ControlFlow::<()>::Continue(())
-            }
-        });
-
-        if let Some(found) = found {
-            debug!("Blocking access to table \"{}\"", found);
-
-            Ok(PluginOutput::Deny(format!(
-                "permission for table \"{}\" denied",
-                found
-            )))
-        } else {
-            Ok(PluginOutput::Allow)
-        }
-    }
-}
--- a/src/pool.rs
+++ b/src/pool.rs
@@ -17,13 +17,10 @@ use std::sync::{
 use std::time::Instant;
 use tokio::sync::Notify;

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

 use crate::auth_passthrough::AuthPassthrough;
-use crate::plugins::prewarmer;
 use crate::server::Server;
 use crate::sharding::ShardingFunction;
 use crate::stats::{AddressStats, ClientStats, PoolStats, ServerStats};
@@ -64,8 +61,6 @@ pub struct PoolIdentifier {
    pub user: String,
 }

-static POOL_REAPER_RATE: u64 = 30_000; // 30 seconds by default
-
 impl PoolIdentifier {
    /// Create a new user/pool identifier.
    pub fn new(db: &str, user: &str) -> PoolIdentifier {
@@ -96,7 +91,6 @@ pub struct PoolSettings {

    // Connecting user.
    pub user: User,
-    pub db: String,

    // Default server role to connect to.
    pub default_role: Option<Role>,
@@ -135,9 +129,6 @@ pub struct PoolSettings {
    pub auth_query: Option<String>,
    pub auth_query_user: Option<String>,
    pub auth_query_password: Option<String>,
-
-    /// Plugins
-    pub plugins: Option<Plugins>,
 }

 impl Default for PoolSettings {
@@ -147,7 +138,6 @@ impl Default for PoolSettings {
            load_balancing_mode: LoadBalancingMode::Random,
            shards: 1,
            user: User::default(),
-            db: String::default(),
            default_role: None,
            query_parser_enabled: false,
            primary_reads_enabled: true,
@@ -162,7 +152,6 @@ impl Default for PoolSettings {
            auth_query: None,
            auth_query_user: None,
            auth_query_password: None,
-            plugins: None,
        }
    }
 }
@@ -202,7 +191,6 @@ pub struct ConnectionPool {
    paused: Arc<AtomicBool>,
    paused_waiter: Arc<Notify>,

-    /// Statistics.
    pub stats: Arc<PoolStats>,

    /// AuthInfo
@@ -323,34 +311,21 @@ impl ConnectionPool {

                        if let Some(apt) = &auth_passthrough {
                            match apt.fetch_hash(&address).await {
-                                Ok(ok) => {
-                                    if let Some(ref pool_auth_hash_value) = *(pool_auth_hash.read())
-                                    {
-                                        if ok != *pool_auth_hash_value {
-                                            warn!(
-                                                "Hash is not the same across shards \
-                                                of the same pool, client auth will \
-                                                be done using last obtained hash. \
-                                                Server: {}:{}, Database: {}",
-                                                server.host, server.port, shard.database,
-                                            );
-                                        }
-                                    }
-
-                                    debug!("Hash obtained for {:?}", address);
-
-                                    {
-                                        let mut pool_auth_hash = pool_auth_hash.write();
-                                        *pool_auth_hash = Some(ok.clone());
-                                    }
-                                }
-                                Err(err) => warn!(
-                                    "Could not obtain password hashes \
-                                        using auth_query config, ignoring. \
-                                        Error: {:?}",
-                                    err,
-                                ),
-                            }
+				Ok(ok) => {
+				    if let Some(ref pool_auth_hash_value) = *(pool_auth_hash.read()) {
+					if ok != *pool_auth_hash_value {
+					    warn!("Hash is not the same across shards of the same pool, client auth will \
+						   be done using last obtained hash. Server: {}:{}, Database: {}", server.host, server.port, shard.database);
+					}
+				    }
+				    debug!("Hash obtained for {:?}", address);
+				    {
+					let mut pool_auth_hash = pool_auth_hash.write();
+					*pool_auth_hash = Some(ok.clone());
+				    }
+				},
+				Err(err) => warn!("Could not obtain password hashes using auth_query config, ignoring. Error: {:?}", err),
+			    }
                        }

                        let manager = ServerPool::new(
@@ -360,11 +335,6 @@ impl ConnectionPool {
                            client_server_map.clone(),
                            pool_stats.clone(),
                            pool_auth_hash.clone(),
-                            match pool_config.plugins {
-                                Some(ref plugins) => Some(plugins.clone()),
-                                None => config.plugins.clone(),
-                            },
-                            pool_config.cleanup_server_connections,
                        );

                        let connect_timeout = match pool_config.connect_timeout {
@@ -377,38 +347,14 @@ impl ConnectionPool {
                            None => config.general.idle_timeout,
                        };

-                        let server_lifetime = match user.server_lifetime {
-                            Some(server_lifetime) => server_lifetime,
-                            None => match pool_config.server_lifetime {
-                                Some(server_lifetime) => server_lifetime,
-                                None => config.general.server_lifetime,
-                            },
-                        };
-
-                        let reaper_rate = *vec![idle_timeout, server_lifetime, POOL_REAPER_RATE]
-                            .iter()
-                            .min()
-                            .unwrap();
-
-                        debug!(
-                            "[pool: {}][user: {}] Pool reaper rate: {}ms",
-                            pool_name, user.username, reaper_rate
-                        );
-
                        let pool = Pool::builder()
                            .max_size(user.pool_size)
-                            .min_idle(user.min_pool_size)
                            .connection_timeout(std::time::Duration::from_millis(connect_timeout))
                            .idle_timeout(Some(std::time::Duration::from_millis(idle_timeout)))
-                            .max_lifetime(Some(std::time::Duration::from_millis(server_lifetime)))
-                            .reaper_rate(std::time::Duration::from_millis(reaper_rate))
-                            .test_on_check_out(false);
-
-                        let pool = if config.general.validate_config {
-                            pool.build(manager).await?
-                        } else {
-                            pool.build_unchecked(manager)
-                        };
+                            .test_on_check_out(false)
+                            .build(manager)
+                            .await
+                            .unwrap();

                        pools.push(pool);
                        servers.push(address);
@@ -444,7 +390,6 @@ impl ConnectionPool {
                        // shards: pool_config.shards.clone(),
                        shards: shard_ids.len(),
                        user: user.clone(),
-                        db: pool_name.clone(),
                        default_role: match pool_config.default_role.as_str() {
                            "any" => None,
                            "replica" => Some(Role::Replica),
@@ -470,10 +415,6 @@ impl ConnectionPool {
                        auth_query: pool_config.auth_query.clone(),
                        auth_query_user: pool_config.auth_query_user.clone(),
                        auth_query_password: pool_config.auth_query_password.clone(),
-                        plugins: match pool_config.plugins {
-                            Some(ref plugins) => Some(plugins.clone()),
-                            None => config.plugins.clone(),
-                        },
                    },
                    validated: Arc::new(AtomicBool::new(false)),
                    paused: Arc::new(AtomicBool::new(false)),
@@ -483,12 +424,10 @@ impl ConnectionPool {
                // Connect to the servers to make sure pool configuration is valid
                // before setting it globally.
                // Do this async and somewhere else, we don't have to wait here.
-                if config.general.validate_config {
-                    let mut validate_pool = pool.clone();
-                    tokio::task::spawn(async move {
-                        let _ = validate_pool.validate().await;
-                    });
-                }
+                let mut validate_pool = pool.clone();
+                tokio::task::spawn(async move {
+                    let _ = validate_pool.validate().await;
+                });

                // There is one pool per database/user pair.
                new_pools.insert(PoolIdentifier::new(pool_name, &user.username), pool);
@@ -639,10 +578,7 @@ impl ConnectionPool {
            {
                Ok(conn) => conn,
                Err(err) => {
-                    error!(
-                        "Connection checkout error for instance {:?}, error: {:?}",
-                        address, err
-                    );
+                    error!("Banning instance {:?}, error: {:?}", address, err);
                    self.ban(address, BanReason::FailedCheckout, Some(client_stats));
                    address.stats.error();
                    client_stats.idle();
@@ -718,7 +654,7 @@ impl ConnectionPool {
                // Health check failed.
                Err(err) => {
                    error!(
-                        "Failed health check on instance {:?}, error: {:?}",
+                        "Banning instance {:?} because of failed health check, {:?}",
                        address, err
                    );
                }
@@ -727,7 +663,7 @@ impl ConnectionPool {
            // Health check timed out.
            Err(err) => {
                error!(
-                    "Health check timeout on instance {:?}, error: {:?}",
+                    "Banning instance {:?} because of health check timeout, {:?}",
                    address, err
                );
            }
@@ -749,16 +685,13 @@ impl ConnectionPool {
            return;
        }

-        error!("Banning instance {:?}, reason: {:?}", address, reason);
-
        let now = chrono::offset::Utc::now().naive_utc();
        let mut guard = self.banlist.write();
-
+        error!("Banning {:?}", address);
        if let Some(client_info) = client_info {
            client_info.ban_error();
            address.stats.error();
        }
-
        guard[address.shard].insert(address.clone(), (reason, now));
    }

@@ -844,6 +777,7 @@ impl ConnectionPool {
        self.databases.len()
    }

+    /// Retrieve all bans for all servers.
    pub fn get_bans(&self) -> Vec<(Address, (BanReason, NaiveDateTime))> {
        let mut bans: Vec<(Address, (BanReason, NaiveDateTime))> = Vec::new();
        let guard = self.banlist.read();
@@ -855,7 +789,7 @@ impl ConnectionPool {
        return bans;
    }

-    /// Get the address from the host url
+    /// Get the address from the host url.
    pub fn get_addresses_from_host(&self, host: &str) -> Vec<Address> {
        let mut addresses = Vec::new();
        for shard in 0..self.shards() {
@@ -894,10 +828,13 @@ impl ConnectionPool {
        &self.addresses[shard][server]
    }

+    /// Get server settings retrieved at connection setup.
    pub fn server_info(&self) -> BytesMut {
        self.server_info.read().clone()
    }

+    /// Calculate how many used connections in the pool
+    /// for the given server.
    fn busy_connection_count(&self, address: &Address) -> u32 {
        let state = self.pool_state(address.shard, address.address_index);
        let idle = state.idle_connections;
@@ -915,29 +852,12 @@ impl ConnectionPool {

 /// Wrapper for the bb8 connection pool.
 pub struct ServerPool {
-    /// Server address.
    address: Address,
-
-    /// Server Postgres user.
    user: User,
-
-    /// Server database.
    database: String,
-
-    /// Client/server mapping.
    client_server_map: ClientServerMap,
-
-    /// Server statistics.
    stats: Arc<PoolStats>,
-
-    /// Server auth hash (for auth passthrough).
    auth_hash: Arc<RwLock<Option<String>>>,
-
-    /// Server plugins.
-    plugins: Option<Plugins>,
-
-    /// Should we clean up dirty connections before putting them into the pool?
-    cleanup_connections: bool,
 }

 impl ServerPool {
@@ -948,8 +868,6 @@ impl ServerPool {
        client_server_map: ClientServerMap,
        stats: Arc<PoolStats>,
        auth_hash: Arc<RwLock<Option<String>>>,
-        plugins: Option<Plugins>,
-        cleanup_connections: bool,
    ) -> ServerPool {
        ServerPool {
            address,
@@ -958,8 +876,6 @@ impl ServerPool {
            client_server_map,
            stats,
            auth_hash,
-            plugins,
-            cleanup_connections,
        }
    }
 }
@@ -989,23 +905,10 @@ impl ManageConnection for ServerPool {
            self.client_server_map.clone(),
            stats.clone(),
            self.auth_hash.clone(),
-            self.cleanup_connections,
        )
        .await
        {
-            Ok(mut conn) => {
-                if let Some(ref plugins) = self.plugins {
-                    if let Some(ref prewarmer) = plugins.prewarmer {
-                        let mut prewarmer = prewarmer::Prewarmer {
-                            enabled: prewarmer.enabled,
-                            server: &mut conn,
-                            queries: &prewarmer.queries,
-                        };
-
-                        prewarmer.run().await?;
-                    }
-                }
-
+            Ok(conn) => {
                stats.idle();
                Ok(conn)
            }
--- a/src/query_router.rs
+++ b/src/query_router.rs
@@ -6,16 +6,13 @@ use once_cell::sync::OnceCell;
 use regex::{Regex, RegexSet};
 use sqlparser::ast::Statement::{Query, StartTransaction};
 use sqlparser::ast::{
-    BinaryOperator, Expr, Ident, JoinConstraint, JoinOperator, SetExpr, Statement, TableFactor,
-    Value,
+    BinaryOperator, Expr, Ident, JoinConstraint, JoinOperator, SetExpr, TableFactor, Value,
 };
 use sqlparser::dialect::PostgreSqlDialect;
 use sqlparser::parser::Parser;

 use crate::config::Role;
-use crate::errors::Error;
 use crate::messages::BytesMutReader;
-use crate::plugins::{Intercept, Plugin, PluginOutput, QueryLogger, TableAccess};
 use crate::pool::PoolSettings;
 use crate::sharding::Sharder;

@@ -132,10 +129,6 @@ impl QueryRouter {
        self.pool_settings = pool_settings;
    }

-    pub fn pool_settings<'a>(&'a self) -> &'a PoolSettings {
-        &self.pool_settings
-    }
-
    /// Try to parse a command and execute it.
    pub fn try_execute_command(&mut self, message_buffer: &BytesMut) -> Option<(Command, String)> {
        let mut message_cursor = Cursor::new(message_buffer);
@@ -331,7 +324,10 @@ impl QueryRouter {
        Some((command, value))
    }

-    pub fn parse(message: &BytesMut) -> Result<Vec<sqlparser::ast::Statement>, Error> {
+    /// Try to infer which server to connect to based on the contents of the query.
+    pub fn infer(&mut self, message: &BytesMut) -> bool {
+        debug!("Inferring role");
+
        let mut message_cursor = Cursor::new(message);

        let code = message_cursor.get_u8() as char;
@@ -357,29 +353,28 @@ impl QueryRouter {
                query
            }

-            _ => return Err(Error::UnsupportedStatement),
+            _ => return false,
        };

-        match Parser::parse_sql(&PostgreSqlDialect {}, &query) {
-            Ok(ast) => Ok(ast),
+        let ast = match Parser::parse_sql(&PostgreSqlDialect {}, &query) {
+            Ok(ast) => ast,
            Err(err) => {
+                // SELECT ... FOR UPDATE won't get parsed correctly.
                debug!("{}: {}", err, query);
-                Err(Error::QueryRouterParserError(err.to_string()))
+                self.active_role = Some(Role::Primary);
+                return false;
            }
-        }
-    }
+        };

-    /// Try to infer which server to connect to based on the contents of the query.
-    pub fn infer(&mut self, ast: &Vec<sqlparser::ast::Statement>) -> Result<(), Error> {
-        debug!("Inferring role");
+        debug!("AST: {:?}", ast);

        if ast.is_empty() {
            // That's weird, no idea, let's go to primary
            self.active_role = Some(Role::Primary);
-            return Err(Error::QueryRouterParserError("empty query".into()));
+            return false;
        }

-        for q in ast {
+        for q in &ast {
            match q {
                // All transactions go to the primary, probably a write.
                StartTransaction { .. } => {
@@ -423,7 +418,7 @@ impl QueryRouter {
            };
        }

-        Ok(())
+        true
    }

    /// Parse the shard number from the Bind message
@@ -788,52 +783,6 @@ impl QueryRouter {
        }
    }

-    /// Add your plugins here and execute them.
-    pub async fn execute_plugins(&self, ast: &Vec<Statement>) -> Result<PluginOutput, Error> {
-        let plugins = match self.pool_settings.plugins {
-            Some(ref plugins) => plugins,
-            None => return Ok(PluginOutput::Allow),
-        };
-
-        if let Some(ref query_logger) = plugins.query_logger {
-            let mut query_logger = QueryLogger {
-                enabled: query_logger.enabled,
-                user: &self.pool_settings.user.username,
-                db: &self.pool_settings.db,
-            };
-
-            let _ = query_logger.run(&self, ast).await;
-        }
-
-        if let Some(ref intercept) = plugins.intercept {
-            let mut intercept = Intercept {
-                enabled: intercept.enabled,
-                config: &intercept,
-            };
-
-            let result = intercept.run(&self, ast).await;
-
-            if let Ok(PluginOutput::Intercept(output)) = result {
-                return Ok(PluginOutput::Intercept(output));
-            }
-        }
-
-        if let Some(ref table_access) = plugins.table_access {
-            let mut table_access = TableAccess {
-                enabled: table_access.enabled,
-                tables: &table_access.tables,
-            };
-
-            let result = table_access.run(&self, ast).await;
-
-            if let Ok(PluginOutput::Deny(error)) = result {
-                return Ok(PluginOutput::Deny(error));
-            }
-        }
-
-        Ok(PluginOutput::Allow)
-    }
-
    fn set_sharding_key(&mut self, sharding_key: i64) -> Option<usize> {
        let sharder = Sharder::new(
            self.pool_settings.shards,
@@ -861,23 +810,12 @@ impl QueryRouter {
    /// Should we attempt to parse queries?
    pub fn query_parser_enabled(&self) -> bool {
        let enabled = match self.query_parser_enabled {
-            None => {
-                debug!(
-                    "Using pool settings, query_parser_enabled: {}",
-                    self.pool_settings.query_parser_enabled
-                );
-                self.pool_settings.query_parser_enabled
-            }
-
-            Some(value) => {
-                debug!(
-                    "Using query parser override, query_parser_enabled: {}",
-                    value
-                );
-                value
-            }
+            None => self.pool_settings.query_parser_enabled,
+            Some(value) => value,
        };

+        debug!("Query parser enabled: {}", enabled);
+
        enabled
    }

@@ -924,7 +862,7 @@ mod test {

        for query in queries {
            // It's a recognized query
-            assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
+            assert!(qr.infer(&query));
            assert_eq!(qr.role(), Some(Role::Replica));
        }
    }
@@ -943,7 +881,7 @@ mod test {

        for query in queries {
            // It's a recognized query
-            assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
+            assert!(qr.infer(&query));
            assert_eq!(qr.role(), Some(Role::Primary));
        }
    }
@@ -955,7 +893,7 @@ mod test {
        let query = simple_query("SELECT * FROM items WHERE id = 5");
        assert!(qr.try_execute_command(&simple_query("SET PRIMARY READS TO on")) != None);

-        assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
+        assert!(qr.infer(&query));
        assert_eq!(qr.role(), None);
    }

@@ -975,7 +913,7 @@ mod test {
        res.put(prepared_stmt);
        res.put_i16(0);

-        assert!(qr.infer(&QueryRouter::parse(&res).unwrap()).is_ok());
+        assert!(qr.infer(&res));
        assert_eq!(qr.role(), Some(Role::Replica));
    }

@@ -1139,11 +1077,11 @@ mod test {
        assert_eq!(qr.role(), None);

        let query = simple_query("INSERT INTO test_table VALUES (1)");
-        assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
+        assert!(qr.infer(&query));
        assert_eq!(qr.role(), Some(Role::Primary));

        let query = simple_query("SELECT * FROM test_table");
-        assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
+        assert!(qr.infer(&query));
        assert_eq!(qr.role(), Some(Role::Replica));

        assert!(qr.query_parser_enabled());
@@ -1175,8 +1113,6 @@ mod test {
            auth_query: None,
            auth_query_password: None,
            auth_query_user: None,
-            db: "test".to_string(),
-            plugins: None,
        };
        let mut qr = QueryRouter::new();
        assert_eq!(qr.active_role, None);
@@ -1206,24 +1142,15 @@ mod test {
        QueryRouter::setup();

        let mut qr = QueryRouter::new();
-        assert!(qr
-            .infer(&QueryRouter::parse(&simple_query("BEGIN; SELECT 1; COMMIT;")).unwrap())
-            .is_ok());
+        assert!(qr.infer(&simple_query("BEGIN; SELECT 1; COMMIT;")));
        assert_eq!(qr.role(), Role::Primary);

-        assert!(qr
-            .infer(&QueryRouter::parse(&simple_query("SELECT 1; SELECT 2;")).unwrap())
-            .is_ok());
+        assert!(qr.infer(&simple_query("SELECT 1; SELECT 2;")));
        assert_eq!(qr.role(), Role::Replica);

-        assert!(qr
-            .infer(
-                &QueryRouter::parse(&simple_query(
-                    "SELECT 123; INSERT INTO t VALUES (5); SELECT 1;"
-                ))
-                .unwrap()
-            )
-            .is_ok());
+        assert!(qr.infer(&simple_query(
+            "SELECT 123; INSERT INTO t VALUES (5); SELECT 1;"
+        )));
        assert_eq!(qr.role(), Role::Primary);
    }

@@ -1250,10 +1177,7 @@ mod test {
            auth_query: None,
            auth_query_password: None,
            auth_query_user: None,
-            db: "test".to_string(),
-            plugins: None,
        };
-
        let mut qr = QueryRouter::new();
        qr.update_pool_settings(pool_settings.clone());

@@ -1284,84 +1208,47 @@ mod test {
        qr.pool_settings.automatic_sharding_key = Some("data.id".to_string());
        qr.pool_settings.shards = 3;

-        assert!(qr
-            .infer(&QueryRouter::parse(&simple_query("SELECT * FROM data WHERE id = 5")).unwrap())
-            .is_ok());
+        assert!(qr.infer(&simple_query("SELECT * FROM data WHERE id = 5")));
        assert_eq!(qr.shard(), 2);

-        assert!(qr
-            .infer(
-                &QueryRouter::parse(&simple_query(
-                    "SELECT one, two, three FROM public.data WHERE id = 6"
-                ))
-                .unwrap()
-            )
-            .is_ok());
+        assert!(qr.infer(&simple_query(
+            "SELECT one, two, three FROM public.data WHERE id = 6"
+        )));
        assert_eq!(qr.shard(), 0);

-        assert!(qr
-            .infer(
-                &QueryRouter::parse(&simple_query(
-                    "SELECT * FROM data
+        assert!(qr.infer(&simple_query(
+            "SELECT * FROM data
            INNER JOIN t2 ON data.id = 5
            AND t2.data_id = data.id
        WHERE data.id = 5"
-                ))
-                .unwrap()
-            )
-            .is_ok());
+        )));
        assert_eq!(qr.shard(), 2);

        // Shard did not move because we couldn't determine the sharding key since it could be ambiguous
        // in the query.
-        assert!(qr
-            .infer(
-                &QueryRouter::parse(&simple_query(
-                    "SELECT * FROM t2 INNER JOIN data ON id = 6 AND data.id = t2.data_id"
-                ))
-                .unwrap()
-            )
-            .is_ok());
+        assert!(qr.infer(&simple_query(
+            "SELECT * FROM t2 INNER JOIN data ON id = 6 AND data.id = t2.data_id"
+        )));
        assert_eq!(qr.shard(), 2);

-        assert!(qr
-            .infer(
-                &QueryRouter::parse(&simple_query(
-                    r#"SELECT * FROM "public"."data" WHERE "id" = 6"#
-                ))
-                .unwrap()
-            )
-            .is_ok());
+        assert!(qr.infer(&simple_query(
+            r#"SELECT * FROM "public"."data" WHERE "id" = 6"#
+        )));
        assert_eq!(qr.shard(), 0);

-        assert!(qr
-            .infer(
-                &QueryRouter::parse(&simple_query(
-                    r#"SELECT * FROM "public"."data" WHERE "data"."id" = 5"#
-                ))
-                .unwrap()
-            )
-            .is_ok());
+        assert!(qr.infer(&simple_query(
+            r#"SELECT * FROM "public"."data" WHERE "data"."id" = 5"#
+        )));
        assert_eq!(qr.shard(), 2);

        // Super unique sharding key
        qr.pool_settings.automatic_sharding_key = Some("*.unique_enough_column_name".to_string());
-        assert!(qr
-            .infer(
-                &QueryRouter::parse(&simple_query(
-                    "SELECT * FROM table_x WHERE unique_enough_column_name = 6"
-                ))
-                .unwrap()
-            )
-            .is_ok());
+        assert!(qr.infer(&simple_query(
+            "SELECT * FROM table_x WHERE unique_enough_column_name = 6"
+        )));
        assert_eq!(qr.shard(), 0);

-        assert!(qr
-            .infer(
-                &QueryRouter::parse(&simple_query("SELECT * FROM table_y WHERE another_key = 5"))
-                    .unwrap()
-            )
-            .is_ok());
+        assert!(qr.infer(&simple_query("SELECT * FROM table_y WHERE another_key = 5")));
        assert_eq!(qr.shard(), 0);
    }

@@ -1385,61 +1272,11 @@ mod test {
        qr.pool_settings.automatic_sharding_key = Some("data.id".to_string());
        qr.pool_settings.shards = 3;

-        assert!(qr
-            .infer(&QueryRouter::parse(&simple_query(stmt)).unwrap())
-            .is_ok());
+        assert!(qr.infer(&simple_query(stmt)));
        assert_eq!(qr.placeholders.len(), 1);

        assert!(qr.infer_shard_from_bind(&bind));
        assert_eq!(qr.shard(), 2);
        assert!(qr.placeholders.is_empty());
    }
-
-    #[tokio::test]
-    async fn test_table_access_plugin() {
-        use crate::config::{Plugins, TableAccess};
-        let table_access = TableAccess {
-            enabled: true,
-            tables: vec![String::from("pg_database")],
-        };
-        let plugins = Plugins {
-            table_access: Some(table_access),
-            intercept: None,
-            query_logger: None,
-            prewarmer: None,
-        };
-
-        QueryRouter::setup();
-        let mut pool_settings = PoolSettings::default();
-        pool_settings.query_parser_enabled = true;
-        pool_settings.plugins = Some(plugins);
-
-        let mut qr = QueryRouter::new();
-        qr.update_pool_settings(pool_settings);
-
-        let query = simple_query("SELECT * FROM pg_database");
-        let ast = QueryRouter::parse(&query).unwrap();
-
-        let res = qr.execute_plugins(&ast).await;
-
-        assert_eq!(
-            res,
-            Ok(PluginOutput::Deny(
-                "permission for table \"pg_database\" denied".to_string()
-            ))
-        );
-    }
-
-    #[tokio::test]
-    async fn test_plugins_disabled_by_defaault() {
-        QueryRouter::setup();
-        let qr = QueryRouter::new();
-
-        let query = simple_query("SELECT * FROM pg_database");
-        let ast = QueryRouter::parse(&query).unwrap();
-
-        let res = qr.execute_plugins(&ast).await;
-
-        assert_eq!(res, Ok(PluginOutput::Allow));
-    }
 }
--- a/src/server.rs
+++ b/src/server.rs
@@ -7,143 +7,22 @@ use parking_lot::{Mutex, RwLock};
 use postgres_protocol::message;
 use std::collections::HashMap;
 use std::io::Read;
-use std::net::IpAddr;
 use std::sync::Arc;
 use std::time::SystemTime;
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, BufStream};
-use tokio::net::TcpStream;
-use tokio_rustls::rustls::{OwnedTrustAnchor, RootCertStore};
-use tokio_rustls::{client::TlsStream, TlsConnector};
+use tokio::io::{AsyncReadExt, BufReader};
+use tokio::net::{
+    tcp::{OwnedReadHalf, OwnedWriteHalf},
+    TcpStream,
+};

-use crate::config::{get_config, Address, User};
+use crate::config::{Address, User};
 use crate::constants::*;
-use crate::dns_cache::{AddrSet, CACHED_RESOLVER};
 use crate::errors::{Error, ServerIdentifier};
 use crate::messages::*;
 use crate::mirrors::MirroringManager;
 use crate::pool::ClientServerMap;
 use crate::scram::ScramSha256;
 use crate::stats::ServerStats;
-use std::io::Write;
-
-use pin_project::pin_project;
-
-#[pin_project(project = SteamInnerProj)]
-pub enum StreamInner {
-    Plain {
-        #[pin]
-        stream: TcpStream,
-    },
-    Tls {
-        #[pin]
-        stream: TlsStream<TcpStream>,
-    },
-}
-
-impl AsyncWrite for StreamInner {
-    fn poll_write(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        buf: &[u8],
-    ) -> std::task::Poll<Result<usize, std::io::Error>> {
-        let this = self.project();
-        match this {
-            SteamInnerProj::Tls { stream } => stream.poll_write(cx, buf),
-            SteamInnerProj::Plain { stream } => stream.poll_write(cx, buf),
-        }
-    }
-
-    fn poll_flush(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Result<(), std::io::Error>> {
-        let this = self.project();
-        match this {
-            SteamInnerProj::Tls { stream } => stream.poll_flush(cx),
-            SteamInnerProj::Plain { stream } => stream.poll_flush(cx),
-        }
-    }
-
-    fn poll_shutdown(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Result<(), std::io::Error>> {
-        let this = self.project();
-        match this {
-            SteamInnerProj::Tls { stream } => stream.poll_shutdown(cx),
-            SteamInnerProj::Plain { stream } => stream.poll_shutdown(cx),
-        }
-    }
-}
-
-impl AsyncRead for StreamInner {
-    fn poll_read(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-        buf: &mut tokio::io::ReadBuf<'_>,
-    ) -> std::task::Poll<std::io::Result<()>> {
-        let this = self.project();
-        match this {
-            SteamInnerProj::Tls { stream } => stream.poll_read(cx, buf),
-            SteamInnerProj::Plain { stream } => stream.poll_read(cx, buf),
-        }
-    }
-}
-
-impl StreamInner {
-    pub fn try_write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        match self {
-            StreamInner::Tls { stream } => {
-                let r = stream.get_mut();
-                let mut w = r.1.writer();
-                w.write(buf)
-            }
-            StreamInner::Plain { stream } => stream.try_write(buf),
-        }
-    }
-}
-
-#[derive(Copy, Clone)]
-struct CleanupState {
-    /// If server connection requires DISCARD ALL before checkin because of set statement
-    needs_cleanup_set: bool,
-
-    /// If server connection requires DISCARD ALL before checkin because of prepare statement
-    needs_cleanup_prepare: bool,
-}
-
-impl CleanupState {
-    fn new() -> Self {
-        CleanupState {
-            needs_cleanup_set: false,
-            needs_cleanup_prepare: false,
-        }
-    }
-
-    fn needs_cleanup(&self) -> bool {
-        self.needs_cleanup_set || self.needs_cleanup_prepare
-    }
-
-    fn set_true(&mut self) {
-        self.needs_cleanup_set = true;
-        self.needs_cleanup_prepare = true;
-    }
-
-    fn reset(&mut self) {
-        self.needs_cleanup_set = false;
-        self.needs_cleanup_prepare = false;
-    }
-}
-
-impl std::fmt::Display for CleanupState {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "SET: {}, PREPARE: {}",
-            self.needs_cleanup_set, self.needs_cleanup_prepare
-        )
-    }
-}

 /// Server state.
 pub struct Server {
@@ -151,11 +30,15 @@ pub struct Server {
    /// port, e.g. 5432, and role, e.g. primary or replica.
    address: Address,

-    /// Server TCP connection.
-    stream: BufStream<StreamInner>,
+    /// Buffered read socket.
+    read: BufReader<OwnedReadHalf>,
+
+    /// Unbuffered write socket (our client code buffers).
+    write: OwnedWriteHalf,

    /// Our server response buffer. We buffer data before we give it to the client.
    buffer: BytesMut,
+    is_async: bool,

    /// Server information the server sent us over on startup.
    server_info: BytesMut,
@@ -173,8 +56,8 @@ pub struct Server {
    /// Is the server broken? We'll remote it from the pool if so.
    bad: bool,

-    /// If server connection requires DISCARD ALL before checkin
-    cleanup_state: CleanupState,
+    /// 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,
@@ -188,16 +71,10 @@ pub struct Server {
    /// Application name using the server at the moment.
    application_name: String,

-    /// Last time that a successful server send or response happened
+    // Last time that a successful server send or response happened
    last_activity: SystemTime,

    mirror_manager: Option<MirroringManager>,
-
-    /// Associated addresses used
-    addr_set: Option<AddrSet>,
-
-    /// Should clean up dirty connections?
-    cleanup_connections: bool,
 }

 impl Server {
@@ -210,26 +87,7 @@ impl Server {
        client_server_map: ClientServerMap,
        stats: Arc<ServerStats>,
        auth_hash: Arc<RwLock<Option<String>>>,
-        cleanup_connections: bool,
    ) -> Result<Server, Error> {
-        let cached_resolver = CACHED_RESOLVER.load();
-        let mut addr_set: Option<AddrSet> = None;
-
-        // If we are caching addresses and hostname is not an IP
-        if cached_resolver.enabled() && address.host.parse::<IpAddr>().is_err() {
-            debug!("Resolving {}", &address.host);
-            addr_set = match cached_resolver.lookup_ip(&address.host).await {
-                Ok(ok) => {
-                    debug!("Obtained: {:?}", ok);
-                    Some(ok)
-                }
-                Err(err) => {
-                    warn!("Error trying to resolve {}, ({:?})", &address.host, err);
-                    None
-                }
-            }
-        };
-
        let mut stream =
            match TcpStream::connect(&format!("{}:{}", &address.host, address.port)).await {
                Ok(stream) => stream,
@@ -241,88 +99,8 @@ impl Server {
                    )));
                }
            };
-
-        // TCP timeouts.
        configure_socket(&stream);

-        let config = get_config();
-
-        let mut stream = if config.general.server_tls {
-            // Request a TLS connection
-            ssl_request(&mut stream).await?;
-
-            let response = match stream.read_u8().await {
-                Ok(response) => response as char,
-                Err(err) => {
-                    return Err(Error::SocketError(format!(
-                        "Server socket error: {:?}",
-                        err
-                    )))
-                }
-            };
-
-            match response {
-                // Server supports TLS
-                'S' => {
-                    debug!("Connecting to server using TLS");
-
-                    let mut root_store = RootCertStore::empty();
-                    root_store.add_server_trust_anchors(
-                        webpki_roots::TLS_SERVER_ROOTS.0.iter().map(|ta| {
-                            OwnedTrustAnchor::from_subject_spki_name_constraints(
-                                ta.subject,
-                                ta.spki,
-                                ta.name_constraints,
-                            )
-                        }),
-                    );
-
-                    let mut tls_config = rustls::ClientConfig::builder()
-                        .with_safe_defaults()
-                        .with_root_certificates(root_store)
-                        .with_no_client_auth();
-
-                    // Equivalent to sslmode=prefer which is fine most places.
-                    // If you want verify-full, change `verify_server_certificate` to true.
-                    if !config.general.verify_server_certificate {
-                        let mut dangerous = tls_config.dangerous();
-                        dangerous.set_certificate_verifier(Arc::new(
-                            crate::tls::NoCertificateVerification {},
-                        ));
-                    }
-
-                    let connector = TlsConnector::from(Arc::new(tls_config));
-                    let stream = match connector
-                        .connect(address.host.as_str().try_into().unwrap(), stream)
-                        .await
-                    {
-                        Ok(stream) => stream,
-                        Err(err) => {
-                            return Err(Error::SocketError(format!("Server TLS error: {:?}", err)))
-                        }
-                    };
-
-                    StreamInner::Tls { stream }
-                }
-
-                // Server does not support TLS
-                'N' => StreamInner::Plain { stream },
-
-                // Something else?
-                m => {
-                    return Err(Error::SocketError(format!(
-                        "Unknown message: {}",
-                        m as char
-                    )));
-                }
-            }
-        } else {
-            StreamInner::Plain { stream }
-        };
-
-        // let (read, write) = split(stream);
-        // let (mut read, mut write) = (ReadInner::Plain { stream: read }, WriteInner::Plain { stream: write });
-
        trace!("Sending StartupMessage");

        // StartupMessage
@@ -468,7 +246,7 @@ impl Server {

                            let sasl_type = String::from_utf8_lossy(&sasl_auth[..sasl_len - 2]);

-                            if sasl_type.contains(SCRAM_SHA_256) {
+                            if sasl_type == SCRAM_SHA_256 {
                                debug!("Using {}", SCRAM_SHA_256);

                                // Generate client message.
@@ -491,7 +269,7 @@ impl Server {
                                res.put_i32(sasl_response.len() as i32);
                                res.put(sasl_response);

-                                write_all_flush(&mut stream, &res).await?;
+                                write_all(&mut stream, res).await?;
                            } else {
                                error!("Unsupported SCRAM version: {}", sasl_type);
                                return Err(Error::ServerError);
@@ -522,7 +300,7 @@ impl Server {
                            res.put_i32(4 + sasl_response.len() as i32);
                            res.put(sasl_response);

-                            write_all_flush(&mut stream, &res).await?;
+                            write_all(&mut stream, res).await?;
                        }

                        SASL_FINAL => {
@@ -666,19 +444,22 @@ impl Server {
                        }
                    };

+                    let (read, write) = stream.into_split();
+
                    let mut server = Server {
                        address: address.clone(),
-                        stream: BufStream::new(stream),
+                        read: BufReader::new(read),
+                        write,
                        buffer: BytesMut::with_capacity(8196),
+                        is_async: false,
                        server_info,
                        process_id,
                        secret_key,
                        in_transaction: false,
                        data_available: false,
                        bad: false,
-                        cleanup_state: CleanupState::new(),
+                        needs_cleanup: false,
                        client_server_map,
-                        addr_set,
                        connected_at: chrono::offset::Utc::now().naive_utc(),
                        stats,
                        application_name: String::new(),
@@ -691,7 +472,6 @@ impl Server {
                                address.mirrors.clone(),
                            )),
                        },
-                        cleanup_connections,
                    };

                    server.set_name("pgcat").await?;
@@ -737,7 +517,7 @@ impl Server {
        bytes.put_i32(process_id);
        bytes.put_i32(secret_key);

-        write_all_flush(&mut stream, &bytes).await
+        write_all(&mut stream, bytes).await
    }

    /// Send messages to the server from the client.
@@ -745,35 +525,39 @@ impl Server {
        self.mirror_send(messages);
        self.stats().data_sent(messages.len());

-        match write_all_flush(&mut self.stream, &messages).await {
+        match write_all_half(&mut self.write, messages).await {
            Ok(_) => {
                // Successfully sent to server
                self.last_activity = SystemTime::now();
                Ok(())
            }
            Err(err) => {
-                error!(
-                    "Terminating server {:?} because of: {:?}",
-                    self.address, err
-                );
+                error!("Terminating server because of: {:?}", err);
                self.bad = true;
                Err(err)
            }
        }
    }

+    /// Switch to async mode, flushing messages as soon
+    /// as we receive them without buffering or waiting for "ReadyForQuery".
+    pub fn switch_async(&mut self, on: bool) {
+        if on {
+            self.is_async = true;
+        } else {
+            self.is_async = false;
+        }
+    }
+
    /// Receive data from the server in response to a client request.
    /// This method must be called multiple times while `self.is_data_available()` is true
    /// in order to receive all data the server has to offer.
    pub async fn recv(&mut self) -> Result<BytesMut, Error> {
        loop {
-            let mut message = match read_message(&mut self.stream).await {
+            let mut message = match read_message(&mut self.read).await {
                Ok(message) => message,
                Err(err) => {
-                    error!(
-                        "Terminating server {:?} because of: {:?}",
-                        self.address, err
-                    );
+                    error!("Terminating server because of: {:?}", err);
                    self.bad = true;
                    return Err(err);
                }
@@ -840,12 +624,12 @@ impl Server {
                                    // This will reduce amount of discard statements sent
                                    if !self.in_transaction {
                                        debug!("Server connection marked for clean up");
-                                        self.cleanup_state.needs_cleanup_set = true;
+                                        self.needs_cleanup = true;
                                    }
                                }
                                "PREPARE\0" => {
                                    debug!("Server connection marked for clean up");
-                                    self.cleanup_state.needs_cleanup_prepare = true;
+                                    self.needs_cleanup = true;
                                }
                                _ => (),
                            }
@@ -860,7 +644,10 @@ impl Server {
                // DataRow
                'D' => {
                    // More data is available after this message, this is not the end of the reply.
-                    self.data_available = true;
+                    // If we're async, flush to client now.
+                    if !self.is_async {
+                        self.data_available = true;
+                    }

                    // Don't flush yet, the more we buffer, the faster this goes...up to a limit.
                    if self.buffer.len() >= 8196 {
@@ -873,7 +660,10 @@ impl Server {

                // CopyOutResponse: copy is starting from the server to the client.
                'H' => {
-                    self.data_available = true;
+                    // If we're in async mode, flush now.
+                    if !self.is_async {
+                        self.data_available = true;
+                    }
                    break;
                }

@@ -893,6 +683,10 @@ impl Server {
                // Keep buffering until ReadyForQuery shows up.
                _ => (),
            };
+
+            if self.is_async {
+                break;
+            }
        }

        let bytes = self.buffer.clone();
@@ -926,23 +720,7 @@ impl Server {
    /// Server & client are out of sync, we must discard this connection.
    /// This happens with clients that misbehave.
    pub fn is_bad(&self) -> bool {
-        if self.bad {
-            return self.bad;
-        };
-        let cached_resolver = CACHED_RESOLVER.load();
-        if cached_resolver.enabled() {
-            if let Some(addr_set) = &self.addr_set {
-                if cached_resolver.has_changed(self.address.host.as_str(), addr_set) {
-                    warn!(
-                        "DNS changed for {}, it was {:?}. Dropping server connection.",
-                        self.address.host.as_str(),
-                        addr_set
-                    );
-                    return true;
-                }
-            }
-        }
-        false
+        self.bad
    }

    /// Get server startup information to forward it to the client.
@@ -975,8 +753,6 @@ impl Server {
    /// It will use the simple query protocol.
    /// Result will not be returned, so this is useful for things like `SET` or `ROLLBACK`.
    pub async fn query(&mut self, query: &str) -> Result<(), Error> {
-        debug!("Running `{}` on server {:?}", query, self.address);
-
        let query = simple_query(query);

        self.send(&query).await?;
@@ -1009,11 +785,10 @@ impl Server {
        // 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.cleanup_state.needs_cleanup() && self.cleanup_connections {
-            warn!("Server returned with session state altered, discarding state ({}) for application {}", self.cleanup_state, self.application_name);
+        if self.needs_cleanup {
+            warn!("Server returned with session state altered, discarding state");
            self.query("DISCARD ALL").await?;
-            self.query("RESET ROLE").await?;
-            self.cleanup_state.reset();
+            self.needs_cleanup = false;
        }

        Ok(())
@@ -1025,12 +800,12 @@ impl Server {
            self.application_name = name.to_string();
            // We don't want `SET application_name` to mark the server connection
            // as needing cleanup
-            let needs_cleanup_before = self.cleanup_state;
+            let needs_cleanup_before = self.needs_cleanup;

            let result = Ok(self
                .query(&format!("SET application_name = '{}'", name))
                .await?);
-            self.cleanup_state = needs_cleanup_before;
+            self.needs_cleanup = needs_cleanup_before;
            result
        } else {
            Ok(())
@@ -1055,7 +830,7 @@ impl Server {

    // Marks a connection as needing DISCARD ALL at checkin
    pub fn mark_dirty(&mut self) {
-        self.cleanup_state.set_true();
+        self.needs_cleanup = true;
    }

    pub fn mirror_send(&mut self, bytes: &BytesMut) {
@@ -1089,7 +864,6 @@ impl Server {
            client_server_map,
            Arc::new(ServerStats::default()),
            Arc::new(RwLock::new(None)),
-            true,
        )
        .await?;
        debug!("Connected!, sending query.");
@@ -1183,27 +957,23 @@ impl Drop for Server {
        // Update statistics
        self.stats.disconnect();

-        let mut bytes = BytesMut::with_capacity(5);
+        let mut bytes = BytesMut::with_capacity(4);
        bytes.put_u8(b'X');
        bytes.put_i32(4);

-        match self.stream.get_mut().try_write(&bytes) {
-            Ok(5) => (),
-            _ => debug!("Dirty shutdown"),
+        match self.write.try_write(&bytes) {
+            Ok(_) => (),
+            Err(_) => debug!("Dirty shutdown"),
        };

+        // Should not matter.
+        self.bad = true;
+
        let now = chrono::offset::Utc::now().naive_utc();
        let duration = now - self.connected_at;

-        let message = if self.bad {
-            "Server connection terminated"
-        } else {
-            "Server connection closed"
-        };
-
        info!(
-            "{} {:?}, session duration: {}",
-            message,
+            "Server connection closed {:?}, session duration: {}",
            self.address,
            crate::format_duration(&duration)
        );
--- a/src/stats.rs
+++ b/src/stats.rs
@@ -107,20 +107,8 @@ impl Collector {
            loop {
                interval.tick().await;

-                // Hold read lock for duration of update to retain all server stats
-                let server_stats = SERVER_STATS.read();
-
-                for stats in server_stats.values() {
-                    if !stats.check_address_stat_average_is_updated_status() {
-                        stats.address_stats().update_averages();
-                        stats.address_stats().reset_current_counts();
-                        stats.set_address_stat_average_is_updated_status(true);
-                    }
-                }
-
-                // Reset to false for next update
-                for stats in server_stats.values() {
-                    stats.set_address_stat_average_is_updated_status(false);
+                for stats in SERVER_STATS.read().values() {
+                    stats.address_stats().update_averages();
                }
            }
        });
--- a/src/stats/address.rs
+++ b/src/stats/address.rs
@@ -1,29 +1,26 @@
+use log::warn;
 use std::sync::atomic::*;
 use std::sync::Arc;

-#[derive(Debug, Clone, Default)]
-struct AddressStatFields {
-    xact_count: Arc<AtomicU64>,
-    query_count: Arc<AtomicU64>,
-    bytes_received: Arc<AtomicU64>,
-    bytes_sent: Arc<AtomicU64>,
-    xact_time: Arc<AtomicU64>,
-    query_time: Arc<AtomicU64>,
-    wait_time: Arc<AtomicU64>,
-    errors: Arc<AtomicU64>,
-}
-
 /// Internal address stats
 #[derive(Debug, Clone, Default)]
 pub struct AddressStats {
-    total: AddressStatFields,
-
-    current: AddressStatFields,
-
-    averages: AddressStatFields,
-
-    // Determines if the averages have been updated since the last time they were reported
-    pub averages_updated: Arc<AtomicBool>,
+    pub total_xact_count: Arc<AtomicU64>,
+    pub total_query_count: Arc<AtomicU64>,
+    pub total_received: Arc<AtomicU64>,
+    pub total_sent: Arc<AtomicU64>,
+    pub total_xact_time: Arc<AtomicU64>,
+    pub total_query_time: Arc<AtomicU64>,
+    pub total_wait_time: Arc<AtomicU64>,
+    pub total_errors: Arc<AtomicU64>,
+    pub avg_query_count: Arc<AtomicU64>,
+    pub avg_query_time: Arc<AtomicU64>,
+    pub avg_recv: Arc<AtomicU64>,
+    pub avg_sent: Arc<AtomicU64>,
+    pub avg_errors: Arc<AtomicU64>,
+    pub avg_xact_time: Arc<AtomicU64>,
+    pub avg_xact_count: Arc<AtomicU64>,
+    pub avg_wait_time: Arc<AtomicU64>,
 }

 impl IntoIterator for AddressStats {
@@ -34,67 +31,67 @@ impl IntoIterator for AddressStats {
        vec![
            (
                "total_xact_count".to_string(),
-                self.total.xact_count.load(Ordering::Relaxed),
+                self.total_xact_count.load(Ordering::Relaxed),
            ),
            (
                "total_query_count".to_string(),
-                self.total.query_count.load(Ordering::Relaxed),
+                self.total_query_count.load(Ordering::Relaxed),
            ),
            (
                "total_received".to_string(),
-                self.total.bytes_received.load(Ordering::Relaxed),
+                self.total_received.load(Ordering::Relaxed),
            ),
            (
                "total_sent".to_string(),
-                self.total.bytes_sent.load(Ordering::Relaxed),
+                self.total_sent.load(Ordering::Relaxed),
            ),
            (
                "total_xact_time".to_string(),
-                self.total.xact_time.load(Ordering::Relaxed),
+                self.total_xact_time.load(Ordering::Relaxed),
            ),
            (
                "total_query_time".to_string(),
-                self.total.query_time.load(Ordering::Relaxed),
+                self.total_query_time.load(Ordering::Relaxed),
            ),
            (
                "total_wait_time".to_string(),
-                self.total.wait_time.load(Ordering::Relaxed),
+                self.total_wait_time.load(Ordering::Relaxed),
            ),
            (
                "total_errors".to_string(),
-                self.total.errors.load(Ordering::Relaxed),
+                self.total_errors.load(Ordering::Relaxed),
            ),
            (
                "avg_xact_count".to_string(),
-                self.averages.xact_count.load(Ordering::Relaxed),
+                self.avg_xact_count.load(Ordering::Relaxed),
            ),
            (
                "avg_query_count".to_string(),
-                self.averages.query_count.load(Ordering::Relaxed),
+                self.avg_query_count.load(Ordering::Relaxed),
            ),
            (
                "avg_recv".to_string(),
-                self.averages.bytes_received.load(Ordering::Relaxed),
+                self.avg_recv.load(Ordering::Relaxed),
            ),
            (
                "avg_sent".to_string(),
-                self.averages.bytes_sent.load(Ordering::Relaxed),
+                self.avg_sent.load(Ordering::Relaxed),
            ),
            (
                "avg_errors".to_string(),
-                self.averages.errors.load(Ordering::Relaxed),
+                self.avg_errors.load(Ordering::Relaxed),
            ),
            (
                "avg_xact_time".to_string(),
-                self.averages.xact_time.load(Ordering::Relaxed),
+                self.avg_xact_time.load(Ordering::Relaxed),
            ),
            (
                "avg_query_time".to_string(),
-                self.averages.query_time.load(Ordering::Relaxed),
+                self.avg_query_time.load(Ordering::Relaxed),
            ),
            (
                "avg_wait_time".to_string(),
-                self.averages.wait_time.load(Ordering::Relaxed),
+                self.avg_wait_time.load(Ordering::Relaxed),
            ),
        ]
        .into_iter()
@@ -102,120 +99,22 @@ impl IntoIterator for AddressStats {
 }

 impl AddressStats {
-    pub fn xact_count_add(&self) {
-        self.total.xact_count.fetch_add(1, Ordering::Relaxed);
-        self.current.xact_count.fetch_add(1, Ordering::Relaxed);
-    }
-
-    pub fn query_count_add(&self) {
-        self.total.query_count.fetch_add(1, Ordering::Relaxed);
-        self.current.query_count.fetch_add(1, Ordering::Relaxed);
-    }
-
-    pub fn bytes_received_add(&self, bytes: u64) {
-        self.total
-            .bytes_received
-            .fetch_add(bytes, Ordering::Relaxed);
-        self.current
-            .bytes_received
-            .fetch_add(bytes, Ordering::Relaxed);
-    }
-
-    pub fn bytes_sent_add(&self, bytes: u64) {
-        self.total.bytes_sent.fetch_add(bytes, Ordering::Relaxed);
-        self.current.bytes_sent.fetch_add(bytes, Ordering::Relaxed);
-    }
-
-    pub fn xact_time_add(&self, time: u64) {
-        self.total.xact_time.fetch_add(time, Ordering::Relaxed);
-        self.current.xact_time.fetch_add(time, Ordering::Relaxed);
-    }
-
-    pub fn query_time_add(&self, time: u64) {
-        self.total.query_time.fetch_add(time, Ordering::Relaxed);
-        self.current.query_time.fetch_add(time, Ordering::Relaxed);
-    }
-
-    pub fn wait_time_add(&self, time: u64) {
-        self.total.wait_time.fetch_add(time, Ordering::Relaxed);
-        self.current.wait_time.fetch_add(time, Ordering::Relaxed);
-    }
-
    pub fn error(&self) {
-        self.total.errors.fetch_add(1, Ordering::Relaxed);
-        self.current.errors.fetch_add(1, Ordering::Relaxed);
+        self.total_errors.fetch_add(1, Ordering::Relaxed);
    }

    pub fn update_averages(&self) {
-        let stat_period_per_second = crate::stats::STAT_PERIOD / 1_000;
-
-        // xact_count
-        let current_xact_count = self.current.xact_count.load(Ordering::Relaxed);
-        let current_xact_time = self.current.xact_time.load(Ordering::Relaxed);
-        self.averages.xact_count.store(
-            current_xact_count / stat_period_per_second,
-            Ordering::Relaxed,
-        );
-        if current_xact_count == 0 {
-            self.averages.xact_time.store(0, Ordering::Relaxed);
-        } else {
-            self.averages
-                .xact_time
-                .store(current_xact_time / current_xact_count, Ordering::Relaxed);
+        let (totals, averages) = self.fields_iterators();
+        for data in totals.iter().zip(averages.iter()) {
+            let (total, average) = data;
+            if let Err(err) = average.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |avg| {
+                let total = total.load(Ordering::Relaxed);
+                let avg = (total - avg) / (crate::stats::STAT_PERIOD / 1_000); // Avg / second
+                Some(avg)
+            }) {
+                warn!("Could not update averages for addresses stats, {:?}", err);
+            }
        }
-
-        // query_count
-        let current_query_count = self.current.query_count.load(Ordering::Relaxed);
-        let current_query_time = self.current.query_time.load(Ordering::Relaxed);
-        self.averages.query_count.store(
-            current_query_count / stat_period_per_second,
-            Ordering::Relaxed,
-        );
-        if current_query_count == 0 {
-            self.averages.query_time.store(0, Ordering::Relaxed);
-        } else {
-            self.averages
-                .query_time
-                .store(current_query_time / current_query_count, Ordering::Relaxed);
-        }
-
-        // bytes_received
-        let current_bytes_received = self.current.bytes_received.load(Ordering::Relaxed);
-        self.averages.bytes_received.store(
-            current_bytes_received / stat_period_per_second,
-            Ordering::Relaxed,
-        );
-
-        // bytes_sent
-        let current_bytes_sent = self.current.bytes_sent.load(Ordering::Relaxed);
-        self.averages.bytes_sent.store(
-            current_bytes_sent / stat_period_per_second,
-            Ordering::Relaxed,
-        );
-
-        // wait_time
-        let current_wait_time = self.current.wait_time.load(Ordering::Relaxed);
-        self.averages.wait_time.store(
-            current_wait_time / stat_period_per_second,
-            Ordering::Relaxed,
-        );
-
-        // errors
-        let current_errors = self.current.errors.load(Ordering::Relaxed);
-        self.averages
-            .errors
-            .store(current_errors / stat_period_per_second, Ordering::Relaxed);
-    }
-
-    pub fn reset_current_counts(&self) {
-        self.current.xact_count.store(0, Ordering::Relaxed);
-        self.current.xact_time.store(0, Ordering::Relaxed);
-        self.current.query_count.store(0, Ordering::Relaxed);
-        self.current.query_time.store(0, Ordering::Relaxed);
-        self.current.bytes_received.store(0, Ordering::Relaxed);
-        self.current.bytes_sent.store(0, Ordering::Relaxed);
-        self.current.wait_time.store(0, Ordering::Relaxed);
-        self.current.errors.store(0, Ordering::Relaxed);
    }

    pub fn populate_row(&self, row: &mut Vec<String>) {
@@ -223,4 +122,28 @@ impl AddressStats {
            row.push(value.to_string());
        }
    }
+
+    fn fields_iterators(&self) -> (Vec<Arc<AtomicU64>>, Vec<Arc<AtomicU64>>) {
+        let mut totals: Vec<Arc<AtomicU64>> = Vec::new();
+        let mut averages: Vec<Arc<AtomicU64>> = Vec::new();
+
+        totals.push(self.total_xact_count.clone());
+        averages.push(self.avg_xact_count.clone());
+        totals.push(self.total_query_count.clone());
+        averages.push(self.avg_query_count.clone());
+        totals.push(self.total_received.clone());
+        averages.push(self.avg_recv.clone());
+        totals.push(self.total_sent.clone());
+        averages.push(self.avg_sent.clone());
+        totals.push(self.total_xact_time.clone());
+        averages.push(self.avg_xact_time.clone());
+        totals.push(self.total_query_time.clone());
+        averages.push(self.avg_query_time.clone());
+        totals.push(self.total_wait_time.clone());
+        averages.push(self.avg_wait_time.clone());
+        totals.push(self.total_errors.clone());
+        averages.push(self.avg_errors.clone());
+
+        (totals, averages)
+    }
 }
--- a/src/stats/server.rs
+++ b/src/stats/server.rs
@@ -139,17 +139,6 @@ impl ServerStats {
        self.address.stats.clone()
    }

-    pub fn check_address_stat_average_is_updated_status(&self) -> bool {
-        self.address.stats.averages_updated.load(Ordering::Relaxed)
-    }
-
-    pub fn set_address_stat_average_is_updated_status(&self, is_checked: bool) {
-        self.address
-            .stats
-            .averages_updated
-            .store(is_checked, Ordering::Relaxed);
-    }
-
    // Helper methods for show_servers
    pub fn pool_name(&self) -> String {
        self.pool_stats.database()
@@ -177,9 +166,12 @@ impl ServerStats {
    }

    pub fn checkout_time(&self, microseconds: u64, application_name: String) {
-        // Update server stats and address aggregation stats
+        // Update server stats and address aggergation stats
        self.set_application(application_name);
-        self.address.stats.wait_time_add(microseconds);
+        self.address
+            .stats
+            .total_wait_time
+            .fetch_add(microseconds, Ordering::Relaxed);
        self.pool_stats
            .maxwait
            .fetch_max(microseconds, Ordering::Relaxed);
@@ -188,8 +180,13 @@ impl ServerStats {
    /// Report a query executed by a client against a server
    pub fn query(&self, milliseconds: u64, application_name: &str) {
        self.set_application(application_name.to_string());
-        self.address.stats.query_count_add();
-        self.address.stats.query_time_add(milliseconds);
+        let address_stats = self.address_stats();
+        address_stats
+            .total_query_count
+            .fetch_add(1, Ordering::Relaxed);
+        address_stats
+            .total_query_time
+            .fetch_add(milliseconds, Ordering::Relaxed);
    }

    /// Report a transaction executed by a client a server
@@ -200,20 +197,29 @@ impl ServerStats {
        self.set_application(application_name.to_string());

        self.transaction_count.fetch_add(1, Ordering::Relaxed);
-        self.address.stats.xact_count_add();
+        self.address
+            .stats
+            .total_xact_count
+            .fetch_add(1, Ordering::Relaxed);
    }

    /// Report data sent to a server
    pub fn data_sent(&self, amount_bytes: usize) {
        self.bytes_sent
            .fetch_add(amount_bytes as u64, Ordering::Relaxed);
-        self.address.stats.bytes_sent_add(amount_bytes as u64);
+        self.address
+            .stats
+            .total_sent
+            .fetch_add(amount_bytes as u64, Ordering::Relaxed);
    }

    /// Report data received from a server
    pub fn data_received(&self, amount_bytes: usize) {
        self.bytes_received
            .fetch_add(amount_bytes as u64, Ordering::Relaxed);
-        self.address.stats.bytes_received_add(amount_bytes as u64);
+        self.address
+            .stats
+            .total_received
+            .fetch_add(amount_bytes as u64, Ordering::Relaxed);
    }
 }
--- a/src/tls.rs
+++ b/src/tls.rs
@@ -4,12 +4,7 @@ use rustls_pemfile::{certs, read_one, Item};
 use std::iter;
 use std::path::Path;
 use std::sync::Arc;
-use std::time::SystemTime;
-use tokio_rustls::rustls::{
-    self,
-    client::{ServerCertVerified, ServerCertVerifier},
-    Certificate, PrivateKey, ServerName,
-};
+use tokio_rustls::rustls::{self, Certificate, PrivateKey};
 use tokio_rustls::TlsAcceptor;

 use crate::config::get_config;
@@ -69,19 +64,3 @@ impl Tls {
        })
    }
 }
-
-pub struct NoCertificateVerification;
-
-impl ServerCertVerifier for NoCertificateVerification {
-    fn verify_server_cert(
-        &self,
-        _end_entity: &Certificate,
-        _intermediates: &[Certificate],
-        _server_name: &ServerName,
-        _scts: &mut dyn Iterator<Item = &[u8]>,
-        _ocsp_response: &[u8],
-        _now: SystemTime,
-    ) -> Result<ServerCertVerified, rustls::Error> {
-        Ok(ServerCertVerified::assertion())
-    }
-}
--- a/tests/python/async_test.py
+++ b/tests/python/async_test.py
@@ -0,0 +1,60 @@
+import psycopg2
+import asyncio
+import asyncpg
+
+PGCAT_HOST = "127.0.0.1"
+PGCAT_PORT = "6432"
+
+
+def regular_main():
+    # Connect to the PostgreSQL database
+    conn = psycopg2.connect(
+        host=PGCAT_HOST,
+        database="sharded_db",
+        user="sharding_user",
+        password="sharding_user",
+        port=PGCAT_PORT,
+    )
+
+    # Open a cursor to perform database operations
+    cur = conn.cursor()
+
+    # Execute a SQL query
+    cur.execute("SELECT 1")
+
+    # Fetch the results
+    rows = cur.fetchall()
+
+    # Print the results
+    for row in rows:
+        print(row[0])
+
+    # Close the cursor and the database connection
+    cur.close()
+    conn.close()
+
+
+async def main():
+    # Connect to the PostgreSQL database
+    conn = await asyncpg.connect(
+        host=PGCAT_HOST,
+        database="sharded_db",
+        user="sharding_user",
+        password="sharding_user",
+        port=PGCAT_PORT,
+    )
+
+    # Execute a SQL query
+    for _ in range(25):
+        rows = await conn.fetch("SELECT 1")
+
+        # Print the results
+        for row in rows:
+            print(row[0])
+
+    # Close the database connection
+    await conn.close()
+
+
+regular_main()
+asyncio.run(main())
--- a/tests/python/requirements.txt
+++ b/tests/python/requirements.txt
@@ -1,2 +1,11 @@
+asyncio==3.4.3
+asyncpg==0.27.0
+black==23.3.0
+click==8.1.3
+mypy-extensions==1.0.0
+packaging==23.1
+pathspec==0.11.1
+platformdirs==3.2.0
+psutil==5.9.1
 psycopg2==2.9.3
-psutil==5.9.1
+tomli==2.0.1
--- a/tests/ruby/admin_spec.rb
+++ b/tests/ruby/admin_spec.rb
@@ -14,12 +14,11 @@ describe "Admin" do
  describe "SHOW STATS" do
    context "clients connect and make one query" do
      it "updates *_query_time and *_wait_time" do
-        connections = Array.new(3) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
-        connections.each do |c|
-          Thread.new { c.async_exec("SELECT pg_sleep(0.25)") }
-        end
-        sleep(1)
-        connections.map(&:close)
+        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)
@@ -27,7 +26,7 @@ describe "Admin" do
        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 be_within(50).of(250)
+        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)
@@ -72,17 +71,15 @@ describe "Admin" do

    context "client connects but issues no queries" do
      it "only affects cl_idle stats" do
-        admin_conn = PG::connect(processes.pgcat.admin_connection_string)
-
-        before_test = admin_conn.async_exec("SHOW POOLS")[0]["sv_idle"]
        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(before_test)
+        expect(results["sv_idle"]).to eq("1")

        connections.map(&:close)
        sleep(1.1)
@@ -90,7 +87,7 @@ describe "Admin" do
        %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(before_test)
+        expect(results["sv_idle"]).to eq("1")
      end
    end

--- a/tests/ruby/helpers/pgcat_helper.rb
+++ b/tests/ruby/helpers/pgcat_helper.rb
@@ -41,24 +41,7 @@ module Helpers
            "1" => { "database" => "shard1", "servers" => [["localhost", primary1.port.to_s, "primary"]] },
            "2" => { "database" => "shard2", "servers" => [["localhost", primary2.port.to_s, "primary"]] },
          },
-          "users" => { "0" => user },
-          "plugins" => {
-            "intercept" => {
-              "enabled" => true,
-              "queries" => {
-                "0" => {
-                  "query" => "select current_database() as a, current_schemas(false) as b",
-                  "schema" => [
-                      ["a", "text"],
-                      ["b", "text"],
-                  ],
-                  "result" => [
-                    ["${DATABASE}", "{public}"],
-                  ]
-                }
-              }
-            }
-          }
+          "users" => { "0" => user }
        }
      }
      pgcat.update_config(pgcat_cfg)
@@ -118,7 +101,7 @@ module Helpers
      end
    end

-    def self.single_shard_setup(pool_name, pool_size, pool_mode="transaction", lb_mode="random", log_level="info", pool_settings={})
+    def self.single_shard_setup(pool_name, pool_size, pool_mode="transaction", lb_mode="random", log_level="info")
      user = {
        "password" => "sharding_user",
        "pool_size" => pool_size,
@@ -134,32 +117,28 @@ module Helpers
      replica1 = PgInstance.new(8432, user["username"], user["password"], "shard0")
      replica2 = PgInstance.new(9432, user["username"], user["password"], "shard0")

-      pool_config = {
-        "default_role" => "any",
-        "pool_mode" => pool_mode,
-        "load_balancing_mode" => lb_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 }
-      }
-
-      pool_config = pool_config.merge(pool_settings)
-
      # Main proxy configs
      pgcat_cfg["pools"] = {
-        "#{pool_name}" => pool_config,
+        "#{pool_name}" => {
+          "default_role" => "any",
+          "pool_mode" => pool_mode,
+          "load_balancing_mode" => lb_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)
--- a/tests/ruby/mirrors_spec.rb
+++ b/tests/ruby/mirrors_spec.rb
@@ -25,7 +25,7 @@ describe "Query Mirroing" do
    processes.pgcat.shutdown
  end

-  xit "can mirror a query" do
+  it "can mirror a query" do
    conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
    runs = 15
    runs.times { conn.async_exec("SELECT 1 + 2") }
--- a/tests/ruby/misc_spec.rb
+++ b/tests/ruby/misc_spec.rb
@@ -241,18 +241,6 @@ describe "Miscellaneous" do

        expect(processes.primary.count_query("DISCARD ALL")).to eq(10)
      end
-
-      it "Resets server roles correctly" 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.async_exec("SET statement_timeout to 5000")
-          conn.close
-        end
-
-        expect(processes.primary.count_query("RESET ROLE")).to eq(10)
-      end 
    end

    context "transaction mode" do
@@ -320,31 +308,6 @@ describe "Miscellaneous" do
        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
      end
    end
-
-    context "server cleanup disabled" do
-      let(:processes) { Helpers::Pgcat.single_shard_setup("sharded_db", 1, "transaction", "random", "info", { "cleanup_server_connections" => false }) }
-
-      it "will not clean up connection state" do
-        conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
-        processes.primary.reset_stats
-        conn.async_exec("SET statement_timeout TO 1000")
-        conn.close
-
-        puts processes.pgcat.logs
-        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
-      end
-
-      it "will not clean up prepared statements" do
-        conn = PG::connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
-        processes.primary.reset_stats
-        conn.async_exec("PREPARE prepared_q (int) AS SELECT $1")
-
-        conn.close
-
-        puts processes.pgcat.logs
-        expect(processes.primary.count_query("DISCARD ALL")).to eq(0)
-      end
-    end
  end

  describe "Idle client timeout" do
--- a/tests/ruby/plugins_spec.rb
+++ b/tests/ruby/plugins_spec.rb
@@ -1,14 +0,0 @@
-require_relative 'spec_helper'
-
-
-describe "Plugins" do
-  let(:processes) { Helpers::Pgcat.three_shard_setup("sharded_db", 5) }
-
-  context "intercept" do
-    it "will intercept an intellij query" do
-      conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
-      res = conn.exec("select current_database() as a, current_schemas(false) as b")
-      expect(res.values).to eq([["sharded_db", "{public}"]])
-    end
-  end
-end
Author	SHA1	Message	Date
Lev Kokotov	fd3623ff13	mm	2023-04-22 08:02:20 -07:00
Lev Kokotov	088f1a7dae	remove debug msg	2023-04-22 07:47:19 -07:00
Lev Kokotov	ab7ac16974	reqs	2023-04-22 07:40:21 -07:00