update proxy protocol parsing to not a rw wrapper (#12035)

## Problem

I believe in all environments we now specify either required/rejected
for proxy-protocol V2 as required. We no longer rely on the supported
flow. This means we no longer need to keep around read bytes incase
they're not in a header.

While I designed ChainRW to be fast (the hot path with an empty buffer
is very easy to branch predict), it's still unnecessary.

## Summary of changes

* Remove the ChainRW wrapper
* Refactor how we read the proxy-protocol header using read_exact.
Slightly worse perf but it's hardly significant.
* Don't try and parse the header if it's rejected.

proxy/src/serverless/mod.rs

@@ -49,7 +49,7 @@ use crate::config::{ProxyConfig, ProxyProtocolV2};
 use crate::context::RequestContext;
 use crate::ext::TaskExt;
 use crate::metrics::Metrics;
-use crate::protocol2::{ChainRW, ConnectHeader, ConnectionInfo, read_proxy_protocol};
+use crate::protocol2::{ConnectHeader, ConnectionInfo, read_proxy_protocol};
 use crate::proxy::run_until_cancelled;
 use crate::rate_limiter::EndpointRateLimiter;
 use crate::serverless::backend::PoolingBackend;
@@ -207,12 +207,12 @@ pub(crate) type AsyncRW = Pin<Box<dyn AsyncReadWrite>>;
 
 #[async_trait]
 trait MaybeTlsAcceptor: Send + Sync + 'static {
-    async fn accept(&self, conn: ChainRW<TcpStream>) -> std::io::Result<AsyncRW>;
+    async fn accept(&self, conn: TcpStream) -> std::io::Result<AsyncRW>;
 }
 
 #[async_trait]
 impl MaybeTlsAcceptor for &'static ArcSwapOption<crate::config::TlsConfig> {
-    async fn accept(&self, conn: ChainRW<TcpStream>) -> std::io::Result<AsyncRW> {
+    async fn accept(&self, conn: TcpStream) -> std::io::Result<AsyncRW> {
         match &*self.load() {
             Some(config) => Ok(Box::pin(
                 TlsAcceptor::from(config.http_config.clone())
@@ -235,33 +235,30 @@ async fn connection_startup(
     peer_addr: SocketAddr,
 ) -> Option<(AsyncRW, ConnectionInfo)> {
     // handle PROXY protocol
-    let (conn, peer) = match read_proxy_protocol(conn).await {
-        Ok(c) => c,
-        Err(e) => {
-            tracing::warn!(?session_id, %peer_addr, "failed to accept TCP connection: invalid PROXY protocol V2 header: {e:#}");
-            return None;
+    let (conn, conn_info) = match config.proxy_protocol_v2 {
+        ProxyProtocolV2::Required => {
+            match read_proxy_protocol(conn).await {
+                Err(e) => {
+                    warn!("per-client task finished with an error: {e:#}");
+                    return None;
+                }
+                // our load balancers will not send any more data. let's just exit immediately
+                Ok((_conn, ConnectHeader::Local)) => {
+                    tracing::debug!("healthcheck received");
+                    return None;
+                }
+                Ok((conn, ConnectHeader::Proxy(info))) => (conn, info),
+            }
         }
-    };
-
-    let conn_info = match peer {
-        // our load balancers will not send any more data. let's just exit immediately
-        ConnectHeader::Local => {
-            tracing::debug!("healthcheck received");
-            return None;
-        }
-        ConnectHeader::Missing if config.proxy_protocol_v2 == ProxyProtocolV2::Required => {
-            tracing::warn!("missing required proxy protocol header");
-            return None;
-        }
-        ConnectHeader::Proxy(_) if config.proxy_protocol_v2 == ProxyProtocolV2::Rejected => {
-            tracing::warn!("proxy protocol header not supported");
-            return None;
-        }
-        ConnectHeader::Proxy(info) => info,
-        ConnectHeader::Missing => ConnectionInfo {
-            addr: peer_addr,
-            extra: None,
-        },
+        // ignore the header - it cannot be confused for a postgres or http connection so will
+        // error later.
+        ProxyProtocolV2::Rejected => (
+            conn,
+            ConnectionInfo {
+                addr: peer_addr,
+                extra: None,
+            },
+        ),
     };
 
     let has_private_peer_addr = match conn_info.addr.ip() {