Add Polly resilience policies

src/MxGateway.Worker/Ipc/WorkerPipeClient.cs

@@ -1,17 +1,24 @@
 using System;
+using System.Collections.Generic;
 using System.IO;
 using System.IO.Pipes;
 using System.Threading;
 using System.Threading.Tasks;
 using MxGateway.Worker.Bootstrap;
+using Polly;
+using Polly.Retry;
 
 namespace MxGateway.Worker.Ipc;
 
 public sealed class WorkerPipeClient : IWorkerPipeClient
 {
     public const int DefaultConnectTimeoutMilliseconds = 30000;
+    public const int DefaultConnectAttemptTimeoutMilliseconds = 2000;
+    public const string ConnectAttemptTimeoutEnvironmentVariableName =
+        "MXGATEWAY_WORKER_PIPE_CONNECT_ATTEMPT_TIMEOUT_MS";
 
     private readonly int _connectTimeoutMilliseconds;
+    private readonly int _connectAttemptTimeoutMilliseconds;
     private readonly Func<Stream, WorkerFrameProtocolOptions, IWorkerLogger?, WorkerPipeSession> _sessionFactory;
     private readonly IWorkerLogger? _logger;
 
@@ -36,6 +43,7 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
         : this(
             null,
             connectTimeoutMilliseconds,
+            ResolveDefaultConnectAttemptTimeoutMilliseconds(),
             (stream, frameOptions, _) => sessionFactory(stream, frameOptions))
     {
     }
@@ -46,6 +54,7 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
         : this(
             logger,
             connectTimeoutMilliseconds,
+            ResolveDefaultConnectAttemptTimeoutMilliseconds(),
             (stream, frameOptions, workerLogger) => new WorkerPipeSession(stream, frameOptions, workerLogger))
     {
     }
@@ -54,6 +63,19 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
         IWorkerLogger? logger,
         int connectTimeoutMilliseconds,
         Func<Stream, WorkerFrameProtocolOptions, IWorkerLogger?, WorkerPipeSession> sessionFactory)
+        : this(
+            logger,
+            connectTimeoutMilliseconds,
+            ResolveDefaultConnectAttemptTimeoutMilliseconds(),
+            sessionFactory)
+    {
+    }
+
+    public WorkerPipeClient(
+        IWorkerLogger? logger,
+        int connectTimeoutMilliseconds,
+        int connectAttemptTimeoutMilliseconds,
+        Func<Stream, WorkerFrameProtocolOptions, IWorkerLogger?, WorkerPipeSession> sessionFactory)
     {
         if (connectTimeoutMilliseconds <= 0)
         {
@@ -62,9 +84,17 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
                 "Worker pipe connect timeout must be greater than zero.");
         }
 
+        if (connectAttemptTimeoutMilliseconds <= 0)
+        {
+            throw new ArgumentOutOfRangeException(
+                nameof(connectAttemptTimeoutMilliseconds),
+                "Worker pipe connect attempt timeout must be greater than zero.");
+        }
+
         _logger = logger;
         _sessionFactory = sessionFactory ?? throw new ArgumentNullException(nameof(sessionFactory));
         _connectTimeoutMilliseconds = connectTimeoutMilliseconds;
+        _connectAttemptTimeoutMilliseconds = connectAttemptTimeoutMilliseconds;
     }
 
     public async Task RunAsync(
@@ -78,28 +108,91 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
 
         WorkerFrameProtocolOptions frameOptions = new(options);
 
-        using NamedPipeClientStream pipe = new(
-            ".",
-            options.PipeName,
-            PipeDirection.InOut,
-            PipeOptions.Asynchronous);
-
-        await ConnectAsync(pipe, cancellationToken).ConfigureAwait(false);
+        using NamedPipeClientStream pipe = await ConnectWithRetryAsync(options.PipeName, cancellationToken)
+            .ConfigureAwait(false);
 
         WorkerPipeSession session = _sessionFactory(pipe, frameOptions, _logger);
         await session.RunAsync(cancellationToken).ConfigureAwait(false);
     }
 
-    private Task ConnectAsync(
-        NamedPipeClientStream pipe,
+    private async Task<NamedPipeClientStream> ConnectWithRetryAsync(
+        string pipeName,
         CancellationToken cancellationToken)
     {
-        return Task.Run(
-            () =>
+        int retryAttempts = Math.Max(
+            0,
+            (_connectTimeoutMilliseconds / Math.Min(_connectTimeoutMilliseconds, _connectAttemptTimeoutMilliseconds)) - 1);
+
+        ResiliencePipeline<NamedPipeClientStream> pipeline = new ResiliencePipelineBuilder<NamedPipeClientStream>()
+            .AddRetry(new RetryStrategyOptions<NamedPipeClientStream>
             {
-                cancellationToken.ThrowIfCancellationRequested();
-                pipe.Connect(_connectTimeoutMilliseconds);
-            },
-            cancellationToken);
+                MaxRetryAttempts = retryAttempts,
+                BackoffType = DelayBackoffType.Exponential,
+                UseJitter = true,
+                Delay = TimeSpan.FromMilliseconds(250),
+                MaxDelay = TimeSpan.FromSeconds(2),
+                ShouldHandle = new PredicateBuilder<NamedPipeClientStream>()
+                    .Handle<Exception>(exception => exception is TimeoutException or IOException),
+                OnRetry = args =>
+                {
+                    args.Outcome.Result?.Dispose();
+                    _logger?.Information(
+                        "WorkerPipeConnectRetry",
+                        new Dictionary<string, object?>
+                        {
+                            ["attempt"] = args.AttemptNumber + 1,
+                            ["pipe_name"] = pipeName,
+                        });
+                    return default;
+                },
+            })
+            .Build();
+
+        return await pipeline.ExecuteAsync(
+                async token => await ConnectSingleAttemptAsync(pipeName, token).ConfigureAwait(false),
+                cancellationToken)
+            .ConfigureAwait(false);
+    }
+
+    private async Task<NamedPipeClientStream> ConnectSingleAttemptAsync(
+        string pipeName,
+        CancellationToken cancellationToken)
+    {
+        NamedPipeClientStream pipe = new(
+            ".",
+            pipeName,
+            PipeDirection.InOut,
+            PipeOptions.Asynchronous);
+
+        try
+        {
+            using CancellationTokenSource attemptTimeout =
+                CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+            attemptTimeout.CancelAfter(_connectAttemptTimeoutMilliseconds);
+
+            await Task.Run(
+                    () =>
+                    {
+                        attemptTimeout.Token.ThrowIfCancellationRequested();
+                        pipe.Connect(_connectAttemptTimeoutMilliseconds);
+                    },
+                    attemptTimeout.Token)
+                .ConfigureAwait(false);
+
+            return pipe;
+        }
+        catch
+        {
+            pipe.Dispose();
+            throw;
+        }
+    }
+
+    private static int ResolveDefaultConnectAttemptTimeoutMilliseconds()
+    {
+        string? configuredValue = Environment.GetEnvironmentVariable(ConnectAttemptTimeoutEnvironmentVariableName);
+        return int.TryParse(configuredValue, out int milliseconds) && milliseconds > 0
+            ? milliseconds
+            : DefaultConnectAttemptTimeoutMilliseconds;
     }
 }

src/MxGateway.Worker/MxGateway.Worker.csproj

@@ -10,6 +10,10 @@
     <GenerateBindingRedirectsOutputType>true</GenerateBindingRedirectsOutputType>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Polly.Core" Version="8.6.6" />
+  </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\MxGateway.Contracts\MxGateway.Contracts.csproj" />
   </ItemGroup>