fix: Stream error handling (#743)

Author: jbelkins

Sources/ClientRuntime/Networking/Streaming/BufferedStream.swift

@@ -13,7 +13,7 @@ import class Foundation.NSRecursiveLock
 /// Note: This class is thread-safe and async-safe.
 /// Note: if data is not read from the stream, the buffer will grow indefinitely until the stream is closed.
 ///       or reach the maximum size of a `Data` object.
-public class BufferedStream: Stream {
+public class BufferedStream: Stream, @unchecked Sendable {
 
     /// Returns the cumulative length of all data so far written to the stream, if known.
     /// For a buffered stream, the length will only be known if the stream has closed.
@@ -130,6 +130,13 @@ public class BufferedStream: Stream {
 
     /// Call this function only while `lock` is locked, to prevent simultaneous access.
     private func _read(upToCount count: Int) throws -> Data? {
+
+        // throw any previously stored error, if there was one
+        // dispose of the error when throwing so it is only thrown once
+        if let error = _error {
+            _error = nil
+            throw error
+        }
         let toRead = min(count, _buffer.count)
         let endPosition = position.advanced(by: toRead)
         let chunk = _buffer[position..<endPosition]
@@ -143,7 +150,6 @@ public class BufferedStream: Stream {
         // if we're closed and there's no data left, return nil
         // this will signal the end of the stream
         if _isClosed && chunk.isEmpty == true {
-            if let error = _error { throw error }
             return nil
         }
 

Tests/ClientRuntimeTests/NetworkingTests/Streaming/BufferedStreamTests.swift

@@ -69,6 +69,20 @@ final class BufferedStreamTests: XCTestCase {
         XCTAssertNil(readData2)
     }
 
+    func test_read_throwsErrorWhenStreamIsClosedWithError() throws {
+        let subject = BufferedStream()
+        try subject.write(contentsOf: testData)
+        subject.closeWithError(TestError.error)
+        do {
+            let readData1 = try subject.read(upToCount: Int.max)
+            XCTFail("Error was expected to be thrown")
+        } catch TestError.error {
+            // Test passes
+        } catch {
+            XCTFail("Unexpected error thrown: \(error.localizedDescription)")
+        }
+    }
+
     // MARK: - readToEnd()
 
     func test_readToEnd_readsToEnd() throws {
@@ -83,6 +97,20 @@ final class BufferedStreamTests: XCTestCase {
         XCTAssertNil(readData2)
     }
 
+    func test_readToEnd_throwsErrorWhenStreamIsClosedWithError() throws {
+        let subject = BufferedStream()
+        try subject.write(contentsOf: testData)
+        subject.closeWithError(TestError.error)
+        do {
+            let readData1 = try subject.readToEnd()
+            XCTFail("Error was expected to be thrown")
+        } catch TestError.error {
+            // Test passes
+        } catch {
+            XCTFail("Unexpected error thrown: \(error.localizedDescription)")
+        }
+    }
+
     // MARK: - readToEndAsync()
 
     func test_readToEndAsync_readsToEnd() async throws {
@@ -97,6 +125,20 @@ final class BufferedStreamTests: XCTestCase {
         XCTAssertNil(readData2)
     }
 
+    func test_readToEndAsync_throwsErrorWhenStreamIsClosedWithError() async throws {
+        let subject = BufferedStream()
+        try subject.write(contentsOf: testData)
+        subject.closeWithError(TestError.error)
+        do {
+            let readData1 = try await subject.readToEndAsync()
+            XCTFail("Error was expected to be thrown")
+        } catch TestError.error {
+            // Test passes
+        } catch {
+            XCTFail("Unexpected error thrown: \(error.localizedDescription)")
+        }
+    }
+
     // MARK: - readAsync(upToCount:)
 
     func test_readAsync_readsAsynchronously() async throws {
@@ -142,6 +184,20 @@ final class BufferedStreamTests: XCTestCase {
         XCTAssertNil(readData3)
     }
 
+    func test_readAsync_throwsErrorWhenStreamIsClosedWithError() async throws {
+        let subject = BufferedStream()
+        try subject.write(contentsOf: testData)
+        subject.closeWithError(TestError.error)
+        do {
+            let readData1 = try await subject.readAsync(upToCount: Int.max)
+            XCTFail("Error was expected to be thrown")
+        } catch TestError.error {
+            // Test passes
+        } catch {
+            XCTFail("Unexpected error thrown: \(error.localizedDescription)")
+        }
+    }
+
     // MARK: - write(contentsOf:)
 
     func test_write_appendsWrittenDataToBuffer() throws {
@@ -189,3 +245,7 @@ final class BufferedStreamTests: XCTestCase {
         XCTAssertEqual(sut.length, testData.count)
     }
 }
+
+private enum TestError: Error, Equatable {
+    case error
+}