Fixes for failed CI

Mr3zee · Mr3zee · commit df7b1ce46ec0 · 2025-06-13T11:20:00.000+02:00
diff --git a/compiler-plugin/compiler-plugin-k2/src/main/pre_2_0_10/kotlinx/rpc/codegen/checkers/FirRpcCheckersVS.kt b/compiler-plugin/compiler-plugin-k2/src/main/pre_2_0_10/kotlinx/rpc/codegen/checkers/FirRpcCheckersVS.kt
@@ -65,8 +65,8 @@ class FirRpcServiceDeclarationCheckerVS(
     }
 }
 
-class FirRpcStrictModeClassCheckerVS(private val ctx: FirCheckersContext) : FirRegularClassChecker(MppCheckerKind.Common) {
+class FirRpcStrictModeClassCheckerVS : FirRegularClassChecker(MppCheckerKind.Common) {
     override fun check(declaration: FirRegularClass, context: CheckerContext, reporter: DiagnosticReporter) {
-        FirRpcStrictModeClassChecker.check(ctx, declaration, context, reporter)
+        FirRpcStrictModeClassChecker.check(declaration, context, reporter)
     }
 }
diff --git a/compiler-plugin/compiler-plugin-k2/src/main/pre_2_0_21/kotlinx/rpc/codegen/checkers/FirRpcCheckersVS.kt b/compiler-plugin/compiler-plugin-k2/src/main/pre_2_0_21/kotlinx/rpc/codegen/checkers/FirRpcCheckersVS.kt
@@ -65,8 +65,8 @@ class FirRpcServiceDeclarationCheckerVS(
     }
 }
 
-class FirRpcStrictModeClassCheckerVS(private val ctx: FirCheckersContext) : FirRegularClassChecker(MppCheckerKind.Common) {
+class FirRpcStrictModeClassCheckerVS : FirRegularClassChecker(MppCheckerKind.Common) {
     override fun check(declaration: FirRegularClass, context: CheckerContext, reporter: DiagnosticReporter) {
-        FirRpcStrictModeClassChecker.check(ctx, declaration, context, reporter)
+        FirRpcStrictModeClassChecker.check(declaration, context, reporter)
     }
 }
diff --git a/compiler-plugin/compiler-plugin-k2/src/main/v_2_2/kotlinx/rpc/codegen/checkers/FirRpcCheckersVS.kt b/compiler-plugin/compiler-plugin-k2/src/main/v_2_2/kotlinx/rpc/codegen/checkers/FirRpcCheckersVS.kt
@@ -71,9 +71,9 @@ class FirRpcServiceDeclarationCheckerVS(
     }
 }
 
-class FirRpcStrictModeClassCheckerVS(private val ctx: FirCheckersContext) : FirRegularClassChecker(MppCheckerKind.Common) {
+class FirRpcStrictModeClassCheckerVS : FirRegularClassChecker(MppCheckerKind.Common) {
     context(context: CheckerContext, reporter: DiagnosticReporter)
     override fun check(declaration: FirRegularClass) {
-        FirRpcStrictModeClassChecker.check(ctx, declaration, context, reporter)
+        FirRpcStrictModeClassChecker.check(declaration, context, reporter)
     }
 }
diff --git a/compiler-plugin/compiler-plugin-k2/src/main/v_2_2_2/kotlinx/rpc/codegen/checkers/FirRpcCheckersVS.kt b/compiler-plugin/compiler-plugin-k2/src/main/v_2_2_2/kotlinx/rpc/codegen/checkers/FirRpcCheckersVS.kt
@@ -71,9 +71,9 @@ class FirRpcServiceDeclarationCheckerVS(
     }
 }
 
