diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 452b6af4..ae87e5b8 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -128,6 +128,7 @@ The app uses a layered service architecture with clear separation of concerns: - `PingService`: TX/RX/Discovery ping orchestration, coordinates with TxTracker/DiscTracker/RxLogger - `ApiQueueService`: Hive-based persistent upload queue with batch POST and retry logic - `ApiService`: HTTP client for MeshMapper API endpoints +- `NetworkStateService`: Android constrained and satellite network monitoring; routine uploads use 60-second pacing and auth uses a 30-second timeout on constrained links - `DeviceModelService`: Loads `assets/device-models.json` for device identification and power reporting **State Management** (`lib/providers/`): diff --git a/android/app/src/main/AndroidManifest.xml b/android/app/src/main/AndroidManifest.xml index cba07486..8e0d5898 100644 --- a/android/app/src/main/AndroidManifest.xml +++ b/android/app/src/main/AndroidManifest.xml @@ -19,6 +19,8 @@ + + @@ -73,6 +75,15 @@ android:name="flutterEmbedding" android:value="2" /> + + + = + mapOf("constrained" to constrained, "satellite" to satellite) + + /// EventSink.success() must run on the main thread; NetworkCallback + /// methods fire on the HandlerThread registered for them. + private fun emit(state: Map) { + mainHandler.post { eventSink?.success(state) } + } + + private fun stopMonitoring() { + networkCallback?.let { + try { + connectivityManager.unregisterNetworkCallback(it) + } catch (_: IllegalArgumentException) { + } + } + networkCallback = null + handlerThread?.quitSafely() + handlerThread = null + } +} diff --git a/lib/services/api_queue_service.dart b/lib/services/api_queue_service.dart index 6988087d..e0b4b316 100644 --- a/lib/services/api_queue_service.dart +++ b/lib/services/api_queue_service.dart @@ -7,13 +7,14 @@ import '../models/api_queue_item.dart'; import '../utils/debug_logger_io.dart'; import 'api_service.dart'; import 'custom_api_service.dart'; +import 'network_state_service.dart'; /// API queue service with batch upload and retry logic /// Ported from apiQueue and batchUpload() in wardrive.js /// /// Features: /// - Queue pings locally with Hive persistence -/// - Batch upload every 50 entries OR 30 seconds +/// - Upload batches contain up to 50 entries and use network-aware timers /// - RX buffering: group by repeater ID (max 4 per batch) /// - Retry with exponential backoff for failed uploads /// - Offline mode: accumulates pings without uploading @@ -21,13 +22,22 @@ class ApiQueueService { static const String _boxName = 'api_queue'; static const int _batchSize = 50; static const Duration _batchTimeout = Duration(seconds: 15); + // Wider cadence while on a constrained (e.g. satellite) link: fewer, larger + // batches beat frequent small ones when every round trip carries high + // per-request latency. + static const Duration _batchTimeoutConstrained = Duration(seconds: 60); + static const Duration _pingFlushTimeout = Duration(seconds: 5); + static const Duration _pingFlushTimeoutConstrained = Duration(seconds: 60); static const int _maxRetries = 5; static const int _maxRxPerRepeater = 4; final ApiService _apiService; + final NetworkStateSource _networkState; Box? _box; Timer? _batchTimer; Timer? _pingFlushTimer; + StreamSubscription? _networkStateSubscription; + late bool _lastIsConstrained; bool _isUploading = false; bool _isRecovering = false; @@ -112,7 +122,15 @@ class ApiQueueService { return true; } - ApiQueueService({required ApiService apiService}) : _apiService = apiService; + ApiQueueService({ + required ApiService apiService, + NetworkStateSource? networkState, + }) : _apiService = apiService, + _networkState = networkState ?? NetworkStateService.instance { + _lastIsConstrained = _networkState.current.isConstrained; + _networkStateSubscription = + _networkState.stream.listen(_handleNetworkState); + } /// Initialize the queue (must be called before use) Future init() async { @@ -148,6 +166,7 @@ class ApiQueueService { // Start batch timer debugLog('[API QUEUE] Starting batch timer...'); _startBatchTimer(); + debugLog('[API QUEUE] init() complete'); } @@ -328,12 +347,7 @@ class ApiQueueService { '[API QUEUE] TX enqueued: $heardRepeats (queue size: $queueSize)'); } onQueueUpdated?.call(queueSize); - _pingFlushTimer?.cancel(); - _pingFlushTimer = Timer(const Duration(seconds: 5), () { - debugLog('[API QUEUE] Ping flush timer fired'); - _flushRxBuffer(); - _uploadBatch(); - }); + _schedulePingFlush(); } /// Enqueue an RX observation @@ -435,12 +449,7 @@ class ApiQueueService { '[API QUEUE] DISC enqueued: $repeaterId ($nodeType) at $latitude, $longitude (queue size: $queueSize)'); } onQueueUpdated?.call(queueSize); - _pingFlushTimer?.cancel(); - _pingFlushTimer = Timer(const Duration(seconds: 5), () { - debugLog('[API QUEUE] Ping flush timer fired'); - _flushRxBuffer(); - _uploadBatch(); - }); + _schedulePingFlush(); } /// Enqueue a TRACE ping result (targeted zero-hop trace) @@ -491,12 +500,7 @@ class ApiQueueService { '[API QUEUE] TRACE enqueued: $repeaterId at $latitude, $longitude (queue size: $queueSize)'); } onQueueUpdated?.call(queueSize); - _pingFlushTimer?.cancel(); - _pingFlushTimer = Timer(const Duration(seconds: 5), () { - debugLog('[API QUEUE] Ping flush timer fired'); - _flushRxBuffer(); - _uploadBatch(); - }); + _schedulePingFlush(); } /// Enqueue a failed DISC discovery (no nodes responded) @@ -539,19 +543,14 @@ class ApiQueueService { '[API QUEUE] DISC drop enqueued at $latitude, $longitude (queue size: $queueSize)'); } onQueueUpdated?.call(queueSize); - _pingFlushTimer?.cancel(); - _pingFlushTimer = Timer(const Duration(seconds: 5), () { - debugLog('[API QUEUE] Ping flush timer fired'); - _flushRxBuffer(); - _uploadBatch(); - }); + _schedulePingFlush(); } /// Report a square where smart pinging held a ping. [held] is `tx` or /// `disc`. The server verifies the square against its own coverage and /// credits it once per session; a dropped one is silent. Modelled on /// enqueueDiscDrop: offline rows honour the airborne pause, a closed box - /// falls back to memory, and the 5 second flush timer sends it on. + /// falls back to memory, and the network-aware flush timer sends it on. Future enqueueDefer({ required double latitude, required double longitude, @@ -584,12 +583,7 @@ class ApiQueueService { '[API QUEUE] DEFER ($held) enqueued at $latitude, $longitude (queue size: $queueSize)'); } onQueueUpdated?.call(queueSize); - _pingFlushTimer?.cancel(); - _pingFlushTimer = Timer(const Duration(seconds: 5), () { - debugLog('[API QUEUE] Ping flush timer fired'); - _flushRxBuffer(); - _uploadBatch(); - }); + _schedulePingFlush(); } // Guard to prevent concurrent RX buffer flushes @@ -637,10 +631,37 @@ class ApiQueueService { } } + void _handleNetworkState(NetworkState state) { + if (state.isConstrained == _lastIsConstrained) return; + + _lastIsConstrained = state.isConstrained; + debugLog('[API QUEUE] Network pacing changed: ' + '${state.isConstrained ? 'constrained' : 'ordinary'}'); + + if (_batchTimer != null) _startBatchTimer(); + if (_pingFlushTimer?.isActive ?? false) _schedulePingFlush(); + } + + void _schedulePingFlush() { + final timeout = + _lastIsConstrained ? _pingFlushTimeoutConstrained : _pingFlushTimeout; + _pingFlushTimer?.cancel(); + _pingFlushTimer = Timer(timeout, () { + debugLog('[API QUEUE] Ping flush timer fired ' + '(${timeout.inSeconds}s delay' + '${_lastIsConstrained ? ', constrained network' : ''})'); + _flushRxBuffer(); + _uploadBatch(); + }); + } + void _startBatchTimer() { + final constrained = _lastIsConstrained; + final timeout = constrained ? _batchTimeoutConstrained : _batchTimeout; _batchTimer?.cancel(); - _batchTimer = Timer.periodic(_batchTimeout, (_) { - debugLog('[API QUEUE] Batch timer fired (15s interval)'); + _batchTimer = Timer.periodic(timeout, (_) { + debugLog('[API QUEUE] Batch timer fired (${timeout.inSeconds}s interval' + '${constrained ? ', constrained network' : ''})'); _flushRxBuffer(); _uploadBatch(); }); @@ -1007,6 +1028,7 @@ class ApiQueueService { void dispose() { _batchTimer?.cancel(); _pingFlushTimer?.cancel(); + _networkStateSubscription?.cancel(); _box?.close(); } } diff --git a/lib/services/api_service.dart b/lib/services/api_service.dart index efee6bea..bfc8783b 100644 --- a/lib/services/api_service.dart +++ b/lib/services/api_service.dart @@ -9,6 +9,7 @@ import 'package:http/http.dart' as http; import '../models/repeater.dart'; import '../utils/debug_logger_io.dart'; import 'meshcore/regional_carpeater_filter.dart'; +import 'network_state_service.dart'; /// Result of a batch upload attempt /// @@ -116,6 +117,7 @@ class ApiService { static const Duration maxWardriveRetryAfter = Duration(hours: 1); final http.Client _client; + final NetworkStateSource _networkState; bool _heartbeatEnabled = false; // Track if heartbeat mode is active String? _sessionId; bool _txAllowed = false; @@ -201,7 +203,11 @@ class ApiService { /// the refusal code, if any. The provider replaces its cache from this. void Function(List keys, String? error)? onRegionalCarpeaters; - ApiService({http.Client? client}) : _client = client ?? http.Client(); + ApiService({ + http.Client? client, + NetworkStateSource? networkState, + }) : _client = client ?? http.Client(), + _networkState = networkState ?? NetworkStateService.instance; /// Send [request], replaying it once when the first attempt was written onto /// a keep-alive socket the server had already closed. @@ -564,6 +570,11 @@ class ApiService { } } + // Give auth attempts more room on a constrained link so a + // high-latency response can arrive before the request times out. + final authTimeout = _networkState.current.isConstrained + ? const Duration(seconds: 30) + : const Duration(seconds: 10); final response = await _send( 'POST /wardrive-api.php/auth', () => _client @@ -572,7 +583,7 @@ class ApiService { headers: {'Content-Type': 'application/json'}, body: json.encode(payload), ) - .timeout(const Duration(seconds: 10)), + .timeout(authTimeout), ); stopwatch.stop(); diff --git a/lib/services/network_state_service.dart b/lib/services/network_state_service.dart new file mode 100644 index 00000000..f6c8801a --- /dev/null +++ b/lib/services/network_state_service.dart @@ -0,0 +1,81 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:flutter/services.dart'; + +import '../utils/debug_logger_io.dart'; + +/// Snapshot of the device's active default network as reported by Android's +/// constrained-networks API. On non-Android platforms (and on Android below +/// API 36, which predates the API) this is always "not constrained". +class NetworkState { + final bool isConstrained; + final bool isSatellite; + + const NetworkState({required this.isConstrained, required this.isSatellite}); + + static const unconstrained = + NetworkState(isConstrained: false, isSatellite: false); + + @override + String toString() => + 'NetworkState(isConstrained: $isConstrained, isSatellite: $isSatellite)'; +} + +/// Read-only network state used by services that adapt request pacing. +abstract interface class NetworkStateSource { + NetworkState get current; + Stream get stream; +} + +/// Surfaces Android's bandwidth-constrained/satellite network signal to Dart. +/// +/// Routed through an EventChannel on `meshmapper/network_state`, backed by +/// MeshMapperNetworkService.kt (registers a ConnectivityManager network +/// callback and reports whether NET_CAPABILITY_NOT_BANDWIDTH_CONSTRAINED is +/// absent or TRANSPORT_SATELLITE is present): +/// https://developer.android.com/develop/connectivity/satellite/constrained-networks +/// +/// Callers that want to adapt heavy network usage (batch uploads, large +/// downloads) should watch [stream] or check [current]. +class NetworkStateService implements NetworkStateSource { + static const _channel = EventChannel('meshmapper/network_state'); + + final _controller = StreamController.broadcast(); + + NetworkState _current = NetworkState.unconstrained; + + NetworkStateService._() { + if (Platform.isAndroid) _startListening(); + } + static final NetworkStateService instance = NetworkStateService._(); + + /// Most recently reported network state. Defaults to "not constrained" + /// until the first native event arrives, and stays there on non-Android + /// platforms (this API is Android-only). + @override + NetworkState get current => _current; + + /// Broadcast stream of network state changes. Never emits on platforms + /// other than Android. + @override + Stream get stream => _controller.stream; + + void _startListening() { + _channel.receiveBroadcastStream().listen( + (event) { + if (event is! Map) return; + final state = NetworkState( + isConstrained: event['constrained'] as bool? ?? false, + isSatellite: event['satellite'] as bool? ?? false, + ); + _current = state; + _controller.add(state); + if (state.isConstrained) { + debugLog('[NETWORK] Constrained network detected: $state'); + } + }, + onError: (e) => debugWarn('[NETWORK] State stream error: $e'), + ); + } +} diff --git a/test/services/api_queue_network_pacing_test.dart b/test/services/api_queue_network_pacing_test.dart new file mode 100644 index 00000000..b2a3aa1f --- /dev/null +++ b/test/services/api_queue_network_pacing_test.dart @@ -0,0 +1,201 @@ +import 'dart:convert'; + +import 'package:fake_async/fake_async.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:mesh_mapper/services/api_queue_service.dart'; +import 'package:mesh_mapper/services/api_service.dart'; +import 'package:mesh_mapper/services/network_state_service.dart'; + +import 'test_network_state_source.dart'; + +void main() { + const constrained = NetworkState(isConstrained: true, isSatellite: true); + + ({ + ApiQueueService queue, + ApiService api, + TestNetworkStateSource network, + List uploads, + }) build(NetworkState initial) { + final network = TestNetworkStateSource(initial); + final uploads = []; + final api = ApiService( + networkState: network, + client: MockClient((request) async { + if (request.url.path.endsWith('/auth')) { + return http.Response( + json.encode({ + 'success': true, + 'session_id': 'YOW-20260912-0001', + 'tx_allowed': true, + 'rx_allowed': true, + }), + 200, + ); + } + uploads.add(DateTime.now()); + return http.Response(json.encode({'success': true}), 200); + }), + ); + final queue = ApiQueueService( + apiService: api, + networkState: network, + ); + return (queue: queue, api: api, network: network, uploads: uploads); + } + + void connect(FakeAsync async, ApiService api) { + api.requestAuth( + reason: 'connect', + publicKey: 'AB', + lat: 45.27, + lon: -75.78, + ); + async.flushMicrotasks(); + } + + void enqueueTx(FakeAsync async, ApiQueueService queue) { + queue.enqueueTx( + latitude: 45.27, + longitude: -75.78, + heardRepeats: '4e(12.25)', + timestamp: 1789200000, + externalAntenna: false, + ); + async.flushMicrotasks(); + } + + void enqueueOtherRoutineItems(FakeAsync async, ApiQueueService queue) { + queue.enqueueDisc( + latitude: 45.27, + longitude: -75.78, + repeaterId: '4e', + nodeType: 'repeater', + localSnr: 12.25, + localRssi: -95, + remoteSnr: 8.5, + pubkeyFull: 'AB', + timestamp: 1789200001, + externalAntenna: false, + ); + queue.enqueueTrace( + latitude: 45.27, + longitude: -75.78, + repeaterId: '4e', + localSnr: 12.25, + localRssi: -95, + remoteSnr: 8.5, + timestamp: 1789200002, + externalAntenna: false, + ); + queue.enqueueDiscDrop( + latitude: 45.27, + longitude: -75.78, + timestamp: 1789200003, + externalAntenna: false, + ); + queue.enqueueDefer( + latitude: 45.27, + longitude: -75.78, + timestamp: 1789200004, + held: 'tx', + ); + async.flushMicrotasks(); + } + + void dispose( + ({ + ApiQueueService queue, + ApiService api, + TestNetworkStateSource network, + List uploads, + }) built) { + built.queue.dispose(); + built.api.dispose(); + built.network.close(); + } + + test('ordinary network uploads five seconds after a ping', () { + fakeAsync((async) { + final built = build(NetworkState.unconstrained); + connect(async, built.api); + enqueueTx(async, built.queue); + + async.elapse(const Duration(seconds: 4)); + expect(built.uploads, isEmpty); + async.elapse(const Duration(seconds: 1)); + expect(built.uploads, hasLength(1)); + + dispose(built); + }); + }); + + test('constrained network paces every routine item at sixty seconds', () { + fakeAsync((async) { + final built = build(constrained); + connect(async, built.api); + enqueueTx(async, built.queue); + enqueueOtherRoutineItems(async, built.queue); + + async.elapse(const Duration(seconds: 59)); + expect(built.uploads, isEmpty); + async.elapse(const Duration(seconds: 1)); + expect(built.uploads, hasLength(1)); + + dispose(built); + }); + }); + + test('becoming constrained postpones a pending ping upload', () { + fakeAsync((async) { + final built = build(NetworkState.unconstrained); + connect(async, built.api); + enqueueTx(async, built.queue); + + async.elapse(const Duration(seconds: 2)); + built.network.emit(constrained); + async.elapse(const Duration(seconds: 59)); + expect(built.uploads, isEmpty); + async.elapse(const Duration(seconds: 1)); + expect(built.uploads, hasLength(1)); + + dispose(built); + }); + }); + + test('leaving a constrained network advances a pending ping upload', () { + fakeAsync((async) { + final built = build(constrained); + connect(async, built.api); + enqueueTx(async, built.queue); + + async.elapse(const Duration(seconds: 10)); + built.network.emit(NetworkState.unconstrained); + async.elapse(const Duration(seconds: 4)); + expect(built.uploads, isEmpty); + async.elapse(const Duration(seconds: 1)); + expect(built.uploads, hasLength(1)); + + dispose(built); + }); + }); + + test('duplicate constrained state does not postpone a pending upload', () { + fakeAsync((async) { + final built = build(constrained); + connect(async, built.api); + enqueueTx(async, built.queue); + + async.elapse(const Duration(seconds: 30)); + built.network.emit(constrained); + async.elapse(const Duration(seconds: 29)); + expect(built.uploads, isEmpty); + async.elapse(const Duration(seconds: 1)); + expect(built.uploads, hasLength(1)); + + dispose(built); + }); + }); +} diff --git a/test/services/api_service_network_timeout_test.dart b/test/services/api_service_network_timeout_test.dart new file mode 100644 index 00000000..79d744ab --- /dev/null +++ b/test/services/api_service_network_timeout_test.dart @@ -0,0 +1,67 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:mesh_mapper/services/api_service.dart'; +import 'package:mesh_mapper/services/network_state_service.dart'; + +import 'test_network_state_source.dart'; + +void main() { + const constrained = NetworkState(isConstrained: true, isSatellite: true); + + ApiService unresponsiveApi(NetworkState state) => ApiService( + networkState: TestNetworkStateSource(state), + client: MockClient( + (_) => Completer().future, + ), + ); + + test('ordinary network auth times out after ten seconds', () { + fakeAsync((async) { + final api = unresponsiveApi(NetworkState.unconstrained); + var completed = false; + api + .requestAuth( + reason: 'connect', + publicKey: 'AB', + lat: 45.27, + lon: -75.78, + ) + .whenComplete(() => completed = true); + async.flushMicrotasks(); + + async.elapse(const Duration(seconds: 9)); + expect(completed, isFalse); + async.elapse(const Duration(seconds: 1)); + expect(completed, isTrue); + + api.dispose(); + }); + }); + + test('constrained network auth times out after thirty seconds', () { + fakeAsync((async) { + final api = unresponsiveApi(constrained); + var completed = false; + api + .requestAuth( + reason: 'connect', + publicKey: 'AB', + lat: 45.27, + lon: -75.78, + ) + .whenComplete(() => completed = true); + async.flushMicrotasks(); + + async.elapse(const Duration(seconds: 29)); + expect(completed, isFalse); + async.elapse(const Duration(seconds: 1)); + expect(completed, isTrue); + + api.dispose(); + }); + }); +} diff --git a/test/services/test_network_state_source.dart b/test/services/test_network_state_source.dart new file mode 100644 index 00000000..9d9a9bcb --- /dev/null +++ b/test/services/test_network_state_source.dart @@ -0,0 +1,23 @@ +import 'dart:async'; + +import 'package:mesh_mapper/services/network_state_service.dart'; + +class TestNetworkStateSource implements NetworkStateSource { + final StreamController _controller = + StreamController.broadcast(sync: true); + + TestNetworkStateSource([this.current = NetworkState.unconstrained]); + + @override + NetworkState current; + + @override + Stream get stream => _controller.stream; + + void emit(NetworkState state) { + current = state; + _controller.add(state); + } + + Future close() => _controller.close(); +}