fix(rt): add special-casing for CrtHttpEngine aws-chunked requests

.changes/f0ec11be-3875-498a-8e33-74c49416f3b8.json

@@ -0,0 +1,8 @@
+{
+    "id": "f0ec11be-3875-498a-8e33-74c49416f3b8",
+    "type": "bugfix",
+    "description": "Fix `aws-chunked` requests in the CRT HTTP engine",
+    "issues": [
+        "https://github.com/awslabs/smithy-kotlin/issues/759"
+    ]
+}

gradle.properties

@@ -44,4 +44,4 @@ kotlinLoggingVersion=2.1.21
 slf4jVersion=1.7.36
 
 # crt
-crtKotlinVersion=0.6.6
+crtKotlinVersion=0.6.7-SNAPSHOT

runtime/protocol/http-client-engines/http-client-engine-crt/common/src/aws/smithy/kotlin/runtime/http/engine/crt/CrtHttpEngine.kt

@@ -10,21 +10,28 @@ import aws.sdk.kotlin.crt.io.*
 import aws.smithy.kotlin.runtime.ClientException
 import aws.smithy.kotlin.runtime.client.ExecutionContext
 import aws.smithy.kotlin.runtime.crt.SdkDefaultIO
+import aws.smithy.kotlin.runtime.http.HttpBody
 import aws.smithy.kotlin.runtime.http.engine.HttpClientEngine
 import aws.smithy.kotlin.runtime.http.engine.HttpClientEngineBase
 import aws.smithy.kotlin.runtime.http.engine.ProxyConfig
 import aws.smithy.kotlin.runtime.http.engine.callContext
 import aws.smithy.kotlin.runtime.http.operation.getLogger
 import aws.smithy.kotlin.runtime.http.request.HttpRequest
 import aws.smithy.kotlin.runtime.http.response.HttpCall
+import aws.smithy.kotlin.runtime.io.SdkBuffer
+import aws.smithy.kotlin.runtime.io.buffer
+import aws.smithy.kotlin.runtime.io.internal.SdkDispatchers
+import aws.smithy.kotlin.runtime.io.readToByteArray
 import aws.smithy.kotlin.runtime.logging.Logger
 import aws.smithy.kotlin.runtime.time.Instant
 import kotlinx.coroutines.job
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeoutOrNull
 
 internal const val DEFAULT_WINDOW_SIZE_BYTES: Int = 16 * 1024
+internal const val CHUNK_BUFFER_SIZE: Long = 64 * 1024
 
 /**
  * [HttpClientEngine] based on the AWS Common Runtime HTTP client
@@ -100,6 +107,12 @@ public class CrtHttpEngine(public val config: CrtHttpEngineConfig) : HttpClientE
         val stream = conn.makeRequest(engineRequest, respHandler)
         stream.activate()
 
+        if (request.isChunked) {
+            withContext(SdkDispatchers.IO) {
+                stream.sendChunkedBody(request.body)
+            }
+        }
+
         val resp = respHandler.waitForResponse()
 
         return HttpCall(request, resp, reqTime, Instant.now(), callContext)
@@ -131,3 +144,32 @@ public class CrtHttpEngine(public val config: CrtHttpEngineConfig) : HttpClientE
         }
     }
 }
+
+internal suspend fun HttpStream.sendChunkedBody(body: HttpBody) {
+    when (body) {
+        is HttpBody.SourceContent -> {
+            val source = body.readFrom()
+            val bufferedSource = source.buffer()
+
+            while (!bufferedSource.exhausted()) {
+                bufferedSource.request(CHUNK_BUFFER_SIZE)
+                this.writeChunk(bufferedSource.buffer.readByteArray(), isFinalChunk = bufferedSource.exhausted())
+            }
+        }
+        is HttpBody.ChannelContent -> {
+            val chan = body.readFrom()
+            var buffer = SdkBuffer()
+            val nextBuffer = SdkBuffer()
+
+            while (!chan.isClosedForRead) {
+                chan.read(buffer, CHUNK_BUFFER_SIZE)
+
+                val isFinalChunk = chan.read(nextBuffer, CHUNK_BUFFER_SIZE) == -1L
+
+                this.writeChunk(buffer.readToByteArray(), isFinalChunk)
+                if (isFinalChunk) break else buffer = nextBuffer
+            }
+        }
+        else -> {}
+    }
+}

runtime/protocol/http-client-engines/http-client-engine-crt/common/src/aws/smithy/kotlin/runtime/http/engine/crt/RequestUtil.kt

@@ -34,7 +34,7 @@ internal val HttpRequest.uri: Uri
 internal fun HttpRequest.toCrtRequest(callContext: CoroutineContext): aws.sdk.kotlin.crt.http.HttpRequest {
     val body = this.body
     check(!body.isDuplex) { "CrtHttpEngine does not yet support full duplex streams" }
-    val bodyStream = when (body) {
+    val bodyStream = if (isChunked) null else when (body) {
         is HttpBody.Empty -> null
         is HttpBody.Bytes -> HttpRequestBodyStream.fromByteArray(body.bytes())
         is HttpBody.ChannelContent -> ReadChannelBodyStream(body.readFrom(), callContext)
@@ -61,3 +61,11 @@ internal fun HttpRequest.toCrtRequest(callContext: CoroutineContext): aws.sdk.ko
 
     return aws.sdk.kotlin.crt.http.HttpRequest(method.name, url.encodedPath, crtHeaders.build(), bodyStream)
 }
+
+/**
+ * @return whether this HttpRequest is a chunked request.
+ * Specifically, this means return `true` if a request contains a `Transfer-Encoding` header with the value `chunked`,
+ * and the body is either [HttpBody.SourceContent] or [HttpBody.ChannelContent].
+ */
+internal val HttpRequest.isChunked: Boolean get() = (this.body is HttpBody.SourceContent || this.body is HttpBody.ChannelContent) &&
+    this.headers.getAll("Transfer-Encoding")?.contains("chunked") == true

