Remove warn-concurrency warnings (#408)

Author: fabianfett

Sources/PostgresNIO/Connection/PostgresConnection+Configuration.swift

@@ -4,12 +4,12 @@ import NIOSSL
 
 extension PostgresConnection {
     /// A configuration object for a connection
-    public struct Configuration {
-    
+    public struct Configuration: Sendable {
+
         // MARK: - TLS
 
         /// The possible modes of operation for TLS encapsulation of a connection.
-        public struct TLS {
+        public struct TLS: Sendable {
             // MARK: Initializers
 
             /// Do not try to create a TLS connection to the server.
@@ -63,7 +63,7 @@ extension PostgresConnection {
         // MARK: - Connection options
 
         /// Describes options affecting how the underlying connection is made.
-        public struct Options {
+        public struct Options: Sendable {
             /// A timeout for connection attempts. Defaults to ten seconds.
             ///
             /// Ignored when using a preexisting communcation channel. (See
@@ -219,7 +219,7 @@ extension PostgresConnection {
     /// the deprecated configuration.
     ///
     /// TODO: Drop with next major release
-    struct InternalConfiguration {
+    struct InternalConfiguration: Sendable {
         enum Connection {
             case unresolvedTCP(host: String, port: Int)
             case unresolvedUDS(path: String)

Sources/PostgresNIO/Connection/PostgresConnection.swift

@@ -144,8 +144,9 @@ public final class PostgresConnection: @unchecked Sendable {
         on eventLoop: EventLoop
     ) -> EventLoopFuture<PostgresConnection> {
 
-        var logger = logger
-        logger[postgresMetadataKey: .connectionID] = "\(connectionID)"
+        var mlogger = logger
+        mlogger[postgresMetadataKey: .connectionID] = "\(connectionID)"
+        let logger = mlogger
 
         // Here we dispatch to the `eventLoop` first before we setup the EventLoopFuture chain, to
         // ensure all `flatMap`s are executed on the EventLoop (this means the enqueuing of the
@@ -567,12 +568,13 @@ extension PostgresConnection {
     ///   - line: The line, the query was started in. Used for better error reporting.
     ///   - onRow: A closure that is invoked for every row.
     /// - Returns: An EventLoopFuture, that allows access to the future ``PostgresQueryMetadata``.
+    @preconcurrency
     public func query(
         _ query: PostgresQuery,
         logger: Logger,
         file: String = #fileID,
         line: Int = #line,
-        _ onRow: @escaping (PostgresRow) throws -> ()
+        _ onRow: @escaping @Sendable (PostgresRow) throws -> ()
     ) -> EventLoopFuture<PostgresQueryMetadata> {
         self.queryStream(query, logger: logger).flatMap { rowStream in
             rowStream.onRow(onRow).flatMapThrowing { () -> PostgresQueryMetadata in
@@ -638,18 +640,19 @@ extension PostgresConnection: PostgresDatabase {
         }
     }
 
+    @preconcurrency
     public func withConnection<T>(_ closure: (PostgresConnection) -> EventLoopFuture<T>) -> EventLoopFuture<T> {
         closure(self)
     }
 }
 
 internal enum PostgresCommands: PostgresRequest {
     case query(PostgresQuery,
-               onMetadata: (PostgresQueryMetadata) -> () = { _ in },
-               onRow: (PostgresRow) throws -> ())
-    case queryAll(PostgresQuery, onResult: (PostgresQueryResult) -> ())
+               onMetadata: @Sendable (PostgresQueryMetadata) -> () = { _ in },
+               onRow: @Sendable (PostgresRow) throws -> ())
+    case queryAll(PostgresQuery, onResult: @Sendable (PostgresQueryResult) -> ())
     case prepareQuery(request: PrepareQueryRequest)
-    case executePreparedStatement(query: PreparedQuery, binds: [PostgresData], onRow: (PostgresRow) throws -> ())
+    case executePreparedStatement(query: PreparedQuery, binds: [PostgresData], onRow: @Sendable (PostgresRow) throws -> ())
 
     func respond(to message: PostgresMessage) throws -> [PostgresMessage]? {
         fatalError("This function must not be called")

Sources/PostgresNIO/Connection/PostgresDatabase+PreparedQuery.swift

@@ -1,4 +1,5 @@
 import NIOCore
+import NIOConcurrencyHelpers
 import struct Foundation.UUID
 
 extension PostgresDatabase {
@@ -14,7 +15,8 @@ extension PostgresDatabase {
         }
     }
 
-    public func prepare(query: String, handler: @escaping (PreparedQuery) -> EventLoopFuture<[[PostgresRow]]>) -> EventLoopFuture<[[PostgresRow]]> {
+    @preconcurrency
+    public func prepare(query: String, handler: @Sendable @escaping (PreparedQuery) -> EventLoopFuture<[[PostgresRow]]>) -> EventLoopFuture<[[PostgresRow]]> {
         prepare(query: query)
         .flatMap { preparedQuery in
             handler(preparedQuery)
@@ -26,7 +28,7 @@ extension PostgresDatabase {
 }
 
 
-public struct PreparedQuery {
+public struct PreparedQuery: Sendable {
     let underlying: PSQLPreparedStatement
     let database: PostgresDatabase
 
@@ -36,11 +38,16 @@ public struct PreparedQuery {
     }
 
     public func execute(_ binds: [PostgresData] = []) -> EventLoopFuture<[PostgresRow]> {
-        var rows: [PostgresRow] = []
-        return self.execute(binds) { rows.append($0) }.map { rows }
+        let rowsBoxed = NIOLockedValueBox([PostgresRow]())
+        return self.execute(binds) { row in
+            rowsBoxed.withLockedValue {
+                $0.append(row)
+            }
+        }.map { rowsBoxed.withLockedValue { $0 } }
     }
 
-    public func execute(_ binds: [PostgresData] = [], _ onRow: @escaping (PostgresRow) throws -> ()) -> EventLoopFuture<Void> {
+    @preconcurrency
+    public func execute(_ binds: [PostgresData] = [], _ onRow: @Sendable @escaping (PostgresRow) throws -> ()) -> EventLoopFuture<Void> {
         let command = PostgresCommands.executePreparedStatement(query: self, binds: binds, onRow: onRow)
         return self.database.send(command, logger: self.database.logger)
     }
@@ -50,15 +57,23 @@ public struct PreparedQuery {
     }
 }
 
-final class PrepareQueryRequest {
+final class PrepareQueryRequest: Sendable {
     let query: String
     let name: String
-    var prepared: PreparedQuery? = nil
-    
-    
+    var prepared: PreparedQuery? {
+        get {
+            self._prepared.withLockedValue { $0 }
+        }
+        set {
+            self._prepared.withLockedValue {
+                $0 = newValue
+            }
+        }
+    }
+    let _prepared: NIOLockedValueBox<PreparedQuery?> = .init(nil)
+
     init(_ query: String, as name: String) {
         self.query = query
         self.name = name
     }
-
 }

Sources/PostgresNIO/Message/PostgresMessage+Error.swift

@@ -2,8 +2,8 @@ import NIOCore
 
 extension PostgresMessage {
     /// First message sent from the frontend during startup.
-    public struct Error: CustomStringConvertible {
-        public enum Field: UInt8, Hashable {
+    public struct Error: CustomStringConvertible, Sendable {
+        public enum Field: UInt8, Hashable, Sendable {
             /// Severity: the field contents are ERROR, FATAL, or PANIC (in an error message),
             /// or WARNING, NOTICE, DEBUG, INFO, or LOG (in a notice message), or a
             //// localized translation of one of these. Always present.

Sources/PostgresNIO/New/NotificationListener.swift

@@ -44,6 +44,7 @@ final class NotificationListener: @unchecked Sendable {
 
     func startListeningSucceeded(handler: PostgresChannelHandler) {
         self.eventLoop.preconditionInEventLoop()
+        let handlerLoopBound = NIOLoopBound(handler, eventLoop: self.eventLoop)
 
         switch self.state {
         case .streamInitialized(let checkedContinuation):
@@ -55,7 +56,7 @@ final class NotificationListener: @unchecked Sendable {
                 switch reason {
                 case .cancelled:
                     eventLoop.execute {
-                        handler.cancelNotificationListener(channel: channel, id: listenerID)
+                        handlerLoopBound.value.cancelNotificationListener(channel: channel, id: listenerID)
                     }
 
                 case .finished:

Sources/PostgresNIO/New/PSQLError.swift

@@ -1,7 +1,8 @@
 import NIOCore
 
 /// An error that is thrown from the PostgresClient.
-public struct PSQLError: Error {
+/// Sendability enforced through Copy on Write semantics
+public struct PSQLError: Error, @unchecked Sendable {
 
     public struct Code: Sendable, Hashable, CustomStringConvertible {
         enum Base: Sendable, Hashable {

Sources/PostgresNIO/New/PSQLRowStream.swift

@@ -96,10 +96,8 @@ final class PSQLRowStream: @unchecked Sendable {
             let yieldResult = source.yield(contentsOf: bufferedRows)
             self.downstreamState = .asyncSequence(source, dataSource)
 
-            self.eventLoop.execute {
-                self.executeActionBasedOnYieldResult(yieldResult, source: dataSource)
-            }
-            
+            self.executeActionBasedOnYieldResult(yieldResult, source: dataSource)
+
         case .finished(let buffer, let commandTag):
             _ = source.yield(contentsOf: buffer)
             source.finish()
@@ -206,7 +204,7 @@ final class PSQLRowStream: @unchecked Sendable {
 
     // MARK: Consume on EventLoop
 
-    func onRow(_ onRow: @escaping (PostgresRow) throws -> ()) -> EventLoopFuture<Void> {
+    func onRow(_ onRow: @Sendable @escaping (PostgresRow) throws -> ()) -> EventLoopFuture<Void> {
         if self.eventLoop.inEventLoop {
             return self.onRow0(onRow)
         } else {

Sources/PostgresNIO/New/PSQLTask.swift

@@ -70,7 +70,7 @@ final class ExtendedQueryContext {
     }
 }
 
-final class PreparedStatementContext{
+final class PreparedStatementContext: Sendable {
     let name: String
     let sql: String
     let bindings: PostgresBindings

Sources/PostgresNIO/New/PostgresChannelHandler.swift

@@ -597,8 +597,10 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
             logger: self.logger,
             promise: promise
         )
+        let loopBound = NIOLoopBound((self, context), eventLoop: self.eventLoop)
         promise.futureResult.whenComplete { result in
-            self.startListenCompleted(result, for: channel, context: context)
+            let (selfTransferred, context) = loopBound.value
+            selfTransferred.startListenCompleted(result, for: channel, context: context)
         }
 
         return .extendedQuery(query)
@@ -643,8 +645,10 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
             logger: self.logger,
             promise: promise
         )
+        let loopBound = NIOLoopBound((self, context), eventLoop: self.eventLoop)
         promise.futureResult.whenComplete { result in
-            self.stopListenCompleted(result, for: channel, context: context)
+            let (selfTransferred, context) = loopBound.value
+            selfTransferred.stopListenCompleted(result, for: channel, context: context)
         }
 
         return .extendedQuery(query)
@@ -693,10 +697,12 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
         context: ChannelHandlerContext
     ) -> PSQLTask {
         let promise = self.eventLoop.makePromise(of: RowDescription?.self)
+        let loopBound = NIOLoopBound((self, context), eventLoop: self.eventLoop)
         promise.futureResult.whenComplete { result in
+            let (selfTransferred, context) = loopBound.value
             switch result {
             case .success(let rowDescription):
-                self.prepareStatementComplete(
+                selfTransferred.prepareStatementComplete(
                     name: preparedStatement.name,
                     rowDescription: rowDescription,
                     context: context
@@ -708,7 +714,7 @@ final class PostgresChannelHandler: ChannelDuplexHandler {
                 } else {
                     psqlError = .connectionError(underlying: error)
                 }
-                self.prepareStatementFailed(
+                selfTransferred.prepareStatementFailed(
                     name: preparedStatement.name,
                     error: psqlError,
                     context: context

Sources/PostgresNIO/New/PostgresCodable.swift

@@ -188,7 +188,7 @@ extension PostgresEncodingContext where JSONEncoder == Foundation.JSONEncoder {
 
 /// A context that is passed to Swift objects that are decoded from the Postgres wire format. Used
 /// to pass further information to the decoding method.
-public struct PostgresDecodingContext<JSONDecoder: PostgresJSONDecoder> {
+public struct PostgresDecodingContext<JSONDecoder: PostgresJSONDecoder>: Sendable {
     /// A ``PostgresJSONDecoder`` used to decode the object from json.
     public var jsonDecoder: JSONDecoder