Add a bad integration test for zero-downtime patches

I think it's best to test this issue through integration tests, but
there seems to be no way to clear the server-side prepared statements
without access to the raw packet stream, as they're not named and thus
cannot be deleted via DEALLOCATE PREPARED.

To make the test work, additional interfaces have to be made available,
which should not be shipped as part of the library for regular use.

Author: sulami

sqlx-core/src/common/statement_cache.rs

@@ -72,4 +72,8 @@ impl<T> StatementCache<T> {
     pub fn is_enabled(&self) -> bool {
         self.capacity() > 0
     }
+
+    pub fn iter(&self) -> impl Iterator<Item=&T> {
+        self.inner.iter().map(|(_, v)| v)
+    }
 }

sqlx-mysql/src/connection/executor.rs

@@ -183,7 +183,7 @@ impl MySqlConnection {
             loop {
                 // query response is a meta-packet which may be one of:
                 //  Ok, Err, ResultSet, or (unhandled) LocalInfileRequest
-                let mut packet = self.inner.stream.recv_packet().await.inspect_err(|_| {
+                let mut packet = self.inner.stream.recv_packet().await.inspect_err(|e| {
                     // if a prepared statement vanished on the server side, we get an error here
                     // clear the statement cache in case the connection got reset to cause re-preparing
                     self.inner.cache_statement.clear();

sqlx-mysql/src/connection/mod.rs

@@ -57,6 +57,19 @@ impl MySqlConnection {
             .status_flags
             .intersects(Status::SERVER_STATUS_IN_TRANS)
     }
+
+    pub async fn nuke_cached_statements(&mut self) -> Result<(), Error> {
+        for (statement_id, _) in self.inner.cache_statement.iter() {
+            self.inner
+                .stream
+                .send_packet(StmtClose {
+                    statement: *statement_id,
+                })
+                .await?;
+        }
+
+        Ok(())
+    }
 }
 
 impl Debug for MySqlConnection {

tests/mysql/mysql.rs

@@ -54,6 +54,26 @@ async fn it_maths() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[sqlx_macros::test]
+async fn it_clears_statement_cache_on_error() -> anyhow::Result<()> {
+    setup_if_needed();
+
+    let query = "SELECT 1";
+
+    let mut conn = new::<MySql>().await?;
+    let _ = sqlx::query(query).fetch_one(&mut conn).await?;
+    assert_eq!(1, conn.cached_statements_size());
+
+    // clear cached statements only on the server side
+    conn.nuke_cached_statements().await?;
+    assert_eq!(1, conn.cached_statements_size());
+
+    assert!(sqlx::query(query).fetch_one(&mut conn).await.is_err());
+    assert_eq!(0, conn.cached_statements_size());
+
+    Ok(())
+}
+
 #[sqlx_macros::test]
 async fn it_can_fail_at_querying() -> anyhow::Result<()> {
     let mut conn = new::<MySql>().await?;