From dfcc0c8d333e788b0142b9c74cd0513409bd3860 Mon Sep 17 00:00:00 2001
From: danielshih <dog830228@gmail.com>
Date: Mon, 27 Jul 2026 01:03:29 +0000
Subject: [PATCH] refactor: streamline replication message handling and SQL
 preparation functions refactor: remove unused methods and streamline
 BufferReader and BufferWriter implementations refactor(error): eliminate
 generic error type and related methods for clarity refactor(lib): adjust
 visibility of PG_EPOCH_OFFSET_SECS constant refactor(lsn): simplify
 SharedLsnFeedback by removing Clone implementation and related tests
 refactor(sql_builder): change build_sql_options to crate visibility
 fix(stream): update error handling to use protocol-specific error type
 refactor(types): update CachePadded and PG_EPOCH_OFFSET_SECS to crate
 visibility

---
 examples/arbitrary-fuzzing/Cargo.lock |  69 ++--
 fuzz/Cargo.lock                       | 561 ++++++++++++++++----------
 macros/Cargo.toml                     |   2 +-
 src/buffer.rs                         |  77 +---
 src/connection/libpq.rs               |  30 +-
 src/connection/native/connection.rs   |  28 +-
 src/connection/native/copy.rs         |   7 +-
 src/connection/native/error.rs        |  62 +--
 src/connection/native/wire.rs         |   4 +-
 src/error.rs                          |  44 +-
 src/lib.rs                            |  20 +-
 src/lsn.rs                            |  77 +---
 src/protocol.rs                       |  50 +++
 src/sql_builder.rs                    |  28 +-
 src/stream.rs                         |  46 +--
 src/types.rs                          |  52 ++-
 16 files changed, 581 insertions(+), 576 deletions(-)

diff --git a/examples/arbitrary-fuzzing/Cargo.lock b/examples/arbitrary-fuzzing/Cargo.lock
index e83ae9d..b86793a 100644
--- a/examples/arbitrary-fuzzing/Cargo.lock
+++ b/examples/arbitrary-fuzzing/Cargo.lock
@@ -28,9 +28,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.17.1"
+version = "1.17.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4342d8937fc7e5dd9b1c60292261c0670c882a2cd1719cfc11b1af41731e32ad"
+checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -39,9 +39,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.42.0"
+version = "0.43.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d9ceb1da931507a12f4fccea479dccd00da1943e1b4ae72d8e502d707361444"
+checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c"
 dependencies = [
  "cc",
  "cmake",
@@ -189,7 +189,7 @@ checksum = "1e567bd82dcff979e4b03460c307b3cdc9e96fde3d73bed1496d2bc75d9dd62a"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.115",
 ]
 
 [[package]]
@@ -230,9 +230,9 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
 [[package]]
 name = "futures-core"
-version = "0.3.32"
+version = "0.3.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
+checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7"
 
 [[package]]
 name = "futures-sink"
