From a87be7d8d645543bcdb6d7a20664f7116089edc7 Mon Sep 17 00:00:00 2001
From: Andrey Kerchin <akerchin@viacode.com>
Date: Thu, 9 Jul 2026 16:18:16 +0400
Subject: [PATCH] Keep ingestion pipeline alive across transient database
 failures

TaskProcessingHostedService.ExecuteAsync wrapped the whole while loop in a
single try/catch, so a transient exception from GetNew() (for example the
Postgres server restarting during maintenance) exited the loop and stopped
the background service permanently while the web host kept serving. The pod
stayed Ready, so nothing restarted it and ingestion silently stopped. Move
the per-iteration work into RunOnceAsync with its own try/catch, so a bad
tick is logged and the loop keeps polling and recovers on the next tick.
This mirrors how IngestionSchedulerHostedService already handles it.

ResetUnfinishedTasks only re-queued InProgress tasks with IsRetriable=true.
A non-retriable task left InProgress by such an interruption stayed stuck
forever and deadlocked 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. Reset every InProgress task at
startup regardless of IsRetriable, since an InProgress task at startup was
by definition interrupted and must be re-queued.
---
 AiCoreApi/Data/Processors/TaskProcessor.cs    |  7 +++-
 .../TaskProcessingHostedService.cs            | 42 +++++++++++++------
 2 files changed, 35 insertions(+), 14 deletions(-)

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