remove debug

lowercase config query
Actually plugins (#421 )
2026-03-23 17:36:28 +00:00 · 2023-05-03 16:46:48 -07:00 · 2023-05-03 16:46:07 -07:00 · 2023-05-03 16:13:45 -07:00 · 2023-05-03 10:00:05 -07:00 · 2023-05-03 09:13:05 -07:00
25 changed files with 2217 additions and 154 deletions
--- a/CONFIG.md
+++ b/CONFIG.md
@@ -49,6 +49,14 @@ 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
@@ -180,6 +188,22 @@ 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,6 +26,27 @@ 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"
@@ -212,6 +233,12 @@ 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"
@@ -223,6 +250,18 @@ dependencies = [
 "subtle",
 ]
 [[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"
@@ -275,6 +314,15 @@ 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"
@@ -410,6 +458,12 @@ 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"
@@ -434,6 +488,17 @@ 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"
@@ -522,6 +587,27 @@ 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"
@@ -542,6 +628,24 @@ 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"
@@ -589,6 +693,12 @@ 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"
@@ -604,6 +714,12 @@ 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"
@@ -629,6 +745,27 @@ 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"
@@ -737,9 +874,15 @@ 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.1"
+version = "1.0.2-alpha1"
 dependencies = [
 "arc-swap",
 "async-trait",
@@ -762,12 +905,15 @@ dependencies = [
 "once_cell",
 "parking_lot",
 "phf",
 "pin-project",
 "postgres-protocol",
 "rand",
 "regex",
 "rustls",
 "rustls-pemfile",
 "serde",
 "serde_derive",
 "serde_json",
 "sha-1",
 "sha2",
 "socket2",
@@ -775,7 +921,10 @@ dependencies = [
 "stringprep",
 "tokio",
 "tokio-rustls",
 "tokio-test",
 "toml",
 "trust-dns-resolver",
 "webpki-roots",
 ]
 [[package]]
@@ -820,6 +969,26 @@ 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"
@@ -865,6 +1034,12 @@ 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"
@@ -915,9 +1090,9 @@ dependencies = [
 [[package]]
 name = "regex"
-version = "1.8.0"
+version = "1.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ac6cf59af1067a3fb53fbe5c88c053764e930f932be1d71d3ffe032cbe147f59"
+checksum = "af83e617f331cc6ae2da5443c602dfa5af81e517212d9d611a5b3ba1777b5370"
 dependencies = [
 "aho-corasick",
 "memchr",
@@ -926,9 +1101,19 @@ 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 = "b6868896879ba532248f33598de5181522d8b3d9d724dfd230911e1a7d4822f5"
+checksum = "52e44394d2086d010551b14b53b1f24e31647570cd1deb0379e2c21b329aba00"
 dependencies = [
 "hostname",
 "quick-error",
 ]
 [[package]]
 name = "ring"
@@ -961,9 +1146,9 @@ dependencies = [
 [[package]]
 name = "rustls"
-version = "0.21.0"
+version = "0.21.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "07180898a28ed6a7f7ba2311594308f595e3dd2e3c3812fa0a80a47b45f17e5d"
+checksum = "c911ba11bc8433e811ce56fde130ccf32f5127cab0e0194e9c68c5a5b671791e"
 dependencies = [
 "log",
 "ring",
@@ -990,6 +1175,12 @@ 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"
@@ -1017,6 +1208,9 @@ 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"
@@ -1029,6 +1223,17 @@ 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.1"
@@ -1113,6 +1318,18 @@ 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]]
@@ -1168,6 +1385,26 @@ 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"
@@ -1235,6 +1472,30 @@ 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"
@@ -1297,9 +1558,21 @@ 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"
@@ -1309,6 +1582,51 @@ 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"
@@ -1354,6 +1672,17 @@ 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"
@@ -1446,6 +1775,21 @@ 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"
@@ -1551,3 +1895,12 @@ 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.1"
+version = "1.0.2-alpha1"
 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 = "1"
+serde = { version = "1", features = ["derive"] }
 serde_derive = "1"
 regex = "1"
 num_cpus = "1"
 once_cell = "1"
-sqlparser = "0.33.0"
+sqlparser = {version = "0.33", features = ["visitor"] }
 log = "0.4"
 arc-swap = "1"
 env_logger = "0.10"
@@ -39,6 +39,13 @@ nix = "0.26.2"
 atomic_enum = "0.2.0"
 postgres-protocol = "0.6.5"
 fallible-iterator = "0.2"
 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"
 [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. |
-| Client TLS | **Stable** | Clients can connect to the pooler using TLS/SSL. |
+| SSL/TLS | **Stable** | Clients can connect to the pooler using TLS. Pooler can connect to Postgres servers using TLS. |
 | 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/pgcat.toml
+++ b/pgcat.toml
@@ -23,6 +23,9 @@ 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
@@ -58,9 +61,15 @@ tcp_keepalives_count = 5
 tcp_keepalives_interval = 5
 # Path to TLS Certificate file to use for TLS connections
-# tls_certificate = "server.cert"
+# tls_certificate = ".circleci/server.cert"
 # Path to TLS private key file to use for TLS connections
-# tls_private_key = "server.key"
+# tls_private_key = ".circleci/server.key"
 # Enable/disable server TLS
 server_tls = false
 # Verify server certificate is completely authentic.
 verify_server_certificate = false
 # User name to access the virtual administrative database (pgbouncer or pgcat)
 # Connecting to that database allows running commands like `SHOW POOLS`, `SHOW DATABASES`, etc..
@@ -137,6 +146,53 @@ 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]
 [plugins.query_logger]
 enabled = false
 [plugins.table_access]
 enabled = false
 tables = [
  "pg_user",
  "pg_roles",
  "pg_database",
 ]
 [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}"],
 ]
 # 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]
@@ -206,6 +262,8 @@ 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,6 +77,8 @@ 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,6 +16,7 @@ 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;
@@ -539,6 +540,7 @@ 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()));
                }
@@ -565,6 +567,8 @@ where
                        }
                        Err(err) => {
                            wrong_password(&mut write, username).await?;
                            return Err(Error::ClientAuthPassthroughError(
                                err.to_string(),
                                client_identifier,
@@ -587,7 +591,15 @@ where
                    client_identifier
                );
-                let fetched_hash = refetch_auth_hash(&pool).await?;
+                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 new_password_hash = md5_hash_second_pass(&fetched_hash, &salt);
                // Ok password changed in server an auth is possible.
@@ -754,6 +766,9 @@ 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.
@@ -804,7 +819,25 @@ where
                'Q' => {
                    if query_router.query_parser_enabled() {
-                        query_router.infer(&message);
+                        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);
                        }
                    }
                }
@@ -812,7 +845,13 @@ where
                    self.buffer.put(&message[..]);
                    if query_router.query_parser_enabled() {
-                        query_router.infer(&message);
+                        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);
                        }
                    }
                    continue;