-class FirRpcStrictModeClassCheckerVS(private val ctx: FirCheckersContext) : FirRegularClassChecker(MppCheckerKind.Common) {
+class FirRpcStrictModeClassCheckerVS : FirRegularClassChecker(MppCheckerKind.Common) {
     context(context: CheckerContext, reporter: DiagnosticReporter)
     override fun check(declaration: FirRegularClass) {
-        FirRpcStrictModeClassChecker.check(ctx, declaration, context, reporter)
+        FirRpcStrictModeClassChecker.check(declaration, context, reporter)
     }
 }
diff --git a/krpc/krpc-client/api/krpc-client.api b/krpc/krpc-client/api/krpc-client.api
@@ -1,14 +1,17 @@
 public abstract class kotlinx/rpc/krpc/client/InitializedKrpcClient : kotlinx/rpc/krpc/client/KrpcClient {
 	public fun <init> (Lkotlinx/rpc/krpc/KrpcConfig$Client;Lkotlinx/rpc/krpc/KrpcTransport;)V
-	public final fun initializeConfig ()Lkotlinx/rpc/krpc/KrpcConfig$Client;
-	public final fun initializeTransport (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	protected final fun initializeConfig ()Lkotlinx/rpc/krpc/KrpcConfig$Client;
+	protected final fun initializeTransport (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
 public abstract class kotlinx/rpc/krpc/client/KrpcClient : kotlinx/rpc/RpcClient, kotlinx/rpc/krpc/internal/KrpcEndpoint {
 	public fun <init> ()V
+	public final fun awaitCompletion (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public final fun call (Lkotlinx/rpc/RpcCall;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public final fun callServerStreaming (Lkotlinx/rpc/RpcCall;)Lkotlinx/coroutines/flow/Flow;
-	public abstract fun initializeConfig ()Lkotlinx/rpc/krpc/KrpcConfig$Client;
-	public abstract fun initializeTransport (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public final fun close (Ljava/lang/String;)V
+	public static synthetic fun close$default (Lkotlinx/rpc/krpc/client/KrpcClient;Ljava/lang/String;ILjava/lang/Object;)V
+	protected abstract fun initializeConfig ()Lkotlinx/rpc/krpc/KrpcConfig$Client;
+	protected abstract fun initializeTransport (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
diff --git a/krpc/krpc-client/src/commonMain/kotlin/kotlinx/rpc/krpc/client/KrpcClient.kt b/krpc/krpc-client/src/commonMain/kotlin/kotlinx/rpc/krpc/client/KrpcClient.kt
@@ -84,13 +84,21 @@ public abstract class KrpcClient : RpcClient, KrpcEndpoint {
      * Close this client, removing all the services and stopping accepting messages.
      */
     public fun close(message: String? = null) {
+        if (!isTransportReady) {
+            return
+        }
+
         internalScope.cancel(message ?: "Client closed")
     }
 
     /**
      * Waits until the client is closed.
      */
     public suspend fun awaitCompletion() {
+        if (!isTransportReady) {
+            return
+        }
+
         internalScope.coroutineContext.job.join()
     }
 
@@ -133,7 +141,8 @@ public abstract class KrpcClient : RpcClient, KrpcEndpoint {
             sendCancellation(CancellationType.ENDPOINT, null, null, closeTransportAfterSending = true)
 
             @OptIn(DelicateCoroutinesApi::class)
-            GlobalScope.launch {
+            @Suppress("detekt.GlobalCoroutineUsage")
+            GlobalScope.launch(CoroutineName("client-request-channels-closing")) {
                 requestChannels.values.forEach { it.close(CancellationException("Client cancelled")) }
                 requestChannels.clear()
             }
diff --git a/krpc/krpc-ktor/krpc-ktor-client/src/commonMain/kotlin/kotlinx/rpc/krpc/ktor/client/KtorClientDsl.kt b/krpc/krpc-ktor/krpc-ktor-client/src/commonMain/kotlin/kotlinx/rpc/krpc/ktor/client/KtorClientDsl.kt
@@ -9,7 +9,6 @@ import io.ktor.client.plugins.*
 import io.ktor.client.plugins.websocket.*
 import io.ktor.client.request.*
 import io.ktor.util.*
-import kotlinx.rpc.RpcClient
 import kotlinx.rpc.krpc.KrpcConfigBuilder
 
 private val KrpcRequestConfigAttributeKey = AttributeKey<KrpcConfigBuilder.Client.() -> Unit>(
diff --git a/krpc/krpc-ktor/krpc-ktor-core/src/jvmTest/kotlin/kotlinx/rpc/krpc/ktor/KtorTransportTest.kt b/krpc/krpc-ktor/krpc-ktor-core/src/jvmTest/kotlin/kotlinx/rpc/krpc/ktor/KtorTransportTest.kt
@@ -152,6 +152,7 @@ class KtorTransportTest {
 
         val service = SlowServiceImpl()
 
+        @Suppress("detekt.GlobalCoroutineUsage")
         val serverJob = GlobalScope.launch(CoroutineName("server")) {
             withContext(newPool) {
                 val server = embeddedServer(
diff --git a/krpc/krpc-server/api/krpc-server.api b/krpc/krpc-server/api/krpc-server.api
@@ -1,5 +1,8 @@
 public abstract class kotlinx/rpc/krpc/server/KrpcServer : kotlinx/rpc/RpcServer, kotlinx/rpc/krpc/internal/KrpcEndpoint {
 	public fun <init> (Lkotlinx/rpc/krpc/KrpcConfig$Server;Lkotlinx/rpc/krpc/KrpcTransport;)V
+	public final fun awaitCompletion (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public final fun close (Ljava/lang/String;)V
+	public static synthetic fun close$default (Lkotlinx/rpc/krpc/server/KrpcServer;Ljava/lang/String;ILjava/lang/Object;)V
 	public fun deregisterService (Lkotlin/reflect/KClass;)V
 	public final fun registerService (Lkotlin/reflect/KClass;Lkotlin/jvm/functions/Function0;)V
 }
diff --git a/krpc/krpc-test/src/commonMain/kotlin/kotlinx/rpc/krpc/test/KrpcTransportTestBase.kt b/krpc/krpc-test/src/commonMain/kotlin/kotlinx/rpc/krpc/test/KrpcTransportTestBase.kt
@@ -391,8 +391,9 @@ abstract class KrpcTransportTestBase {
             return@runTest
         }
 
-        val inContinuation = Semaphore(1)
+        val inContinuation = Semaphore(permits = 1, acquiredPermits = 1)
         val running = AtomicTest()
+        val gate = CompletableDeferred<Unit>()
 
         // start a coroutine that block the thread after continuation
         val c1 = async(Dispatchers.Default) { // make a rpc call
@@ -405,13 +406,14 @@ abstract class KrpcTransportTestBase {
             inContinuation.release()
 
             // let's block the thread
+            gate.complete(Unit)
             while (running.atomic.value) {
                 // do nothing
             }
         }
 
         // wait, till the Rpc continuation thread is blocked
-        delay(100)
+        gate.await()
         assertTrue(inContinuation.tryAcquire())
 
         val c2 = async { // make a call
diff --git a/krpc/krpc-test/src/commonTest/kotlin/kotlinx/rpc/krpc/test/LocalTransport.kt b/krpc/krpc-test/src/commonTest/kotlin/kotlinx/rpc/krpc/test/LocalTransport.kt
@@ -4,6 +4,7 @@
 
 package kotlinx.rpc.krpc.test
 
+import kotlinx.atomicfu.atomic
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
@@ -12,6 +13,8 @@ import kotlinx.coroutines.job
 import kotlinx.rpc.krpc.KrpcTransport
 import kotlinx.rpc.krpc.KrpcTransportMessage
 import kotlin.coroutines.CoroutineContext
+import kotlin.time.Clock
+import kotlin.time.ExperimentalTime
 
 class LocalTransport(parentScope: CoroutineScope? = null) : CoroutineScope {
     override val coroutineContext = parentScope?.run { SupervisorJob(coroutineContext.job) }
@@ -20,10 +23,15 @@ class LocalTransport(parentScope: CoroutineScope? = null) : CoroutineScope {
     private val clientIncoming = Channel<KrpcTransportMessage>()
     private val serverIncoming = Channel<KrpcTransportMessage>()
 
+    val lastMessageSentOnClient = atomic(0L)
+    val lastMessageSentOnServer = atomic(0L)
+
     val client: KrpcTransport = object : KrpcTransport {
         override val coroutineContext: CoroutineContext = Job(this@LocalTransport.coroutineContext.job)
 
+        @OptIn(ExperimentalTime::class)
         override suspend fun send(message: KrpcTransportMessage) {
+            lastMessageSentOnClient.getAndSet(Clock.System.now().toEpochMilliseconds())
             serverIncoming.send(message)
         }
 
@@ -35,7 +43,9 @@ class LocalTransport(parentScope: CoroutineScope? = null) : CoroutineScope {
     val server: KrpcTransport = object : KrpcTransport {
         override val coroutineContext: CoroutineContext = Job(this@LocalTransport.coroutineContext)
 
+        @OptIn(ExperimentalTime::class)
         override suspend fun send(message: KrpcTransportMessage) {
+            lastMessageSentOnServer.getAndSet(Clock.System.now().toEpochMilliseconds())
             clientIncoming.send(message)
         }
 
diff --git a/krpc/krpc-test/src/commonTest/kotlin/kotlinx/rpc/krpc/test/cancellation/CancellationService.kt b/krpc/krpc-test/src/commonTest/kotlin/kotlinx/rpc/krpc/test/cancellation/CancellationService.kt
@@ -61,6 +61,7 @@ class CancellationServiceImpl : CancellationService {
 
     @OptIn(DelicateCoroutinesApi::class)
     override suspend fun outgoingStreamAsync(stream: Flow<Int>) {
+        @Suppress("detekt.GlobalCoroutineUsage")
         GlobalScope.launch {
             consume(stream)
         }
diff --git a/krpc/krpc-test/src/jvmTest/kotlin/kotlinx/rpc/krpc/test/api/WireSamplingTestScope.kt b/krpc/krpc-test/src/jvmTest/kotlin/kotlinx/rpc/krpc/test/api/WireSamplingTestScope.kt
@@ -43,9 +43,13 @@ import org.jetbrains.krpc.test.api.util.SamplingServiceImpl
 import java.nio.file.Files
 import java.nio.file.Path
 import kotlin.io.path.name
+import kotlin.reflect.full.callSuspend
+import kotlin.reflect.full.memberFunctions
 import kotlin.reflect.full.memberProperties
 import kotlin.reflect.jvm.isAccessible
+import kotlin.time.Clock
 import kotlin.time.Duration.Companion.seconds
+import kotlin.time.ExperimentalTime
 
 @Suppress("RedundantUnitReturnType")
 fun wireSamplingTest(name: String, sampling: suspend WireSamplingTestScope.() -> Unit): TestResult {
@@ -76,6 +80,8 @@ class WireSamplingTestScope(private val sampleName: String, scope: TestScope) :
         val fails = mutableListOf<String>()
         formats.forEach { format ->
             val finishedToolkit = WireToolkit(this, format).apply {
+                initClient()
+
                 server // init server
 
                 service.block()
@@ -111,6 +117,13 @@ class WireSamplingTestScope(private val sampleName: String, scope: TestScope) :
         fails.failIfAnyCauses()
     }
 
+    private suspend fun WireToolkit.initClient() {
+        KrpcClient::class.memberFunctions.single { it.name == "initializeTransport" }.apply {
+            isAccessible = true
+            callSuspend(client)
+        }
+    }
+
     var skipOldServerTests: Boolean = true
 
     suspend fun runSimulatorTests() {
@@ -147,6 +160,7 @@ class WireSamplingTestScope(private val sampleName: String, scope: TestScope) :
         val oldClientToolkit = WireToolkit(this, format, logger)
         logger.info { "Running wire test: old client (version: $version) with current server on $format format" }
 
+        oldClientToolkit.initClient()
         oldClientToolkit.server // init server
         for ((role, _, message) in dump.filter { it.phase == Phase.Send }) {
             when (role) {
@@ -255,8 +269,25 @@ private class WireToolkit(scope: CoroutineScope, format: SamplingFormat, val log
     }
 
     suspend fun stop() {
-        transport.coroutineContext.job.cancelAndJoin()
+        @OptIn(ExperimentalTime::class)
+        while (true) {
+            val now = Clock.System.now().toEpochMilliseconds()
+            if (now - transport.lastMessageSentOnClient.value > 400 &&
+                now - transport.lastMessageSentOnServer.value > 400
+            ) {
+                break
+            }
+
+            delay(100)
+        }
+
         RpcInternalDumpLoggerContainer.set(null)
+
+        client.close()
+        server.close()
+        client.awaitCompletion()
+        server.awaitCompletion()
+        transport.coroutineContext.job.cancelAndJoin()
     }
 
     init {
@@ -301,7 +332,7 @@ data class DumpLog(
                 .map { it.trim() }
                 .filter { !it.startsWith("//") && it.isNotBlank() }
                 .map { line ->
-                    val (prefix, log) = line.split("\$", limit = 2).map { it.trim() }
+                    val (prefix, log) = line.split("$", limit = 2).map { it.trim() }
                     val (role, phase) = prefix.split(" ")
 
                     DumpLog(Role.fromText(role), Phase.fromText(phase), log)

Original file line number	Diff line number	Diff line change
`@@ -65,8 +65,8 @@ class FirRpcServiceDeclarationCheckerVS(`
`65`	`65`	`}`
`66`	`66`	`}`
`67`	`67`
`68`		`-class FirRpcStrictModeClassCheckerVS(private val ctx: FirCheckersContext) : FirRegularClassChecker(MppCheckerKind.Common) {`
	`68`	`+class FirRpcStrictModeClassCheckerVS : FirRegularClassChecker(MppCheckerKind.Common) {`
`69`	`69`	`override fun check(declaration: FirRegularClass, context: CheckerContext, reporter: DiagnosticReporter) {`
`70`		`- FirRpcStrictModeClassChecker.check(ctx, declaration, context, reporter)`
	`70`	`+ FirRpcStrictModeClassChecker.check(declaration, context, reporter)`
`71`	`71`	`}`
`72`	`72`	`}`
Original file line number	Diff line number	Diff line change
`@@ -71,9 +71,9 @@ class FirRpcServiceDeclarationCheckerVS(`
`71`	`71`	`}`
`72`	`72`	`}`
`73`	`73`
`74`		`-class FirRpcStrictModeClassCheckerVS(private val ctx: FirCheckersContext) : FirRegularClassChecker(MppCheckerKind.Common) {`
	`74`	`+class FirRpcStrictModeClassCheckerVS : FirRegularClassChecker(MppCheckerKind.Common) {`
`75`	`75`	`context(context: CheckerContext, reporter: DiagnosticReporter)`
`76`	`76`	`override fun check(declaration: FirRegularClass) {`
`77`		`- FirRpcStrictModeClassChecker.check(ctx, declaration, context, reporter)`
	`77`	`+ FirRpcStrictModeClassChecker.check(declaration, context, reporter)`
`78`	`78`	`}`
`79`	`79`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,8 @@`
`1`	`1`	`public abstract class kotlinx/rpc/krpc/server/KrpcServer : kotlinx/rpc/RpcServer, kotlinx/rpc/krpc/internal/KrpcEndpoint {`
`2`	`2`	`public fun <init> (Lkotlinx/rpc/krpc/KrpcConfig$Server;Lkotlinx/rpc/krpc/KrpcTransport;)V`
	`3`	`+ public final fun awaitCompletion (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;`
	`4`	`+ public final fun close (Ljava/lang/String;)V`
	`5`	`+ public static synthetic fun close$default (Lkotlinx/rpc/krpc/server/KrpcServer;Ljava/lang/String;ILjava/lang/Object;)V`
`3`	`6`	`public fun deregisterService (Lkotlin/reflect/KClass;)V`
`4`	`7`	`public final fun registerService (Lkotlin/reflect/KClass;Lkotlin/jvm/functions/Function0;)V`
`5`	`8`	`}`