Skip to content

Commit

Permalink
Tolerate both CancellationError and URLError in CancellationTests (#45)
Browse files Browse the repository at this point in the history
### Motivation

When cancelling a Swift concurrency task during a streaming request then
the error returned might be `URLError` with `.cancelled` code or
`CancellationError`. The tests tried to be smart and expect just one of
these depending on which stage of the request we were at, but there are
still some races, and this test fails very rarely because a
`CancellationError` was thrown instead of a `URLError`.

### Modifications

This patch updates the test to tolerate both kinds of error at this
stage of the request.

### Result

The test will pass if the error is either `URLError` with `.cancelled`
or `CancellationError`, and continue to fail if there is any other kind
of error or no error.

### Test Plan

Existing tests, which failed when run repeatedly for 1k runs, now pass
when run for 10k runs.
  • Loading branch information
simonjbeaumont authored Jan 16, 2024
1 parent aac0a82 commit 3d1f6e7
Show file tree
Hide file tree
Showing 3 changed files with 19 additions and 16 deletions.
2 changes: 1 addition & 1 deletion Tests/OpenAPIURLSessionTests/NIOAsyncHTTP1TestServer.swift
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ final class AsyncTestHTTP1Server {
for try await connectionChannel in inbound {
group.addTask {
do {
debug("Sevrer handling new connection")
debug("Server handling new connection")
try await connectionHandler(connectionChannel)
debug("Server done handling connection")
} catch { debug("Server error handling connection: \(error)") }
Expand Down
8 changes: 4 additions & 4 deletions Tests/OpenAPIURLSessionTests/TaskCancellationTests.swift
Original file line number Diff line number Diff line change
Expand Up @@ -150,11 +150,11 @@ func testTaskCancelled(_ cancellationPoint: CancellationPoint, transport: URLSes
await XCTAssertThrowsError(try await task.value) { error in XCTAssertTrue(error is CancellationError) }
case .beforeSendingRequestBody, .partwayThroughSendingRequestBody:
await XCTAssertThrowsError(try await task.value) { error in
guard let urlError = error as? URLError else {
XCTFail()
return
switch error {
case is CancellationError: break
case is URLError: XCTAssertEqual((error as! URLError).code, .cancelled)
default: XCTFail("Unexpected error: \(error)")
}
XCTAssertEqual(urlError.code, .cancelled)
}
case .beforeConsumingResponseBody, .partwayThroughConsumingResponseBody, .afterConsumingResponseBody:
try await task.value
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,44 +20,47 @@ struct MockAsyncSequence<Element>: AsyncSequence, Sendable where Element: Sendab
var elementsToVend: [Element]
private let _elementsVended: LockedValueBox<[Element]>
var elementsVended: [Element] { _elementsVended.withValue { $0 } }
private let semaphore: DispatchSemaphore?
private let gateOpeningsStream: AsyncStream<Void>
private let gateOpeningsContinuation: AsyncStream<Void>.Continuation

init(elementsToVend: [Element], gatingProduction: Bool) {
self.elementsToVend = elementsToVend
self._elementsVended = LockedValueBox([])
self.semaphore = gatingProduction ? DispatchSemaphore(value: 0) : nil
(self.gateOpeningsStream, self.gateOpeningsContinuation) = AsyncStream.makeStream(of: Void.self)
if !gatingProduction { openGate() }
}

func openGate(for count: Int) { for _ in 0..<count { semaphore?.signal() } }
func openGate(for count: Int) { for _ in 0..<count { self.gateOpeningsContinuation.yield() } }

func openGate() {
openGate(for: elementsToVend.count + 1) // + 1 for the nil
}

func makeAsyncIterator() -> AsyncIterator {
AsyncIterator(elementsToVend: elementsToVend[...], semaphore: semaphore, elementsVended: _elementsVended)
AsyncIterator(
elementsToVend: elementsToVend[...],
gateOpenings: gateOpeningsStream.makeAsyncIterator(),
elementsVended: _elementsVended
)
}

final class AsyncIterator: AsyncIteratorProtocol {
var elementsToVend: ArraySlice<Element>
var semaphore: DispatchSemaphore?
var gateOpenings: AsyncStream<Void>.Iterator
var elementsVended: LockedValueBox<[Element]>

init(
elementsToVend: ArraySlice<Element>,
semaphore: DispatchSemaphore?,
gateOpenings: AsyncStream<Void>.Iterator,
elementsVended: LockedValueBox<[Element]>
) {
self.elementsToVend = elementsToVend
self.semaphore = semaphore
self.gateOpenings = gateOpenings
self.elementsVended = elementsVended
}

func next() async throws -> Element? {
await withCheckedContinuation { continuation in
semaphore?.wait()
continuation.resume()
}
guard await gateOpenings.next() != nil else { throw CancellationError() }
guard let element = elementsToVend.popFirst() else { return nil }
elementsVended.withValue { $0.append(element) }
return element
Expand Down

0 comments on commit 3d1f6e7

Please sign in to comment.