From fc0ccc94a3ee010db57a005cda17e728045c9eae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Sun, 2 Aug 2026 11:56:13 +0200 Subject: [PATCH 01/12] refactor(driver-multiplatform-wire): Use a simple Deferred instead of a Channel in the common case --- .../src/commonMain/kotlin/MongoWireClient.kt | 48 +++++++++---------- 1 file changed, 24 insertions(+), 24 deletions(-) diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt index 6715b11d..b380c88c 100644 --- a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt @@ -31,7 +31,6 @@ import opensavvy.ktmongo.bson.multiplatform.BsonDocument import opensavvy.ktmongo.bson.multiplatform.BsonFactory import opensavvy.ktmongo.dsl.LowLevelApi import kotlin.coroutines.CoroutineContext -import kotlin.coroutines.cancellation.CancellationException @LowLevelApi interface MongoWireClient : AutoCloseable { @@ -71,16 +70,19 @@ private class SocketWireClient( coroutineScope: CoroutineScope, ) : MongoWireClient { + private sealed class ResponseHandler { + data class Single(val result: CompletableDeferred) : ResponseHandler() + data class Multiple(val result: SendChannel) : ResponseHandler() + } + private class Request( val data: Buffer, - val output: Channel, - val expectsMultipleResponses: Boolean, + val output: ResponseHandler, ) private class SentMessage( val requestId: Int, - val output: Channel, - val expectsMultipleResponses: Boolean, + val output: ResponseHandler, ) private class Response( @@ -91,7 +93,7 @@ private class SocketWireClient( private class ResponseWithHandler( val response: Response, - val output: SendChannel, + val output: ResponseHandler, ) /** @@ -170,7 +172,7 @@ private class SocketWireClient( writeSocket.flush() log("$requestId was sent") - sentChannel.send(SentMessage(requestId, request.output, expectsMultipleResponses = request.expectsMultipleResponses)) + sentChannel.send(SentMessage(requestId, request.output)) } } @@ -207,8 +209,7 @@ private class SocketWireClient( receivedChannel: ReceiveChannel, triagedChannel: SendChannel, ) { - val waiting = HashMap>() - val requestsExpectingMultipleResponses = HashSet() + val waiting = HashMap() while (currentCoroutineContext().isActive && socket.isActive) { select { @@ -219,16 +220,13 @@ private class SocketWireClient( sentChannel.onReceive { message -> log("${message.requestId} expects an answer") waiting[message.requestId] = message.output - if (message.expectsMultipleResponses) - requestsExpectingMultipleResponses.add(message.requestId) } receivedChannel.onReceive { response -> val handler = waiting[response.responseTo] ?: error("Received the message ${response.requestId} in response to ${response.responseTo}, but no known message with ID ${response.responseTo} has been sent by this client.\nCurrently in-flight requests: ${waiting.keys.sorted()}") triagedChannel.send(ResponseWithHandler(response, handler)) - if (response.responseTo !in requestsExpectingMultipleResponses) { - requestsExpectingMultipleResponses.remove(response.responseTo) + if (handler is ResponseHandler.Single) { waiting.remove(response.responseTo) } } @@ -291,13 +289,16 @@ private class SocketWireClient( val body = sections.singleOrNull { it is MessageSection.Body } as? MessageSection.Body ?: error("An OP_MSG message must have a single body section, found: $sections") - received.output.send( - Message.OpMsg( - body, - sections.asSequence() - .filterIsInstance(), - ) + val response = Message.OpMsg( + body, + sections.asSequence() + .filterIsInstance(), ) + + when (received.output) { + is ResponseHandler.Single -> received.output.result.complete(response) + is ResponseHandler.Multiple -> received.output.result.send(response) + } } } @@ -398,17 +399,16 @@ private class SocketWireClient( val output = Channel() log("Preparing to write $message…") val buffer = writeMessage(message) - requestChannel.send(Request(buffer, output, expectsMultipleResponses = true)) + requestChannel.send(Request(buffer, ResponseHandler.Multiple(output))) return output } override suspend fun sendSingle(message: Message): Message { - val output = Channel() + val output = CompletableDeferred() log("Preparing to write $message…") val buffer = writeMessage(message) - requestChannel.send(Request(buffer, output, expectsMultipleResponses = false)) - val message = output.receive() - output.close(CancellationException("We expected a single response, and we received it, so this channel was closed.")) + requestChannel.send(Request(buffer, ResponseHandler.Single(output))) + val message = output.await() return message } -- 2.51.2 From 09f1f579b00cfa7c86493f154fb5bc5f172ca426 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Sat, 4 Jul 2026 18:35:36 +0200 Subject: [PATCH 02/12] test(driver-multiplatform-wire): Launch the client in the foreground scope --- .../src/commonTest/kotlin/Utils.kt | 54 +++++++++++-------- 1 file changed, 32 insertions(+), 22 deletions(-) diff --git a/driver-multiplatform-wire/src/commonTest/kotlin/Utils.kt b/driver-multiplatform-wire/src/commonTest/kotlin/Utils.kt index c66ec1c4..84490fc0 100644 --- a/driver-multiplatform-wire/src/commonTest/kotlin/Utils.kt +++ b/driver-multiplatform-wire/src/commonTest/kotlin/Utils.kt @@ -23,37 +23,45 @@ import io.ktor.network.sockets.* import kotlinx.coroutines.* import opensavvy.ktmongo.dsl.LowLevelApi import opensavvy.prepared.runner.testballoon.preparedSuite -import opensavvy.prepared.suite.backgroundScope import opensavvy.prepared.suite.cleanUp +import opensavvy.prepared.suite.foregroundScope import opensavvy.prepared.suite.prepared import opensavvy.prepared.suite.shared val mongoAddress by shared { - val job = Job() - currentCoroutineContext().job.invokeOnCompletion { e -> job.cancel("Finished searching for the address. (ended with: $e)") } - val manager = SelectorManager(Dispatchers.Default + job + CoroutineName("FindMongoAddressManager")) + coroutineScope { + val manager = CompletableDeferred() + val managerJob = launch(Dispatchers.Default + CoroutineName("FindMongoAddressManager")) { + SelectorManager(currentCoroutineContext()) + .also { manager.complete(it) } + } - val candidateHostNames = listOf("localhost", "mongo") + val candidateHostNames = listOf("localhost", "mongo") - println("Searching for the address of the running MongoDB instance…") - for (hostName in candidateHostNames) { - val address = InetSocketAddress(hostName, 27017) + println(" Searching for the address of the running MongoDB instance…") + for (hostName in candidateHostNames) { + val address = InetSocketAddress(hostName, 27017) - try { - println("» Trying $address…") - aSocket(manager).tcp().connect(address.hostname, address.port) { - socketTimeout = 100 - }.use { - println(" Connected successfully!") + try { + println("» Trying $address…") + aSocket(manager.await()).tcp().connect(address.hostname, address.port) { + socketTimeout = 100 + }.use { + println(" Connected successfully!") + } + manager.await().close() + managerJob.cancel("We found a valid address.") + println(" Closed the socket.") + return@coroutineScope address + } catch (e: Exception) { + println(" Could not connect to MongoDB on socket $address • $e") } - println(" Closed the socket.") - return@shared address - } catch (e: Exception) { - println(" Could not connect to MongoDB on socket $address • $e") } - } - error("Could not find on which port MongoDB is running.") + manager.await().close() + error("Could not find on which port MongoDB is running.") + // No need to cancel the manager, it will be killed by the exception + } } val MongoWireClient by prepared { @@ -62,7 +70,7 @@ val MongoWireClient by prepared { MongoWireClient( hostName = socket.hostname, port = socket.port, - coroutineContext = backgroundScope.coroutineContext + coroutineContext = foregroundScope.coroutineContext ).also { cleanUp("Close $it") { it.close() @@ -73,7 +81,7 @@ val MongoWireClient by prepared { val SocketTest by preparedSuite { test("Create and close a socket") { println("Creating socket manager…") - val manager = SelectorManager(backgroundScope.coroutineContext + Dispatchers.Default + CoroutineName("FindMongoAddressManager")) + val manager = SelectorManager(currentCoroutineContext() + Dispatchers.Default + CoroutineName("FindMongoAddressManager")) println("Creating address…") val address = InetSocketAddress("google.com", 80) @@ -83,5 +91,7 @@ val SocketTest by preparedSuite { println("Connected!") } println("Disconnected!") + + manager.close() } } -- 2.51.2 From 3ab9e2a2da60c3d90b65d082370ad1736cabc0b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Sun, 5 Jul 2026 13:00:49 +0200 Subject: [PATCH 03/12] fix(driver-multiplatform-wire): Ensure the client properly closes --- .../src/commonMain/kotlin/MongoWireClient.kt | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt index b380c88c..625d3323 100644 --- a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt @@ -67,9 +67,11 @@ private class SocketWireClient( private val socket: Socket, private val selectorManager: SelectorManager, private val factory: BsonFactory, - coroutineScope: CoroutineScope, + coroutineScope: CoroutineScope, // Should contain a Job dedicated to this client ) : MongoWireClient { + private val actorsJob = coroutineScope.coroutineContext.job + private sealed class ResponseHandler { data class Single(val result: CompletableDeferred) : ResponseHandler() data class Multiple(val result: SendChannel) : ResponseHandler() @@ -108,6 +110,9 @@ private class SocketWireClient( } init { + // Ensure that no resources can leak + actorsJob.invokeOnCompletion { close() } + /** * When the [sendActor] has sent a message into the socket, it adds a message in here. * @@ -413,6 +418,7 @@ private class SocketWireClient( } override fun close() { + actorsJob.cancel("${this::class}.close() has been called") socket.close() selectorManager.close() } @@ -427,7 +433,9 @@ suspend fun MongoWireClient( factory: BsonFactory = BsonFactory(), coroutineContext: CoroutineContext, ): MongoWireClient { - val selectorManager = SelectorManager(coroutineContext + Dispatchers.Default + CoroutineName("ktmongo-socket")) + val innerJob = Job(coroutineContext.job) + + val selectorManager = SelectorManager(coroutineContext + innerJob + Dispatchers.Default + CoroutineName("ktmongo-socket")) val socket = aSocket(selectorManager).tcp().connect(hostName, port) { socketTimeout = 1000 keepAlive = true @@ -437,6 +445,6 @@ suspend fun MongoWireClient( socket = socket, selectorManager = selectorManager, factory = factory, - coroutineScope = CoroutineScope(coroutineContext + CoroutineName("ktmongo-client")) + coroutineScope = CoroutineScope(coroutineContext + innerJob + CoroutineName("ktmongo-client")) ) } -- 2.51.2 From d91b82b6785ca82afdcd15fb019d8914d750aed9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Tue, 7 Jul 2026 20:06:46 +0200 Subject: [PATCH 04/12] refactor(driver-multiplatform-wire): Remove the dummy operations Since the module is becoming useful, it's a good time to delete these prototypal hardcoded operations. This commit also changes the collections used by the tests to ensure no two tests use the same collection name, as a write just after a drop strongly lowers the performance of MongoDB. --- .../src/commonMain/kotlin/Message.kt | 63 ----------- .../src/commonTest/kotlin/1_Connect.kt | 105 ++++++++++++++++-- 2 files changed, 97 insertions(+), 71 deletions(-) diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/Message.kt b/driver-multiplatform-wire/src/commonMain/kotlin/Message.kt index 5e959d95..3e01d474 100644 --- a/driver-multiplatform-wire/src/commonMain/kotlin/Message.kt +++ b/driver-multiplatform-wire/src/commonMain/kotlin/Message.kt @@ -16,16 +16,6 @@ package opensavvy.ktmongo.multiplatform.wire -import kotlinx.serialization.Serializable -import opensavvy.ktmongo.bson.encode -import opensavvy.ktmongo.bson.multiplatform.BsonDocument -import opensavvy.ktmongo.bson.multiplatform.BsonFactory -import opensavvy.ktmongo.bson.types.ObjectId -import opensavvy.ktmongo.bson.types.ObjectIdGenerator -import opensavvy.ktmongo.dsl.LowLevelApi -import kotlin.concurrent.atomics.ExperimentalAtomicApi -import kotlin.time.ExperimentalTime - sealed interface Message { /** @@ -57,57 +47,4 @@ sealed interface Message { .joinToString(prefix = "OpMsg(", postfix = ")") } - companion object { - - @OptIn(LowLevelApi::class) - fun Find(): OpMsg = OpMsg( - MessageSection.Body( - eager( - BsonFactory().buildDocument { - writeString("find", "test-basic") - writeDocument("filter") {} - writeString("\$db", "test-basic") - } - ) - ) - ) - - @Serializable - data class DataTest @OptIn(ExperimentalTime::class) constructor( - val _id: ObjectId, - val name: String, - val age: Int, - ) - - @OptIn(LowLevelApi::class, ExperimentalTime::class, ExperimentalAtomicApi::class) - fun Insert(): OpMsg = OpMsg( - MessageSection.Body( - eager( - BsonFactory().buildDocument { - writeString("insert", "test-basic") - writeBoolean("ordered", true) - writeString("\$db", "test-basic") - } - ) - ), - MessageSection.DocumentSequence( - id = "documents", - listOf( - eager(BsonFactory().encode(DataTest(ObjectIdGenerator.Default().newId(), "Bob", 18)) as BsonDocument) - ) - ) - ) - - @OptIn(LowLevelApi::class) - fun Drop(): OpMsg = OpMsg( - MessageSection.Body( - eager( - BsonFactory().buildDocument { - writeString("drop", "test-basic") - writeString("\$db", "test-basic") - } - ) - ) - ) - } } diff --git a/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt b/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt index 208cf28c..fba1c4a5 100644 --- a/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt +++ b/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt @@ -14,14 +14,31 @@ * limitations under the License. */ -@file:OptIn(LowLevelApi::class, ExperimentalBsonPathApi::class) +@file:OptIn(LowLevelApi::class, ExperimentalBsonPathApi::class, ExperimentalAtomicApi::class) package opensavvy.ktmongo.multiplatform.wire +import kotlinx.serialization.Serializable import opensavvy.ktmongo.bson.ExperimentalBsonPathApi +import opensavvy.ktmongo.bson.encode +import opensavvy.ktmongo.bson.multiplatform.BsonDocument +import opensavvy.ktmongo.bson.multiplatform.BsonFactory import opensavvy.ktmongo.bson.selectFirst +import opensavvy.ktmongo.bson.types.ObjectId +import opensavvy.ktmongo.bson.types.ObjectIdGenerator.Default import opensavvy.ktmongo.dsl.LowLevelApi +import opensavvy.ktmongo.multiplatform.wire.Message.OpMsg +import opensavvy.ktmongo.multiplatform.wire.MessageSection.Body +import opensavvy.ktmongo.multiplatform.wire.MessageSection.DocumentSequence import opensavvy.prepared.runner.testballoon.preparedSuite +import kotlin.concurrent.atomics.ExperimentalAtomicApi + +@Serializable +private data class DataTest( + val _id: ObjectId, + val name: String, + val age: Int, +) val ConnectTest by preparedSuite { @@ -37,20 +54,48 @@ val ConnectTest by preparedSuite { test("Send a find on a collection that doesn't exist") { val client = MongoWireClient() - val output = client.send(Message.Find()) + val output = client.send( + OpMsg( + Body( + eager( + BsonFactory().buildDocument { + writeString("find", "test-basic-1") + writeDocument("filter") {} + writeString("\$db", "test-basic") + } + ) + ) + ) + ) println("Awaiting response…") val response = output.receive() check(response is Message.OpMsg) check(response.body.document["ok"]?.decodeDouble() == 1.0) - check(response.body.document["cursor"]?.decodeDocument()?.get("ns")?.decodeString() == "test-basic.test-basic") + check(response.body.document["cursor"]?.decodeDocument()?.get("ns")?.decodeString() == "test-basic.test-basic-1") } test("Insert an element") { val client = MongoWireClient() - val output = client.send(Message.Insert()) + val output = client.send(OpMsg( + Body( + eager( + BsonFactory().buildDocument { + writeString("insert", "test-basic-2") + writeBoolean("ordered", true) + writeString("\$db", "test-basic") + } + ) + ), + DocumentSequence( + id = "documents", + listOf( + eager(BsonFactory().encode(DataTest(Default().newId(), "Bob", 18)) as BsonDocument) + ) + ) + )) println("Awaiting response…") val response = output.receive() @@ -63,7 +108,16 @@ val ConnectTest by preparedSuite { test("Drop a collection") { val client = MongoWireClient() - val output = client.send(Message.Drop()) + val output = client.send(OpMsg( + Body( + eager( + BsonFactory().buildDocument { + writeString("drop", "test-basic-2") + writeString("\$db", "test-basic") + } + ) + ) + )) println("Awaiting response…") val response = output.receive() @@ -75,8 +129,34 @@ val ConnectTest by preparedSuite { test("Find an element that was just inserted") { val client = MongoWireClient() - val insertOutput = client.send(Message.Insert()) - val findOutput = client.send(Message.Find()) + val insertOutput = client.send(OpMsg( + Body( + eager( + BsonFactory().buildDocument { + writeString("insert", "test-basic-3") + writeBoolean("ordered", true) + writeString("\$db", "test-basic") + } + ) + ), + DocumentSequence( + id = "documents", + listOf( + eager(BsonFactory().encode(DataTest(Default().newId(), "Bob", 18)) as BsonDocument) + ) + ) + )) + val findOutput = client.send(OpMsg( + Body( + eager( + BsonFactory().buildDocument { + writeString("find", "test-basic-3") + writeDocument("filter") {} + writeString("\$db", "test-basic") + } + ) + ) + )) println("Awaiting response…") val insertResponse = insertOutput.receive() @@ -89,6 +169,15 @@ val ConnectTest by preparedSuite { check(findResponse.body.document["ok"]?.decodeDouble() == 1.0) check(findResponse.body.document.selectFirst("$.cursor.firstBatch[0].name") == "Bob") - val _ = client.send(Message.Drop()) + val _ = client.send(OpMsg( + Body( + eager( + BsonFactory().buildDocument { + writeString("drop", "test-basic-3") + writeString("\$db", "test-basic") + } + ) + ) + )) } } -- 2.51.2 From 77a2f7683fc003e1d5ba7c240a19be5307e746dd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Tue, 7 Jul 2026 20:30:38 +0200 Subject: [PATCH 05/12] test(driver-multiplatform): Launch the client in the foreground scope The background scope auto-cancels everything. By using the foreground scope, it will be more visible if something is not properly cancelled. --- .../src/commonTest/kotlin/utils/FindHostname.kt | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt b/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt index ac427a7e..59d20925 100644 --- a/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt +++ b/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt @@ -21,7 +21,8 @@ import kotlinx.coroutines.cancel import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.job import opensavvy.ktmongo.multiplatform.MongoClient -import opensavvy.prepared.suite.backgroundScope +import opensavvy.prepared.suite.cleanUp +import opensavvy.prepared.suite.foregroundScope import opensavvy.prepared.suite.prepared import opensavvy.prepared.suite.shared import kotlin.concurrent.atomics.ExperimentalAtomicApi @@ -55,9 +56,15 @@ private val mongoAddress by shared { val MongoClient by prepared { val address = mongoAddress() - MongoClient( + val client = MongoClient( hostname = address, port = 27017, - coroutineContext = backgroundScope.coroutineContext, + coroutineContext = foregroundScope.coroutineContext, ) + + cleanUp("MongoClient") { + client.close() + } + + client } -- 2.51.2 From 299c172026e32d40330be2b6640194119ccc369c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Tue, 7 Jul 2026 20:31:22 +0200 Subject: [PATCH 06/12] feat(driver-multiplatform): Add MongoClient.close() --- driver-multiplatform/src/commonMain/kotlin/MongoClient.kt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/driver-multiplatform/src/commonMain/kotlin/MongoClient.kt b/driver-multiplatform/src/commonMain/kotlin/MongoClient.kt index fffa4eb6..4f23f534 100644 --- a/driver-multiplatform/src/commonMain/kotlin/MongoClient.kt +++ b/driver-multiplatform/src/commonMain/kotlin/MongoClient.kt @@ -70,7 +70,7 @@ class MongoClient internal constructor( internal val wire: MongoWireClient, val factory: BsonFactory, val context: BsonContext, -) { +) : AutoCloseable { /** * Creates a [MongoDatabase] object. @@ -83,6 +83,9 @@ class MongoClient internal constructor( fun database(name: String): MongoDatabase = MongoDatabaseImpl(this, name) + override fun close() { + wire.close() + } } /** -- 2.51.2 From 156088255c2d077ef56f888bfa85104b79225cb0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Tue, 7 Jul 2026 20:31:49 +0200 Subject: [PATCH 07/12] test(driver-multiplatform): Ensure each test runs in a different collection --- .../src/commonTest/kotlin/commands/MultiplatformInsert.kt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/driver-multiplatform/src/commonTest/kotlin/commands/MultiplatformInsert.kt b/driver-multiplatform/src/commonTest/kotlin/commands/MultiplatformInsert.kt index 0507ba1a..00f59acc 100644 --- a/driver-multiplatform/src/commonTest/kotlin/commands/MultiplatformInsert.kt +++ b/driver-multiplatform/src/commonTest/kotlin/commands/MultiplatformInsert.kt @@ -20,6 +20,7 @@ import kotlinx.serialization.Serializable import opensavvy.ktmongo.bson.types.ObjectId import opensavvy.ktmongo.multiplatform.utils.MongoClient import opensavvy.prepared.runner.testballoon.preparedSuite +import opensavvy.prepared.suite.random.randomInt @Serializable private data class User( @@ -30,10 +31,12 @@ private data class User( val MultiplatformInsert by preparedSuite { + val collectionPostfix by randomInt(0, Int.MAX_VALUE) + test("Simple insert") { val client = MongoClient() val database = client.database("ktmongo-test-1") - val collection = database.collection("users") + val collection = database.collection("users-${collectionPostfix()}") collection.insertOne( User( -- 2.51.2 From efdfefe28c59c5b6d61e100eff471f3ff853f155 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Tue, 7 Jul 2026 20:42:23 +0200 Subject: [PATCH 08/12] refactor(driver-multiplatform): Simplify coroutines code --- .../src/commonTest/kotlin/utils/FindHostname.kt | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt b/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt index 59d20925..1bd176cf 100644 --- a/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt +++ b/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt @@ -16,10 +16,7 @@ package opensavvy.ktmongo.multiplatform.utils -import kotlinx.coroutines.Job -import kotlinx.coroutines.cancel -import kotlinx.coroutines.coroutineScope -import kotlinx.coroutines.job +import kotlinx.coroutines.currentCoroutineContext import opensavvy.ktmongo.multiplatform.MongoClient import opensavvy.prepared.suite.cleanUp import opensavvy.prepared.suite.foregroundScope @@ -33,11 +30,7 @@ private suspend fun tryConnect( ): Boolean { try { println("KtMongo • Attempting to connect to $hostname") - val _ = coroutineScope { - val job = Job() - this.coroutineContext.job.invokeOnCompletion { e -> job.cancel("Finished searching for the address. (ended with: $e)") } - MongoClient(hostname = hostname, coroutineContext = coroutineContext + job) - } + MongoClient(hostname = hostname, coroutineContext = currentCoroutineContext()).close() return true } catch (e: Throwable) { println("KtMongo • Could not connect to $hostname: $e") -- 2.51.2 From 47fa394a5c738ddf980cd557f2eb7159c2337703 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Thu, 9 Jul 2026 15:22:32 +0200 Subject: [PATCH 09/12] test: Increase the timeout for tests that use a real connection The database becomes overloaded by our test suite, which makes the test suite flaky. By using a large timeout, we hope that the database can recover from the overload and the tests can proceed normally. --- driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt | 4 +++- driver-multiplatform/src/commonTest/kotlin/Connection.kt | 4 +++- .../src/commonTest/kotlin/commands/MultiplatformInsert.kt | 4 +++- test-legacy/src/commonTest/kotlin/AggregationTests.kt | 4 ++-- test-legacy/src/commonTest/kotlin/ArraysTest.kt | 4 ++-- test-legacy/src/commonTest/kotlin/BasicReadWriteTest.kt | 4 ++-- test-legacy/src/commonTest/kotlin/FilteredCollectionTest.kt | 4 ++-- test-legacy/src/commonTest/kotlin/MapsTest.kt | 4 ++-- test-legacy/src/commonTest/kotlin/SortTest.kt | 4 ++-- test/src/commonMain/kotlin/Entrypoint.kt | 4 +++- 10 files changed, 24 insertions(+), 16 deletions(-) diff --git a/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt b/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt index fba1c4a5..262d20c6 100644 --- a/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt +++ b/driver-multiplatform-wire/src/commonTest/kotlin/1_Connect.kt @@ -31,7 +31,9 @@ import opensavvy.ktmongo.multiplatform.wire.Message.OpMsg import opensavvy.ktmongo.multiplatform.wire.MessageSection.Body import opensavvy.ktmongo.multiplatform.wire.MessageSection.DocumentSequence import opensavvy.prepared.runner.testballoon.preparedSuite +import opensavvy.prepared.suite.config.CoroutineTimeout import kotlin.concurrent.atomics.ExperimentalAtomicApi +import kotlin.time.Duration.Companion.minutes @Serializable private data class DataTest( @@ -40,7 +42,7 @@ private data class DataTest( val age: Int, ) -val ConnectTest by preparedSuite { +val ConnectTest by preparedSuite(preparedConfig = CoroutineTimeout(15.minutes)) { test("Connect to the database") { val client = MongoWireClient() diff --git a/driver-multiplatform/src/commonTest/kotlin/Connection.kt b/driver-multiplatform/src/commonTest/kotlin/Connection.kt index 9180d252..62026cde 100644 --- a/driver-multiplatform/src/commonTest/kotlin/Connection.kt +++ b/driver-multiplatform/src/commonTest/kotlin/Connection.kt @@ -18,8 +18,10 @@ package opensavvy.ktmongo.multiplatform import opensavvy.ktmongo.multiplatform.utils.MongoClient import opensavvy.prepared.runner.testballoon.preparedSuite +import opensavvy.prepared.suite.config.CoroutineTimeout +import kotlin.time.Duration.Companion.minutes -val MultiplatformConnection by preparedSuite { +val MultiplatformConnection by preparedSuite(preparedConfig = CoroutineTimeout(15.minutes)) { test("Connect") { val client = MongoClient() diff --git a/driver-multiplatform/src/commonTest/kotlin/commands/MultiplatformInsert.kt b/driver-multiplatform/src/commonTest/kotlin/commands/MultiplatformInsert.kt index 00f59acc..15f62d3b 100644 --- a/driver-multiplatform/src/commonTest/kotlin/commands/MultiplatformInsert.kt +++ b/driver-multiplatform/src/commonTest/kotlin/commands/MultiplatformInsert.kt @@ -20,7 +20,9 @@ import kotlinx.serialization.Serializable import opensavvy.ktmongo.bson.types.ObjectId import opensavvy.ktmongo.multiplatform.utils.MongoClient import opensavvy.prepared.runner.testballoon.preparedSuite +import opensavvy.prepared.suite.config.CoroutineTimeout import opensavvy.prepared.suite.random.randomInt +import kotlin.time.Duration.Companion.minutes @Serializable private data class User( @@ -29,7 +31,7 @@ private data class User( val age: Int, ) -val MultiplatformInsert by preparedSuite { +val MultiplatformInsert by preparedSuite(preparedConfig = CoroutineTimeout(15.minutes)) { val collectionPostfix by randomInt(0, Int.MAX_VALUE) diff --git a/test-legacy/src/commonTest/kotlin/AggregationTests.kt b/test-legacy/src/commonTest/kotlin/AggregationTests.kt index 50a0ca4e..ceb347f9 100644 --- a/test-legacy/src/commonTest/kotlin/AggregationTests.kt +++ b/test-legacy/src/commonTest/kotlin/AggregationTests.kt @@ -27,9 +27,9 @@ import opensavvy.ktmongo.dsl.LowLevelApi import opensavvy.ktmongo.test.testCollection import opensavvy.prepared.runner.testballoon.preparedSuite import opensavvy.prepared.suite.config.CoroutineTimeout -import kotlin.time.Duration.Companion.seconds +import kotlin.time.Duration.Companion.minutes -val AggregationTests by preparedSuite(preparedConfig = CoroutineTimeout(30.seconds)) { +val AggregationTests by preparedSuite(preparedConfig = CoroutineTimeout(15.minutes)) { @Serializable data class Song( val creationDate: Int = 0, // default values are required for dumping the collection diff --git a/test-legacy/src/commonTest/kotlin/ArraysTest.kt b/test-legacy/src/commonTest/kotlin/ArraysTest.kt index 79b5414b..a8402f03 100644 --- a/test-legacy/src/commonTest/kotlin/ArraysTest.kt +++ b/test-legacy/src/commonTest/kotlin/ArraysTest.kt @@ -20,7 +20,7 @@ import kotlinx.serialization.Serializable import opensavvy.ktmongo.test.testCollection import opensavvy.prepared.runner.testballoon.preparedSuite import opensavvy.prepared.suite.config.CoroutineTimeout -import kotlin.time.Duration.Companion.seconds +import kotlin.time.Duration.Companion.minutes @Serializable data class ArrayUser( @@ -29,7 +29,7 @@ data class ArrayUser( val friends: List = emptyList(), ) -val ArraysTest by preparedSuite(preparedConfig = CoroutineTimeout(30.seconds)) { +val ArraysTest by preparedSuite(preparedConfig = CoroutineTimeout(15.minutes)) { val users by testCollection("arrays") suite("Not empty array") { diff --git a/test-legacy/src/commonTest/kotlin/BasicReadWriteTest.kt b/test-legacy/src/commonTest/kotlin/BasicReadWriteTest.kt index 54ceaadb..cd15b59e 100644 --- a/test-legacy/src/commonTest/kotlin/BasicReadWriteTest.kt +++ b/test-legacy/src/commonTest/kotlin/BasicReadWriteTest.kt @@ -23,7 +23,7 @@ import opensavvy.ktmongo.test.testCollection import opensavvy.prepared.runner.testballoon.preparedSuite import opensavvy.prepared.suite.config.CoroutineTimeout import kotlin.concurrent.atomics.ExperimentalAtomicApi -import kotlin.time.Duration.Companion.seconds +import kotlin.time.Duration.Companion.minutes import kotlin.time.ExperimentalTime data class User( @@ -32,7 +32,7 @@ data class User( val age: Int, ) -val BasicReadWriteTest by preparedSuite(preparedConfig = CoroutineTimeout(30.seconds)) { +val BasicReadWriteTest by preparedSuite(preparedConfig = CoroutineTimeout(15.minutes)) { val id1 = ObjectId("69908384e10bb0a5f7d17c5b") val id2 = ObjectId("699083ade6e89e315640258c") val id3 = ObjectId("699083c9e493dd6c2e60c664") diff --git a/test-legacy/src/commonTest/kotlin/FilteredCollectionTest.kt b/test-legacy/src/commonTest/kotlin/FilteredCollectionTest.kt index 77a15caf..6a68b016 100644 --- a/test-legacy/src/commonTest/kotlin/FilteredCollectionTest.kt +++ b/test-legacy/src/commonTest/kotlin/FilteredCollectionTest.kt @@ -22,9 +22,9 @@ import opensavvy.ktmongo.test.testCollection import opensavvy.prepared.runner.testballoon.preparedSuite import opensavvy.prepared.suite.config.CoroutineTimeout import opensavvy.prepared.suite.prepared -import kotlin.time.Duration.Companion.seconds +import kotlin.time.Duration.Companion.minutes -val FilteredCollectionTest by preparedSuite(preparedConfig = CoroutineTimeout(30.seconds)) { +val FilteredCollectionTest by preparedSuite(preparedConfig = CoroutineTimeout(15.minutes)) { @Serializable data class User( val name: String = "MISSING", diff --git a/test-legacy/src/commonTest/kotlin/MapsTest.kt b/test-legacy/src/commonTest/kotlin/MapsTest.kt index eec0cc02..a4b310af 100644 --- a/test-legacy/src/commonTest/kotlin/MapsTest.kt +++ b/test-legacy/src/commonTest/kotlin/MapsTest.kt @@ -20,7 +20,7 @@ import kotlinx.serialization.Serializable import opensavvy.ktmongo.test.testCollection import opensavvy.prepared.runner.testballoon.preparedSuite import opensavvy.prepared.suite.config.CoroutineTimeout -import kotlin.time.Duration.Companion.seconds +import kotlin.time.Duration.Companion.minutes @Serializable data class MapsUser( @@ -29,7 +29,7 @@ data class MapsUser( val friends: Map = emptyMap(), ) -val MapsTest by preparedSuite(preparedConfig = CoroutineTimeout(30.seconds)) { +val MapsTest by preparedSuite(preparedConfig = CoroutineTimeout(15.minutes)) { val users by testCollection("maps") suite("Not empty map") { diff --git a/test-legacy/src/commonTest/kotlin/SortTest.kt b/test-legacy/src/commonTest/kotlin/SortTest.kt index 8f9d5c99..140c39fb 100644 --- a/test-legacy/src/commonTest/kotlin/SortTest.kt +++ b/test-legacy/src/commonTest/kotlin/SortTest.kt @@ -23,7 +23,7 @@ import opensavvy.ktmongo.test.testCollection import opensavvy.prepared.runner.testballoon.preparedSuite import opensavvy.prepared.suite.config.CoroutineTimeout import kotlin.concurrent.atomics.ExperimentalAtomicApi -import kotlin.time.Duration.Companion.seconds +import kotlin.time.Duration.Companion.minutes import kotlin.time.ExperimentalTime import kotlin.time.Instant @@ -42,7 +42,7 @@ data class SortUser( private fun Instant.truncateToMilliseconds(): Instant = Instant.fromEpochMilliseconds(this.toEpochMilliseconds()) -val SortTest by preparedSuite(preparedConfig = CoroutineTimeout(30.seconds)) { +val SortTest by preparedSuite(preparedConfig = CoroutineTimeout(15.minutes)) { val users by testCollection("basic-sorts") test("Sort by date") { diff --git a/test/src/commonMain/kotlin/Entrypoint.kt b/test/src/commonMain/kotlin/Entrypoint.kt index d3dae491..834be57e 100644 --- a/test/src/commonMain/kotlin/Entrypoint.kt +++ b/test/src/commonMain/kotlin/Entrypoint.kt @@ -22,12 +22,14 @@ import kotlinx.coroutines.ensureActive import opensavvy.ktmongo.api.MongoClient import opensavvy.ktmongo.tests.api.operations.* import opensavvy.prepared.suite.* +import opensavvy.prepared.suite.config.CoroutineTimeout import kotlin.coroutines.CoroutineContext +import kotlin.time.Duration.Companion.minutes fun SuiteDsl.verifyClient( name: String, createClient: suspend (connectionString: String, coroutineContext: CoroutineContext) -> MongoClient, -) = suite(name) { +) = suite(name, CoroutineTimeout(15.minutes)) { suspend fun verifyClientConnected(client: MongoClient): Boolean = try { val count = client.use { -- 2.51.2 From 7c00aef9b0ea6803979c26043d5c2b8d7d1c2d3d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Fri, 10 Jul 2026 20:44:51 +0200 Subject: [PATCH 10/12] refactor(driver-multiplatform-wire): Log the timing of the requests --- .../src/commonMain/kotlin/MongoWireClient.kt | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt index 625d3323..4286bf10 100644 --- a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt @@ -31,6 +31,8 @@ import opensavvy.ktmongo.bson.multiplatform.BsonDocument import opensavvy.ktmongo.bson.multiplatform.BsonFactory import opensavvy.ktmongo.dsl.LowLevelApi import kotlin.coroutines.CoroutineContext +import kotlin.time.DurationUnit +import kotlin.time.TimeSource @LowLevelApi interface MongoWireClient : AutoCloseable { @@ -105,11 +107,16 @@ private class SocketWireClient( */ private val requestChannel = Channel(Channel.RENDEZVOUS) + // Helps debugging time-sensitive operations for now. Will need to be removed when stabilizing, and be replaced by a proper observability framework. + private val start = TimeSource.Monotonic.markNow() + private fun log(message: String) { - println("KtMongo • $message") + println("» KtMongo +${start.elapsedNow().toString(DurationUnit.MILLISECONDS, decimals = 0)} • $message") } init { + log("Creating client for socket ${socket.remoteAddress}") + // Ensure that no resources can leak actorsJob.invokeOnCompletion { close() } -- 2.51.2 From 8f77466ed01788933f670b9a4e0c51206e4ce112 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Sat, 11 Jul 2026 09:28:25 +0200 Subject: [PATCH 11/12] feat(driver-multiplatform-wire): Remove the socket timeout --- .../src/commonMain/kotlin/MongoWireClient.kt | 1 - 1 file changed, 1 deletion(-) diff --git a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt index 4286bf10..adaa6410 100644 --- a/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt +++ b/driver-multiplatform-wire/src/commonMain/kotlin/MongoWireClient.kt @@ -444,7 +444,6 @@ suspend fun MongoWireClient( val selectorManager = SelectorManager(coroutineContext + innerJob + Dispatchers.Default + CoroutineName("ktmongo-socket")) val socket = aSocket(selectorManager).tcp().connect(hostName, port) { - socketTimeout = 1000 keepAlive = true } -- 2.51.2 From 85765e45403ff59318f8ab29686010c1e5aa236c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ivan=20=E2=80=9CCLOVIS=E2=80=9D=20Canet?= Date: Sat, 18 Jul 2026 09:33:23 +0200 Subject: [PATCH 12/12] test(driver-multiplatform): Isolate the client into its own coroutine scope to ensure it is completely torn down before continuing --- .../src/commonTest/kotlin/utils/FindHostname.kt | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt b/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt index 1bd176cf..18cc607e 100644 --- a/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt +++ b/driver-multiplatform/src/commonTest/kotlin/utils/FindHostname.kt @@ -16,7 +16,7 @@ package opensavvy.ktmongo.multiplatform.utils -import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.coroutineScope import opensavvy.ktmongo.multiplatform.MongoClient import opensavvy.prepared.suite.cleanUp import opensavvy.prepared.suite.foregroundScope @@ -29,11 +29,13 @@ private suspend fun tryConnect( hostname: String, ): Boolean { try { - println("KtMongo • Attempting to connect to $hostname") - MongoClient(hostname = hostname, coroutineContext = currentCoroutineContext()).close() + println(" Attempting to connect to $hostname") + coroutineScope { + MongoClient(hostname = hostname, coroutineContext = coroutineContext).close() + } return true } catch (e: Throwable) { - println("KtMongo • Could not connect to $hostname: $e") + println(" Could not connect to $hostname: ${e.stackTraceToString()}") return false } } -- 2.51.2