From b66d3a0277ea0e2ff7ed84dc595f5790dc9bff11 Mon Sep 17 00:00:00 2001 From: Anton Martynenko Date: Wed, 19 Nov 2025 13:01:30 +0100 Subject: [PATCH] parallel processing for routes and regions --- SatelliteProvider.Api/Program.cs | 6 +- SatelliteProvider.Api/appsettings.json | 9 +- .../Configs/ProcessingConfig.cs | 3 + .../GoogleMapsDownloaderV2.cs | 255 +++++++++++++----- .../RegionProcessingService.cs | 42 ++- .../RegionRequestQueue.cs | 21 +- SatelliteProvider.Services/RegionService.cs | 17 +- .../RouteProcessingService.cs | 111 ++++---- 8 files changed, 331 insertions(+), 133 deletions(-) diff --git a/SatelliteProvider.Api/Program.cs b/SatelliteProvider.Api/Program.cs index b898dac..5f18489 100644 --- a/SatelliteProvider.Api/Program.cs +++ b/SatelliteProvider.Api/Program.cs @@ -32,7 +32,11 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); var processingConfig = builder.Configuration.GetSection("ProcessingConfig").Get() ?? new ProcessingConfig(); -builder.Services.AddSingleton(sp => new RegionRequestQueue(processingConfig.QueueCapacity)); +builder.Services.AddSingleton(sp => +{ + var logger = sp.GetRequiredService>(); + return new RegionRequestQueue(processingConfig.QueueCapacity, logger); +}); builder.Services.AddSingleton(); builder.Services.AddHostedService(); builder.Services.AddSingleton(); diff --git a/SatelliteProvider.Api/appsettings.json b/SatelliteProvider.Api/appsettings.json index 0063b7c..946513b 100644 --- a/SatelliteProvider.Api/appsettings.json +++ b/SatelliteProvider.Api/appsettings.json @@ -4,7 +4,9 @@ "MinimumLevel": { "Default": "Information", "Override": { - "Microsoft.AspNetCore": "Warning" + "Microsoft.AspNetCore": "Warning", + "SatelliteProvider.Services.RegionRequestQueue": "Debug", + "SatelliteProvider.Services.GoogleMapsDownloaderV2": "Debug" } }, "WriteTo": [ @@ -33,7 +35,10 @@ }, "ProcessingConfig": { "MaxConcurrentDownloads": 4, + "MaxConcurrentRegions": 3, "DefaultZoomLevel": 20, - "QueueCapacity": 100 + "QueueCapacity": 1000, + "DelayBetweenRequestsMs": 50, + "SessionTokenReuseCount": 100 } } diff --git a/SatelliteProvider.Common/Configs/ProcessingConfig.cs b/SatelliteProvider.Common/Configs/ProcessingConfig.cs index b8806c5..7480f83 100644 --- a/SatelliteProvider.Common/Configs/ProcessingConfig.cs +++ b/SatelliteProvider.Common/Configs/ProcessingConfig.cs @@ -3,7 +3,10 @@ namespace SatelliteProvider.Common.Configs; public class ProcessingConfig { public int MaxConcurrentDownloads { get; set; } = 4; + public int MaxConcurrentRegions { get; set; } = 3; public int DefaultZoomLevel { get; set; } = 20; public int QueueCapacity { get; set; } = 100; + public int DelayBetweenRequestsMs { get; set; } = 50; + public int SessionTokenReuseCount { get; set; } = 100; } diff --git a/SatelliteProvider.Services/GoogleMapsDownloaderV2.cs b/SatelliteProvider.Services/GoogleMapsDownloaderV2.cs index 523c958..86c9347 100644 --- a/SatelliteProvider.Services/GoogleMapsDownloaderV2.cs +++ b/SatelliteProvider.Services/GoogleMapsDownloaderV2.cs @@ -28,18 +28,24 @@ public class GoogleMapsDownloaderV2 private readonly ILogger _logger; private readonly string _apiKey; private readonly StorageConfig _storageConfig; + private readonly ProcessingConfig _processingConfig; private readonly IHttpClientFactory _httpClientFactory; + private readonly SemaphoreSlim _downloadSemaphore; + private static readonly System.Collections.Concurrent.ConcurrentDictionary> _activeDownloads = new(); public GoogleMapsDownloaderV2( ILogger logger, IOptions mapConfig, IOptions storageConfig, + IOptions processingConfig, IHttpClientFactory httpClientFactory) { _logger = logger; _apiKey = mapConfig.Value.ApiKey; _storageConfig = storageConfig.Value; + _processingConfig = processingConfig.Value; _httpClientFactory = httpClientFactory; + _downloadSemaphore = new SemaphoreSlim(_processingConfig.MaxConcurrentDownloads, _processingConfig.MaxConcurrentDownloads); } private record SessionResponse(string Session); @@ -247,15 +253,13 @@ public class GoogleMapsDownloaderV2 centerGeoPoint.Lat, centerGeoPoint.Lon, radiusM, zoomLevel); _logger.LogInformation("Tile range: X=[{XMin}, {XMax}], Y=[{YMin}, {YMax}]", xMin, xMax, yMin, yMax); - var downloadedTiles = new List(); + var tilesToDownload = new List<(int x, int y, GeoPoint center, double tileSizeMeters)>(); int skippedCount = 0; for (var y = yMin; y <= yMax; y++) { for (var x = xMin; x <= xMax; x++) { - token.ThrowIfCancellationRequested(); - var tileCenter = GeoUtils.TileToWorldPos(x, y, zoomLevel); var existingTile = existingTiles.FirstOrDefault(t => @@ -266,83 +270,190 @@ public class GoogleMapsDownloaderV2 if (existingTile != null) { skippedCount++; - _logger.LogInformation("Skipping tile ({X}, {Y}) - already exists at {FilePath}", x, y, existingTile.FilePath); continue; } var tileSizeMeters = CalculateTileSizeInMeters(zoomLevel, tileCenter.Lat); - - try - { - var sessionToken = await GetSessionToken(); - var server = (x + y) % 4; - var url = string.Format(TILE_URL_TEMPLATE, server, x, y, zoomLevel, sessionToken); - - var timestamp = DateTime.UtcNow.ToString("yyyyMMddHHmmss"); - var subdirectory = _storageConfig.GetTileSubdirectoryPath(zoomLevel, x, y); - Directory.CreateDirectory(subdirectory); - - var filePath = _storageConfig.GetTileFilePath(zoomLevel, x, y, timestamp); - - var imageBytes = await ExecuteWithRetryAsync(async () => - { - using var httpClient = _httpClientFactory.CreateClient(); - httpClient.DefaultRequestHeaders.UserAgent.ParseAdd(USER_AGENT); - - var response = await httpClient.GetAsync(url, token); - - if (!response.IsSuccessStatusCode) - { - var errorBody = await response.Content.ReadAsStringAsync(token); - _logger.LogError("Tile download failed. Tile: ({X}, {Y}), Status: {StatusCode}, URL: {Url}, Response: {Response}", - x, y, response.StatusCode, url, errorBody); - } - - response.EnsureSuccessStatusCode(); - - return await response.Content.ReadAsByteArrayAsync(token); - }, cancellationToken: token); - - await File.WriteAllBytesAsync(filePath, imageBytes, token); - - _logger.LogInformation("Downloaded tile ({X}, {Y}) to {FilePath}, center=({Lat:F6}, {Lon:F6}), size={Size:F2}m", - x, y, filePath, tileCenter.Lat, tileCenter.Lon, tileSizeMeters); - - downloadedTiles.Add(new DownloadedTileInfoV2( - x, y, zoomLevel, tileCenter.Lat, tileCenter.Lon, filePath, tileSizeMeters)); - } - catch (TaskCanceledException ex) - { - _logger.LogError(ex, "Tile download cancelled for ({X}, {Y}). This may be due to HttpClient timeout or explicit cancellation.", x, y); - throw; - } - catch (OperationCanceledException ex) - { - _logger.LogError(ex, "Tile download operation cancelled for ({X}, {Y})", x, y); - throw; - } - catch (RateLimitException ex) - { - _logger.LogError(ex, "Rate limit exceeded for tile ({X}, {Y}). Google Maps API is throttling requests. Consider reducing concurrent requests or adding delays.", x, y); - throw; - } - catch (HttpRequestException ex) - { - _logger.LogError(ex, "HTTP request failed for tile ({X}, {Y}). StatusCode: {StatusCode}, Message: {Message}", - x, y, ex.StatusCode, ex.Message); - throw; - } - catch (Exception ex) - { - _logger.LogError(ex, "Unexpected error downloading tile ({X}, {Y}). Type: {ExceptionType}, Message: {Message}", - x, y, ex.GetType().Name, ex.Message); - throw; - } + tilesToDownload.Add((x, y, tileCenter, tileSizeMeters)); } } - _logger.LogInformation("Downloaded {Count} new tiles, skipped {Skipped} existing tiles", downloadedTiles.Count, skippedCount); + _logger.LogInformation("Need to download {Count} tiles (skipped {Skipped} existing), using {MaxConcurrent} parallel downloads", + tilesToDownload.Count, skippedCount, _processingConfig.MaxConcurrentDownloads); + + if (tilesToDownload.Count == 0) + { + _logger.LogInformation("All tiles already exist, returning empty list"); + return new List(); + } + + _logger.LogInformation("Getting initial session token before starting {Count} downloads", tilesToDownload.Count); + var sessionToken = await GetSessionToken(); + _logger.LogInformation("Session token obtained, starting parallel downloads"); + + var downloadTasks = new List>(); + int sessionTokenUsageCount = 0; + + for (int i = 0; i < tilesToDownload.Count; i++) + { + var tileInfo = tilesToDownload[i]; + + if (sessionTokenUsageCount >= _processingConfig.SessionTokenReuseCount) + { + _logger.LogInformation("Session token usage limit reached ({Count}), requesting new token", sessionTokenUsageCount); + sessionToken = await GetSessionToken(); + sessionTokenUsageCount = 0; + _logger.LogInformation("New session token obtained, continuing downloads"); + } + + var currentToken = sessionToken; + var tileIndex = i; + sessionTokenUsageCount++; + + var downloadTask = DownloadTileAsync( + tileInfo.x, + tileInfo.y, + tileInfo.center, + tileInfo.tileSizeMeters, + zoomLevel, + currentToken, + tileIndex, + tilesToDownload.Count, + token); + + downloadTasks.Add(downloadTask); + } + + _logger.LogInformation("All {Count} download tasks created, waiting for completion", downloadTasks.Count); + var results = await Task.WhenAll(downloadTasks); + _logger.LogInformation("Task.WhenAll completed, processing results"); + + var downloadedTiles = results.Where(r => r != null).Cast().ToList(); + + _logger.LogInformation("Parallel download completed: {Downloaded} tiles downloaded, {Skipped} skipped, {Failed} failed", + downloadedTiles.Count, skippedCount, tilesToDownload.Count - downloadedTiles.Count); + return downloadedTiles; } + + private async Task DownloadTileAsync( + int x, + int y, + GeoPoint tileCenter, + double tileSizeMeters, + int zoomLevel, + string? sessionToken, + int tileIndex, + int totalTiles, + CancellationToken token) + { + var tileKey = $"{zoomLevel}_{x}_{y}"; + + var downloadTask = _activeDownloads.GetOrAdd(tileKey, _ => PerformDownloadAsync( + x, y, tileCenter, tileSizeMeters, zoomLevel, sessionToken, tileIndex, totalTiles, token)); + + try + { + return await downloadTask; + } + finally + { + _activeDownloads.TryRemove(tileKey, out _); + } + } + + private async Task PerformDownloadAsync( + int x, + int y, + GeoPoint tileCenter, + double tileSizeMeters, + int zoomLevel, + string? sessionToken, + int tileIndex, + int totalTiles, + CancellationToken token) + { + _logger.LogDebug("Tile ({X},{Y}) [{Index}/{Total}]: Waiting for semaphore slot", x, y, tileIndex + 1, totalTiles); + await _downloadSemaphore.WaitAsync(token); + _logger.LogDebug("Tile ({X},{Y}) [{Index}/{Total}]: Acquired semaphore slot, starting download", x, y, tileIndex + 1, totalTiles); + + try + { + if (_processingConfig.DelayBetweenRequestsMs > 0) + { + await Task.Delay(_processingConfig.DelayBetweenRequestsMs, token); + } + + var server = (x + y) % 4; + var url = string.Format(TILE_URL_TEMPLATE, server, x, y, zoomLevel, sessionToken); + + var timestamp = DateTime.UtcNow.ToString("yyyyMMddHHmmss"); + var subdirectory = _storageConfig.GetTileSubdirectoryPath(zoomLevel, x, y); + Directory.CreateDirectory(subdirectory); + + var filePath = _storageConfig.GetTileFilePath(zoomLevel, x, y, timestamp); + + var imageBytes = await ExecuteWithRetryAsync(async () => + { + using var httpClient = _httpClientFactory.CreateClient(); + httpClient.DefaultRequestHeaders.UserAgent.ParseAdd(USER_AGENT); + + var response = await httpClient.GetAsync(url, token); + + if (!response.IsSuccessStatusCode) + { + var errorBody = await response.Content.ReadAsStringAsync(token); + _logger.LogError("Tile download failed. Tile: ({X}, {Y}), Status: {StatusCode}, Response: {Response}", + x, y, response.StatusCode, errorBody); + } + + response.EnsureSuccessStatusCode(); + + return await response.Content.ReadAsByteArrayAsync(token); + }, cancellationToken: token); + + await File.WriteAllBytesAsync(filePath, imageBytes, token); + + if ((tileIndex + 1) % 10 == 0 || tileIndex == 0 || tileIndex == totalTiles - 1) + { + _logger.LogInformation("Progress: {Current}/{Total} tiles downloaded - tile ({X}, {Y})", + tileIndex + 1, totalTiles, x, y); + } + + return new DownloadedTileInfoV2( + x, y, zoomLevel, tileCenter.Lat, tileCenter.Lon, filePath, tileSizeMeters); + } + catch (TaskCanceledException ex) + { + _logger.LogError(ex, "Tile download cancelled for ({X}, {Y})", x, y); + throw; + } + catch (OperationCanceledException ex) + { + _logger.LogError(ex, "Tile download operation cancelled for ({X}, {Y})", x, y); + throw; + } + catch (RateLimitException ex) + { + _logger.LogError(ex, "Rate limit exceeded for tile ({X}, {Y})", x, y); + throw; + } + catch (HttpRequestException ex) + { + _logger.LogError(ex, "HTTP request failed for tile ({X}, {Y}). StatusCode: {StatusCode}", + x, y, ex.StatusCode); + throw; + } + catch (Exception ex) + { + _logger.LogError(ex, "Unexpected error downloading tile ({X}, {Y})", x, y); + throw; + } + finally + { + _logger.LogDebug("Tile ({X},{Y}) [{Index}/{Total}]: Releasing semaphore slot", x, y, tileIndex + 1, totalTiles); + _downloadSemaphore.Release(); + } + } } diff --git a/SatelliteProvider.Services/RegionProcessingService.cs b/SatelliteProvider.Services/RegionProcessingService.cs index c2f37b9..127d00a 100644 --- a/SatelliteProvider.Services/RegionProcessingService.cs +++ b/SatelliteProvider.Services/RegionProcessingService.cs @@ -1,5 +1,7 @@ using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using SatelliteProvider.Common.Configs; using SatelliteProvider.Common.Interfaces; namespace SatelliteProvider.Services; @@ -8,21 +10,49 @@ public class RegionProcessingService : BackgroundService { private readonly IRegionRequestQueue _queue; private readonly IRegionService _regionService; + private readonly ProcessingConfig _processingConfig; private readonly ILogger _logger; public RegionProcessingService( IRegionRequestQueue queue, IRegionService regionService, + IOptions processingConfig, ILogger logger) { _queue = queue; _regionService = regionService; + _processingConfig = processingConfig.Value; _logger = logger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - _logger.LogInformation("Region Processing Service started"); + _logger.LogInformation("Region Processing Service started with {MaxConcurrent} parallel workers", + _processingConfig.MaxConcurrentRegions); + + var workers = new List(); + for (int i = 0; i < _processingConfig.MaxConcurrentRegions; i++) + { + var workerId = i + 1; + workers.Add(ProcessRegionWorkerAsync(workerId, stoppingToken)); + } + + await Task.WhenAll(workers); + + _logger.LogInformation("Region Processing Service stopped"); + } + + private async Task ProcessRegionWorkerAsync(int workerId, CancellationToken stoppingToken) + { + if (workerId > 1) + { + var startupDelay = Random.Shared.Next(100, 500); + _logger.LogInformation("Region worker {WorkerId} starting with {Delay}ms delay to reduce contention", + workerId, startupDelay); + await Task.Delay(startupDelay, stoppingToken); + } + + _logger.LogInformation("Region worker {WorkerId} started", workerId); while (!stoppingToken.IsCancellationRequested) { @@ -32,9 +62,13 @@ public class RegionProcessingService : BackgroundService if (request != null) { - _logger.LogInformation("Dequeued region request {RegionId}", request.Id); + _logger.LogInformation("Worker {WorkerId}: Dequeued region request {RegionId}", + workerId, request.Id); await _regionService.ProcessRegionAsync(request.Id, stoppingToken); + + _logger.LogInformation("Worker {WorkerId}: Completed region {RegionId}", + workerId, request.Id); } } catch (OperationCanceledException) @@ -43,11 +77,11 @@ public class RegionProcessingService : BackgroundService } catch (Exception ex) { - _logger.LogError(ex, "Error processing region request"); + _logger.LogError(ex, "Worker {WorkerId}: Error processing region request", workerId); } } - _logger.LogInformation("Region Processing Service stopped"); + _logger.LogInformation("Region worker {WorkerId} stopped", workerId); } } diff --git a/SatelliteProvider.Services/RegionRequestQueue.cs b/SatelliteProvider.Services/RegionRequestQueue.cs index 827d426..0864def 100644 --- a/SatelliteProvider.Services/RegionRequestQueue.cs +++ b/SatelliteProvider.Services/RegionRequestQueue.cs @@ -1,4 +1,5 @@ using System.Threading.Channels; +using Microsoft.Extensions.Logging; using SatelliteProvider.Common.DTO; using SatelliteProvider.Common.Interfaces; @@ -7,19 +8,33 @@ namespace SatelliteProvider.Services; public class RegionRequestQueue : IRegionRequestQueue { private readonly Channel _queue; + private readonly ILogger? _logger; + private int _totalEnqueued = 0; + private int _totalDequeued = 0; - public RegionRequestQueue(int capacity) + public RegionRequestQueue(int capacity, ILogger? logger = null) { var options = new BoundedChannelOptions(capacity) { FullMode = BoundedChannelFullMode.Wait }; _queue = Channel.CreateBounded(options); + _logger = logger; + _logger?.LogInformation("RegionRequestQueue created with capacity {Capacity}", capacity); } public async ValueTask EnqueueAsync(RegionRequest request, CancellationToken cancellationToken = default) { + var queueDepthBefore = Count; + _totalEnqueued++; + _logger?.LogDebug("Enqueuing region {RegionId} (queue depth before: {Depth}, total enqueued: {Total})", + request.Id, queueDepthBefore, _totalEnqueued); + await _queue.Writer.WriteAsync(request, cancellationToken); + + var queueDepthAfter = Count; + _logger?.LogDebug("Enqueued region {RegionId} (queue depth after: {Depth})", + request.Id, queueDepthAfter); } public async ValueTask DequeueAsync(CancellationToken cancellationToken = default) @@ -28,6 +43,10 @@ public class RegionRequestQueue : IRegionRequestQueue { if (_queue.Reader.TryRead(out var request)) { + _totalDequeued++; + var queueDepth = Count; + _logger?.LogDebug("Dequeued region {RegionId} (queue depth: {Depth}, total dequeued: {Total})", + request.Id, queueDepth, _totalDequeued); return request; } } diff --git a/SatelliteProvider.Services/RegionService.cs b/SatelliteProvider.Services/RegionService.cs index f56fa66..c892a02 100644 --- a/SatelliteProvider.Services/RegionService.cs +++ b/SatelliteProvider.Services/RegionService.cs @@ -79,20 +79,22 @@ public class RegionService : IRegionService public async Task ProcessRegionAsync(Guid id, CancellationToken cancellationToken = default) { - _logger.LogInformation("Processing region {RegionId}", id); + _logger.LogInformation("Processing region {RegionId} - START", id); var startTime = DateTime.UtcNow; var region = await _regionRepository.GetByIdAsync(id); if (region == null) { - _logger.LogWarning("Region {RegionId} not found", id); + _logger.LogWarning("Region {RegionId} not found in database", id); return; } + _logger.LogInformation("Region {RegionId}: Updating status to 'processing' (was: {OldStatus})", id, region.Status); region.Status = "processing"; region.UpdatedAt = DateTime.UtcNow; await _regionRepository.UpdateAsync(region); + _logger.LogInformation("Region {RegionId}: Creating timeout CTS (5 minutes)", id); using var timeoutCts = new CancellationTokenSource(TimeSpan.FromMinutes(5)); using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, timeoutCts.Token); @@ -103,32 +105,33 @@ public class RegionService : IRegionService try { - _logger.LogInformation("Downloading tiles for region {RegionId} at ({Lat}, {Lon}) size {Size}m zoom {Zoom}", + _logger.LogInformation("Region {RegionId}: Step 1 - Starting download for location ({Lat}, {Lon}) size {Size}m zoom {Zoom}", id, region.Latitude, region.Longitude, region.SizeMeters, region.ZoomLevel); var processingStartTime = DateTime.UtcNow; - _logger.LogInformation("Checking for existing tiles in region {RegionId}", id); + _logger.LogInformation("Region {RegionId}: Step 2 - Checking for existing tiles", id); var tilesBeforeDownload = await _tileService.GetTilesByRegionAsync( region.Latitude, region.Longitude, region.SizeMeters, region.ZoomLevel); var existingTileIds = new HashSet(tilesBeforeDownload.Select(t => t.Id)); - _logger.LogInformation("Found {Count} existing tiles for region {RegionId}", existingTileIds.Count, id); + _logger.LogInformation("Region {RegionId}: Step 3 - Found {Count} existing tiles", id, existingTileIds.Count); - _logger.LogInformation("Starting tile download for region {RegionId}", id); + _logger.LogInformation("Region {RegionId}: Step 4 - Calling DownloadAndStoreTilesAsync", id); tiles = await _tileService.DownloadAndStoreTilesAsync( region.Latitude, region.Longitude, region.SizeMeters, region.ZoomLevel, linkedCts.Token); + _logger.LogInformation("Region {RegionId}: Step 5 - DownloadAndStoreTilesAsync completed with {Count} tiles", id, tiles.Count); tilesDownloaded = tiles.Count(t => !existingTileIds.Contains(t.Id)); tilesReused = tiles.Count(t => existingTileIds.Contains(t.Id)); - _logger.LogInformation("Region {RegionId}: Downloaded {Downloaded} tiles, Reused {Reused} tiles", + _logger.LogInformation("Region {RegionId}: Step 6 - Downloaded {Downloaded} new tiles, Reused {Reused} existing tiles", id, tilesDownloaded, tilesReused); var readyDir = _storageConfig.ReadyDirectory; diff --git a/SatelliteProvider.Services/RouteProcessingService.cs b/SatelliteProvider.Services/RouteProcessingService.cs index 0bc4b86..742c9dc 100644 --- a/SatelliteProvider.Services/RouteProcessingService.cs +++ b/SatelliteProvider.Services/RouteProcessingService.cs @@ -97,76 +97,95 @@ public class RouteProcessingService : BackgroundService using var scope = _serviceProvider.CreateScope(); var regionService = scope.ServiceProvider.GetRequiredService(); - _logger.LogInformation("Route {RouteId}: Starting sequential region processing for {PointCount} points", + _logger.LogInformation("Route {RouteId}: Starting parallel region processing for {PointCount} points", routeId, routePointsList.Count); - var firstPoint = routePointsList.First(); - var regionId = Guid.NewGuid(); + var queuedRegionIds = new List(); - await regionService.RequestRegionAsync( - regionId, - firstPoint.Latitude, - firstPoint.Longitude, - route.RegionSizeMeters, - route.ZoomLevel, - stitchTiles: false); + foreach (var point in routePointsList) + { + var regionId = Guid.NewGuid(); + + await regionService.RequestRegionAsync( + regionId, + point.Latitude, + point.Longitude, + route.RegionSizeMeters, + route.ZoomLevel, + stitchTiles: false); + + await _routeRepository.LinkRouteToRegionAsync(routeId, regionId); + queuedRegionIds.Add(regionId); + } - await _routeRepository.LinkRouteToRegionAsync(routeId, regionId); - - _logger.LogInformation("Route {RouteId}: Queued first region {RegionId} (1/{TotalPoints})", - routeId, regionId, routePointsList.Count); + _logger.LogInformation("Route {RouteId}: Queued all {Count} regions for parallel processing. Region IDs: {RegionIds}", + routeId, queuedRegionIds.Count, string.Join(", ", queuedRegionIds.Take(5)) + (queuedRegionIds.Count > 5 ? "..." : "")); return; } - var lastRegion = await _regionRepository.GetByIdAsync(regionIdsList.Last()); - if (lastRegion == null) + var regions = new List(); + foreach (var regionId in regionIdsList) { - return; + var region = await _regionRepository.GetByIdAsync(regionId); + if (region != null) + { + regions.Add(region); + } } - if (lastRegion.Status == "queued" || lastRegion.Status == "processing") - { - return; - } + var completedRegions = regions.Where(r => r.Status == "completed").ToList(); + var failedRegions = regions.Where(r => r.Status == "failed").ToList(); + var processingRegions = regions.Where(r => r.Status == "queued" || r.Status == "processing").ToList(); - if (lastRegion.Status == "failed") + var hasEnoughCompleted = completedRegions.Count >= routePointsList.Count; + var activeRegions = completedRegions.Count + processingRegions.Count; + var shouldRetryFailed = failedRegions.Count > 0 && !hasEnoughCompleted && activeRegions < routePointsList.Count; + + if (hasEnoughCompleted) { - _logger.LogError("Route {RouteId}: Region {RegionId} failed. Stopping route processing.", - routeId, lastRegion.Id); + _logger.LogInformation("Route {RouteId}: Have {Completed} completed regions (required: {Required}). Generating final maps. Ignoring {Processing} processing and {Failed} failed regions.", + routeId, completedRegions.Count, routePointsList.Count, processingRegions.Count, failedRegions.Count); - route.MapsReady = false; - route.UpdatedAt = DateTime.UtcNow; - await _routeRepository.UpdateRouteAsync(route); + await GenerateRouteMapsAsync(routeId, route, completedRegions.Take(routePointsList.Count).Select(r => r.Id), cancellationToken); return; } - if (regionIdsList.Count < routePointsList.Count) + if (shouldRetryFailed) { + _logger.LogWarning("Route {RouteId}: {FailedCount} region(s) failed: {FailedRegions}. Active regions: {ActiveCount}/{RequiredCount}. Attempting to retry with new regions.", + routeId, failedRegions.Count, string.Join(", ", failedRegions.Select(r => r.Id)), activeRegions, routePointsList.Count); + using var scope = _serviceProvider.CreateScope(); var regionService = scope.ServiceProvider.GetRequiredService(); - var nextPoint = routePointsList[regionIdsList.Count]; - var regionId = Guid.NewGuid(); + foreach (var failedRegion in failedRegions) + { + var newRegionId = Guid.NewGuid(); + _logger.LogInformation("Route {RouteId}: Retrying failed region {OldRegionId} with new region {NewRegionId}", + routeId, failedRegion.Id, newRegionId); + + await regionService.RequestRegionAsync( + newRegionId, + failedRegion.Latitude, + failedRegion.Longitude, + failedRegion.SizeMeters, + failedRegion.ZoomLevel, + stitchTiles: false); + + await _routeRepository.LinkRouteToRegionAsync(routeId, newRegionId); + } - await regionService.RequestRegionAsync( - regionId, - nextPoint.Latitude, - nextPoint.Longitude, - route.RegionSizeMeters, - route.ZoomLevel, - stitchTiles: false); - - await _routeRepository.LinkRouteToRegionAsync(routeId, regionId); - - _logger.LogInformation("Route {RouteId}: Queued next region {RegionId} ({CurrentRegion}/{TotalPoints})", - routeId, regionId, regionIdsList.Count + 1, routePointsList.Count); + _logger.LogInformation("Route {RouteId}: Queued {Count} retry regions", routeId, failedRegions.Count); return; } - _logger.LogInformation("Route {RouteId}: All {RegionCount} regions completed, generating final maps", - routeId, regionIdsList.Count); - - await GenerateRouteMapsAsync(routeId, route, regionIdsList, cancellationToken); + var anyProcessing = processingRegions.Count > 0; + if (anyProcessing) + { + _logger.LogInformation("Route {RouteId}: Progress - {Completed}/{Required} regions completed, {Processing} still processing, {Failed} failed (will retry if needed)", + routeId, completedRegions.Count, routePointsList.Count, processingRegions.Count, failedRegions.Count); + return; + } } private async Task GenerateRouteMapsAsync(