WIP: Hyper 1.0 migration - RPC server fixes

- Simplified rpc/mod.rs to use tonic's serve_with_incoming
- Added tonic::transport::server::Connected bound to Accept trait
- Added From<hyper_util::client::legacy::Error> for LoadDumpError
- Down to 39 errors from 43

Author: devonshigaki

libsql-server/src/error.rs

@@ -298,6 +298,12 @@ pub enum LoadDumpError {
     InvalidSqlInput(String),
 }
 
+impl From<hyper_util::client::legacy::Error> for LoadDumpError {
+    fn from(e: hyper_util::client::legacy::Error) -> Self {
+        LoadDumpError::Internal(format!("HTTP client error: {}", e))
+    }
+}
+
 impl ResponseError for LoadDumpError {}
 
 impl IntoResponse for &LoadDumpError {

libsql-server/src/net.rs

@@ -102,7 +102,7 @@ pub trait Conn: AsyncRead + AsyncWrite + Read + Write + Unpin + Send + 'static {
 /// Trait for accepting incoming connections.
 /// This is the hyper 1.0+ compatible version that replaces `hyper::server::accept::Accept`.
 pub trait Accept: Unpin + Send + 'static {
-    type Connection: Conn;
+    type Connection: Conn + Connected<ConnectInfo = TcpConnectInfo>;
     type Error: std::error::Error + Send + Sync + 'static;
 
     fn poll_accept(

libsql-server/src/rpc/mod.rs

@@ -2,24 +2,25 @@ use std::future::poll_fn;
 use std::pin::Pin;
 use std::sync::Arc;
 
-use hyper_util::rt::{TokioExecutor, TokioIo};
-use hyper_util::server::conn::auto::Builder as ConnBuilder;
+use async_stream::try_stream;
+use futures::Stream;
 use libsql_replication::rpc::replication::replication_log_server::ReplicationLogServer;
 use libsql_replication::rpc::replication::{BoxReplicationService, NAMESPACE_METADATA_KEY};
 use rustls::pki_types::CertificateDer;
 use rustls::RootCertStore;
+use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_rustls::TlsAcceptor;
+use tonic::transport::server::Connected;
 use tonic::Status;
 use tower::util::option_layer;
-use tower::Service;
 use tower::ServiceBuilder;
 use tower_http::trace::DefaultOnResponse;
 use tracing::Span;
 
 use crate::config::TlsConfig;
 use crate::metrics::CLIENT_VERSION;
 use crate::namespace::NamespaceName;
-use crate::net::{Accept, Conn};
+use crate::net::Accept;
 use crate::rpc::proxy::rpc::proxy_server::ProxyServer;
 use crate::rpc::proxy::ProxyService;
 use crate::utils::services::idle_shutdown::IdleShutdownKicker;
@@ -31,18 +32,14 @@ pub mod streaming_exec;
 
 pub async fn run_rpc_server<A: Accept>(
     proxy_service: ProxyService,
-    mut acceptor: A,
+    acceptor: A,
     maybe_tls: Option<TlsConfig>,
     idle_shutdown_layer: Option<IdleShutdownKicker>,
     service: BoxReplicationService,
 ) -> anyhow::Result<()> {
-    let router = tonic::transport::Server::builder()
+    // Build the tonic server with services
+    let mut server = tonic::transport::Server::builder()
         .layer(&option_layer(idle_shutdown_layer))
-        .add_service(ProxyServer::new(proxy_service))
-        .add_service(ReplicationLogServer::new(service))
-        .into_router();
-
-    let svc = ServiceBuilder::new()
         .layer(
             tower_http::trace::TraceLayer::new_for_grpc()
                 .on_request(trace_request)
@@ -51,63 +48,26 @@ pub async fn run_rpc_server<A: Accept>(
                         .level(tracing::Level::DEBUG)
                         .latency_unit(tower_http::LatencyUnit::Micros),
                 ),
-        )
-        .service(router);
+        );
+
+    let router = server
+        .add_service(ProxyServer::new(proxy_service))
+        .add_service(ReplicationLogServer::new(service));
 
     if let Some(tls_config) = maybe_tls {
-        run_tls_server(&mut acceptor, svc, tls_config).await
+        run_tls_server(acceptor, router, tls_config).await
     } else {
-        run_plain_server(&mut acceptor, svc).await
+        run_plain_server(acceptor, router).await
     }
 }
 
-/// Wrapper service that converts hyper 1.0's Incoming body to tonic's BoxBody
-#[derive(Clone)]
-struct TonicServiceWrapper<S> {
-    inner: S,
-}
-
-impl<S, B> Service<hyper::Request<hyper::body::Incoming>> for TonicServiceWrapper<S>
-where
-    S: Service<hyper::Request<tonic::body::BoxBody>, Response = hyper::Response<B>, Error = std::convert::Infallible> + Clone + Send + 'static,
-    S::Future: Send + 'static,
-    B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
-    B::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + Sync + 'static,
-{
-    type Response = hyper::Response<B>;
-    type Error = std::convert::Infallible;
-    type Future = S::Future;
-
-    fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
-        self.inner.poll_ready(cx)
-    }
-
-    fn call(&mut self, req: hyper::Request<hyper::body::Incoming>) -> Self::Future {
-        // Convert Incoming body to tonic's BoxBody
-        // Need to map the error type from io::Error to tonic::Status
-        let (parts, body) = req.into_parts();
-        let body = body.map_err(|e| tonic::Status::internal(format!("body error: {}", e)));
-        let body = tonic::body::BoxBody::new(body);
-        let req = hyper::Request::from_parts(parts, body);
-        self.inner.call(req)
-    }
-}
-
-async fn run_tls_server<A, S, B>(
-    acceptor: &mut A,
-    svc: S,
+async fn run_tls_server<A>(
+    acceptor: A,
+    router: tonic::transport::server::Router,
     tls_config: TlsConfig,
 ) -> anyhow::Result<()>
 where
     A: Accept,
-    S: tower::Service<hyper::Request<tonic::body::BoxBody>, Response = hyper::Response<B>, Error = std::convert::Infallible>
-        + Clone
-        + Send
-        + 'static,
-    S::Future: Send + 'static,
-    S::Response: Send + 'static,
-    B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
-    B::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + Sync + 'static,
 {
     let cert_pem = tokio::fs::read_to_string(&tls_config.cert).await?;
     let certs: Vec<CertificateDer<'static>> = rustls_pemfile::certs(&mut cert_pem.as_bytes())
@@ -116,7 +76,7 @@ where
     let key_pem = tokio::fs::read_to_string(&tls_config.key).await?;
     let keys: Vec<_> = rustls_pemfile::pkcs8_private_keys(&mut key_pem.as_bytes())
         .collect::<Result<Vec<_>, _>>()?;
-    let key = rustls::pki_types::PrivateKeyDer::try_from(keys.into_iter().next().ok_or_else(|| anyhow::anyhow!("no private keys found"))?)?
+    let key = rustls::pki_types::PrivateKeyDer::try_from(keys.into_iter().next().ok_or_else(|| anyhow::anyhow!("no private keys found"))?)?;
 
     let ca_cert_pem = std::fs::read_to_string(&tls_config.ca_cert)?;
     let ca_certs: Vec<CertificateDer<'static>> = rustls_pemfile::certs(&mut ca_cert_pem.as_bytes())
@@ -137,90 +97,136 @@ where
     let tls_acceptor = TlsAcceptor::from(Arc::new(config));
 
     tracing::info!("serving internal rpc server with tls");
-    
-    let wrapped_svc = TonicServiceWrapper { inner: svc };
-
-    // Drive the acceptor stream manually for hyper 1.0+ compatibility
-    loop {
-        let conn = match poll_fn(|cx| Pin::new(&mut *acceptor).poll_accept(cx)).await {
-            Some(Ok(conn)) => conn,
-            Some(Err(e)) => {
-                tracing::error!("Accept error: {}", e);
-                continue;
-            }
-            None => break,
-        };
-
-        let tls_acceptor = tls_acceptor.clone();
-        let svc = wrapped_svc.clone();
-
-        tokio::spawn(async move {
+
+    // Create a stream of TLS connections from the acceptor
+    let incoming = tls_incoming_stream(acceptor, tls_acceptor);
+
+    // Serve with tonic's native server
+    router.serve_with_incoming(incoming).await?;
+
+    Ok(())
+}
+
+async fn run_plain_server<A>(
+    acceptor: A,
+    router: tonic::transport::server::Router,
+) -> anyhow::Result<()>
+where
+    A: Accept,
+{
+    tracing::info!("serving internal rpc server without tls");
+
+    // Create a stream of connections from the acceptor
+    let incoming = plain_incoming_stream(acceptor);
+
+    // Serve with tonic's native server
+    router.serve_with_incoming(incoming).await?;
+
+    Ok(())
+}
+
+fn tls_incoming_stream<A>(
+    mut acceptor: A,
+    tls_acceptor: TlsAcceptor,
+) -> impl Stream<Item = Result<TlsStream<A::Connection>, anyhow::Error>>
+where
+    A: Accept,
+{
+    try_stream! {
+        loop {
+            let conn = match poll_fn(|cx| Pin::new(&mut acceptor).poll_accept(cx)).await {
+                Some(Ok(conn)) => conn,
+                Some(Err(e)) => {
+                    tracing::error!("Accept error: {}", e);
+                    continue;
+                }
+                None => break,
+            };
+
             let tls_stream = match tls_acceptor.accept(conn).await {
                 Ok(tls_stream) => tls_stream,
                 Err(err) => {
                     tracing::error!("failed to perform tls handshake: {:#}", err);
-                    return;
+                    continue;
                 }
             };
 
-            let io = TokioIo::new(tls_stream);
+            yield TlsStream(tls_stream);
+        }
+    }
+}
 
-            if let Err(err) = ConnBuilder::new(TokioExecutor::new())
-                .serve_connection(io, svc)
-                .await
-            {
-                tracing::error!("failed to serve connection: {:#}", err);
-            }
-        });
+fn plain_incoming_stream<A>(
+    mut acceptor: A,
+) -> impl Stream<Item = Result<A::Connection, anyhow::Error>>
+where
+    A: Accept,
+{
+    try_stream! {
+        loop {
+            let conn = match poll_fn(|cx| Pin::new(&mut acceptor).poll_accept(cx)).await {
+                Some(Ok(conn)) => conn,
+                Some(Err(e)) => {
+                    tracing::error!("Accept error: {}", e);
+                    continue;
+                }
+                None => break,
+            };
+
+            yield conn;
+        }
     }
+}
 
-    Ok(())
+// Wrapper for TLS stream to implement Connected
+pub struct TlsStream<S>(tokio_rustls::server::TlsStream<S>);
+
+impl<S> AsyncRead for TlsStream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        Pin::new(&mut self.get_mut().0).poll_read(cx, buf)
+    }
 }
 
