refactor(proxy): migrate PostgreSQL protocol to pg-proto

PG_PROTO_FOLLOWUPS.md

@@ -0,0 +1,40 @@
+# pg-proto follow-ups
+
+The proxy migration delegates framing, typed messages, startup/authentication
+state, typed startup middleware, async runtime middleware, demultiplexing, and
+compile-time checked bounded pipeline dispatch to `pg-proto` 0.5.0. The
+remaining transport adapters below would be better eliminated in `pg-proto`
+itself.
+
+## Preserve buffered transport state across a split
+
+`Buffered::into_inner()` cannot return retained inbound bytes, pending outbound
+bytes, or demultiplexer state. The proxy must finish startup through
+`ReadyForQuery` before splitting the bidirectional stream and then construct new
+`Buffered` values for concurrent frontend/backend processing.
+
+An `into_parts`/`from_parts` API, or a buffer-preserving split API, would let a
+proxy change transport ownership without risking loss of bytes already read past
+a message boundary. It should preserve both codec buffers and the demultiplexer.
+
+## Transfer a demultiplexer between transport owners
+
+The proxy routes backend messages through pg-proto's demultiplexer, but startup
+and concurrent runtime currently use separate `Buffered` owners. A supported way
+to extract and restore `Demux` state would retain startup parameter status,
+cancellation-key, and readiness state without application bookkeeping.
+
+This may naturally be solved by the buffer-preserving transport-parts API above.
+
+## Publish the example proxy driver as a library API
+
+`pg-proto` demonstrates a clean `Buffered` + `Middleware` forwarding loop in
+`examples/proxy_support`, but does not expose a configurable proxy driver from
+the crate. CipherStash therefore still owns connection orchestration, concurrent
+forwarding, and the small amount of glue that invokes middleware.
+
+A library-level proxy builder should accept downstream/upstream transports,
+startup and authentication policy, typed frontend/backend middleware, timeout
+policy, and an output strategy. It should own framing, phase transitions,
+bounded pipeline dispatch, demultiplexing, and shutdown. Applications would then
+only supply policy and message transformations.

PG_PROTO_MIGRATION_PLAN.md

