diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index cff0a1281..fe0e8ec41 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -171,7 +171,23 @@ and answers each poll from that buffer — `Ok` while connected, `Err` while reconnecting — so the channel/state/staleness are untouched); the `data:` payload is the same JSON the GET returned. A bursty feed (EEW, silent between events) uses connection-open liveness; a continuous feed (RTS, ~1 Hz) uses -event-recency. `RealtimeService` + `AppLifecycleListener` +event-recency. Neither trusts an open socket on its own: the servers write a +`: ping` comment on an idle stream, the parser surfaces it, and a connection +that receives nothing at all for `sseSilenceLimit` (2.2 × the ping interval, +`sse_client.dart`) is half-open — a handover or NAT timeout that never sent a +FIN — so the source answers `Err` and replaces it, and silence never reads as +calm. A change of network (`core/platform/network_changes.dart`: Android's +default-network callback, iOS's `NWPathMonitor`) is fanned out by +`RealtimeService` to every source, and an SSE source drops its connection at +once and reopens after a random 1–2 s settle with its backoff reset, rather +than waiting out the silence limit; the backoff itself is jittered ±50 % so a +region's audience does not reconnect in lockstep. While the EEW stream is down +(past a 3 s grace), its source polls the one-shot GET every 10 s through the +same decode and agency filters and answers from that snapshot, flagged as a +fallback (`RealtimeSource.servingFallback`): the channel shows it but never +classifies it better than `stale` (`classifyStatus(fallback:)`), since a +snapshot every ten seconds cannot keep `live`'s promise, and polling stops the +moment the stream answers. `RealtimeService` + `AppLifecycleListener` pause polling on background and resume (recompute status → resync clock → refetch) on foreground; background alerting is push's job. A safety-critical feed that is `stale`/`offline` must never be presented as current. diff --git a/android/app/src/main/AndroidManifest.xml b/android/app/src/main/AndroidManifest.xml index 46d44e6f3..ef2b373de 100644 --- a/android/app/src/main/AndroidManifest.xml +++ b/android/app/src/main/AndroidManifest.xml @@ -1,5 +1,9 @@ + + diff --git a/android/app/src/main/kotlin/com/exptech/dpip/MainActivity.kt b/android/app/src/main/kotlin/com/exptech/dpip/MainActivity.kt index 0a1319606..28a31db22 100644 --- a/android/app/src/main/kotlin/com/exptech/dpip/MainActivity.kt +++ b/android/app/src/main/kotlin/com/exptech/dpip/MainActivity.kt @@ -60,6 +60,9 @@ class MainActivity : FlutterActivity() { EventChannel(messenger, CompassChannel.NAME) .setStreamHandler(CompassChannel(applicationContext)) + EventChannel(messenger, NetworkChannel.NAME) + .setStreamHandler(NetworkChannel(applicationContext)) + MethodChannel(messenger, PlainChannelsChannel.NAME) .setMethodCallHandler(PlainChannelsChannel(applicationContext)) diff --git a/android/app/src/main/kotlin/com/exptech/dpip/NetworkChannel.kt b/android/app/src/main/kotlin/com/exptech/dpip/NetworkChannel.kt new file mode 100644 index 000000000..b4afe8b18 --- /dev/null +++ b/android/app/src/main/kotlin/com/exptech/dpip/NetworkChannel.kt @@ -0,0 +1,106 @@ +package com.exptech.dpip + +import android.content.Context +import android.net.ConnectivityManager +import android.net.Network +import android.net.NetworkCapabilities +import android.os.Handler +import android.os.Looper +import io.flutter.plugin.common.EventChannel + +/** + * Streams which network the device's traffic currently leaves through, once + * on listen and again each time that changes, so the realtime feeds can + * replace connections bound to a network that is gone. + * + * Each event is an identity string: `none` with no default network, otherwise + * `:`. The handle is what makes it an identity — + * it changes whenever the network does, Wi-Fi to a different Wi-Fi included, + * which a transport type alone would miss. The Dart side only compares them. + * + * One system callback, so a channel rather than a connectivity package. The + * default-network callback reports the network apps actually use (a validated + * Wi-Fi beats cellular), not every interface that happens to be up, and it is + * delivered on the main looper, where the event sink must be called. + */ +class NetworkChannel(context: Context) : EventChannel.StreamHandler { + + companion object { + const val NAME = "com.exptech.dpip/network" + } + + private val connectivity = + context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager + private val main = Handler(Looper.getMainLooper()) + private var sink: EventChannel.EventSink? = null + private var callback: ConnectivityManager.NetworkCallback? = null + + /** + * The transport first seen for each network handle. A network's transport + * never changes, but reading its capabilities from inside a callback can + * race and come back null — and "other:7" then "wifi:7" would read as a + * change of network, tearing down a healthy connection for nothing. + * Remembering the first answer keeps one handle one identity. + */ + private val transports = HashMap() + + override fun onListen(arguments: Any?, events: EventChannel.EventSink?) { + sink = events + val callback = object : ConnectivityManager.NetworkCallback() { + override fun onAvailable(network: Network) { + sink?.success(identity(network)) + } + + // Whatever is the default *now*, not a hard "none": if the loss of + // the old network is reported after its replacement came up, the + // last word must still be the network actually in use. The lost + // one itself can linger as "active" for a moment; that is none. + override fun onLost(network: Network) { + val active = connectivity.activeNetwork + sink?.success(identity(if (active == network) null else active)) + } + } + try { + events?.success(identity(connectivity.activeNetwork)) + connectivity.registerDefaultNetworkCallback(callback, main) + this.callback = callback + } catch (error: RuntimeException) { + // SecurityException without ACCESS_NETWORK_STATE, or the per-app + // callback quota. The feeds still fall back on their own silence + // limit, so this is a degraded path, not a failure to start. + events?.error("UNAVAILABLE", error.message, null) + } + } + + override fun onCancel(arguments: Any?) { + callback?.let { + try { + connectivity.unregisterNetworkCallback(it) + } catch (_: IllegalArgumentException) { + // Never registered — nothing to undo. + } + } + callback = null + sink = null + } + + private fun identity(network: Network?): String { + if (network == null) return "none" + val handle = network.networkHandle + // A handle is never reused while the process lives; a long session + // only ever needs the last few. + if (transports.size > 16) transports.clear() + val transport = transports.getOrPut(handle) { + val caps = connectivity.getNetworkCapabilities(network) + when { + caps == null -> "other" + caps.hasTransport(NetworkCapabilities.TRANSPORT_WIFI) -> "wifi" + caps.hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) -> "cellular" + caps.hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET) -> "ethernet" + caps.hasTransport(NetworkCapabilities.TRANSPORT_VPN) -> "vpn" + else -> "other" + } + } + return "$transport:$handle" + } +} diff --git a/ios/Runner.xcodeproj/project.pbxproj b/ios/Runner.xcodeproj/project.pbxproj index f16faa4ff..34cff7394 100644 --- a/ios/Runner.xcodeproj/project.pbxproj +++ b/ios/Runner.xcodeproj/project.pbxproj @@ -40,6 +40,7 @@ 9F74F772B6E8760DD47AB2FE /* Sounds/normal.aiff in Resources */ = {isa = PBXBuildFile; fileRef = B916667D1B2356583B174E80 /* Sounds/normal.aiff */; }; A8D382D04B4ACD327E29F46B /* Sounds/eq.aiff in Resources */ = {isa = PBXBuildFile; fileRef = 682D0165E2FF3895C5B252C5 /* Sounds/eq.aiff */; }; AA0000000000000000000C02 /* CompassPlugin.swift in Sources */ = {isa = PBXBuildFile; fileRef = AA0000000000000000000C01 /* CompassPlugin.swift */; }; + AA0000000000000000000H02 /* NetworkPlugin.swift in Sources */ = {isa = PBXBuildFile; fileRef = AA0000000000000000000H01 /* NetworkPlugin.swift */; }; AA0000000000000000000D02 /* DeviceInfoPlugin.swift in Sources */ = {isa = PBXBuildFile; fileRef = AA0000000000000000000D01 /* DeviceInfoPlugin.swift */; }; NL0G00000000000000000002 /* NativeLogPlugin.swift in Sources */ = {isa = PBXBuildFile; fileRef = NL0G00000000000000000001 /* NativeLogPlugin.swift */; }; NL0G00000000000000000004 /* NativeLogTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = NL0G00000000000000000003 /* NativeLogTests.swift */; }; @@ -166,6 +167,7 @@ 97C147021CF9000F007C117D /* Info.plist */ = {isa = PBXFileReference; lastKnownFileType = text.plist.xml; path = Info.plist; sourceTree = ""; }; A382CD9DEA741E45DBF741D7 /* Sounds/rain.aiff */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = audio.aiff; path = Sounds/rain.aiff; sourceTree = ""; }; AA0000000000000000000C01 /* CompassPlugin.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CompassPlugin.swift; sourceTree = ""; }; + AA0000000000000000000H01 /* NetworkPlugin.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NetworkPlugin.swift; sourceTree = ""; }; AA0000000000000000000D01 /* DeviceInfoPlugin.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = DeviceInfoPlugin.swift; sourceTree = ""; }; NL0G00000000000000000001 /* NativeLogPlugin.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NativeLogPlugin.swift; sourceTree = ""; }; NL0G00000000000000000003 /* NativeLogTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NativeLogTests.swift; sourceTree = ""; }; @@ -446,6 +448,7 @@ 1498D2331E8E89220040F4C2 /* GeneratedPluginRegistrant.m */, 74858FAE1ED2DC5600515810 /* AppDelegate.swift */, AA0000000000000000000C01 /* CompassPlugin.swift */, + AA0000000000000000000H01 /* NetworkPlugin.swift */, AA0000000000000000000E01 /* ScreenWakePlugin.swift */, AA0000000000000000000G01 /* LocationTrackStore.swift */, AA0000000000000000000F01 /* ApnsTokenPlugin.swift */, @@ -753,6 +756,7 @@ files = ( 74858FAF1ED2DC5600515810 /* AppDelegate.swift in Sources */, AA0000000000000000000C02 /* CompassPlugin.swift in Sources */, + AA0000000000000000000H02 /* NetworkPlugin.swift in Sources */, AA0000000000000000000E02 /* ScreenWakePlugin.swift in Sources */, AA0000000000000000000G02 /* LocationTrackStore.swift in Sources */, AA0000000000000000000F02 /* ApnsTokenPlugin.swift in Sources */, diff --git a/ios/Runner/AppDelegate.swift b/ios/Runner/AppDelegate.swift index 316f24a4d..08fb8849a 100644 --- a/ios/Runner/AppDelegate.swift +++ b/ios/Runner/AppDelegate.swift @@ -49,6 +49,7 @@ import UserNotifications // lifecycle (registering a bare channel here can silently stop firing). DeviceInfoPlugin.register(with: registry.registrar(forPlugin: "DeviceInfoPlugin")!) CompassPlugin.register(with: registry.registrar(forPlugin: "CompassPlugin")!) + NetworkPlugin.register(with: registry.registrar(forPlugin: "NetworkPlugin")!) MapSnapshotPlugin.register(with: registry.registrar(forPlugin: "MapSnapshotPlugin")!) MapCachePlugin.register(with: registry.registrar(forPlugin: "MapCachePlugin")!) StorageScanPlugin.register(with: registry.registrar(forPlugin: "StorageScanPlugin")!) diff --git a/ios/Runner/NetworkPlugin.swift b/ios/Runner/NetworkPlugin.swift new file mode 100644 index 000000000..788e4adc1 --- /dev/null +++ b/ios/Runner/NetworkPlugin.swift @@ -0,0 +1,69 @@ +import Flutter +import Network + +/// EventChannel plugin reporting which network the device's traffic currently +/// leaves through, once on listen and again whenever the path changes, so the +/// realtime feeds can replace connections bound to a network that is gone. +/// +/// Each event is an identity string: `none` when no path is satisfied, +/// otherwise `:` for the preferred interface +/// (`availableInterfaces` is ordered by preference). The Dart side only +/// compares consecutive identities, and drops the repeats `NWPathMonitor` +/// sends when some other property of the same path moves. +/// +/// Coarser than Android's network handle: one Wi-Fi network to another keeps +/// `wifi:en0`, and `NWPath` exposes nothing that reliably tells them apart. +/// That case is left to the feeds' silence limit; every change of interface +/// type, and every drop to no network and back, is caught here. +/// +/// One system monitor, so a channel rather than a connectivity package. +/// Registered as a real `FlutterPlugin` for the implicit-engine lifecycle. +public class NetworkPlugin: NSObject, FlutterPlugin, FlutterStreamHandler { + private var monitor: NWPathMonitor? + private var sink: FlutterEventSink? + + public static func register(with registrar: FlutterPluginRegistrar) { + let channel = FlutterEventChannel( + name: "com.exptech.dpip/network", + binaryMessenger: registrar.messenger()) + channel.setStreamHandler(NetworkPlugin()) + } + + public func onListen( + withArguments arguments: Any?, + eventSink events: @escaping FlutterEventSink + ) -> FlutterError? { + sink = events + let monitor = NWPathMonitor() + // Delivered on the main queue, where the event sink must be called; a + // change of path is rare enough that it costs the main thread nothing. + monitor.pathUpdateHandler = { [weak self] path in + self?.sink?(NetworkPlugin.identity(of: path)) + } + monitor.start(queue: .main) + self.monitor = monitor + return nil + } + + public func onCancel(withArguments arguments: Any?) -> FlutterError? { + monitor?.cancel() + monitor = nil + sink = nil + return nil + } + + static func identity(of path: NWPath) -> String { + guard path.status == .satisfied, let interface = path.availableInterfaces.first + else { return "none" } + let type: String + switch interface.type { + case .wifi: type = "wifi" + case .cellular: type = "cellular" + case .wiredEthernet: type = "ethernet" + case .loopback: type = "loopback" + case .other: type = "other" + @unknown default: type = "other" + } + return "\(type):\(interface.name)" + } +} diff --git a/lib/bootstrap.dart b/lib/bootstrap.dart index 4aa573917..4e83d98c5 100644 --- a/lib/bootstrap.dart +++ b/lib/bootstrap.dart @@ -17,6 +17,7 @@ import 'package:dpip/core/logging/log_store.dart'; import 'package:dpip/core/network/api_client.dart'; import 'package:dpip/core/platform/background_location.dart'; import 'package:dpip/core/platform/install_source.dart'; +import 'package:dpip/core/platform/network_changes.dart'; import 'package:dpip/core/platform/widget_location_catalog_coordinator.dart'; import 'package:dpip/core/platform/widget_snapshot_writer.dart'; import 'package:dpip/core/network/dio_client.dart'; @@ -303,7 +304,13 @@ Future bootstrap() async { ); AppTime.install(serverClock); AppTime.sync().ignore(); - final realtimeService = RealtimeService(serverClock); + // Network changes reach the realtime spine so a feed's held-open connection + // is replaced as soon as the network it was bound to goes, instead of when + // its silence limit runs out. + final realtimeService = RealtimeService( + serverClock, + networkChanges: NetworkChanges.stream, + ); // Push: best-effort and off the first frame — a missing push environment or // slow FCM registration must never gate launch. The token is only consumed diff --git a/lib/core/network/api_client.dart b/lib/core/network/api_client.dart index b707869e4..b6c0ce457 100644 --- a/lib/core/network/api_client.dart +++ b/lib/core/network/api_client.dart @@ -210,18 +210,24 @@ class ApiClient { /// Opens a streaming GET against [tier] as a raw byte stream, trying each /// region host in failover order (same policy as [request]). /// - /// Built for long-lived responses (Server-Sent Events): the idle receive - /// timeout is **disabled** (a feed may be legitimately silent for minutes — - /// the EEW stream sends nothing between earthquakes), while the connect - /// timeout still bounds the initial handshake so a dead host fails over. The - /// returned [StreamedResponse.stream] ends or errors when the server closes - /// the connection — the caller treats that as "reconnect me". [cancel] aborts - /// an in-progress connection (e.g. on dispose). + /// Built for long-lived responses (Server-Sent Events). The client-wide 10 s + /// receive timeout does not apply — a feed may be legitimately silent for + /// minutes (the EEW stream sends nothing between earthquakes) — and is + /// replaced by [idleTimeout]: the longest the response may go without a + /// single byte before the stream errors with a receive timeout. Pass the + /// longest silence the server can produce on a healthy connection (its ping + /// cadence, with margin); zero disables it. The connect timeout still bounds + /// the initial handshake so a dead host fails over. The returned + /// [StreamedResponse.stream] ends or errors when the server closes the + /// connection, or when [idleTimeout] passes — the caller treats both as + /// "reconnect me". [cancel] aborts an in-progress connection (e.g. on + /// dispose). Future openStream( ApiTier tier, String path, { Map? query, Map? headers, + Duration idleTimeout = Duration.zero, }) async { final hosts = hostsFor(tier); for (var i = 0; i < hosts.length; i++) { @@ -234,9 +240,12 @@ class ApiClient { options: Options( method: 'GET', responseType: ResponseType.stream, - // Disabled, not merged from BaseOptions' 10s: a quiet SSE feed must - // not be torn down mid-connection just because no bytes arrived. - receiveTimeout: Duration.zero, + // Not merged from BaseOptions' 10s: a quiet SSE feed must not be + // torn down for ten seconds of nothing. Dio applies this between + // body chunks, so it bounds silence, not the stream's length — a + // half-open socket (no FIN after a handover) errors out instead of + // waiting forever. + receiveTimeout: idleTimeout, headers: headers, ), ); diff --git a/lib/core/network/sse_client.dart b/lib/core/network/sse_client.dart index c7d7838b5..bba080f54 100644 --- a/lib/core/network/sse_client.dart +++ b/lib/core/network/sse_client.dart @@ -5,6 +5,31 @@ import 'package:dpip/core/network/api_client.dart'; import 'package:dpip/core/network/api_region.dart'; import 'package:dpip/core/network/sse_event.dart'; +/// How often DPIP's SSE servers write a `: ping` comment on a stream that has +/// nothing else to send — the LB hubs (EEW) and the TREM stream (RTS) both. +/// +/// A fact about the servers, not a client setting: the silence limit below is +/// derived from it, so if the servers ping more often this is the one number to +/// change, and detection of a dead connection speeds up with it. +const Duration sseServerPingInterval = Duration(seconds: 60); + +/// How long an SSE connection may deliver nothing at all — no frame, no ping — +/// before it is treated as dead: 2.2 × [sseServerPingInterval]. +/// +/// A connection can die without closing. A network handover or a NAT mapping +/// that times out leaves a half-open socket: no FIN, no RST, no error, just +/// silence, and a client that reads silence as "nothing to report" shows a +/// live, calm feed that can no longer receive an alert. The servers' pings are +/// what make silence meaningful, so the bound is counted in pings. +/// +/// Two intervals, not one, plus 10 %: the LB hub only checks whether a ping is +/// due when its wait ends, and a wake that finds nothing new restarts that +/// wait — so the gap between two bytes on a healthy stream can approach two +/// full intervals. The 10 % covers scheduling and transit. Any tighter and a +/// healthy quiet stream would be torn down; any looser and a dead one passes +/// for quiet that much longer. +const Duration sseSilenceLimit = Duration(seconds: 132); + /// Opens a Server-Sent Events connection and parses its `text/event-stream` /// body into [SseEvent] frames — the transport seam a streaming feed sits on. /// @@ -37,6 +62,11 @@ class HttpSseClient implements SseClient { tier, path, query: query, + // The transport's own backstop for a half-open socket: it errors the + // stream after this long without a byte, so a dead connection is + // replaced even while nothing is polling the source (the app sits + // behind a notification shade, where the socket is deliberately kept). + idleTimeout: sseSilenceLimit, headers: const { 'Accept': 'text/event-stream', 'Cache-Control': 'no-cache', @@ -58,9 +88,11 @@ class HttpSseClient implements SseClient { /// Chunk-boundary safe: `utf8.decoder` and `LineSplitter` both buffer partial /// input across network chunks, so a multi-byte character or a line split /// mid-chunk is reassembled correctly. A frame is emitted on each blank-line - /// delimiter; `:`-prefixed comment lines (SSE heartbeats) are skipped; - /// multiple `data:` lines are joined with `\n` per the spec (trailing newline - /// stripped). Exposed for unit testing the parser directly. + /// delimiter; each `:`-prefixed comment line (a server keep-alive) is emitted + /// at once as an [SseEvent.comment], without touching the frame being + /// assembled around it; multiple `data:` lines are joined with `\n` per the + /// spec (trailing newline stripped). Exposed for unit testing the parser + /// directly. /// /// Uses `utf8.decoder.bind` rather than `bytes.transform(utf8.decoder)`: Dio's /// response stream is a `Stream` at runtime, and `Stream.transform` @@ -88,7 +120,13 @@ class HttpSseClient implements SseClient { dirty = false; continue; } - if (line.startsWith(':')) continue; // comment / heartbeat + if (line.startsWith(':')) { + // A keep-alive: no content, but proof the connection still delivers. + var text = line.substring(1); + if (text.startsWith(' ')) text = text.substring(1); + yield SseEvent.comment(text); + continue; + } dirty = true; final colon = line.indexOf(':'); final field = colon == -1 ? line : line.substring(0, colon); diff --git a/lib/core/network/sse_event.dart b/lib/core/network/sse_event.dart index 4517f5b80..61ed39bc5 100644 --- a/lib/core/network/sse_event.dart +++ b/lib/core/network/sse_event.dart @@ -10,10 +10,24 @@ /// - [retry] is the server's reconnect hint (`retry:` in ms), used to pace /// reconnection. /// +/// A `:` comment line is surfaced too, as its own [SseEvent.comment]. The spec +/// says to ignore comments, and as *content* they are ignored — [isDefault] is +/// false, so nothing decodes one — but the servers write `: ping` on a stream +/// with nothing else to say, and that ping is the only way a client can tell a +/// quiet connection from one that silently died. Dropping it in the parser +/// would leave the source blind to exactly the failure it most needs to see. +/// /// `id:` is not modelled: DPIP's feeds are snapshot streams (each event is the /// current state), so Last-Event-ID resumption is not used. class SseEvent { - const SseEvent({this.name, this.data = '', this.retry}); + const SseEvent({this.name, this.data = '', this.retry}) : isComment = false; + + /// A `:` comment line — a server keep-alive. [data] holds the text after the + /// colon (`ping`, `ping 1700000000000`), for logging only. + const SseEvent.comment([this.data = '']) + : name = null, + retry = null, + isComment = true; /// The `event:` type, or null for the default (unnamed) event. final String? name; @@ -24,10 +38,18 @@ class SseEvent { /// The server's `retry:` reconnect hint, or null if the frame carried none. final Duration? retry; + /// Whether this is a comment line rather than a frame: proof the connection + /// is alive, and nothing else. + final bool isComment; + /// Whether this is the default event that carries the feed payload (the one - /// the source decodes); named events (e.g. `info`) are ignored for payload. - bool get isDefault => name == null || name!.isEmpty || name == 'message'; + /// the source decodes); named events (e.g. `info`) and comments are ignored + /// for payload. + bool get isDefault => + !isComment && (name == null || name!.isEmpty || name == 'message'); @override - String toString() => 'SseEvent(name: $name, data: $data, retry: $retry)'; + String toString() => isComment + ? 'SseEvent.comment($data)' + : 'SseEvent(name: $name, data: $data, retry: $retry)'; } diff --git a/lib/core/platform/network_changes.dart b/lib/core/platform/network_changes.dart new file mode 100644 index 000000000..d01282483 --- /dev/null +++ b/lib/core/platform/network_changes.dart @@ -0,0 +1,70 @@ +/// Reports when the device's network changes underneath the app — Wi-Fi to +/// cellular, one Wi-Fi to another, offline to online — so a feed holding a +/// connection open can replace one bound to a network that is gone. +/// +/// A platform channel rather than a package, like the rest of `core/platform/`: +/// Android's default-network callback and iOS's `NWPathMonitor` each answer the +/// one question asked — which network traffic currently leaves through — and +/// each side is a single system callback. +/// +/// The platform sends an opaque identity per report (`wifi:`, +/// `cellular:pdp_ip0`, `none`); only whether it differs from the last one +/// matters here. Without reports the stream simply never fires, and the feeds' +/// own silence limit still catches a dead connection, only later. +library; + +import 'package:dpip/core/logging/log.dart'; +import 'package:flutter/foundation.dart' show visibleForTesting; +import 'package:flutter/services.dart'; + +abstract final class NetworkChanges { + const NetworkChanges._(); + + static const EventChannel _channel = EventChannel('com.exptech.dpip/network'); + + /// One event per real change of network, carrying the new identity. + /// + /// Single-subscription: the realtime service is its one listener. + static Stream get stream => + changesOf(_channel.receiveBroadcastStream()); + + /// The change filter, over any stream of identity reports. + /// + /// The first report is the network the app started on — a baseline, not a + /// change. After it, a report equal to the previous one is not a change + /// either: iOS re-reports the same path whenever any property of it moves + /// (cost, constrained mode), and acting on those would tear down a healthy + /// connection for nothing. An error from the platform is logged and skipped; + /// the reports after it still count. + /// + /// `Stream.multi` rather than an `async*` generator: a generator parked in + /// `await for` does not let a cancel through until the next report arrives, + /// which for a network that never changes again is never — the platform + /// callback would outlive its only listener. + @visibleForTesting + static Stream changesOf(Stream reports) => + Stream.multi((out) { + String? previous; + final subscription = reports.listen( + (report) { + final identity = '$report'; + if (previous == null) { + Log.debug('network: $identity'); + } else if (identity != previous) { + Log.info('network changed: $previous → $identity'); + out.add(identity); + } + previous = identity; + }, + // The OS refused the callback, or a report failed in transit. The + // feeds fall back on their silence limit meanwhile. + onError: (Object error) => + Log.warning('network change report failed: $error'), + onDone: out.close, + ); + out + ..onPause = subscription.pause + ..onResume = subscription.resume + ..onCancel = subscription.cancel; + }); +} diff --git a/lib/core/realtime/realtime_channel.dart b/lib/core/realtime/realtime_channel.dart index fabc938f3..ea89b9df3 100644 --- a/lib/core/realtime/realtime_channel.dart +++ b/lib/core/realtime/realtime_channel.dart @@ -40,6 +40,11 @@ abstract interface class RealtimeChannelBase { /// Forces an immediate poll, completing when a poll result has landed. Future refreshNow(); + /// Tells the source the device's network changed, so a held-open connection + /// bound to the old one can be replaced (see + /// [RealtimeSource.networkChanged]). Polling itself is unaffected. + void networkChanged(); + /// Cancels polling and closes the stream. void dispose(); } @@ -63,6 +68,9 @@ abstract interface class RealtimeChannelBase { /// re-flip the feed to live after [resume]. /// - **Emit once per real transition**: a new state is pushed only when the /// observable status or payload changes, not on every 1 Hz poll. +/// - **A fallback is never live**: data the source reports as coming from a +/// secondary path ([RealtimeSource.servingFallback]) is kept and shown, but +/// classifies at best as stale until the feed's own path answers again. class RealtimeChannel implements RealtimeChannelBase { RealtimeChannel({ required this._source, @@ -103,6 +111,10 @@ class RealtimeChannel implements RealtimeChannelBase { /// feeds like EEW; `serverNow − payloadTime` for payload-timestamped feeds). Duration? _ageBase; + /// Whether the current data came from the source's fallback path, which + /// caps its status at stale (see [RealtimeSource.servingFallback]). + bool _fallback = false; + /// Bumped whenever polling is invalidated (pause/dispose), so a fetch that /// completes after the fact is discarded instead of applied. int _epoch = 0; @@ -160,6 +172,12 @@ class RealtimeChannel implements RealtimeChannelBase { await _fetch(); } + @override + void networkChanged() { + if (_disposed) return; + _source.networkChanged(); + } + @override void dispose() { _disposed = true; @@ -187,6 +205,7 @@ class RealtimeChannel implements RealtimeChannelBase { sinceStart: _elapsed.elapsed - _startMark, staleAfter: _config.staleAfter, offlineAfter: _config.offlineAfter, + fallback: _fallback, ); /// Recomputes status from elapsed time only (no network) and emits on change. @@ -221,6 +240,13 @@ class RealtimeChannel implements RealtimeChannelBase { ? Duration.zero : now.difference(payloadTime); _lastSuccessMark = _elapsed.elapsed; + // A fallback answers every poll from its last snapshot until the + // next one lands, so a repeat is no newer than when it first came: + // it keeps that time rather than claiming this poll's. + final fallback = _source.servingFallback; + final repeat = + fallback && _fallback && identical(value, _current.data); + _fallback = fallback; final status = _classify(); final changed = status != _current.status || @@ -232,8 +258,8 @@ class RealtimeChannel implements RealtimeChannelBase { _current = RealtimeState( status: status, data: value, - dataTime: payloadTime ?? now, - fetchedAt: now, + dataTime: repeat ? _current.dataTime : payloadTime ?? now, + fetchedAt: repeat ? _current.fetchedAt : now, consecutiveFailures: 0, ); if (changed) _publish(); diff --git a/lib/core/realtime/realtime_service.dart b/lib/core/realtime/realtime_service.dart index cde61eee6..6ed9b206a 100644 --- a/lib/core/realtime/realtime_service.dart +++ b/lib/core/realtime/realtime_service.dart @@ -12,18 +12,26 @@ import 'package:dpip/core/realtime/ticker.dart'; /// [onBackground] directly (the `AppLifecycleListener` adapter lives in /// `realtime_lifecycle.dart`) and the resync cadence via an injected [Ticker]. class RealtimeService { + /// [networkChanges] fires once per real change of the device's network + /// (production: `NetworkChanges.stream` from `core/platform/`); each event is + /// fanned out to every channel from [startAll] until [dispose]. Null — tests, + /// and any platform without the channel — means connections are left to + /// notice a dead network through their own silence limit. RealtimeService( this._clock, { this._ticker = const SystemTicker(), this._clockSyncInterval = const Duration(seconds: 60), + this._networkChanges, }); final ServerClock _clock; final Ticker _ticker; final Duration _clockSyncInterval; + final Stream? _networkChanges; final List _channels = []; TickerHandle? _clockSyncHandle; + StreamSubscription? _networkSubscription; /// The shared corrected clock (also each channel's injected `Clock`). ServerClock get clock => _clock; @@ -51,6 +59,20 @@ class RealtimeService { } } _startClockSync(); + _networkSubscription ??= _networkChanges?.listen( + (_) => _onNetworkChanged(), + ); + } + + /// The device moved to a different network: every channel's source gets the + /// chance to replace a connection bound to the old one. Not gated on the + /// lifecycle — a source paused for the background ignores it itself, and + /// one merely interrupted (a notification shade) is holding a socket that + /// the change has just killed. + void _onNetworkChanged() { + for (final channel in _channels) { + channel.networkChanged(); + } } /// App went to background: pause every channel (stop polling, keep state), @@ -97,6 +119,8 @@ class RealtimeService { /// Disposes every channel and stops the resync ticker. void dispose() { _stopClockSync(); + _networkSubscription?.cancel(); + _networkSubscription = null; for (final channel in _channels) { channel.dispose(); } diff --git a/lib/core/realtime/realtime_source.dart b/lib/core/realtime/realtime_source.dart index 707a104a3..fcb75c09d 100644 --- a/lib/core/realtime/realtime_source.dart +++ b/lib/core/realtime/realtime_source.dart @@ -24,6 +24,16 @@ abstract class RealtimeSource { /// whose default `==` is identity (e.g. `List`). bool sameData(T? a, T? b) => identical(a, b) || a == b; + /// Whether the payload the latest successful [fetch] returned came from a + /// slower, secondary path rather than the feed's own — an SSE feed answering + /// from a periodic REST snapshot while its stream reconnects. + /// + /// The channel reads this after every successful fetch and never lets such a + /// payload classify as `live` (see `classifyStatus`): it is shown, flagged + /// as possibly out of date, and the feed returns to `live` only when its own + /// path answers again. A source with a single path leaves this false. + bool get servingFallback => false; + /// Whether a fetch failure means there is simply no data for the requested /// point in time, rather than a fault worth counting. The channel still /// records it as `lastFailure` — that is what a replay page reads to say the @@ -57,6 +67,21 @@ abstract class RealtimeSource { /// so the channel ages the feed with its ordinary logic. void resume() {} + /// The device's network changed underneath the app — Wi-Fi to cellular, one + /// Wi-Fi to another, offline to online. + /// + /// A poll source needs nothing here: its next request goes out on whatever + /// network is current. A **connection-holding** source is different, because + /// its socket is bound to the network it was opened on. When that network + /// goes, the socket rarely says so: no FIN arrives, and the connection just + /// stops delivering until something notices. Overriding this lets the source + /// replace it at once instead of waiting out its own silence limit. + /// + /// Called whether or not the app is in the foreground; a source that has + /// released its transport for the background simply ignores it, because + /// resuming opens a fresh connection anyway. + void networkChanged() {} + /// Releases any transport the source holds (e.g. an open SSE connection). A /// poll source is stateless per [fetch] and uses this no-op default; a /// connection-holding source (SSE) overrides it. The channel calls this from diff --git a/lib/core/realtime/sse_realtime_source.dart b/lib/core/realtime/sse_realtime_source.dart index 379ab3cab..052b5c1cd 100644 --- a/lib/core/realtime/sse_realtime_source.dart +++ b/lib/core/realtime/sse_realtime_source.dart @@ -1,11 +1,13 @@ import 'dart:async'; import 'dart:convert'; import 'dart:io'; +import 'dart:math'; import 'dart:typed_data'; import 'package:dpip/core/error/failure.dart'; import 'package:dpip/core/error/result.dart'; import 'package:dpip/core/logging/log.dart'; +import 'package:dpip/core/network/sse_client.dart'; import 'package:dpip/core/network/sse_event.dart'; import 'package:dpip/core/realtime/elapsed.dart'; import 'package:dpip/core/realtime/realtime_source.dart'; @@ -13,11 +15,18 @@ import 'package:flutter/foundation.dart' show protected; /// How a source decides an SSE feed is currently "alive", given that a poll /// channel measures freshness from the last successful [RealtimeSource.fetch]. +/// +/// Under either mode a connection that has received nothing at all — not a +/// frame, not a server ping — for longer than [SseLiveness.silenceLimit] is +/// dead, whatever it last said: the source reports a failure and replaces it. enum SseLivenessMode { - /// The connection being **open** is the liveness signal, independent of event - /// recency. For a **bursty** feed that is silent between events (EEW sends - /// nothing between earthquakes), so a quiet-but-connected feed is correctly - /// `live` (connected, no alert) rather than aging to offline on silence. + /// The connection being **open and still delivering** is the liveness + /// signal, independent of event recency. For a **bursty** feed that is + /// silent between events (EEW sends nothing between earthquakes), so a + /// quiet-but-connected feed is correctly `live` (connected, no alert) rather + /// than aging to offline on silence. "Delivering" is what the servers' pings + /// are for: open alone is not enough, because a socket left half-open by a + /// network handover stays open forever and delivers nothing. connectionOpen, /// A **recent event** is the liveness signal: `fetch` returns `Err` once no @@ -26,20 +35,27 @@ enum SseLivenessMode { eventRecency, } -/// Liveness policy for [SseRealtimeSource]: a [SseLivenessMode] plus the window -/// used by [SseLivenessMode.eventRecency]. +/// Liveness policy for [SseRealtimeSource]: a [SseLivenessMode], the window +/// used by [SseLivenessMode.eventRecency], and the silence bound both share. class SseLiveness { - const SseLiveness.connectionOpen() + const SseLiveness.connectionOpen({this.silenceLimit = sseSilenceLimit}) : mode = SseLivenessMode.connectionOpen, window = null; - const SseLiveness.eventRecency(Duration this.window) - : mode = SseLivenessMode.eventRecency; + const SseLiveness.eventRecency( + Duration this.window, { + this.silenceLimit = sseSilenceLimit, + }) : mode = SseLivenessMode.eventRecency; final SseLivenessMode mode; /// The recency window for [SseLivenessMode.eventRecency]; null otherwise. final Duration? window; + + /// How long the connection may receive nothing — no frame, no ping — before + /// it is treated as dead and replaced. Defaults to [sseSilenceLimit], which + /// is derived from the servers' ping cadence. + final Duration silenceLimit; } /// A [RealtimeSource] whose transport is a Server-Sent Events connection, @@ -56,6 +72,13 @@ class SseLiveness { /// whatever [decode] parses from each event's `data:`, i.e. the same JSON the /// one-shot GET returns. /// +/// A drop the socket never reports counts as a drop too. Every byte that +/// arrives — a frame or a server ping — is timed, and a connection silent past +/// [SseLiveness.silenceLimit] is cancelled and reconnected, with [fetch] +/// answering `Err` meanwhile. Without that, a socket a network handover left +/// half-open would read as a connected, quiet feed for as long as the app +/// stayed open. +/// /// Subclasses supply the feed specifics ([decode] plus /// [RealtimeSource.timestampOf]/[RealtimeSource.sameData]); the connection /// factory, liveness policy and payload event name come through the @@ -66,10 +89,12 @@ abstract class SseRealtimeSource extends RealtimeSource { this._liveness = const SseLiveness.connectionOpen(), Elapsed? elapsed, Future Function(Duration delay)? delay, + Random? random, this._label = 'sse', this._payloadEvent = 'g', }) : _elapsed = elapsed ?? SystemElapsed(), - _delay = delay ?? _defaultDelay; + _delay = delay ?? _defaultDelay, + _random = random ?? Random(); static Future _defaultDelay(Duration d) => Future.delayed(d); @@ -78,6 +103,9 @@ abstract class SseRealtimeSource extends RealtimeSource { final SseLiveness _liveness; final Elapsed _elapsed; final Future Function(Duration delay) _delay; + + /// Spreads reconnect delays; injectable so a test can pin the spread. + final Random _random; final String _label; /// The event name whose `data:` is a base64-gzipped payload: `g` under the @@ -122,6 +150,11 @@ abstract class SseRealtimeSource extends RealtimeSource { /// decoded elsewhere when that happens cannot bring it back. int _snapshotEpoch = 0; + /// When the current connection last received anything — a frame or a ping — + /// or, until it does, when it was opened. What [SseLiveness.silenceLimit] is + /// measured from. + Duration? _lastByteMark; + /// Decodes a default-event `data:` payload into `T`. The payload is the same /// JSON the one-shot GET returns, so this mirrors the repository's mapping. T decode(String data); @@ -160,6 +193,10 @@ abstract class SseRealtimeSource extends RealtimeSource { /// stream's sleep mode sends an RTS frame only while a station is alerting) /// returns its calm value here: its silence *is* the answer, and reading it /// as a dead feed would age a healthy connection to offline between events. + /// + /// Only while the connection is still delivering, though. Past + /// [SseLiveness.silenceLimit] without even a ping, [fetch] reports a failure + /// and never consults this — a dead socket cannot vouch for calm. T? get quiet => null; @override @@ -168,6 +205,10 @@ abstract class SseRealtimeSource extends RealtimeSource { return const Err(NetworkFailure('SSE source disposed')); } _ensureStarted(); + if (_silentTooLong) { + _dropSilentConnection(); + return Err(NetworkFailure('$_label SSE went silent')); + } if (!_connected) return Err(NetworkFailure('$_label SSE not connected')); if (_hasSnapshot && _isFresh) return Ok(_latest as T); final quiet = this.quiet; @@ -175,6 +216,28 @@ abstract class SseRealtimeSource extends RealtimeSource { return Err(NetworkFailure('$_label SSE has nothing fresh')); } + /// Whether the open connection has gone longer than the silence limit + /// without receiving a byte — the signature of a half-open socket. + bool get _silentTooLong { + final mark = _lastByteMark; + return _subscription != null && + mark != null && + _elapsed.elapsed - mark > _liveness.silenceLimit; + } + + /// Replaces a connection that stopped delivering. Cancelling raises no + /// `onDone`, so the reconnect is queued here, exactly as a real close would. + void _dropSilentConnection() { + final silence = _elapsed.elapsed - _lastByteMark!; + Log.warning( + '[$_label] SSE silent for ${silence.inSeconds}s ' + '(limit ${_liveness.silenceLimit.inSeconds}s) — ' + 'treating the connection as dead and reconnecting', + ); + _subscription?.cancel(); + _onClosed(); + } + /// Whether the buffered payload counts as fresh under the liveness policy. bool get _isFresh { final window = _liveness.window; @@ -200,6 +263,7 @@ abstract class SseRealtimeSource extends RealtimeSource { _subscription = null; _connected = false; _dropSnapshot(); + _lastByteMark = null; } /// Re-arms the lazy open, so the channel's first post-resume `fetch` builds a @@ -214,6 +278,47 @@ abstract class SseRealtimeSource extends RealtimeSource { _attempt = 0; } + /// Replaces the connection after a change of network, with the backoff + /// reset and after a short random settle of 1–2 s. + /// + /// The current connection is dropped at once rather than trusted until the + /// new one is up: it was bound to the old network, and the common cases — a + /// handover to cellular, a drop to no network at all — leave it dead without + /// any error to say so. Answering `Err` for the second or two until the new + /// connection speaks is honest; answering from the old one would claim a + /// liveness nothing is backing. + /// + /// The settle is for the new network, whose routes and DNS are often not + /// usable in the first instant, and for the server: a cell handover moves a + /// whole area's phones at once, and a fixed delay would bring them all back + /// in the same instant. A further change inside the settle restarts it. + /// + /// Before the first [fetch], or while paused for the background, there is + /// nothing to replace — the next open is on the new network anyway. + @override + void networkChanged() { + if (_disposed || _paused || !_started) return; + Log.info('[$_label] network changed — replacing the SSE connection'); + _generation++; // a backoff counting down belongs to the old network + _attempt = 0; + _pending?.cancel(); + _pending = null; + _subscription?.cancel(); + _subscription = null; + _connected = false; + _hasSnapshot = false; + _lastByteMark = null; + final generation = _generation; + _delay(_networkSettleDelay()).then((_) { + if (_disposed || generation != _generation) return; + _openConnection(); + }); + } + + /// 1–2 s, uniformly at random, to the millisecond. + Duration _networkSettleDelay() => + Duration(milliseconds: 1000 + _random.nextInt(1000)); + void _ensureStarted() { if (_started || _disposed || _paused) return; _started = true; @@ -233,6 +338,9 @@ abstract class SseRealtimeSource extends RealtimeSource { // rather than fail, and an orphaned SSE socket is exactly the kind of leak // this class exists to avoid. _subscription?.cancel(); + // Silence counts from the open: a connection that never says anything is + // as dead as one that stopped. + _lastByteMark = _elapsed.elapsed; _subscription = _connect().listen( _onEvent, onError: _onError, @@ -296,6 +404,9 @@ abstract class SseRealtimeSource extends RealtimeSource { if (_disposed) return; _connected = true; _attempt = 0; // the connection is delivering — reset the backoff + _lastByteMark = _elapsed.elapsed; + // A server ping: proof the connection is alive, and nothing more. + if (event.isComment) return; final retry = event.retry; if (retry != null) _serverRetry = retry; // A payload arrives either as the default event (plain JSON) or as the @@ -389,12 +500,20 @@ abstract class SseRealtimeSource extends RealtimeSource { }); } - /// 1s, 2s, then capped at the server's `retry:` hint (default 3s). Reset to 1s - /// whenever a connection delivers an event, so a transient blip recovers fast - /// while a sustained outage does not hammer the server. + /// 1s, 2s, then capped at the server's `retry:` hint (default 3s), each + /// spread ±50 % at random. Reset to 1s whenever a connection delivers an + /// event, so a transient blip recovers fast while a sustained outage does + /// not hammer the server. + /// + /// The spread is for the server, not this client. When a region drops, every + /// phone connected to it loses its stream in the same instant; on a fixed + /// schedule they would all come back in the same instant too, once a second + /// and then every three, and the region that just recovered takes the whole + /// audience as one spike on each beat. Jitter turns the spike into a ramp. Duration _backoffDelay() { - final base = Duration(seconds: 1 << _attempt.clamp(0, 2)); - return base < _serverRetry ? base : _serverRetry; + final step = Duration(seconds: 1 << _attempt.clamp(0, 2)); + final base = step < _serverRetry ? step : _serverRetry; + return base * (0.5 + _random.nextDouble()); } @override diff --git a/lib/core/realtime/staleness.dart b/lib/core/realtime/staleness.dart index b6202a272..9cf2edbfa 100644 --- a/lib/core/realtime/staleness.dart +++ b/lib/core/realtime/staleness.dart @@ -14,11 +14,20 @@ import 'package:dpip/core/realtime/realtime_state.dart'; /// A negative [age] (data timestamped in the future — a clock anomaly) is treated /// as **not current**: the classifier fails toward [RealtimeStatus.stale] rather /// than presenting possibly-bad data as live. +/// +/// [fallback] marks data delivered by a feed's secondary path — an SSE feed +/// answering from a periodic REST snapshot while its stream reconnects. Such +/// data is real and recent, so it is shown rather than blanked, but it is at +/// best [RealtimeStatus.stale], never [RealtimeStatus.live]: `live` promises +/// that a change reaches the screen within the feed's own latency, and a +/// snapshot taken every few seconds cannot keep that promise however fresh the +/// last one was. It still ages to offline on the same schedule. RealtimeStatus classifyStatus({ required Duration? age, required Duration sinceStart, required Duration staleAfter, required Duration offlineAfter, + bool fallback = false, }) { if (age == null) { return sinceStart <= offlineAfter @@ -26,7 +35,9 @@ RealtimeStatus classifyStatus({ : RealtimeStatus.offline; } if (age.isNegative) return RealtimeStatus.stale; - if (age <= staleAfter) return RealtimeStatus.live; + if (age <= staleAfter) { + return fallback ? RealtimeStatus.stale : RealtimeStatus.live; + } if (age <= offlineAfter) return RealtimeStatus.stale; return RealtimeStatus.offline; } diff --git a/lib/features/earthquake/data/eew_realtime_source.dart b/lib/features/earthquake/data/eew_realtime_source.dart index 819026b96..5e3e53863 100644 --- a/lib/features/earthquake/data/eew_realtime_source.dart +++ b/lib/features/earthquake/data/eew_realtime_source.dart @@ -1,6 +1,13 @@ +import 'dart:async'; import 'dart:convert'; +import 'dart:math'; +import 'package:dpip/core/error/failure.dart'; +import 'package:dpip/core/error/result.dart'; +import 'package:dpip/core/logging/log.dart'; +import 'package:dpip/core/network/api_exception.dart'; import 'package:dpip/core/network/sse_event.dart'; +import 'package:dpip/core/realtime/elapsed.dart'; import 'package:dpip/core/realtime/sse_realtime_source.dart'; import 'package:dpip/features/earthquake/domain/eew.dart'; import 'package:flutter/foundation.dart' show listEquals; @@ -13,30 +20,208 @@ import 'package:flutter/foundation.dart' show listEquals; /// data format is unchanged by the move to SSE. /// /// Liveness is [SseLiveness.connectionOpen] (the default): the EEW stream is -/// **silent between earthquakes** (empirically tens of seconds with no bytes), -/// so an open connection — not event recency — is the liveness signal; a -/// quiet-but-connected feed stays `live` (connected, no alert) instead of aging -/// to offline. This mirrors the previous fetch-freshness choice ([timestampOf] -/// is null): feed liveness, not payload age, drives staleness. +/// **silent between earthquakes** apart from the server's keep-alive pings, so +/// an open connection that is still delivering — not event recency — is the +/// liveness signal; a quiet-but-connected feed stays `live` (connected, no +/// alert) instead of aging to offline, while one that has missed its pings is +/// dropped and reconnected rather than trusted. This mirrors the previous +/// fetch-freshness choice ([timestampOf] is null): feed liveness, not payload +/// age, drives staleness. +/// +/// **While the stream is down, the one-shot GET stands in for it.** A network +/// that will not hold a long-lived connection — or a stream that keeps +/// dropping — would otherwise leave the alert list frozen at whatever it was +/// when the stream last spoke, and an earthquake that starts during the outage +/// would never reach the screen at all. So once the stream has been failing +/// for [fallbackGrace], the source polls the same list every +/// [fallbackInterval] and answers from the latest snapshot, through the same +/// decode and agency filters as a streamed frame. +/// +/// That answer is flagged as a fallback ([servingFallback]), and the channel +/// never classifies it better than stale: a snapshot every ten seconds cannot +/// promise what `live` promises, that an alert reaches the screen within a +/// second of the server having it. The data is shown — flagged as possibly out +/// of date, which is the truth — and the feed is `live` again only once the +/// stream answers, at which point polling stops at once and any poll still in +/// flight is discarded. class EewRealtimeSource extends SseRealtimeSource> { /// [connect] opens one EEW SSE connection — in production /// `EarthquakeApi.openEewSse`; the source calls it again to reconnect. /// - /// [cwaOnly] is read fresh on every [decode] (a closure, not a captured - /// bool) so toggling `EewCwaOnlySettings` takes effect on the very next - /// event instead of needing the connection to be rebuilt. + /// [poll] fetches the same alert list once, as raw JSON — in production + /// `EarthquakeApi.getEewRealtime`. Without it there is no fallback. + /// + /// [cwaOnly] is read fresh on every [decode] and every poll (a closure, not + /// a captured bool) so toggling `EewCwaOnlySettings` takes effect on the + /// very next event instead of needing the connection to be rebuilt. EewRealtimeSource( Stream Function() connect, { - required this.cwaOnly, - }) : super(connect: connect, label: 'eew'); + required bool Function() cwaOnly, + Future> Function()? poll, + Elapsed? elapsed, + Future Function(Duration delay)? delay, + Random? random, + }) : this._( + connect, + cwaOnly, + poll, + elapsed ?? SystemElapsed(), + delay, + random, + ); + + EewRealtimeSource._( + Stream Function() connect, + this.cwaOnly, + this._poll, + this._monotonic, + Future Function(Duration delay)? delay, + Random? random, + ) : super( + connect: connect, + label: 'eew', + elapsed: _monotonic, + delay: delay, + random: random, + ); + + /// How long the stream must have been failing before the fallback starts: + /// past the feed's stale threshold, so the ordinary reconnect after a drop — + /// a second or two — costs the server no extra request, and an app that is + /// just starting does not poll before its stream has had a chance. + static const Duration fallbackGrace = Duration(seconds: 3); + + /// How often the alert list is polled while the stream is down. + static const Duration fallbackInterval = Duration(seconds: 10); + + /// How long one polled snapshot is answered with after it lands: one + /// interval, plus room for the next poll's round trip. Past it the polls + /// have stopped landing, the source answers `Err`, and the channel ages the + /// feed to offline on its usual schedule. + static const Duration fallbackHold = Duration(seconds: 15); final bool Function() cwaOnly; + final Future> Function()? _poll; + final Elapsed _monotonic; + + /// When the stream started failing, or null while it answers. + Duration? _downSince; + + /// When the last poll was sent, or null if none has been this outage. + Duration? _lastPollAt; + + bool _polling = false; + bool _closed = false; + + /// Bumped whenever the fallback stops, so a poll still in flight cannot + /// land after the stream came back and replace a streamed answer. + int _pollEpoch = 0; + + /// The latest polled snapshot and when it landed. + ({List alerts, Duration at})? _snapshot; + + bool _servingFallback = false; @override - List decode(String data) { + bool get servingFallback => _servingFallback; + + @override + Future>> fetch() async { + final streamed = await super.fetch(); + final poll = _poll; + if (poll == null || _closed) return streamed; + if (streamed.isOk) { + _stopFallback(); + return streamed; + } + + final now = _monotonic.elapsed; + final downSince = _downSince ??= now; + final lastPoll = _lastPollAt; + if (!_polling && + now - downSince >= fallbackGrace && + (lastPoll == null || now - lastPoll >= fallbackInterval)) { + _startPoll(poll, now); + } + + final snapshot = _snapshot; + if (snapshot == null || now - snapshot.at > fallbackHold) { + _servingFallback = false; + return streamed; + } + _servingFallback = true; + return Ok(snapshot.alerts); + } + + void _startPoll(Future> Function() poll, Duration now) { + if (_lastPollAt == null) { + Log.warning( + '[eew] SSE down for ${fallbackGrace.inSeconds}s — polling the alert ' + 'list every ${fallbackInterval.inSeconds}s until it is back', + ); + } + _polling = true; + _lastPollAt = now; + final epoch = _pollEpoch; + unawaited(_runPoll(poll, epoch)); + } + + Future _runPoll( + Future> Function() poll, + int epoch, + ) async { + final result = await guardResult( + () async => _alertsFrom(await poll()), + // An unreachable server is what an outage looks like and is already + // being logged once a second by the channel; a body that does not + // decode is a fault, and keeps the full report. + shouldLog: (failure) => + failure is! NetworkFailure && failure is! TimeoutFailure, + ); + if (epoch != _pollEpoch) return; // the stream came back meanwhile + _polling = false; + switch (result) { + case Ok(:final value): + _snapshot = (alerts: value, at: _monotonic.elapsed); + case Err(:final failure): + Log.warning('[eew] fallback poll failed: ${failure.message}'); + } + } + + void _stopFallback() { + if (_lastPollAt != null) Log.info('[eew] SSE is back — polling stopped'); + _pollEpoch++; + _downSince = null; + _lastPollAt = null; + _polling = false; + _snapshot = null; + _servingFallback = false; + } + + /// A background stint ends the outage as far as the fallback is concerned: + /// resuming opens a fresh stream, which deserves its own grace before any + /// poll — and a snapshot from before the pause is not one to answer with. + @override + void pause() { + _stopFallback(); + super.pause(); + } + + @override + void dispose() { + _closed = true; + _stopFallback(); + super.dispose(); + } + + @override + List decode(String data) => _alertsFrom(jsonDecode(data) as List); + + /// The one mapping both paths share: JMA never, and only CWA when the user + /// asked for that. + List _alertsFrom(List raw) { final all = [ - for (final item in jsonDecode(data) as List) - Eew.fromJson(item as Map), + for (final item in raw) Eew.fromJson(item as Map), ]; final onlyCwa = cwaOnly(); return [ diff --git a/lib/features/earthquake/data/rts_realtime_source.dart b/lib/features/earthquake/data/rts_realtime_source.dart index 02470a5a2..7c2936b54 100644 --- a/lib/features/earthquake/data/rts_realtime_source.dart +++ b/lib/features/earthquake/data/rts_realtime_source.dart @@ -17,7 +17,9 @@ import 'package:dpip/features/earthquake/domain/rts_live_demand.dart'; /// should be talking means trouble, and the channel ages it to stale. Asleep, /// the server sends a frame only while a station is alerting, so silence is /// the normal, calm answer: the source reports an empty frame (nothing -/// alerting) for as long as the connection is open, instead of a failure. +/// alerting) for as long as the connection is open and still delivering the +/// server's pings, instead of a failure. A connection that has missed them is +/// not calm but dead, and is reported and replaced as such. /// /// A change of speed swaps the connection without a gap (see /// [SseRealtimeSource.renew]). Waking is immediate — the monitor has just @@ -74,7 +76,8 @@ class RtsRealtimeSource extends SseRealtimeSource { renew(); } - /// Asleep, a connection with nothing to say is reporting calm. + /// Asleep, a connection with nothing to say is reporting calm — up to the + /// silence limit, past which the base reports a failure without asking. @override Rts? get quiet => _live ? null : const Rts(); diff --git a/lib/features/earthquake/earthquake_providers.dart b/lib/features/earthquake/earthquake_providers.dart index e86c5d31a..04bfea2f0 100644 --- a/lib/features/earthquake/earthquake_providers.dart +++ b/lib/features/earthquake/earthquake_providers.dart @@ -74,7 +74,8 @@ List earthquakeProviders(SharedDeps deps) { final boxGrid = const RtsBoxGridSource().load(); // Live EEW over SSE (`/api/v2/eq/eew?sse=1`) — bursty, connection-open - // liveness. Debug runs with `DPIP_DEMO_MONITOR=1` swap in a synthetic alert + // liveness, with the one-shot GET polled as a flagged fallback while the + // stream is down. Debug runs with `DPIP_DEMO_MONITOR=1` swap in a synthetic alert // instead (see monitor_demo.dart) — its parameters are read from the newest // real earthquake report, so the 強震監視器 shows a plausible wavefront // without waiting for a live event. @@ -82,8 +83,11 @@ List earthquakeProviders(SharedDeps deps) { ? DemoEewSource(reports) as RealtimeSource> : kStartupEewDemoEnabled ? StartupEewDemoSource() as RealtimeSource> - : EewRealtimeSource(api.openEewSse, cwaOnly: () => eewCwaOnly.enabled) - as RealtimeSource>; + : EewRealtimeSource( + api.openEewSse, + cwaOnly: () => eewCwaOnly.enabled, + poll: api.getEewRealtime, + ) as RealtimeSource>; final eewChannel = RealtimeChannel>( source: eewSource, clock: deps.serverClock, diff --git a/test/core/network/api_client_test.dart b/test/core/network/api_client_test.dart index aa72348e9..11f4eb569 100644 --- a/test/core/network/api_client_test.dart +++ b/test/core/network/api_client_test.dart @@ -1,3 +1,4 @@ +import 'dart:async'; import 'dart:typed_data'; import 'package:dio/dio.dart'; @@ -323,6 +324,36 @@ void main() { expect(adapter.hits, hasLength(1)); }); + test( + 'openStream errors a stream that falls silent past idleTimeout', + () async { + // A half-open socket after a network handover sends nothing and never + // closes. Without a bound the stream waits on it forever. + final silent = StreamController(); + addTearDown(silent.close); + final adapter = _FakeAdapter((_, _) => ResponseBody(silent.stream, 200)); + final opened = await clientWith(adapter).openStream( + ApiTier.lbApi, + '/sse', + idleTimeout: const Duration(milliseconds: 50), + ); + silent.add(Uint8List.fromList([1])); + await expectLater( + opened.stream, + emitsInOrder([ + isA(), + emitsError( + isA().having( + (e) => e.type, + 'type', + DioExceptionType.receiveTimeout, + ), + ), + ]), + ); + }, + ); + test('a stream 5xx fails over and a 4xx does not', () async { final health = EndpointHealthMonitor(); final failing = _FakeAdapter( diff --git a/test/core/network/sse_client_test.dart b/test/core/network/sse_client_test.dart index 0b8ac7668..b0393f0b2 100644 --- a/test/core/network/sse_client_test.dart +++ b/test/core/network/sse_client_test.dart @@ -54,12 +54,27 @@ void main() { expect(events.map((e) => e.data), ['\n', 'a\n', '']); }); - test('skips comment / heartbeat lines', () async { + test('surfaces a ping as a comment, never as a payload', () async { + // The servers' keep-alive. It carries nothing, but it is the only thing + // that tells a quiet connection from a dead one, so it has to get out. final events = await HttpSseClient.parse( - _bytes([': keep-alive\ndata: x\n\n']), + _bytes([': ping\n\n', ':ping 1700000000000\n\n']), ).toList(); - expect(events, hasLength(1)); - expect(events.single.data, 'x'); + expect(events, hasLength(2)); + expect(events.every((e) => e.isComment), isTrue); + expect(events.any((e) => e.isDefault), isFalse); + expect(events.map((e) => e.data), ['ping', 'ping 1700000000000']); + }); + + test('a comment inside a frame leaves the frame intact', () async { + final events = await HttpSseClient.parse( + _bytes(['event: g\n: keep-alive\ndata: x\n\n']), + ).toList(); + expect(events, hasLength(2)); + expect(events.first.isComment, isTrue); + expect(events.last.name, 'g'); + expect(events.last.data, 'x'); + expect(events.last.isComment, isFalse); }); test( diff --git a/test/core/network/sse_event_test.dart b/test/core/network/sse_event_test.dart index 53b0d61b2..b802982e5 100644 --- a/test/core/network/sse_event_test.dart +++ b/test/core/network/sse_event_test.dart @@ -9,6 +9,15 @@ void main() { expect(const SseEvent(name: 'info').isDefault, isFalse); }); + test('a comment is never the payload event', () { + const ping = SseEvent.comment('ping'); + expect(ping.isComment, isTrue); + expect(ping.isDefault, isFalse, reason: 'unnamed, yet not a payload'); + expect(ping.name, isNull); + expect(const SseEvent(data: '{}').isComment, isFalse); + expect(ping.toString(), 'SseEvent.comment(ping)'); + }); + test('toString names the fields a log line needs', () { expect( const SseEvent( diff --git a/test/core/platform/network_changes_test.dart b/test/core/platform/network_changes_test.dart new file mode 100644 index 000000000..2f92da20f --- /dev/null +++ b/test/core/platform/network_changes_test.dart @@ -0,0 +1,85 @@ +/// Each event out of [NetworkChanges] tears down and rebuilds every realtime +/// connection, so what counts as a change is the whole contract: the network +/// the app starts on is not one, a repeat of the same network is not one +/// (iOS re-reports a path whenever any property of it moves), and a failed +/// report must not end the watch. +library; + +import 'dart:async'; + +import 'package:dpip/core/platform/network_changes.dart'; +import 'package:flutter/services.dart'; +import 'package:flutter_test/flutter_test.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + test('the first report is a baseline, not a change', () async { + final changes = await NetworkChanges.changesOf( + Stream.fromIterable(['wifi:100']), + ).toList(); + expect(changes, isEmpty); + }); + + test('only a different network is a change', () async { + final changes = await NetworkChanges.changesOf( + Stream.fromIterable([ + 'wifi:100', + 'wifi:100', // the same path, re-reported + 'cellular:200', // Wi-Fi gone, on cellular + 'none', // offline + 'none', + 'cellular:201', // back on a new cellular network + 'wifi:300', // a different Wi-Fi + ]), + ).toList(); + expect(changes, ['cellular:200', 'none', 'cellular:201', 'wifi:300']); + }); + + test('a failed report is skipped, not the end of the stream', () async { + final reports = StreamController(); + final done = NetworkChanges.changesOf(reports.stream).toList(); + reports + ..add('wifi:1') + ..addError(PlatformException(code: 'UNAVAILABLE')) + ..add('cellular:2'); + await reports.close(); + expect(await done, ['cellular:2']); + }); + + test('cancelling reaches the platform at once', () async { + // The reports may never come again; the listener must still be able to + // let go of the platform callback. + final reports = StreamController(); + final subscription = NetworkChanges.changesOf(reports.stream).listen(null); + reports.add('wifi:1'); + await pumpEventQueue(); + expect(reports.hasListener, isTrue); + await subscription.cancel(); + expect(reports.hasListener, isFalse); + }); + + test('the channel reports reach the stream', () async { + const channel = EventChannel('com.exptech.dpip/network'); + final messenger = + TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger; + final events = StreamController(); + messenger.setMockStreamHandler( + channel, + MockStreamHandler.inline( + onListen: (_, sink) => events.stream.listen(sink.success), + ), + ); + addTearDown(() => messenger.setMockStreamHandler(channel, null)); + + final seen = []; + final subscription = NetworkChanges.stream.listen(seen.add); + events + ..add('wifi:1') + ..add('cellular:2'); + await pumpEventQueue(); + expect(seen, ['cellular:2']); + await subscription.cancel(); + await events.close(); + }); +} diff --git a/test/core/realtime/realtime_channel_test.dart b/test/core/realtime/realtime_channel_test.dart index a1cfa1bed..9c106de7d 100644 --- a/test/core/realtime/realtime_channel_test.dart +++ b/test/core/realtime/realtime_channel_test.dart @@ -70,6 +70,17 @@ class _FakeSource extends RealtimeSource { @override bool isIgnorableFailure(Failure failure) => ignorable?.call(failure) ?? false; + + int networkChanges = 0; + + /// What [servingFallback] reports for the next fetch's answer. + bool fallback = false; + + @override + bool get servingFallback => fallback; + + @override + void networkChanged() => networkChanges++; } /// Flushes pending microtasks so broadcast emissions and unawaited fetches land. @@ -353,4 +364,100 @@ void main() { expect(source.fetchCount, fetches); expect(events.length, emissions); }); + + test('a network change reaches the source, never after dispose', () async { + channel.start(); + channel.networkChanged(); + expect(source.networkChanges, 1); + + channel.dispose(); + channel.networkChanged(); + expect(source.networkChanges, 1); + }); + + group('fallback answers', () { + test('are shown but never live', () async { + source + ..next = const Ok(7) + ..fallback = true; + channel.start(); + await pump(); + expect(channel.state.status, RealtimeStatus.stale); + expect(channel.state.data, 7, reason: 'shown, not blanked'); + + ticker.fire(); // every later poll answered the same way + await pump(); + expect(channel.state.status, RealtimeStatus.stale); + }); + + test('a repeat keeps the time its snapshot first arrived', () async { + final value = [1]; + final fallbackSource = _ListSource() + ..next = Ok(value) + ..fallback = true; + final listChannel = RealtimeChannel>( + source: fallbackSource, + clock: clock, + elapsed: elapsed, + ticker: ticker, + config: RealtimeConfig.eew, + ); + addTearDown(listChannel.dispose); + listChannel.start(); + await pump(); + final first = listChannel.state.dataTime; + + clock.current = clock.current.add(const Duration(seconds: 5)); + await listChannel.refreshNow(); // the same snapshot again + expect(listChannel.state.dataTime, first); + + fallbackSource.next = Ok([1]); // a new snapshot, equal content + await listChannel.refreshNow(); + expect(listChannel.state.dataTime, clock.current); + }); + + test('the feed is live again the moment its own path answers', () async { + source + ..next = const Ok(7) + ..fallback = true; + channel.start(); + await pump(); + expect(channel.state.status, RealtimeStatus.stale); + + source + ..next = const Ok(8) + ..fallback = false; + await channel.refreshNow(); + expect(channel.state.status, RealtimeStatus.live); + expect(channel.state.data, 8); + }); + + test('age to offline once the fallback stops answering too', () async { + source + ..next = const Ok(7) + ..fallback = true; + channel.start(); + await pump(); + + source.next = const Err(TimeoutFailure('down')); + elapsed.advance(const Duration(seconds: 11)); + ticker.fire(); + await pump(); + expect(channel.state.status, RealtimeStatus.offline); + }); + }); +} + +class _ListSource extends RealtimeSource> { + Result> next = const Ok([]); + bool fallback = false; + + @override + Future>> fetch() async => next; + + @override + bool get servingFallback => fallback; + + @override + DateTime? timestampOf(List value) => null; } diff --git a/test/core/realtime/realtime_service_test.dart b/test/core/realtime/realtime_service_test.dart index cba76aed6..948b3ac26 100644 --- a/test/core/realtime/realtime_service_test.dart +++ b/test/core/realtime/realtime_service_test.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:dpip/core/error/result.dart'; import 'package:dpip/core/realtime/clock.dart'; import 'package:dpip/core/realtime/elapsed.dart'; @@ -68,6 +70,8 @@ class _RecordingChannel implements RealtimeChannelBase { @override Future refreshNow() async => log.add('$id:refresh'); @override + void networkChanged() => log.add('$id:network'); + @override void dispose() => log.add('$id:dispose'); } @@ -155,4 +159,51 @@ void main() { expect(source.calls, greaterThanOrEqualTo(1), reason: 'immediate resync'); expect(ticker.startCount, 1, reason: 'periodic sync restarted'); }); + + group('network changes', () { + late StreamController changes; + + setUp(() { + changes = StreamController(); + service = RealtimeService( + ServerClock(_FixedClock(), _FakeElapsed(), source), + ticker: ticker, + networkChanges: changes.stream, + ); + log.clear(); + service.register(a); + service.register(b); + }); + + test('each change reaches every channel once started', () async { + service.startAll(); + log.clear(); + changes.add(null); + await pumpEventQueue(); + expect(log, ['a:network', 'b:network']); + }); + + test('nothing listens before startAll', () { + expect(changes.hasListener, isFalse); + service.startAll(); + expect(changes.hasListener, isTrue); + }); + + test('a change after an interruption still reaches the channels', () async { + // Behind a notification shade the socket is kept on purpose; a change of + // network is exactly what kills it, so the shade must not mute this. + service.startAll(); + service.onInterrupted(); + log.clear(); + changes.add(null); + await pumpEventQueue(); + expect(log, ['a:network', 'b:network']); + }); + + test('dispose stops listening', () { + service.startAll(); + service.dispose(); + expect(changes.hasListener, isFalse); + }); + }); } diff --git a/test/core/realtime/sse_realtime_source_test.dart b/test/core/realtime/sse_realtime_source_test.dart index 6cb5ff571..d48a4733b 100644 --- a/test/core/realtime/sse_realtime_source_test.dart +++ b/test/core/realtime/sse_realtime_source_test.dart @@ -2,7 +2,9 @@ import 'dart:async'; import 'dart:typed_data'; import 'dart:convert'; import 'dart:io'; +import 'dart:math'; +import 'package:dpip/core/network/sse_client.dart'; import 'package:dpip/core/network/sse_event.dart'; import 'package:dpip/core/realtime/elapsed.dart'; import 'package:dpip/core/realtime/sse_realtime_source.dart'; @@ -14,6 +16,18 @@ class _FakeElapsed implements Elapsed { Duration get elapsed => value; } +/// A [Random] whose `nextDouble` the test sets — 0 is the shortest spread, +/// just under 1 the longest. +class _FixedRandom implements Random { + double next = 0.5; + @override + double nextDouble() => next; + @override + int nextInt(int max) => (next * max).floor(); + @override + bool nextBool() => next >= 0.5; +} + /// A concrete [SseRealtimeSource] over `String` payloads (decode = identity), /// throwing on the sentinel `'bad'` so decode-failure resilience is testable. class _TestSseSource extends SseRealtimeSource { @@ -22,6 +36,7 @@ class _TestSseSource extends SseRealtimeSource { SseLiveness? liveness, super.elapsed, super.delay, + super.random, }) : super( liveness: liveness ?? const SseLiveness.connectionOpen(), label: 'test', @@ -50,6 +65,7 @@ class _BytesSseSource extends _TestSseSource { super.liveness, super.elapsed, super.delay, + super.random, }); @override @@ -64,18 +80,24 @@ class _Harness { liveness: liveness, elapsed: elapsed, delay: _delay, + random: random, ) : _TestSseSource( connect: _connect, liveness: liveness, elapsed: elapsed, delay: _delay, + random: random, ); } final connects = >[]; final delays = >[]; + + /// The duration each [delays] entry was asked to wait. + final waits = []; final elapsed = _FakeElapsed(); + final random = _FixedRandom(); late final _TestSseSource source; Stream _connect() { @@ -84,12 +106,22 @@ class _Harness { return controller.stream; } - Future _delay(Duration _) { + Future _delay(Duration wait) { final completer = Completer(); delays.add(completer); + waits.add(wait); return completer.future; } + /// Drops the current connection and lets the queued reconnect fire, + /// leaving the new connection open but silent. + Future dropAndReconnect() async { + await current.close(); + await pumpEventQueue(); + delays.last.complete(); + await pumpEventQueue(); + } + StreamController get current => connects.last; } @@ -115,16 +147,86 @@ void main() { }, ); - test('stays Ok while the connection is open but silent', () async { + test('stays Ok through silence up to the limit, and no further', () async { final h = _Harness(); await h.source.fetch(); h.current.add(const SseEvent(data: 'a')); await pumpEventQueue(); - // No further events (an EEW-quiet period) — still live because open. - expect((await h.source.fetch()).valueOrNull, 'a'); + // No further events (an EEW-quiet period) — still live while the + // silence is one a healthy stream can produce between two pings. + h.elapsed.value = sseSilenceLimit; expect((await h.source.fetch()).valueOrNull, 'a'); expect(h.connects, hasLength(1), reason: 'no reconnect churn while open'); + expect(h.delays, isEmpty); + + // One tick past it, the socket is presumed half-open: a handover or a + // NAT timeout that never sent a FIN. Not live, and replaced. + final old = h.current; + h.elapsed.value = sseSilenceLimit + const Duration(seconds: 1); + expect((await h.source.fetch()).isOk, isFalse, reason: 'dead, not calm'); + expect(old.hasListener, isFalse, reason: 'the dead socket is released'); + expect(h.delays, hasLength(1), reason: 'a reconnect is queued'); + expect( + (await h.source.fetch()).isOk, + isFalse, + reason: 'still not live until the new connection speaks', + ); + + h.delays.single.complete(); + await pumpEventQueue(); + expect(h.connects, hasLength(2)); + h.current.add(const SseEvent(data: 'b')); + await pumpEventQueue(); + expect((await h.source.fetch()).valueOrNull, 'b'); + }); + + test('a server ping keeps a quiet connection alive', () async { + final h = _Harness(); + await h.source.fetch(); + h.current.add(const SseEvent(data: 'a')); + await pumpEventQueue(); + + // Pings land every interval; the data never changes. + for (var i = 1; i <= 5; i++) { + h.elapsed.value = sseServerPingInterval * i; + h.current.add(const SseEvent.comment('ping')); + await pumpEventQueue(); + expect((await h.source.fetch()).valueOrNull, 'a'); + } + expect(h.connects, hasLength(1)); + expect(h.delays, isEmpty); + }); + + test('a ping is not a payload', () async { + final h = _Harness(); + await h.source.fetch(); + h.current.add(const SseEvent.comment('ping')); + await pumpEventQueue(); + expect( + (await h.source.fetch()).isOk, + isFalse, + reason: 'connected, but nothing to report yet', + ); + }); + + test('a connection that opens and never speaks is replaced', () async { + final h = _Harness(); + await h.source.fetch(); + expect(h.connects, hasLength(1)); + + h.elapsed.value = sseSilenceLimit + const Duration(seconds: 1); + await h.source.fetch(); + expect(h.connects.single.hasListener, isFalse); + expect(h.delays, hasLength(1)); + }); + + test('the silence limit is two ping intervals and then some', () { + // Two, because the LB hub can push a ping out to almost two intervals + // after the last byte; any tighter and a healthy quiet stream is torn + // down. Pinned so a change to either number is a deliberate one. + expect(sseSilenceLimit, sseServerPingInterval * 2.2); + expect(const SseLiveness.connectionOpen().silenceLimit, sseSilenceLimit); }); test('a compressed frame reaches decodeBytes, never decode', () async { @@ -199,6 +301,61 @@ void main() { expect((await h.source.fetch()).valueOrNull, 'b'); }); + test( + 'backoff steps 1s, 2s, then the server hint, each spread ±50%', + () async { + final h = _Harness(); + await h.source.fetch(); + + h.random.next = 0.5; // the middle of the spread: the plain schedule + for (var i = 0; i < 4; i++) { + await h.dropAndReconnect(); + } + expect(h.waits, const [ + Duration(seconds: 1), + Duration(seconds: 2), + Duration(seconds: 3), // the default `retry:` hint caps it + Duration(seconds: 3), + ]); + + h.random.next = 0; + await h.dropAndReconnect(); + expect( + h.waits.last, + const Duration(milliseconds: 1500), + reason: '-50%', + ); + + h.random.next = 0.999; + await h.dropAndReconnect(); + expect( + h.waits.last, + const Duration(milliseconds: 4497), + reason: 'just under +50%', + ); + }, + ); + + test( + 'a delivering connection resets the backoff to the first step', + () async { + final h = _Harness(); + await h.source.fetch(); + h.random.next = 0; + await h.dropAndReconnect(); + await h.dropAndReconnect(); + expect(h.waits, const [ + Duration(milliseconds: 500), + Duration(seconds: 1), + ]); + + h.current.add(const SseEvent(data: 'a')); // the new connection speaks + await pumpEventQueue(); + await h.dropAndReconnect(); + expect(h.waits.last, const Duration(milliseconds: 500)); + }, + ); + test('a decode failure keeps the last good payload', () async { final h = _Harness(); await h.source.fetch(); @@ -249,6 +406,30 @@ void main() { await pumpEventQueue(); expect((await h.source.fetch()).valueOrNull, 'b', reason: 'refreshed'); }); + + test( + 'a frozen connection is also replaced once past the silence limit', + () async { + final h = _Harness( + liveness: const SseLiveness.eventRecency( + Duration(seconds: 2), + silenceLimit: Duration(seconds: 30), + ), + ); + await h.source.fetch(); + h.current.add(const SseEvent(data: 'a')); + await pumpEventQueue(); + + h.elapsed.value = const Duration(seconds: 30); + await h.source.fetch(); + expect(h.delays, isEmpty, reason: 'stale, but not yet presumed dead'); + + h.elapsed.value = const Duration(seconds: 31); + await h.source.fetch(); + expect(h.connects.single.hasListener, isFalse); + expect(h.delays, hasLength(1)); + }, + ); }); group('SseRealtimeSource — background pause', () { @@ -391,6 +572,105 @@ void main() { }); }); + group('SseRealtimeSource — network change', () { + test( + 'drops the connection at once and reopens after a 1–2 s settle', + () async { + final h = _Harness(); + await h.source.fetch(); + h.current.add(const SseEvent(data: 'a')); + await pumpEventQueue(); + final old = h.current; + + h.random.next = 0.25; + h.source.networkChanged(); + expect(old.hasListener, isFalse, reason: 'bound to the old network'); + expect( + (await h.source.fetch()).isOk, + isFalse, + reason: + 'nothing vouches for the feed until the new connection speaks', + ); + expect(h.waits.single, const Duration(milliseconds: 1250)); + expect(h.connects, hasLength(1), reason: 'not before the settle'); + + h.delays.single.complete(); + await pumpEventQueue(); + expect(h.connects, hasLength(2)); + h.current.add(const SseEvent(data: 'b')); + await pumpEventQueue(); + expect((await h.source.fetch()).valueOrNull, 'b'); + }, + ); + + test('the settle spans 1 to 2 seconds', () async { + final h = _Harness(); + await h.source.fetch(); + h.random.next = 0; + h.source.networkChanged(); + h.random.next = 0.9999; + h.source.networkChanged(); + expect(h.waits, const [ + Duration(seconds: 1), + Duration(milliseconds: 1999), + ]); + }); + + test('resets the backoff and orphans the one counting down', () async { + final h = _Harness(); + await h.source.fetch(); + await h.dropAndReconnect(); + await h.dropAndReconnect(); + await h.current.close(); // a third drop: its backoff is now 3s + await pumpEventQueue(); + expect(h.connects, hasLength(3)); + + h.source.networkChanged(); + h.delays[h.delays.length - 2].complete(); // the old backoff fires late + await pumpEventQueue(); + expect( + h.connects, + hasLength(3), + reason: 'it belonged to the old network', + ); + + h.delays.last.complete(); // the settle + await pumpEventQueue(); + expect(h.connects, hasLength(4)); + + // The new connection fails before speaking: back to the first step. + h.random.next = 0.5; + await h.current.close(); + await pumpEventQueue(); + expect(h.waits.last, const Duration(seconds: 1)); + }); + + test('changes in quick succession open one connection', () async { + final h = _Harness(); + await h.source.fetch(); + h.source + ..networkChanged() // Wi-Fi lost + ..networkChanged(); // cellular up + for (final delay in h.delays) { + delay.complete(); + } + await pumpEventQueue(); + expect(h.connects, hasLength(2)); + expect(h.connects.where((c) => c.hasListener), hasLength(1)); + }); + + test('is ignored before the first fetch and while paused', () async { + final h = _Harness(); + h.source.networkChanged(); + expect(h.delays, isEmpty); + + await h.source.fetch(); + h.source.pause(); + h.source.networkChanged(); + expect(h.delays, isEmpty, reason: 'resuming opens a fresh one anyway'); + }); + }); + group('SseRealtimeSource — named payload event', () { test( 'a topic-named frame is read when that is the payload event', diff --git a/test/core/realtime/staleness_test.dart b/test/core/realtime/staleness_test.dart index 710f07ee9..7f0daddb7 100644 --- a/test/core/realtime/staleness_test.dart +++ b/test/core/realtime/staleness_test.dart @@ -9,13 +9,17 @@ void main() { const staleAfter = Duration(seconds: 3); const offlineAfter = Duration(seconds: 10); - RealtimeStatus at({Duration? age, Duration sinceStart = Duration.zero}) => - classifyStatus( - age: age, - sinceStart: sinceStart, - staleAfter: staleAfter, - offlineAfter: offlineAfter, - ); + RealtimeStatus at({ + Duration? age, + Duration sinceStart = Duration.zero, + bool fallback = false, + }) => classifyStatus( + age: age, + sinceStart: sinceStart, + staleAfter: staleAfter, + offlineAfter: offlineAfter, + fallback: fallback, + ); group('classifyStatus — no data yet', () { test('within the connect window → connecting', () { @@ -74,4 +78,43 @@ void main() { }, ); }); + + group('classifyStatus — fallback data (a stream answering from a poll)', () { + test('fresh → stale, never live', () { + expect(at(age: Duration.zero, fallback: true), RealtimeStatus.stale); + expect( + at(age: const Duration(seconds: 3), fallback: true), + RealtimeStatus.stale, + ); + }); + + test('ages on the same schedule past the stale window', () { + expect( + at(age: const Duration(seconds: 10), fallback: true), + RealtimeStatus.stale, + ); + expect( + at(age: const Duration(seconds: 10, milliseconds: 1), fallback: true), + RealtimeStatus.offline, + ); + }); + + test('negative age → stale, as without it', () { + expect( + at(age: const Duration(seconds: -5), fallback: true), + RealtimeStatus.stale, + ); + }); + + test('no data yet → the same connect/offline split', () { + expect( + at(sinceStart: const Duration(seconds: 5), fallback: true), + RealtimeStatus.connecting, + ); + expect( + at(sinceStart: const Duration(seconds: 11), fallback: true), + RealtimeStatus.offline, + ); + }); + }); } diff --git a/test/features/earthquake/data/earthquake_api_test.dart b/test/features/earthquake/data/earthquake_api_test.dart index 37d56cb33..96331643e 100644 --- a/test/features/earthquake/data/earthquake_api_test.dart +++ b/test/features/earthquake/data/earthquake_api_test.dart @@ -15,12 +15,14 @@ import 'package:dio/dio.dart'; import 'package:dpip/core/network/api_client.dart'; import 'package:dpip/core/network/api_paths.dart'; import 'package:dpip/core/network/api_region.dart'; +import 'package:dpip/core/network/sse_client.dart'; import 'package:dpip/features/earthquake/data/earthquake_api.dart'; import 'package:flutter_test/flutter_test.dart'; class _RecordingApiClient implements ApiClient { final List<(ApiTier, String, Map?)> getCalls = []; final List<(ApiTier, String, Map?)> streamCalls = []; + final List idleTimeouts = []; dynamic nextGetResult; Object? getError; @@ -44,8 +46,10 @@ class _RecordingApiClient implements ApiClient { String path, { Map? query, Map? headers, + Duration idleTimeout = Duration.zero, }) async { streamCalls.add((tier, path, query)); + idleTimeouts.add(idleTimeout); return StreamedResponse( (nextStream ?? const Stream>.empty()).cast(), () {}, @@ -257,5 +261,18 @@ void main() { expect(events.single.data, '[]'); }, ); + + test( + 'both streams bound their silence by the server ping, not forever', + () async { + // A zero idle timeout is how a half-open socket used to wait out a + // handover indefinitely. + client.nextStream = _sseBody('[]'); + await api.openEewSse().toList(); + client.nextStream = _sseBody('x'); + await api.openTremSse(live: false).toList(); + expect(client.idleTimeouts, [sseSilenceLimit, sseSilenceLimit]); + }, + ); }); } diff --git a/test/features/earthquake/eew_realtime_source_test.dart b/test/features/earthquake/eew_realtime_source_test.dart index 8c30a7181..d4e06f218 100644 --- a/test/features/earthquake/eew_realtime_source_test.dart +++ b/test/features/earthquake/eew_realtime_source_test.dart @@ -1,6 +1,9 @@ +import 'dart:async'; import 'dart:convert'; +import 'package:dio/dio.dart'; import 'package:dpip/core/network/sse_event.dart'; +import 'package:dpip/core/realtime/elapsed.dart'; import 'package:dpip/features/earthquake/data/eew_realtime_source.dart'; import 'package:dpip/features/earthquake/domain/eew.dart'; import 'package:flutter_test/flutter_test.dart'; @@ -169,4 +172,238 @@ void main() { expect(source.sameData([_eew(serial: 1)], [_eew(serial: 2)]), isFalse); expect(source.sameData(const [], [_eew()]), isFalse); }); + + group('REST fallback while the stream is down', () { + late _Fallback f; + + setUp(() => f = _Fallback()); + tearDown(() => f.source.dispose()); + + test( + 'no poll while the stream answers, or in its first seconds down', + () async { + await f.source.fetch(); // opens the stream + f.stream.add(const SseEvent(data: '[]')); + await pumpEventQueue(); + f.at(const Duration(seconds: 30)); + expect((await f.source.fetch()).isOk, isTrue); + expect(f.polls, isEmpty, reason: 'a live stream needs no fallback'); + + await f.stream.close(); // the stream drops + await pumpEventQueue(); + f.at(const Duration(seconds: 31)); + expect((await f.source.fetch()).isOk, isFalse); + f.at(const Duration(seconds: 33)); + await f.source.fetch(); + expect(f.polls, isEmpty, reason: 'an ordinary reconnect costs nothing'); + + f.at(const Duration(seconds: 34)); + await f.source.fetch(); + expect(f.polls, hasLength(1), reason: 'past the grace: poll'); + }, + ); + + test('a polled alert arrives, filtered like a streamed one, flagged as ' + 'fallback', () async { + await f.downFor(EewRealtimeSource.fallbackGrace); + expect(f.polls, hasLength(1)); + f.polls.single.complete([_wire('cwa', 'a'), _wire('jma', 'b')]); + await pumpEventQueue(); + + final result = await f.source.fetch(); + expect(result.valueOrNull?.map((e) => e.id), [ + 'a', + ], reason: 'JMA dropped'); + expect(f.source.servingFallback, isTrue); + + f.cwaOnly = true; + f.at(f.now + EewRealtimeSource.fallbackInterval); + await f.source.fetch(); + f.polls.last.complete([_wire('cwa', 'a'), _wire('scdzj', 'c')]); + await pumpEventQueue(); + expect((await f.source.fetch()).valueOrNull?.map((e) => e.id), [ + 'a', + ], reason: 'cwaOnly applies to a polled list too'); + }); + + test('polls once per interval, never two at a time', () async { + await f.downFor(EewRealtimeSource.fallbackGrace); + expect(f.polls, hasLength(1)); + + // The first poll hangs past the interval: no second one stacks on it. + f.at(f.now + const Duration(seconds: 12)); + await f.source.fetch(); + expect(f.polls, hasLength(1)); + + f.polls.single.complete(const []); + await pumpEventQueue(); + await f.source.fetch(); // 12 s since the last send: due + expect(f.polls, hasLength(2)); + f.polls.last.complete(const []); + await pumpEventQueue(); + + f.at(f.now + const Duration(seconds: 9)); + await f.source.fetch(); + expect(f.polls, hasLength(2), reason: 'not yet an interval since'); + f.at(f.now + const Duration(seconds: 1)); + await f.source.fetch(); + expect(f.polls, hasLength(3)); + }); + + test( + 'a snapshot is answered with for the hold, then the source fails', + () async { + await f.downFor(EewRealtimeSource.fallbackGrace); + f.polls.single.complete([_wire('cwa', 'a')]); + await pumpEventQueue(); + final landed = f.now; + + // Every later poll fails: the snapshot is served until the hold runs + // out, and then the channel is told the truth. + f.at(landed + EewRealtimeSource.fallbackHold); + expect((await f.source.fetch()).isOk, isTrue); + f.polls.last.completeError( + DioException( + requestOptions: RequestOptions(), + type: DioExceptionType.connectionError, + ), + ); + await pumpEventQueue(); + f.at( + landed + EewRealtimeSource.fallbackHold + const Duration(seconds: 1), + ); + expect((await f.source.fetch()).isOk, isFalse); + expect(f.source.servingFallback, isFalse); + }, + ); + + test( + 'the stream coming back stops polling and discards a poll in flight', + () async { + await f.downFor(EewRealtimeSource.fallbackGrace); + final inFlight = f.polls.single; + + f.reconnect(); + await pumpEventQueue(); + f.stream.add(const SseEvent(data: '[]')); + await pumpEventQueue(); + final streamed = await f.source.fetch(); + expect(streamed.valueOrNull, isEmpty); + expect(f.source.servingFallback, isFalse); + + inFlight.complete([_wire('cwa', 'late')]); // lands after the stream + await pumpEventQueue(); + f.at(f.now + const Duration(minutes: 1)); + expect((await f.source.fetch()).valueOrNull, isEmpty); + expect( + f.polls, + hasLength(1), + reason: 'no polling while the stream is up', + ); + }, + ); + + test('pausing for the background ends the fallback', () async { + await f.downFor(EewRealtimeSource.fallbackGrace); + f.polls.single.complete([_wire('cwa', 'a')]); + await pumpEventQueue(); + expect((await f.source.fetch()).isOk, isTrue); + + f.source + ..pause() + ..resume(); + expect( + (await f.source.fetch()).isOk, + isFalse, + reason: 'a snapshot from before the background is not an answer', + ); + expect(f.source.servingFallback, isFalse); + }); + + test('without a poll the source is the stream alone', () async { + final source = EewRealtimeSource( + () => const Stream.empty(), + cwaOnly: () => false, + elapsed: f.elapsed, + delay: (_) => Completer().future, + ); + addTearDown(source.dispose); + f.at(const Duration(minutes: 5)); + expect((await source.fetch()).isOk, isFalse); + expect(source.servingFallback, isFalse); + }); + }); +} + +class _FakeElapsed implements Elapsed { + Duration value = Duration.zero; + @override + Duration get elapsed => value; +} + +Map _wire(String author, String id) => { + 'author': author, + 'id': id, + 'serial': 1, + 'status': 0, + 'final': 0, + 'eq': { + 'time': 1700000000000, + 'lon': 121.5, + 'lat': 24.0, + 'depth': 10.0, + 'mag': 5.0, + 'loc': 'test', + 'max': 4, + }, +}; + +/// An EEW source whose stream, clock, reconnect delay and fallback polls are +/// all driven by hand. +class _Fallback { + _Fallback() { + source = EewRealtimeSource( + () { + final controller = StreamController(); + streams.add(controller); + return controller.stream; + }, + cwaOnly: () => cwaOnly, + poll: () { + final completer = Completer>(); + polls.add(completer); + return completer.future; + }, + elapsed: elapsed, + delay: (_) { + final completer = Completer(); + reconnects.add(completer); + return completer.future; + }, + ); + } + + late final EewRealtimeSource source; + final elapsed = _FakeElapsed(); + final streams = >[]; + final polls = >>[]; + final reconnects = >[]; + bool cwaOnly = false; + + StreamController get stream => streams.last; + Duration get now => elapsed.value; + void at(Duration t) => elapsed.value = t; + + /// Opens the stream, drops it, and fetches across [down] of outage. + Future downFor(Duration down) async { + await source.fetch(); + await stream.close(); + await pumpEventQueue(); + await source.fetch(); // the outage starts here + at(now + down); + await source.fetch(); + } + + /// Lets the queued reconnect fire. + void reconnect() => reconnects.last.complete(); } diff --git a/test/features/earthquake/replay_session_test.dart b/test/features/earthquake/replay_session_test.dart index b6549f4e5..d74063ce9 100644 --- a/test/features/earthquake/replay_session_test.dart +++ b/test/features/earthquake/replay_session_test.dart @@ -101,6 +101,7 @@ class _RecordingApiClient implements ApiClient { String path, { Map? query, Map? headers, + Duration idleTimeout = Duration.zero, }) => throw UnimplementedError(); @override diff --git a/test/features/earthquake/rts_realtime_source_test.dart b/test/features/earthquake/rts_realtime_source_test.dart index e0e6932a9..b71e8a2c2 100644 --- a/test/features/earthquake/rts_realtime_source_test.dart +++ b/test/features/earthquake/rts_realtime_source_test.dart @@ -17,6 +17,7 @@ import 'dart:convert'; import 'dart:io'; import 'dart:typed_data'; +import 'package:dpip/core/network/sse_client.dart'; import 'package:dpip/core/network/sse_event.dart'; import 'package:dpip/core/realtime/elapsed.dart'; import 'package:dpip/features/earthquake/data/rts_realtime_source.dart'; @@ -109,6 +110,41 @@ void main() { }); }); + test('asleep, silence past the ping bound is a dead feed, not calm', () { + fakeAsync((async) { + final connections = _Connections(); + final elapsed = _FakeElapsed(); + final source = RtsRealtimeSource( + connections.connect, + demand: RtsLiveDemand(), + elapsed: elapsed, + delay: (_) => Completer().future, + ); + + source.fetch(); + connections.last.add(_info); + async.flushMicrotasks(); + + // Pings keep a sleeping stream calm however long nothing alerts. + elapsed.value = sseServerPingInterval; + connections.last.add(const SseEvent.comment('ping')); + async.flushMicrotasks(); + elapsed.value = sseServerPingInterval + sseSilenceLimit; + source.fetch().then((result) { + expect(result.valueOrNull?.stations, isEmpty, reason: 'calm'); + }); + async.flushMicrotasks(); + + // A half-open socket sends no pings: past the bound it cannot vouch for + // calm, and it is dropped for a fresh one. + elapsed.value += const Duration(seconds: 1); + source.fetch().then((result) => expect(result.isOk, isFalse)); + async.flushMicrotasks(); + expect(connections.last.hasListener, isFalse); + source.dispose(); + }); + }); + test('live, a connection that has gone quiet is a failure, not calm', () { fakeAsync((async) { final connections = _Connections();