using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; using PARR.Core.Common.Interfaces.RabbitServices; using PARR.Core.Repositories.Interfaces.TaskRepositories; using PARR.Core.Services.TaskServices.Interfaces; using PARR.Core.Services.TaskServices.Models; using PARR.Core.Services.TaskServices.Providers; using PARR.Domain.Common.Rabbit.Messages; using PARR.Domain.Entities.TaskEntities; using PARR.Domain.Enums; using PARR.Domain.Settings; namespace PARR.Core.Services.TaskServices.Implementations { internal class TaskReconciliationService : ITaskReconciliationService { private readonly ILogger logger; private readonly ITaskTypeRepository taskTypeRepository; private readonly ITaskRepository taskRepository; private readonly ITaskErrorRepository taskErrorRepository; private readonly IReconciliationSettings reconciliationSettings; private readonly IRabbitService rabbitService; private readonly ITaskMqSettingsProvider taskMqSettingsProvider; public TaskReconciliationService( ILogger logger, ITaskTypeRepository taskTypeRepository, ITaskRepository taskRepository, ITaskErrorRepository taskErrorRepository, IReconciliationSettings reconciliationSettings, IRabbitService rabbitService, ITaskMqSettingsProvider taskMqSettingsProvider ) { this.logger = logger; this.taskTypeRepository = taskTypeRepository; this.taskRepository = taskRepository; this.taskErrorRepository = taskErrorRepository; this.reconciliationSettings = reconciliationSettings; this.rabbitService = rabbitService; this.taskMqSettingsProvider = taskMqSettingsProvider; } public async Task RunAsync() { var report = new ReconciliationReport { StartedAt = DateTimeOffset.UtcNow }; logger.LogInformation("Начало выполнения Reconciliation Job."); // Загружаем все типы задач (для получения максимальной длительности) var taskTypes = await taskTypeRepository.Get().AsNoTracking().ToDictionaryAsync(t => t.Code, t => t); if (taskTypes.Count == 0) { logger.LogWarning("Типы задач не настроены в БД (таблица {TaskTypes})", nameof(TaskType)); report.Messages.Add("TaskTypes пустые. Не настроены в БД."); return report; } // Для каждого типа ищем зависшие задачи foreach (var taskType in taskTypes.Values) { var stuckTasks = await FindStuckTasksAsync(taskType); report.CheckedCount += stuckTasks.Count; foreach (var task in stuckTasks) { await ProcessStuckTaskAsync(task, taskType, report); } } report.CompletedAt = DateTimeOffset.UtcNow; if (report.HasChanged) logger.LogInformation("Reconciliation Job завершен. Проверено: {Checked}, Исправлено: {Fixed}, Retry: {Retried}, Failed: {Failed}. Длительность: {Duration}", report.CheckedCount, report.FixedCount, report.RetriedCount, report.FailedCount, report.Duration); else logger.LogDebug("Reconciliation Job завершен. Зависших задач не найдено"); return report; } /// /// Поиск зависших задач для конкретного типа. /// /// /// private async Task> FindStuckTasksAsync(TaskType taskType) { // Вычисляем порог: задача зависла, если выполняется дольше чем MaxExecutionTimeMinutes var timeoutThreshold = DateTimeOffset.UtcNow.AddMinutes(-taskType.MaxExecutionTimeMinutes); logger.LogDebug("Поиск зависших задач типа {Type}. Порог: {Threshold}", taskType.Code, timeoutThreshold); return await taskRepository.Get() .Where(t => //t.TypeCode == taskType.Code && t.DateModified < timeoutThreshold && t.TypeCode == taskType.Code // если это вновь созданная задача и она упала, то у нее может не быть DateModified, тогда сравним с DateCreated && (t.DateModified ?? t.DateCreated) < timeoutThreshold && ( // Зависла в Processing t.StatusCode == TaskItemStatusEnum.Processing //Зависли в Pending || (t.StatusCode == TaskItemStatusEnum.Pending /*&& t.RetryCount > 0*/) ) ) .OrderBy(t => t.DateModified) // Сначала самые старые .Take(reconciliationSettings.MaxTaskPerRun) // Защита от перегрузки .ToListAsync(); } /// /// Обработка одной зависшей задачи. /// /// /// /// /// private async System.Threading.Tasks.Task ProcessStuckTaskAsync(TaskItem task, TaskType taskType, ReconciliationReport report) { logger.LogWarning("Обнаружена одна зависшая задача {TaskId} (тип {Type}). Последнее обновление: {LastModified}", task.Id, task.TypeCode, task.DateModified); // Записываем ошибку в историю var error = new TaskError { Id = Guid.NewGuid(), TaskId = task.Id, AttemptNumber = task.RetryCount + 1, // показываем планируемую попытку ErrorMessage = $"Превышен порог выполнения задачи. LastDateModified: {task.DateModified}" }; await taskErrorRepository.CreateAsync(error); await taskErrorRepository.CommitAsync(); // Если лимит исчерпан - ставим Failed if (task.RetryCount >= taskType.MaxRetries) { task.StatusCode = TaskItemStatusEnum.Failed; task.ProcessedAt = DateTimeOffset.UtcNow; task.DateModified = DateTimeOffset.UtcNow; await taskRepository.CommitAsync(); report.FailedCount++; report.Messages.Add($"Задача {task.Id}: закончились попытки."); return; } // Сначала пытаемся отправить в очередь var mqSettings = taskMqSettingsProvider.GetSettings(taskType.Code); var message = new TaskMessage { TaskId = task.Id, TypeCode = task.TypeCode }; var sendResult = await rabbitService.SendAsync(mqSettings, new List { message }); if (!sendResult.IsSuccess) { // Очередь недоступна — НЕ меняем БД, НЕ тратим RetryCount // Задача останется в текущем статусе и будет найдена в следующий раз logger.LogError("Не удалось отправить задачу {TaskId} в очередь. Повторим в следующей итерации.", task.Id); report.Messages.Add($"Задача {task.Id}: не смогла отправиться в очередь, обработается в следующей итерации."); return; // Выходим без сохранения изменений } // Очередь успешна - обновляем БД // Ставим задачу в ожидание task.StatusCode = TaskItemStatusEnum.Pending; // Может получиться так, что воркер возьмет задачу быстрее, чем в БД у нее установится статус Pending. // Если это произойдет, ничего страшного, воркер это не отработает. Задача опять попадет в ProcessStuckTaskAsync и повторно отправится в очередь. task.RetryCount++; task.DateModified = DateTimeOffset.UtcNow; task.ProgressPercent = 0; await taskRepository.CommitAsync(); report.RetriedCount++; report.Messages.Add($"Задача {task.Id}: опубликована в очередь, попытка #{task.RetryCount}"); report.FixedCount++; } } }