feat: Retry write operation on GRPC exception

src/BullOak.Repositories.EventStore.Test.Integration/BullOak.Repositories.EventStore.Test.Integration.csproj

@@ -24,6 +24,8 @@
     <PackageReference Include="OneOf" Version="3.0.243" />
     <PackageReference Include="Polly" Version="7.2.3" />
     <PackageReference Include="BullOak.Repositories" Version="3.0.0-alpha-003" />
+    <PackageReference Include="SpecFlow" Version="3.9.74" />
+    <PackageReference Include="SpecFlow.Tools.MsBuild.Generation" Version="3.9.74" />
     <PackageReference Include="SpecFlow.xUnit" Version="3.9.74" />
     <PackageReference Include="System.Collections.Immutable" Version="7.0.0" />
     <PackageReference Include="xunit.runner.console" Version="2.4.2">

src/BullOak.Repositories.EventStore/BullOak.Repositories.EventStore.csproj

@@ -50,5 +50,7 @@
     <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
     <PackageReference Include="Grpc.Net.Client" Version="2.53.0" />
     <PackageReference Include="OneOf" Version="3.0.243" />
+    <PackageReference Include="Polly" Version="7.2.3" />
+    <PackageReference Include="Polly.Contrib.WaitAndRetry" Version="1.1.1" />
   </ItemGroup>
 </Project>

src/BullOak.Repositories.EventStore/Streams/GrpcEventWriter.cs

@@ -1,17 +1,24 @@
 namespace BullOak.Repositories.EventStore.Streams;
 
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using Exceptions;
+using BullOak.Repositories.Exceptions;
 using global::EventStore.Client;
 using EventData = global::EventStore.Client.EventData;
+using Polly;
+using Polly.Retry;
 
 public class GrpcEventWriter : IStoreEventsToStream
 {
     private readonly EventStoreClient client;
 
+    // TODO: Get this via .ctor dependency injection
+    private readonly AsyncRetryPolicy retryPolicy =
+        GrpcWriteRetryPolicy.GetRetryPolicy(GrpcWriteRetrySettings.DefaultRetrySettings);
+
     public GrpcEventWriter(EventStoreClient client)
     {
         this.client = client;
@@ -55,29 +62,33 @@ private async Task<int> AppendToStream
         CancellationToken cancellationToken
     )
     {
-        IWriteResult writeResult;
+        var result = await retryPolicy.ExecuteAndCaptureAsync(async () =>
+            revision == -1 ?
+                await client.AppendToStreamAsync(
+                        streamId,
+                        StreamState.NoStream,
+                        eventsToAdd.Select(eventObject => eventObject.CreateV20EventData(dateTimeProvider)),
+                        options => options.ThrowOnAppendFailure = false,
+                        cancellationToken: cancellationToken)
+                    .ConfigureAwait(false)
+            :
+                await client.AppendToStreamAsync(
+                        streamId,
+                        StreamRevision.FromInt64(revision),
+                        eventsToAdd.Select(eventObject => eventObject.CreateV20EventData(dateTimeProvider)),
+                        options => options.ThrowOnAppendFailure = false,
+                        cancellationToken: cancellationToken)
+                    .ConfigureAwait(false)
+        );
 
-        if (revision == -1)
-        {
-            writeResult = await client.AppendToStreamAsync(
-                    streamId,
-                    StreamState.NoStream,
-                    eventsToAdd.Select(eventObject => eventObject.CreateV20EventData(dateTimeProvider)),
-                    options => options.ThrowOnAppendFailure = false,
-                    cancellationToken: cancellationToken)
-                .ConfigureAwait(false);
-        }
-        else
+        if (result.Outcome != OutcomeType.Successful)
         {
-            writeResult = await client.AppendToStreamAsync(
-                    streamId,
-                    StreamRevision.FromInt64(revision),
-                    eventsToAdd.Select(eventObject => eventObject.CreateV20EventData(dateTimeProvider)),
-                    options => options.ThrowOnAppendFailure = false,
-                    cancellationToken: cancellationToken)
-                .ConfigureAwait(false);
+            // TODO: Use more specific exception
+            throw new InvalidOperationException();
         }
 
+        var writeResult = result.Result;
+
         if (writeResult is WrongExpectedVersionResult)
             throw new ConcurrencyException(streamId, null);
 

src/BullOak.Repositories.EventStore/Streams/GrpcWriteRetryPolicy.cs

@@ -0,0 +1,17 @@
+using Grpc.Core;
+using Polly;
+using Polly.Contrib.WaitAndRetry;
+using Polly.Retry;
+
+namespace BullOak.Repositories.EventStore.Streams;
+
+public static class GrpcWriteRetryPolicy
+{
+    public static AsyncRetryPolicy GetRetryPolicy(GrpcWriteRetrySettings retrySettings) =>
+        Policy
+            .Handle<global::EventStore.Client.NotLeaderException>()
+            .Or<RpcException>(e => retrySettings.RetryableStatusCodes.Contains(e.StatusCode))
+            .WaitAndRetryAsync(
+                Backoff.DecorrelatedJitterBackoffV2(retrySettings.IntervalDelta, retrySettings.Limit, fastFirst: retrySettings.FastFirst),
+                (ex, timespan, context) => retrySettings.OnRetry(ex, timespan, context));
+}

src/BullOak.Repositories.EventStore/Streams/GrpcWriteRetrySettings.cs

@@ -0,0 +1,35 @@
+using System;
+using System.Collections.Generic;
+using Grpc.Core;
+
+namespace BullOak.Repositories.EventStore.Streams;
+
+public class GrpcWriteRetrySettings
+{
+    public TimeSpan IntervalDelta { get; }
+    public int Limit { get; }
+    public bool FastFirst { get; }
+
+    public HashSet<StatusCode> RetryableStatusCodes { get; }
+
+    public Action<Exception, TimeSpan, Polly.Context> OnRetry { get; init; }
+
+    private GrpcWriteRetrySettings(
+        TimeSpan intervalDelta,
+        int limit,
+        bool fastFirst = true,
+        HashSet<StatusCode> retryableStatusCodes = null,
+        Action<Exception, TimeSpan, Polly.Context> onRetry = null)
+    {
+        IntervalDelta = intervalDelta;
+        Limit = limit;
+        FastFirst = fastFirst;
+        RetryableStatusCodes = retryableStatusCodes ?? new HashSet<StatusCode>();
+        OnRetry = onRetry ?? NoOp;
+    }
+
+    // With default settings we will only retry on NotLeaderException
+    public static GrpcWriteRetrySettings DefaultRetrySettings = new(TimeSpan.FromMilliseconds(200), 5);
+
+    private static Action<Exception, TimeSpan, Polly.Context> NoOp => (_, _, _) => { };
+}