From 823c7855306d68c42173fd273251e2b889d65979 Mon Sep 17 00:00:00 2001 From: YuYu1015 Date: Fri, 16 Oct 2026 06:31:22 +0800 Subject: [PATCH 1/4] fix(eew): drop SSE connections that miss the server's pings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Platform: all Fix(zh-Hant): 修正網路切換後地震速報連線已中斷卻仍顯示連線正常，現在會自動發現並重新連線 Fix(en-US): the earthquake alert feed no longer shows as connected after a network switch silently cut it off; it now notices and reconnects --- ARCHITECTURE.md | 7 +- lib/core/network/api_client.dart | 29 +++-- lib/core/network/sse_client.dart | 46 +++++++- lib/core/network/sse_event.dart | 30 +++++- lib/core/realtime/sse_realtime_source.dart | 82 ++++++++++++-- .../earthquake/data/eew_realtime_source.dart | 12 ++- .../earthquake/data/rts_realtime_source.dart | 7 +- test/core/network/api_client_test.dart | 31 ++++++ test/core/network/sse_client_test.dart | 23 +++- test/core/network/sse_event_test.dart | 9 ++ .../realtime/sse_realtime_source_test.dart | 101 +++++++++++++++++- .../earthquake/data/earthquake_api_test.dart | 17 +++ .../earthquake/replay_session_test.dart | 1 + .../earthquake/rts_realtime_source_test.dart | 36 +++++++ 14 files changed, 389 insertions(+), 42 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index cff0a1281..9f7244429 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -171,7 +171,12 @@ 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. `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/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/realtime/sse_realtime_source.dart b/lib/core/realtime/sse_realtime_source.dart index 379ab3cab..c82530250 100644 --- a/lib/core/realtime/sse_realtime_source.dart +++ b/lib/core/realtime/sse_realtime_source.dart @@ -6,6 +6,7 @@ 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 +14,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 +34,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 +71,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 @@ -122,6 +144,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 +187,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 +199,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 +210,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 +257,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 @@ -233,6 +291,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 +357,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 diff --git a/lib/features/earthquake/data/eew_realtime_source.dart b/lib/features/earthquake/data/eew_realtime_source.dart index 819026b96..b2d32557b 100644 --- a/lib/features/earthquake/data/eew_realtime_source.dart +++ b/lib/features/earthquake/data/eew_realtime_source.dart @@ -13,11 +13,13 @@ 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. class EewRealtimeSource extends SseRealtimeSource> { /// [connect] opens one EEW SSE connection — in production /// `EarthquakeApi.openEewSse`; the source calls it again to reconnect. 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/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/realtime/sse_realtime_source_test.dart b/test/core/realtime/sse_realtime_source_test.dart index 6cb5ff571..ed102b931 100644 --- a/test/core/realtime/sse_realtime_source_test.dart +++ b/test/core/realtime/sse_realtime_source_test.dart @@ -3,6 +3,7 @@ import 'dart:typed_data'; import 'dart:convert'; import 'dart:io'; +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'; @@ -115,16 +116,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 { @@ -249,6 +320,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', () { 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/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(); From 0cb0fc2effcdb97b001965eb0b2b7a8f555b988b Mon Sep 17 00:00:00 2001 From: YuYu1015 Date: Fri, 16 Oct 2026 06:34:08 +0800 Subject: [PATCH 2/4] perf(eew): spread SSE reconnects with random jitter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Platform: all Optimization(zh-Hant): 伺服器中斷後各裝置錯開時間重新連線，恢復時不再一起湧入，速報連線更快恢復 Optimization(en-US): devices now reconnect at staggered times after a server outage instead of all at once, so the alert feed recovers faster --- ARCHITECTURE.md | 3 +- lib/core/realtime/sse_realtime_source.dart | 26 ++++-- .../realtime/sse_realtime_source_test.dart | 88 ++++++++++++++++++- 3 files changed, 109 insertions(+), 8 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 9f7244429..7bb96d42d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -176,7 +176,8 @@ event-recency. Neither trusts an open socket on its own: the servers write a 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. `RealtimeService` + `AppLifecycleListener` +calm. Reconnect backoff is jittered ±50 % so a region's audience does not +reconnect in lockstep. `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/lib/core/realtime/sse_realtime_source.dart b/lib/core/realtime/sse_realtime_source.dart index c82530250..d3a7374bf 100644 --- a/lib/core/realtime/sse_realtime_source.dart +++ b/lib/core/realtime/sse_realtime_source.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'dart:convert'; import 'dart:io'; +import 'dart:math'; import 'dart:typed_data'; import 'package:dpip/core/error/failure.dart'; @@ -88,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); @@ -100,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 @@ -453,12 +459,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/test/core/realtime/sse_realtime_source_test.dart b/test/core/realtime/sse_realtime_source_test.dart index ed102b931..189149ada 100644 --- a/test/core/realtime/sse_realtime_source_test.dart +++ b/test/core/realtime/sse_realtime_source_test.dart @@ -2,6 +2,7 @@ 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'; @@ -15,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 { @@ -23,6 +36,7 @@ class _TestSseSource extends SseRealtimeSource { SseLiveness? liveness, super.elapsed, super.delay, + super.random, }) : super( liveness: liveness ?? const SseLiveness.connectionOpen(), label: 'test', @@ -51,6 +65,7 @@ class _BytesSseSource extends _TestSseSource { super.liveness, super.elapsed, super.delay, + super.random, }); @override @@ -65,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() { @@ -85,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; } @@ -270,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(); From dca904599213f3641784d8fe04a9cc0d7891f5fd Mon Sep 17 00:00:00 2001 From: YuYu1015 Date: Fri, 16 Oct 2026 06:51:10 +0800 Subject: [PATCH 3/4] fix(eew): reconnect live feeds as soon as the network changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Platform: all Fix(zh-Hant): 切換 Wi-Fi 與行動網路或重新連上網路後，地震速報與即時測站連線會立即重新建立，不必等到逾時 Fix(en-US): switching between Wi-Fi and mobile data, or coming back online, now reconnects the earthquake alert and station feeds at once instead of waiting for a timeout --- ARCHITECTURE.md | 9 +- android/app/src/main/AndroidManifest.xml | 4 + .../kotlin/com/exptech/dpip/MainActivity.kt | 3 + .../kotlin/com/exptech/dpip/NetworkChannel.kt | 106 ++++++++++++++++++ ios/Runner.xcodeproj/project.pbxproj | 4 + ios/Runner/AppDelegate.swift | 1 + ios/Runner/NetworkPlugin.swift | 69 ++++++++++++ lib/bootstrap.dart | 9 +- lib/core/platform/network_changes.dart | 70 ++++++++++++ lib/core/realtime/realtime_channel.dart | 11 ++ lib/core/realtime/realtime_service.dart | 24 ++++ lib/core/realtime/realtime_source.dart | 15 +++ lib/core/realtime/sse_realtime_source.dart | 41 +++++++ test/core/platform/network_changes_test.dart | 85 ++++++++++++++ test/core/realtime/realtime_channel_test.dart | 15 +++ test/core/realtime/realtime_service_test.dart | 51 +++++++++ .../realtime/sse_realtime_source_test.dart | 99 ++++++++++++++++ 17 files changed, 613 insertions(+), 3 deletions(-) create mode 100644 android/app/src/main/kotlin/com/exptech/dpip/NetworkChannel.kt create mode 100644 ios/Runner/NetworkPlugin.swift create mode 100644 lib/core/platform/network_changes.dart create mode 100644 test/core/platform/network_changes_test.dart diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 7bb96d42d..f4e9c530c 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -176,8 +176,13 @@ event-recency. Neither trusts an open socket on its own: the servers write a 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. Reconnect backoff is jittered ±50 % so a region's audience does not -reconnect in lockstep. `RealtimeService` + `AppLifecycleListener` +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. `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/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..0d82d925a 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(); } @@ -160,6 +165,12 @@ class RealtimeChannel implements RealtimeChannelBase { await _fetch(); } + @override + void networkChanged() { + if (_disposed) return; + _source.networkChanged(); + } + @override void dispose() { _disposed = true; 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..69ef7b23b 100644 --- a/lib/core/realtime/realtime_source.dart +++ b/lib/core/realtime/realtime_source.dart @@ -57,6 +57,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 d3a7374bf..052b5c1cd 100644 --- a/lib/core/realtime/sse_realtime_source.dart +++ b/lib/core/realtime/sse_realtime_source.dart @@ -278,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; 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..42cbc64f8 100644 --- a/test/core/realtime/realtime_channel_test.dart +++ b/test/core/realtime/realtime_channel_test.dart @@ -70,6 +70,11 @@ class _FakeSource extends RealtimeSource { @override bool isIgnorableFailure(Failure failure) => ignorable?.call(failure) ?? false; + + int networkChanges = 0; + + @override + void networkChanged() => networkChanges++; } /// Flushes pending microtasks so broadcast emissions and unawaited fetches land. @@ -353,4 +358,14 @@ 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); + }); } 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 189149ada..d48a4733b 100644 --- a/test/core/realtime/sse_realtime_source_test.dart +++ b/test/core/realtime/sse_realtime_source_test.dart @@ -572,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', From b084e0d220b1aa9e12ed3ba41f6911d188d74606 Mon Sep 17 00:00:00 2001 From: YuYu1015 Date: Fri, 16 Oct 2026 07:00:24 +0800 Subject: [PATCH 4/4] fix(eew): poll the alert list while the live stream is down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Platform: all Fix(zh-Hant): 速報即時連線中斷時改為每 10 秒查詢一次，期間發布的速報仍會出現在地震速報頁，並標示資料可能已過期 Fix(en-US): while the live alert connection is down, the app checks for alerts every 10 seconds, so one issued meanwhile still appears on the early warning page, marked as possibly out of date --- ARCHITECTURE.md | 9 +- lib/core/realtime/realtime_channel.dart | 19 +- lib/core/realtime/realtime_source.dart | 10 + lib/core/realtime/staleness.dart | 13 +- .../earthquake/data/eew_realtime_source.dart | 199 ++++++++++++++- .../earthquake/earthquake_providers.dart | 10 +- test/core/realtime/realtime_channel_test.dart | 92 +++++++ test/core/realtime/staleness_test.dart | 57 ++++- .../earthquake/eew_realtime_source_test.dart | 237 ++++++++++++++++++ 9 files changed, 623 insertions(+), 23 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f4e9c530c..fe0e8ec41 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -181,8 +181,13 @@ 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. `RealtimeService` + -`AppLifecycleListener` +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/lib/core/realtime/realtime_channel.dart b/lib/core/realtime/realtime_channel.dart index 0d82d925a..ea89b9df3 100644 --- a/lib/core/realtime/realtime_channel.dart +++ b/lib/core/realtime/realtime_channel.dart @@ -68,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, @@ -108,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; @@ -198,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. @@ -232,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 || @@ -243,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_source.dart b/lib/core/realtime/realtime_source.dart index 69ef7b23b..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 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 b2d32557b..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; @@ -20,25 +27,201 @@ import 'package:flutter/foundation.dart' show listEquals; /// 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/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/realtime/realtime_channel_test.dart b/test/core/realtime/realtime_channel_test.dart index 42cbc64f8..9c106de7d 100644 --- a/test/core/realtime/realtime_channel_test.dart +++ b/test/core/realtime/realtime_channel_test.dart @@ -73,6 +73,12 @@ class _FakeSource extends RealtimeSource { int networkChanges = 0; + /// What [servingFallback] reports for the next fetch's answer. + bool fallback = false; + + @override + bool get servingFallback => fallback; + @override void networkChanged() => networkChanges++; } @@ -368,4 +374,90 @@ void main() { 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/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/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(); }