diff --git a/AiCoreApi/Data/Processors/TaskProcessor.cs b/AiCoreApi/Data/Processors/TaskProcessor.cs index 47bb3a8..639c9dd 100644 --- a/AiCoreApi/Data/Processors/TaskProcessor.cs +++ b/AiCoreApi/Data/Processors/TaskProcessor.cs @@ -156,7 +156,12 @@ await db.Tasks.Where(t => public async Task ResetUnfinishedTasks() { await using var db = await _dbFactory.CreateDbContextAsync(); - await db.Tasks.Where(t => (t.State == TaskState.InProgress) && t.IsRetriable) + // Any task still marked InProgress at startup was interrupted (the worker that owned it + // is gone), so it must be re-queued regardless of IsRetriable. Otherwise a non-retriable + // task stays stuck InProgress forever and blocks the scheduler: GetStale().Take(N) keeps + // selecting the same stalest data sources, sees their orphaned "active" task and skips + // them, so no data source is ever synced again. + await db.Tasks.Where(t => t.State == TaskState.InProgress) .ExecuteUpdateAsync(t => t.SetProperty(x => x.State, TaskState.New)); } } diff --git a/AiCoreApi/Services/ProcessingServices/TaskProcessingHostedService.cs b/AiCoreApi/Services/ProcessingServices/TaskProcessingHostedService.cs index ce57ba2..ce77eff 100644 --- a/AiCoreApi/Services/ProcessingServices/TaskProcessingHostedService.cs +++ b/AiCoreApi/Services/ProcessingServices/TaskProcessingHostedService.cs @@ -37,28 +37,44 @@ protected override async Task ExecuteAsync(CancellationToken ct) { while (await timer.WaitForNextTickAsync(ct)) { - if (!_instanceSync.IsMainInstance) - continue; + await RunOnceAsync(ct); + } + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) { } + catch (Exception ex) + { + _logger.LogError(ex, "TaskProcessingHostedService crashed."); + } + } - using var scope = _scopeFactory.CreateScope(); - var taskProcessor = scope.ServiceProvider.GetRequiredService(); + private async Task RunOnceAsync(CancellationToken ct) + { + // A transient failure (e.g. the database restarting during maintenance) must not + // terminate the polling loop: keep the try/catch inside the loop iteration so the + // service keeps polling and recovers on the next tick instead of dying permanently. + try + { + if (!_instanceSync.IsMainInstance) + return; - var tasks = await taskProcessor.GetNew(); + using var scope = _scopeFactory.CreateScope(); + var taskProcessor = scope.ServiceProvider.GetRequiredService(); - foreach (var task in tasks) - { - if (!TryLockActivity(task.IngestionId)) - continue; + var tasks = await taskProcessor.GetNew(); - _ = ProcessOneAsync(task, ct) - .ContinueWith(_ => UnlockActivity(task.IngestionId), TaskScheduler.Default); - } + foreach (var task in tasks) + { + if (!TryLockActivity(task.IngestionId)) + continue; + + _ = ProcessOneAsync(task, ct) + .ContinueWith(_ => UnlockActivity(task.IngestionId), TaskScheduler.Default); } } catch (OperationCanceledException) when (ct.IsCancellationRequested) { } catch (Exception ex) { - _logger.LogError(ex, "TaskProcessingHostedService crashed."); + _logger.LogError(ex, "TaskProcessingHostedService iteration error."); } }