runtime/protocol/http-client-engines/http-client-engine-crt/common/test/aws/smithy/kotlin/runtime/http/engine/crt/RequestConversionTest.kt

@@ -11,12 +11,12 @@ import aws.smithy.kotlin.runtime.http.*
 import aws.smithy.kotlin.runtime.http.content.ByteArrayContent
 import aws.smithy.kotlin.runtime.http.request.HttpRequest
 import aws.smithy.kotlin.runtime.io.SdkByteReadChannel
+import aws.smithy.kotlin.runtime.io.SdkSource
+import aws.smithy.kotlin.runtime.io.source
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.cancel
 import kotlin.coroutines.EmptyCoroutineContext
-import kotlin.test.Test
-import kotlin.test.assertEquals
-import kotlin.test.assertFalse
+import kotlin.test.*
 
 class RequestConversionTest {
     private fun byteStreamFromContents(contents: String): ByteStream =
@@ -103,4 +103,44 @@ class RequestConversionTest {
         val crtRequest = request.toCrtRequest(testContext)
         assertFalse(crtRequest.headers.contains("Content-Length"))
     }
+
+    @Test
+    fun testEngineSetsNullBodyForChannelContentChunkedRequests() {
+        val testData = ByteArray(1024) { 0 }
+
+        val request = HttpRequest(
+            HttpMethod.POST,
+            Url.parse("https://test.aws.com?foo=bar"),
+            Headers.invoke { append("Transfer-Encoding", "chunked") },
+            object : HttpBody.ChannelContent() {
+                override val contentLength: Long = testData.size.toLong()
+                override fun readFrom(): SdkByteReadChannel = SdkByteReadChannel(testData)
+            },
+        )
+
+        val testContext = EmptyCoroutineContext + Job()
+        val crtRequest = request.toCrtRequest(testContext)
+        assertNotNull(request.body)
+        assertNull(crtRequest.body)
+    }
+
+    @Test
+    fun testEngineSetsNullBodyForSourceContentChunkedRequests() {
+        val testData = ByteArray(1024) { 0 }
+
+        val request = HttpRequest(
+            HttpMethod.POST,
+            Url.parse("https://test.aws.com?foo=bar"),
+            Headers.invoke { append("Transfer-Encoding", "chunked") },
+            object : HttpBody.SourceContent() {
+                override val contentLength: Long = testData.size.toLong()
+                override fun readFrom(): SdkSource = testData.source()
+            },
+        )
+
+        val testContext = EmptyCoroutineContext + Job()
+        val crtRequest = request.toCrtRequest(testContext)
+        assertNotNull(request.body)
+        assertNull(crtRequest.body)
+    }
 }

runtime/protocol/http-client-engines/http-client-engine-crt/common/test/aws/smithy/kotlin/runtime/http/engine/crt/SdkStreamResponseHandlerTest.kt

