Keep ingestion pipeline alive across transient database failures

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));
         }
     }

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<ITaskProcessor>();
+        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<ITaskProcessor>();
 
-                    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.");
             }
         }