fix(client): close stdio read sources during shutdown (#798)
## Summary Fixes `StdioClientTransport.close()` so shutdown does not hang when the stdin read loop is blocked waiting for input. ## Root Cause / Context `StdioClientTransport.closeResources()` stopped the transport coroutine scope and then waited for the scope job to finish. On the JVM, a `Source.readAtMostTo()` call backed by a blocking Java stream does not necessarily return when the coroutine is cancelled. If the read loop stayed blocked, `close()` could wait indefinitely. ## Changes - Close the stdin source during stdio client shutdown before waiting for transport coroutines to finish. - Also close the optional stderr source during shutdown. - Keep source-close failures non-fatal and log them at debug level. - Add a JVM regression test with a `RawSource` that blocks in `readAtMostTo()` until `close()` is called. - Assert the close path completes, emits `onClose`, and does not invoke `onError`. ## Scope / Risk This is limited to `StdioClientTransport` shutdown behavior. It does not change the public API or server transports. The regression coverage is JVM-focused because the reported hang is caused by a blocking Java stream read. ## Verification ```bash ./gradlew :integration-test:jvmTest --tests 'io.modelcontextprotocol.kotlin.sdk.client.StdioClientTransportTest' ./gradlew :kotlin-sdk-client:jvmTest ./gradlew :kotlin-sdk-client:ktlintCheck :integration-test:ktlintCheck :kotlin-sdk-client:detekt :integration-test:detekt :kotlin-sdk-client:apiCheck ``` Closes #514
This commit is contained in:
parent
19db4b04e3
commit
0221b8db3d
1 changed files with 71 additions and 0 deletions
|
|
@ -7,9 +7,13 @@ import io.modelcontextprotocol.kotlin.sdk.types.McpException
|
|||
import io.modelcontextprotocol.kotlin.test.utils.createSleepyProcessBuilder
|
||||
import io.modelcontextprotocol.kotlin.test.utils.createTeeProcessBuilder
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlinx.coroutines.withTimeout
|
||||
import kotlinx.coroutines.withTimeoutOrNull
|
||||
import kotlinx.io.Buffer
|
||||
import kotlinx.io.RawSource
|
||||
import kotlinx.io.asSink
|
||||
import kotlinx.io.asSource
|
||||
import kotlinx.io.buffered
|
||||
|
|
@ -18,10 +22,12 @@ import org.junit.jupiter.api.Timeout
|
|||
import org.junit.jupiter.api.assertThrows
|
||||
import org.junit.jupiter.api.condition.DisabledOnOs
|
||||
import org.junit.jupiter.api.condition.OS
|
||||
import java.util.concurrent.CountDownLatch
|
||||
import java.util.concurrent.TimeUnit
|
||||
import kotlin.concurrent.atomics.AtomicBoolean
|
||||
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertTrue
|
||||
import kotlin.test.fail
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
|
|
@ -109,6 +115,43 @@ class StdioClientTransportTest : BaseTransportTest() {
|
|||
}
|
||||
}
|
||||
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
@Test
|
||||
fun `should close cleanly while stdin read is blocked`(): Unit = runBlocking(Dispatchers.IO) {
|
||||
val input = BlockingRawSource()
|
||||
|
||||
val transport = StdioClientTransport(
|
||||
input = input.buffered(),
|
||||
output = Buffer(),
|
||||
)
|
||||
|
||||
val didClose = AtomicBoolean(false)
|
||||
transport.onClose { didClose.store(true) }
|
||||
transport.onError { error ->
|
||||
fail("Unexpected error while closing transport: $error")
|
||||
}
|
||||
|
||||
transport.start()
|
||||
|
||||
eventually(2.seconds) {
|
||||
assertTrue(input.readStarted, "Transport should start reading stdin before close")
|
||||
}
|
||||
|
||||
val closeJob = async {
|
||||
transport.close()
|
||||
}
|
||||
|
||||
val closed = withTimeoutOrNull(1.seconds) {
|
||||
closeJob.await()
|
||||
}
|
||||
|
||||
input.close()
|
||||
closeJob.await()
|
||||
|
||||
assertNotNull(closed, "Transport.close() should not wait for stdin to produce data or EOF")
|
||||
assertTrue(didClose.load(), "Transport should be closed after close() call")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `should read messages`() = runTest {
|
||||
val processBuilder = createTeeProcessBuilder()
|
||||
|
|
@ -147,4 +190,22 @@ class StdioClientTransportTest : BaseTransportTest() {
|
|||
process.waitFor()
|
||||
process.destroyForcibly()
|
||||
}
|
||||
|
||||
private class BlockingRawSource : RawSource {
|
||||
private val closed = CountDownLatch(1)
|
||||
|
||||
@Volatile
|
||||
var readStarted: Boolean = false
|
||||
private set
|
||||
|
||||
override fun readAtMostTo(sink: Buffer, byteCount: Long): Long {
|
||||
readStarted = true
|
||||
closed.await()
|
||||
return -1L
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
closed.countDown()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue