Fix statement close on error (#32)

* close statement on user error

* fix fluent tests

.github/workflows/test.yml

@@ -78,7 +78,14 @@ jobs:
     container: 
       image: vapor/swift:5.2
     services:
-      mysql:
+      mysql-a:
+        image: mysql
+        env:
+          MYSQL_ALLOW_EMPTY_PASSWORD: true
+          MYSQL_DATABASE: vapor_database
+          MYSQL_USER: vapor_username
+          MYSQL_PASSWORD: vapor_password
+      mysql-b:
         image: mysql
         env:
           MYSQL_ALLOW_EMPTY_PASSWORD: true
@@ -94,4 +101,5 @@ jobs:
     - run: swift test --enable-test-discovery --sanitize=thread
       working-directory: ./fluent-mysql-driver
       env:
-        MYSQL_HOSTNAME: mysql
+        MYSQL_HOSTNAME_A: mysql-a
+        MYSQL_HOSTNAME_B: mysql-b

Sources/MySQLNIO/MySQLConnectionHandler.swift

@@ -254,7 +254,11 @@ final class MySQLConnectionHandler: ChannelDuplexHandler {
         if commandState.done {
             let current = self.queue.removeFirst()
             self.commandState = .ready
-            current.promise.succeed(())
+            if let error = commandState.error {
+                current.promise.fail(error)
+            } else {
+                current.promise.succeed(())
+            }
             self.sendEnqueuedCommandIfReady(context: context)
         }
         if commandState.resetSequence {

Sources/MySQLNIO/MySQLDatabase.swift

@@ -21,11 +21,18 @@ public struct MySQLCommandState {
     let response: [MySQLPacket]
     let done: Bool
     let resetSequence: Bool
+    var error: Error?
 
-    public init(response: [MySQLPacket] = [], done: Bool = false, resetSequence: Bool = false) {
+    public init(
+        response: [MySQLPacket] = [],
+        done: Bool = false,
+        resetSequence: Bool = false,
+        error: Error? = nil
+    ) {
         self.response = response
         self.done = done
         self.resetSequence = resetSequence
+        self.error = error
     }
 }
 

Sources/MySQLNIO/MySQLQueryCommand.swift

@@ -176,12 +176,15 @@ private final class MySQLQueryCommand: MySQLCommand {
             }
         }
         var packet = MySQLPacket()
-        MySQLProtocol.COM_STMT_CLOSE(statementID: self.ok!.statementID).encode(into: &packet)
-        if let error = self.lastUserError {
-            throw error
-        } else {
-            return .init(response: [packet], done: true, resetSequence: true)
-        }
+        MySQLProtocol.COM_STMT_CLOSE(
+            statementID: self.ok!.statementID
+        ).encode(into: &packet)
+        return .init(
+            response: [packet],
+            done: true,
+            resetSequence: true,
+            error: self.lastUserError
+        )
     }
 
     func activate(capabilities: MySQLProtocol.CapabilityFlags) throws -> MySQLCommandState {

Tests/MySQLNIOTests/NIOMySQLTests.swift

@@ -1,7 +1,8 @@
 import XCTest
 @testable import MySQLNIO
+import Logging
 
-final class NIOMySQLTests: XCTestCase {
+final class MySQLNIOTests: XCTestCase {
     private var group: EventLoopGroup!
     private var eventLoop: EventLoop {
         return self.group.next()
@@ -390,12 +391,51 @@ final class NIOMySQLTests: XCTestCase {
             XCTAssertEqual(rows[0].column("d").flatMap { Decimal(mysqlData: $0) }?.description, "3.1415926")
         }
     }
+
+    // https://github.com/vapor/mysql-nio/issues/30
+    func testPreparedStatement_maxOpen() throws {
+        let conn = try MySQLConnection.test(on: self.eventLoop).wait()
+        defer { try! conn.close().wait() }
+
+        let result = try conn.simpleQuery("SHOW VARIABLES LIKE 'max_prepared_stmt_count';").wait()
+        let max = result[0].column("Value")!.int!
+        conn.logger.info("max_prepared_stmt_count=\(max)")
+
+        struct TestError: Error { }
+        for i in 0..<(max + 1) {
+            if i % (max / 10) == 0 {
+                conn.logger.info("max_prepared_stmt_count iteration \(i)/\(max + 1)")
+            }
+            do {
+                _ = try conn.query("SELECT @@version", onRow: { row in
+                    throw TestError()
+                }).wait()
+                XCTFail("Query should have errored")
+            } catch is TestError {
+                // expected
+            }
+        }
+    }
 
     override func setUp() {
         self.group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+        XCTAssert(isLoggingConfigured)
     }
 
     override func tearDown() {
         try! self.group.syncShutdownGracefully()
     }
 }
+
+let isLoggingConfigured: Bool = {
+    LoggingSystem.bootstrap { label in
+        var handler = StreamLogHandler.standardOutput(label: label)
+        handler.logLevel = env("LOG_LEVEL").flatMap { Logger.Level(rawValue: $0) } ?? .debug
+        return handler
+    }
+    return true
+}()
+
+func env(_ name: String) -> String? {
+    ProcessInfo.processInfo.environment[name]
+}

Tests/MySQLNIOTests/Utilities.swift

@@ -17,7 +17,3 @@ extension MySQLConnection {
         }
     }
 }
-
-private func env(_ name: String) -> String? {
-    getenv(name).flatMap { String(cString: $0) }
-}

docker-compose.yml

@@ -1,6 +1,15 @@
 version: '3'
 
 services:
+  test:
+    image: mysql
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: "true"
+      MYSQL_DATABASE: vapor_database
+      MYSQL_USER: vapor_username
+      MYSQL_PASSWORD: vapor_password
+    ports:
+      - 3306:3306
   mysql-8_0:
     image: mysql:8.0
     environment: