feat(api, Core, TaskReconciliationWorker): Воркер обработки зависших задач. Доработка логики.

PARR.API.sln

@@ -98,6 +98,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PARR.Infrastructure", "PARR
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PARR.WorkloadBuilderWorker", "PARR.WorkloadBuilderWorker\PARR.WorkloadBuilderWorker.csproj", "{835BD6CF-024E-47F5-BB02-C904AC9733D9}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "PARR.TaskReconciliationWorker", "PARR.TaskReconciliationWorker\PARR.TaskReconciliationWorker.csproj", "{DB701295-1696-4E1A-9088-D3AECF823BA6}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -274,6 +276,10 @@ Global
 		{835BD6CF-024E-47F5-BB02-C904AC9733D9}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{835BD6CF-024E-47F5-BB02-C904AC9733D9}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{835BD6CF-024E-47F5-BB02-C904AC9733D9}.Release|Any CPU.Build.0 = Release|Any CPU
+		{DB701295-1696-4E1A-9088-D3AECF823BA6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{DB701295-1696-4E1A-9088-D3AECF823BA6}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{DB701295-1696-4E1A-9088-D3AECF823BA6}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{DB701295-1696-4E1A-9088-D3AECF823BA6}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

PARR.API/Program.cs

@@ -6,8 +6,6 @@ using PARR.API.Installers;
 using PARR.API.Settings;
 using PARR.Core;
 using PARR.DAL;
-using PARR.Domain.Enums;
-using PARR.Domain.Settings;
 using PARR.Infrastructure;
 using Serilog;
 using System.Reflection;

PARR.Core/DependencyInjection.cs

@@ -8,6 +8,7 @@ using PARR.Core.Services.Task.Handlers.Factory;
 using PARR.Core.Services.Task.Implementations;
 using PARR.Core.Services.Task.Interfaces;
 using PARR.Core.Services.Task.Providers;
+using PARR.Core.Services.Task.ReconciliationHosted;
 using PARR.Core.Services.Workload.Implementations;
 using PARR.Core.Services.Workload.Interfaces;
 using PARR.Domain.Enums;
@@ -108,6 +109,30 @@ namespace PARR.Core
         }
 
 
+        /// <summary>
+        /// Подключение сервисов для TaskReconciliation
+        /// </summary>
+        /// <param name="services"></param>
+        /// <param name="configureSettings"></param>
+        /// <returns></returns>
+        public static IServiceCollection AddTaskReconciliation(this IServiceCollection services, IReconciliationSettings reconciliationSettings)
+        {
+            if (reconciliationSettings == null)
+                throw new ArgumentNullException(nameof(reconciliationSettings));
+
+            // Настройки 
+            services.AddSingleton(reconciliationSettings);
+
+            // Сервис
+            services.AddScoped<ITaskReconciliationService, TaskReconciliationService>();
+
+            // Воркер
+            services.AddHostedService<ReconciliationHostedService>();
+
+            return services;
+        }
+
+
         private static IServiceCollection AddBllMapping(this IServiceCollection services)
         {
             // AutoMapper

PARR.Core/Services/Task/Implementations/TaskManagementService.cs

@@ -89,6 +89,10 @@ namespace PARR.Core.Services.Task.Implementations
                 // Не пробрасываем исключение дальше — задача создана, просто не в очереди.
                 // задача останется в БД со статусом pending, и Reconciliation Task позже ее возмет в работу
                 logger.LogError("Не удалось опубликовать задачу {TaskId} в очередь. Задача осталась в БД со статусом Pending.", task.Id);
+
+                // так как задачу не смогли отправить в очередь повторно, обновим ей DateModified, относительно нее считается время обработки, и задача встанет на повтор позже сама
+                task.DateModified = DateTimeOffset.UtcNow;
+                await taskRepository.CommitAsync();
             }
 
             return task.Id;

PARR.Core/Services/Task/Implementations/TaskReconciliationService.cs

@@ -98,12 +98,15 @@ namespace PARR.Core.Services.Task.Implementations
 
             return await taskRepository.Get()
                             .Where(t =>
-                                        t.TypeCode == taskType.Code && t.DateModified < timeoutThreshold &&
+                                        //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)
+                                                || (t.StatusCode == TaskItemStatusEnum.Pending /*&& t.RetryCount > 0*/)
                                             )
                                         )
                             .OrderBy(t => t.DateModified) // Сначала самые старые

PARR.Core/Services/Task/ReconciliationHosted/ReconciliationHostedService.cs

