Merge branch 'main' into t2_grpc_retry

Filter/Classify, generate-keywords/evaluate-fit и Predict/Status оборачиваются в GrpcRetry; мутирующие вызовы (train/reset/telegram) не повторяются. Тесты учли повторы.

src/core/Deal.Infrastructure/Integrations/Resilience/GrpcRetry.cs

@@ -0,0 +1,66 @@
+using Deal.SharedKernel.Resilience;
+using Grpc.Core;
+
+namespace Deal.Infrastructure.Integrations.Resilience;
+
+/// <summary>
+/// Повтор транзиентных gRPC-сбоев клиентов автономных сервисов.
+/// </summary>
+public static class GrpcRetry
+{
+    /// <summary>
+    /// Число повторов после первой попытки.
+    /// </summary>
+    public const int RetryCount = 2;
+
+    /// <summary>
+    /// Базовая задержка повтора (далее — экспоненциально с джиттером).
+    /// </summary>
+    public static readonly TimeSpan BaseDelay = TimeSpan.FromMilliseconds(200);
+
+    /// <summary>
+    /// Выполняет gRPC-вызов с повтором транзиентных сбоев.
+    /// </summary>
+    /// <param name="operation">Вызов (принимает токен отмены).</param>
+    /// <param name="cancellationToken">Токен отмены.</param>
+    /// <returns>Ответ вызова.</returns>
+    public static Task<TResult> ExecuteAsync<TResult>(
+        Func<CancellationToken, Task<TResult>> operation,
+        CancellationToken cancellationToken)
+        => ExecuteAsync(operation, DefaultDelayAsync, cancellationToken);
+
+    /// <summary>
+    /// Выполняет gRPC-вызов с повтором и заданной паузой между попытками.
+    /// </summary>
+    /// <param name="operation">Вызов (принимает токен отмены).</param>
+    /// <param name="delayAsync">Пауза между попытками (в тестах — мгновенная).</param>
+    /// <param name="cancellationToken">Токен отмены.</param>
+    /// <returns>Ответ вызова.</returns>
+    public static Task<TResult> ExecuteAsync<TResult>(
+        Func<CancellationToken, Task<TResult>> operation,
+        Func<TimeSpan, CancellationToken, Task> delayAsync,
+        CancellationToken cancellationToken)
+        => RetryExecutor.ExecuteAsync(
+            operation,
+            RetryCount,
+            BaseDelay,
+            IsTransient,
+            delayAsync,
+            cancellationToken);
+
+    /// <summary>
+    /// Признак транзиентного сбоя транспорта (недоступность/дедлайн).
+    /// </summary>
+    /// <param name="exception">Исключение вызова.</param>
+    /// <returns>True — сбой имеет смысл повторить.</returns>
+    public static bool IsTransient(Exception exception)
+        => exception is RpcException rpc
+            && rpc.StatusCode is StatusCode.Unavailable or StatusCode.DeadlineExceeded;
+
+    // Экспоненциальная задержка с джиттером 0.5–1.5× (сглаживает синхронные ретраи воркеров).
+    private static Task DefaultDelayAsync(TimeSpan delay, CancellationToken cancellationToken)
+    {
+        double factor = 0.5 + Random.Shared.NextDouble();
+        return Task.Delay(TimeSpan.FromMilliseconds(delay.TotalMilliseconds * factor), cancellationToken);
+    }
+}

src/core/Deal.Infrastructure/Integrations/Services/GrpcAiClassifier.cs

@@ -3,6 +3,7 @@ using Deal.Contracts.Integrations.Models;
 using Deal.Grpc.Ai;
 using Deal.Infrastructure.Integrations.Exceptions;
 using Deal.Infrastructure.Integrations.Models;
+using Deal.Infrastructure.Integrations.Resilience;
 using Deal.Modules.Pipeline.Application.Services;
 using Deal.SharedKernel.Tenants.Abstractions;
 using Deal.SharedKernel.Tenants.Models;
