From 7cff6059a60a471d6a9506a6c0a1779d439be89b Mon Sep 17 00:00:00 2001 From: Doug Borg Date: Thu, 12 Feb 2026 14:32:20 -0700 Subject: [PATCH 1/2] Parallelize quadrant fetching and add smart 429 retry (#109) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When Overpass queries hit the 50k node limit, _fetchSplitAreas() splits the area into 4 quadrants. Previously these were fetched sequentially — at max split depth (3) that's up to 64 serial HTTP requests, which is the primary cause of the slow "splitting" network status users report. This change: - Parallelizes quadrant fetches with Future.wait() instead of a for-loop - Adds a dynamic concurrency limiter (_AsyncSemaphore) sized from the Overpass /api/status slot count, so we never exceed the server's per-IP rate limit - Replaces the "sleep 30s and give up on 429" behavior with smart retry: polls /api/status until a slot is available (modeled after OSMnx and OSMPythonTools), then retries up to 2 times - Resizes the semaphore on retry with the latest observed slot count, adapting to changing server conditions Refactors for testability: - HTTP client injection in OverpassService (following RoutingService pattern) - DI constructors + forTesting() factories on NodeDataManager and NodeSpatialCache - splitBounds made static + @visibleForTesting for direct unit testing 18 new tests covering: splitBounds geometry, getSlotCount/waitForSlot parsing, fetchWithSplitting (happy path, split, max depth, rate limit retry + cap), partial/total quadrant failure, recursive splitting, and semaphore init deduplication. Co-Authored-By: Claude Opus 4.6 --- lib/services/node_data_manager.dart | 240 +++++++--- lib/services/node_spatial_cache.dart | 3 + lib/services/overpass_service.dart | 65 ++- pubspec.lock | 2 +- pubspec.yaml | 1 + test/services/node_data_manager_test.dart | 541 ++++++++++++++++++++++ 6 files changed, 775 insertions(+), 77 deletions(-) create mode 100644 test/services/node_data_manager_test.dart diff --git a/lib/services/node_data_manager.dart b/lib/services/node_data_manager.dart index 2dbaeeb3..b6002b9e 100644 --- a/lib/services/node_data_manager.dart +++ b/lib/services/node_data_manager.dart @@ -14,19 +14,89 @@ import 'map_data_submodules/nodes_from_local.dart'; import 'offline_area_service.dart'; import 'offline_areas/offline_area_models.dart'; +/// Resizable async semaphore for limiting concurrent Overpass requests. +class _AsyncSemaphore { + int _maxConcurrent; + int _current = 0; + final _waiters = >[]; + + _AsyncSemaphore(int maxConcurrent) : _maxConcurrent = maxConcurrent < 1 ? 1 : maxConcurrent; + + int get maxConcurrent => _maxConcurrent; + + /// Resize the semaphore. If capacity increased, wake up queued waiters. + void resize(int newMax) { + _maxConcurrent = newMax < 1 ? 1 : newMax; + // Wake exactly the number of newly available slots. + // Can't use _current in the loop condition because woken waiters + // haven't incremented it yet (their continuations are microtasks). + var available = _maxConcurrent - _current; + while (available > 0 && _waiters.isNotEmpty) { + _waiters.removeAt(0).complete(); + available--; + } + } + + Future run(Future Function() fn) async { + while (_current >= _maxConcurrent) { + final completer = Completer(); + _waiters.add(completer); + await completer.future; + } + _current++; + try { + return await fn(); + } finally { + _current--; + if (_waiters.isNotEmpty && _current < _maxConcurrent) { + _waiters.removeAt(0).complete(); + } + } + } +} + /// Coordinates node data fetching between cache, Overpass, and OSM API. /// Simple interface: give me nodes for this view with proper caching and error handling. class NodeDataManager extends ChangeNotifier { static final NodeDataManager _instance = NodeDataManager._(); factory NodeDataManager() => _instance; - NodeDataManager._(); - final OverpassService _overpassService = OverpassService(); - final NodeSpatialCache _cache = NodeSpatialCache(); - + NodeDataManager._({ + OverpassService? overpassService, + NodeSpatialCache? cache, + }) : _overpassService = overpassService ?? OverpassService(), + _cache = cache ?? NodeSpatialCache(); + + @visibleForTesting + factory NodeDataManager.forTesting({ + OverpassService? overpassService, + NodeSpatialCache? cache, + }) => NodeDataManager._(overpassService: overpassService, cache: cache); + + final OverpassService _overpassService; + final NodeSpatialCache _cache; + + // Concurrency limiter for Overpass requests + _AsyncSemaphore? _overpassSemaphore; + Future<_AsyncSemaphore>? _semaphoreInitFuture; + + Future<_AsyncSemaphore> _getOrCreateSemaphore() { + return _semaphoreInitFuture ??= _createSemaphore().catchError((e, st) { + _semaphoreInitFuture = null; // Allow retry on next fetch + Error.throwWithStackTrace(e, st); + }); + } + + Future<_AsyncSemaphore> _createSemaphore() async { + final slots = await _overpassService.getSlotCount(); + _overpassSemaphore = _AsyncSemaphore(slots); + debugPrint('[NodeDataManager] Overpass semaphore: $slots slots'); + return _overpassSemaphore!; + } + // Track ongoing user-initiated requests for status reporting final Set _userInitiatedRequests = {}; - + /// Get nodes for the given bounds and profiles. /// Returns cached data immediately if available, otherwise fetches from appropriate source. Future> getNodesFor({ @@ -43,7 +113,7 @@ class NodeDataManager extends ChangeNotifier { if (isUserInitiated) { NetworkStatus.instance.clear(); } - + if (uploadMode == UploadMode.sandbox) { // Offline + Sandbox = no nodes (local cache is production data) debugPrint('[NodeDataManager] Offline + Sandbox mode: returning no nodes'); @@ -51,7 +121,7 @@ class NodeDataManager extends ChangeNotifier { } else { // Offline + Production = use local offline areas (instant) final offlineNodes = await fetchLocalNodes(bounds: bounds, profiles: profiles); - + // Add offline nodes to cache so they integrate with the rest of the system if (offlineNodes.isNotEmpty) { _cache.addOrUpdateNodes(offlineNodes); @@ -59,7 +129,7 @@ class NodeDataManager extends ChangeNotifier { _cache.markAreaAsFetched(bounds, offlineNodes); notifyListeners(); } - + // Show brief success for user-initiated offline loads with data if (isUserInitiated && offlineNodes.isNotEmpty) { WidgetsBinding.instance.addPostFrameCallback((_) { @@ -71,7 +141,7 @@ class NodeDataManager extends ChangeNotifier { NetworkStatus.instance.setNoData(); }); } - + return offlineNodes; } } @@ -79,15 +149,15 @@ class NodeDataManager extends ChangeNotifier { // Handle sandbox mode (always fetch from OSM API, but integrate with cache system for UI) if (uploadMode == UploadMode.sandbox) { debugPrint('[NodeDataManager] Sandbox mode: fetching from OSM API'); - + // Track user-initiated requests for status reporting final requestKey = '${bounds.hashCode}_${profiles.map((p) => p.id).join('_')}_$uploadMode'; - + if (isUserInitiated && _userInitiatedRequests.contains(requestKey)) { debugPrint('[NodeDataManager] Sandbox request already in progress for this area'); return _cache.getNodesFor(bounds); } - + // Start status tracking for user-initiated requests if (isUserInitiated) { _userInitiatedRequests.add(requestKey); @@ -96,7 +166,7 @@ class NodeDataManager extends ChangeNotifier { } else { debugPrint('[NodeDataManager] Starting background sandbox request (no status reporting)'); } - + try { final nodes = await fetchOsmApiNodes( bounds: bounds, @@ -104,7 +174,7 @@ class NodeDataManager extends ChangeNotifier { uploadMode: uploadMode, maxResults: 0, ); - + // Add nodes to cache for UI integration (even though we don't rely on cache for subsequent fetches) if (nodes.isNotEmpty) { _cache.addOrUpdateNodes(nodes); @@ -113,10 +183,10 @@ class NodeDataManager extends ChangeNotifier { // Mark area as fetched even with no nodes so UI knows we've checked this area _cache.markAreaAsFetched(bounds, []); } - + // Update UI notifyListeners(); - + // Set success after the next frame renders, but only for user-initiated requests if (isUserInitiated) { WidgetsBinding.instance.addPostFrameCallback((_) { @@ -124,12 +194,12 @@ class NodeDataManager extends ChangeNotifier { }); debugPrint('[NodeDataManager] User-initiated sandbox request completed successfully: ${nodes.length} nodes'); } - + return nodes; - + } catch (e) { debugPrint('[NodeDataManager] Sandbox fetch failed: $e'); - + // Only report errors for user-initiated requests if (isUserInitiated) { if (e is RateLimitError) { @@ -141,7 +211,7 @@ class NodeDataManager extends ChangeNotifier { } debugPrint('[NodeDataManager] User-initiated sandbox request failed: $e'); } - + // Return whatever we have in cache for this area (likely empty for sandbox) return _cache.getNodesFor(bounds); } finally { @@ -159,13 +229,13 @@ class NodeDataManager extends ChangeNotifier { // Not cached - need to fetch final requestKey = '${bounds.hashCode}_${profiles.map((p) => p.id).join('_')}_$uploadMode'; - + // Only allow one user-initiated request per area at a time if (isUserInitiated && _userInitiatedRequests.contains(requestKey)) { debugPrint('[NodeDataManager] User request already in progress for this area'); return _cache.getNodesFor(bounds); } - + // Start status tracking for user-initiated requests only if (isUserInitiated) { _userInitiatedRequests.add(requestKey); @@ -177,10 +247,10 @@ class NodeDataManager extends ChangeNotifier { try { final nodes = await fetchWithSplitting(bounds, profiles, isUserInitiated: isUserInitiated); - + // Update cache and notify listeners notifyListeners(); - + // Set success after the next frame renders, but only for user-initiated requests if (isUserInitiated) { WidgetsBinding.instance.addPostFrameCallback((_) { @@ -188,12 +258,12 @@ class NodeDataManager extends ChangeNotifier { }); debugPrint('[NodeDataManager] User-initiated request completed successfully'); } - + return nodes; - + } catch (e) { debugPrint('[NodeDataManager] Fetch failed: $e'); - + // Only report errors for user-initiated requests if (isUserInitiated) { if (e is RateLimitError) { @@ -205,7 +275,7 @@ class NodeDataManager extends ChangeNotifier { } debugPrint('[NodeDataManager] User-initiated request failed: $e'); } - + // Return whatever we have in cache for this area return _cache.getNodesFor(bounds); } finally { @@ -217,88 +287,108 @@ class NodeDataManager extends ChangeNotifier { /// Fetch nodes with automatic area splitting if needed Future> fetchWithSplitting( - LatLngBounds bounds, + LatLngBounds bounds, List profiles, { int splitDepth = 0, + int rateLimitRetries = 0, bool isUserInitiated = false, }) async { const maxSplitDepth = 3; // 4^3 = 64 max sub-areas - + try { // Expand bounds slightly to reduce edge effects final expandedBounds = _expandBounds(bounds, 1.2); - - final nodes = await _overpassService.fetchNodes( - bounds: expandedBounds, - profiles: profiles, + + final semaphore = await _getOrCreateSemaphore(); + final nodes = await semaphore.run( + () => _overpassService.fetchNodes( + bounds: expandedBounds, + profiles: profiles, + ), ); - + // Success - cache the data for the expanded area _cache.markAreaAsFetched(expandedBounds, nodes); return nodes; - + } on NodeLimitError { // Hit node limit or timeout - split area if not too deep if (splitDepth >= maxSplitDepth) { debugPrint('[NodeDataManager] Max split depth reached, giving up'); return []; } - + debugPrint('[NodeDataManager] Splitting area (depth: $splitDepth)'); - + // Only report splitting status for user-initiated requests if (isUserInitiated && splitDepth == 0) { NetworkStatus.instance.setSplitting(); } - + return _fetchSplitAreas(bounds, profiles, splitDepth + 1, isUserInitiated: isUserInitiated); - + } on RateLimitError { - // Rate limited - wait and return empty - debugPrint('[NodeDataManager] Rate limited, backing off'); - await Future.delayed(const Duration(seconds: 30)); - return []; + if (rateLimitRetries >= 2) { + debugPrint('[NodeDataManager] Max rate limit retries reached, giving up'); + return []; + } + + debugPrint('[NodeDataManager] Rate limited, polling for slot (retry ${rateLimitRetries + 1}/2)'); + if (isUserInitiated) NetworkStatus.instance.setRateLimited(); + + // Poll until slot available; resize semaphore with fresh slot count + final slots = await _overpassService.waitForSlot(); + _overpassSemaphore?.resize(slots); + debugPrint('[NodeDataManager] Semaphore resized to $slots slots'); + + return fetchWithSplitting( + bounds, profiles, + splitDepth: splitDepth, + rateLimitRetries: rateLimitRetries + 1, + isUserInitiated: isUserInitiated, + ); } } - /// Fetch data by splitting area into quadrants + /// Fetch data by splitting area into quadrants (parallel) Future> _fetchSplitAreas( - LatLngBounds bounds, + LatLngBounds bounds, List profiles, int splitDepth, { bool isUserInitiated = false, }) async { - final quadrants = _splitBounds(bounds); - final allNodes = []; - - for (final quadrant in quadrants) { - try { - final nodes = await fetchWithSplitting( - quadrant, - profiles, - splitDepth: splitDepth, - isUserInitiated: isUserInitiated, - ); - allNodes.addAll(nodes); - } catch (e) { - debugPrint('[NodeDataManager] Quadrant fetch failed: $e'); - // Continue with other quadrants - } - } - + final quadrants = splitBounds(bounds); + + final results = await Future.wait( + quadrants.map((quadrant) async { + try { + return await fetchWithSplitting( + quadrant, profiles, + splitDepth: splitDepth, + isUserInitiated: isUserInitiated, + ); + } catch (e) { + debugPrint('[NodeDataManager] Quadrant fetch failed: $e'); + return []; + } + }), + ); + + final allNodes = results.expand((nodes) => nodes).toList(); debugPrint('[NodeDataManager] Split fetch complete: ${allNodes.length} total nodes'); return allNodes; } /// Split bounds into 4 quadrants - List _splitBounds(LatLngBounds bounds) { + @visibleForTesting + static List splitBounds(LatLngBounds bounds) { final centerLat = (bounds.north + bounds.south) / 2; final centerLng = (bounds.east + bounds.west) / 2; - + return [ // Southwest LatLngBounds(LatLng(bounds.south, bounds.west), LatLng(centerLat, centerLng)), - // Southeast + // Southeast LatLngBounds(LatLng(bounds.south, centerLng), LatLng(centerLat, bounds.east)), // Northwest LatLngBounds(LatLng(centerLat, bounds.west), LatLng(bounds.north, centerLng)), @@ -311,10 +401,10 @@ class NodeDataManager extends ChangeNotifier { LatLngBounds _expandBounds(LatLngBounds bounds, double factor) { final centerLat = (bounds.north + bounds.south) / 2; final centerLng = (bounds.east + bounds.west) / 2; - + final latSpan = (bounds.north - bounds.south) * factor / 2; final lngSpan = (bounds.east - bounds.west) * factor / 2; - + return LatLngBounds( LatLng(centerLat - latSpan, centerLng - lngSpan), LatLng(centerLat + latSpan, centerLng + lngSpan), @@ -347,7 +437,7 @@ class NodeDataManager extends ChangeNotifier { }) async { // Clear any cached data for this area _cache.clear(); - + // Re-fetch as user-initiated request await getNodesFor( bounds: bounds, @@ -374,16 +464,16 @@ class NodeDataManager extends ChangeNotifier { Future preloadOfflineNodes() async { try { final offlineAreaService = OfflineAreaService(); - + for (final area in offlineAreaService.offlineAreas) { if (area.status != OfflineAreaStatus.complete) continue; - + // Load nodes from this offline area final nodes = await fetchLocalNodes( bounds: area.bounds, profiles: [], // Empty profiles = load all nodes ); - + if (nodes.isNotEmpty) { _cache.addOrUpdateNodes(nodes); // Mark the offline area as having coverage so submit buttons work @@ -391,7 +481,7 @@ class NodeDataManager extends ChangeNotifier { debugPrint('[NodeDataManager] Preloaded ${nodes.length} offline nodes from area ${area.name}'); } } - + notifyListeners(); } catch (e) { debugPrint('[NodeDataManager] Error preloading offline nodes: $e'); @@ -400,4 +490,4 @@ class NodeDataManager extends ChangeNotifier { /// Get cache statistics String get cacheStats => _cache.stats.toString(); -} \ No newline at end of file +} diff --git a/lib/services/node_spatial_cache.dart b/lib/services/node_spatial_cache.dart index 35d90442..4e27fd90 100644 --- a/lib/services/node_spatial_cache.dart +++ b/lib/services/node_spatial_cache.dart @@ -13,6 +13,9 @@ class NodeSpatialCache { factory NodeSpatialCache() => _instance; NodeSpatialCache._(); + @visibleForTesting + NodeSpatialCache.forTesting(); + final List _fetchedAreas = []; final Map _nodes = {}; // nodeId -> node diff --git a/lib/services/overpass_service.dart b/lib/services/overpass_service.dart index 0d0f7ca5..bce36e32 100644 --- a/lib/services/overpass_service.dart +++ b/lib/services/overpass_service.dart @@ -13,11 +13,13 @@ import 'http_client.dart'; /// Single responsibility: Make requests, handle network errors, return data. class OverpassService { static const String _endpoint = 'https://overpass-api.de/api/interpreter'; + static const String _statusEndpoint = 'https://overpass-api.de/api/status'; + static const int defaultSlotCount = 4; + final http.Client _client; OverpassService({http.Client? client}) : _client = client ?? UserAgentClient(); - /// Fetch surveillance nodes from Overpass API with proper retry logic. /// Throws NetworkError for retryable failures, NodeLimitError for area splitting. Future> fetchNodes({ @@ -99,6 +101,67 @@ class OverpassService { throw NetworkError('Max retries exceeded'); } + /// Query Overpass /api/status to get the rate limit (slot count per IP). + Future getSlotCount() async { + try { + final response = await _client.get(Uri.parse(_statusEndpoint)) + .timeout(const Duration(seconds: 5)); + if (response.statusCode == 200) { + final match = RegExp(r'Rate limit:\s*(\d+)').firstMatch(response.body); + if (match != null) return int.parse(match.group(1)!); + } + } catch (e) { + debugPrint('[OverpassService] Failed to get slot count: $e'); + } + return defaultSlotCount; + } + + /// Poll /api/status until a slot is available. Returns observed slot count. + /// + /// Uses [elapsedFn] to track elapsed time. Defaults to a [Stopwatch]-based + /// implementation; tests can inject a fake to control time progression. + Future waitForSlot({ + Duration maxWait = const Duration(minutes: 2), + Duration Function()? elapsedFn, + }) async { + final stopwatch = Stopwatch()..start(); + final elapsed = elapsedFn ?? () => stopwatch.elapsed; + int observedSlots = defaultSlotCount; + + while (elapsed() < maxWait) { + try { + final response = await _client.get(Uri.parse(_statusEndpoint)) + .timeout(const Duration(seconds: 5)); + + if (response.statusCode == 200) { + // Always parse slot count while we have the response + final slotMatch = RegExp(r'Rate limit:\s*(\d+)').firstMatch(response.body); + if (slotMatch != null) observedSlots = int.parse(slotMatch.group(1)!); + + // Slots available → ready + if (response.body.contains('slots available now')) return observedSlots; + + // Parse "in N seconds" from "Slot available after: ..., in N seconds." + final match = RegExp(r'in (\d+) seconds').firstMatch(response.body); + if (match != null) { + final wait = int.parse(match.group(1)!).clamp(1, 30); + debugPrint('[OverpassService] Waiting $wait seconds for slot'); + await Future.delayed(Duration(seconds: wait)); + continue; + } + } + } catch (e) { + debugPrint('[OverpassService] Status check failed: $e'); + } + + // Fallback: wait 5 seconds and re-poll + await Future.delayed(const Duration(seconds: 5)); + } + + debugPrint('[OverpassService] Max wait time exceeded, proceeding anyway'); + return observedSlots; + } + /// Build Overpass QL query for given bounds and profiles String _buildQuery(LatLngBounds bounds, List profiles) { final nodeClauses = profiles.map((profile) { diff --git a/pubspec.lock b/pubspec.lock index 76982e6e..b61873af 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -178,7 +178,7 @@ packages: source: hosted version: "0.2.3" fake_async: - dependency: transitive + dependency: "direct dev" description: name: fake_async sha256: "5368f224a74523e8d2e7399ea1638b37aecfca824a3cc4dfdf77bf1fa905ac44" diff --git a/pubspec.yaml b/pubspec.yaml index 14f81d21..3b955f8a 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -42,6 +42,7 @@ dependencies: dev_dependencies: flutter_test: sdk: flutter + fake_async: ^1.3.3 mocktail: ^1.0.4 flutter_launcher_icons: ^0.14.4 flutter_lints: ^6.0.0 diff --git a/test/services/node_data_manager_test.dart b/test/services/node_data_manager_test.dart new file mode 100644 index 00000000..d279b3d4 --- /dev/null +++ b/test/services/node_data_manager_test.dart @@ -0,0 +1,541 @@ +import 'package:fake_async/fake_async.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:latlong2/latlong.dart'; +import 'package:flutter_map/flutter_map.dart'; +import 'package:mocktail/mocktail.dart'; + +import 'package:deflockapp/models/node_profile.dart'; +import 'package:deflockapp/models/osm_node.dart'; +import 'package:deflockapp/services/overpass_service.dart'; +import 'package:deflockapp/services/node_data_manager.dart'; +import 'package:deflockapp/services/node_spatial_cache.dart'; + +class MockOverpassService extends Mock implements OverpassService {} + +class MockNodeSpatialCache extends Mock implements NodeSpatialCache {} + +class MockHttpClient extends Mock implements http.Client {} + +void main() { + final testBounds = LatLngBounds( + const LatLng(38.0, -78.0), + const LatLng(39.0, -77.0), + ); + + final testProfiles = [ + NodeProfile( + id: 'test', + name: 'Test Profile', + tags: const {'man_made': 'surveillance'}, + ), + ]; + + OsmNode makeNode(int id, {double lat = 38.5, double lng = -77.5}) => OsmNode( + id: id, + coord: LatLng(lat, lng), + tags: const {'man_made': 'surveillance'}, + ); + + setUpAll(() { + registerFallbackValue(testBounds); + registerFallbackValue([]); + registerFallbackValue([]); + registerFallbackValue(Uri.parse('https://example.com')); + registerFallbackValue(const Duration(seconds: 1)); + }); + + group('splitBounds', () { + test('splits into 4 correct quadrants with center at midpoint', () { + final bounds = LatLngBounds( + const LatLng(0.0, 0.0), + const LatLng(10.0, 10.0), + ); + + final quadrants = NodeDataManager.splitBounds(bounds); + + expect(quadrants, hasLength(4)); + + // Southwest + expect(quadrants[0].south, 0.0); + expect(quadrants[0].west, 0.0); + expect(quadrants[0].north, 5.0); + expect(quadrants[0].east, 5.0); + + // Southeast + expect(quadrants[1].south, 0.0); + expect(quadrants[1].west, 5.0); + expect(quadrants[1].north, 5.0); + expect(quadrants[1].east, 10.0); + + // Northwest + expect(quadrants[2].south, 5.0); + expect(quadrants[2].west, 0.0); + expect(quadrants[2].north, 10.0); + expect(quadrants[2].east, 5.0); + + // Northeast + expect(quadrants[3].south, 5.0); + expect(quadrants[3].west, 5.0); + expect(quadrants[3].north, 10.0); + expect(quadrants[3].east, 10.0); + }); + + test('quadrants tile exactly - no gaps or overlaps', () { + final bounds = LatLngBounds( + const LatLng(10.0, 20.0), + const LatLng(30.0, 40.0), + ); + + final quadrants = NodeDataManager.splitBounds(bounds); + + // Total area should equal original + final totalLatSpan = quadrants.map((q) => q.north - q.south).reduce((a, b) => a + b); + final totalLngSpan = quadrants.map((q) => q.east - q.west).reduce((a, b) => a + b); + + // Each quadrant is half the span, and we have 4 quadrants (2x2) + // Total lat span = 2 * half = full span + expect(totalLatSpan, closeTo((bounds.north - bounds.south) * 2, 1e-10)); + expect(totalLngSpan, closeTo((bounds.east - bounds.west) * 2, 1e-10)); + + // Verify edges align at center + final centerLat = (bounds.north + bounds.south) / 2; + final centerLng = (bounds.east + bounds.west) / 2; + + for (final q in quadrants) { + // Every quadrant edge should be either an original edge or the center + expect( + q.south == bounds.south || q.south == centerLat, + isTrue, + reason: 'south edge ${q.south} should be original south or center', + ); + expect( + q.north == bounds.north || q.north == centerLat, + isTrue, + reason: 'north edge ${q.north} should be original north or center', + ); + expect( + q.west == bounds.west || q.west == centerLng, + isTrue, + reason: 'west edge ${q.west} should be original west or center', + ); + expect( + q.east == bounds.east || q.east == centerLng, + isTrue, + reason: 'east edge ${q.east} should be original east or center', + ); + } + }); + }); + + group('OverpassService.getSlotCount', () { + late MockHttpClient mockClient; + late OverpassService service; + + setUp(() { + mockClient = MockHttpClient(); + service = OverpassService(client: mockClient); + }); + + test('parses Rate limit from status response', () async { + when(() => mockClient.get(any())).thenAnswer( + (_) async => http.Response( + 'Connected as: 123456\n' + 'Current time: 2025-01-01T00:00:00Z\n' + 'Rate limit: 6\n' + '2 slots available now.', + 200, + ), + ); + + final count = await service.getSlotCount(); + expect(count, 6); + }); + + test('falls back to defaultSlotCount on HTTP failure', () async { + when(() => mockClient.get(any())).thenAnswer( + (_) async => http.Response('Server Error', 500), + ); + + final count = await service.getSlotCount(); + expect(count, OverpassService.defaultSlotCount); + }); + + test('falls back to defaultSlotCount on network error', () async { + when(() => mockClient.get(any())).thenThrow( + http.ClientException('Connection refused'), + ); + + final count = await service.getSlotCount(); + expect(count, OverpassService.defaultSlotCount); + }); + }); + + group('OverpassService.waitForSlot', () { + late MockHttpClient mockClient; + late OverpassService service; + + setUp(() { + mockClient = MockHttpClient(); + service = OverpassService(client: mockClient); + }); + + test('returns immediately when slots available now', () async { + when(() => mockClient.get(any())).thenAnswer( + (_) async => http.Response( + 'Rate limit: 6\n2 slots available now.', + 200, + ), + ); + + final slots = await service.waitForSlot(); + expect(slots, 6); + verify(() => mockClient.get(any())).called(1); + }); + + test('waits and re-polls when "in N seconds" in response', () { + FakeAsync().run((fake) { + var fakeElapsed = Duration.zero; + var callCount = 0; + when(() => mockClient.get(any())).thenAnswer((_) async { + callCount++; + if (callCount == 1) { + return http.Response( + 'Rate limit: 4\nSlot available after: 2025-01-01T00:00:03Z, in 1 seconds.', + 200, + ); + } + return http.Response( + 'Rate limit: 4\n2 slots available now.', + 200, + ); + }); + + late int slots; + service.waitForSlot(elapsedFn: () => fakeElapsed).then((s) => slots = s); + + fakeElapsed = const Duration(seconds: 1); + fake.elapse(const Duration(seconds: 2)); + + expect(slots, 4); + expect(callCount, 2); + }); + }); + + test('falls back to 5s poll on unparseable response', () { + FakeAsync().run((fake) { + var fakeElapsed = Duration.zero; + var callCount = 0; + when(() => mockClient.get(any())).thenAnswer((_) async { + callCount++; + if (callCount == 1) { + return http.Response('some garbage response', 200); + } + return http.Response( + 'Rate limit: 4\n1 slots available now.', + 200, + ); + }); + + late int slots; + service.waitForSlot(elapsedFn: () => fakeElapsed).then((s) => slots = s); + + fakeElapsed = const Duration(seconds: 5); + fake.elapse(const Duration(seconds: 6)); + + expect(slots, 4); + expect(callCount, 2); + }); + }); + + test('returns updated slot count if Rate limit changes', () { + FakeAsync().run((fake) { + var fakeElapsed = Duration.zero; + var callCount = 0; + when(() => mockClient.get(any())).thenAnswer((_) async { + callCount++; + if (callCount == 1) { + return http.Response( + 'Rate limit: 4\nSlot available after: ..., in 1 seconds.', + 200, + ); + } + return http.Response( + 'Rate limit: 8\n3 slots available now.', + 200, + ); + }); + + late int slots; + service.waitForSlot(elapsedFn: () => fakeElapsed).then((s) => slots = s); + + fakeElapsed = const Duration(seconds: 1); + fake.elapse(const Duration(seconds: 2)); + + expect(slots, 8); + }); + }); + + test('returns default slot count when maxWait deadline expires', () { + FakeAsync().run((fake) { + var fakeElapsed = Duration.zero; + when(() => mockClient.get(any())).thenAnswer( + (_) async => http.Response('Rate limit: 6\nNo slots right now.', 200), + ); + + late int slots; + service.waitForSlot( + maxWait: const Duration(seconds: 10), + elapsedFn: () => fakeElapsed, + ).then((s) => slots = s); + + // Advance past maxWait + fakeElapsed = const Duration(seconds: 11); + fake.elapse(const Duration(seconds: 6)); + + expect(slots, 6); + }); + }); + }); + + group('fetchWithSplitting', () { + late MockOverpassService mockOverpass; + late MockNodeSpatialCache mockCache; + late NodeDataManager manager; + + setUp(() { + mockOverpass = MockOverpassService(); + mockCache = MockNodeSpatialCache(); + manager = NodeDataManager.forTesting( + overpassService: mockOverpass, + cache: mockCache, + ); + + // Default: semaphore init returns 4 slots + when(() => mockOverpass.getSlotCount()).thenAnswer((_) async => 4); + + // Default: cache operations are no-ops + when(() => mockCache.markAreaAsFetched(any(), any())).thenReturn(null); + }); + + test('happy path - returns nodes and caches them', () async { + final nodes = [makeNode(1), makeNode(2)]; + + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async => nodes); + + final result = await manager.fetchWithSplitting(testBounds, testProfiles); + + expect(result, hasLength(2)); + verify(() => mockCache.markAreaAsFetched(any(), any())).called(1); + }); + + test('NodeLimitError splits into 4 and combines results', () async { + var callCount = 0; + + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async { + callCount++; + if (callCount == 1) { + throw NodeLimitError('too many nodes'); + } + return [makeNode(callCount)]; + }); + + final result = await manager.fetchWithSplitting(testBounds, testProfiles); + + // First call throws, then 4 quadrant calls succeed + expect(result, hasLength(4)); + verify(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).called(5); // 1 initial + 4 quadrants + }); + + test('max depth + NodeLimitError returns empty', () async { + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenThrow(NodeLimitError('too many nodes')); + + final result = await manager.fetchWithSplitting( + testBounds, testProfiles, + splitDepth: 3, + ); + + expect(result, isEmpty); + }); + + test('RateLimitError polls for slot, resizes semaphore, retries', () async { + var callCount = 0; + + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async { + callCount++; + if (callCount == 1) { + throw RateLimitError('rate limited'); + } + return [makeNode(1)]; + }); + + when(() => mockOverpass.waitForSlot(maxWait: any(named: 'maxWait'))) + .thenAnswer((_) async => 6); + + final result = await manager.fetchWithSplitting(testBounds, testProfiles); + + expect(result, hasLength(1)); + verify(() => mockOverpass.waitForSlot(maxWait: any(named: 'maxWait'))).called(1); + }); + + test('RateLimitError x3 gives up after 2 retries', () async { + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenThrow(RateLimitError('rate limited')); + + when(() => mockOverpass.waitForSlot(maxWait: any(named: 'maxWait'))) + .thenAnswer((_) async => 4); + + final result = await manager.fetchWithSplitting(testBounds, testProfiles); + + expect(result, isEmpty); + // Called twice (retry 1 and retry 2), third attempt gives up + verify(() => mockOverpass.waitForSlot(maxWait: any(named: 'maxWait'))).called(2); + }); + }); + + group('_fetchSplitAreas (via fetchWithSplitting)', () { + late MockOverpassService mockOverpass; + late MockNodeSpatialCache mockCache; + late NodeDataManager manager; + + setUp(() { + mockOverpass = MockOverpassService(); + mockCache = MockNodeSpatialCache(); + manager = NodeDataManager.forTesting( + overpassService: mockOverpass, + cache: mockCache, + ); + + when(() => mockOverpass.getSlotCount()).thenAnswer((_) async => 4); + when(() => mockCache.markAreaAsFetched(any(), any())).thenReturn(null); + }); + + test('partial failure - 1 quadrant throws, other 3 return nodes', () async { + var callCount = 0; + + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async { + callCount++; + if (callCount == 1) { + // Initial call: trigger split + throw NodeLimitError('too many nodes'); + } + if (callCount == 2) { + // First quadrant: network error + throw NetworkError('connection failed'); + } + // Other 3 quadrants succeed + return [makeNode(callCount)]; + }); + + final result = await manager.fetchWithSplitting(testBounds, testProfiles); + + // 3 of 4 quadrants returned 1 node each + expect(result, hasLength(3)); + }); + + test('all quadrants fail returns empty', () async { + var callCount = 0; + + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async { + callCount++; + if (callCount == 1) { + throw NodeLimitError('too many nodes'); + } + throw NetworkError('connection failed'); + }); + + final result = await manager.fetchWithSplitting(testBounds, testProfiles); + expect(result, isEmpty); + }); + + test('recursive splitting - depth-1 NodeLimitError, depth-2 success', () async { + var callCount = 0; + + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async { + callCount++; + // First 5 calls all hit node limit (1 initial + 4 quadrants at depth 1) + if (callCount <= 5) { + throw NodeLimitError('too many nodes'); + } + // Depth-2 calls succeed + return [makeNode(callCount)]; + }); + + final result = await manager.fetchWithSplitting(testBounds, testProfiles); + + // 4 quadrants at depth 1 each split into 4 = 16 depth-2 fetches + expect(result, hasLength(16)); + }); + }); + + group('semaphore initialization', () { + late MockOverpassService mockOverpass; + late MockNodeSpatialCache mockCache; + late NodeDataManager manager; + + setUp(() { + mockOverpass = MockOverpassService(); + mockCache = MockNodeSpatialCache(); + manager = NodeDataManager.forTesting( + overpassService: mockOverpass, + cache: mockCache, + ); + + when(() => mockCache.markAreaAsFetched(any(), any())).thenReturn(null); + }); + + test('concurrent calls to semaphore init return same instance', () async { + var getSlotCallCount = 0; + when(() => mockOverpass.getSlotCount()).thenAnswer((_) async { + getSlotCallCount++; + // Simulate slow network + await Future.delayed(const Duration(milliseconds: 10)); + return 4; + }); + + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async => [makeNode(1)]); + + // Launch two concurrent fetches + final results = await Future.wait([ + manager.fetchWithSplitting(testBounds, testProfiles), + manager.fetchWithSplitting(testBounds, testProfiles), + ]); + + // Both should succeed + expect(results[0], hasLength(1)); + expect(results[1], hasLength(1)); + + // getSlotCount should only be called once (shared init future) + expect(getSlotCallCount, 1); + }); + }); +} From 9bebba6f7c771440ef0f23c4810fe11d33fd85d0 Mon Sep 17 00:00:00 2001 From: Doug Borg Date: Thu, 12 Feb 2026 17:13:31 -0700 Subject: [PATCH 2/2] Cancel stale Overpass fetch requests via generation counter When the user pans/zooms mid-fetch, queued sub-requests now bail out instead of blocking the semaphore. Each getNodesFor() call increments _fetchGeneration; fetchWithSplitting checks _isStale(generation) at 6 cooperative checkpoints (before semaphore, inside lambda, before splitting, before/after waitForSlot, top of _fetchSplitAreas). Null generation (offline download, existing tests) is never stale, so there is zero regression risk for non-production-path callers. Co-Authored-By: Claude Opus 4.6 --- lib/services/map_data_provider.dart | 6 +- lib/services/node_data_manager.dart | 86 +++++++++++++--- test/services/node_data_manager_test.dart | 116 +++++++++++++++++++++- 3 files changed, 188 insertions(+), 20 deletions(-) diff --git a/lib/services/map_data_provider.dart b/lib/services/map_data_provider.dart index 6f5e99b5..231de6ac 100644 --- a/lib/services/map_data_provider.dart +++ b/lib/services/map_data_provider.dart @@ -60,7 +60,11 @@ class MapDataProvider { throw OfflineModeException("Cannot fetch remote nodes for offline area download in offline mode."); } - // For downloads, always fetch fresh data (don't use cache) + // For downloads, always fetch fresh data (don't use cache). + // Note: passes null generation, so downloads are never cancelled by stale-fetch + // detection and will hold semaphore slots until complete. This is intentional — + // offline downloads should run to completion — but means concurrent downloads + // can block foreground map fetches via the shared semaphore. return _nodeDataManager.fetchWithSplitting(bounds, profiles); } diff --git a/lib/services/node_data_manager.dart b/lib/services/node_data_manager.dart index b6002b9e..58dd8618 100644 --- a/lib/services/node_data_manager.dart +++ b/lib/services/node_data_manager.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:collection'; import 'package:flutter/widgets.dart'; import 'package:latlong2/latlong.dart'; import 'package:flutter_map/flutter_map.dart'; @@ -18,7 +19,7 @@ import 'offline_areas/offline_area_models.dart'; class _AsyncSemaphore { int _maxConcurrent; int _current = 0; - final _waiters = >[]; + final _waiters = Queue>(); _AsyncSemaphore(int maxConcurrent) : _maxConcurrent = maxConcurrent < 1 ? 1 : maxConcurrent; @@ -32,7 +33,7 @@ class _AsyncSemaphore { // haven't incremented it yet (their continuations are microtasks). var available = _maxConcurrent - _current; while (available > 0 && _waiters.isNotEmpty) { - _waiters.removeAt(0).complete(); + _waiters.removeFirst().complete(); available--; } } @@ -49,7 +50,7 @@ class _AsyncSemaphore { } finally { _current--; if (_waiters.isNotEmpty && _current < _maxConcurrent) { - _waiters.removeAt(0).complete(); + _waiters.removeFirst().complete(); } } } @@ -80,6 +81,24 @@ class NodeDataManager extends ChangeNotifier { _AsyncSemaphore? _overpassSemaphore; Future<_AsyncSemaphore>? _semaphoreInitFuture; + // Generation counter for cancelling stale fetch requests. + // Each new getNodesFor() call increments this; queued work checks before proceeding. + int _fetchGeneration = 0; + int? _lastLoggedStaleGeneration; + + bool _isStale(int? generation) { + if (generation == null || generation == _fetchGeneration) return false; + if (_lastLoggedStaleGeneration != generation) { + _lastLoggedStaleGeneration = generation; + debugPrint('[NodeDataManager] Fetch generation $generation is stale ' + '(current: $_fetchGeneration), cancelling remaining work'); + } + return true; + } + + @visibleForTesting + void advanceFetchGeneration() => _fetchGeneration++; + Future<_AsyncSemaphore> _getOrCreateSemaphore() { return _semaphoreInitFuture ??= _createSemaphore().catchError((e, st) { _semaphoreInitFuture = null; // Allow retry on next fetch @@ -245,8 +264,15 @@ class NodeDataManager extends ChangeNotifier { debugPrint('[NodeDataManager] Starting background request (no status reporting)'); } + final generation = ++_fetchGeneration; try { - final nodes = await fetchWithSplitting(bounds, profiles, isUserInitiated: isUserInitiated); + final nodes = await fetchWithSplitting(bounds, profiles, + isUserInitiated: isUserInitiated, generation: generation); + + // If this fetch became stale (user panned away), skip UI updates + if (_isStale(generation)) { + return _cache.getNodesFor(bounds); + } // Update cache and notify listeners notifyListeners(); @@ -264,8 +290,8 @@ class NodeDataManager extends ChangeNotifier { } catch (e) { debugPrint('[NodeDataManager] Fetch failed: $e'); - // Only report errors for user-initiated requests - if (isUserInitiated) { + // Skip error reporting for stale requests + if (isUserInitiated && !_isStale(generation)) { if (e is RateLimitError) { NetworkStatus.instance.setRateLimited(); } else if (e.toString().contains('timeout')) { @@ -285,30 +311,43 @@ class NodeDataManager extends ChangeNotifier { } } - /// Fetch nodes with automatic area splitting if needed + /// Fetch nodes with automatic area splitting if needed. + /// When [generation] is non-null, the request is cancelled if a newer + /// generation has started (user panned/zoomed away). Future> fetchWithSplitting( LatLngBounds bounds, List profiles, { int splitDepth = 0, int rateLimitRetries = 0, bool isUserInitiated = false, + int? generation, }) async { const maxSplitDepth = 3; // 4^3 = 64 max sub-areas + // Checkpoint 1: bail before entering semaphore + if (_isStale(generation)) return []; + try { // Expand bounds slightly to reduce edge effects final expandedBounds = _expandBounds(bounds, 1.2); final semaphore = await _getOrCreateSemaphore(); - final nodes = await semaphore.run( - () => _overpassService.fetchNodes( - bounds: expandedBounds, - profiles: profiles, - ), + // Checkpoint 2: stale request woke from queue — don't make HTTP call + final nodes = await semaphore.run>( + () { + if (_isStale(generation)) return Future.value([]); + return _overpassService.fetchNodes( + bounds: expandedBounds, + profiles: profiles, + ); + }, ); - // Success - cache the data for the expanded area - _cache.markAreaAsFetched(expandedBounds, nodes); + // Cache real data even if stale (valid for if user pans back). + // Skip marking area if stale and got empty result (short-circuited). + if (nodes.isNotEmpty || !_isStale(generation)) { + _cache.markAreaAsFetched(expandedBounds, nodes); + } return nodes; } on NodeLimitError { @@ -318,6 +357,9 @@ class NodeDataManager extends ChangeNotifier { return []; } + // Checkpoint 3: don't spawn 4 new sub-requests for stale fetch + if (_isStale(generation)) return []; + debugPrint('[NodeDataManager] Splitting area (depth: $splitDepth)'); // Only report splitting status for user-initiated requests @@ -325,7 +367,8 @@ class NodeDataManager extends ChangeNotifier { NetworkStatus.instance.setSplitting(); } - return _fetchSplitAreas(bounds, profiles, splitDepth + 1, isUserInitiated: isUserInitiated); + return _fetchSplitAreas(bounds, profiles, splitDepth + 1, + isUserInitiated: isUserInitiated, generation: generation); } on RateLimitError { if (rateLimitRetries >= 2) { @@ -333,11 +376,18 @@ class NodeDataManager extends ChangeNotifier { return []; } + // Checkpoint 4: don't wait up to 2 minutes for a stale request + if (_isStale(generation)) return []; + debugPrint('[NodeDataManager] Rate limited, polling for slot (retry ${rateLimitRetries + 1}/2)'); if (isUserInitiated) NetworkStatus.instance.setRateLimited(); // Poll until slot available; resize semaphore with fresh slot count final slots = await _overpassService.waitForSlot(); + + // Checkpoint 5: became stale during the wait + if (_isStale(generation)) return []; + _overpassSemaphore?.resize(slots); debugPrint('[NodeDataManager] Semaphore resized to $slots slots'); @@ -346,6 +396,7 @@ class NodeDataManager extends ChangeNotifier { splitDepth: splitDepth, rateLimitRetries: rateLimitRetries + 1, isUserInitiated: isUserInitiated, + generation: generation, ); } } @@ -356,7 +407,11 @@ class NodeDataManager extends ChangeNotifier { List profiles, int splitDepth, { bool isUserInitiated = false, + int? generation, }) async { + // Checkpoint 6: don't spawn quadrants for stale tree + if (_isStale(generation)) return []; + final quadrants = splitBounds(bounds); final results = await Future.wait( @@ -366,6 +421,7 @@ class NodeDataManager extends ChangeNotifier { quadrant, profiles, splitDepth: splitDepth, isUserInitiated: isUserInitiated, + generation: generation, ); } catch (e) { debugPrint('[NodeDataManager] Quadrant fetch failed: $e'); diff --git a/test/services/node_data_manager_test.dart b/test/services/node_data_manager_test.dart index d279b3d4..515bc23c 100644 --- a/test/services/node_data_manager_test.dart +++ b/test/services/node_data_manager_test.dart @@ -89,12 +89,9 @@ void main() { final quadrants = NodeDataManager.splitBounds(bounds); - // Total area should equal original + // Summing all quadrant spans gives 2x original (2 rows + 2 columns of half-spans) final totalLatSpan = quadrants.map((q) => q.north - q.south).reduce((a, b) => a + b); final totalLngSpan = quadrants.map((q) => q.east - q.west).reduce((a, b) => a + b); - - // Each quadrant is half the span, and we have 4 quadrants (2x2) - // Total lat span = 2 * half = full span expect(totalLatSpan, closeTo((bounds.north - bounds.south) * 2, 1e-10)); expect(totalLngSpan, closeTo((bounds.east - bounds.west) * 2, 1e-10)); @@ -494,6 +491,117 @@ void main() { }); }); + group('stale fetch cancellation', () { + late MockOverpassService mockOverpass; + late MockNodeSpatialCache mockCache; + late NodeDataManager manager; + + setUp(() { + mockOverpass = MockOverpassService(); + mockCache = MockNodeSpatialCache(); + manager = NodeDataManager.forTesting( + overpassService: mockOverpass, + cache: mockCache, + ); + + when(() => mockOverpass.getSlotCount()).thenAnswer((_) async => 4); + when(() => mockCache.markAreaAsFetched(any(), any())).thenReturn(null); + }); + + test('stale generation skips fetch entirely', () async { + manager.advanceFetchGeneration(); + + final result = await manager.fetchWithSplitting( + testBounds, testProfiles, + generation: 0, + ); + + expect(result, isEmpty); + verifyNever(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )); + }); + + test('stale generation inside semaphore lambda prevents HTTP call', () async { + when(() => mockOverpass.getSlotCount()).thenAnswer((_) async { + manager.advanceFetchGeneration(); + return 4; + }); + + final result = await manager.fetchWithSplitting( + testBounds, testProfiles, + generation: 0, + ); + + expect(result, isEmpty); + verifyNever(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )); + }); + + test('stale generation prevents recursive splitting', () async { + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async { + manager.advanceFetchGeneration(); + throw NodeLimitError('too many nodes'); + }); + + final result = await manager.fetchWithSplitting( + testBounds, testProfiles, + generation: 0, + ); + + expect(result, isEmpty); + // Only the initial call, no quadrant fetches + verify(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).called(1); + }); + + test('stale generation skips waitForSlot', () async { + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async { + manager.advanceFetchGeneration(); + throw RateLimitError('rate limited'); + }); + + final result = await manager.fetchWithSplitting( + testBounds, testProfiles, + generation: 0, + ); + + expect(result, isEmpty); + verifyNever(() => mockOverpass.waitForSlot(maxWait: any(named: 'maxWait'))); + }); + + test('null generation is never stale (backward compat)', () async { + final nodes = [makeNode(1), makeNode(2)]; + + when(() => mockOverpass.fetchNodes( + bounds: any(named: 'bounds'), + profiles: any(named: 'profiles'), + )).thenAnswer((_) async => nodes); + + // Advance generation many times + for (var i = 0; i < 10; i++) { + manager.advanceFetchGeneration(); + } + + // Call without generation parameter — null generation is never stale + final result = await manager.fetchWithSplitting(testBounds, testProfiles); + + expect(result, hasLength(2)); + verify(() => mockCache.markAreaAsFetched(any(), any())).called(1); + }); + }); + group('semaphore initialization', () { late MockOverpassService mockOverpass; late MockNodeSpatialCache mockCache;