fix(eew): detect and recover from silently dropped live feeds

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.

android/app/src/main/AndroidManifest.xml

@@ -1,5 +1,9 @@
 <manifest xmlns:android="http://schemas.android.com/apk/res/android">
     <uses-permission android:name="android.permission.INTERNET"/>
+    <!-- Network changes (NetworkChannel): a normal permission, granted at
+         install, that the default-network callback requires. Declared here
+         rather than left to whichever plugin's manifest happens to merge it. -->
+    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE"/>
     <uses-permission android:name="android.permission.POST_NOTIFICATIONS"/>
     <!-- Notifications (awesome_notifications): vibrate, wake the screen for
          critical alerts, and reschedule after a reboot. -->

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))
 

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
+ * `<transport>:<network handle>`. 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<Long, String>()
+
+    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"
+    }
+}

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 = "<group>"; };
 		A382CD9DEA741E45DBF741D7 /* Sounds/rain.aiff */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = audio.aiff; path = Sounds/rain.aiff; sourceTree = "<group>"; };
 		AA0000000000000000000C01 /* CompassPlugin.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CompassPlugin.swift; sourceTree = "<group>"; };
+		AA0000000000000000000H01 /* NetworkPlugin.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NetworkPlugin.swift; sourceTree = "<group>"; };
 		AA0000000000000000000D01 /* DeviceInfoPlugin.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = DeviceInfoPlugin.swift; sourceTree = "<group>"; };
 		NL0G00000000000000000001 /* NativeLogPlugin.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NativeLogPlugin.swift; sourceTree = "<group>"; };
 		NL0G00000000000000000003 /* NativeLogTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = NativeLogTests.swift; sourceTree = "<group>"; };
@@ -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 */,

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")!)

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 `<type>:<interface name>` 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)"
+  }
+}

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<void> 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

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<StreamedResponse> openStream(
     ApiTier tier,
     String path, {
     Map<String, dynamic>? query,
     Map<String, dynamic>? 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,
           ),
         );