From 79ff299d87c0626d63cb25e7dde06a7df46eae83 Mon Sep 17 00:00:00 2001 From: Roman Sorochak Date: Tue, 13 Oct 2026 23:48:20 +0200 Subject: [PATCH 1/2] Retry when the reconnect's header request fails After a dropped stream, retryOnError() reopens the source through seek(at:) -> performOpen -> initialRequest, a bytes=0-1 request whose callback only handled .response. If that request failed - the server or network still down a second after the drop - the .stream(.failure) event fell through `default: break`: no stream request was started and no retry was scheduled, so playback stayed in buffering until the user acted. The Retrier itself is idle at that point (its callback's seek already cancelled it); nothing called it again. Route the initial request's failure through handleFailedStreamEvent, the same path the stream's own failures take, so the source keeps retrying (or waits for the network) until a reconnect succeeds. The new RemoteAudioSourceRetryTests drives a loopback server that cuts the stream mid-body and refuses connections for 1.5 s after it; without this change no audio arrives from a later connection and the test times out. Co-Authored-By: Claude Opus 5.5 (cherry picked from commit 8942f0d76819ebaf99d533886287040272f1c763) --- .../Audio Source/RemoteAudioSource.swift | 5 + .../RemoteAudioSourceRetryTests.swift | 150 ++++++++++++++++++ 2 files changed, 155 insertions(+) create mode 100644 AudioStreamingTests/Streaming/Audio Source/RemoteAudioSourceRetryTests.swift diff --git a/AudioStreaming/Streaming/Audio Source/RemoteAudioSource.swift b/AudioStreaming/Streaming/Audio Source/RemoteAudioSource.swift index 522a6c2..d805229 100644 --- a/AudioStreaming/Streaming/Audio Source/RemoteAudioSource.swift +++ b/AudioStreaming/Streaming/Audio Source/RemoteAudioSource.swift @@ -229,6 +229,11 @@ public class RemoteAudioSource: AudioStreamSource { self?.parseResponseHeader(response: urlResponse) task.cancel() completion() + case let .stream(.failure(error)): + // This request is also the first step of every reconnect (`retryOnError` → `seek`), + // so its failure has to go through the retry path like the stream's own failure; + // dropping it leaves the source with no request and no pending retry. + self?.handleFailedStreamEvent(error: error) default: break } diff --git a/AudioStreamingTests/Streaming/Audio Source/RemoteAudioSourceRetryTests.swift b/AudioStreamingTests/Streaming/Audio Source/RemoteAudioSourceRetryTests.swift new file mode 100644 index 0000000..24ea071 --- /dev/null +++ b/AudioStreamingTests/Streaming/Audio Source/RemoteAudioSourceRetryTests.swift @@ -0,0 +1,150 @@ +import Network +import XCTest +@testable import AudioStreaming + +final class RemoteAudioSourceRetryTests: XCTestCase { + private var server: DroppingStreamServer! + private var source: RemoteAudioSource! + private var delegate: SourceDelegateSpy! + + override func setUpWithError() throws { + try super.setUpWithError() + server = try DroppingStreamServer(refuseWindow: 1.5) + try server.start() + XCTAssertGreaterThan(server.port, 0, "server failed to bind a port") + } + + override func tearDownWithError() throws { + source?.close() + source = nil + server?.stop() + server = nil + try super.tearDownWithError() + } + + /// A stream drops mid-body and the first reconnect fails too: the source has to keep retrying + /// until a later reconnect succeeds, and deliver that connection's audio. + func test_Source_Keeps_Retrying_When_The_First_Reconnect_Fails() throws { + let resumed = expectation(description: "audio from a reconnect after the failed one") + delegate = SourceDelegateSpy(onData: { data in + if data.contains(DroppingStreamServer.laterStreamByte) { resumed.fulfill() } + }) + let url = try XCTUnwrap(URL(string: "http://127.0.0.1:\(server.port)/live.mp3")) + source = RemoteAudioSource(networking: NetworkingClient(), + metadataStreamSource: MetadataStreamProcessor(parser: MetadataParser().eraseToAnyParser()), + icycastHeadersProcessor: IcycastHeadersProcessor(), + netStatusProvider: ConnectedNetStatusProvider(), + retrier: Retrier(interval: .seconds(1), maxInterval: 5, underlyingQueue: nil), + url: url, + httpMethod: nil, + httpBody: nil, + underlyingQueue: DispatchQueue(label: "remote.audio.source.retry.tests"), + httpHeaders: [:]) + source.delegate = delegate + + source.seek(at: 0) + + wait(for: [resumed], timeout: 10) + XCTAssertGreaterThan(server.refusedConnections, 0, "the first reconnect was expected to fail") + } +} + +// MARK: - Helpers + +private final class ConnectedNetStatusProvider: NetStatusProvider { + var isConnected: Bool { true } + var connectionType: NetConnectionType { .wifi(connected: true) } + func start(connectionChange _: @escaping (NetConnectionType) -> Void) {} + func stop() {} +} + +private final class SourceDelegateSpy: AudioStreamSourceDelegate { + private let onData: (Data) -> Void + + init(onData: @escaping (Data) -> Void) { + self.onData = onData + } + + func dataAvailable(source _: CoreAudioStreamSource, data: Data) { onData(data) } + func errorOccurred(source _: CoreAudioStreamSource, error _: Error) {} + func endOfFileOccurred(source _: CoreAudioStreamSource) {} + func metadataReceived(data _: [String: String]) {} +} + +/// A loopback HTTP server that behaves like a live stream on a flaky network: the first stream +/// connection sends some audio and is cut mid-body, every connection in the `refuseWindow` after +/// that is closed before a response, and connections after the window are served again. +/// The first stream's body is filled with `firstStreamByte`, every later one with `laterStreamByte`. +private final class DroppingStreamServer { + static let firstStreamByte: UInt8 = 0x01 + static let laterStreamByte: UInt8 = 0x02 + + private let listener: NWListener + private let queue = DispatchQueue(label: "dropping.stream.server") + private let refuseWindow: TimeInterval + private var droppedAt: Date? + private var connections: [NWConnection] = [] + private(set) var port: UInt16 = 0 + private var _refusedConnections = 0 + var refusedConnections: Int { queue.sync { _refusedConnections } } + + init(refuseWindow: TimeInterval) throws { + self.refuseWindow = refuseWindow + let params = NWParameters.tcp + params.allowLocalEndpointReuse = true + listener = try NWListener(using: params, on: .any) + } + + func start() throws { + let ready = DispatchSemaphore(value: 0) + listener.stateUpdateHandler = { state in + if case .ready = state { ready.signal() } + } + listener.newConnectionHandler = { [weak self] connection in + self?.handle(connection) + } + listener.start(queue: queue) + _ = ready.wait(timeout: .now() + 5) + port = listener.port?.rawValue ?? 0 + } + + func stop() { + listener.cancel() + queue.sync { + connections.forEach { $0.cancel() } + connections.removeAll() + } + } + + private func handle(_ connection: NWConnection) { + connections.append(connection) + connection.start(queue: queue) + connection.receive(minimumIncompleteLength: 1, maximumLength: 65536) { [weak self, weak connection] data, _, _, _ in + guard let self, let connection else { return } + let request = data.flatMap { String(data: $0, encoding: .utf8) } ?? "" + self.respond(to: request, on: connection) + } + } + + private func respond(to request: String, on connection: NWConnection) { + if let droppedAt, Date().timeIntervalSince(droppedAt) < refuseWindow { + _refusedConnections += 1 + connection.cancel() + return + } + let isRangeProbe = request.range(of: "\r\nRange:", options: .caseInsensitive) != nil + let isFirstStream = droppedAt == nil && !isRangeProbe + let header = "HTTP/1.1 200 OK\r\n" + + "Content-Type: audio/mpeg\r\n" + + "Content-Length: 100000000\r\n\r\n" + var response = Data(header.utf8) + let fill = droppedAt == nil ? Self.firstStreamByte : Self.laterStreamByte + response.append(Data(repeating: fill, count: 4096)) + connection.send(content: response, completion: .contentProcessed { [weak self, weak connection] _ in + guard isFirstStream, let self, let connection else { return } + // Closing well short of Content-Length is a dropped connection to URLSession. + self.droppedAt = Date() + connection.cancel() + }) + } +} From 9e0a33412abf5ee1b088581ac796e61b1cf386eb Mon Sep 17 00:00:00 2001 From: Roman Sorochak Date: Wed, 14 Oct 2026 01:22:24 +0200 Subject: [PATCH 2/2] Shorten the initial-request failure comment --- AudioStreaming/Streaming/Audio Source/RemoteAudioSource.swift | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/AudioStreaming/Streaming/Audio Source/RemoteAudioSource.swift b/AudioStreaming/Streaming/Audio Source/RemoteAudioSource.swift index d805229..774d916 100644 --- a/AudioStreaming/Streaming/Audio Source/RemoteAudioSource.swift +++ b/AudioStreaming/Streaming/Audio Source/RemoteAudioSource.swift @@ -230,9 +230,7 @@ public class RemoteAudioSource: AudioStreamSource { task.cancel() completion() case let .stream(.failure(error)): - // This request is also the first step of every reconnect (`retryOnError` → `seek`), - // so its failure has to go through the retry path like the stream's own failure; - // dropping it leaves the source with no request and no pending retry. + // Every reconnect starts with this request, so its failure must reach the retry path. self?.handleFailedStreamEvent(error: error) default: break