fix(rt): okhttp engine crashing on Android when coroutine is cancelled while uploading request body (#723)

Author: aajtodd

.changes/841987e7-c68c-4ff6-8426-f7bf51373d79.json

@@ -0,0 +1,8 @@
+{
+    "id": "841987e7-c68c-4ff6-8426-f7bf51373d79",
+    "type": "bugfix",
+    "description": "Fix OkHttp engine crashing on Android when coroutine is cancelled while uploading request body",
+    "issues": [
+        "awslabs/aws-sdk-kotlin#733"
+    ]
+}

runtime/protocol/http-client-engines/http-client-engine-okhttp/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp/ByteChannelRequestBody.kt

@@ -10,6 +10,7 @@ import kotlinx.coroutines.*
 import okhttp3.MediaType
 import okhttp3.RequestBody
 import okio.BufferedSink
+import java.io.IOException
 import java.nio.ByteBuffer
 import kotlin.coroutines.CoroutineContext
 
@@ -76,7 +77,9 @@ private inline fun <T> withJob(job: CompletableJob, block: () -> T): T {
         return block()
     } catch (ex: Exception) {
         job.completeExceptionally(ex)
-        throw ex
+        // wrap all exceptions thrown from inside `okhttp3.RequestBody#writeTo(..)` as an IOException
+        // see https://github.com/awslabs/aws-sdk-kotlin/issues/733
+        throw IOException(ex)
     } finally {
         job.complete()
     }

runtime/protocol/http-client-engines/http-client-engine-okhttp/jvm/test/aws/smithy/kotlin/runtime/http/engine/okhttp/ByteChannelRequestBodyTest.kt

@@ -15,10 +15,9 @@ import kotlinx.coroutines.test.runTest
 import okio.Buffer
 import okio.BufferedSink
 import org.junit.jupiter.api.Test
+import java.io.IOException
 import kotlin.coroutines.EmptyCoroutineContext
-import kotlin.test.assertEquals
-import kotlin.test.assertFalse
-import kotlin.test.assertTrue
+import kotlin.test.*
 import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 
@@ -106,11 +105,17 @@ class ByteChannelRequestBodyTest {
             override val contentLength: Long = content.size.toLong()
             override fun readFrom(): SdkByteReadChannel = chan
         }
+
         val job = launch(Dispatchers.IO) {
             val callContext = coroutineContext + Job(coroutineContext.job)
             val actual = ByteChannelRequestBody(body, callContext)
             val buffer = Buffer()
-            actual.writeTo(buffer)
+            // see https://github.com/awslabs/aws-sdk-kotlin/issues/733 for why we expect
+            // this to be an IOException
+            val ex = assertFailsWith<IOException> {
+                actual.writeTo(buffer)
+            }
+            assertIs<CancellationException>(ex.cause)
         }
         delay(100.milliseconds)
 

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

@@ -165,4 +165,49 @@ class UploadTest : AbstractEngineTest() {
             assertEquals("application/xml", reqContentType, "No Content-Type set on ${client.engine}")
         }
     }
+
+    /**
+     * Test cancelling the request in the middle of processing the request body
+     * This recreates [aws-sdk-kotlin#733](https://github.com/awslabs/aws-sdk-kotlin/issues/733)
+     * but does not result in a test failure (instead you will see
+     * 'Exception in thread "OkHttp Dispatcher" ... ' in the test output if not handled).
+     *
+     * If we exposed configuring the `Dispatcher` on OkHttpEngine builder then we could
+     * plug in a custom executor/thread factory to make the actual assertions (in a JVM test sourceSet).
+     * Although that would only work so long as okhttp continues to use `execute` rather than `submit` internally.
+     */
+    @OptIn(DelicateCoroutinesApi::class)
+    @Test
+    fun testUploadCancellation() = testEngines {
+        test { env, client ->
+            val chan = SdkByteChannel(autoFlush = true)
+
+            val writeJob = GlobalScope.launch {
+                val seedData = ByteArray(6017) { it.toByte() }
+                chan.writeFully(seedData)
+            }
+
+            val req = HttpRequest {
+                method = HttpMethod.POST
+                testSetup(env)
+                url.path = "/upload/content"
+                body = chan.toHttpBody(16 * 1024 * 1024)
+            }
+
+            val job = GlobalScope.launch {
+                val call = client.call(req)
+                try {
+                    // wait for seed data to have been written
+                    writeJob.join()
+                    delay(5000)
+                } finally {
+                    call.complete()
+                    chan.cancel(null)
+                }
+            }
+
+            writeJob.join()
+            job.cancelAndJoin()
+        }
+    }
 }