[DEVEX-227] Added retry policy handling for optimistic concurrency of decision making

Author: oskardudycz

src/Kurrent.Client/Streams/DecisionMaking/Decider.cs

@@ -1,9 +1,13 @@
 using EventStore.Client;
 using Kurrent.Client.Core.Serialization;
 using Kurrent.Client.Streams.GettingState;
+using Polly;
+using Polly.Retry;
 
 namespace Kurrent.Client.Streams.DecisionMaking;
 
+using static AsyncDecider;
+
 public delegate ValueTask<Message[]> CommandHandler<in TState>(TState state, CancellationToken ct = default);
 
 public record AsyncDecider<TState, TCommand>(
@@ -56,43 +60,51 @@ resolvedEvent.DeserializedData is TEvent @event
 public class DecideOptions<TState> where TState : notnull {
 	public GetStreamStateOptions<TState>? GetStateOptions       { get; set; }
 	public AppendToStreamOptions?         AppendToStreamOptions { get; set; }
+	public IAsyncPolicy<IWriteResult>?    RetryPolicy           { get; set; }
 }
 
 public static class KurrentClientDecisionMakingExtensions {
-	public static async Task<IWriteResult> DecideAsync<TState>(
+	public static Task<IWriteResult> DecideAsync<TState>(
 		this KurrentClient eventStore,
 		string streamName,
 		CommandHandler<TState> decide,
 		IStateBuilder<TState> stateBuilder,
 		DecideOptions<TState>? options,
-		CancellationToken ct = default
-	) where TState : notnull {
-		var (state, streamPosition, position) =
-			await eventStore.GetStateAsync(streamName, stateBuilder, options?.GetStateOptions, ct);
-
-		var events = await decide(state, ct);
-
-		if (events.Length == 0) {
-			return new SuccessResult(
-				streamPosition.HasValue ? StreamRevision.FromStreamPosition(streamPosition.Value) : StreamRevision.None,
-				position ?? Position.Start
-			);
-		}
-
-		var appendToStreamOptions = options?.AppendToStreamOptions ?? new AppendToStreamOptions();
-
-		if (streamPosition.HasValue)
-			appendToStreamOptions.ExpectedStreamRevision = StreamRevision.FromStreamPosition(streamPosition.Value);
-		else
-			appendToStreamOptions.ExpectedStreamState = StreamState.NoStream;
-
-		return await eventStore.AppendToStreamAsync(
-			streamName,
-			events.Cast<object>(),
-			appendToStreamOptions,
-			cancellationToken: ct
+		CancellationToken cancellationToken = default
+	) where TState : notnull =>
+		DecideRetryPolicy(options).ExecuteAsync(
+			async ct => {
+				var (state, streamPosition, position) =
+					await eventStore.GetStateAsync(streamName, stateBuilder, options?.GetStateOptions, ct);
+
+				var messages = await decide(state, ct);
+
+				if (messages.Length == 0) {
+					return new SuccessResult(
+						streamPosition.HasValue
+							? StreamRevision.FromStreamPosition(streamPosition.Value)
+							: StreamRevision.None,
+						position ?? Position.Start
+					);
+				}
+
+				var appendToStreamOptions = options?.AppendToStreamOptions ?? new AppendToStreamOptions();
+
+				if (streamPosition.HasValue)
+					appendToStreamOptions.ExpectedStreamRevision ??=
+						StreamRevision.FromStreamPosition(streamPosition.Value);
+				else
+					appendToStreamOptions.ExpectedStreamState ??= StreamState.NoStream;
+
+				return await eventStore.AppendToStreamAsync(
+					streamName,
+					messages,
+					appendToStreamOptions,
+					ct
+				);
+			},
+			cancellationToken
 		);
-	}
 
 	public static Task<IWriteResult> DecideAsync<TState, TCommand>(
 		this KurrentClient eventStore,
@@ -183,3 +195,24 @@ public static Task<IWriteResult> DecideAsync<TState, TEvent>(
 			ct
 		);
 }
+
+public static class AsyncDecider {
+	public static readonly IAsyncPolicy<IWriteResult> DefaultRetryPolicy =
+		Policy<IWriteResult>
+			.Handle<WrongExpectedVersionException>()
+			.WaitAndRetryAsync(
+				retryCount: 3,
+				sleepDurationProvider: retryAttempt => TimeSpan.FromMilliseconds(20 * retryAttempt)
+			);
+
+	public static bool HasUserProvidedExpectedVersioning(AppendToStreamOptions? options) =>
+		options != null && (options.ExpectedStreamState.HasValue || options.ExpectedStreamRevision.HasValue);
+
+	public static IAsyncPolicy<IWriteResult> DecideRetryPolicy<TState>(DecideOptions<TState>? options)
+		where TState : notnull =>
+		options?.RetryPolicy ??
+		(HasUserProvidedExpectedVersioning(options?.AppendToStreamOptions)
+			// it doesn't make sense to retry, as expected state will be always the same
+			? Policy.NoOpAsync<IWriteResult>() 
+			: DefaultRetryPolicy);
+}