Fix connection exhaustion and closing of connections on errors (#10790)

* fix connection exhaustion and closing of connections on errors

* added as maintainer

* clean up

Author: inemtsev

frameworks/Kotlin/ktor/benchmark_config.json

@@ -1,6 +1,6 @@
 {
   "framework": "ktor",
-  "maintainers": ["ShreckYe"],
+  "maintainers": ["ShreckYe", "inemtsev"],
   "tests": [
     {
       "default": {

frameworks/Kotlin/ktor/ktor-r2dbc/src/main/kotlin/org/jetbrains/ktor/benchmarks/Hello.kt

@@ -107,8 +107,8 @@ fun Application.main() {
                     .then(Mono.from(connection.commitTransaction()))
             },
                 Connection::close,
-                { connection, _ -> connection.rollbackTransaction() },
-                { connection -> connection.rollbackTransaction() }
+                { connection, _ -> Mono.from(connection.rollbackTransaction()).then(Mono.from(connection.close())) },
+                { connection -> Mono.from(connection.rollbackTransaction()).then(Mono.from(connection.close())) }
             ).awaitFirstOrNull()
 
             call.respondJson(updatedWorlds)
@@ -126,11 +126,14 @@ private suspend fun ConnectionFactory.fetchWorlds(
 ): List<World> {
     if (count <= 0) return emptyList()
     val concurrency = min(count, 32)
-    return Mono.usingWhen(create(), { connection ->
-        Flux.range(0, count)
-            .flatMap({ selectWorldPublisher(connection) }, concurrency)
-            .collectList()
-    }, Connection::close).awaitSingle()
+    return Flux.range(0, count)
+        .flatMap({
+            Mono.usingWhen(create(), { connection ->
+                selectWorldPublisher(connection)
+            }, Connection::close)
+        }, concurrency)
+        .collectList()
+        .awaitSingle()
 }
 
 private fun selectWorld(connection: Connection): Mono<World> =
@@ -186,7 +189,6 @@ private fun configurePostgresR2DBC(config: ApplicationConfig): ConnectionFactory
         .maxSize(config.property("db.maxPoolSize").getString().toInt())
         .maxIdleTime(Duration.ofSeconds(30))
         .maxAcquireTime(Duration.ofSeconds(5))
-        .validationQuery("SELECT 1")
         .build()
 
     return ConnectionPool(cp)

frameworks/Kotlin/ktor/ktor-r2dbc/src/main/resources/application.conf

@@ -18,8 +18,8 @@ db {
     host = "tfb-database"
     port = 5432
     database = "hello_world"
-    initPoolSize = 512
-    maxPoolSize = 512
+    initPoolSize = 32
+    maxPoolSize = 256
     username = "benchmarkdbuser"
     password = "benchmarkdbpass"
     //url = "r2dbc:postgresql://"${db.host}":"${db.port}"/"${db.database}"?loggerLevel=OFF&disableColumnSanitiser=true&assumeMinServerVersion=16&sslmode=disable&maxSize="${db.poolSize}