Log storage_query errors on warn

Logging the storage_query errors on warn allows better insights
into why a connection might be rejected by a client. Currently,
a user of Restate runs into the problem that the connection gets
rejected but it is unclear why.

This fixes #740.

src/storage_query/Cargo.toml

@@ -35,3 +35,4 @@ paste = { workspace = true}
 schemars = { workspace = true, optional = true }
 serde = { workspace = true }
 thiserror = { workspace = true }
+tracing = { workspace = true }

src/storage_query/src/pgwire_server.rs

@@ -8,6 +8,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::net::SocketAddr;
 use std::sync::Arc;
 
 use async_trait::async_trait;
@@ -40,6 +41,7 @@ use pgwire::api::{ClientInfo, MakeHandler, StatelessMakeHandler, Type};
 use pgwire::error::{ErrorInfo, PgWireError, PgWireResult};
 use pgwire::messages::data::DataRow;
 use pgwire::tokio::process_socket;
+use tracing::warn;
 
 pub(crate) struct HandlerFactory {
     processor: Arc<StatelessMakeHandler<DfSessionService>>,
@@ -65,19 +67,23 @@ impl HandlerFactory {
         }
     }
 
-    pub fn spawn_connection(&self, incoming_socket: TcpStream) {
+    pub fn spawn_connection(&self, incoming_socket: TcpStream, addr: SocketAddr) {
         let authenticator_ref = self.authenticator.make();
         let processor_ref = self.processor.make();
         let placeholder_ref = self.placeholder.make();
         tokio::spawn(async move {
-            let _ = process_socket(
+            let result = process_socket(
                 incoming_socket,
                 None,
                 authenticator_ref,
                 processor_ref,
                 placeholder_ref,
             )
             .await;
+
+            if let Err(err) = result {
+                warn!("Failed processing socket for connection '{addr}': {err}");
+            }
         });
     }
 }

src/storage_query/src/service.rs

@@ -16,6 +16,7 @@ use std::io::ErrorKind;
 use std::net::SocketAddr;
 use tokio::net::TcpListener;
 use tokio::select;
+use tracing::warn;
 
 pub type GenericError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
@@ -66,8 +67,11 @@ impl PostgresQueryService {
         loop {
             select! {
                 incoming_socket = listener.accept() => {
-                    if let Ok((stream, _addr)) = incoming_socket {
-                        factory.spawn_connection(stream);
+                    match incoming_socket {
+                        Ok((stream, addr)) => factory.spawn_connection(stream, addr),
+                        Err(err) => {
+                            warn!("Failed to accept storage query connection: {err}");
+                        }
                     }
                 },
                 _ = &mut shutdown => {