@@ -28,6 +28,7 @@ class SdkStreamResponseHandlerTest {
         override fun activate() {}
         override fun close() { closed = true }
         override fun incrementWindow(size: Int) {}
+        override fun writeChunk(chunkData: ByteArray, isFinalChunk: Boolean) {}
     }
 
     private class MockHttpClientConnection : HttpClientConnection {

runtime/protocol/http-client-engines/http-client-engine-crt/jvm/test/aws/smithy/kotlin/runtime/http/engine/crt/SendChunkedBodyTest.kt

@@ -0,0 +1,105 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package aws.smithy.kotlin.runtime.http.engine.crt
+
+import aws.sdk.kotlin.crt.http.HttpStream
+import aws.smithy.kotlin.runtime.http.toHttpBody
+import aws.smithy.kotlin.runtime.io.SdkByteReadChannel
+import aws.smithy.kotlin.runtime.io.readToByteArray
+import aws.smithy.kotlin.runtime.io.source
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runTest
+import kotlin.test.*
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class SendChunkedBodyTest {
+    private class MockHttpStream(override val responseStatusCode: Int) : HttpStream {
+        var closed: Boolean = false
+        var numChunksWritten = 0
+        override fun activate() {}
+        override fun close() { closed = true }
+        override fun incrementWindow(size: Int) {}
+        override fun writeChunk(chunkData: ByteArray, isFinalChunk: Boolean) { numChunksWritten += 1 }
+    }
+
+    @Test
+    fun testSourceContent() = runTest {
+        val stream = MockHttpStream(200)
+
+        val chunkedBytes = """
+           100;chunk-signature=${"0".repeat(64)}\r\n${"0".repeat(256)}\r\n\r\n 
+        """.trimIndent().toByteArray()
+
+        val source = chunkedBytes.source()
+
+        stream.sendChunkedBody(source.toHttpBody(chunkedBytes.size.toLong()))
+
+        // source should be fully consumed with 1 chunk written
+        assertEquals(0, source.readToByteArray().size)
+        assertEquals(1, stream.numChunksWritten)
+    }
+
+    @Test
+    fun testChannelContentMultipleChunks() = runTest {
+        val stream = MockHttpStream(200)
+
+        val chunkSize = (CHUNK_BUFFER_SIZE * 5).toInt()
+
+        val chunkedBytes = """
+           ${chunkSize.toString(16)};chunk-signature=${"0".repeat(64)}\r\n${"0".repeat(chunkSize)}\r\n\r\n 
+        """.trimIndent().toByteArray()
+
+        val source = chunkedBytes.source()
+
+        stream.sendChunkedBody(source.toHttpBody(chunkedBytes.size.toLong()))
+
+        // source should be fully consumed
+        assertEquals(0, source.readToByteArray().size)
+
+        // there should definitely be more than 1 call to `writeChunk`, but in practice we don't care how many there are
+        assertTrue(stream.numChunksWritten > 1)
+    }
+
+    @Test
+    fun testChannelContent() = runTest {
+        val stream = MockHttpStream(200)
+
+        val chunkedBytes = """
+           100;chunk-signature=${"0".repeat(64)}\r\n${"0".repeat(256)}\r\n\r\n 
+        """.trimIndent().toByteArray()
+
+        val channel = SdkByteReadChannel(chunkedBytes)
+
+        stream.sendChunkedBody(channel.toHttpBody(chunkedBytes.size.toLong()))
+
+        // channel should be fully consumed with 1 chunk written
+        assertEquals(0, channel.availableForRead)
+        assertTrue(channel.isClosedForRead)
+        assertEquals(1, stream.numChunksWritten)
+    }
+
+    @Test
+    fun testSourceContentMultipleChunks() = runTest {
+        val stream = MockHttpStream(200)
+
+        val chunkSize = (CHUNK_BUFFER_SIZE * 5).toInt()
+
+        val chunkedBytes = """
+           ${chunkSize.toString(16)};chunk-signature=${"0".repeat(64)}\r\n${"0".repeat(chunkSize)}\r\n\r\n 
+        """.trimIndent().toByteArray()
+
+        val channel = SdkByteReadChannel(chunkedBytes)
+
+        stream.sendChunkedBody(channel.toHttpBody(chunkedBytes.size.toLong()))
+
+        // source should be fully consumed
+        assertEquals(0, channel.availableForRead)
+        assertTrue(channel.isClosedForRead)
+
+        // there should definitely be more than 1 call to `writeChunk`, but in practice we don't care how many there are
+        assertTrue(stream.numChunksWritten > 1)
+    }
+}