@@ -75,14 +76,16 @@ public sealed class GrpcAiClassifier : IAiClassifier
             string prompt = await _contextBuilder.BuildFilterPromptAsync(ct);
             ProviderConfig providerConfig = await _providerConfigBuilder.BuildAsync(ct);
             AiService.AiServiceClient client = _connection.CreateClient();
-            FilterReply reply = await client.FilterAsync(
+            FilterReply reply = await GrpcRetry.ExecuteAsync(
-                new FilterRequest
+                token => client.FilterAsync(
-                {
+                    new FilterRequest
-                    Prompt = prompt,
+                    {
-                    Text = SliceCodePoints(text, MaxFilterTextCodePoints), // python L193: text[:4000]
+                        Prompt = prompt,
-                    ProviderConfig = providerConfig,
+                        Text = SliceCodePoints(text, MaxFilterTextCodePoints), // python L193: text[:4000]
-                },
+                        ProviderConfig = providerConfig,
-                CallOptions(tenantId.Value, ct));
+                    },
+                    CallOptions(tenantId.Value, token)).ResponseAsync,
+                ct);
             await _usageRecorder.AddAsync(reply.Usage, providerConfig.ProviderId, providerConfig.Model, ct);
 
             return new AiFilterResultDto(
@@ -113,14 +116,16 @@ public sealed class GrpcAiClassifier : IAiClassifier
         ClassifyReply reply;
         try
         {
-            reply = await client.ClassifyAsync(
+            reply = await GrpcRetry.ExecuteAsync(
-                new ClassifyRequest
+                token => client.ClassifyAsync(
-                {
+                    new ClassifyRequest
-                    SystemPrompt = systemPrompt,
+                    {
-                    UserContext = userContext,
+                        SystemPrompt = systemPrompt,
-                    ProviderConfig = providerConfig,
+                        UserContext = userContext,
-                },
+                        ProviderConfig = providerConfig,
-                CallOptions(tenantId.Value, ct));
+                    },
+                    CallOptions(tenantId.Value, token)).ResponseAsync,
+                ct);
         }
         catch (RpcException exception)
         {

src/core/Deal.Infrastructure/Integrations/Services/GrpcAiTools.cs

@@ -3,6 +3,7 @@ using Deal.Contracts.Integrations.Models;
 using Deal.Grpc.Ai;
 using Deal.Infrastructure.Integrations.Exceptions;
 using Deal.Infrastructure.Integrations.Models;
+using Deal.Infrastructure.Integrations.Resilience;
 using Deal.SharedKernel.Tenants.Abstractions;
 using Deal.SharedKernel.Tenants.Models;
 using Grpc.Core;
@@ -74,13 +75,15 @@ public sealed class GrpcAiTools : IAiTools
         {
             AiService.AiServiceClient client = _connection.CreateClient();
             ProviderConfig providerConfig = await _providerConfigBuilder.BuildAsync(ct);
-            GenerateKeywordsReply reply = await client.GenerateKeywordsAsync(
+            GenerateKeywordsReply reply = await GrpcRetry.ExecuteAsync(
-                new GenerateKeywordsRequest
+                token => client.GenerateKeywordsAsync(
-                {
+                    new GenerateKeywordsRequest
-                    Description = SliceCodePoints(description ?? string.Empty, MaxDescriptionCodePoints),
+                    {
-                    ProviderConfig = providerConfig,
+                        Description = SliceCodePoints(description ?? string.Empty, MaxDescriptionCodePoints),
-                },
+                        ProviderConfig = providerConfig,
-                CallOptions(tenantId.Value, ct));
+                    },
+                    CallOptions(tenantId.Value, token)).ResponseAsync,
+                ct);
             await _usageRecorder.AddAsync(reply.Usage, providerConfig.ProviderId, providerConfig.Model, ct);
             return new AiGenerateKeywordsResultDto(
                 Ok: true,
@@ -125,7 +128,9 @@ public sealed class GrpcAiTools : IAiTools
                 }
             }
 
-            EvaluateFitReply reply = await client.EvaluateFitAsync(request, CallOptions(tenantId.Value, ct));
+            EvaluateFitReply reply = await GrpcRetry.ExecuteAsync(
+                token => client.EvaluateFitAsync(request, CallOptions(tenantId.Value, token)).ResponseAsync,
+                ct);
             await _usageRecorder.AddAsync(reply.Usage, providerConfig.ProviderId, providerConfig.Model, ct);
             return new AiEvaluateFitResultDto(
                 Fit: reply.Fit,

src/core/Deal.Infrastructure/Integrations/Services/GrpcMlClient.cs

@@ -4,6 +4,7 @@ using Deal.Contracts.Integrations.Models;
 using Deal.Grpc.Ml;
 using Deal.Infrastructure.Integrations.Abstractions;
 using Deal.Infrastructure.Integrations.Models;
+using Deal.Infrastructure.Integrations.Resilience;
 using Deal.Modules.Kanban.Application.Abstractions;
 using Deal.Modules.Kanban.Application.Models;
 using Deal.Modules.Settings.Application.Abstractions;
@@ -124,9 +125,11 @@ public sealed class GrpcMlClient : IMlClient, IMlTrainClient
         try
         {
             MlService.MlServiceClient client = _connection.CreateClient();
-            PredictReply reply = await client.PredictAsync(
+            PredictReply reply = await GrpcRetry.ExecuteAsync(
-                new PredictRequest { Text = text ?? string.Empty },
+                token => client.PredictAsync(
-                CallOptions(tenantId.Value, TimeSpan.FromSeconds(PredictDeadlineSeconds), ct));
+                    new PredictRequest { Text = text ?? string.Empty },
+                    CallOptions(tenantId.Value, TimeSpan.FromSeconds(PredictDeadlineSeconds), token)).ResponseAsync,
+                ct);
 
             await _usageRecorder.AddEstimatedAsync(text, TokenUsageSources.Local, TokenUsageSources.Ml, ct);
             return MapPredict(reply);
@@ -220,9 +223,11 @@ public sealed class GrpcMlClient : IMlClient, IMlTrainClient
         try
         {
             MlService.MlServiceClient client = _connection.CreateClient();
-            StatusReply reply = await client.StatusAsync(
+            StatusReply reply = await GrpcRetry.ExecuteAsync(
-                new StatusRequest(),
+                token => client.StatusAsync(
-                CallOptions(tenantId.Value, TimeSpan.FromSeconds(StatusDeadlineSeconds), ct));
+                    new StatusRequest(),
+                    CallOptions(tenantId.Value, TimeSpan.FromSeconds(StatusDeadlineSeconds), token)).ResponseAsync,
+                ct);
             MlServiceStatusDto service = MapStatus(reply);
             _statusCache.Set(tenantId.Value, service, reachable: true);
             return _statusCache.TryGet(tenantId.Value, out MlStatusCache.Snapshot updated)

src/core/Deal.SharedKernel/Resilience/RetryExecutor.cs

@@ -0,0 +1,49 @@
+namespace Deal.SharedKernel.Resilience;
+
+/// <summary>
+/// Повтор операции при транзиентном сбое.
+/// </summary>
+public static class RetryExecutor
+{
+    /// <summary>
+    /// Выполняет операцию, повторяя её при транзиентном сбое с задержкой.
+    /// </summary>
+    /// <param name="operation">Операция (принимает токен отмены).</param>
+    /// <param name="retryCount">Число повторов после первой попытки.</param>
+    /// <param name="baseDelay">Базовая задержка; для повтора N — baseDelay * 2^N.</param>
+    /// <param name="shouldRetry">Предикат транзиентности сбоя.</param>
+    /// <param name="delayAsync">Пауза между попытками (в тестах — мгновенная).</param>
+    /// <param name="cancellationToken">Токен отмены.</param>
+    /// <returns>Результат первой успешной попытки.</returns>
+    public static async Task<TResult> ExecuteAsync<TResult>(
+        Func<CancellationToken, Task<TResult>> operation,
+        int retryCount,
+        TimeSpan baseDelay,
+        Func<Exception, bool> shouldRetry,
+        Func<TimeSpan, CancellationToken, Task> delayAsync,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(operation);
+        ArgumentOutOfRangeException.ThrowIfNegative(retryCount);
+        ArgumentNullException.ThrowIfNull(shouldRetry);
+        ArgumentNullException.ThrowIfNull(delayAsync);
+
+        for (int attempt = 0; ; attempt++)
+        {
+            try
+            {
+                return await operation(cancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception exception) when (attempt < retryCount
+                && shouldRetry(exception)
+                && !cancellationToken.IsCancellationRequested)
+            {
+                await delayAsync(BackoffDelay(baseDelay, attempt), cancellationToken).ConfigureAwait(false);
+            }
+        }
+    }
+
+    // Задержка повтора с экспоненциальным ростом от базовой.
+    private static TimeSpan BackoffDelay(TimeSpan baseDelay, int attempt)
+        => TimeSpan.FromMilliseconds(baseDelay.TotalMilliseconds * Math.Pow(2, attempt));
+}

src/core/tests/Deal.Tests.Unit/Contracts/GrpcMlClientTests.cs

@@ -4,6 +4,7 @@ using Deal.Infrastructure.Data;
 using Deal.Infrastructure.Integrations.Abstractions;
 using Deal.Infrastructure.Integrations.Models;
 using Deal.Infrastructure.Integrations.Options;
+using Deal.Infrastructure.Integrations.Resilience;
 using Deal.Infrastructure.Integrations.Services;
 using Deal.Modules.Kanban.Application.Models;
 using Deal.Modules.Settings.Application.Models;
@@ -91,7 +92,8 @@ public sealed class GrpcMlClientTests
             Assert.Null(result.Margin);
             Assert.Empty(result.Terms);
             Assert.Null(result.Type);
-            Assert.Single(service.RequestTenantIds);
+            // Недоступность транспорта повторяется — на сервер приходит первая попытка и повторы.
+            Assert.Equal(GrpcRetry.RetryCount + 1, service.RequestTenantIds.Count);
         });
     }
 
@@ -148,7 +150,8 @@ public sealed class GrpcMlClientTests
             Assert.False(down.Reachable);
             Assert.False(down.Service.Ready);
             Assert.False(down.Stats.Reachable);
-            Assert.Equal(1, service.StatusCalls);
+            // При недоступности транспорта идёт повтор — считаем все попытки.
+            Assert.Equal(GrpcRetry.RetryCount + 1, service.StatusCalls);
 
             // «Поднялся»: после TTL 15 с следующий StatusAsync обновляет кэш (ready=true, reachable=true).
             service.StatusUnavailable = false;
@@ -165,7 +168,7 @@ public sealed class GrpcMlClientTests
             Assert.True(up.Reachable);
             Assert.True(up.Service.Ready);
             Assert.Equal(3, up.Service.Learned);
-            Assert.Equal(2, service.StatusCalls);
+            Assert.Equal(GrpcRetry.RetryCount + 2, service.StatusCalls);
         });
     }
 

