Release 2.3.7 (#3852)

CHANGELOG.md

@@ -1,3 +1,12 @@
+# 2.3.7
+> Published 28 November 2023
+
+### Bugfixes
+* Server ContentNegotiation no longer allows multiple decoders for one Content-Type ([KTOR-5410](https://youtrack.jetbrains.com/issue/KTOR-5410/Server-ContentNegotiation-no-longer-allows-multiple-decoders-for-one-Content-Type))
+* High Native Server Memory Usage ([KTOR-6321](https://youtrack.jetbrains.com/issue/KTOR-6321/High-Native-Server-Memory-Usage))
+* WebSockets: Confusing error message when server doesn't respond with Upgrade ([KTOR-6397](https://youtrack.jetbrains.com/issue/KTOR-6397/WebSockets-Confusing-error-message-when-server-doesnt-respond-with-Upgrade))
+* ContentNegotiation: Adding charset to content type of JacksonConverter breaks request matching ([KTOR-6420](https://youtrack.jetbrains.com/issue/KTOR-6420))
+
 # 2.3.6
 > Published 7 November 2023
 

build.gradle.kts

@@ -97,6 +97,10 @@ plugins {
     id("com.osacky.doctor") version "0.8.1"
 }
 
+doctor {
+    enableTestCaching = false
+}
+
 allprojects {
     group = "io.ktor"
     version = configuredVersion

buildSrc/src/main/kotlin/test/server/tests/WebSockets.kt

@@ -64,6 +64,9 @@ internal fun Application.webSockets() {
                     }
                 }
             }
+            get("500") {
+                throw IllegalStateException()
+            }
         }
     }
 }

gradle.properties

@@ -10,7 +10,7 @@ ktor.ide.jvmAndCommonOnly=true
 kotlin.code.style=official
 
 # config
-version=2.3.6
+version=2.3.7
 
 # gradle
 org.gradle.daemon=true
@@ -37,3 +37,9 @@ slf4j_version=1.7.36
 junit_version=4.13.2
 logback_version=1.2.11
 
+kotlin.daemon.jvmargs=-Xmx6g -XX:+HeapDumpOnOutOfMemoryError
+kotlin.daemon.useFallbackStrategy=false
+
+# dokka
+# workaround for resolving platform dependencies, see https://github.com/Kotlin/dokka/issues/3153
+org.jetbrains.dokka.classpath.useNativeDistributionAccessor=true

ktor-client/ktor-client-cio/jvmAndNix/test/CIOEngineTest.kt

@@ -151,6 +151,21 @@ class CIOEngineTest {
         }
     }
 
+    @Test
+    fun testErrorMessageWhenServerDontRespondWithUpgrade() = testWithEngine(CIO) {
+        config {
+            install(WebSockets)
+        }
+
+        test { client ->
+            kotlin.test.assertFailsWith<WebSocketException> {
+                client.webSocket("$TEST_WEBSOCKET_SERVER/websockets/500") {}
+            }.apply {
+                assertEquals(message, "Handshake exception, expected status code 101 but was 500")
+            }
+        }
+    }
+
     private suspend fun sendExpectRequest(socket: ServerSocket, client: HttpClient, body: String? = null) =
         client.post {
             val serverPort = (socket.localAddress as InetSocketAddress).port

ktor-client/ktor-client-core/common/src/io/ktor/client/plugins/websocket/WebSockets.kt

@@ -170,12 +170,26 @@ public class WebSockets internal constructor(
             }
 
             scope.responsePipeline.intercept(HttpResponsePipeline.Transform) { (info, session) ->
-                if (session !is WebSocketSession) {
+                val response = this.context.response
+                val status = response.status
+                val requestContent = response.request.content
+
+                if (requestContent !is WebSocketContent) {
                     LOGGER.trace("Skipping non-websocket response from ${context.request.url}: $session")
                     return@intercept
                 }
-                LOGGER.trace("Receive websocket session from ${context.request.url}: $session")
+                if (status != HttpStatusCode.SwitchingProtocols) {
+                    throw WebSocketException(
+                        "Handshake exception, expected status code ${HttpStatusCode.SwitchingProtocols.value} but was ${status.value}" // ktlint-disable max-line-length
+                    )
+                }
+                if (session !is WebSocketSession) {
+                    throw WebSocketException(
+                        "Handshake exception, expected `WebSocketSession` content but was $session"
+                    )
+                }
 
+                LOGGER.trace("Receive websocket session from ${context.request.url}: $session")
                 val clientSession: ClientWebSocketSession = when (info.type) {
                     DefaultClientWebSocketSession::class -> {
                         val defaultSession = plugin.convertSessionToDefault(session)
@@ -195,8 +209,7 @@ public class WebSockets internal constructor(
                     else -> DelegatingClientWebSocketSession(context, session)
                 }
 
-                val response = HttpResponseContainer(info, clientSession)
-                proceedWith(response)
+                proceedWith(HttpResponseContainer(info, clientSession))
             }
         }
     }

ktor-client/ktor-client-core/js/src/io/ktor/client/engine/js/JsClientEngine.kt

@@ -100,7 +100,7 @@ internal class JsClientEngine(
         val session = JsWebSocketSession(callContext, socket)
 
         return HttpResponseData(
-            HttpStatusCode.OK,
+            HttpStatusCode.SwitchingProtocols,
             requestTime,
             Headers.Empty,
             HttpProtocolVersion.HTTP_1_1,

ktor-client/ktor-client-darwin/darwin/src/io/ktor/client/engine/darwin/internal/DarwinWebsocketSession.kt

@@ -146,7 +146,7 @@ internal class DarwinWebsocketSession(
 
     fun didOpen() {
         val response = HttpResponseData(
-            HttpStatusCode.OK,
+            HttpStatusCode.SwitchingProtocols,
             requestTime,
             Headers.Empty,
             HttpProtocolVersion.HTTP_1_1,

ktor-client/ktor-client-java/jvm/src/io/ktor/client/engine/java/JavaHttpWebSocket.kt

@@ -156,7 +156,7 @@ internal class JavaHttpWebSocket(
         webSocket = builder.buildAsync(requestData.url.toURI(), this).await()
 
         return HttpResponseData(
-            HttpStatusCode.OK,
+            HttpStatusCode.SwitchingProtocols,
             requestTime,
             Headers.Empty,
             HttpProtocolVersion.HTTP_1_1,

ktor-io/common/src/io/ktor/utils/io/ByteChannelSequential.kt

@@ -753,6 +753,7 @@ public abstract class ByteChannelSequentialBase(
             flushBuffer.release()
         } else {
             flush()
+            writable.release()
         }
 
         slot.cancel(cause)

ktor-io/common/src/io/ktor/utils/io/core/Buffer.kt

@@ -329,7 +329,8 @@ public open class Buffer(public val memory: Memory) {
     }
 
     override fun toString(): String {
-        return "Buffer($readRemaining used, $writeRemaining free, ${startGap + endGap} reserved of $capacity)"
+        return "Buffer[0x${hashCode().toString(16)}]" +
+            "($readRemaining used, $writeRemaining free, ${startGap + endGap} reserved of $capacity)"
     }
 
     public companion object {

ktor-io/common/src/io/ktor/utils/io/core/BytePacketBuilder.kt

@@ -86,7 +86,8 @@ public class BytePacketBuilder(
         }
     }
 
+    @OptIn(ExperimentalStdlibApi::class)
     override fun toString(): String {
-        return "BytePacketBuilder($size bytes written)"
+        return "BytePacketBuilder[0x${hashCode()}]"
     }
 }

ktor-io/common/src/io/ktor/utils/io/core/ByteReadPacket.kt

@@ -38,8 +38,9 @@ public class ByteReadPacket internal constructor(
     final override fun closeSource() {
     }
 
+    @OptIn(ExperimentalStdlibApi::class)
     override fun toString(): String {
-        return "ByteReadPacket($remaining bytes remaining)"
+        return "ByteReadPacket[${hashCode()}]"
     }
 
     public companion object {

ktor-io/common/src/io/ktor/utils/io/core/internal/ChunkBuffer.kt

@@ -127,22 +127,7 @@ public open class ChunkBuffer(
     }
 
     public companion object {
-        public val Pool: ObjectPool<ChunkBuffer> = object : ObjectPool<ChunkBuffer> {
-            override val capacity: Int
-                get() = DefaultChunkedBufferPool.capacity
-
-            override fun borrow(): ChunkBuffer {
-                return DefaultChunkedBufferPool.borrow()
-            }
-
-            override fun recycle(instance: ChunkBuffer) {
-                DefaultChunkedBufferPool.recycle(instance)
-            }
-
-            override fun dispose() {
-                DefaultChunkedBufferPool.dispose()
-            }
-        }
+        public val Pool: ObjectPool<ChunkBuffer> get() = DefaultChunkedBufferPool
 
         /**
          * A pool that always returns [ChunkBuffer.Empty]

ktor-io/posix/src/io/ktor/utils/io/pool/DefaultPool.kt → ktor-io/posix/src/io/ktor/utils/io/pool/DefaultPool.posix.kt

@@ -12,8 +12,17 @@ public actual abstract class DefaultPool<T : Any> actual constructor(
     )
     protected val lock: SynchronizedObject = SynchronizedObject()
 
-    private val instances = atomicArrayOfNulls<Any?>(capacity)
-    private var size by atomic(0)
+    private val instances = mutableListOf<T>()
+
+    private val _allocated = atomic(0)
+    private val _released = atomic(0)
+    private val _recycled = atomic(0)
+
+    public val inCache: Int get() = instances.size
+    public val inUsed: Int get() = _allocated.value - _released.value
+    public val allocated: Int get() = _allocated.value
+    public val released: Int get() = _released.value
+    public val recycled: Int get() = _recycled.value
 
     protected actual abstract fun produceInstance(): T
     protected actual open fun disposeInstance(instance: T) {}
@@ -23,38 +32,35 @@ public actual abstract class DefaultPool<T : Any> actual constructor(
 
     @Suppress("DEPRECATION")
     public actual final override fun borrow(): T = synchronized(lock) {
-        if (size == 0) return produceInstance()
-        val idx = --size
-
-        @Suppress("UNCHECKED_CAST")
-        val instance = instances[idx].value as T
-        instances[idx].value = null
+        if (instances.isEmpty()) {
+            _allocated.incrementAndGet()
+            return@synchronized produceInstance()
+        }
 
-        return clearInstance(instance)
+        val result = instances.removeAt(instances.lastIndex)
+        clearInstance(result)
+        return@synchronized result
     }
 
     @Suppress("DEPRECATION")
     public actual final override fun recycle(instance: T) {
         synchronized(lock) {
+            _recycled.incrementAndGet()
             validateInstance(instance)
-            if (size == capacity) {
-                disposeInstance(instance)
+            if (instances.size < capacity) {
+                instances.add(instance)
             } else {
-                instances[size++].value = instance
+                _released.incrementAndGet()
+                disposeInstance(instance)
             }
         }
     }
 
     @Suppress("DEPRECATION")
     public actual final override fun dispose() {
         synchronized(lock) {
-            for (i in 0 until size) {
-                @Suppress("UNCHECKED_CAST")
-                val instance = instances[i].value as T
-                instances[i].value = null
-                disposeInstance(instance)
-            }
-            size = 0
+            instances.forEach { disposeInstance(it) }
+            instances.clear()
         }
     }
 }

ktor-network/jvmAndNix/test/io/ktor/network/sockets/tests/TCPSocketTest.kt

@@ -123,8 +123,6 @@ class TCPSocketTest {
             assertFailsWith<CancellationException> {
                 readChannel.readByte()
             }
-
-            assertTrue(client.isClosed)
         }
     }
 }

ktor-network/nix/src/io/ktor/network/sockets/CIOReader.kt

@@ -17,43 +17,47 @@ internal fun CoroutineScope.attachForReadingImpl(
     descriptor: Int,
     selectable: Selectable,
     selector: SelectorManager
-): WriterJob = writer(Dispatchers.Unconfined, userChannel) {
-    while (!channel.isClosedForWrite) {
-        var close = false
-        val count = channel.write { memory, startIndex, endIndex ->
-            val bufferStart = memory.pointer + startIndex
-            val size = endIndex - startIndex
-            val bytesRead = recv(descriptor, bufferStart, size.convert(), 0).toInt()
-
-            when (bytesRead) {
-                0 -> close = true
-                -1 -> {
-                    if (errno == EAGAIN) return@write 0
-                    throw PosixException.forErrno()
+): WriterJob = writer(Dispatchers.IO, userChannel) {
+    try {
+        while (!channel.isClosedForWrite) {
+            var close = false
+            val count = channel.write { memory, startIndex, endIndex ->
+                val bufferStart = memory.pointer + startIndex
+                val size = endIndex - startIndex
+                val bytesRead = recv(descriptor, bufferStart, size.convert(), 0).toInt()
+
+                when (bytesRead) {
+                    0 -> close = true
+                    -1 -> {
+                        if (errno == EAGAIN) return@write 0
+                        throw PosixException.forErrno()
+                    }
                 }
-            }
-
-            bytesRead
-        }
 
-        channel.flush()
-        if (close) {
-            channel.close()
-            break
-        }
+                bytesRead
+            }
 
-        if (count == 0) {
-            try {
-                selector.select(selectable, SelectInterest.READ)
-            } catch (_: IOException) {
+            channel.flush()
+            if (close) {
+                channel.close()
                 break
             }
+
+            if (count == 0) {
+                try {
+                    selector.select(selectable, SelectInterest.READ)
+                } catch (_: IOException) {
+                    break
+                }
+            }
         }
-    }
 
-    channel.closedCause?.let { throw it }
-}.apply {
-    invokeOnCompletion {
+        channel.closedCause?.let { throw it }
+    } catch (cause: Throwable) {
+        channel.close(cause)
+        throw cause
+    } finally {
         shutdown(descriptor, SHUT_RD)
+        channel.close()
     }
 }

ktor-network/nix/src/io/ktor/network/sockets/CIOWriter.kt

@@ -18,7 +18,7 @@ internal fun CoroutineScope.attachForWritingImpl(
     descriptor: Int,
     selectable: Selectable,
     selector: SelectorManager
-): ReaderJob = reader(Dispatchers.Unconfined, userChannel) {
+): ReaderJob = reader(Dispatchers.IO, userChannel) {
     val source = channel
     var sockedClosed = false
     var needSelect = false

ktor-server/ktor-server-plugins/ktor-server-call-logging/jvm/test/io/ktor/server/plugins/callloging/CallLoggingTest.kt

@@ -79,19 +79,13 @@ class CallLoggingTest {
 
         assertTrue(messages.size >= 3, "It should be at least 3 message logged:\n$messages")
         assertTrue {
-            messages[messages.size - 3].startsWith(
-                "INFO: Application started: class io.ktor.server.application.Application(0x$hash)"
-            )
+            messages.contains("INFO: Application started: io.ktor.server.application.Application@$hash")
         }
         assertTrue {
-            messages[messages.size - 2].startsWith(
-                "INFO: Application stopping: class io.ktor.server.application.Application(0x$hash)"
-            )
+            messages.contains("INFO: Application stopping: io.ktor.server.application.Application@$hash")
         }
         assertTrue {
-            messages[messages.size - 1].startsWith(
-                "INFO: Application stopped: class io.ktor.server.application.Application(0x$hash)"
-            )
+            messages.contains("INFO: Application stopped: io.ktor.server.application.Application@$hash")
         }
     }