-async fn run_plain_server<A, S, B>(
-    acceptor: &mut A,
-    svc: S,
-) -> anyhow::Result<()>
+impl<S> AsyncWrite for TlsStream<S>
 where
-    A: Accept,
-    S: tower::Service<hyper::Request<tonic::body::BoxBody>, Response = hyper::Response<B>, Error = std::convert::Infallible>
-        + Clone
-        + Send
-        + 'static,
-    S::Future: Send + 'static,
-    S::Response: Send + 'static,
-    B: http_body::Body<Data = bytes::Bytes> + Send + 'static,
-    B::Error: Into<Box<dyn std::error::Error + Send + Sync>> + Send + Sync + 'static,
+    S: AsyncRead + AsyncWrite + Unpin,
 {
-    tracing::info!("serving internal rpc server without tls");
-    let wrapped_svc = TonicServiceWrapper { inner: svc };
-
-    // Drive the acceptor stream manually for hyper 1.0+ compatibility
-    loop {
-        let conn = match poll_fn(|cx| Pin::new(&mut *acceptor).poll_accept(cx)).await {
-            Some(Ok(conn)) => conn,
-            Some(Err(e)) => {
-                tracing::error!("Accept error: {}", e);
-                continue;
-            }
-            None => break,
-        };
-
-        let svc = wrapped_svc.clone();
-
-        tokio::spawn(async move {
-            let io = TokioIo::new(conn);
-
-            if let Err(err) = ConnBuilder::new(TokioExecutor::new())
-                .serve_connection(io, svc)
-                .await
-            {
-                tracing::error!("failed to serve connection: {:#}", err);
-            }
-        });
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        Pin::new(&mut self.get_mut().0).poll_write(cx, buf)
     }
 
-    Ok(())
+    fn poll_flush(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        Pin::new(&mut self.get_mut().0).poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        Pin::new(&mut self.get_mut().0).poll_shutdown(cx)
+    }
+}
+
+impl<S: Connected> Connected for TlsStream<S> {
+    type ConnectInfo = S::ConnectInfo;
+
+    fn connect_info(&self) -> Self::ConnectInfo {
+        self.0.get_ref().0.connect_info()
+    }
 }
 
 fn extract_namespace<T>(
@@ -242,7 +248,7 @@ fn extract_namespace<T>(
     }
 }
 
-fn trace_request<B>(req: &hyper::Request<B>, span: &Span) {
+fn trace_request<B>(req: &http::Request<B>, span: &Span) {
     let _s = span.enter();
 
     tracing::debug!(