From 15f48c9c65f3ae152eceb3e49785fd18b54f00e5 Mon Sep 17 00:00:00 2001 From: weishu Date: Mon, 17 Aug 2026 15:28:14 +0800 Subject: [PATCH] feat(ios): SSE client + reconnect state machine (A-M1c) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HapiClient/SSE per docs/api/client-contract/sse.md and the web reference (web/src/hooks/useSSE.ts): - SSELineParser: incremental frame parser (LF/CRLF, multi-line data, comments, retry ignored); per-block ids only — sticky-cursor semantics live in the caller so id-less heartbeats can never blank the cursor. - ReconnectPolicy + SSETimings: 0s first retry, 1s x2 capped at 30s, 300s slow ceiling after 8 attempts, 0-500ms injectable jitter; connect timeout 10s, staleness 90s, foreground-resume 45s, watchdog tick 10s. - SSETransport protocol + URLSessionSSETransport (URLSession.bytes with dedicated config: 300s idle timeout, 7d resource, no waitsForConnectivity). - actor SSEClient: idle/connecting/connected/backoff/suspended machine; handshake-gated connected state surfacing resume ok|gap; per-subscription cursor sent as ?lastEventId, advanced only after consumer yield (at-least-once); one free token-refresh retry per connect cycle on 401; suspend defers retries, resume applies the 45s staleness check; acceptEncodingIdentity escape hatch (gzip streaming verification TODO). - NetworkPathObserving + NWPathObserver: path change while connected is treated as a transport error (immediate reconnect). - swift-testing suite on a fake transport + manual clock: parser edge cases, exact backoff schedule + seeded jitter bounds, handshake gating, ok/gap verdicts, cursor replay + isolation, heartbeat vs watchdog, 90s staleness, connect timeout, suspend/resume, event ordering, unknown-type passthrough, 401 bypass cap, path-change reconnect. --- ios/Packages/HapiKit/Package.swift | 3 +- .../HapiClient/SSE/NetworkPathMonitor.swift | 59 ++ .../HapiClient/SSE/ReconnectPolicy.swift | 76 +++ .../Sources/HapiClient/SSE/SSEClient.swift | 528 ++++++++++++++++++ .../Sources/HapiClient/SSE/SSEClock.swift | 33 ++ .../HapiClient/SSE/SSELineParser.swift | 115 ++++ .../Sources/HapiClient/SSE/SSETransport.swift | 93 +++ .../SSE/ReconnectPolicyTests.swift | 93 +++ .../HapiClientTests/SSE/SSEClientTests.swift | 419 ++++++++++++++ .../SSE/SSELineParserTests.swift | 114 ++++ .../HapiClientTests/SSE/SSETestSupport.swift | 332 +++++++++++ ios/README.md | 19 +- 12 files changed, 1881 insertions(+), 3 deletions(-) create mode 100644 ios/Packages/HapiKit/Sources/HapiClient/SSE/NetworkPathMonitor.swift create mode 100644 ios/Packages/HapiKit/Sources/HapiClient/SSE/ReconnectPolicy.swift create mode 100644 ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClient.swift create mode 100644 ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClock.swift create mode 100644 ios/Packages/HapiKit/Sources/HapiClient/SSE/SSELineParser.swift create mode 100644 ios/Packages/HapiKit/Sources/HapiClient/SSE/SSETransport.swift create mode 100644 ios/Packages/HapiKit/Tests/HapiClientTests/SSE/ReconnectPolicyTests.swift create mode 100644 ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSEClientTests.swift create mode 100644 ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSELineParserTests.swift create mode 100644 ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSETestSupport.swift diff --git a/ios/Packages/HapiKit/Package.swift b/ios/Packages/HapiKit/Package.swift index 781247fc..bf63d023 100644 --- a/ios/Packages/HapiKit/Package.swift +++ b/ios/Packages/HapiKit/Package.swift @@ -21,6 +21,7 @@ let package = Package( // Transport layer: APIClient, auth, SSE, stores (lands in M1). .target(name: "HapiClient", dependencies: ["HapiProtocol"]), .testTarget(name: "HapiProtocolTests", dependencies: ["HapiProtocol"]), - .testTarget(name: "HapiClientTests", dependencies: ["HapiClient"]), + // HapiProtocol is named directly by SSE tests (SyncEvent assertions). + .testTarget(name: "HapiClientTests", dependencies: ["HapiClient", "HapiProtocol"]), ] ) diff --git a/ios/Packages/HapiKit/Sources/HapiClient/SSE/NetworkPathMonitor.swift b/ios/Packages/HapiKit/Sources/HapiClient/SSE/NetworkPathMonitor.swift new file mode 100644 index 00000000..0b75b096 --- /dev/null +++ b/ios/Packages/HapiKit/Sources/HapiClient/SSE/NetworkPathMonitor.swift @@ -0,0 +1,59 @@ +import Foundation +#if canImport(Network) +import Network +#endif + +/// A snapshot of the device's network path. +public struct NetworkPathUpdate: Equatable, Sendable { + /// Whether the path can carry traffic (`NWPath.Status.satisfied`). + public var isSatisfied: Bool + /// Cellular / personal-hotspot style paths (informational). + public var isExpensive: Bool + + public init(isSatisfied: Bool, isExpensive: Bool = false) { + self.isSatisfied = isSatisfied + self.isExpensive = isExpensive + } +} + +/// Source of network-path change notifications for `SSEClient`. +/// +/// The stream's FIRST element is the baseline path reported on subscription +/// (NWPathMonitor always fires once immediately); every subsequent element is +/// an actual change. `SSEClient` skips the baseline and treats any later +/// update while connected as a transport error — the old socket is almost +/// certainly routed over a path that no longer exists, and reconnecting +/// immediately beats waiting for the 90 s staleness watchdog. +public protocol NetworkPathObserving: Sendable { + func pathUpdates() -> AsyncStream +} + +#if canImport(Network) +/// Production observer backed by `NWPathMonitor`. +public struct NWPathObserver: NetworkPathObserving { + /// `NWPathMonitor` is not Sendable; it is confined to its own dispatch + /// queue and only ever touched from the update handler / termination + /// callback, so boxing it is safe. + private final class MonitorBox: @unchecked Sendable { + let monitor = NWPathMonitor() + } + + public init() {} + + public func pathUpdates() -> AsyncStream { + AsyncStream { continuation in + let box = MonitorBox() + box.monitor.pathUpdateHandler = { path in + continuation.yield(NetworkPathUpdate( + isSatisfied: path.status == .satisfied, + isExpensive: path.isExpensive + )) + } + box.monitor.start(queue: DispatchQueue(label: "run.hapi.sse.path-monitor")) + continuation.onTermination = { _ in + box.monitor.cancel() + } + } + } +} +#endif diff --git a/ios/Packages/HapiKit/Sources/HapiClient/SSE/ReconnectPolicy.swift b/ios/Packages/HapiKit/Sources/HapiClient/SSE/ReconnectPolicy.swift new file mode 100644 index 00000000..d9b9c353 --- /dev/null +++ b/ios/Packages/HapiKit/Sources/HapiClient/SSE/ReconnectPolicy.swift @@ -0,0 +1,76 @@ +import Foundation + +/// Normative SSE timing constants, in milliseconds. +/// +/// Values come from `docs/api/client-contract/sse.md` ("Reconnect policy") +/// and the web reference client (`web/src/hooks/useSSE.ts`). +public enum SSETimings { + /// The hub emits a heartbeat frame every 30 s (informational). + public static let heartbeatIntervalMs = 30_000 + /// No frames of any kind for 90 s ⇒ tear down and reconnect. + public static let stalenessThresholdMs = 90_000 + /// On app-foreground, reconnect immediately when the last frame is older + /// than this. One missed heartbeat interval after an OS suspend is + /// already enough to distrust the socket (no FIN/RST ever surfaces). + public static let foregroundResumeStalenessMs = 45_000 + /// Staleness check interval. Checks are skipped while suspended. + public static let watchdogTickMs = 10_000 + /// An attempt that has not reached OPEN within 10 s is likely hung on a + /// dead pooled socket — abandon it and retry on a fresh connection. + public static let connectTimeoutMs = 10_000 +} + +/// Pure backoff schedule for SSE reconnects. +/// +/// Delay for attempt `n` (0-based, reset to 0 on every successful open): +/// +/// - attempt 0 → 0 ms (the first retry is immediate; backoff is for +/// *repeated* failures, not the initial recovery), +/// - attempt n ≥ 1 → `min(cap, base × 2^(n-1))`, +/// - cap is 30 s, widening to 300 s once `n ≥ 8` (a hub that stays +/// unreachable is usually down for hours, and every retry through a relay +/// costs a TLS handshake), +/// - plus uniform jitter of 0–500 ms on every delay, immediate ones included. +/// +/// Jitter is injectable: the generic `delayMs(forAttempt:using:)` draws from +/// any `RandomNumberGenerator` so tests can seed it, and callers that manage +/// randomness themselves can add `exponentialDelayMs(forAttempt:)` + their +/// own draw from `jitterRangeMs` (this is what `SSEClient` does). +public struct ReconnectPolicy: Equatable, Sendable { + public var baseDelayMs: Int + public var maxDelayMs: Int + public var slowMaxDelayMs: Int + public var slowAfterAttempts: Int + public var jitterRangeMs: ClosedRange + + public init( + baseDelayMs: Int = 1_000, + maxDelayMs: Int = 30_000, + slowMaxDelayMs: Int = 300_000, + slowAfterAttempts: Int = 8, + jitterRangeMs: ClosedRange = 0...500 + ) { + self.baseDelayMs = baseDelayMs + self.maxDelayMs = maxDelayMs + self.slowMaxDelayMs = slowMaxDelayMs + self.slowAfterAttempts = slowAfterAttempts + self.jitterRangeMs = jitterRangeMs + } + + /// The deterministic part of the schedule, without jitter. + public func exponentialDelayMs(forAttempt attempt: Int) -> Int { + guard attempt >= 1 else { + return 0 + } + let cap = attempt >= slowAfterAttempts ? slowMaxDelayMs : maxDelayMs + // Clamp the exponent so the shift can never overflow; 2^30 × base is + // already far past every cap. + let exponent = min(attempt - 1, 30) + return min(cap, baseDelayMs << exponent) + } + + /// Full delay including a jitter draw from the supplied generator. + public func delayMs(forAttempt attempt: Int, using rng: inout R) -> Int { + exponentialDelayMs(forAttempt: attempt) + Int.random(in: jitterRangeMs, using: &rng) + } +} diff --git a/ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClient.swift b/ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClient.swift new file mode 100644 index 00000000..600952ed --- /dev/null +++ b/ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClient.swift @@ -0,0 +1,528 @@ +import Foundation +import HapiProtocol + +/// Which events this subscription receives (sse.md "Dual-subscription +/// model"). The reference client holds one `.global` client for the whole +/// app session plus one `.session` client per open chat. +public enum SSEScope: Equatable, Hashable, Sendable { + /// `all=true` — every event in the token's namespace. + case global + /// `sessionId=` — events for one session. + case session(String) +} + +public struct SSEClientConfiguration: Sendable { + /// Hub origin, e.g. `https://hub.example`. The path is always replaced + /// with `/api/events` (mirrors `new URL(path, baseUrl)` on the web). + public var baseUrl: URL + /// Called before every connection attempt; expected to return a fresh + /// JWT (refreshing when its cached one is stale) or `nil` when the + /// client cannot authenticate right now (the attempt fails into + /// backoff). Re-asked once, without counting a backoff attempt, when a + /// connect fails with 401. + public var tokenProvider: @Sendable () async -> String? + public var scope: SSEScope + /// Initial `?visibility=` value: `"visible"` or `"hidden"` (anything + /// else reads as hidden hub-side). Later changes go through + /// `POST /api/visibility` with the handshake's `subscriptionId`. + public var visibility: String + /// When true, sends `Accept-Encoding: identity` so the hub skips gzip. + /// + /// TODO(gzip): sse.md requires verifying on-device that URLSession + /// surfaces gzip-decoded bytes incrementally (one visible frame per + /// server flush) rather than buffering until EOF. Until that M1 + /// verification runs against a real hub, this flag is the escape hatch: + /// flip it to `true` if frames only arrive in bulk on disconnect. + public var acceptEncodingIdentity: Bool + public var reconnectPolicy: ReconnectPolicy + /// Jitter draw added to every backoff delay. Injectable so tests are + /// deterministic; defaults to uniform over the policy's jitter range. + public var jitterMs: @Sendable () -> Int + + public init( + baseUrl: URL, + tokenProvider: @escaping @Sendable () async -> String?, + scope: SSEScope, + visibility: String = "visible", + acceptEncodingIdentity: Bool = false, + reconnectPolicy: ReconnectPolicy = ReconnectPolicy(), + jitterMs: (@Sendable () -> Int)? = nil + ) { + self.baseUrl = baseUrl + self.tokenProvider = tokenProvider + self.scope = scope + self.visibility = visibility + self.acceptEncodingIdentity = acceptEncodingIdentity + self.reconnectPolicy = reconnectPolicy + let range = reconnectPolicy.jitterRangeMs + self.jitterMs = jitterMs ?? { Int.random(in: range) } + } +} + +public enum SSEConnectionState: Equatable, Sendable { + case idle + case connecting + /// Entered only on the first decoded + /// `connection-changed {status: "connected"}` frame — a 200 response + /// alone is not "up" (the hub's handshake is the source of truth). + case connected + /// Waiting `reconnectPolicy` delay before attempt `attempt` retries + /// (0-based; attempt 0 retries immediately, jitter only). + case backoff(attempt: Int) + /// Parked by `suspend()` with no live connection; retries are deferred + /// until `resume(...)`. + case suspended +} + +public enum SSEClientEvent: Equatable, Sendable { + /// The hub's subscribe handshake. `resume == .ok` ⇒ the replay that + /// follows contains every missed event, skip the REST resync; `.gap` + /// (including an absent/unknown wire verdict) ⇒ full refetch. + /// `subscriptionId` is needed for `POST /api/visibility` and is new on + /// every reconnect. + case handshake(resume: ResumeVerdict, subscriptionId: String?) + /// A decoded broadcast frame, in arrival order. Heartbeats are consumed + /// internally (watchdog) and never surface; unknown event types surface + /// as `SyncEvent.unknown` and never kill the stream. + case event(SyncEvent) + case stateChanged(SSEConnectionState) +} + +/// SSE subscription client: one actor instance per subscription +/// (scope + hub + namespace). Owns the whole reconnect state machine: +/// +/// idle → connecting → connected +/// ↑ │ transport error / staleness / path change +/// └─ backoff(n) ┘ (suspended parks between attempts) +/// +/// Behavior follows `docs/api/client-contract/sse.md` and the web reference +/// `web/src/hooks/useSSE.ts`: +/// +/// - Handshake-gated: `.connected` only after the first decoded +/// `connection-changed {status: "connected"}`; its `resume` verdict is +/// surfaced as `.handshake`. +/// - Sticky cursor: only non-empty frame `id`s move `lastEventId`; id-less +/// heartbeats can never blank it. The cursor is scoped to this instance — +/// a client for a different subscription must be a new instance, so a +/// cursor can never replay against the wrong filter set. +/// - At-least-once: the cursor advances only after the event was yielded to +/// the consumer stream. (`AsyncStream` buffers, so "yielded" means handed +/// to the consumer's buffer — consumers needing durable handling must +/// drain promptly and treat redelivery as idempotent.) +/// - Watchdog: 10 s ticks; connect attempts that have not opened within +/// 10 s are abandoned; 90 s without any bytes tears the connection down. +/// Checks are skipped while suspended. +/// - Backoff per `ReconnectPolicy`; the attempt counter resets on every +/// successful open, so the first retry after a healthy connection is +/// immediate. +/// - `suspend()`/`resume(staleThresholdCheck:)` for app lifecycle: retries +/// are deferred while suspended; on resume a surviving connection that +/// has been silent ≥ 45 s is torn down and rebuilt immediately. +/// - 401 connect failures re-ask the token provider once per connect cycle +/// without counting a backoff attempt (the provider refreshes); a second +/// consecutive 401 backs off normally. +/// +/// Lifecycle: `start()` once, consume the returned stream, `stop()` when +/// done. A stopped client is finished — build a new instance to resubscribe. +public actor SSEClient { + private enum AttemptExit: Equatable, Sendable { + case unauthorized + case failed + case stopped + } + + public let configuration: SSEClientConfiguration + private let transport: any SSETransport + private let clock: any SSEClock + private let pathObserver: (any NetworkPathObserving)? + private let decoder = JSONDecoder() + + public private(set) var state: SSEConnectionState = .idle + /// Sticky resume cursor for THIS subscription (nil until the first + /// id-carrying frame). Callers may persist it keyed by + /// (hub, namespace, filter set) and seed a future instance via + /// `seedCursor(_:)` before `start()`. + public private(set) var lastEventId: String? + /// From the latest handshake; new on every reconnect. Needed for + /// `POST /api/visibility`. + public private(set) var subscriptionId: String? + + private var continuation: AsyncStream.Continuation? + private var consumerGeneration = 0 + private var runTask: Task? + private var pathTask: Task? + private var currentAttemptTask: Task? + private var sleepTask: Task? + private var resumeWaiters: [CheckedContinuation] = [] + + private var stopped = false + private var suspended = false + private var connectAttempt = 0 + private var authRetryUsed = false + + // Per-attempt bookkeeping, read by the watchdog and lifecycle checks. + // Internal (not private) so @testable tests can synchronize on them. + private(set) var transportOpen = false + private(set) var handshakeSeen = false + private(set) var connectStartedAtMs = 0 + private(set) var lastActivityAtMs = 0 + + public init( + configuration: SSEClientConfiguration, + transport: any SSETransport = URLSessionSSETransport(), + clock: any SSEClock = SystemSSEClock(), + pathObserver: (any NetworkPathObserving)? = nil + ) { + self.configuration = configuration + self.transport = transport + self.clock = clock + self.pathObserver = pathObserver + } + + // MARK: - Lifecycle + + /// Starts the connection loop (first call) and returns the event stream. + /// Calling again replaces the consumer stream; calling after `stop()` + /// returns an immediately finished stream. + public func start() -> AsyncStream { + let (stream, continuation) = AsyncStream.makeStream(of: SSEClientEvent.self) + if stopped { + continuation.finish() + return stream + } + self.continuation?.finish() + self.continuation = continuation + consumerGeneration += 1 + let generation = consumerGeneration + continuation.onTermination = { [weak self] _ in + // Nobody is listening anymore: shut the connection down (guarded + // by generation so replacing the consumer does not stop us). + Task { await self?.consumerFinished(generation: generation) } + } + if runTask == nil { + runTask = Task { await self.run() } + if let pathObserver { + pathTask = Task { await self.observePath(pathObserver) } + } + } + return stream + } + + /// Permanently stops the client: tears down the connection, cancels any + /// pending retry, and finishes the event stream. Idempotent. + public func stop() { + stopped = true + suspended = false + drainResumeWaiters() + sleepTask?.cancel() + currentAttemptTask?.cancel() + pathTask?.cancel() + pathTask = nil + runTask?.cancel() + setState(.idle) + finishStream() + } + + /// App went to background: defer retries. A live connection is kept + /// (the OS may or may not preserve it); watchdog checks pause so a + /// frozen app cannot mis-detect staleness on wake. + public func suspend() { + guard !stopped, !suspended else { return } + suspended = true + // A pending backoff sleep is abandoned; the run loop parks at the + // top until resume() and then retries immediately. + sleepTask?.cancel() + } + + /// App returned to foreground. A parked retry runs immediately. When a + /// connection survived suspension, `staleThresholdCheck` distrusts it if + /// it has been silent for ≥ 45 s (`SSETimings.foregroundResumeStalenessMs`) + /// — an OS suspend can kill the socket without any error surfacing — and + /// reconnects immediately. + public func resume(staleThresholdCheck: Bool = true) { + guard !stopped, suspended else { return } + suspended = false + drainResumeWaiters() + if staleThresholdCheck, transportOpen, + clock.nowMs() - lastActivityAtMs >= SSETimings.foregroundResumeStalenessMs { + currentAttemptTask?.cancel() + } + } + + /// Seeds the resume cursor before `start()` (e.g. one persisted from a + /// previous process, keyed to this same subscription filter set — never + /// seed a cursor recorded under a different scope/hub/namespace). + public func seedCursor(_ id: String?) { + guard runTask == nil, !stopped else { return } + lastEventId = id + } + + private func consumerFinished(generation: Int) { + guard generation == consumerGeneration else { return } + stop() + } + + private func finishStream() { + continuation?.finish() + continuation = nil + } + + // MARK: - Run loop + + private func run() async { + while !stopped { + if suspended { + await parkUntilResumed() + continue + } + setState(.connecting) + let token = await configuration.tokenProvider() + if stopped { break } + if suspended { continue } + guard let token, let request = makeRequest(token: token) else { + // Cannot authenticate right now; the provider may recover + // (refresh in flight, credentials being restored) — retry on + // the normal schedule. + await scheduleBackoff() + continue + } + let exit = await performAttempt(request: request) + if stopped { break } + if exit == .stopped { continue } // loop condition exits + if exit == .unauthorized, !authRetryUsed { + // The token was rejected: ask the provider again (it + // refreshes) and retry immediately. Not a backoff attempt, + // and capped at once per connect cycle so a hub that keeps + // 401-ing still backs off. + authRetryUsed = true + continue + } + if suspended { continue } // parks at loop top; retry deferred + await scheduleBackoff() + } + setState(.idle) + finishStream() + } + + private func parkUntilResumed() async { + setState(.suspended) + while suspended, !stopped { + await withCheckedContinuation { (waiter: CheckedContinuation) in + if !suspended || stopped { + waiter.resume() + } else { + resumeWaiters.append(waiter) + } + } + } + } + + private func drainResumeWaiters() { + let waiters = resumeWaiters + resumeWaiters.removeAll() + for waiter in waiters { + waiter.resume() + } + } + + private func scheduleBackoff() async { + let attempt = connectAttempt + connectAttempt += 1 + setState(.backoff(attempt: attempt)) + let delay = configuration.reconnectPolicy.exponentialDelayMs(forAttempt: attempt) + + max(0, configuration.jitterMs()) + await backoffSleep(ms: delay) + } + + private func backoffSleep(ms: Int) async { + guard ms > 0, !suspended, !stopped else { return } + let clock = self.clock + let task = Task { try? await clock.sleep(ms: ms) } + sleepTask = task + await task.value + sleepTask = nil + } + + // MARK: - One connection attempt + + private func performAttempt(request: URLRequest) async -> AttemptExit { + connectStartedAtMs = clock.nowMs() + lastActivityAtMs = connectStartedAtMs + transportOpen = false + handshakeSeen = false + let attemptTask = Task { await self.readAttempt(request: request) } + currentAttemptTask = attemptTask + let watchdogTask = Task { await self.runWatchdog(killing: attemptTask) } + let exit = await attemptTask.value + watchdogTask.cancel() + currentAttemptTask = nil + transportOpen = false + handshakeSeen = false + return exit + } + + private func readAttempt(request: URLRequest) async -> AttemptExit { + let stream = transport.connect(request) + var parser = SSELineParser() + do { + for try await transportEvent in stream { + if Task.isCancelled { return cancellationExit() } + switch transportEvent { + case .connected(let response): + let status = response.statusCode + if status == 401 { return .unauthorized } + guard (200...299).contains(status) else { return .failed } + transportOpen = true + // Mirrors the web client's onopen: the backoff counter + // resets on every successful open, and the one free + // auth retry re-arms for the next cycle. + connectAttempt = 0 + authRetryUsed = false + lastActivityAtMs = clock.nowMs() + case .bytes(let data): + // Any received bytes — heartbeats, comments, partial + // frames — count as activity for the staleness clock. + lastActivityAtMs = clock.nowMs() + for frame in parser.consume(data) { + handleFrame(frame) + } + } + } + // Server closed the stream (or iteration ended on cancellation). + return Task.isCancelled ? cancellationExit() : .failed + } catch { + if Task.isCancelled || error is CancellationError { + return cancellationExit() + } + return .failed + } + } + + private func cancellationExit() -> AttemptExit { + stopped ? .stopped : .failed + } + + private func handleFrame(_ frame: SSEFrame) { + let decoded: SyncEvent + do { + decoded = try decoder.decode(SyncEvent.self, from: Data(frame.data.utf8)) + } catch { + // Known event type with a malformed payload (unknown types decode + // to .unknown and never throw). Mirror the web client: drop the + // frame, keep the stream alive, and leave the cursor BEHIND it — + // never advance past something that was not handled. + return + } + + switch decoded { + case .heartbeat: + // Pure liveness; activity was already stamped per-chunk. Carries + // no id, so the sticky cursor is untouched by construction. + return + case .connectionChanged(_, let payload): + if !handshakeSeen, payload?.status == "connected" { + handshakeSeen = true + subscriptionId = payload?.subscriptionId + setState(.connected) + // Absent verdict = older hub = cannot prove continuity: gap. + continuation?.yield(.handshake( + resume: payload?.resume ?? .gap, + subscriptionId: payload?.subscriptionId + )) + } else { + continuation?.yield(.event(decoded)) + } + default: + continuation?.yield(.event(decoded)) + } + + // At-least-once: the cursor moves only after the yield above, and + // only for frames that actually carried an id (sticky otherwise). + if let id = frame.id, !id.isEmpty { + lastEventId = id + } + } + + private func runWatchdog(killing attemptTask: Task) async { + while !Task.isCancelled { + try? await clock.sleep(ms: SSETimings.watchdogTickMs) + if Task.isCancelled { return } + if suspended { continue } // checks are skipped while backgrounded + let now = clock.nowMs() + if !transportOpen { + // Likely hung on a dead pooled socket; abandon the attempt + // and retry on a fresh connection. + if now - connectStartedAtMs >= SSETimings.connectTimeoutMs { + attemptTask.cancel() + return + } + } else if now - lastActivityAtMs >= SSETimings.stalenessThresholdMs { + attemptTask.cancel() + return + } + } + } + + private func observePath(_ observer: any NetworkPathObserving) async { + // The first element is the baseline emitted on subscription, not a + // change — see NetworkPathObserving. + var isBaseline = true + for await _ in observer.pathUpdates() { + if Task.isCancelled { return } + if isBaseline { + isBaseline = false + continue + } + // A path change while connected: the socket is almost certainly + // bound to a route that no longer exists. Treat it as a transport + // error; the attempt counter was reset on open, so the reconnect + // is immediate. + if !stopped, !suspended, transportOpen { + currentAttemptTask?.cancel() + } + } + } + + // MARK: - Request building + + private func makeRequest(token: String) -> URLRequest? { + guard var components = URLComponents(url: configuration.baseUrl, resolvingAgainstBaseURL: false) else { + return nil + } + components.path = "/api/events" + // Query-param auth, exactly like the web reference (buildEventsUrl): + // token, visibility, all|sessionId, lastEventId. + var items = [ + URLQueryItem(name: "token", value: token), + URLQueryItem(name: "visibility", value: configuration.visibility), + ] + switch configuration.scope { + case .global: + items.append(URLQueryItem(name: "all", value: "true")) + case .session(let sessionId): + items.append(URLQueryItem(name: "sessionId", value: sessionId)) + } + if let lastEventId, !lastEventId.isEmpty { + items.append(URLQueryItem(name: "lastEventId", value: lastEventId)) + } + components.queryItems = items + guard let url = components.url else { return nil } + var request = URLRequest(url: url) + request.httpMethod = "GET" + request.setValue("text/event-stream", forHTTPHeaderField: "Accept") + request.setValue("no-cache", forHTTPHeaderField: "Cache-Control") + if configuration.acceptEncodingIdentity { + // See SSEClientConfiguration.acceptEncodingIdentity (gzip TODO). + request.setValue("identity", forHTTPHeaderField: "Accept-Encoding") + } + return request + } + + // MARK: - State + + private func setState(_ newState: SSEConnectionState) { + guard state != newState else { return } + state = newState + continuation?.yield(.stateChanged(newState)) + } +} diff --git a/ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClock.swift b/ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClock.swift new file mode 100644 index 00000000..6b4967a8 --- /dev/null +++ b/ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClock.swift @@ -0,0 +1,33 @@ +import Dispatch +import Foundation + +/// Minimal time source used by `SSEClient` for the connect timeout, the +/// staleness watchdog, and backoff sleeps. +/// +/// Deliberately NOT the stdlib `Clock` protocol: tests need a manually +/// advanced clock whose `sleep` parks on a continuation until test code +/// moves time forward, and the stdlib protocol's associated types and +/// tolerance machinery buy nothing here. +public protocol SSEClock: Sendable { + /// Current time in milliseconds on a monotonically increasing scale. + /// Only ever used for differences; the origin is arbitrary. + func nowMs() -> Int + /// Suspends for `ms`. Must throw `CancellationError` promptly when the + /// surrounding task is cancelled — the client relies on it to abort + /// watchdog ticks and pending backoff sleeps. + func sleep(ms: Int) async throws +} + +/// Production clock: monotonic uptime + `Task.sleep`. +public struct SystemSSEClock: SSEClock { + public init() {} + + public func nowMs() -> Int { + Int(DispatchTime.now().uptimeNanoseconds / 1_000_000) + } + + public func sleep(ms: Int) async throws { + guard ms > 0 else { return } + try await Task.sleep(nanoseconds: UInt64(ms) * 1_000_000) + } +} diff --git a/ios/Packages/HapiKit/Sources/HapiClient/SSE/SSELineParser.swift b/ios/Packages/HapiKit/Sources/HapiClient/SSE/SSELineParser.swift new file mode 100644 index 00000000..2a33bb6b --- /dev/null +++ b/ios/Packages/HapiKit/Sources/HapiClient/SSE/SSELineParser.swift @@ -0,0 +1,115 @@ +import Foundation + +/// One dispatched SSE message. +/// +/// `id` is the value of the `id:` field seen in THIS event block only — `nil` +/// when the block carried no `id:` line (heartbeat / connection-changed / +/// toast frames). Cursor stickiness is deliberately NOT the parser's job: the +/// caller keeps its last-seen id and lets only non-nil (and non-empty) frame +/// ids overwrite it, so an id-less heartbeat can never reset the cursor. +public struct SSEFrame: Equatable, Sendable { + public var id: String? + /// Value of the `event:` field, `nil` for default `message` frames. + /// The hub only ever sends unnamed frames; kept for completeness. + public var event: String? + /// All `data:` lines of the block joined with `\n`. + public var data: String + + public init(id: String? = nil, event: String? = nil, data: String) { + self.id = id + self.event = event + self.data = data + } +} + +/// Incremental parser for a `text/event-stream` byte stream. +/// +/// Feed it raw chunks as they arrive (`consume`); it buffers partial lines +/// across chunk boundaries, understands LF and CRLF line endings, and emits +/// an `SSEFrame` for every blank-line dispatch that accumulated at least one +/// `data:` line. Field handling follows the WHATWG EventSource algorithm: +/// +/// - `data:` — multi-line, joined with `\n` at dispatch. +/// - `id:` — sets the block id (ignored when it contains U+0000, per spec). +/// Like EventSource's last-event-id buffer, an id from a block that never +/// dispatched (no data lines) carries over into the next block; a real +/// dispatch consumes it. +/// - `event:` — sets the frame's event name. +/// - `retry:` — ignored; reconnect policy is entirely client-owned (sse.md). +/// - `: comment` lines and unknown fields — ignored. +/// +/// Not handled (the hub never produces them): lone-CR line endings and a +/// leading UTF-8 BOM. Create a fresh parser per connection; state must never +/// bleed across reconnects. +public struct SSELineParser: Sendable { + private var pendingLine = Data() + private var dataLines: [String] = [] + private var eventType: String? + private var eventId: String? + + public init() {} + + /// Consumes one chunk and returns every frame completed by it, in order. + public mutating func consume(_ chunk: Data) -> [SSEFrame] { + var frames: [SSEFrame] = [] + for byte in chunk { + if byte == 0x0A { // LF terminates a line; strip an optional CR. + if pendingLine.last == 0x0D { + pendingLine.removeLast() + } + process(line: pendingLine, into: &frames) + pendingLine.removeAll(keepingCapacity: true) + } else { + pendingLine.append(byte) + } + } + return frames + } + + private mutating func process(line: Data, into frames: inout [SSEFrame]) { + if line.isEmpty { + // Blank line: dispatch. Per spec, no data lines means no event — + // the event type buffer resets, but the id buffer survives so a + // (hypothetical) id-only block still tags the next dispatch. + if !dataLines.isEmpty { + frames.append(SSEFrame(id: eventId, event: eventType, data: dataLines.joined(separator: "\n"))) + dataLines.removeAll(keepingCapacity: true) + eventId = nil + } + eventType = nil + return + } + if line.first == 0x3A { // ':' — comment line. + return + } + + let field: String + let value: String + if let colonIndex = line.firstIndex(of: 0x3A) { + field = String(decoding: line[line.startIndex.. AsyncThrowingStream +} + +/// Production transport on `URLSession.bytes(for:)`. +/// +/// Uses a dedicated session configuration because the defaults are wrong for +/// a long-lived stream that manages its own retries: +/// +/// - `timeoutIntervalForRequest` 300 s — this is URLSession's *idle* timeout +/// (time between chunks). It must exceed the 30 s heartbeat interval by a +/// wide margin (≥ 120 s required; the 90 s staleness watchdog in +/// `SSEClient` fires long before it anyway). +/// - `timeoutIntervalForResource` 7 days — the stream is expected to live +/// "forever"; the default 7-minute-per-resource cap would kill it. +/// - `waitsForConnectivity` false — a dead link must surface as an error +/// immediately so the reconnect state machine owns the schedule. +/// - No caching, ephemeral storage — an event stream must never be served +/// from or written to a cache. +public struct URLSessionSSETransport: SSETransport { + private let session: URLSession + + public init(session: URLSession = URLSessionSSETransport.makeSession()) { + self.session = session + } + + public static func makeSession() -> URLSession { + let configuration = URLSessionConfiguration.ephemeral + configuration.timeoutIntervalForRequest = 300 + configuration.timeoutIntervalForResource = 7 * 24 * 60 * 60 + configuration.waitsForConnectivity = false + configuration.requestCachePolicy = .reloadIgnoringLocalCacheData + configuration.urlCache = nil + configuration.httpCookieStorage = nil + return URLSession(configuration: configuration) + } + + public func connect(_ request: URLRequest) -> AsyncThrowingStream { + let session = self.session + return AsyncThrowingStream { continuation in + let task = Task { + do { + let (bytes, response) = try await session.bytes(for: request) + guard let http = response as? HTTPURLResponse else { + throw SSETransportError.notHTTP + } + continuation.yield(.connected(http)) + // AsyncBytes iterates single bytes (buffered internally); + // coalesce to one chunk per line so the parser is called + // O(lines), not O(bytes). + var chunk = Data() + chunk.reserveCapacity(1024) + for try await byte in bytes { + chunk.append(byte) + if byte == 0x0A { + continuation.yield(.bytes(chunk)) + chunk.removeAll(keepingCapacity: true) + } + } + if !chunk.isEmpty { + continuation.yield(.bytes(chunk)) + } + continuation.finish() + } catch { + continuation.finish(throwing: error) + } + } + continuation.onTermination = { _ in + task.cancel() + } + } + } +} diff --git a/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/ReconnectPolicyTests.swift b/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/ReconnectPolicyTests.swift new file mode 100644 index 00000000..d2f554eb --- /dev/null +++ b/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/ReconnectPolicyTests.swift @@ -0,0 +1,93 @@ +import Foundation +import Testing +@testable import HapiClient + +/// Deterministic seedable RNG (SplitMix64) for jitter tests. +struct SplitMix64: RandomNumberGenerator { + private var state: UInt64 + + init(seed: UInt64) { + state = seed + } + + mutating func next() -> UInt64 { + state &+= 0x9E37_79B9_7F4A_7C15 + var z = state + z = (z ^ (z >> 30)) &* 0xBF58_476D_1CE4_E5B9 + z = (z ^ (z >> 27)) &* 0x94D0_49BB_1331_11EB + return z ^ (z >> 31) + } +} + +@Suite("Reconnect policy") +struct ReconnectPolicyTests { + @Test func exactScheduleIncludingSlowCeiling() { + let policy = ReconnectPolicy() + let expected = [ + 0: 0, // first retry is immediate (jitter only) + 1: 1_000, + 2: 2_000, + 3: 4_000, + 4: 8_000, + 5: 16_000, + 6: 30_000, // 32 s capped at 30 s + 7: 30_000, // 64 s capped at 30 s + 8: 128_000, // slow ceiling (300 s) active, exponential resumes + 9: 256_000, + 10: 300_000, // 512 s capped at the slow ceiling + 11: 300_000, + 25: 300_000, + ] + for (attempt, delay) in expected { + #expect(policy.exponentialDelayMs(forAttempt: attempt) == delay, "attempt \(attempt)") + } + } + + @Test func hugeAttemptCountsDoNotOverflow() { + let policy = ReconnectPolicy() + #expect(policy.exponentialDelayMs(forAttempt: 100) == 300_000) + #expect(policy.exponentialDelayMs(forAttempt: Int(Int32.max)) == 300_000) + } + + @Test func negativeAttemptTreatedAsImmediate() { + #expect(ReconnectPolicy().exponentialDelayMs(forAttempt: -3) == 0) + } + + @Test func jitterStaysInsideBoundsWithSeededRNG() { + let policy = ReconnectPolicy() + var rng = SplitMix64(seed: 0xDEAD_BEEF) + for attempt in 0...20 { + for _ in 0..<50 { + let base = policy.exponentialDelayMs(forAttempt: attempt) + let full = policy.delayMs(forAttempt: attempt, using: &rng) + let jitter = full - base + #expect(jitter >= 0 && jitter <= 500, "attempt \(attempt) jitter \(jitter)") + } + } + } + + @Test func seededJitterIsDeterministic() { + let policy = ReconnectPolicy() + var first = SplitMix64(seed: 42) + var second = SplitMix64(seed: 42) + let a = (0..<20).map { _ in policy.delayMs(forAttempt: 3, using: &first) } + let b = (0..<20).map { _ in policy.delayMs(forAttempt: 3, using: &second) } + #expect(a == b) + // The draws themselves vary (jitter is actually applied). + #expect(Set(a).count > 1) + } + + @Test func timingConstantsMatchContract() { + #expect(SSETimings.heartbeatIntervalMs == 30_000) + #expect(SSETimings.stalenessThresholdMs == 90_000) + #expect(SSETimings.foregroundResumeStalenessMs == 45_000) + #expect(SSETimings.watchdogTickMs == 10_000) + #expect(SSETimings.connectTimeoutMs == 10_000) + let policy = ReconnectPolicy() + #expect(policy.baseDelayMs == 1_000) + #expect(policy.maxDelayMs == 30_000) + #expect(policy.slowMaxDelayMs == 300_000) + #expect(policy.slowAfterAttempts == 8) + #expect(policy.jitterRangeMs == 0...500) + } +} diff --git a/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSEClientTests.swift b/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSEClientTests.swift new file mode 100644 index 00000000..9296cf53 --- /dev/null +++ b/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSEClientTests.swift @@ -0,0 +1,419 @@ +import Foundation +import HapiProtocol +import Testing +@testable import HapiClient + +/// State-machine tests for `SSEClient`, all against a fake transport and a +/// manually advanced clock (no real timers, no sockets). Real-time waits +/// appear only in bounded polls that synchronize with the actor's async +/// processing. +@Suite("SSEClient state machine") +struct SSEClientTests { + typealias Harness = ( + client: SSEClient, + transport: FakeTransport, + clock: ManualClock, + tokens: TokenSource, + box: EventBox + ) + + private func makeHarness( + scope: SSEScope = .global, + visibility: String = "visible", + tokens: [String?] = ["tok-1"], + pathObserver: FakePathObserver? = nil, + transport: FakeTransport = FakeTransport(), + clock: ManualClock = ManualClock() + ) async -> Harness { + let tokenSource = TokenSource(tokens) + let configuration = SSEClientConfiguration( + baseUrl: URL(string: "https://hub.test")!, + tokenProvider: { tokenSource.next() }, + scope: scope, + visibility: visibility, + jitterMs: { 0 } + ) + let client = SSEClient( + configuration: configuration, + transport: transport, + clock: clock, + pathObserver: pathObserver + ) + let stream = await client.start() + let (box, _) = collect(stream) + return (client, transport, clock, tokenSource, box) + } + + /// Advances the manual clock once the watchdog (or any sleeper) has + /// actually parked — the watchdog registers its sleep asynchronously, so + /// blind advances could set a deadline the test never crosses. + private func advanceOnceSleeperParked(_ clock: ManualClock, byMs delta: Int) async { + #expect(await waitUntil { clock.activeSleeperCount() >= 1 }) + clock.advance(byMs: delta) + } + + // MARK: - Handshake + + @Test func handshakeGatesConnectedState() async throws { + let h = await makeHarness() + let connection = try #require(await h.transport.nextConnection()) + // Query-param auth, exactly like the web reference client. + #expect(connection.queryItems["token"] == "tok-1") + #expect(connection.queryItems["visibility"] == "visible") + #expect(connection.queryItems["all"] == "true") + #expect(connection.queryItems["sessionId"] == nil) + #expect(connection.queryItems["lastEventId"] == nil) + #expect(connection.request.value(forHTTPHeaderField: "Accept") == "text/event-stream") + #expect(connection.request.value(forHTTPHeaderField: "Accept-Encoding") == nil) + + h.clock.advance(byMs: 5) + connection.open() + #expect(await waitUntil { await h.client.transportOpen }) + // A 200 alone is not "up": the state machine stays .connecting until + // the hub's connection-changed handshake, heartbeats included. + h.clock.advance(byMs: 1) + connection.sendHeartbeat() + #expect(await waitUntil { await h.client.lastActivityAtMs == 6 }) + #expect(await h.client.state == .connecting) + #expect(h.box.handshakes.isEmpty) + #expect(!h.box.states.contains(.connected)) + + connection.sendHandshake(resume: "ok", subscriptionId: "sub-9") + #expect(await waitUntil { !h.box.handshakes.isEmpty }) + #expect(h.box.handshakes == [.handshake(resume: .ok, subscriptionId: "sub-9")]) + #expect(await h.client.state == .connected) + #expect(await h.client.subscriptionId == "sub-9") + // stateChanged(.connected) precedes the handshake yield. + let events = h.box.snapshot + let connectedIndex = try #require(events.firstIndex(of: .stateChanged(.connected))) + let handshakeIndex = try #require(events.firstIndex(of: .handshake(resume: .ok, subscriptionId: "sub-9"))) + #expect(connectedIndex < handshakeIndex) + await h.client.stop() + } + + @Test func resumeVerdictSurfacesGapWhenAbsent() async throws { + let h = await makeHarness() + let connection = try #require(await h.transport.nextConnection()) + connection.open() + connection.sendHandshake(resume: nil, subscriptionId: "sub-2") + #expect(await waitUntil { !h.box.handshakes.isEmpty }) + // Older hubs omit the verdict; absence must read as gap (full resync). + #expect(h.box.handshakes == [.handshake(resume: .gap, subscriptionId: "sub-2")]) + await h.client.stop() + } + + // MARK: - Cursor + + @Test func cursorSentOnReconnectAndIsolatedPerSubscription() async throws { + let h = await makeHarness() + let first = try #require(await h.transport.nextConnection()) + first.open() + first.sendHandshake() + first.sendEvent( + "{\"type\":\"session-removed\",\"namespace\":\"ns\",\"sessionId\":\"s-9\"}", + id: "018f:41:aa" + ) + #expect(await waitUntil { !h.box.syncEvents.isEmpty }) + #expect(h.box.syncEvents == [.sessionRemoved(namespace: "ns", sessionId: "s-9")]) + #expect(await h.client.lastEventId == "018f:41:aa") + + // Server closes; attempt counter was reset on open, so the reconnect + // is immediate and replays the cursor of THIS subscription. + first.finish() + let second = try #require(await h.transport.nextConnection()) + #expect(second.queryItems["lastEventId"] == "018f:41:aa") + #expect(second.queryItems["all"] == "true") + + // A different subscription must never inherit that cursor: fresh + // client, session scope, same hub. + let other = await makeHarness( + scope: .session("s-42"), + visibility: "hidden", + tokens: ["tok-B"], + transport: h.transport, + clock: h.clock + ) + let otherConnection = try #require(await other.transport.nextConnection()) + #expect(otherConnection.queryItems["sessionId"] == "s-42") + #expect(otherConnection.queryItems["all"] == nil) + #expect(otherConnection.queryItems["visibility"] == "hidden") + #expect(otherConnection.queryItems["token"] == "tok-B") + #expect(otherConnection.queryItems["lastEventId"] == nil) + await h.client.stop() + await other.client.stop() + } + + @Test func heartbeatKeepsWatchdogAliveButDoesNotMoveCursor() async throws { + let h = await makeHarness() + let connection = try #require(await h.transport.nextConnection()) + connection.open() + connection.sendHandshake() + connection.sendEvent( + "{\"type\":\"scheduled-matured\",\"namespace\":\"ns\",\"sessionId\":\"s-1\"}", + id: "018f:7:aa" + ) + #expect(await waitUntil { await h.client.lastEventId == "018f:7:aa" }) + + // 50 s of silence, then a heartbeat: watchdog stays quiet and the + // cursor keeps the last real id (heartbeat frames carry no id). + await advanceOnceSleeperParked(h.clock, byMs: 50_000) + connection.sendHeartbeat() + #expect(await waitUntil { await h.client.lastActivityAtMs == 50_000 }) + await advanceOnceSleeperParked(h.clock, byMs: 50_000) // 100 s total, only 50 s since the heartbeat + #expect(await h.transport.expectNoConnection()) + #expect(h.transport.totalConnections() == 1) + #expect(await h.client.state == .connected) + #expect(await h.client.lastEventId == "018f:7:aa") + + // Full silence from here: the 90 s threshold tears it down and the + // reconnect replays the heartbeat-untouched cursor. + await advanceOnceSleeperParked(h.clock, byMs: 90_000) + let second = try #require(await h.transport.nextConnection()) + #expect(second.queryItems["lastEventId"] == "018f:7:aa") + #expect(connection.isTerminated) + await h.client.stop() + } + + // MARK: - Watchdog + + @Test func ninetySecondsOfSilenceTriggersReconnect() async throws { + let h = await makeHarness() + let first = try #require(await h.transport.nextConnection()) + first.open() + first.sendHandshake() + #expect(await waitUntil { !h.box.handshakes.isEmpty }) + await advanceOnceSleeperParked(h.clock, byMs: 90_000) + _ = try #require(await h.transport.nextConnection()) + #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 0)) }) + #expect(first.isTerminated) + await h.client.stop() + } + + @Test func hungConnectAttemptIsAbandonedAfterTimeout() async throws { + let h = await makeHarness() + let first = try #require(await h.transport.nextConnection()) + // Never reaches OPEN; the 10 s connect deadline abandons it. + await advanceOnceSleeperParked(h.clock, byMs: 10_000) + _ = try #require(await h.transport.nextConnection()) + #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 0)) }) + #expect(first.isTerminated) + await h.client.stop() + } + + // MARK: - Suspend / resume + + @Test func suspendDefersRetriesAndResumeRunsThemImmediately() async throws { + let h = await makeHarness() + let first = try #require(await h.transport.nextConnection()) + first.finish() // attempt 0 fails → immediate retry + let second = try #require(await h.transport.nextConnection()) + second.finish() // attempt 1 → 1 s backoff + #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 1)) }) + + await h.client.suspend() + #expect(await waitUntil { await h.client.state == .suspended }) + + // Arbitrary time passes; a suspended client schedules nothing. + h.clock.advance(byMs: 600_000) + #expect(await h.transport.expectNoConnection()) + #expect(h.transport.totalConnections() == 2) + + await h.client.resume() + _ = try #require(await h.transport.nextConnection()) + #expect(h.transport.totalConnections() == 3) + await h.client.stop() + } + + @Test func resumeWithStaleConnectionReconnectsImmediately() async throws { + let h = await makeHarness() + let first = try #require(await h.transport.nextConnection()) + first.open() + first.sendHandshake() + #expect(await waitUntil { !h.box.handshakes.isEmpty }) + + await h.client.suspend() + // 50 s pass while suspended — beyond the 45 s foreground threshold. + // The watchdog itself must stay quiet (checks skip while suspended). + h.clock.advance(byMs: 50_000) + #expect(await h.transport.expectNoConnection()) + + await h.client.resume() + _ = try #require(await h.transport.nextConnection()) + #expect(first.isTerminated) + await h.client.stop() + } + + @Test func resumeWithFreshConnectionKeepsIt() async throws { + let h = await makeHarness() + let first = try #require(await h.transport.nextConnection()) + first.open() + first.sendHandshake() + #expect(await waitUntil { !h.box.handshakes.isEmpty }) + + await h.client.suspend() + h.clock.advance(byMs: 10_000) // well under the 45 s threshold + await h.client.resume() + #expect(await h.transport.expectNoConnection()) + #expect(!first.isTerminated) + #expect(await h.client.state == .connected) + await h.client.stop() + } + + // MARK: - Event delivery + + @Test func eventsAreYieldedInArrivalOrder() async throws { + let h = await makeHarness() + let connection = try #require(await h.transport.nextConnection()) + connection.open() + connection.sendHandshake() + connection.send( + "id: e-1\ndata: {\"type\":\"session-removed\",\"sessionId\":\"s-1\"}\n\n" + + "id: e-2\ndata: {\"type\":\"messages-invalidated\",\"sessionId\":\"s-2\"}\n\n" + + "id: e-3\ndata: {\"type\":\"scheduled-matured\",\"sessionId\":\"s-3\"}\n\n" + ) + #expect(await waitUntil { h.box.syncEvents.count == 3 }) + #expect(h.box.syncEvents == [ + .sessionRemoved(namespace: nil, sessionId: "s-1"), + .messagesInvalidated(namespace: nil, sessionId: "s-2"), + .scheduledMatured(namespace: nil, sessionId: "s-3"), + ]) + #expect(await h.client.lastEventId == "e-3") + await h.client.stop() + } + + @Test func unknownEventTypesPassThroughWithoutKillingTheStream() async throws { + let h = await makeHarness() + let connection = try #require(await h.transport.nextConnection()) + connection.open() + connection.sendHandshake() + connection.sendEvent( + "{\"type\":\"quantum-flux\",\"namespace\":\"ns\",\"data\":{\"weird\":true}}", + id: "e-8" + ) + connection.sendEvent( + "{\"type\":\"session-removed\",\"namespace\":\"ns\",\"sessionId\":\"s-1\"}", + id: "e-9" + ) + #expect(await waitUntil { h.box.syncEvents.count == 2 }) + #expect(h.box.syncEvents == [ + .unknown(type: "quantum-flux", namespace: "ns"), + .sessionRemoved(namespace: "ns", sessionId: "s-1"), + ]) + #expect(await h.client.lastEventId == "e-9") + #expect(await h.client.state == .connected) + await h.client.stop() + } + + @Test func malformedKnownEventIsSkippedWithoutAdvancingCursor() async throws { + let h = await makeHarness() + let connection = try #require(await h.transport.nextConnection()) + connection.open() + connection.sendHandshake() + connection.sendEvent( + "{\"type\":\"session-removed\",\"namespace\":\"ns\",\"sessionId\":\"s-1\"}", + id: "good-1" + ) + #expect(await waitUntil { await h.client.lastEventId == "good-1" }) + // Known type, missing required field: dropped, stream stays up, and + // the cursor stays BEHIND the unhandled frame (at-least-once). + connection.sendEvent("{\"type\":\"session-removed\",\"namespace\":\"ns\"}", id: "bad-1") + let moved = await waitUntil(timeoutMs: 300) { await h.client.lastEventId == "bad-1" } + #expect(!moved) + connection.sendEvent("{\"type\":\"messages-invalidated\",\"sessionId\":\"s-2\"}", id: "good-2") + #expect(await waitUntil { await h.client.lastEventId == "good-2" }) + #expect(h.box.syncEvents.count == 2) + #expect(await h.client.state == .connected) + await h.client.stop() + } + + // MARK: - Auth + + @Test func unauthorizedConnectGetsOneTokenRefreshBypassPerCycle() async throws { + let h = await makeHarness(tokens: ["tok-1", "tok-2", "tok-3"]) + let first = try #require(await h.transport.nextConnection()) + #expect(first.queryItems["token"] == "tok-1") + first.open(status: 401) + + // Free retry: the provider is asked again (it refreshes) and the + // reconnect is immediate, without counting a backoff attempt. + let second = try #require(await h.transport.nextConnection()) + #expect(second.queryItems["token"] == "tok-2") + + // The bypass is capped at once per connect cycle: a second + // consecutive 401 fails into normal backoff. + second.open(status: 401) + let third = try #require(await h.transport.nextConnection()) + #expect(third.queryItems["token"] == "tok-3") + #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 0)) }) + + // Any other non-2xx fails into backoff too (503: hub not ready). + third.open(status: 503) + #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 1)) }) + + // Exactly two backoffs total — none between the first 401 and its + // free retry. + let backoffs = h.box.states.filter { + if case .backoff = $0 { return true } + return false + } + #expect(backoffs == [.backoff(attempt: 0), .backoff(attempt: 1)]) + #expect(h.tokens.callCount() == 3) + await h.client.stop() + } + + // MARK: - Network path + + @Test func networkPathChangeWhileConnectedForcesReconnect() async throws { + let observer = FakePathObserver() + let h = await makeHarness(pathObserver: observer) + let first = try #require(await h.transport.nextConnection()) + first.open() + first.sendHandshake() + #expect(await waitUntil { !h.box.handshakes.isEmpty }) + + #expect(await waitUntil { observer.subscriberCount() >= 1 }) + observer.emitChange(isSatisfied: true) // e.g. wifi → cellular + _ = try #require(await h.transport.nextConnection()) + #expect(first.isTerminated) + await h.client.stop() + } + + // MARK: - Lifecycle + + @Test func stopTearsDownAndFinishesTheStream() async throws { + let h = await makeHarness() + let first = try #require(await h.transport.nextConnection()) + first.open() + first.sendHandshake() + #expect(await waitUntil { !h.box.handshakes.isEmpty }) + + await h.client.stop() + #expect(await waitUntil { first.isTerminated }) + #expect(await h.client.state == .idle) + #expect(await waitUntil { h.box.states.last == .idle }) + + // start() after stop() yields a finished stream, not a revived client. + let revived = await h.client.start() + var iterator = revived.makeAsyncIterator() + #expect(await iterator.next() == nil) + } + + @Test func seededCursorIsSentOnFirstConnect() async throws { + let transport = FakeTransport() + let clock = ManualClock() + let tokenSource = TokenSource(["tok-1"]) + let configuration = SSEClientConfiguration( + baseUrl: URL(string: "https://hub.test")!, + tokenProvider: { tokenSource.next() }, + scope: .global, + jitterMs: { 0 } + ) + let client = SSEClient(configuration: configuration, transport: transport, clock: clock) + await client.seedCursor("018f:99:zz") + let stream = await client.start() + _ = collect(stream) + let connection = try #require(await transport.nextConnection()) + #expect(connection.queryItems["lastEventId"] == "018f:99:zz") + await client.stop() + } +} diff --git a/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSELineParserTests.swift b/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSELineParserTests.swift new file mode 100644 index 00000000..f355bfca --- /dev/null +++ b/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSELineParserTests.swift @@ -0,0 +1,114 @@ +import Foundation +import Testing +@testable import HapiClient + +@Suite("SSE line parser") +struct SSELineParserTests { + private func frames(_ text: String) -> [SSEFrame] { + var parser = SSELineParser() + return parser.consume(Data(text.utf8)) + } + + @Test func joinsMultiLineDataWithNewline() { + let result = frames("data: line1\ndata: line2\ndata: line3\n\n") + #expect(result == [SSEFrame(data: "line1\nline2\nline3")]) + } + + @Test func emitsIdOnlyForTheBlockThatCarriedIt() { + // A dispatched id is consumed; the next block starts with no id. + // Cursor stickiness across id-less frames is the caller's job. + let result = frames("id: 018f:1:aa\ndata: first\n\ndata: second\n\n") + #expect(result.count == 2) + #expect(result[0] == SSEFrame(id: "018f:1:aa", data: "first")) + #expect(result[1].id == nil) + #expect(result[1].data == "second") + } + + @Test func heartbeatShapedFrameHasNilId() { + let result = frames("id: 42\ndata: {\"a\":1}\n\ndata: {\"type\":\"heartbeat\"}\n\n") + #expect(result.map(\.id) == ["42", nil]) + } + + @Test func idFromNonDispatchedBlockTagsNextDispatch() { + // Mirrors the EventSource last-event-id buffer: an id-only block sets + // the id even though nothing dispatches, and the next data block + // carries it. + let result = frames("id: X\n\ndata: y\n\n") + #expect(result == [SSEFrame(id: "X", data: "y")]) + } + + @Test func handlesCRLFLineEndings() { + let result = frames("id: 5\r\nevent: named\r\ndata: hello\r\ndata: world\r\n\r\n") + #expect(result == [SSEFrame(id: "5", event: "named", data: "hello\nworld")]) + } + + @Test func ignoresInterleavedComments() { + let result = frames(": keepalive\ndata: a\n: mid-frame comment\ndata: b\n\n: trailing\n") + #expect(result == [SSEFrame(data: "a\nb")]) + } + + @Test func reassemblesFramesSplitAcrossChunks() { + let text = "id: 018f:9:bb\r\ndata: {\"type\":\"x\"}\r\n\r\ndata: tail\n\n" + let whole = frames(text) + + // Byte-at-a-time feeding must produce the identical frames, + // including a CRLF split between the CR and the LF. + var parser = SSELineParser() + var collected: [SSEFrame] = [] + for byte in Data(text.utf8) { + collected += parser.consume(Data([byte])) + } + #expect(collected == whole) + #expect(whole.count == 2) + #expect(whole[0] == SSEFrame(id: "018f:9:bb", data: "{\"type\":\"x\"}")) + + // And an arbitrary mid-line split. + var parser2 = SSELineParser() + let bytes = Data(text.utf8) + let cut = 7 + var collected2 = parser2.consume(bytes.prefix(cut)) + collected2 += parser2.consume(bytes.dropFirst(cut)) + #expect(collected2 == whole) + } + + @Test func ignoresRetryAndUnknownFields() { + let result = frames("retry: 5000\nfancy-field: nope\ndata: a\n\n") + #expect(result == [SSEFrame(data: "a")]) + } + + @Test func lineWithoutColonIsFieldWithEmptyValue() { + // Per spec, "data" alone is the data field with an empty value. + let result = frames("data\n\n") + #expect(result == [SSEFrame(data: "")]) + } + + @Test func stripsExactlyOneLeadingSpace() { + #expect(frames("data:no-space\n\n") == [SSEFrame(data: "no-space")]) + #expect(frames("data: two-spaces\n\n") == [SSEFrame(data: " two-spaces")]) + } + + @Test func emptyDataFieldDispatchesEmptyString() { + #expect(frames("data:\n\n") == [SSEFrame(data: "")]) + } + + @Test func blankLineWithoutDataDispatchesNothingAndResetsEventType() { + let result = frames("event: named\n\ndata: x\n\n") + // The first block had no data: no frame, and its event type must not + // leak into the following block. + #expect(result == [SSEFrame(data: "x")]) + } + + @Test func idContainingNulIsIgnored() { + let result = frames("id: a\u{0000}b\ndata: x\n\n") + #expect(result == [SSEFrame(id: nil, data: "x")]) + } + + @Test func incompleteTrailingBlockIsNotEmitted() { + var parser = SSELineParser() + let first = parser.consume(Data("data: complete\n\ndata: partial".utf8)) + #expect(first == [SSEFrame(data: "complete")]) + // The unterminated tail only dispatches once its blank line arrives. + let second = parser.consume(Data("\n\n".utf8)) + #expect(second == [SSEFrame(data: "partial")]) + } +} diff --git a/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSETestSupport.swift b/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSETestSupport.swift new file mode 100644 index 00000000..9f71829d --- /dev/null +++ b/ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSETestSupport.swift @@ -0,0 +1,332 @@ +import Foundation +import HapiProtocol +@testable import HapiClient + +// MARK: - Manual clock + +/// Deterministic clock: `sleep` parks on a continuation until test code +/// advances time. Cancellation resumes the sleeper with `CancellationError`, +/// including the cancel-before-park race. +final class ManualClock: SSEClock, @unchecked Sendable { + private struct Sleeper { + let id: Int + let deadlineMs: Int + let continuation: CheckedContinuation + } + + private let lock = NSLock() + private var currentMs = 0 + private var nextSleeperId = 0 + private var sleepers: [Sleeper] = [] + private var cancelledIds: Set = [] + + func nowMs() -> Int { + lock.withLock { currentMs } + } + + func sleep(ms: Int) async throws { + let id: Int = lock.withLock { + nextSleeperId += 1 + return nextSleeperId + } + try await withTaskCancellationHandler { + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation) in + enum Immediate { case resume, cancel, park } + let action: Immediate = lock.withLock { + if cancelledIds.remove(id) != nil { + return .cancel + } + if ms <= 0 { + return .resume + } + sleepers.append(Sleeper(id: id, deadlineMs: currentMs + ms, continuation: continuation)) + return .park + } + switch action { + case .resume: continuation.resume() + case .cancel: continuation.resume(throwing: CancellationError()) + case .park: break + } + } + } onCancel: { + let sleeper: Sleeper? = lock.withLock { + if let index = sleepers.firstIndex(where: { $0.id == id }) { + return sleepers.remove(at: index) + } + cancelledIds.insert(id) + return nil + } + sleeper?.continuation.resume(throwing: CancellationError()) + } + } + + /// Number of currently parked sleepers. Tests gate clock advances on + /// this: the client's watchdog registers its sleep asynchronously, and + /// advancing before it parked would leave it waiting for a deadline the + /// test never reaches. + func activeSleeperCount() -> Int { + lock.withLock { sleepers.count } + } + + func advance(byMs delta: Int) { + let due: [Sleeper] = lock.withLock { + currentMs += delta + let now = currentMs + let ready = sleepers.filter { $0.deadlineMs <= now }.sorted { $0.deadlineMs < $1.deadlineMs } + sleepers.removeAll { $0.deadlineMs <= now } + return ready + } + for sleeper in due { + sleeper.continuation.resume() + } + } +} + +// MARK: - Fake transport + +final class Flag: @unchecked Sendable { + private let lock = NSLock() + private var value = false + + func set() { + lock.withLock { value = true } + } + + var isSet: Bool { + lock.withLock { value } + } +} + +final class FakeConnection: @unchecked Sendable { + let request: URLRequest + private let continuation: AsyncThrowingStream.Continuation + private let terminated = Flag() + + init(request: URLRequest, continuation: AsyncThrowingStream.Continuation) { + self.request = request + self.continuation = continuation + let flag = terminated + continuation.onTermination = { _ in flag.set() } + } + + var isTerminated: Bool { + terminated.isSet + } + + func open(status: Int = 200) { + let response = HTTPURLResponse( + url: request.url ?? URL(string: "https://hub.test")!, + statusCode: status, + httpVersion: "HTTP/1.1", + headerFields: nil + )! + continuation.yield(.connected(response)) + } + + func send(_ text: String) { + continuation.yield(.bytes(Data(text.utf8))) + } + + func sendEvent(_ json: String, id: String? = nil) { + var block = "" + if let id { + block += "id: \(id)\n" + } + block += "data: \(json)\n\n" + send(block) + } + + func sendHeartbeat(timestamp: Int = 1) { + sendEvent("{\"type\":\"heartbeat\",\"namespace\":\"ns\",\"data\":{\"timestamp\":\(timestamp)}}") + } + + func sendHandshake(resume: String? = "ok", subscriptionId: String = "sub-1") { + let resumeField = resume.map { ",\"resume\":\"\($0)\"" } ?? "" + sendEvent("{\"type\":\"connection-changed\",\"data\":{\"status\":\"connected\",\"subscriptionId\":\"\(subscriptionId)\"\(resumeField)}}") + } + + func finish() { + continuation.finish() + } + + func fail(_ error: Error = URLError(.networkConnectionLost)) { + continuation.finish(throwing: error) + } + + var queryItems: [String: String] { + guard let url = request.url, + let components = URLComponents(url: url, resolvingAgainstBaseURL: false) else { + return [:] + } + var result: [String: String] = [:] + for item in components.queryItems ?? [] { + result[item.name] = item.value ?? "" + } + return result + } +} + +final class FakeTransport: SSETransport, @unchecked Sendable { + private let lock = NSLock() + private var pending: [FakeConnection] = [] + private var connectCount = 0 + + func connect(_ request: URLRequest) -> AsyncThrowingStream { + let (stream, continuation) = AsyncThrowingStream.makeStream(of: SSETransportEvent.self) + let connection = FakeConnection(request: request, continuation: continuation) + lock.withLock { + connectCount += 1 + pending.append(connection) + } + return stream + } + + func totalConnections() -> Int { + lock.withLock { connectCount } + } + + /// Real-time bounded wait for the next unclaimed connection. + func nextConnection(timeoutMs: Int = 2_000) async -> FakeConnection? { + let deadline = Date().addingTimeInterval(Double(timeoutMs) / 1_000) + while Date() < deadline { + let claimed: FakeConnection? = lock.withLock { + pending.isEmpty ? nil : pending.removeFirst() + } + if let claimed { + return claimed + } + try? await Task.sleep(nanoseconds: 5_000_000) + } + return nil + } + + /// True when no NEW connection shows up within the grace window. + func expectNoConnection(forMs ms: Int = 200) async -> Bool { + let deadline = Date().addingTimeInterval(Double(ms) / 1_000) + while Date() < deadline { + let hasPending = lock.withLock { !pending.isEmpty } + if hasPending { + return false + } + try? await Task.sleep(nanoseconds: 5_000_000) + } + return lock.withLock { pending.isEmpty } + } +} + +// MARK: - Fake path observer + +final class FakePathObserver: NetworkPathObserving, @unchecked Sendable { + private let lock = NSLock() + private var continuations: [AsyncStream.Continuation] = [] + + func pathUpdates() -> AsyncStream { + let (stream, continuation) = AsyncStream.makeStream(of: NetworkPathUpdate.self) + lock.withLock { continuations.append(continuation) } + // NWPathMonitor always fires the current path on subscription. + continuation.yield(NetworkPathUpdate(isSatisfied: true)) + return stream + } + + /// Tests gate `emitChange` on this: the client subscribes from an async + /// task, and a change emitted before subscription would be lost. + func subscriberCount() -> Int { + lock.withLock { continuations.count } + } + + func emitChange(isSatisfied: Bool = true) { + let all = lock.withLock { continuations } + for continuation in all { + continuation.yield(NetworkPathUpdate(isSatisfied: isSatisfied)) + } + } +} + +// MARK: - Token source + +final class TokenSource: @unchecked Sendable { + private let lock = NSLock() + private var queue: [String?] + private(set) var calls = 0 + + /// Returns the queued tokens in order, sticking on the last one. + init(_ tokens: [String?]) { + queue = tokens + } + + func next() -> String? { + lock.withLock { + calls += 1 + if queue.count > 1 { + return queue.removeFirst() + } + return queue.first ?? nil + } + } + + func callCount() -> Int { + lock.withLock { calls } + } +} + +// MARK: - Event collection + +final class EventBox: @unchecked Sendable { + private let lock = NSLock() + private var storage: [SSEClientEvent] = [] + + func append(_ event: SSEClientEvent) { + lock.withLock { storage.append(event) } + } + + var snapshot: [SSEClientEvent] { + lock.withLock { storage } + } + + var syncEvents: [SyncEvent] { + snapshot.compactMap { + if case .event(let event) = $0 { return event } + return nil + } + } + + var handshakes: [SSEClientEvent] { + snapshot.filter { + if case .handshake = $0 { return true } + return false + } + } + + var states: [SSEConnectionState] { + snapshot.compactMap { + if case .stateChanged(let state) = $0 { return state } + return nil + } + } +} + +func collect(_ stream: AsyncStream) -> (box: EventBox, task: Task) { + let box = EventBox() + let task = Task { + for await event in stream { + box.append(event) + } + } + return (box, task) +} + +/// Real-time bounded poll for an async condition. +@discardableResult +func waitUntil( + timeoutMs: Int = 2_000, + _ condition: @Sendable () async -> Bool +) async -> Bool { + let deadline = Date().addingTimeInterval(Double(timeoutMs) / 1_000) + while Date() < deadline { + if await condition() { + return true + } + try? await Task.sleep(nanoseconds: 5_000_000) + } + return await condition() +} diff --git a/ios/README.md b/ios/README.md index 3a101378..f08f7fe7 100644 --- a/ios/README.md +++ b/ios/README.md @@ -47,8 +47,23 @@ ios/ The chat pipeline + message window logic (ported from web/src/chat/**) land in M2, validated against shared/fixtures/**. - HapiClient Transport layer: APIClient, AuthManager (single-flight - refresh), SSEClient, @Observable stores, snapshots. + HapiClient Transport layer. As of M1c: + SSE/ actor SSEClient — handshake-gated connect + (resume ok/gap surfaced), sticky per- + subscription cursor with at-least-once + replay, 10 s connect timeout + 90 s + staleness watchdog, backoff per sse.md + (1 s ×2 → 30 s, 300 s after 8 attempts, + 0–500 ms jitter), suspend/resume with the + 45 s foreground staleness check, NWPath + change → immediate reconnect. Plus + SSELineParser, ReconnectPolicy/SSETimings, + URLSessionSSETransport (dedicated config; + gzip streaming-decompression verification + is still TODO — `acceptEncodingIdentity` + is the fallback flag until then). + APIClient + AuthManager (single-flight refresh), + @Observable stores and snapshots land with M1b+. ``` The app target stays thin; features live in `HapiKit` so they are testable