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/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,30 @@ 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());
+
+    // one query fails as the statement is not cached server-side any more, client-side cache is cleared
+    assert!(sqlx::query(query).fetch_one(&mut conn).await.is_err());
+    assert_eq!(0, conn.cached_statements_size());
+
+    // next query succeeds again
+    let _ = sqlx::query(query).fetch_one(&mut conn).await?;
+
+    Ok(())
+}
+
 #[sqlx_macros::test]
 async fn it_can_fail_at_querying() -> anyhow::Result<()> {
     let mut conn = new::<MySql>().await?;