From 38d34098fd33856f5b4c5f9e81bc5cb89ae3c4e3 Mon Sep 17 00:00:00 2001 From: Brandon Moore <16313090+thatcube@users.noreply.github.com> Date: Sun, 18 Oct 2026 10:32:33 -0400 Subject: [PATCH 1/2] fix(hls): inherit isolation for persistent body reads Keep the non-Sendable chunked framing on the cancellation handler caller's isolation rather than sending it to the generic executor. Add chunked read coverage through cancellation handlers. Copilot-Session: e3687c50-89d2-4a1a-86ea-dc2326878fc6 --- CHANGELOG.md | 7 ++++- .../HLSIngest/PersistentHTTP1Transport.swift | 5 +++- .../Issue732PersistentHTTP1Tests.swift | 29 +++++++++++++++++++ docs/architecture.md | 6 ++++ 4 files changed, 45 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 12d83a9d..ef7c9b3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,12 @@ the public-API contract. ## [Unreleased] -_Nothing yet._ +### Fixed + +- Keep persistent HTTP/1.1 body framing in the caller's isolation while reading + chunks. Swift 6.4 can otherwise reject the cancellation-handler read with + "sending 'current' risks causing data races"; the chunked decoder remains + single-consumer, without unchecked Sendable conformance or an executor hop. ## [7.40.0] - 2026-10-18 diff --git a/Sources/AetherEngine/IO/HLSIngest/PersistentHTTP1Transport.swift b/Sources/AetherEngine/IO/HLSIngest/PersistentHTTP1Transport.swift index 4b1a6dea..591ace02 100644 --- a/Sources/AetherEngine/IO/HLSIngest/PersistentHTTP1Transport.swift +++ b/Sources/AetherEngine/IO/HLSIngest/PersistentHTTP1Transport.swift @@ -574,7 +574,10 @@ final class PersistentHTTP1Connection: @unchecked Sendable { /// The next piece of a body under `framing`, nil at its end, plus the framing after it. No /// cancellation handler: the caller decides whether an abandoned body is drained or closed. - func readChunk(_ framing: Framing, progress: () -> String) async throws -> (Data?, Framing) { + func readChunk( + _ framing: Framing, progress: () -> String, + isolation: isolated (any Actor)? = #isolation + ) async throws -> (Data?, Framing) { guard let task = stateLock.withLock({ retired ? nil : task }) else { throw CancellationError() } while true { switch framing { diff --git a/Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift b/Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift index 5f79be76..2e207f3d 100644 --- a/Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift +++ b/Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift @@ -386,6 +386,35 @@ final class Issue732PersistentHTTP1Tests: XCTestCase { XCTAssertEqual(server.connectionCount, 1) } + func testChunkedReadsKeepFramingAcrossCancellationHandlers() async throws { + let server = try XCTUnwrap(KeepAliveOrigin()) + defer { server.stop() } + let expected = Data((0..<5000).map { UInt8($0 & 0xFF) }) + server.route = { _ in .init(body: expected, chunked: true) } + let url = server.url("/chunked") + let origin = try XCTUnwrap(PersistentHTTP1Transport.Origin(url)) + let connection = PersistentHTTP1Connection(origin: origin, label: "isolation") + defer { connection.retire(reason: "test finished") } + let head = try await connection.sendRequest(target: url, headers: [:]) + var framing = PersistentHTTP1Connection.framing(for: head) + var received = Data() + + while true { + let current = framing + let (chunk, advanced) = try await withTaskCancellationHandler { + try await connection.readChunk(current, progress: { "test body" }) + } onCancel: { + connection.retire(reason: "test cancelled") + } + framing = advanced + guard let chunk else { break } + received.append(chunk) + } + + XCTAssertEqual(received, expected) + XCTAssertNil(PersistentHTTP1Connection.reuseRefusal(head, framing)) + } + func testRefusalKeepsTheConnectionAndReportsItsStatus() async throws { let server = try XCTUnwrap(KeepAliveOrigin()) defer { server.stop() } diff --git a/docs/architecture.md b/docs/architecture.md index 45d8c0ab..41d46733 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -358,6 +358,12 @@ at a time, until a 120 s time limit fired at 151 s because its watchdog is a tas pool. `LIBDISPATCH_COOPERATIVE_POOL_STRICT=1` narrows the pool to one thread and reproduces that on any Mac; the full suite has to pass under it. +Persistent HTTP/1.1 body reads inherit the caller's isolation through +`PersistentHTTP1Connection.readChunk`'s isolated parameter. Its framing can own a +mutable chunked-body decoder, so a cancellation-handler read must not send that +single-consumer state to a different executor. Cancellation still abandons the +body through its existing drain-or-close path. + ## Subtitle recognition scheduling Bitmap subtitle OCR awaits each image's recognition on a dedicated utility thread. From bc85e609b0cbacdfc42cf001e5dfa0c8ecd14db6 Mon Sep 17 00:00:00 2001 From: Brandon Moore <16313090+thatcube@users.noreply.github.com> Date: Sun, 18 Oct 2026 10:38:48 -0400 Subject: [PATCH 2/2] fix(hls): preserve isolation through socket progress reads Keep the progress closure in the same isolation through nextBytes as well; exercise a progress callback that reads the consumer's accumulated bytes. Copilot-Session: e3687c50-89d2-4a1a-86ea-dc2326878fc6 --- .../AetherEngine/IO/HLSIngest/PersistentHTTP1Transport.swift | 5 ++++- Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/Sources/AetherEngine/IO/HLSIngest/PersistentHTTP1Transport.swift b/Sources/AetherEngine/IO/HLSIngest/PersistentHTTP1Transport.swift index 591ace02..65b58f83 100644 --- a/Sources/AetherEngine/IO/HLSIngest/PersistentHTTP1Transport.swift +++ b/Sources/AetherEngine/IO/HLSIngest/PersistentHTTP1Transport.swift @@ -734,7 +734,10 @@ final class PersistentHTTP1Connection: @unchecked Sendable { } /// Bytes left over from the head read first, then the socket. nil is EOF. - private func nextBytes(_ task: URLSessionStreamTask, upTo limit: Int, progress: () -> String) async throws -> Data? { + private func nextBytes( + _ task: URLSessionStreamTask, upTo limit: Int, progress: () -> String, + isolation: isolated (any Actor)? = #isolation + ) async throws -> Data? { let buffered = stateLock.withLock { () -> Data in let slice = pending.prefix(limit) pending.removeFirst(slice.count) diff --git a/Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift b/Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift index 2e207f3d..757dc74e 100644 --- a/Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift +++ b/Tests/AetherEngineTests/Issue732PersistentHTTP1Tests.swift @@ -402,7 +402,7 @@ final class Issue732PersistentHTTP1Tests: XCTestCase { while true { let current = framing let (chunk, advanced) = try await withTaskCancellationHandler { - try await connection.readChunk(current, progress: { "test body" }) + try await connection.readChunk(current, progress: { "\(received.count) bytes" }) } onCancel: { connection.retire(reason: "test cancelled") }