@@ -0,0 +1,54 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using PARR.Core.Common.Interfaces;
+using PARR.Core.Services.Task.Interfaces;
+using PARR.Domain.Settings;
+
+namespace PARR.Core.Services.Task.ReconciliationHosted
+{
+    /// <summary>
+    /// Фоновая задача для периодического запуска Reconciliation Job. 
+    /// Обработка зависших задач.
+    /// </summary>
+    internal class ReconciliationHostedService : BackgroundService
+    {
+        // Это реализация воркера, подключается напрямую в воркер
+
+        private readonly ILogger<ReconciliationHostedService> logger;
+        private readonly IReconciliationSettings reconciliationSettings;
+        private readonly IServiceProvider serviceProvider;
+        private readonly IIntervalService intervalService;
+
+        public ReconciliationHostedService(
+            ILogger<ReconciliationHostedService> logger,
+            IReconciliationSettings reconciliationSettings,
+            IServiceProvider serviceProvider,
+            IIntervalService intervalService
+            )
+        {
+            this.logger = logger;
+            this.reconciliationSettings = reconciliationSettings;
+            this.serviceProvider = serviceProvider;
+            this.intervalService = intervalService;
+        }
+
+        protected override async System.Threading.Tasks.Task ExecuteAsync(CancellationToken stoppingToken)
+        {
+            await intervalService.IntervalInitAsync(async () =>
+            {
+                await using (var scope = serviceProvider.CreateAsyncScope())
+                {
+                    var reconciliationService = scope.ServiceProvider.GetRequiredService<ITaskReconciliationService>();
+
+                    // Запускаем проверку
+                    var report = await reconciliationService.RunAsync();
+
+                    if (report.HasChanged)
+                        logger.LogInformation("Reconciliation Job выполнен. Проверено: {Checked}, Исправлено: {Fixed}", report.CheckedCount, report.FixedCount);
+                }
+
+            }, reconciliationSettings.CheckInterval);
+        }
+    }
+}

PARR.Core/Services/Workload/Implementations/WorkloadCacheBuilderService.cs

@@ -2,7 +2,6 @@
 using Microsoft.Extensions.Logging;
 using PARR.Core.Common.Interfaces;
 using PARR.Core.Common.Interfaces.RabbitServices;
-using PARR.Core.Services.Task.Handlers;
 using PARR.Core.Services.Task.Handlers.Factory;
 using PARR.Core.Services.Workload.Interfaces;
 using PARR.Domain.Common.Rabbit.Messages;

PARR.TaskReconciliationWorker/PARR.TaskReconciliationWorker.csproj

@@ -0,0 +1,25 @@
+<Project Sdk="Microsoft.NET.Sdk.Worker">
+
+  <PropertyGroup>
+    <TargetFramework>net7.0</TargetFramework>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <UserSecretsId>dotnet-PARR.TaskReconciliationWorker-90386488-d496-4f4d-8dcd-7dcc50882876</UserSecretsId>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Elastic.CommonSchema.Serilog" Version="8.6.1" />
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="7.0.1" />
+    <PackageReference Include="Serilog.Extensions.Hosting" Version="7.0.0" />
+    <PackageReference Include="Serilog.Settings.Configuration" Version="7.0.1" />
+    <PackageReference Include="Serilog.Sinks.Console" Version="4.1.0" />
+    <PackageReference Include="Serilog.Sinks.File" Version="5.0.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\PARR.Core\PARR.Core.csproj" />
+    <ProjectReference Include="..\PARR.DAL\PARR.DAL.csproj" />
+    <ProjectReference Include="..\PARR.Domain\PARR.Domain.csproj" />
+    <ProjectReference Include="..\PARR.Infrastructure\PARR.Infrastructure.csproj" />
+  </ItemGroup>
+</Project>

PARR.TaskReconciliationWorker/Program.cs