@@ -0,0 +1,38 @@
+# pg-proto Migration
+
+## Summary
+
+- Create `/Users/jamessadler/cipherstash/proxy-pg-proto` from current `main` (`15b7f996`) on branch `refactor/pg-proto`.
+- Save this migration plan as `PG_PROTO_MIGRATION_PLAN.md` in that worktree’s repository root.
+- Limit the initial deliverable to the worktree, branch, and plan document; implementation follows separately.
+- Target a full migration to published [`pg-proto` 0.2.1](https://crates.io/crates/pg-proto/0.2.1), covering codecs, startup/authentication, runtime protocol-state validation, and bounded pipeline orchestration.
+
+## Implementation Changes
+
+- Replace handwritten framing, startup packet parsing, message codes, and message serialization with direction-specific `pg-proto` frontend/backend codecs.
+- Convert CipherStash-specific behavior into adapters over `pg-proto` messages:
+  - Preserve Parse/Query SQL rewriting and parameter OID mapping.
+  - Preserve Bind format-code semantics, nulls, parameter reshaping, and encryption.
+  - Preserve ParameterDescription, RowDescription, and DataRow rewriting and batched decryption.
+  - Retain diagnostic-response factories while emitting `pg-proto` response types.
+- Use `pg-proto` pre-startup and authentication APIs for SSL negotiation, startup, cancellation, client-facing MD5 authentication, and upstream cleartext/MD5/SCRAM authentication. Continue using existing TLS configuration and certificate policy.
+- Pair downstream server-role and upstream client-role runtime FSMs through `Intermediary`. Advance both sides for forwarded messages and only the affected side for locally intercepted or synthesised messages.
+- Preserve concurrent client-to-server and server-to-client processing, connection timeouts, response ordering, metrics, logging, schema reloads, and row buffering.
+- Track protocol state even when encryption mapping is disabled. Preserve one-to-one cancellation forwarding; do not introduce pooling or cancellation-key translation.
+- Reject unknown message tags as protocol errors, matching `pg-proto`’s fail-closed behavior.
+- Remove obsolete handwritten protocol modules and direct low-level dependencies once unused. Preserve the public configuration and CLI surfaces; retain existing `ProtocolError` variants for source compatibility even where `pg-proto` supersedes them.
+
+## Test Plan
+
+- Port existing message round-trip and rewrite tests to `pg-proto` message fixtures.
+- Add coverage for partial/oversized frames, malformed messages, unknown-tag rejection, SSL/TLS startup, cancellation, and all supported authentication modes.
+- Exercise simple queries and extended Parse/Bind/Describe/Execute/Close/Sync pipelines, including pipelining and error draining through Sync.
+- Verify text/binary formats, nulls, reshaped parameters, prepared statements, portals, COPY messages, asynchronous backend messages, and buffered DataRow decryption.
+- Run formatting, clippy, proxy unit tests, and TCP/TLS integration suites. Unset `CS_PROMETHEUS__ENABLED` for the baseline unit suite; its current environment value causes the otherwise unrelated Prometheus test to fail.
+
+## Assumptions
+
+- The existing untracked `.claude/worktrees/` directory remains untouched.
+- The worktree path and branch are currently available.
+- No compatibility feature flag or dual protocol implementation is required.
+- The plan document is left as an uncommitted worktree change unless a commit is requested separately.

packages/cipherstash-proxy/Cargo.toml

@@ -5,7 +5,6 @@ edition = "2021"
 
 [dependencies]
 async-trait = "0.1"
-aws-lc-rs = "1.13.3"
 bigdecimal = { version = "0.4.6", features = ["serde-json"] }
 blake3 = "1"
 arc-swap = "1.7.1"
@@ -29,8 +28,7 @@ md-5 = "0.10.6"
 metrics = "0.24.3"
 metrics-exporter-prometheus = "0.17"
 moka = { version = "0.12", features = ["future"] }
-oid-registry = "0.8"
-pg_escape = "0.1.1"
+pg-proto = "0.5.0"
 postgres-protocol = "0.6.7"
 postgres-types = { version = "0.2.8", features = ["with-serde_json-1"] }
 rand = "0.9"
@@ -43,7 +41,6 @@ rustls-platform-verifier = "0.5.0"
 rustls-pki-types = "1.10.0"
 serde = "1.0"
 serde_json = "1.0"
-socket2 = "0.5.7"
 sqltk = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
@@ -52,13 +49,11 @@ tokio-postgres = { version = "0.7", features = [
     "with-serde_json-1",
 ] }
 tokio-postgres-rustls = "0.13.0"
-tokio-rustls = "0.26.0"
 tokio-util = { version = "0.7.13", features = ["rt"] }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 uuid = { version = "1.11.0", features = ["serde", "v4"] }
 vitaminc-protected = "0.1.0-pre4.2"
-x509-parser = "0.17.0"
 
 
 [dev-dependencies]

packages/cipherstash-proxy/src/connect/channel_writer.rs

@@ -1,4 +1,7 @@
-use bytes::BytesMut;
+use pg_proto::{
+    codec::{BackendMessage, Frontend},
+    transport::Buffered,
+};
 use tokio::{
     io::{AsyncWrite, AsyncWriteExt},
     sync::mpsc::{self, UnboundedReceiver, UnboundedSender},
@@ -7,15 +10,15 @@ use tracing::{debug, error};
 
 use crate::log::PROTOCOL;
 
-pub type Receiver = UnboundedReceiver<BytesMut>;
-pub type Sender = UnboundedSender<BytesMut>;
+pub type Receiver = UnboundedReceiver<BackendMessage>;
+pub type Sender = UnboundedSender<BackendMessage>;
 
 #[derive(Debug)]
 pub struct ChannelWriter<W>
 where
     W: AsyncWrite + Unpin,
 {
-    writer: W,
+    writer: Buffered<W, Frontend>,
     receiver: Receiver,
     sender: Sender,
     client_id: i32,
@@ -26,11 +29,13 @@ where
     W: AsyncWrite + Unpin,
 {
     pub fn new(writer: W, client_id: i32) -> Self {
-        let (sender, receiver): (UnboundedSender<BytesMut>, UnboundedReceiver<BytesMut>) =
-            mpsc::unbounded_channel();
+        let (sender, receiver): (
+            UnboundedSender<BackendMessage>,
+            UnboundedReceiver<BackendMessage>,
+        ) = mpsc::unbounded_channel();
 
         ChannelWriter {
-            writer,
+            writer: Buffered::new_frontend(writer),
             receiver,
             sender,
             client_id,
@@ -48,14 +53,19 @@ where
         // but we're holding one of them ourselves!
         drop(self.sender);
 
-        while let Some(bytes) = self.receiver.recv().await {
+        while let Some(message) = self.receiver.recv().await {
             debug!(target: PROTOCOL,
                 client_id = self.client_id,
                 msg = "Writing",
-                ?bytes
+                ?message
             );
 
-            match self.writer.write_all(&bytes).await {
+            let result = message.to_frame().and_then(|frame| self.writer.push(frame));
+            let result = match result {
+                Ok(()) => self.writer.flush().await,
+                Err(error) => Err(error),
+            };
+            match result {
                 Ok(_) => {
                     debug!(target: PROTOCOL,
                         client_id = self.client_id,
@@ -89,7 +99,7 @@ where
         }
 
         // Shutdown the write half to send FIN and properly close the connection
-        if let Err(err) = self.writer.shutdown().await {
+        if let Err(err) = self.writer.into_inner().shutdown().await {
             error!(target: PROTOCOL,
                 client_id = self.client_id,
                 msg = "Error shutting down writer",