forward COM_PING to backend connection

When the connection is idle in some proxies/connection pool's, it's
common to refresh the connection by sending a COM_PING(MySQL) or
select 1 query(Postgres). Currently we are not forwarding the COM_PING
to the backend connection, which can cause the connection to be closed
by the backend server.

Fixes: REA-4887
Fixes: #1387

Release-Note-Core: Fixes an issue that we misse to forward COM_PING to
  backend connection to keep the connection alive.
Change-Id: I92178a37d58f8336a669c2f259ca9a478b856973
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/8112
Tested-by: Buildkite CI
Reviewed-by: Jason Brown <jason.b@readyset.io>

Author: altmannmarcelo

mysql-srv/src/lib.rs

@@ -48,6 +48,10 @@
 //!     }
 //!     async fn on_close(&mut self, _: DeallocateId) {}
 //!
+//!     async fn on_ping(&mut self) -> io::Result<()> {
+//!         Ok(())
+//!     }
+//!
 //!     async fn on_reset(&mut self) -> io::Result<()> {
 //!         Ok(())
 //!     }
@@ -293,6 +297,9 @@ pub trait MySqlShim<W: AsyncWrite + Unpin + Send> {
         results: QueryResultWriter<'_, W>,
     ) -> QueryResultsResponse;
 
+    /// Called when the client issue a ping command.
+    async fn on_ping(&mut self) -> io::Result<()>;
+
     /// Called when the client issues a reset command
     async fn on_reset(&mut self) -> io::Result<()>;
 
@@ -818,6 +825,7 @@ impl<B: MySqlShim<W> + Send, R: AsyncRead + Unpin, W: AsyncWrite + Unpin + Send>
                         .await?;
                 }
                 Command::Ping => {
+                    self.shim.on_ping().await?;
                     writers::write_ok_packet(&mut self.writer, 0, 0, StatusFlags::empty()).await?;
                     self.writer.flush().await?;
                 }

mysql-srv/tests/main.rs

@@ -91,6 +91,10 @@ where
 
     async fn on_close(&mut self, _: DeallocateId) {}
 
+    async fn on_ping(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+
     async fn on_reset(&mut self) -> io::Result<()> {
         Ok(())
     }

readyset-adapter/src/backend.rs

@@ -859,6 +859,14 @@ where
             .unwrap_or_else(|| DB::DEFAULT_DB_VERSION.to_string())
     }
 
+    /// Send ping on the upstream connection, if it exists
+    pub async fn ping(&mut self) -> Result<(), DB::Error> {
+        if let Some(upstream) = &mut self.upstream {
+            upstream.ping().await
+        } else {
+            Ok(())
+        }
+    }
     /// Reset the current upstream connection
     pub async fn reset(&mut self) -> Result<(), DB::Error> {
         if let Some(upstream) = &mut self.upstream {

readyset-adapter/src/upstream_database.rs

@@ -110,6 +110,9 @@ pub trait UpstreamDatabase: Sized + Send {
         database: &str,
     ) -> Result<(), Self::Error>;
 
+    /// Ping the upstream connection to see if it is still alive
+    async fn ping(&mut self) -> Result<(), Self::Error>;
+
     /// Reset the connection to the upstream database
     async fn reset(&mut self) -> Result<(), Self::Error>;
 
@@ -280,6 +283,14 @@ where
             .await
     }
 
+    async fn ping(&mut self) -> Result<(), Self::Error> {
+        if let Some(u) = &mut self.upstream {
+            u.ping().await
+        } else {
+            Ok(())
+        }
+    }
+
     async fn reset(&mut self) -> Result<(), Self::Error> {
         self.upstream().await?.reset().await
     }

readyset-mysql/src/backend.rs

@@ -746,6 +746,11 @@ where
         let _ = self.noria.remove_statement(statement_id).await;
     }
 
+    async fn on_ping(&mut self) -> std::io::Result<()> {
+        self.ping()
+            .await
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))
+    }
     async fn on_reset(&mut self) -> io::Result<()> {
         let _ = match self.reset().await {
             Ok(()) => Ok(()),

readyset-mysql/src/upstream.rs

@@ -257,6 +257,11 @@ impl UpstreamDatabase for MySqlUpstream {
         Ok(self.conn.ping().await.is_ok())
     }
 
+    async fn ping(&mut self) -> Result<(), Self::Error> {
+        self.conn.ping().await.map_err(Error::MySql)?;
+        Ok(())
+    }
+
     async fn change_user(
         &mut self,
         user: &str,

readyset-psql/src/upstream.rs

@@ -239,6 +239,11 @@ impl UpstreamDatabase for PostgreSqlUpstream {
         Ok(!self.client.simple_query("select 1").await?.is_empty())
     }
 
+    async fn ping(&mut self) -> Result<(), Self::Error> {
+        self.client.simple_query("select 1").await?;
+        Ok(())
+    }
+
     async fn change_user(
         &mut self,
         _user: &str,