src/core/tests/Deal.Tests.Unit/Contracts/PipelineWorkerGrpcAiTests.cs

@@ -4,6 +4,7 @@ using Deal.Grpc.Ai;
 using Deal.Infrastructure.Data;
 using Deal.Infrastructure.Integrations.Models;
 using Deal.Infrastructure.Integrations.Options;
+using Deal.Infrastructure.Integrations.Resilience;
 using Deal.Infrastructure.Integrations.Services;
 using Deal.Modules.Kanban.Application.Models;
 using Deal.Modules.Pipeline.Application.Models;
@@ -101,8 +102,9 @@ public sealed class PipelineWorkerGrpcAiTests
             Assert.Equal(1, result.AiFail);
             Assert.Equal(1, result.AiStored);
             Assert.Single(result.CreatedCards);
-            Assert.Equal(1, service.FilterCalls);
+            // Недоступность транспорта повторяется — на сервер приходит первая попытка и повторы.
-            Assert.Equal(1, service.ClassifyCalls);
+            Assert.Equal(GrpcRetry.RetryCount + 1, service.FilterCalls);
+            Assert.Equal(GrpcRetry.RetryCount + 1, service.ClassifyCalls);
         });
     }
 

src/core/tests/Deal.Tests.Unit/Support/GrpcRetryTests.cs

