KTOR-9561 Rethrow closedCause after copyTo

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

@@ -162,6 +162,7 @@ public suspend fun ByteReadChannel.copyTo(channel: ByteWriteChannel): Long {
         channel.flush()
     }
 
+    rethrowCloseCauseIfNeeded()
     return result
 }
 
@@ -216,6 +217,7 @@ public suspend fun ByteReadChannel.copyTo(channel: ByteWriteChannel, limit: Long
         channel.flush()
     }
 
+    rethrowCloseCauseIfNeeded()
     return limit - remaining
 }
 

ktor-io/common/test/ByteReadChannelOperationsTest.kt

@@ -115,6 +115,42 @@ class ByteReadChannelOperationsTest {
         }
     }
 
+    @Test
+    fun `copyTo propagates closedCause from cancelled source`() = runTest {
+        val src = ByteChannel()
+        val dst = ByteChannel()
+        src.writeFully(byteArrayOf(1, 2, 3))
+        src.flush()
+        src.cancel(IOException("source cancelled"))
+        assertFailsWith<IOException> {
+            src.copyTo(dst)
+        }
+        assertTrue(src.isClosedForRead)
+    }
+
+    @Test
+    fun `copyTo with limit propagates closedCause from cancelled source`() = runTest {
+        val src = ByteChannel()
+        val dst = ByteChannel()
+        src.writeFully(byteArrayOf(1, 2, 3))
+        src.flush()
+        src.cancel(IOException("source cancelled"))
+        assertFailsWith<IOException> {
+            src.copyTo(dst, limit = 1024L)
+        }
+        assertTrue(src.isClosedForRead)
+    }
+
+    @Test
+    fun `copyTo does not throw on normal close`() = runTest {
+        val src = ByteChannel()
+        val dst = ByteChannel()
+        src.writeFully(byteArrayOf(1, 2, 3))
+        src.flushAndClose()
+        val copied = src.copyTo(dst)
+        assertEquals(3, copied)
+    }
+
     @Test
     fun readFully() = runTest {
         val expected = ByteArray(10) { it.toByte() }