Updated timeouts and added more logs

Author: Mr3zee

krpc/krpc-core/src/commonTest/kotlin/kotlinx/rpc/krpc/KrpcSendHandlerTest.kt

@@ -111,7 +111,7 @@ internal abstract class KrpcSendHandlerBaseTest {
     }
 
     protected fun runTest(
-        timeout: Duration = 10.seconds,
+        timeout: Duration = 30.seconds,
         body: suspend TestScope.(Channel<KrpcTransportMessage>, KrpcSendHandler) -> Unit,
     ) = runTestWithCoroutinesProbes(timeout = timeout) {
         val channel = Channel<KrpcTransportMessage>(

tests/krpc-protocol-compatibility-tests/src/test/kotlin/kotlinx/rpc/krpc/test/compat/KrpcProtocolCompatibilityTests.kt

@@ -124,23 +124,37 @@ class KrpcProtocolCompatibilityTests : KrpcProtocolCompatibilityTestsBase() {
     }
 
     @TestFactory
-    fun clientStreamCancellation() = matrixTest { service, impl ->
+    fun clientStreamCancellation() = matrixTest(
+        exclude = listOf(
+            Versions.v0_9.client,
+            Versions.v0_9.server,
+            Versions.v0_8.client,
+            Versions.v0_8.server,
+        ),
+    ) { service, impl ->
         val job = launch {
+            println("[clientStreamCancellation] launching")
             service.clientStreamCancellation(flow {
                 emit(1)
+                println("[clientStreamCancellation] emit 1")
                 impl.fence.await()
+                println("[clientStreamCancellation] after fence")
             })
+            println("[clientStreamCancellation] after service call")
         }
 
         impl.entered.await()
+        println("[clientStreamCancellation] entered")
         job.cancelAndJoin()
+        println("[clientStreamCancellation] cancelled")
         impl.cancelled.await(1)
+        println("[clientStreamCancellation] awaited cancellation")
 
         assertNoErrorsInLogs()
     }
 
     @TestFactory
-    fun fastProducer() = matrixTest(timeout = 60.seconds) { service, impl ->
+    fun fastProducer() = matrixTest(timeout = 240.seconds) { service, impl ->
         val root = LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME) as ch.qos.logback.classic.Logger
 
         val async = async {

tests/krpc-protocol-compatibility-tests/src/test/kotlin/kotlinx/rpc/krpc/test/compat/service/TestService.kt

@@ -78,13 +78,20 @@ class TestServiceImpl : TestService, CompatServiceImpl {
     override suspend fun clientStreamCancellation(n: Flow<Int>) {
         try {
             n.collect {
+                println("[clientStreamCancellation] collected $it")
                 if (it != 0) {
                     entered.complete(Unit)
                 }
             }
         } catch (e: CancellationException) {
+            println("[clientStreamCancellation] cancelled on server")
             cancelled.increment()
             throw e
+        } catch (e: Throwable) {
+            println("[clientStreamCancellation] caught $e")
+            throw e
+        } finally {
+            println("[clientStreamCancellation] finally")
         }
     }
 

tests/test-utils/src/commonMain/kotlin/kotlinx/rpc/test/WaitCounter.kt

@@ -22,13 +22,19 @@ class WaitCounter {
     fun increment() {
         lock.withLock {
             val current = counter.incrementAndGet()
-            waiters[current]?.forEach { it.resume(Unit) }
+            (0..current).forEach {
+                waiters[it]?.forEach { continuation ->
+                    continuation.resume(Unit)
+                }
+
+                waiters.remove(it)
+            }
         }
     }
 
     suspend fun await(value: Int) = suspendCancellableCoroutine {
         lock.withLock {
-            if (counter.value == value) {
+            if (counter.value >= value) {
                 it.resume(Unit)
             } else {
                 waiters[value] = waiters[value].orEmpty() + it