@@ -0,0 +1,49 @@
+using Elastic.CommonSchema.Serilog;
+using PARR.Core;
+using PARR.DAL;
+using PARR.Infrastructure;
+using PARR.TaskReconciliationWorker.Settings;
+using Serilog;
+
+var builder = Host.CreateApplicationBuilder();
+
+builder.Services.AddLogging(config =>
+{
+    config.ClearProviders();
+
+    var logger = new LoggerConfiguration();
+
+    if (builder.Environment.IsProduction())
+        logger.WriteTo.Console(new EcsTextFormatter());
+    else
+        logger.WriteTo.Console();
+
+    logger.ReadFrom.Configuration(builder.Configuration);
+
+    config.AddSerilog(logger.CreateLogger());
+});
+
+// --- Настройки ---
+var workerSettings = new WorkerSettings();
+builder.Configuration.GetSection(nameof(WorkerSettings)).Bind(workerSettings);
+
+builder.Services.InstallDalServices(builder.Configuration);
+builder.Services.AddCoreServices(builder.Configuration);
+builder.Services.AddInfrastructureServices(builder.Configuration);
+builder.Configuration.AddDalConfigurations(builder.Services);
+builder.Services.AddDallSettings(builder.Configuration);
+
+builder.Services.AddTaskManagement((provider, taskType) =>
+{
+    // настройки очередей
+    //var workerSettings = provider.GetRequiredService<WorkerSettings>();
+    if (workerSettings.MqSettings.Tasks.TryGetValue(taskType, out var settings))
+        return settings;
+
+    throw new ArgumentException($"Настройки для типа задания {taskType} не найдены в конфигурации.");
+});
+
+builder.Services.AddTaskReconciliation(workerSettings);
+
+var host = builder.Build();
+host.Run();

PARR.TaskReconciliationWorker/Properties/launchSettings.json

@@ -0,0 +1,11 @@
+{
+  "profiles": {
+    "PARR.TaskReconciliationWorker": {
+      "commandName": "Project",
+      "dotnetRunMessages": true,
+      "environmentVariables": {
+        "DOTNET_ENVIRONMENT": "Development"
+      }
+    }
+  }
+}

PARR.TaskReconciliationWorker/Settings/WorkerSettings.cs

@@ -0,0 +1,18 @@
+using PARR.Domain.Enums;
+using PARR.Domain.Settings;
+
+namespace PARR.TaskReconciliationWorker.Settings
+{
+    internal class WorkerSettings : IReconciliationSettings
+    {
+        public TimeSpan CheckInterval { get; set; } = TimeSpan.FromMinutes(10);
+        public int MaxTaskPerRun { get; set; } = 100;
+
+        public MqSettings MqSettings { get; set; } = new();
+    }
+
+    internal class MqSettings
+    {
+        public Dictionary<TaskTypeEnum, MqSettingsBase> Tasks { get; set; } = new();
+    }
+}

PARR.TaskReconciliationWorker/appsettings.Development.json

@@ -0,0 +1,38 @@
+{
+  "ConnectionStrings": {
+    "RedisConnection": "10.99.253.216:6379,password=ParrP@ssPtk202MMdevDvs"
+  },
+  "Logging": {
+    "LogLevel": {
+      "Default": "Debug",
+      "Microsoft.Hosting.Lifetime": "Information"
+    }
+  },
+  "Serilog": {
+    "MinimumLevel": {
+      "Default": "Debug",
+      "Override": {
+        "Microsoft": "Warning",
+        "Microsoft.Hosting.Lifetime": "Information"
+      }
+    },
+    "WriteTo": [
+      {
+        "Name": "File",
+        "Args": {
+          "path": "log/log-.txt",
+          "rollingInterval": "Day"
+        }
+      }
+    ]
+  },
+  "WorkerSettings": {
+    "MqSettings": {
+      "Tasks": {
+        "Workload": {
+          "HostName": "10.99.253.216"
+        }
+      }
+    }
+  }
+}

PARR.TaskReconciliationWorker/appsettings.json

@@ -0,0 +1,39 @@
+{
+  "ConnectionStrings": {
+    "DefaultConnection": "Server=10.99.253.184;Database=parr;User Id=app_parr; Password=PosdfkhT&)%sdfligL&%5546;",
+    "RedisConnection": "parr-redis:6379,password=ParrP@ssPtk202MMdevDvs"
+  },
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Microsoft.EntityFrameworkCore": "Error",
+      "Microsoft.EntityFrameworkCore.Database.Command": "Warning",
+      "Microsoft.AspNetCore": "Warning"
+    }
+  },
+  "Serilog": {
+    "MinimumLevel": {
+      "Default": "Information",
+      "Override": {
+        "Microsoft": "Warning",
+        "Microsoft.EntityFrameworkCore": "Error",
+        "Microsoft.EntityFrameworkCore.Database.Command": "Warning",
+        "Microsoft.Hosting.Lifetime": "Information"
+      }
+    }
+  },
+  "WorkerSettings": {
+    "CheckInterval": "00:10:00",
+    "MaxTaskPerRun": 100,
+    "MqSettings": {
+      "Tasks": {
+        "Workload": {
+          "HostName": "parr-rabbitmq",
+          "QueueName": "parr-task-workload",
+          "User": "task_workload_writer",
+          "Password": "UFtsduifytIUR$^85382Wt1few"
+        }
+      }
+    }
+  }
+}