@@ -0,0 +1,93 @@
+using Deal.Infrastructure.Integrations.Resilience;
+using Grpc.Core;
+
+namespace Deal.Tests.Unit.Support;
+
+/// <summary>
+/// Тесты <see cref="GrpcRetry"/> — повтор транзиентных gRPC-сбоев.
+/// </summary>
+public sealed class GrpcRetryTests
+{
+    [Theory]
+    [InlineData(StatusCode.Unavailable)]
+    [InlineData(StatusCode.DeadlineExceeded)]
+    public void IsTransient_TransportFailures_True(StatusCode statusCode)
+    {
+        Assert.True(GrpcRetry.IsTransient(new RpcException(new Status(statusCode, "сбой"))));
+    }
+
+    [Theory]
+    [InlineData(StatusCode.NotFound)]
+    [InlineData(StatusCode.InvalidArgument)]
+    [InlineData(StatusCode.Internal)]
+    public void IsTransient_ApplicationFailures_False(StatusCode statusCode)
+    {
+        Assert.False(GrpcRetry.IsTransient(new RpcException(new Status(statusCode, "сбой"))));
+    }
+
+    [Fact]
+    public void IsTransient_NonRpcException_False()
+    {
+        Assert.False(GrpcRetry.IsTransient(new InvalidOperationException("сбой")));
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_TransientThenSuccess_Retries()
+    {
+        int calls = 0;
+
+        string result = await GrpcRetry.ExecuteAsync(
+            _ =>
+            {
+                calls++;
+                return calls < 2
+                    ? Task.FromException<string>(new RpcException(new Status(StatusCode.Unavailable, "down")))
+                    : Task.FromResult("ok");
+            },
+            InstantDelayAsync,
+            CancellationToken.None);
+
+        Assert.Equal("ok", result);
+        Assert.Equal(2, calls);
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_TransientExhausted_ThrowsRpcException()
+    {
+        int calls = 0;
+
+        RpcException thrown = await Assert.ThrowsAsync<RpcException>(
+            () => GrpcRetry.ExecuteAsync(
+                _ =>
+                {
+                    calls++;
+                    return Task.FromException<string>(new RpcException(new Status(StatusCode.Unavailable, "down")));
+                },
+                InstantDelayAsync,
+                CancellationToken.None));
+
+        Assert.Equal(StatusCode.Unavailable, thrown.StatusCode);
+        Assert.Equal(GrpcRetry.RetryCount + 1, calls);
+    }
+
+    [Fact]
+    public async Task ExecuteAsync_ApplicationFailure_NotRetried()
+    {
+        int calls = 0;
+
+        await Assert.ThrowsAsync<RpcException>(
+            () => GrpcRetry.ExecuteAsync(
+                _ =>
+                {
+                    calls++;
+                    return Task.FromException<string>(new RpcException(new Status(StatusCode.InvalidArgument, "bad")));
+                },
+                InstantDelayAsync,
+                CancellationToken.None));
+
+        Assert.Equal(1, calls);
+    }
+
+    private static Task InstantDelayAsync(TimeSpan delay, CancellationToken cancellationToken)
+        => Task.CompletedTask;
+}

src/core/tests/Deal.Tests.Unit/Support/RetryExecutorTests.cs

@@ -0,0 +1,158 @@
+using Deal.SharedKernel.Resilience;
+
+namespace Deal.Tests.Unit.Support;
+
+/// <summary>
+/// Тесты <see cref="RetryExecutor"/> — повтор транзиентных сбоев.
+/// </summary>
+public sealed class RetryExecutorTests
+{
+    private const int RetryCount = 2;
+    private static readonly TimeSpan BaseDelay = TimeSpan.FromMilliseconds(10);
+
+    [Fact]
+    public async Task FirstAttemptSucceeds_NoRetry()
+    {
+        var delays = new List<TimeSpan>();
+        int calls = 0;
+
+        string result = await ExecuteAsync(
+            () =>
+            {
+                calls++;
+                return Task.FromResult("ok");
+            },
+            shouldRetry: _ => true,
+            delays);
+
+        Assert.Equal("ok", result);
+        Assert.Equal(1, calls);
+        Assert.Empty(delays);
+    }
+
+    [Fact]
+    public async Task TransientFailureThenSuccess_RetriesUntilSuccess()
+    {
+        var delays = new List<TimeSpan>();
+        int calls = 0;
+
+        string result = await ExecuteAsync(
+            () =>
+            {
+                calls++;
+                return calls < 3
+                    ? throw new InvalidOperationException("транзиент")
+                    : Task.FromResult("ok");
+            },
+            shouldRetry: _ => true,
+            delays);
+
+        Assert.Equal("ok", result);
+        Assert.Equal(3, calls);
+        Assert.Equal(2, delays.Count);
+    }
+
+    [Fact]
+    public async Task RetriesExhausted_ThrowsLastFailure()
+    {
+        var delays = new List<TimeSpan>();
+        int calls = 0;
+
+        InvalidOperationException thrown = await Assert.ThrowsAsync<InvalidOperationException>(
+            () => ExecuteAsync<string>(
+                async () =>
+                {
+                    calls++;
+                    await Task.Yield();
+                    throw new InvalidOperationException($"сбой {calls}");
+                },
+                shouldRetry: _ => true,
+                delays));
+
+        Assert.Equal(3, calls); // первая попытка + 2 повтора
+        Assert.Equal("сбой 3", thrown.Message);
+    }
+
+    [Fact]
+    public async Task NonTransientFailure_NotRetried()
+    {
+        var delays = new List<TimeSpan>();
+        int calls = 0;
+
+        await Assert.ThrowsAsync<InvalidOperationException>(
+            () => ExecuteAsync<string>(
+                async () =>
+                {
+                    calls++;
+                    await Task.Yield();
+                    throw new InvalidOperationException("не транзиент");
+                },
+                shouldRetry: _ => false,
+                delays));
+
+        Assert.Equal(1, calls);
+        Assert.Empty(delays);
+    }
+
+    [Fact]
+    public async Task Cancellation_DoesNotRetry()
+    {
+        var delays = new List<TimeSpan>();
+        int calls = 0;
+        using var cts = new CancellationTokenSource();
+        cts.Cancel();
+
+        await Assert.ThrowsAsync<InvalidOperationException>(
+            () => ExecuteAsync<string>(
+                async () =>
+                {
+                    calls++;
+                    await Task.Yield();
+                    throw new InvalidOperationException("сбой");
+                },
+                shouldRetry: _ => true,
+                delays,
+                cts.Token));
+
+        Assert.Equal(1, calls);
+    }
+
+    [Fact]
+    public async Task Backoff_GrowsExponentially()
+    {
+        var delays = new List<TimeSpan>();
+        int calls = 0;
+
+        await ExecuteAsync(
+            () =>
+            {
+                calls++;
+                return calls < 3
+                    ? throw new InvalidOperationException("транзиент")
+                    : Task.FromResult(1);
+            },
+            shouldRetry: _ => true,
+            delays);
+
+        Assert.Equal(2, delays.Count);
+        Assert.Equal(BaseDelay, delays[0]);
+        Assert.Equal(BaseDelay * 2, delays[1]);
+    }
+
+    private static Task<TResult> ExecuteAsync<TResult>(
+        Func<Task<TResult>> operation,
+        Func<Exception, bool> shouldRetry,
+        List<TimeSpan> delays,
+        CancellationToken cancellationToken = default)
+        => RetryExecutor.ExecuteAsync(
+            _ => operation(),
+            RetryCount,
+            BaseDelay,
+            shouldRetry,
+            (delay, _) =>
+            {
+                delays.Add(delay);
+                return Task.CompletedTask;
+            },
+            cancellationToken);
+}