feat(transaction): Adding withTransaction

Sources/PostgresNIO/Pool/PostgresClient.swift

@@ -303,6 +303,28 @@ public final class PostgresClient: Sendable, ServiceLifecycle.Service {
 
         return try await closure(connection)
     }
+
+    /// Lease a connection for the provided `closure`'s lifetime.
+    /// A transation starts with call to withConnection
+    /// A transaction should end with a call to COMMIT or ROLLBACK
+    /// COMMIT is called upon successful completion and ROLLBACK is called should any steps fail
+    ///
+    /// - Parameter closure: A closure that uses the passed `PostgresConnection`. The closure **must not** capture
+    ///                      the provided `PostgresConnection`.
+    /// - Returns: The closure's return value.
+    public func withTransaction<Result>(logger: Logger, _ process: (PostgresConnection) async throws -> Result) async throws -> Result {
+        try await withConnection { connection in
+            do {
+                try await connection.query("BEGIN;", logger: logger)
+                let value = try await process(connection)
+                try await connection.query("COMMIT;", logger: logger)
+                return value
+            } catch {
+                try await connection.query("ROLLBACK;", logger: logger)
+                throw error
+            }
+        }
+    }
 
     /// Run a query on the Postgres server the client is connected to.
     ///

Tests/IntegrationTests/PostgresClientTests.swift

@@ -42,6 +42,41 @@ final class PostgresClientTests: XCTestCase {
             taskGroup.cancelAll()
         }
     }
+
+    func testTransaction() async throws {
+        var mlogger = Logger(label: "test")
+        mlogger.logLevel = .debug
+        let logger = mlogger
+        let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 8)
+        self.addTeardownBlock {
+            try await eventLoopGroup.shutdownGracefully()
+        }
+
+        let clientConfig = PostgresClient.Configuration.makeTestConfiguration()
+        let client = PostgresClient(configuration: clientConfig, eventLoopGroup: eventLoopGroup, backgroundLogger: logger)
+
+        await withThrowingTaskGroup(of: Void.self) { taskGroup in
+            taskGroup.addTask {
+                await client.run()
+            }
+
+            let iterations = 1000
+
+            for _ in 0..<iterations {
+                taskGroup.addTask {
+                    try await client.withTransaction(logger: logger) { connection in
+                        _ = try await connection.query("SELECT 1", logger: logger)
+                    }
+                }
+            }
+
+            for _ in 0..<iterations {
+                _ = await taskGroup.nextResult()!
+            }
+
+            taskGroup.cancelAll()
+        }
+    }
 
     func testQueryDirectly() async throws {
         var mlogger = Logger(label: "test")