@@ -846,6 +885,18 @@ 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.
@@ -1074,6 +1125,27 @@ 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(
@@ -1113,6 +1185,14 @@ 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[..]);
                    }
@@ -1144,6 +1224,24 @@ where
                    'S' => {
                        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;
                            }
                            _ => (),
                        };
                        self.buffer.put(&message[..]);
                        let first_message_code = (*self.buffer.get(0).unwrap_or(&0)) as char;
--- a/src/config.rs
+++ b/src/config.rs
@@ -12,6 +12,7 @@ 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;
@@ -181,7 +182,9 @@ 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,
 }
@@ -194,12 +197,34 @@ 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 {
@@ -231,6 +256,12 @@ 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,
@@ -246,6 +277,9 @@ 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,
@@ -254,9 +288,17 @@ 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,
    // Support for auth query
    pub auth_query: Option<String>,
    pub auth_query_user: Option<String>,
    pub auth_query_password: Option<String>,
@@ -271,6 +313,10 @@ impl General {
        5432
    }
    pub fn default_server_lifetime() -> u64 {
        1000 * 60 * 60 * 24 // 24 hours
    }
    pub fn default_connect_timeout() -> u64 {
        1000
    }
@@ -298,6 +344,10 @@ impl General {
        60000
    }
    pub fn default_dns_max_ttl() -> u64 {
        30
    }
    pub fn default_healthcheck_timeout() -> u64 {
        1000
    }
@@ -340,13 +390,18 @@ 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,
        }
    }
 }
@@ -411,6 +466,8 @@ pub struct Pool {
    pub idle_timeout: Option<u64>,
    pub server_lifetime: Option<u64>,
    pub sharding_function: ShardingFunction,
    #[serde(default = "Pool::default_automatic_sharding_key")]
@@ -515,6 +572,10 @@ impl Pool {
            None => None,
        };
        for (_, user) in &self.users {
            user.validate()?;
        }
        Ok(())
    }
 }
@@ -539,6 +600,7 @@ impl Default for Pool {
            auth_query: None,
            auth_query_user: None,
            auth_query_password: None,
            server_lifetime: None,
        }
    }
 }
@@ -588,7 +650,7 @@ impl Shard {
        if primary_count > 1 {
            error!(
-                "Shard {} has more than on primary configured",
+                "Shard {} has more than one primary configured",
                self.database
            );
            return Err(Error::BadConfig);
@@ -617,6 +679,56 @@ impl Default for Shard {
    }
 }
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)]
 pub struct Plugins {
    pub intercept: Option<Intercept>,
    pub table_access: Option<TableAccess>,
    pub query_logger: Option<QueryLogger>,
 }
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)]
 pub struct Intercept {
    pub enabled: bool,
    pub queries: BTreeMap<String, Query>,
 }
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)]
 pub struct TableAccess {
    pub enabled: bool,
    pub tables: Vec<String>,
 }
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)]
 pub struct QueryLogger {
    pub enabled: bool,
 }
 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)]
 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 {
@@ -635,6 +747,7 @@ pub struct Config {
    pub path: String,
    pub general: General,
    pub plugins: Option<Plugins>,
    pub pools: HashMap<String, Pool>,
 }
@@ -672,6 +785,7 @@ impl Default for Config {
            path: Self::default_path(),
            general: General::default(),
            pools: HashMap::default(),
            plugins: None,
        }
    }
 }
@@ -791,6 +905,10 @@ 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);
@@ -809,6 +927,11 @@ 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
        );
        for (pool_name, pool_config) in &self.pools {
            // TODO: Make this output prettier (maybe a table?)
@@ -867,12 +990,26 @@ 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(),
                }
            );
            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
@@ -886,6 +1023,15 @@ 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(),
                    }
                );
            }
        }
    }
@@ -896,7 +1042,13 @@ 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);
        }
@@ -904,7 +1056,14 @@ 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);
            }
@@ -914,7 +1073,13 @@ 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);
                }
            }
@@ -1012,6 +1177,7 @@ 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) => {
@@ -1019,14 +1185,18 @@ pub async fn reload_config(client_server_map: ClientServerMap) -> Result<bool, E
            return Err(Error::BadConfig);
        }
    };
    let new_config = get_config();
-    if old_config.pools != new_config.pools {
+    match CachedResolver::from_config().await {
-        info!("Pool configuration changed");
+        Ok(_) => (),
        Err(err) => error!("DNS cache reinitialization error: {:?}", err),
    };
    if old_config != new_config {
        info!("Config changed, reloading");
        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
@@ -0,0 +1,410 @@
 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)]
