fix: properly close byte streams when using CRT engine

Author: ianbotsf

.changes/0c737e4d-adb6-45f4-95d3-7123f3455171.json

@@ -0,0 +1,5 @@
+{
+    "id": "0c737e4d-adb6-45f4-95d3-7123f3455171",
+    "type": "bugfix",
+    "description": "Fix occasional stream leak due to race condition in CRT engine"
+}

runtime/crt-util/common/src/aws/smithy/kotlin/runtime/crt/ReadChannelBodyStream.kt

@@ -15,6 +15,9 @@ import kotlinx.atomicfu.atomic
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
 import kotlin.coroutines.CoroutineContext
+import kotlin.time.Duration.Companion.milliseconds
+
+private val POLLING_DELAY = 100.milliseconds
 
 /**
  * write as much of [outgoing] to [dest] as possible
@@ -41,6 +44,21 @@ public class ReadChannelBodyStream(
         producerJob.invokeOnCompletion { cause ->
             bodyChan.cancel(cause)
         }
+
+        // Poll the channel's `isClosedForRead` and complete when it's true. This works around a timing issue when the
+        // write side of the channel finishes sending bytes but doesn't call `close` in time for the CRT's
+        // `sendRequestBody` loop to pick it up. If CRT reads all the bytes it expects, it ceases calling
+        // `sendRequestBody`, which risks leaving the producer job open indefinitely. This polling loop catches any
+        // missed channel closures and ends the producer job to avoid that issue.
+        launch(coroutineContext + CoroutineName("body-channel-watchdog")) {
+            while (producerJob.isActive) {
+                if (bodyChan.isClosedForRead) {
+                    producerJob.complete()
+                    return@launch
+                }
+                delay(POLLING_DELAY)
+            }
+        }
     }
 
     // lie - CRT tries to control this via normal seek operations (e.g. when they calculate a hash for signing

runtime/protocol/http-client-engines/test-suite/common/test/aws/smithy/kotlin/runtime/http/test/UploadTest.kt

@@ -7,23 +7,18 @@ package aws.smithy.kotlin.runtime.http.test
 
 import aws.smithy.kotlin.runtime.content.ByteStream
 import aws.smithy.kotlin.runtime.hashing.sha256
-import aws.smithy.kotlin.runtime.http.HttpBody
-import aws.smithy.kotlin.runtime.http.HttpMethod
-import aws.smithy.kotlin.runtime.http.HttpStatusCode
+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.http.request.headers
 import aws.smithy.kotlin.runtime.http.response.complete
 import aws.smithy.kotlin.runtime.http.test.util.AbstractEngineTest
 import aws.smithy.kotlin.runtime.http.test.util.test
 import aws.smithy.kotlin.runtime.http.test.util.testSetup
-import aws.smithy.kotlin.runtime.http.toHttpBody
 import aws.smithy.kotlin.runtime.io.SdkByteChannel
 import aws.smithy.kotlin.runtime.io.SdkByteReadChannel
 import aws.smithy.kotlin.runtime.util.encodeToHex
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.*
 import kotlin.test.Test
 import kotlin.test.assertEquals
 
@@ -84,6 +79,39 @@ class UploadTest : AbstractEngineTest() {
         }
     }
 
+    @Test
+    fun testUploadWithClosingDelay() = testEngines {
+        test { env, client ->
+            val data = ByteArray(16) { it.toByte() }
+            val sha = data.sha256().encodeToHex()
+            val ch = SdkByteChannel(autoFlush = true)
+            val content = object : HttpBody.Streaming() {
+                override val contentLength: Long = data.size.toLong()
+                override fun readFrom(): SdkByteReadChannel = ch
+            }
+
+            val req = HttpRequest {
+                method = HttpMethod.POST
+                testSetup(env)
+                url.path = "/upload/content"
+                body = content
+            }
+
+            coroutineScope {
+                launch {
+                    ch.writeFully(data)
+                    delay(1000)
+                    // CRT will have stopped polling by now
+                    ch.close()
+                }
+                val call = client.call(req)
+                call.complete()
+                assertEquals(HttpStatusCode.OK, call.response.status)
+                assertEquals(sha, call.response.headers["content-sha256"])
+            }
+        }
+    }
+
     @Test
     fun testUploadWithWrappedStream() = testEngines {
         // test custom ByteStream behavior