Server Netty. Ensure connection reset detection works with HTTP2.

Author: zibet27

ktor-server/ktor-server-cio/jvm/test/io/ktor/tests/server/cio/CIOEngineTestJvm.kt

@@ -90,5 +90,6 @@ class CIOHttpRequestLifecycleTest :
     HttpRequestLifecycleTest<CIOApplicationEngine, CIOApplicationEngine.Configuration>(CIO) {
     init {
         enableSsl = false
+        enableHttp2 = false
     }
 }

ktor-server/ktor-server-core/common/src/io/ktor/server/http/HttpRequestLifecycle.kt

@@ -19,6 +19,12 @@ public class HttpRequestLifecycleConfig internal constructor() {
      * When `true`, cancels the call coroutine context if the other peer resets the client connection.
      * When `false` (default), request processing continues even if the connection is closed.
      *
+     * **When to use this property: **
+     * - Set to `true` for long-running or resource-intensive requests where you want to stop processing
+     *   immediately when the client disconnects (e.g., streaming, batch processing, heavy computations)
+     * - Keep as `false` (default) for short requests, or when you need to complete processing regardless
+     *   of client connection status (e.g., important side effects, database transactions)
+     *
      * Example:
      * ```kotlin
      * install(HttpRequestLifecycle) {
@@ -43,7 +49,8 @@ public val HttpRequestCloseHandlerKey: AttributeKey<() -> Unit> = AttributeKey<(
  * the plugin will automatically cancel the request handling coroutine if the client disconnects,
  * preventing unnecessary processing and freeing up resources.
  *
- * Remember, coroutine cancellation doesn't stop blocking operations, so check [call.coroutineContext.isActive] if needed.
+ * Remember, when the coroutine context is canceled, the next suspension point will throw [CancellationException], but until
+ * that moment it doesn't stop any blocking operations, so call `call.coroutineContext.ensureActive` if needed.
  * Plugin only works for CIO and Netty engines. Other implementations fail on closed connection only when trying to write some response.
  *
  * This is particularly useful for:
@@ -63,17 +70,14 @@ public val HttpRequestCloseHandlerKey: AttributeKey<() -> Unit> = AttributeKey<(
  *         try {
  *             // Long-running operation
  *             repeat(100) {
- *                 delay(100)
- *                 // Process data...
+ *                  // throws an exception if the client disconnects during processing
+ *                  call.coroutineContext.ensureActive()
+ *                  // Process more data...
+ *                  logger.info("Very important work.")
  *             }
  *             call.respond("Completed")
  *         } catch (e: CancellationException) {
- *             if (e.rootCause is ConnectionClosedException) {
- *                 // Client disconnected, clean up resources
- *                 logger.info("Request cancelled due to client disconnect")
- *             } else {
- *                 throw e
- *             }
+ *             // Handle client disconnected, clean up resources
  *         }
  *     }
  * }

ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/NettyApplicationCallHandler.kt

@@ -26,6 +26,7 @@ internal class NettyApplicationCallHandler(
     private val enginePipeline: EnginePipeline
 ) : ChannelInboundHandlerAdapter(), CoroutineScope {
     private var currentJob: Job? = null
+    private var currentCall: PipelineCall? = null
 
     override val coroutineContext: CoroutineContext = userCoroutineContext
 
@@ -36,22 +37,21 @@ internal class NettyApplicationCallHandler(
         }
     }
 
-    private var onConnectionClose: (() -> Unit)? = null
-
-    override fun handlerRemoved(ctx: ChannelHandlerContext?) {
-        if (ctx?.channel()?.isActive == false) {
-            onConnectionClose?.invoke()
-            onConnectionClose = null
+    internal fun onConnectionClose(context: ChannelHandlerContext) {
+        if (context.channel().isActive) {
+            return
+        }
+        currentCall?.let {
+            currentCall = null
+            @OptIn(InternalAPI::class)
+            it.attributes.getOrNull(HttpRequestCloseHandlerKey)?.invoke()
         }
     }
 
-    @OptIn(InternalAPI::class)
     private fun handleRequest(context: ChannelHandlerContext, call: PipelineCall) {
         val callContext = CallHandlerCoroutineName + NettyDispatcher.CurrentContext(context)
-        onConnectionClose = {
-            call.attributes.getOrNull(HttpRequestCloseHandlerKey)?.invoke()
-        }
 
+        currentCall = call
         currentJob = launch(callContext, start = CoroutineStart.UNDISPATCHED) {
             when {
                 call is NettyHttp1ApplicationCall && !call.request.isValid() -> {
@@ -82,8 +82,7 @@ internal class NettyApplicationCallHandler(
     }
 
     override fun channelInactive(ctx: ChannelHandlerContext) {
-        onConnectionClose?.invoke()
-        onConnectionClose = null
+        onConnectionClose(ctx)
         ctx.fireChannelInactive()
     }
 

ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/http1/NettyHttp1Handler.kt

@@ -88,7 +88,8 @@ internal class NettyHttp1Handler(
     }
 
     override fun channelInactive(context: ChannelHandlerContext) {
-        context.pipeline().remove(NettyApplicationCallHandler::class.java)
+        val handler = context.pipeline().remove(NettyApplicationCallHandler::class.java)
+        handler.onConnectionClose(context)
         context.fireChannelInactive()
     }
 

ktor-server/ktor-server-netty/jvm/src/io/ktor/server/netty/http2/NettyHttp2Handler.kt

@@ -160,7 +160,7 @@ internal class NettyHttp2Handler(
         try {
             Http2FrameCodec::class.javaObjectType.getDeclaredField("streamKey")
                 .also { it.isAccessible = true }
-        } catch (cause: Throwable) {
+        } catch (_: Throwable) {
             null
         }
     }
@@ -174,7 +174,7 @@ internal class NettyHttp2Handler(
 
         try {
             function.invoke(this, streamKey, childStream)
-        } catch (cause: Throwable) {
+        } catch (_: Throwable) {
             return false
         }
 
@@ -187,7 +187,7 @@ internal class NettyHttp2Handler(
     private tailrec fun Class<*>.findIdField(): Field {
         val idField = try {
             getDeclaredField("id")
-        } catch (t: NoSuchFieldException) {
+        } catch (_: NoSuchFieldException) {
             null
         }
         if (idField != null) {

ktor-server/ktor-server-netty/jvm/test/io/ktor/tests/server/netty/NettyEngineTest.kt

@@ -388,4 +388,9 @@ class NettyH2cEnabledTest :
 }
 
 class NettyHttpRequestLifecycleTest :
-    HttpRequestLifecycleTest<NettyApplicationEngine, NettyApplicationEngine.Configuration>(Netty)
+    HttpRequestLifecycleTest<NettyApplicationEngine, NettyApplicationEngine.Configuration>(Netty) {
+    init {
+        enableSsl = true
+        enableHttp2 = true
+    }
+}

ktor-server/ktor-server-test-base/jvm/src/io/ktor/server/test/base/EngineTestBaseJvm.kt

@@ -7,7 +7,6 @@ package io.ktor.server.test.base
 import io.ktor.client.*
 import io.ktor.client.engine.apache.*
 import io.ktor.client.engine.cio.*
-import io.ktor.client.plugins.*
 import io.ktor.client.request.*
 import io.ktor.client.statement.*
 import io.ktor.http.*
@@ -249,10 +248,10 @@ actual abstract class EngineTestBase<
         builder: suspend HttpRequestBuilder.() -> Unit,
         block: suspend HttpResponse.(Int) -> Unit
     ) {
-        withUrl("http://127.0.0.1:$port$path", port, builder, block)
+        withHttp1("http://127.0.0.1:$port$path", port, builder, block)
 
         if (enableSsl) {
-            withUrl("https://127.0.0.1:$sslPort$path", sslPort, builder, block)
+            withHttp1("https://127.0.0.1:$sslPort$path", sslPort, builder, block)
         }
 
         if (enableHttp2 && enableSsl) {
@@ -270,7 +269,7 @@ actual abstract class EngineTestBase<
         }
     }
 
-    private suspend fun withUrl(
+    protected suspend fun withHttp1(
         urlString: String,
         port: Int,
         builder: suspend HttpRequestBuilder.() -> Unit,
@@ -284,22 +283,13 @@ actual abstract class EngineTestBase<
         }
     }
 
-    private suspend fun withHttp2(
+    protected suspend fun withHttp2(
         url: String,
         port: Int,
         builder: suspend HttpRequestBuilder.() -> Unit,
         block: suspend HttpResponse.(Int) -> Unit
     ) {
-        HttpClient(Apache) {
-            followRedirects = false
-            expectSuccess = false
-            engine {
-                pipelining = true
-                sslContext = SSLContext.getInstance("SSL").apply {
-                    init(null, trustAllCertificates, SecureRandom())
-                }
-            }
-        }.use { client ->
+        createApacheClient().use { client ->
             client.prepareRequest(url) {
                 builder()
             }.execute { response ->
@@ -310,33 +300,48 @@ actual abstract class EngineTestBase<
 
     companion object {
         val keyStoreFile: File = File("build/temp.jks")
-        lateinit var keyStore: KeyStore
-        lateinit var sslContext: SSLContext
-        lateinit var trustManager: X509TrustManager
+        val keyStore: KeyStore by lazy { generateCertificate(keyStoreFile) }
         lateinit var client: HttpClient
 
-        @BeforeAll
-        @JvmStatic
-        fun setupAll() {
-            keyStore = generateCertificate(keyStoreFile, algorithm = "SHA256withECDSA", keySizeInBits = 256)
+        fun createTrustManager(): X509TrustManager {
+            val sslContext = SSLContext.getInstance("TLS")
             val tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm())
             tmf.init(keyStore)
-            sslContext = SSLContext.getInstance("TLS")
             sslContext.init(null, tmf.trustManagers, null)
-            trustManager = tmf.trustManagers.first { it is X509TrustManager } as X509TrustManager
+            return tmf.trustManagers.first { it is X509TrustManager } as X509TrustManager
+        }
 
-            client = HttpClient(CIO) {
+        fun createCIOClient(): HttpClient {
+            return HttpClient(CIO) {
                 engine {
-                    https.trustManager = trustManager
+                    https.trustManager = createTrustManager()
                     https.serverName = "localhost"
                     requestTimeout = 0
                 }
+                followRedirects = false
+                expectSuccess = false
+            }
+        }
 
+        fun createApacheClient(): HttpClient {
+            return HttpClient(Apache) {
                 followRedirects = false
                 expectSuccess = false
+                engine {
+                    pipelining = true
+                    sslContext = SSLContext.getInstance("SSL").apply {
+                        init(null, trustAllCertificates, SecureRandom())
+                    }
+                }
             }
         }
 
+        @BeforeAll
+        @JvmStatic
+        fun setupAll() {
+            client = createCIOClient()
+        }
+
         @AfterAll
         @JvmStatic
         fun cleanup() {
@@ -354,13 +359,14 @@ actual abstract class EngineTestBase<
                 }
             } while (true)
         }
-    }
 
-    private val trustAllCertificates = arrayOf<X509TrustManager>(
-        object : X509TrustManager {
-            override fun getAcceptedIssuers(): Array<X509Certificate> = emptyArray()
-            override fun checkClientTrusted(certs: Array<X509Certificate>, authType: String) {}
-            override fun checkServerTrusted(certs: Array<X509Certificate>, authType: String) {}
-        }
-    )
+        private val trustAllCertificates = arrayOf<X509TrustManager>(
+            @Suppress("CustomX509TrustManager")
+            object : X509TrustManager {
+                override fun getAcceptedIssuers(): Array<X509Certificate> = emptyArray()
+                override fun checkClientTrusted(certs: Array<X509Certificate>, authType: String) {}
+                override fun checkServerTrusted(certs: Array<X509Certificate>, authType: String) {}
+            }
+        )
+    }
 }

ktor-server/ktor-server-test-suites/jvm/src/io/ktor/server/testing/suites/ConfigTestSuite.kt

@@ -10,8 +10,6 @@ import java.util.concurrent.*
 import kotlin.system.*
 import kotlin.test.*
 
-var count = 0
-
 abstract class ConfigTestSuite(val engine: ApplicationEngineFactory<*, *>) {
 
     @Test
@@ -58,4 +56,8 @@ abstract class ConfigTestSuite(val engine: ApplicationEngineFactory<*, *>) {
 
         assertTrue(time < 100, "Stop time is $time")
     }
+
+    private companion object {
+        var count = 0
+    }
 }