+#[derive(Debug, PartialEq, Clone)]
 pub enum Error {
    SocketError(String),
    ClientSocketError(String, ClientIdentifier),
@@ -19,10 +19,13 @@ 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,11 +1,17 @@
 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,6 +36,7 @@ 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;
@@ -60,36 +61,19 @@ use std::str::FromStr;
 use std::sync::Arc;
 use tokio::sync::broadcast;
-mod admin;
+use pgcat::config::{get_config, reload_config, VERSION};
-mod auth_passthrough;
+use pgcat::dns_cache;
-mod client;
+use pgcat::messages::configure_socket;
-mod config;
+use pgcat::pool::{ClientServerMap, ConnectionPool};
-mod constants;
+use pgcat::prometheus::start_metric_server;
-mod errors;
+use pgcat::stats::{Collector, Reporter, REPORTER};
 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>> {
-    multi_logger::MultiLogger::init().unwrap();
+    pgcat::multi_logger::MultiLogger::init().unwrap();
    info!("Welcome to PgCat! Meow. (Version {})", VERSION);
-    if !query_router::QueryRouter::setup() {
+    if !pgcat::query_router::QueryRouter::setup() {
        error!("Could not setup query router");
        std::process::exit(exitcode::CONFIG);
    }
@@ -107,7 +91,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
        let runtime = Builder::new_multi_thread().worker_threads(1).build()?;
        runtime.block_on(async {
-            match config::parse(&config_file).await {
+            match pgcat::config::parse(&config_file).await {
                Ok(_) => (),
                Err(err) => {
                    error!("Config parse error: {:?}", err);
@@ -166,6 +150,12 @@ 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(_) => (),
@@ -295,7 +285,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
                    tokio::task::spawn(async move {
                        let start = chrono::offset::Utc::now().naive_utc();
-                        match client::client_entrypoint(
+                        match pgcat::client::client_entrypoint(
                            socket,
                            client_server_map,
                            shutdown_rx,
@@ -326,7 +316,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
                            Err(err) => {
                                match err {
-                                    errors::Error::ClientBadStartup => debug!("Client disconnected with error {:?}", err),
+                                    pgcat::errors::Error::ClientBadStartup => debug!("Client disconnected with error {:?}", err),
                                    _ => warn!("Client disconnected with error {:?}", err),
                                }
--- a/src/messages.rs
+++ b/src/messages.rs
@@ -20,6 +20,10 @@ pub enum DataType {
    Text,
    Int4,
    Numeric,
    Bool,
    Oid,
    AnyArray,
    Any,
 }
 impl From<&DataType> for i32 {
@@ -28,6 +32,10 @@ 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,
        }
    }
 }
@@ -116,7 +124,10 @@ 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(stream: &mut TcpStream, user: &str, database: &str) -> Result<(), Error> {
+pub async fn startup<S>(stream: &mut S, user: &str, database: &str) -> Result<(), Error>
 where
    S: tokio::io::AsyncWrite + std::marker::Unpin,
 {
    let mut bytes = BytesMut::with_capacity(25);
    bytes.put_i32(196608); // Protocol number
@@ -150,6 +161,21 @@ pub async fn startup(stream: &mut TcpStream, user: &str, database: &str) -> Resu
    }
 }
 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();
@@ -425,6 +451,10 @@ 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);
@@ -463,6 +493,29 @@ 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());
@@ -505,6 +558,29 @@ 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/plugins/intercept.rs
+++ b/src/plugins/intercept.rs
@@ -0,0 +1,288 @@
 //! The intercept plugin.
 //!
 //! It intercepts queries and returns fake results.
 use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use bytes::{BufMut, BytesMut};
 use once_cell::sync::Lazy;
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
 use sqlparser::ast::Statement;
 use std::collections::HashMap;
 use log::{debug, info};
 use std::sync::Arc;
 use crate::{
    config::Intercept as InterceptConfig,
    errors::Error,
    messages::{command_complete, data_row_nullable, row_description, DataType},
    plugins::{Plugin, PluginOutput},
    pool::{PoolIdentifier, PoolMap},
    query_router::QueryRouter,
 };
 pub static CONFIG: Lazy<ArcSwap<HashMap<PoolIdentifier, InterceptConfig>>> =
    Lazy::new(|| ArcSwap::from_pointee(HashMap::new()));
 /// Check if the interceptor plugin has been enabled.
 pub fn enabled() -> bool {
    !CONFIG.load().is_empty()
 }
 pub fn setup(intercept_config: &InterceptConfig, pools: &PoolMap) {
    let mut config = HashMap::new();
    for (identifier, _) in pools.iter() {
        let mut intercept_config = intercept_config.clone();
        intercept_config.substitute(&identifier.db, &identifier.user);
        config.insert(identifier.clone(), intercept_config);
    }
    CONFIG.store(Arc::new(config));
    info!("Intercepting {} queries", intercept_config.queries.len());
 }
 pub fn disable() {
    CONFIG.store(Arc::new(HashMap::new()));
 }
 // 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;
 #[async_trait]
 impl Plugin for Intercept {
    async fn run(
        &mut self,
        query_router: &QueryRouter,
        ast: &Vec<Statement>,
    ) -> Result<PluginOutput, Error> {
        if ast.is_empty() {
            return Ok(PluginOutput::Allow);
        }
        let mut result = BytesMut::new();
        let query_map = match CONFIG.load().get(&PoolIdentifier::new(
            &query_router.pool_settings().db,
            &query_router.pool_settings().user.username,
        )) {
            Some(query_map) => query_map.clone(),
            None => return Ok(PluginOutput::Allow),
        };
        for q in ast {
            // Normalization
            let q = q.to_string().to_ascii_lowercase();
            for (_, target) in query_map.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)
        }
    }
 }
 /// Make IntelliJ SQL plugin believe it's talking to an actual database
 /// instead of PgCat.
 #[allow(dead_code)]
 fn fool_datagrip(database: &str, user: &str) -> Value {
    json!([
        {
            "query": "select current_database() as a, current_schemas(false) as b",
            "schema": [
                {
                    "name": "a",
                    "data_type": "text",
                },
                {
                    "name": "b",
                    "data_type": "anyarray",
                },
            ],
            "result": [
                [database, "{public}"],
            ],
        },
        {
            "query": "select current_database(), current_schema(), current_user",
            "schema": [
                {
                    "name": "current_database",
                    "data_type": "text",
                },
                {
                    "name": "current_schema",
                    "data_type": "text",
                },
                {
                    "name": "current_user",
                    "data_type": "text",
                }
            ],
            "result": [
                ["sharded_db", "public", "sharding_user"],
            ],
        },
        {
            "query": "select cast(n.oid as bigint) as id, datname as name, d.description, datistemplate as is_template, datallowconn as allow_connections, pg_catalog.pg_get_userbyid(n.datdba) as \"owner\" from pg_catalog.pg_database as n left join pg_catalog.pg_shdescription as d on n.oid = d.objoid order by case when datname = pg_catalog.current_database() then -cast(1 as bigint) else cast(n.oid as bigint) end",
            "schema": [
                {
                    "name": "id",
                    "data_type": "oid",
                },
                {
                    "name": "name",
                    "data_type": "text",
                },
                {
                    "name": "description",
                    "data_type": "text",
                },
                {
                    "name": "is_template",
                    "data_type": "bool",
                },
                {
                    "name": "allow_connections",
                    "data_type": "bool",
                },
                {
                    "name": "owner",
                    "data_type": "text",
                }
            ],
            "result": [
                ["16387", database, "", "f", "t", user],
            ]
        },
        {
            "query": "select cast(r.oid as bigint) as role_id, rolname as role_name, rolsuper as is_super, rolinherit as is_inherit, rolcreaterole as can_createrole, rolcreatedb as can_createdb, rolcanlogin as can_login, rolreplication as is_replication, rolconnlimit as conn_limit, rolvaliduntil as valid_until, rolbypassrls as bypass_rls, rolconfig as config, d.description from pg_catalog.pg_roles as r left join pg_catalog.pg_shdescription as d on d.objoid = r.oid",
            "schema": [
                {
                    "name": "role_id",
                    "data_type": "oid",
                },
                {
                    "name": "role_name",
                    "data_type": "text",
                },
                {
                    "name": "is_super",
                    "data_type": "bool",
                },
                {
                    "name": "is_inherit",
                    "data_type": "bool",
                },
                {
                    "name": "can_createrole",
                    "data_type": "bool",
                },
                {
                    "name": "can_createdb",
                    "data_type": "bool",
                },
                {
                    "name": "can_login",
                    "data_type": "bool",
                },
                {
                    "name": "is_replication",
                    "data_type": "bool",
                },
                {
                    "name": "conn_limit",
                    "data_type": "int4",
                },
                {
                    "name": "valid_until",
                    "data_type": "text",
                },
                {
                    "name": "bypass_rls",
                    "data_type": "bool",
                },
                {
                    "name": "config",
                    "data_type": "text",
                },
                {
                    "name": "description",
                    "data_type": "text",
                },
            ],
            "result": [
                ["10", "postgres", "f", "t", "f", "f", "t", "f", "-1", "", "f", "", ""],
                ["16419", user, "f", "t", "f", "f", "t", "f", "-1", "", "f", "", ""],
            ]
        }
    ])
 }
--- a/src/plugins/mod.rs
+++ b/src/plugins/mod.rs
@@ -0,0 +1,43 @@
 //! 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 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/query_logger.rs
+++ b/src/plugins/query_logger.rs
@@ -0,0 +1,49 @@
 //! Log all queries to stdout (or somewhere else, why not).
 use crate::{
    errors::Error,
    plugins::{Plugin, PluginOutput},
    query_router::QueryRouter,
 };
 use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use log::info;
 use once_cell::sync::Lazy;
 use sqlparser::ast::Statement;
 use std::sync::Arc;
 static ENABLED: Lazy<ArcSwap<bool>> = Lazy::new(|| ArcSwap::from_pointee(false));
 pub struct QueryLogger;
 pub fn setup() {
    ENABLED.store(Arc::new(true));
    info!("Logging queries to stdout");
 }
 pub fn disable() {
    ENABLED.store(Arc::new(false));
 }
 pub fn enabled() -> bool {
    **ENABLED.load()
 }
 #[async_trait]
 impl Plugin for QueryLogger {
    async fn run(
        &mut self,
        _query_router: &QueryRouter,
        ast: &Vec<Statement>,
    ) -> Result<PluginOutput, Error> {
        let query = ast
            .iter()
            .map(|q| q.to_string())
            .collect::<Vec<String>>()
            .join("; ");
        info!("{}", query);
        Ok(PluginOutput::Allow)
    }
 }
--- a/src/plugins/table_access.rs
+++ b/src/plugins/table_access.rs
@@ -0,0 +1,73 @@
 //! 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::{
    config::TableAccess as TableAccessConfig,
    errors::Error,
    plugins::{Plugin, PluginOutput},
    query_router::QueryRouter,
 };
 use log::{debug, info};
 use arc_swap::ArcSwap;
 use core::ops::ControlFlow;
 use once_cell::sync::Lazy;
 use std::sync::Arc;
 static CONFIG: Lazy<ArcSwap<Vec<String>>> = Lazy::new(|| ArcSwap::from_pointee(vec![]));
 pub fn setup(config: &TableAccessConfig) {
    CONFIG.store(Arc::new(config.tables.clone()));
    info!("Blocking access to {} tables", config.tables.len());
 }
 pub fn enabled() -> bool {
    !CONFIG.load().is_empty()
 }
 pub fn disable() {
    CONFIG.store(Arc::new(vec![]));
 }
 pub struct TableAccess;
 #[async_trait]
 impl Plugin for TableAccess {
    async fn run(
        &mut self,
        _query_router: &QueryRouter,
        ast: &Vec<Statement>,
    ) -> Result<PluginOutput, Error> {
        let mut found = None;
        let forbidden_tables = CONFIG.load();
        visit_relations(ast, |relation| {
            let relation = relation.to_string();
            let parts = relation.split(".").collect::<Vec<&str>>();
            let table_name = parts.last().unwrap();
            if forbidden_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
@@ -61,6 +61,8 @@ 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 {
@@ -91,6 +93,7 @@ pub struct PoolSettings {
    // Connecting user.
    pub user: User,
    pub db: String,
    // Default server role to connect to.
    pub default_role: Option<Role>,
@@ -138,6 +141,7 @@ 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,
@@ -311,21 +315,34 @@ impl ConnectionPool {
                        if let Some(apt) = &auth_passthrough {
                            match apt.fetch_hash(&address).await {
-				Ok(ok) => {
+                                Ok(ok) => {
-				    if let Some(ref pool_auth_hash_value) = *(pool_auth_hash.read()) {
+                                    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 \
+                                        if ok != *pool_auth_hash_value {
-						   be done using last obtained hash. Server: {}:{}, Database: {}", server.host, server.port, shard.database);
+                                            warn!(
-					}
+                                                "Hash is not the same across shards \
-				    }
+                                                of the same pool, client auth will \
-				    debug!("Hash obtained for {:?}", address);
+                                                be done using last obtained hash. \
-				    {
+                                                Server: {}:{}, Database: {}",
-					let mut pool_auth_hash = pool_auth_hash.write();
+                                                server.host, server.port, shard.database,
-					*pool_auth_hash = Some(ok.clone());
+                                            );
-				    }
+                                        }
-				},
+                                    }
-				Err(err) => warn!("Could not obtain password hashes using auth_query config, ignoring. Error: {:?}", err),
+
-			    }
+                                    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(
@@ -347,14 +364,31 @@ 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 reaper rate: {}ms", 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)
                            .build(manager)
-                            .await
+                            .await?;
                            .unwrap();
                        pools.push(pool);
                        servers.push(address);
@@ -390,6 +424,7 @@ 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),
@@ -434,6 +469,32 @@ impl ConnectionPool {
            }
        }
        if let Some(ref plugins) = config.plugins {
            if let Some(ref intercept) = plugins.intercept {
                if intercept.enabled {
                    crate::plugins::intercept::setup(intercept, &new_pools);
                } else {
                    crate::plugins::intercept::disable();
                }
            }
            if let Some(ref table_access) = plugins.table_access {
                if table_access.enabled {
                    crate::plugins::table_access::setup(table_access);
                } else {
                    crate::plugins::table_access::disable();
                }
            }
            if let Some(ref query_logger) = plugins.query_logger {
                if query_logger.enabled {
                    crate::plugins::query_logger::setup();
                } else {
                    crate::plugins::query_logger::disable();
                }
            }
        }
        POOLS.store(Arc::new(new_pools.clone()));
        Ok(())
    }
--- a/src/query_router.rs
+++ b/src/query_router.rs
@@ -6,13 +6,19 @@ use once_cell::sync::OnceCell;
 use regex::{Regex, RegexSet};
 use sqlparser::ast::Statement::{Query, StartTransaction};
 use sqlparser::ast::{
-    BinaryOperator, Expr, Ident, JoinConstraint, JoinOperator, SetExpr, TableFactor, Value,
+    BinaryOperator, Expr, Ident, JoinConstraint, JoinOperator, SetExpr, Statement, 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, query_logger, table_access, Intercept, Plugin, PluginOutput, QueryLogger,
    TableAccess,
 };
 use crate::pool::PoolSettings;
 use crate::sharding::Sharder;
@@ -129,6 +135,10 @@ 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);
@@ -324,10 +334,7 @@ impl QueryRouter {
        Some((command, value))
    }
-    /// Try to infer which server to connect to based on the contents of the query.
+    pub fn parse(message: &BytesMut) -> Result<Vec<sqlparser::ast::Statement>, Error> {
    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;
@@ -353,28 +360,29 @@ impl QueryRouter {
                query
            }
-            _ => return false,
+            _ => return Err(Error::UnsupportedStatement),
        };
-        let ast = match Parser::parse_sql(&PostgreSqlDialect {}, &query) {
+        match Parser::parse_sql(&PostgreSqlDialect {}, &query) {
-            Ok(ast) => ast,
+            Ok(ast) => Ok(ast),
            Err(err) => {
                // SELECT ... FOR UPDATE won't get parsed correctly.
                debug!("{}: {}", err, query);
-                self.active_role = Some(Role::Primary);
+                Err(Error::QueryRouterParserError(err.to_string()))
                return false;
            }
-        };
+        }
    }
-        debug!("AST: {:?}", ast);
+    /// 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");
        if ast.is_empty() {
            // That's weird, no idea, let's go to primary
            self.active_role = Some(Role::Primary);
-            return false;
+            return Err(Error::QueryRouterParserError("empty query".into()));
        }
-        for q in &ast {
+        for q in ast {
            match q {
                // All transactions go to the primary, probably a write.
                StartTransaction { .. } => {
@@ -418,7 +426,7 @@ impl QueryRouter {
            };
        }
-        true
+        Ok(())
    }
    /// Parse the shard number from the Bind message
@@ -783,6 +791,34 @@ impl QueryRouter {
        }
    }
    /// Add your plugins here and execute them.
    pub async fn execute_plugins(&self, ast: &Vec<Statement>) -> Result<PluginOutput, Error> {
        if query_logger::enabled() {
            let mut query_logger = QueryLogger {};
            let _ = query_logger.run(&self, ast).await;
        }
        if intercept::enabled() {
            let mut intercept = Intercept {};
            let result = intercept.run(&self, ast).await;
            if let Ok(PluginOutput::Intercept(output)) = result {
                return Ok(PluginOutput::Intercept(output));
            }
        }
        if table_access::enabled() {
            let mut table_access = TableAccess {};
            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,
@@ -810,11 +846,22 @@ impl QueryRouter {
    /// Should we attempt to parse queries?
    pub fn query_parser_enabled(&self) -> bool {
        let enabled = match self.query_parser_enabled {
-            None => self.pool_settings.query_parser_enabled,
+            None => {
-            Some(value) => value,
+                debug!(
-        };
+                    "Using pool settings, query_parser_enabled: {}",
                    self.pool_settings.query_parser_enabled
                );
                self.pool_settings.query_parser_enabled
            }
-        debug!("Query parser enabled: {}", enabled);
+            Some(value) => {
                debug!(
                    "Using query parser override, query_parser_enabled: {}",
                    value
                );
                value
            }
        };
        enabled
    }
@@ -862,7 +909,7 @@ mod test {
        for query in queries {
            // It's a recognized query
-            assert!(qr.infer(&query));
+            assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
            assert_eq!(qr.role(), Some(Role::Replica));
        }
    }
@@ -881,7 +928,7 @@ mod test {
        for query in queries {
            // It's a recognized query
-            assert!(qr.infer(&query));
+            assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
            assert_eq!(qr.role(), Some(Role::Primary));
        }
    }
@@ -893,7 +940,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(&query));
+        assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
        assert_eq!(qr.role(), None);
    }
@@ -913,7 +960,7 @@ mod test {
        res.put(prepared_stmt);
        res.put_i16(0);
-        assert!(qr.infer(&res));
+        assert!(qr.infer(&QueryRouter::parse(&res).unwrap()).is_ok());
        assert_eq!(qr.role(), Some(Role::Replica));
    }
@@ -1077,11 +1124,11 @@ mod test {
        assert_eq!(qr.role(), None);
        let query = simple_query("INSERT INTO test_table VALUES (1)");
-        assert!(qr.infer(&query));
+        assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
        assert_eq!(qr.role(), Some(Role::Primary));
        let query = simple_query("SELECT * FROM test_table");
-        assert!(qr.infer(&query));
+        assert!(qr.infer(&QueryRouter::parse(&query).unwrap()).is_ok());
        assert_eq!(qr.role(), Some(Role::Replica));
        assert!(qr.query_parser_enabled());
@@ -1113,6 +1160,7 @@ mod test {
            auth_query: None,
            auth_query_password: None,
            auth_query_user: None,
            db: "test".to_string(),
        };
        let mut qr = QueryRouter::new();
        assert_eq!(qr.active_role, None);
@@ -1142,15 +1190,24 @@ mod test {
        QueryRouter::setup();
        let mut qr = QueryRouter::new();
-        assert!(qr.infer(&simple_query("BEGIN; SELECT 1; COMMIT;")));
+        assert!(qr
            .infer(&QueryRouter::parse(&simple_query("BEGIN; SELECT 1; COMMIT;")).unwrap())
            .is_ok());
        assert_eq!(qr.role(), Role::Primary);
-        assert!(qr.infer(&simple_query("SELECT 1; SELECT 2;")));
+        assert!(qr
            .infer(&QueryRouter::parse(&simple_query("SELECT 1; SELECT 2;")).unwrap())
            .is_ok());
        assert_eq!(qr.role(), Role::Replica);
-        assert!(qr.infer(&simple_query(
+        assert!(qr
-            "SELECT 123; INSERT INTO t VALUES (5); SELECT 1;"
+            .infer(
-        )));
+                &QueryRouter::parse(&simple_query(
                    "SELECT 123; INSERT INTO t VALUES (5); SELECT 1;"
                ))
                .unwrap()
            )
            .is_ok());
        assert_eq!(qr.role(), Role::Primary);
    }
@@ -1177,6 +1234,7 @@ mod test {
            auth_query: None,
            auth_query_password: None,
            auth_query_user: None,
            db: "test".to_string(),
        };
        let mut qr = QueryRouter::new();
        qr.update_pool_settings(pool_settings.clone());
@@ -1208,47 +1266,84 @@ mod test {
        qr.pool_settings.automatic_sharding_key = Some("data.id".to_string());
        qr.pool_settings.shards = 3;
-        assert!(qr.infer(&simple_query("SELECT * FROM data WHERE id = 5")));
+        assert!(qr
            .infer(&QueryRouter::parse(&simple_query("SELECT * FROM data WHERE id = 5")).unwrap())
            .is_ok());
        assert_eq!(qr.shard(), 2);
-        assert!(qr.infer(&simple_query(
+        assert!(qr
-            "SELECT one, two, three FROM public.data WHERE id = 6"
+            .infer(
-        )));
+                &QueryRouter::parse(&simple_query(
                    "SELECT one, two, three FROM public.data WHERE id = 6"
                ))
                .unwrap()
            )
            .is_ok());
        assert_eq!(qr.shard(), 0);
-        assert!(qr.infer(&simple_query(
+        assert!(qr
-            "SELECT * FROM data
+            .infer(
                &QueryRouter::parse(&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(&simple_query(
+        assert!(qr
-            "SELECT * FROM t2 INNER JOIN data ON id = 6 AND data.id = t2.data_id"
+            .infer(
-        )));
+                &QueryRouter::parse(&simple_query(
                    "SELECT * FROM t2 INNER JOIN data ON id = 6 AND data.id = t2.data_id"
                ))
                .unwrap()
            )
            .is_ok());
        assert_eq!(qr.shard(), 2);
-        assert!(qr.infer(&simple_query(
+        assert!(qr
-            r#"SELECT * FROM "public"."data" WHERE "id" = 6"#
+            .infer(
-        )));
+                &QueryRouter::parse(&simple_query(
                    r#"SELECT * FROM "public"."data" WHERE "id" = 6"#
                ))
                .unwrap()
            )
            .is_ok());
        assert_eq!(qr.shard(), 0);
-        assert!(qr.infer(&simple_query(
+        assert!(qr
-            r#"SELECT * FROM "public"."data" WHERE "data"."id" = 5"#
+            .infer(
-        )));
+                &QueryRouter::parse(&simple_query(
                    r#"SELECT * FROM "public"."data" WHERE "data"."id" = 5"#
                ))
                .unwrap()
            )
            .is_ok());
        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(&simple_query(
+        assert!(qr
-            "SELECT * FROM table_x WHERE unique_enough_column_name = 6"
+            .infer(
-        )));
+                &QueryRouter::parse(&simple_query(
                    "SELECT * FROM table_x WHERE unique_enough_column_name = 6"
                ))
                .unwrap()
            )
            .is_ok());
        assert_eq!(qr.shard(), 0);
-        assert!(qr.infer(&simple_query("SELECT * FROM table_y WHERE another_key = 5")));
+        assert!(qr
            .infer(
                &QueryRouter::parse(&simple_query("SELECT * FROM table_y WHERE another_key = 5"))
                    .unwrap()
            )
            .is_ok());
        assert_eq!(qr.shard(), 0);
    }
@@ -1272,11 +1367,40 @@ mod test {
        qr.pool_settings.automatic_sharding_key = Some("data.id".to_string());
        qr.pool_settings.shards = 3;
-        assert!(qr.infer(&simple_query(stmt)));
+        assert!(qr
            .infer(&QueryRouter::parse(&simple_query(stmt)).unwrap())
            .is_ok());
        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::TableAccess;
        let ta = TableAccess {
            enabled: true,
            tables: vec![String::from("pg_database")],
        };
        crate::plugins::table_access::setup(&ta);
        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::Deny(
                "permission for table \"pg_database\" denied".to_string()
            ))
        );
    }
 }
--- a/src/server.rs
+++ b/src/server.rs
@@ -7,22 +7,101 @@ 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::{AsyncReadExt, BufReader};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, BufStream};
-use tokio::net::{
+use tokio::net::TcpStream;
-    tcp::{OwnedReadHalf, OwnedWriteHalf},
+use tokio_rustls::rustls::{OwnedTrustAnchor, RootCertStore};
-    TcpStream,
+use tokio_rustls::{client::TlsStream, TlsConnector};
 };
-use crate::config::{Address, User};
+use crate::config::{get_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),
        }
    }
 }
 /// Server state.
 pub struct Server {
@@ -30,11 +109,8 @@ pub struct Server {
    /// port, e.g. 5432, and role, e.g. primary or replica.
    address: Address,
-    /// Buffered read socket.
+    /// Server TCP connection.
-    read: BufReader<OwnedReadHalf>,
+    stream: BufStream<StreamInner>,
    /// 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,
@@ -74,6 +150,9 @@ pub struct Server {
    last_activity: SystemTime,
    mirror_manager: Option<MirroringManager>,
    // Associated addresses used
    addr_set: Option<AddrSet>,
 }
 impl Server {
@@ -87,6 +166,24 @@ impl Server {
        stats: Arc<ServerStats>,
        auth_hash: Arc<RwLock<Option<String>>>,
    ) -> 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,
@@ -98,8 +195,88 @@ 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
@@ -245,7 +422,7 @@ impl Server {
                            let sasl_type = String::from_utf8_lossy(&sasl_auth[..sasl_len - 2]);
-                            if sasl_type == SCRAM_SHA_256 {
+                            if sasl_type.contains(SCRAM_SHA_256) {
                                debug!("Using {}", SCRAM_SHA_256);
                                // Generate client message.
@@ -268,7 +445,7 @@ impl Server {
                                res.put_i32(sasl_response.len() as i32);
                                res.put(sasl_response);
-                                write_all(&mut stream, res).await?;
+                                write_all_flush(&mut stream, &res).await?;
                            } else {
                                error!("Unsupported SCRAM version: {}", sasl_type);
                                return Err(Error::ServerError);
@@ -299,7 +476,7 @@ impl Server {
                            res.put_i32(4 + sasl_response.len() as i32);
                            res.put(sasl_response);
-                            write_all(&mut stream, res).await?;
+                            write_all_flush(&mut stream, &res).await?;
                        }
                        SASL_FINAL => {
@@ -443,12 +620,9 @@ impl Server {
                        }
                    };
                    let (read, write) = stream.into_split();
                    let mut server = Server {
                        address: address.clone(),
-                        read: BufReader::new(read),
+                        stream: BufStream::new(stream),
                        write,
                        buffer: BytesMut::with_capacity(8196),
                        server_info,
                        process_id,
@@ -458,6 +632,7 @@ impl Server {
                        bad: false,
                        needs_cleanup: false,
                        client_server_map,
                        addr_set,
                        connected_at: chrono::offset::Utc::now().naive_utc(),
                        stats,
                        application_name: String::new(),
@@ -515,7 +690,7 @@ impl Server {
        bytes.put_i32(process_id);
        bytes.put_i32(secret_key);
-        write_all(&mut stream, bytes).await
+        write_all_flush(&mut stream, &bytes).await
    }
    /// Send messages to the server from the client.
@@ -523,7 +698,7 @@ impl Server {
        self.mirror_send(messages);
        self.stats().data_sent(messages.len());
-        match write_all_half(&mut self.write, messages).await {
+        match write_all_flush(&mut self.stream, &messages).await {
            Ok(_) => {
                // Successfully sent to server
                self.last_activity = SystemTime::now();
@@ -542,7 +717,7 @@ impl Server {
    /// 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.read).await {
+            let mut message = match read_message(&mut self.stream).await {
                Ok(message) => message,
                Err(err) => {
                    error!("Terminating server because of: {:?}", err);
@@ -698,7 +873,23 @@ 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 {
-        self.bad
+        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
    }
    /// Get server startup information to forward it to the client.
@@ -935,13 +1126,13 @@ impl Drop for Server {
        // Update statistics
        self.stats.disconnect();
-        let mut bytes = BytesMut::with_capacity(4);
+        let mut bytes = BytesMut::with_capacity(5);
        bytes.put_u8(b'X');
        bytes.put_i32(4);
-        match self.write.try_write(&bytes) {
+        match self.stream.get_mut().try_write(&bytes) {
-            Ok(_) => (),
+            Ok(5) => (),
-            Err(_) => debug!("Dirty shutdown"),
+            _ => debug!("Dirty shutdown"),
        };
        // Should not matter.
--- a/src/tls.rs
+++ b/src/tls.rs
@@ -4,7 +4,12 @@ use rustls_pemfile::{certs, read_one, Item};
 use std::iter;
 use std::path::Path;
 use std::sync::Arc;
-use tokio_rustls::rustls::{self, Certificate, PrivateKey};
+use std::time::SystemTime;
 use tokio_rustls::rustls::{
    self,
    client::{ServerCertVerified, ServerCertVerifier},
    Certificate, PrivateKey, ServerName,
 };
 use tokio_rustls::TlsAcceptor;
 use crate::config::get_config;
@@ -64,3 +69,19 @@ 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/ruby/admin_spec.rb
+++ b/tests/ruby/admin_spec.rb
@@ -71,15 +71,17 @@ 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("1")
+        expect(results["sv_idle"]).to eq(before_test)
        connections.map(&:close)
        sleep(1.1)
@@ -87,7 +89,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("1")
+        expect(results["sv_idle"]).to eq(before_test)
      end
    end
--- a/tests/ruby/mirrors_spec.rb
+++ b/tests/ruby/mirrors_spec.rb
@@ -25,7 +25,7 @@ describe "Query Mirroing" do
    processes.pgcat.shutdown
  end
-  it "can mirror a query" do
+  xit "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/plugins_spec.rb
+++ b/tests/ruby/plugins_spec.rb
@@ -0,0 +1,14 @@
 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	35c20dc9b4	remove debug	2023-05-03 16:46:48 -07:00
Lev Kokotov	72ce2be5c7	lowercase config query	2023-05-03 16:46:07 -07:00
Lev Kokotov	811885f464	Actually plugins (#421 ) * more plugins * clean up * fix tests * fix flakey test	2023-05-03 16:13:45 -07:00
dependabot[bot]	d5e329fec5	chore(deps): bump regex from 1.8.0 to 1.8.1 (#413 ) Bumps [regex](https://github.com/rust-lang/regex) from 1.8.0 to 1.8.1. - [Release notes](https://github.com/rust-lang/regex/releases) - [Changelog](https://github.com/rust-lang/regex/blob/master/CHANGELOG.md) - [Commits](https://github.com/rust-lang/regex/commits/1.8.1) --- updated-dependencies: - dependency-name: regex dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2023-05-03 10:00:05 -07:00
Lev Kokotov	09e54e1175	Plugins! (#420 ) * Some queries * Plugins!! * cleanup * actual names * the actual plugins * comment * fix tests * Tests * unused errors * Increase reaper rate to actually enforce settings * ok	2023-05-03 09:13:05 -07:00
dependabot[bot]	23819c8549	chore(deps): bump rustls from 0.21.0 to 0.21.1 (#419 ) Bumps [rustls](https://github.com/rustls/rustls) from 0.21.0 to 0.21.1. - [Release notes](https://github.com/rustls/rustls/releases) - [Changelog](https://github.com/rustls/rustls/blob/main/RELEASE_NOTES.md) - [Commits](https://github.com/rustls/rustls/compare/v/0.21.0...v/0.21.1) --- updated-dependencies: - dependency-name: rustls dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2023-05-02 07:32:44 -07:00
Jose Fernández	7dfbd993f2	Add dns_cache for server addresses as in pgbouncer (#249 ) * Add dns_cache so server addresses are cached and invalidated when DNS changes. Adds a module to deal with dns_cache feature. It's main struct is CachedResolver, which is a simple thread safe hostname <-> Ips cache with the ability to refresh resolutions every `dns_max_ttl` seconds. This way, a client can check whether its ip address has changed. * Allow reloading dns cached * Add documentation for dns_cached	2023-05-02 10:26:40 +02:00
Lev Kokotov	3601130ba1	Readme update (#418 ) * Readme update * m * wording	2023-04-30 09:44:25 -07:00
Lev Kokotov	0d504032b2	Server TLS (#417 ) * Server TLS * Finish up TLS * thats it * diff * remove dead code * maybe? * dirty shutdown * skip flakey test * remove unused error * fetch config once	2023-04-30 09:41:46 -07:00
Lev Kokotov	4a87b4807d	Add more pool settings (#416 ) * Add some pool settings * fmt	2023-04-26 16:33:26 -07:00
Shawn	cb5ff40a59	fix typo (#415 ) chore: typo	2023-04-26 08:28:54 -07:00