await worker shutdown after dropping SqliteConnection

Signed-off-by: Andrew Whitehead <[email protected]>

sqlx-core/src/sqlite/connection/mod.rs

@@ -64,9 +64,12 @@ impl Connection for SqliteConnection {
 
     fn close(mut self) -> BoxFuture<'static, Result<(), Error>> {
         Box::pin(async move {
-            self.statements.clear();
-            self.statement.take();
-            self.worker.shutdown().await
+            let shutdown = self.worker.shutdown();
+            // Drop the statement worker and any outstanding statements, which should
+            // cover all references to the connection handle outside of the worker thread
+            drop(self);
+            // Ensure the worker thread has terminated
+            shutdown.await
         })
     }
 

sqlx-core/src/sqlite/statement/worker.rs

@@ -140,13 +140,19 @@ impl StatementWorker {
     /// A `WorkerCrashed` error may be returned if the thread has already stopped.
     /// Subsequent calls to `step()`, `reset()`, or this method will fail with
     /// `WorkerCrashed`. Ensure that any associated statements are dropped first.
-    pub(crate) async fn shutdown(&mut self) -> Result<(), Error> {
+    pub(crate) fn shutdown(&mut self) -> impl Future<Output = Result<(), Error>> {
         let (tx, rx) = oneshot::channel();
 
-        self.tx
+        let send_res = self
+            .tx
             .send(StatementWorkerCommand::Shutdown { tx })
-            .map_err(|_| Error::WorkerCrashed)?;
+            .map_err(|_| Error::WorkerCrashed);
 
-        rx.await.map_err(|_| Error::WorkerCrashed)
+        async move {
+            send_res?;
+
+            // wait for the response
+            rx.await.map_err(|_| Error::WorkerCrashed)
+        }
     }
 }