Server Netty. Fix rejected execution during engine stop (#8671)

Author: zibet27

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

@@ -35,6 +35,11 @@ internal class NettyApplicationCallHandler(
         }
     }
 
+    override fun channelUnregistered(ctx: ChannelHandlerContext?) {
+        currentJob?.cancel()
+        super.channelUnregistered(ctx)
+    }
+
     private fun handleRequest(context: ChannelHandlerContext, call: PipelineCall) {
         val callContext = CallHandlerCoroutineName + NettyDispatcher.CurrentContext(context)
 

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

@@ -280,29 +280,32 @@ public class NettyApplicationEngine(
         workerEventGroup.shutdownGracefully().sync()
     }
 
+    private inline fun <R> withStopException(crossinline block: () -> R) {
+        runCatching(block).onFailure {
+            environment.log.error("Exception thrown during engine stop", it)
+        }
+    }
+
     override fun stop(gracePeriodMillis: Long, timeoutMillis: Long) {
-        cancellationJob?.complete()
+        if (cancellationJob?.complete() != true) {
+            // Engine was already stopped or started stop process
+            return
+        }
         monitor.raise(ApplicationStopPreparing, environment)
         val channelFutures = channels?.mapNotNull { if (it.isOpen) it.close() else null }.orEmpty()
+        channelFutures.forEach { future ->
+            withStopException { future.sync() }
+        }
 
-        try {
-            val shutdownConnections =
-                connectionEventGroup.shutdownGracefully(gracePeriodMillis, timeoutMillis, TimeUnit.MILLISECONDS)
-
-            val shutdownWorkers =
-                workerEventGroup.shutdownGracefully(gracePeriodMillis, timeoutMillis, TimeUnit.MILLISECONDS)
-            if (configuration.shareWorkGroup) {
-                shutdownWorkers.await()
-            } else {
-                val shutdownCall =
-                    callEventGroup.shutdownGracefully(gracePeriodMillis, timeoutMillis, TimeUnit.MILLISECONDS)
-                shutdownWorkers.await()
-                shutdownCall.await()
+        val shutdownConnections = connectionEventGroup.shutdownGracefully(gracePeriodMillis, timeoutMillis, TimeUnit.MILLISECONDS)
+        val shutdownWorkers = workerEventGroup.shutdownGracefully(gracePeriodMillis, timeoutMillis, TimeUnit.MILLISECONDS)
+        if (!configuration.shareWorkGroup) {
+            withStopException {
+                callEventGroup.shutdownGracefully(gracePeriodMillis, timeoutMillis, TimeUnit.MILLISECONDS).sync()
             }
-            shutdownConnections.await()
-        } finally {
-            channelFutures.forEach { it.sync() }
         }
+        withStopException { shutdownConnections.sync() }
+        withStopException { shutdownWorkers.sync() }
     }
 
     override fun toString(): String {

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

@@ -51,7 +51,7 @@ public abstract class NettyApplicationResponse(
         // because it should've been set by commitHeaders earlier
         val chunked = headers[HttpHeaders.TransferEncoding] == "chunked"
 
-        if (responseMessageSent) return
+        if (responseMessageSent || !context.channel().isActive) return
 
         val message = responseMessage(chunked, bytes)
         responseChannel = when (message) {
@@ -111,7 +111,7 @@ public abstract class NettyApplicationResponse(
     }
 
     internal fun sendResponse(chunked: Boolean = true, content: ByteReadChannel) {
-        if (responseMessageSent) return
+        if (responseMessageSent || !context.channel().isActive) return
 
         responseChannel = content
         responseMessage = when {

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

@@ -23,12 +23,8 @@ import io.ktor.websocket.*
 import io.netty.buffer.ByteBuf
 import io.netty.buffer.Unpooled
 import io.netty.handler.codec.http.HttpResponseStatus
-import io.netty.handler.codec.http2.DefaultHttp2Headers
-import io.netty.handler.codec.http2.DefaultHttp2HeadersDecoder
-import io.netty.handler.codec.http2.DefaultHttp2HeadersEncoder
+import io.netty.handler.codec.http2.*
 import io.netty.handler.codec.http2.Http2CodecUtil.readUnsignedInt
-import io.netty.handler.codec.http2.Http2Flags
-import io.netty.handler.codec.http2.Http2FrameTypes
 import kotlinx.coroutines.flow.consumeAsFlow
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -381,7 +377,7 @@ class NettyH2cEnabledTest :
             total += n
             val s = buf.decodeToString(0, total)
             val end = s.indexOf("\r\n\r\n")
-            if (end >= 0) return s.substring(0, end + 4)
+            if (end >= 0) return s.take(end + 4)
             require(total < maxBytes) { "HTTP/1.1 headers exceed $maxBytes bytes" }
         }
     }