@@ -389,7 +389,7 @@ dependencies = [
 
 [[package]]
 name = "pg_walstream"
-version = "0.8.0"
+version = "0.8.1"
 dependencies = [
  "aws-lc-rs",
  "bytes",
@@ -495,9 +495,9 @@ dependencies = [
 
 [[package]]
 name = "rustls"
-version = "0.23.41"
+version = "0.23.42"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f"
+checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138"
 dependencies = [
  "aws-lc-rs",
  "log",
@@ -537,9 +537,9 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
 
 [[package]]
 name = "serde"
-version = "1.0.228"
+version = "1.0.229"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
+checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
 dependencies = [
  "serde_core",
  "serde_derive",
@@ -547,22 +547,22 @@ dependencies = [
 
 [[package]]
 name = "serde_core"
-version = "1.0.228"
+version = "1.0.229"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad"
+checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.228"
+version = "1.0.229"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79"
+checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 3.0.3",
 ]
 
 [[package]]
@@ -593,9 +593,9 @@ dependencies = [
 
 [[package]]
 name = "socket2"
-version = "0.6.4"
+version = "0.6.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51"
+checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
 dependencies = [
  "libc",
  "windows-sys 0.61.2",
@@ -629,6 +629,17 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "syn"
+version = "3.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
 [[package]]
 name = "tinyvec"
 version = "1.11.0"
@@ -646,9 +657,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.52.3"
+version = "1.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe"
+checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
 dependencies = [
  "bytes",
  "libc",
@@ -667,7 +678,7 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.115",
 ]
 
 [[package]]
@@ -682,9 +693,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-util"
-version = "0.7.18"
+version = "0.7.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
+checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52"
 dependencies = [
  "bytes",
  "futures-core",
@@ -795,7 +806,7 @@ dependencies = [
  "bumpalo",
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.115",
  "wasm-bindgen-shared",
 ]
 
@@ -810,9 +821,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-roots"
-version = "1.0.8"
+version = "1.0.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bf85cb06032201fa7c6f829d7db5a7e5aa45bcc0655327713065f6f0576731bf"
+checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a"
 dependencies = [
  "rustls-pki-types",
 ]
@@ -838,7 +849,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.115",
 ]
 
 [[package]]
@@ -849,7 +860,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn",
+ "syn 2.0.115",
 ]
 
 [[package]]
diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock
index 2c643c9..b5874ea 100644
--- a/fuzz/Cargo.lock
+++ b/fuzz/Cargo.lock
@@ -2,15 +2,6 @@
 # It is not intended for manual editing.
 version = 4
 
-[[package]]
-name = "aho-corasick"
-version = "1.1.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
-dependencies = [
- "memchr",
-]
-
 [[package]]
 name = "android_system_properties"
 version = "0.1.5"
@@ -36,38 +27,43 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
 
 [[package]]
-name = "bindgen"
-version = "0.64.0"
+name = "aws-lc-rs"
+version = "1.17.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c4243e6031260db77ede97ad86c27e501d646a27ab57b59a574f725d98ab1fb4"
+checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1"
 dependencies = [
- "bitflags 1.3.2",
- "cexpr",
- "clang-sys",
- "lazy_static",
- "lazycell",
- "log",
- "peeking_take_while",
- "proc-macro2",
- "quote",
- "regex",
- "rustc-hash",
- "shlex 1.3.0",
- "syn 1.0.109",
- "which",
+ "aws-lc-sys",
+ "untrusted 0.7.1",
+ "zeroize",
 ]
 
 [[package]]
-name = "bitflags"
-version = "1.3.2"
+name = "aws-lc-sys"
+version = "0.43.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
+checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c"
+dependencies = [
+ "cc",
+ "cmake",
+ "dunce",
+ "fs_extra",
+ "pkg-config",
+]
 
 [[package]]
-name = "bitflags"
-version = "2.13.0"
+name = "base64"
+version = "0.22.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8"
+checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
+
+[[package]]
+name = "block-buffer"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa"
+dependencies = [
+ "hybrid-array",
+]
 
 [[package]]
 name = "bumpalo"
@@ -75,11 +71,17 @@ version = "3.20.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
 
+[[package]]
+name = "byteorder"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
+
 [[package]]
 name = "bytes"
-version = "1.12.0"
+version = "1.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ae3f5d315924270530207e2a68396c3cc547f6dca3fbdca317cfb1a51edb593"
+checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
 dependencies = [
  "serde",
 ]
@@ -93,16 +95,7 @@ dependencies = [
  "find-msvc-tools",
  "jobserver",
  "libc",
- "shlex 2.0.1",
-]
-
-[[package]]
-name = "cexpr"
-version = "0.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766"
-dependencies = [
- "nom",
+ "shlex",
 ]
 
 [[package]]
@@ -111,6 +104,17 @@ version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
 
+[[package]]
+name = "chacha20"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "rand_core",
+]
+
 [[package]]
 name = "chrono"
 version = "0.4.45"
@@ -118,30 +122,65 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"
 dependencies = [
  "iana-time-zone",
- "js-sys",
  "num-traits",
  "serde",
- "wasm-bindgen",
  "windows-link",
 ]
 
 [[package]]
-name = "clang-sys"
-version = "1.8.1"
+name = "cmake"
+version = "0.1.58"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4"
+checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678"
 dependencies = [
- "glob",
- "libc",
- "libloading",
+ "cc",
 ]
 
+[[package]]
+name = "cmov"
+version = "0.5.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a"
+
+[[package]]
+name = "const-oid"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c"
+
 [[package]]
 name = "core-foundation-sys"
 version = "0.8.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
 
+[[package]]
+name = "cpufeatures"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201"
+dependencies = [
+ "libc",
+]
+
+[[package]]
+name = "crypto-common"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453"
+dependencies = [
+ "hybrid-array",
+]
+
+[[package]]
+name = "ctutils"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7d5515a3834141de9eafb9717ad39eea8247b5674e6066c404e8c4b365d2a29e"
+dependencies = [
+ "cmov",
+]
+
 [[package]]
 name = "derive_arbitrary"
 version = "1.4.2"
@@ -154,20 +193,28 @@ dependencies = [
 ]
 
 [[package]]
-name = "either"
-version = "1.16.0"
+name = "digest"
+version = "0.11.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e"
+checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2"
+dependencies = [
+ "block-buffer",
+ "const-oid",
+ "crypto-common",
+ "ctutils",
+]
 
 [[package]]
-name = "errno"
-version = "0.3.14"
+name = "dunce"
+version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
-dependencies = [
- "libc",
- "windows-sys 0.61.2",
-]
+checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
+
+[[package]]
+name = "fallible-iterator"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7"
 
 [[package]]
 name = "find-msvc-tools"
@@ -175,11 +222,17 @@ version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
 
+[[package]]
+name = "fs_extra"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
+
 [[package]]
 name = "futures-core"
-version = "0.3.32"
+version = "0.3.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
+checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7"
 
 [[package]]
 name = "futures-sink"
@@ -205,6 +258,17 @@ dependencies = [
  "slab",
 ]
 
+[[package]]
+name = "getrandom"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "wasi",
+]
+
 [[package]]
 name = "getrandom"
 version = "0.3.4"
@@ -213,23 +277,38 @@ checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
 dependencies = [
  "cfg-if",
  "libc",
- "r-efi",
+ "r-efi 5.3.0",
  "wasip2",
 ]
 
 [[package]]
-name = "glob"
-version = "0.3.3"
+name = "getrandom"
+version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280"
+checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi 6.0.0",
+ "rand_core",
+]
 
 [[package]]
-name = "home"
-version = "0.5.12"
+name = "hmac"
+version = "0.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d"
+checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f"
 dependencies = [
- "windows-sys 0.61.2",
+ "digest",
+]
+
+[[package]]
+name = "hybrid-array"
+version = "0.4.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c"
+dependencies = [
+ "typenum",
 ]
 
 [[package]]
@@ -262,7 +341,7 @@ version = "0.1.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33"
 dependencies = [
- "getrandom",
+ "getrandom 0.3.4",
  "libc",
 ]
 
@@ -277,18 +356,6 @@ dependencies = [
  "wasm-bindgen",
 ]
 
-[[package]]
-name = "lazy_static"
-version = "1.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
-
-[[package]]
-name = "lazycell"
-version = "1.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
-
 [[package]]
 name = "libc"
 version = "0.2.186"
@@ -305,33 +372,6 @@ dependencies = [
  "cc",
 ]
 
-[[package]]
-name = "libloading"
-version = "0.8.9"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55"
-dependencies = [
- "cfg-if",
- "windows-link",
-]
-
-[[package]]
-name = "libpq-sys"
-version = "0.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ef060ac05c207c85da15f4eb629100c8782e0db4c06a3c91c86be9c18ae8a23"
-dependencies = [
- "bindgen",
- "pkg-config",
- "vcpkg",
-]
-
-[[package]]
-name = "linux-raw-sys"
-version = "0.4.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab"
-
 [[package]]
 name = "log"
 version = "0.4.32"
@@ -339,16 +379,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a"
 
 [[package]]
-name = "memchr"
-version = "2.8.2"
+name = "md-5"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4"
+checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98"
+dependencies = [
+ "cfg-if",
+ "digest",
+]
 
 [[package]]
-name = "minimal-lexical"
-version = "0.2.1"
+name = "memchr"
+version = "2.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
+checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
 
 [[package]]
 name = "mio"
@@ -361,16 +405,6 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
-[[package]]
-name = "nom"
-version = "7.1.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a"
-dependencies = [
- "memchr",
- "minimal-lexical",
-]
-
 [[package]]
 name = "num-traits"
 version = "0.2.19"
@@ -386,12 +420,6 @@ version = "1.21.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
 
-[[package]]
-name = "peeking_take_while"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099"
-
 [[package]]
 name = "pg-walstream-arbitrary-fuzzing"
 version = "0.1.0"
@@ -402,18 +430,23 @@ dependencies = [
 
 [[package]]
 name = "pg_walstream"
-version = "0.6.3"
+version = "0.8.1"
 dependencies = [
+ "aws-lc-rs",
  "bytes",
  "chrono",
  "futures-core",
- "libpq-sys",
  "memchr",
+ "postgres-protocol",
+ "rustls",
  "serde",
  "smallvec",
+ "socket2",
  "tokio",
+ "tokio-rustls",
  "tokio-util",
  "tracing",
+ "webpki-roots",
 ]
 
 [[package]]
@@ -439,6 +472,24 @@ version = "0.3.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e"
 
+[[package]]
+name = "postgres-protocol"
+version = "0.6.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514"
+dependencies = [
+ "base64",
+ "byteorder",
+ "bytes",
+ "fallible-iterator",
+ "hmac",
+ "md-5",
+ "memchr",
+ "rand",
+ "sha2",
+ "stringprep",
+]
+
 [[package]]
 name = "proc-macro2"
 version = "1.0.106"
@@ -464,51 +515,76 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
 
 [[package]]
-name = "regex"
-version = "1.12.4"
+name = "r-efi"
+version = "6.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1292b7759ae1cb9ec195452d1390a074f0cd8541ab7a5a8c31cd6db45d4a6ba"
+checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
+
+[[package]]
+name = "rand"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
 dependencies = [
- "aho-corasick",
- "memchr",
- "regex-automata",
- "regex-syntax",
+ "chacha20",
+ "getrandom 0.4.3",
+ "rand_core",
 ]
 
 [[package]]
-name = "regex-automata"
-version = "0.4.14"
+name = "rand_core"
+version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f"
+checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
+
+[[package]]
+name = "ring"
+version = "0.17.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7"
 dependencies = [
- "aho-corasick",
- "memchr",
- "regex-syntax",
+ "cc",
+ "cfg-if",
+ "getrandom 0.2.17",
+ "libc",
+ "untrusted 0.9.0",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
-name = "regex-syntax"
-version = "0.8.11"
+name = "rustls"
+version = "0.23.42"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4"
+checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138"
+dependencies = [
+ "aws-lc-rs",
+ "log",
+ "once_cell",
+ "rustls-pki-types",
+ "rustls-webpki",
+ "subtle",
+ "zeroize",
+]
 
 [[package]]
-name = "rustc-hash"
-version = "1.1.0"
+name = "rustls-pki-types"
+version = "1.15.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2"
+checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96"
+dependencies = [
+ "zeroize",
+]
 
 [[package]]
-name = "rustix"
-version = "0.38.44"
+name = "rustls-webpki"
+version = "0.103.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154"
+checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e"
 dependencies = [
- "bitflags 2.13.0",
- "errno",
- "libc",
- "linux-raw-sys",
- "windows-sys 0.59.0",
+ "aws-lc-rs",
+ "ring",
+ "rustls-pki-types",
+ "untrusted 0.9.0",
 ]
 
 [[package]]
@@ -519,9 +595,9 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
 
 [[package]]
 name = "serde"
-version = "1.0.228"
+version = "1.0.229"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
+checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
 dependencies = [
  "serde_core",
  "serde_derive",
@@ -529,29 +605,34 @@ dependencies = [
 
 [[package]]
 name = "serde_core"
-version = "1.0.228"
+version = "1.0.229"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad"
+checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.228"
+version = "1.0.229"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79"
+checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.118",
+ "syn 3.0.3",
 ]
 
 [[package]]
-name = "shlex"
-version = "1.3.0"
+name = "sha2"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64"
+checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "digest",
+]
 
 [[package]]
 name = "shlex"
@@ -576,19 +657,36 @@ dependencies = [
 
 [[package]]
 name = "socket2"
-version = "0.6.4"
+version = "0.6.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51"
+checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
 dependencies = [
  "libc",
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "stringprep"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1"
+dependencies = [
+ "unicode-bidi",
+ "unicode-normalization",
+ "unicode-properties",
+]
+
+[[package]]
+name = "subtle"
+version = "2.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
+
 [[package]]
 name = "syn"
-version = "1.0.109"
+version = "2.0.118"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237"
+checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -597,20 +695,35 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "2.0.118"
+version = "3.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422"
+checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3"
 dependencies = [
  "proc-macro2",
  "quote",
  "unicode-ident",
 ]
 
+[[package]]
+name = "tinyvec"
+version = "1.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bb4ebadaa0af04fab11ae01eb5f9fdb5f9c5b875506e210e71c07873528baa7f"
+dependencies = [
+ "tinyvec_macros",
+]
+
+[[package]]
+name = "tinyvec_macros"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
+
 [[package]]
 name = "tokio"
-version = "1.52.3"
+version = "1.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe"
+checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
 dependencies = [
  "bytes",
  "libc",
@@ -632,11 +745,21 @@ dependencies = [
  "syn 2.0.118",
 ]
 
+[[package]]
+name = "tokio-rustls"
+version = "0.26.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61"
+dependencies = [
+ "rustls",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-util"
-version = "0.7.18"
+version = "0.7.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
+checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52"
 dependencies = [
  "bytes",
  "futures-core",
@@ -652,21 +775,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
 dependencies = [
  "pin-project-lite",
- "tracing-attributes",
  "tracing-core",
 ]
 
-[[package]]
-name = "tracing-attributes"
-version = "0.1.31"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.118",
-]
-
 [[package]]
 name = "tracing-core"
 version = "0.1.36"
@@ -676,6 +787,18 @@ dependencies = [
  "once_cell",
 ]
 
+[[package]]
+name = "typenum"
+version = "1.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
+
+[[package]]
+name = "unicode-bidi"
+version = "0.3.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5"
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.24"
@@ -683,10 +806,31 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
 
 [[package]]
-name = "vcpkg"
-version = "0.2.15"
+name = "unicode-normalization"
+version = "0.1.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5fd4f6878c9cb28d874b009da9e8d183b5abc80117c40bbd187a1fde336be6e8"
+dependencies = [
+ "tinyvec",
+]
+
+[[package]]
+name = "unicode-properties"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d"
+
+[[package]]
+name = "untrusted"
+version = "0.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
+checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a"
+
+[[package]]
+name = "untrusted"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
 
 [[package]]
 name = "wasi"
@@ -749,15 +893,12 @@ dependencies = [
 ]
 
 [[package]]
-name = "which"
-version = "4.4.2"
+name = "webpki-roots"
+version = "1.0.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "87ba24419a2078cd2b0f2ede2691b6c66d8e47836da3b6db8265ebad47afbfc7"
+checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a"
 dependencies = [
- "either",
- "home",
- "once_cell",
- "rustix",
+ "rustls-pki-types",
 ]
 
 [[package]]
@@ -821,9 +962,9 @@ dependencies = [
 
 [[package]]
 name = "windows-sys"
-version = "0.59.0"
+version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b"
+checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
 dependencies = [
  "windows-targets",
 ]
@@ -906,3 +1047,9 @@ name = "wit-bindgen"
 version = "0.57.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
+
+[[package]]
+name = "zeroize"
+version = "1.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e"
diff --git a/macros/Cargo.toml b/macros/Cargo.toml
index f9be2a4..da29405 100644
--- a/macros/Cargo.toml
+++ b/macros/Cargo.toml
@@ -15,6 +15,6 @@ categories = ["database"]
 proc-macro = true
 
 [dependencies]
-syn = { version = "2.0.118", features = ["full"] }
+syn = { version = "2.0.118" }
 quote = "1.0.46"
 proc-macro2 = "1.0.106"
diff --git a/src/buffer.rs b/src/buffer.rs
index 6e112fe..005da53 100644
--- a/src/buffer.rs
+++ b/src/buffer.rs
@@ -62,13 +62,6 @@ impl BufferReader {
         Self { data }
     }
 
-    #[inline]
-    pub fn from_vec(data: Vec<u8>) -> Self {
-        Self {
-            data: Bytes::from(data),
-        }
-    }
-
     /// Get remaining bytes in the buffer
     #[inline]
     pub fn remaining(&self) -> usize {
@@ -251,21 +244,11 @@ impl BufferWriter {
         self.data.len()
     }
 
-    /// Get bytes written so far
-    pub fn bytes_written(&self) -> usize {
-        self.data.len()
-    }
-
     /// Get the data as bytes
     pub fn freeze(self) -> Bytes {
         self.data.freeze()
     }
 
-    /// Get the data as a `Vec<u8>`
-    pub fn into_vec(self) -> Vec<u8> {
-        self.data.to_vec()
-    }
-
     /// Write a single byte.
     #[inline]
     pub fn write_u8(&mut self, value: u8) {
@@ -335,21 +318,10 @@ impl BufferWriter {
         self.data.put_slice(s.as_bytes());
     }
 
-    /// Reserve capacity for at least additional bytes
-    pub fn reserve(&mut self, additional: usize) {
-        self.data.reserve(additional);
-    }
-
     /// Clear the buffer, resetting length to 0
     pub fn clear(&mut self) {
         self.data.clear();
     }
-
-    /// Get remaining capacity
-    pub fn capacity(&self) -> usize {
-        self.data.capacity()
-    }
-
     /// Get a reference to the internal data  
     pub fn as_bytes(&self) -> &[u8] {
         &self.data
@@ -404,7 +376,7 @@ mod tests {
         writer.write_u16(0x0203);
         writer.write_u32(0x04050607);
 
-        assert_eq!(writer.bytes_written(), 7);
+        assert_eq!(writer.position(), 7);
 
         let data = writer.freeze();
         assert_eq!(&data[..], &[0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07]);
@@ -588,15 +560,6 @@ mod tests {
         assert_eq!(reader.remaining(), 4); // "rest"
     }
 
-    #[test]
-    fn test_buffer_reader_from_vec() {
-        let data = vec![0x01, 0x02, 0x03];
-        let mut reader = BufferReader::from_vec(data);
-
-        assert_eq!(reader.read_u8().unwrap(), 0x01);
-        assert_eq!(reader.remaining(), 2);
-    }
-
     #[test]
     fn test_buffer_writer_signed_integers() {
         let mut writer = BufferWriter::new();
@@ -618,7 +581,7 @@ mod tests {
     fn test_buffer_writer_with_capacity() {
         let mut writer = BufferWriter::with_capacity(100);
         writer.write_u64(0x0102030405060708);
-        assert_eq!(writer.bytes_written(), 8);
+        assert_eq!(writer.position(), 8);
     }
 
     #[test]
@@ -659,14 +622,6 @@ mod tests {
         assert_eq!(reader.remaining(), 2);
     }
 
-    #[test]
-    fn test_buffer_reader_from_vec_full() {
-        let data = vec![0xAA, 0xBB, 0xCC, 0xDD];
-        let mut reader = BufferReader::from_vec(data);
-        assert_eq!(reader.read_u32().unwrap(), 0xAABBCCDD);
-        assert_eq!(reader.remaining(), 0);
-    }
-
     #[test]
     fn test_buffer_writer_write_i16() {
         let mut writer = BufferWriter::new();
@@ -718,32 +673,17 @@ mod tests {
         let data = writer.freeze();
         assert_eq!(&data[..], b"hello");
     }
-
-    #[test]
-    fn test_buffer_writer_reserve() {
-        let mut writer = BufferWriter::new();
-        writer.reserve(1024);
-        assert!(writer.capacity() >= 1024);
-    }
-
     #[test]
     fn test_buffer_writer_clear() {
         let mut writer = BufferWriter::new();
         writer.write_u8(0x01);
         writer.write_u8(0x02);
-        assert_eq!(writer.bytes_written(), 2);
+        assert_eq!(writer.position(), 2);
 
         writer.clear();
-        assert_eq!(writer.bytes_written(), 0);
         assert_eq!(writer.position(), 0);
     }
 
-    #[test]
-    fn test_buffer_writer_capacity() {
-        let writer = BufferWriter::with_capacity(256);
-        assert!(writer.capacity() >= 256);
-    }
-
     #[test]
     fn test_buffer_writer_as_bytes() {
         let mut writer = BufferWriter::new();
@@ -764,16 +704,7 @@ mod tests {
     fn test_buffer_writer_default() {
         let mut writer = BufferWriter::default();
         writer.write_u8(0xFF);
-        assert_eq!(writer.bytes_written(), 1);
-    }
-
-    #[test]
-    fn test_buffer_writer_into_vec() {
-        let mut writer = BufferWriter::new();
-        writer.write_u8(0x01);
-        writer.write_u16(0x0203);
-        let vec = writer.into_vec();
-        assert_eq!(vec, vec![0x01, 0x02, 0x03]);
+        assert_eq!(writer.position(), 1);
     }
 
     #[test]
diff --git a/src/connection/libpq.rs b/src/connection/libpq.rs
index 8283556..7e48b21 100644
--- a/src/connection/libpq.rs
+++ b/src/connection/libpq.rs
@@ -29,20 +29,15 @@
 //!
 //! This ensures that no thread is blocked waiting for network I/O, maximizing
 //! throughput and enabling efficient concurrent processing of multiple replication streams.
-use crate::buffer::BufferWriter;
 use crate::error::{ReplicationError, Result};
 use crate::protocol::build_hot_standby_feedback_message;
-use crate::types::{
-    format_lsn, system_time_to_postgres_timestamp, BaseBackupOptions, ReplicationSlotOptions,
-    SlotType, XLogRecPtr,
-};
+use crate::types::{format_lsn, BaseBackupOptions, ReplicationSlotOptions, SlotType, XLogRecPtr};
 use bytes::{BufMut, Bytes, BytesMut};
 use pq_sys::*;
 use std::collections::VecDeque;
 use std::ffi::{CStr, CString};
 use std::os::raw::c_void;
 use std::os::unix::io::RawFd;
-use std::time::SystemTime;
 use std::{ptr, slice};
 use tokio::io::unix::AsyncFd;
 use tokio_util::sync::CancellationToken;
@@ -343,19 +338,12 @@ impl PgReplicationConnection {
     ) -> Result<()> {
         self.ensure_replication_mode()?;
 
-        let timestamp = system_time_to_postgres_timestamp(SystemTime::now());
-
-        // Build the standby status update message using BufferWriter
-        let mut buffer = BufferWriter::with_capacity(34); // 1 + 8 + 8 + 8 + 8 + 1
-
-        buffer.write_u8(b'r'); // Message type
-        buffer.write_u64(received_lsn);
-        buffer.write_u64(flushed_lsn);
-        buffer.write_u64(applied_lsn);
-        buffer.write_i64(timestamp);
-        buffer.write_u8(if reply_requested { 1 } else { 0 });
-
-        let reply_data = buffer.freeze();
+        let reply_data = crate::protocol::build_standby_status_update_message(
+            received_lsn,
+            flushed_lsn,
+            applied_lsn,
+            reply_requested,
+        );
         self.put_copy_data_and_flush(&reply_data).await?;
 
         info!(
@@ -832,8 +820,8 @@ impl PgReplicationConnection {
 
     /// Start a base backup with options
     pub fn base_backup(&mut self, options: &BaseBackupOptions) -> Result<PgResult> {
-        crate::sql_builder::check_base_backup_version(self.server_version(), options)?;
-        let base_backup_sql = crate::sql_builder::build_base_backup_sql(options)?;
+        let base_backup_sql =
+            crate::sql_builder::prepare_base_backup(self.server_version(), options)?;
 
         debug!("Starting base backup: {}", base_backup_sql);
         let result = self.exec(&base_backup_sql)?;
diff --git a/src/connection/native/connection.rs b/src/connection/native/connection.rs
index c567fb9..9272087 100644
--- a/src/connection/native/connection.rs
+++ b/src/connection/native/connection.rs
@@ -8,7 +8,6 @@ use std::collections::VecDeque;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc as std_mpsc;
 use std::sync::Arc;
-use std::time::SystemTime;
 use tokio::sync::{mpsc, oneshot};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, info};
@@ -18,13 +17,9 @@ use super::startup::{self, Transport};
 use super::{copy, query, wire};
 use super::{NativePgResult, NativeResultStatus};
 
-use crate::buffer::BufferWriter;
 use crate::error::{ReplicationError, Result};
 use crate::protocol::build_hot_standby_feedback_message;
-use crate::types::{
-    format_lsn, system_time_to_postgres_timestamp, BaseBackupOptions, ReplicationSlotOptions,
-    SlotType, XLogRecPtr,
-};
+use crate::types::{format_lsn, BaseBackupOptions, ReplicationSlotOptions, SlotType, XLogRecPtr};
 
 // A `NativeConnection` is a handle. A dedicated worker thread owns the socket on
 // its own current-thread runtime, so all I/O stays on one reactor and the
@@ -684,17 +679,13 @@ impl NativeConnection {
     ) -> Result<()> {
         self.ensure_replication_mode()?;
 
-        let timestamp = system_time_to_postgres_timestamp(SystemTime::now());
-
-        let mut buffer = BufferWriter::with_capacity(34);
-        buffer.write_u8(b'r');
-        buffer.write_u64(received_lsn);
-        buffer.write_u64(flushed_lsn);
-        buffer.write_u64(applied_lsn);
-        buffer.write_i64(timestamp);
-        buffer.write_u8(if reply_requested { 1 } else { 0 });
-
-        self.put_copy_data(buffer.freeze()).await?;
+        let reply_data = crate::protocol::build_standby_status_update_message(
+            received_lsn,
+            flushed_lsn,
+            applied_lsn,
+            reply_requested,
+        );
+        self.put_copy_data(reply_data).await?;
 
         info!(
             "Sent standby status update: received={}, flushed={}, applied={}, reply_requested={}",
@@ -930,8 +921,7 @@ impl NativeConnection {
 
     /// Start a base backup with options.
     pub fn base_backup(&mut self, options: &BaseBackupOptions) -> Result<NativePgResult> {
-        crate::sql_builder::check_base_backup_version(self.server_version(), options)?;
-        let sql = crate::sql_builder::build_base_backup_sql(options)?;
+        let sql = crate::sql_builder::prepare_base_backup(self.server_version(), options)?;
 
         debug!("Starting base backup: {}", sql);
         let result = self.exec(&sql)?;
diff --git a/src/connection/native/copy.rs b/src/connection/native/copy.rs
index c78b613..da5b3f7 100644
--- a/src/connection/native/copy.rs
+++ b/src/connection/native/copy.rs
@@ -16,11 +16,8 @@ use tokio_util::sync::CancellationToken;
 /// Prevents unbounded queue growth under extreme throughput.
 const MAX_DRAIN_BATCH: usize = 4096;
 
-/// Minimum header size: 1 (tag) + 4 (length) = 5 bytes.
-const HEADER_LEN: usize = 5;
-
-/// Maximum allowed message body length (128 MiB), matching wire.rs.
-const MAX_MESSAGE_LEN: usize = 128 * 1024 * 1024;
+// Header framing and body-length bounds are shared with the query path (single source in `wire`).
+use super::wire::{HEADER_LEN, MAX_MESSAGE_LEN};
 
 /// Headroom reserved on `read_buf` before each socket read.
 ///
diff --git a/src/connection/native/error.rs b/src/connection/native/error.rs
index fc6aadb..2f448fc 100644
--- a/src/connection/native/error.rs
+++ b/src/connection/native/error.rs
@@ -71,27 +71,6 @@ pub fn parse_error_fields(payload: &[u8]) -> PgErrorFields {
 mod tests {
     use super::*;
 
-    fn error_response_to_replication_error(msg: &[u8]) -> crate::error::ReplicationError {
-        let fields = parse_error_fields(&msg[5..]); // skip tag + length
-        let error_lower = fields.message.to_lowercase();
-
-        if error_lower.contains("authentication")
-            || error_lower.contains("password")
-            || fields.code.starts_with("28")
-        {
-            crate::error::ReplicationError::authentication(format!(
-                "PostgreSQL authentication failed: {}",
-                fields
-            ))
-        } else if fields.severity == "FATAL" || fields.severity == "PANIC" {
-            crate::error::ReplicationError::permanent_connection(format!(
-                "PostgreSQL fatal error: {}",
-                fields
-            ))
-        } else {
-            crate::error::ReplicationError::protocol(format!("PostgreSQL error: {}", fields))
-        }
-    }
     #[test]
     fn test_parse_error_fields() {
         let mut payload = Vec::new();
@@ -161,8 +140,9 @@ mod tests {
     }
 
     #[test]
-    fn test_error_response_to_replication_error_fatal() {
-        // Build a raw 'E' message with FATAL severity
+    fn test_parse_error_fields_from_framed_message() {
+        // A full ErrorResponse frame: tag 'E' + i32 length + field payload.
+        // Verifies fields parse out of the framed form (payload starts at msg[5..]).
         let mut payload = Vec::new();
         payload.push(b'S');
         payload.extend_from_slice(b"FATAL\0");
@@ -177,38 +157,10 @@ mod tests {
         msg.extend_from_slice(&len.to_be_bytes());
         msg.extend_from_slice(&payload);
 
-        let err = error_response_to_replication_error(&msg);
-        // Should be a permanent connection error for FATAL
-        let err_str = err.to_string();
-        assert!(
-            err_str.contains("FATAL") || err_str.contains("terminating"),
-            "Got: {err_str}"
-        );
-    }
-
-    #[test]
-    fn test_error_response_to_replication_error_auth() {
-        // Build a raw 'E' message with auth error (SQLSTATE 28000)
-        let mut payload = Vec::new();
-        payload.push(b'S');
-        payload.extend_from_slice(b"FATAL\0");
-        payload.push(b'C');
-        payload.extend_from_slice(b"28000\0");
-        payload.push(b'M');
-        payload.extend_from_slice(b"password authentication failed\0");
-        payload.push(0);
-
-        let mut msg = vec![b'E'];
-        let len = (4 + payload.len()) as i32;
-        msg.extend_from_slice(&len.to_be_bytes());
-        msg.extend_from_slice(&payload);
-
-        let err = error_response_to_replication_error(&msg);
-        let err_str = err.to_string();
-        assert!(
-            err_str.contains("authentication") || err_str.contains("password"),
-            "Got: {err_str}"
-        );
+        let fields = parse_error_fields(&msg[5..]);
+        assert_eq!(fields.severity, "FATAL");
+        assert_eq!(fields.code, "57P01");
+        assert!(fields.message.contains("terminating"));
     }
 
     #[test]
diff --git a/src/connection/native/wire.rs b/src/connection/native/wire.rs
index 171468f..6b317a8 100644
--- a/src/connection/native/wire.rs
+++ b/src/connection/native/wire.rs
@@ -9,7 +9,7 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use crate::error::ReplicationError;
 
 /// Minimum message size: 1 (tag) + 4 (length) = 5 bytes header.
-const HEADER_LEN: usize = 5;
+pub(crate) const HEADER_LEN: usize = 5;
 
 /// Maximum allowed message body length (128 MiB).
 ///
@@ -17,7 +17,7 @@ const HEADER_LEN: usize = 5;
 /// `body_len` close to `i32::MAX` (2 GiB). PostgreSQL's own max message size
 /// is 1 GiB, but 128 MiB is more than sufficient for any replication message
 /// and provides a safety margin for large TOAST values.
-const MAX_MESSAGE_LEN: usize = 128 * 1024 * 1024;
+pub(crate) const MAX_MESSAGE_LEN: usize = 128 * 1024 * 1024;
 
 /// Read a single complete PostgreSQL backend message from the transport.
 ///
diff --git a/src/error.rs b/src/error.rs
index 82539f0..6c3d466 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -47,9 +47,6 @@ pub enum ReplicationError {
     /// String conversion errors (from CString operations)
     StringConversion(std::ffi::NulError),
 
-    /// Generic replication errors
-    Generic(String),
-
     /// Deserialization errors (when converting RowData to user types)
     Deserialize(String),
 
@@ -79,7 +76,6 @@ impl core::fmt::Display for ReplicationError {
             Self::Io(err) => write!(f, "IO error: {err}"),
             #[cfg(feature = "std")]
             Self::StringConversion(err) => write!(f, "String conversion error: {err}"),
-            Self::Generic(msg) => write!(f, "Replication error: {msg}"),
             Self::Deserialize(msg) => write!(f, "Deserialization error: {msg}"),
             Self::Backend(msg) => write!(f, "Backend worker error: {msg}"),
             Self::StreamStopped(lsn) => {
@@ -143,11 +139,6 @@ impl ReplicationError {
     }
 
     /// Create a new replication connection error
-    pub fn replication_connection<S: Into<String>>(msg: S) -> Self {
-        ReplicationError::ReplicationConnection(msg.into())
-    }
-
-    /// Create a new connection error (alias for replication_connection)
     pub fn connection<S: Into<String>>(msg: S) -> Self {
         ReplicationError::ReplicationConnection(msg.into())
     }
@@ -177,11 +168,6 @@ impl ReplicationError {
         ReplicationError::Config(msg.into())
     }
 
-    /// Create a new generic error
-    pub fn generic<S: Into<String>>(msg: S) -> Self {
-        ReplicationError::Generic(msg.into())
-    }
-
     /// Create a new deserialization error
     pub fn deserialize<S: Into<String>>(msg: S) -> Self {
         ReplicationError::Deserialize(msg.into())
@@ -202,7 +188,9 @@ impl ReplicationError {
         ReplicationError::StreamStopped(lsn)
     }
 
-    /// Check if the error is transient (can be retried)
+    /// Check if the error is transient (can be retried).
+    ///
+    /// Note: this is an advisory classification for consumers (logging, metrics, custom retry policies). It is NOT the predicate the library itself uses to drive retries — the streaming layer retries anything that is not [`is_permanent`](Self::is_permanent), [`is_cancelled`](Self::is_cancelled), or the internal stream-stopped terminal. So the "grey zone" variants (e.g. `Protocol`, `Buffer`, `Config`, `Deserialize`) return `false` here yet are still retried by the stream.
     pub fn is_transient(&self) -> bool {
         #[cfg(feature = "std")]
         if matches!(self, ReplicationError::Io(_)) {
@@ -339,15 +327,6 @@ mod tests {
         assert!(!err.is_permanent());
     }
 
-    #[test]
-    fn test_generic_error() {
-        let err = ReplicationError::generic("something went wrong");
-        match err {
-            ReplicationError::Generic(msg) => assert_eq!(msg, "something went wrong"),
-            _ => panic!("Expected Generic error"),
-        }
-    }
-
     #[test]
     fn test_connection_alias() {
         let err = ReplicationError::connection("test");
@@ -366,14 +345,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn test_replication_connection_display() {
-        let err = ReplicationError::replication_connection("slot error");
-        assert_eq!(err.to_string(), "Replication connection error: slot error");
-        assert!(err.is_transient());
-        assert!(!err.is_permanent());
-    }
-
     #[test]
     fn test_replication_slot_display() {
         let err = ReplicationError::replication_slot("slot not found");
@@ -395,15 +366,6 @@ mod tests {
         assert!(!err.is_cancelled());
     }
 
-    #[test]
-    fn test_generic_error_display() {
-        let err = ReplicationError::generic("unknown issue");
-        assert_eq!(err.to_string(), "Replication error: unknown issue");
-        assert!(!err.is_transient());
-        assert!(!err.is_permanent());
-        assert!(!err.is_cancelled());
-    }
-
     #[test]
     fn test_io_error_conversion() {
         let io_err = std::io::Error::new(std::io::ErrorKind::NotFound, "file not found");
diff --git a/src/lib.rs b/src/lib.rs
index e318360..4407a87 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -69,16 +69,13 @@
 //!
 //! #[tokio::main]
 //! async fn main() -> Result<(), Box<dyn std::error::Error>> {
-//!     let config = ReplicationStreamConfig::new(
-//!         "my_slot".to_string(),
-//!         "my_publication".to_string(),
-//!         2,
-//!         StreamingMode::On,
-//!         Duration::from_secs(10),
-//!         Duration::from_secs(30),
-//!         Duration::from_secs(60),
-//!         RetryConfig::default(),
-//!     );
+//!     let config = ReplicationStreamConfig::builder("my_slot", "my_publication")
+//!         .with_protocol_version(2)
+//!         .with_streaming_mode(StreamingMode::On)
+//!         .with_feedback_interval(Duration::from_secs(10))
+//!         .with_connection_timeout(Duration::from_secs(30))
+//!         .with_health_check_interval(Duration::from_secs(60))
+//!         .with_retry_config(RetryConfig::default());
 //!
 //!     let mut stream = LogicalReplicationStream::new(
 //!         "postgresql://postgres:password@localhost:5432/mydb?replication=database",
@@ -210,7 +207,6 @@ pub use types::{
     Xid,
     // Constants
     INVALID_XLOG_REC_PTR,
-    PG_EPOCH_OFFSET_SECS,
 };
 
 // `system_time_to_postgres_timestamp` needs a `SystemTime`, so it is std-only.
@@ -224,7 +220,7 @@ pub use protocol::{
     RelationInfo, ReplicationState, StreamingReplicationMessage, TupleData,
 };
 
-// pgoutput encoder. `encode_message` emits pgoutput, unlike `ChangeEvent::encode`.
+// pgoutput encoder: `encode_message` emits pgoutput wire bytes.
 pub use pgoutput_encode::{
     encode_message, encode_message_to_bytes, encode_streaming_message,
     encode_streaming_message_to_bytes,
diff --git a/src/lsn.rs b/src/lsn.rs
index 4df5295..4fc7754 100644
--- a/src/lsn.rs
+++ b/src/lsn.rs
@@ -12,8 +12,8 @@
 //! we need a thread-safe way to share the committed LSN from consumer back to producer
 //! for accurate feedback to PostgreSQL.
 
-use crate::prelude::*;
-use crate::types::{format_lsn, CachePadded, XLogRecPtr};
+use crate::types::{CachePadded, XLogRecPtr};
+use crate::{format_lsn, prelude::*};
 use core::sync::atomic::{AtomicU64, Ordering};
 use tracing::{debug, info};
 
@@ -228,7 +228,7 @@ impl SharedLsnFeedback {
         (flushed, applied)
     }
 
-    /// Log current LSN state (for debugging)
+    /// Log current LSN state
     pub fn log_state(&self, prefix: &str) {
         let flushed = self.get_flushed_lsn();
         let applied = self.get_applied_lsn();
@@ -247,15 +247,6 @@ impl Default for SharedLsnFeedback {
     }
 }
 
-impl Clone for SharedLsnFeedback {
-    fn clone(&self) -> Self {
-        Self {
-            flushed_lsn: CachePadded::new(AtomicU64::new(self.flushed_lsn.load(Ordering::Acquire))),
-            applied_lsn: CachePadded::new(AtomicU64::new(self.applied_lsn.load(Ordering::Acquire))),
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -318,22 +309,6 @@ mod tests {
         assert_eq!(applied, 50);
     }
 
-    #[test]
-    fn test_clone() {
-        let feedback = SharedLsnFeedback::new();
-        feedback.update_flushed_lsn(100);
-        feedback.update_applied_lsn(50);
-
-        let cloned = feedback.clone();
-        assert_eq!(cloned.get_flushed_lsn(), 100);
-        assert_eq!(cloned.get_applied_lsn(), 50);
-
-        // Modifying the clone should not affect the original
-        cloned.update_applied_lsn(200);
-        assert_eq!(feedback.get_applied_lsn(), 50);
-        assert_eq!(cloned.get_applied_lsn(), 200);
-    }
-
     #[test]
     fn test_new_shared() {
         let feedback = SharedLsnFeedback::new_shared();
@@ -431,52 +406,6 @@ mod tests {
         assert_eq!(a, 0);
     }
 
-    #[test]
-    fn test_log_state() {
-        // Just ensure log_state doesn't panic
-        let feedback = SharedLsnFeedback::new();
-        feedback.update_flushed_lsn(1000);
-        feedback.update_applied_lsn(500);
-        feedback.log_state("test_prefix");
-    }
-
-    #[test]
-    fn test_clone_independence() {
-        let feedback = SharedLsnFeedback::new();
-        feedback.update_flushed_lsn(500);
-        feedback.update_applied_lsn(300);
-
-        let cloned = feedback.clone();
-
-        // Verify cloned has same initial values
-        assert_eq!(cloned.get_flushed_lsn(), 500);
-        assert_eq!(cloned.get_applied_lsn(), 300);
-
-        // Modify original
-        feedback.update_flushed_lsn(1000);
-        feedback.update_applied_lsn(800);
-
-        // Clone should not be affected
-        assert_eq!(cloned.get_flushed_lsn(), 500);
-        assert_eq!(cloned.get_applied_lsn(), 300);
-
-        // Modify clone
-        cloned.update_flushed_lsn(2000);
-        cloned.update_applied_lsn(1500);
-
-        // Original should not be affected
-        assert_eq!(feedback.get_flushed_lsn(), 1000);
-        assert_eq!(feedback.get_applied_lsn(), 800);
-    }
-
-    #[test]
-    fn test_log_state_with_nonzero_lsns() {
-        let feedback = SharedLsnFeedback::new();
-        feedback.update_flushed_lsn(0x16B374D848);
-        feedback.update_applied_lsn(0x16B374D800);
-        feedback.log_state("replication");
-    }
-
     #[test]
     fn test_equal_lsn_no_update() {
         let feedback = SharedLsnFeedback::new();
diff --git a/src/protocol.rs b/src/protocol.rs
index c11b84f..ba3e91c 100644
--- a/src/protocol.rs
+++ b/src/protocol.rs
@@ -2009,6 +2009,37 @@ pub fn build_hot_standby_feedback_message(
     Ok(buffer.freeze())
 }
 
+/// Build a Standby Status Update ('r') message body — the client→server feedback reply
+/// sent inside a `CopyData` frame during logical replication.
+///
+/// The message format is:
+/// - Byte1('r'): Message type identifier
+/// - Int64: received LSN (last WAL byte written)
+/// - Int64: flushed LSN
+/// - Int64: applied LSN (last WAL byte replayed)
+/// - Int64: client timestamp
+/// - Byte1: 1 to request an immediate server reply, else 0
+#[cfg(feature = "std")]
+pub(crate) fn build_standby_status_update_message(
+    received_lsn: XLogRecPtr,
+    flushed_lsn: XLogRecPtr,
+    applied_lsn: XLogRecPtr,
+    reply_requested: bool,
+) -> Bytes {
+    let timestamp = system_time_to_postgres_timestamp(SystemTime::now());
+    // `b'r'` here is the client→server COPY feedback tag; it is distinct from the
+    // identically-valued server→client `message_types::ROLLBACK_PREPARED` tag.
+    let mut buffer = BufferWriter::with_capacity(34); // 1 + 8*4 + 1
+    buffer.write_u8(b'r');
+    buffer.write_u64(received_lsn);
+    buffer.write_u64(flushed_lsn);
+    buffer.write_u64(applied_lsn);
+    buffer.write_i64(timestamp);
+    buffer.write_u8(if reply_requested { 1 } else { 0 });
+
+    buffer.freeze()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -2761,6 +2792,25 @@ mod tests {
         assert_eq!(catalog_xmin_epoch, 2);
     }
 
+    #[test]
+    fn test_build_standby_status_update_message() {
+        let message =
+            build_standby_status_update_message(0x1122_3344_5566_7788, 0xAABB_CCDD, 0x99, true);
+        assert_eq!(message.len(), 34); // 1 tag + 8*4 LSNs/ts + 1 reply flag
+
+        let mut reader = BufferReader::new(&message);
+        assert_eq!(reader.read_u8().unwrap(), b'r');
+        assert_eq!(reader.read_u64().unwrap(), 0x1122_3344_5566_7788); // received
+        assert_eq!(reader.read_u64().unwrap(), 0xAABB_CCDD); // flushed
+        assert_eq!(reader.read_u64().unwrap(), 0x99); // applied
+        let _timestamp = reader.read_i64().unwrap(); // wall-clock, just verify readable
+        assert_eq!(reader.read_u8().unwrap(), 1); // reply_requested = true
+
+        // reply_requested = false → trailing flag byte is 0
+        let no_reply = build_standby_status_update_message(1, 2, 3, false);
+        assert_eq!(no_reply[no_reply.len() - 1], 0);
+    }
+
     #[test]
     fn test_column_data_binary_bytes() {
         let data = Bytes::from_static(&[0xDE, 0xAD, 0xBE, 0xEF]);
diff --git a/src/sql_builder.rs b/src/sql_builder.rs
index 3c01907..26f4fb8 100644
--- a/src/sql_builder.rs
+++ b/src/sql_builder.rs
@@ -687,6 +687,15 @@ mod version_preflight {
         Ok(())
     }
 
+    /// Preflight + build `BASE_BACKUP`.
+    pub(crate) fn prepare_base_backup(
+        server_version: i32,
+        options: &BaseBackupOptions,
+    ) -> Result<String> {
+        check_base_backup_version(server_version, options)?;
+        build_base_backup_sql(options)
+    }
+
     /// Whether any `BASE_BACKUP` option is set — i.e. the emitted SQL would use the parenthesized option list rather than a bare `BASE_BACKUP`.
     fn base_backup_has_options(o: &BaseBackupOptions) -> bool {
         o.label.is_some()
@@ -811,6 +820,21 @@ mod version_preflight {
             assert!(check_base_backup_version(0, &inc).is_ok());
         }
 
+        #[test]
+        fn prepare_base_backup_builds_and_gates() {
+            // Unknown version (0) passes preflight and builds a bare BASE_BACKUP.
+            let sql = prepare_base_backup(0, &BaseBackupOptions::default()).unwrap();
+            assert!(sql.starts_with("BASE_BACKUP"), "{sql}");
+
+            // Preflight is wired in: INCREMENTAL is rejected below PG17, accepted at PG17+.
+            let inc = BaseBackupOptions {
+                incremental: true,
+                ..Default::default()
+            };
+            assert!(prepare_base_backup(160000, &inc).is_err());
+            assert!(prepare_base_backup(170000, &inc).is_ok());
+        }
+
         #[test]
         fn format_server_version_renders_major_minor() {
             assert_eq!(format_server_version(140023), "14.23");
@@ -822,7 +846,7 @@ mod version_preflight {
 
 #[cfg(any(feature = "libpq", feature = "rustls-tls"))]
 pub(crate) use version_preflight::{
-    check_base_backup_version, prepare_alter_slot, prepare_create_slot, prepare_read_slot,
+    prepare_alter_slot, prepare_base_backup, prepare_create_slot, prepare_read_slot,
 };
 
 /// Options for building a `CREATE SUBSCRIPTION` SQL statement.
@@ -959,7 +983,7 @@ pub fn build_drop_subscription_sql(name: &str) -> Result<String> {
 /// Format a list of options as ` (opt1, opt2, ...)`.
 /// Returns an empty string if the list is empty.
 #[inline]
-pub fn build_sql_options(options: &[String]) -> String {
+pub(crate) fn build_sql_options(options: &[String]) -> String {
     if options.is_empty() {
         String::new()
     } else {
diff --git a/src/stream.rs b/src/stream.rs
index b60ed55..01b23d5 100644
--- a/src/stream.rs
+++ b/src/stream.rs
@@ -1141,6 +1141,22 @@ impl LogicalReplicationStream {
         Ok(())
     }
 
+    /// Fetch the consumer's flushed/applied LSNs from the shared tracker, each capped
+    /// to `last_received_lsn` — a consumer must never report progress ahead of what we
+    /// have actually received. A zero from the tracker means "unset" and stays zero.
+    #[inline]
+    fn capped_feedback_lsns(&self) -> (u64, u64) {
+        let (f, a) = self.shared_lsn_feedback.get_feedback_lsn();
+        let cap = |v: u64| {
+            if v > 0 {
+                v.min(self.state.last_received_lsn)
+            } else {
+                0
+            }
+        };
+        (cap(f), cap(a))
+    }
+
     /// Check if feedback should be sent and send it
     ///
     /// Implements intelligent throttling: only sends feedback if the configured
@@ -1157,17 +1173,7 @@ impl LogicalReplicationStream {
         }
 
         // Get current LSN values that would be sent
-        let (f, a) = self.shared_lsn_feedback.get_feedback_lsn();
-        let flushed_lsn = if f > 0 {
-            f.min(self.state.last_received_lsn)
-        } else {
-            0
-        };
-        let applied_lsn = if a > 0 {
-            a.min(self.state.last_received_lsn)
-        } else {
-            0
-        };
+        let (flushed_lsn, applied_lsn) = self.capped_feedback_lsns();
 
         // Only send feedback if LSN values have changed
         if self.state.lsn_has_changed(flushed_lsn, applied_lsn) {
@@ -1195,19 +1201,9 @@ impl LogicalReplicationStream {
             return Ok(());
         }
 
-        // This allows the consumer to update these values after committing to destination
-        let (f, a) = self.shared_lsn_feedback.get_feedback_lsn();
-        // Cap LSN values to last_received_lsn (consumer shouldn't be ahead, but handle gracefully)
-        let flushed_lsn = if f > 0 {
-            f.min(self.state.last_received_lsn)
-        } else {
-            0
-        };
-        let applied_lsn = if a > 0 {
-            a.min(self.state.last_received_lsn)
-        } else {
-            0
-        };
+        // Fetch the consumer's committed progress (capped to what we've received).
+        // This allows the consumer to update these values after committing to destination.
+        let (flushed_lsn, applied_lsn) = self.capped_feedback_lsns();
 
         // Update local state from shared feedback for consistency
         if flushed_lsn > self.state.last_flushed_lsn {
@@ -3762,7 +3758,7 @@ mod tests {
     #[tokio::test]
     async fn test_timeout_or_error_inner_error() {
         let result = timeout_or_error(Duration::from_secs(5), async {
-            Err::<i32, _>(ReplicationError::generic("inner error".to_string()))
+            Err::<i32, _>(ReplicationError::protocol("inner error".to_string()))
         })
         .await;
         assert!(result.is_err());
diff --git a/src/types.rs b/src/types.rs
index 1057ad2..97ccd6a 100644
--- a/src/types.rs
+++ b/src/types.rs
@@ -14,7 +14,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 // PostgreSQL constants
 /// Seconds from Unix epoch (1970-01-01) to PostgreSQL epoch (2000-01-01)
-pub const PG_EPOCH_OFFSET_SECS: i64 = 946_684_800;
+pub(crate) const PG_EPOCH_OFFSET_SECS: i64 = 946_684_800;
 /// Invalid/zero LSN pointer
 pub const INVALID_XLOG_REC_PTR: u64 = 0;
 
@@ -136,19 +136,26 @@ pub fn system_time_to_postgres_timestamp(time: SystemTime) -> TimestampTz {
 }
 
 /// Convert PostgreSQL timestamp (microseconds since 2000-01-01) into `chrono::DateTime<Utc>`.
+///
+/// Never panics. This runs on the live BEGIN/COMMIT/PREPARE parse path, so a corrupt or
+/// adversarial timestamp must not abort the replication task: an overflowing or
+/// out-of-range value is clamped to `DateTime::<Utc>::MIN_UTC` / `MAX_UTC` instead.
 pub fn postgres_timestamp_to_chrono(ts: i64) -> chrono::DateTime<chrono::Utc> {
-    use chrono::{TimeZone, Utc};
+    use chrono::{DateTime, TimeZone, Utc};
 
-    // Convert back to Unix epoch microseconds
-    let unix_micros = ts + PG_EPOCH_OFFSET_SECS * 1_000_000;
+    // Rebase onto the Unix epoch, saturating rather than overflowing on extreme inputs.
+    let unix_micros = ts.saturating_add(PG_EPOCH_OFFSET_SECS * 1_000_000);
 
-    let secs = unix_micros / 1_000_000;
-    let micros = (unix_micros % 1_000_000) as u32;
+    // Euclidean div/rem keeps the subsecond part in [0, 1_000_000) for negative values too
+    // (plain `%` yields a negative remainder that wraps when cast to u32).
+    let secs = unix_micros.div_euclid(1_000_000);
+    let subsec_nanos = (unix_micros.rem_euclid(1_000_000) as u32) * 1000;
 
-    // Construct chrono DateTime<Utc>
-    Utc.timestamp_opt(secs, micros * 1000)
-        .single()
-        .expect("Invalid timestamp conversion")
+    match Utc.timestamp_opt(secs, subsec_nanos).single() {
+        Some(dt) => dt,
+        None if secs < 0 => DateTime::<Utc>::MIN_UTC,
+        None => DateTime::<Utc>::MAX_UTC,
+    }
 }
 
 /// Parse LSN from string format (e.g., "0/12345678")
@@ -1295,6 +1302,10 @@ impl ChangeEvent {
     ///
     /// This is significantly faster than JSON for both encoding and decoding,
     /// and produces a smaller payload.
+    #[deprecated(
+        since = "0.9.0",
+        note = "`ChangeEvent` derives `serde::Serialize`; serialize with a binary serde codec (e.g. bincode or postcard) instead. This bespoke codec duplicates serde and  is scheduled for removal in a future release. Note: it is NOT the pgoutput wire  format — for that, use `pg_walstream::encode_message`."
+    )]
     pub fn encode(&self, buf: &mut BytesMut) {
         // LSN
         buf.extend_from_slice(&self.lsn.0.to_be_bytes());
@@ -1586,6 +1597,10 @@ impl ChangeEvent {
     }
 
     /// Decode a `ChangeEvent` from binary data produced by [`encode`](Self::encode).
+    #[deprecated(
+        since = "0.9.0",
+        note = "`ChangeEvent` derives `serde::Deserialize`; deserialize with a binary serde codec (e.g. bincode or postcard) instead. This bespoke codec duplicates serde and  is scheduled for removal in a future release. Note: it does NOT parse the pgoutput wire format — WAL parsing lives in `pg_walstream::protocol`."
+    )]
     pub fn decode(data: &[u8]) -> Result<Self> {
         let mut reader = BufferReader::new(data);
 
@@ -1951,6 +1966,7 @@ fn micros_to_chrono(micros: i64) -> Result<chrono::DateTime<chrono::Utc>> {
 
 #[cfg(test)]
 mod tests {
+    #![allow(deprecated)] // round-trip tests intentionally exercise the deprecated ChangeEvent::encode/decode
     use super::*;
     use bytes::Bytes;
     use chrono::{TimeZone, Utc};
@@ -2018,6 +2034,22 @@ mod tests {
         assert!(diff < 2, "Round trip difference too large: {diff}");
     }
 
+    #[test]
+    fn test_postgres_timestamp_extremes_do_not_panic() {
+        // Runs on the live parse path — corrupt/adversarial values must clamp, not panic.
+        assert_eq!(
+            postgres_timestamp_to_chrono(i64::MAX),
+            chrono::DateTime::<Utc>::MAX_UTC
+        );
+        assert_eq!(
+            postgres_timestamp_to_chrono(i64::MIN),
+            chrono::DateTime::<Utc>::MIN_UTC
+        );
+        // A negative pre-epoch value keeps a well-formed (non-wrapped) subsecond part.
+        let dt = postgres_timestamp_to_chrono(-1);
+        assert!(dt.timestamp_subsec_micros() < 1_000_000);
+    }
+
     #[test]
     fn test_replica_identity_from_byte() {
         assert_eq!(