Propagate errors

Author: mariosangiorgio

Sources/PostgresNIO/New/Connection State Machine/PreparedStatementStateMachine.swift

@@ -4,15 +4,14 @@ struct PreparedStatementStateMachine {
     enum State {
         case preparing([PreparedStatementContext])
         case prepared(RowDescription?)
-        // TODO: store errors
-        //        case error(PostgresBackendMessage.ErrorResponse)
+        case error(PSQLError)
     }
 
     enum Action {
-        case none
         case prepareStatement
         case waitForAlreadyInFlightPreparation
         case executePendingStatements([PreparedStatementContext], RowDescription?)
+        case returnError([PreparedStatementContext], PSQLError)
     }
 
     var preparedStatements: [String: State]
@@ -30,6 +29,8 @@ struct PreparedStatementStateMachine {
                 return .waitForAlreadyInFlightPreparation
             case .prepared(let rowDescription):
                 return .executePendingStatements([context], rowDescription)
+            case .error(let error):
+                return .returnError([context], error)
             }
         } else {
             self.preparedStatements[name] = .preparing([context])
@@ -41,24 +42,27 @@ struct PreparedStatementStateMachine {
         name: String,
         rowDescription: RowDescription?
     ) -> Action {
-        guard let state = self.preparedStatements[name] else {
+        guard case .preparing(let statements) = self.preparedStatements[name] else {
             preconditionFailure("Preparation completed for an unexpected statement")
         }
-        switch state {
-        case .preparing(let statements):
-            // When sending the bindings we are going to ask for binary data.
-            if var rowDescription {
-                for i in 0..<rowDescription.columns.count {
-                    rowDescription.columns[i].format = .binary
-                }
-                self.preparedStatements[name] = .prepared(rowDescription)
-                return .executePendingStatements(statements, rowDescription)
-            } else {
-                self.preparedStatements[name] = .prepared(nil)
-                return .executePendingStatements(statements, nil)
+        // When sending the bindings we are going to ask for binary data.
+        if var rowDescription {
+            for i in 0..<rowDescription.columns.count {
+                rowDescription.columns[i].format = .binary
             }
-        case .prepared(_):
-            return .none
+            self.preparedStatements[name] = .prepared(rowDescription)
+            return .executePendingStatements(statements, rowDescription)
+        } else {
+            self.preparedStatements[name] = .prepared(nil)
+            return .executePendingStatements(statements, nil)
+        }
+    }
+
+    mutating func errorHappened(name: String, error: PSQLError) -> Action {
+        guard case .preparing(let statements) = self.preparedStatements[name] else {
+            preconditionFailure("Preparation completed for an unexpected statement")
         }
+        self.preparedStatements[name] = .error(error)
+        return .returnError(statements, error)
     }
 }

Sources/PostgresNIO/New/PostgresChannelHandler.swift

@@ -243,8 +243,6 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
                 name: preparedStatement.name,
                 context: preparedStatement
             ) {
-            case .none:
-                return
             case .prepareStatement:
                 let promise = self.eventLoop.makePromise(of: RowDescription?.self)
                 promise.futureResult.whenSuccess { rowDescription in
@@ -254,6 +252,13 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
                         context: context
                     )
                 }
+                promise.futureResult.whenFailure { error in
+                    self.prepareStatementFailed(
+                        name: preparedStatement.name,
+                        error: error as! PSQLError,
+                        context: context
+                    )
+                }
                 psqlTask = .preparedStatement(.init(
                     name: preparedStatement.name,
                     query: preparedStatement.sql,
@@ -277,6 +282,11 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
                     self.run(action, with: context)
                 }
                 return
+            case .returnError(let pendingStatements, let error):
+                for statement in pendingStatements {
+                    statement.promise.fail(error)
+                }
+                return
             }
         }
 
@@ -735,6 +745,23 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
             self.run(action, with: context)
         }
     }
+
+    private func prepareStatementFailed(
+        name: String,
+        error: PSQLError,
+        context: ChannelHandlerContext
+    ) {
+        let action = self.preparedStatementState.errorHappened(
+            name: name,
+            error: error
+        )
+        guard case .returnError(let statements, let error) = action else {
+            preconditionFailure("Expected to have pending statements to execute")
+        }
+        for statement in statements {
+            statement.promise.fail(error)
+        }
+    }
 }
 
 extension PostgresChannelHandler: PSQLRowsDataSource {