feat(api, Core, Domain, Dal): Для entity моделей, для DateModified создан атрибут ManualControl - отключает автоматическое управление датой. TaskReconciliationService - сервис по управлению задачами в статусах ошибки, зависла. ITaskMqSettingsProvider - провайдер для управления настройками MQ в зависимости от TaskType
This commit is contained in:
Mikhail Trubnikov
@@ -0,0 +1,187 @@
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using PARR.Core.Common.Interfaces.RabbitServices;
|
||||
using PARR.Core.Repositories.Interfaces.TaskRepositories;
|
||||
using PARR.Core.Services.Task.Interfaces;
|
||||
using PARR.Core.Services.Task.Models;
|
||||
using PARR.Core.Services.Task.Providers;
|
||||
using PARR.Domain.Common.Rabbit.Messages;
|
||||
using PARR.Domain.Entities.TaskEntities;
|
||||
using PARR.Domain.Enums;
|
||||
using PARR.Domain.Settings;
|
||||
|
||||
namespace PARR.Core.Services.Task.Implementations
|
||||
{
|
||||
internal class TaskReconciliationService : ITaskReconciliationService
|
||||
{
|
||||
private readonly ILogger<TaskReconciliationService> 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<TaskReconciliationService> 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<ReconciliationReport> 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;
|
||||
}
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// Поиск зависших задач для конкретного типа.
|
||||
/// </summary>
|
||||
/// <param name="taskType"></param>
|
||||
/// <returns></returns>
|
||||
private async Task<List<TaskItem>> 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 &&
|
||||
(
|
||||
// Зависла в Processing
|
||||
t.StatusCode == TaskItemStatusEnum.Processing
|
||||
//Зависли в Pending
|
||||
|| (t.StatusCode == TaskItemStatusEnum.Pending && t.RetryCount > 0)
|
||||
)
|
||||
)
|
||||
.OrderBy(t => t.DateModified) // Сначала самые старые
|
||||
.Take(reconciliationSettings.MaxTaskPerRun) // Защита от перегрузки
|
||||
.ToListAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Обработка одной зависшей задачи.
|
||||
/// </summary>
|
||||
/// <param name="task"></param>
|
||||
/// <param name="taskType"></param>
|
||||
/// <param name="report"></param>
|
||||
/// <returns></returns>
|
||||
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<object> { 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;
|
||||
|
||||
await taskRepository.CommitAsync();
|
||||
|
||||
report.RetriedCount++;
|
||||
report.Messages.Add($"Задача {task.Id}: опубликована в очередь, попытка #{task.RetryCount}");
|
||||
report.FixedCount++;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user