create parallel test for critical section conflict in InMemoryEventRepository
MODIFY
WalletWasabi.Tests/UnitTests/EventSourcing/InMemoryEventRepositoryTests.cs
WalletWasabi.Tests/UnitTests/EventSourcing/InMemoryEventRepositoryTests.cs
@@ -17,6 +17,7 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
public class InMemoryEventRepositoryTests
{
private const string TestRoundAggregate = "TestRoundAggregate";
+ private readonly TimeSpan _semaphoreWaitTimeout = TimeSpan.FromSeconds(1);
private IEventRepository EventRepository { get; init; }
private TestInMemoryEventRepository TestEventRepository { get; init; }
@@ -219,39 +220,159 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
.SequenceEqual(new[] { "1" }));
}
- [Fact]
- public async Task AppendEvents_AppendInParallelForManualThreadedDebug_Async()
+ [Theory]
+ [InlineData(nameof(TestInMemoryEventRepository.LockedSemaphore))]
+ [InlineData(nameof(TestInMemoryEventRepository.AppendedSemaphore))]
+ [InlineData(nameof(TestInMemoryEventRepository.UnlockedSemaphore))]
+ public async Task AppendEvents_CriticalSectionConflicts_Async(string conflictAfter)
+
{
// Arrange
var events_1 = new[] { new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a") };
var events_2 = new[] { new TestWrappedEvent(2, "b"), new TestWrappedEvent(3, "b") };
- TestEventRepository.AppendedSemaphoreToWait.Release(2);
- TestEventRepository.UnlockedSemaphoreToWait.Release(2);
// Act
- async Task Action1()
+ async Task Append1()
{
await EventRepository.AppendEventsAsync(nameof(TestRoundAggregate), "1", events_1!);
}
- async Task Action2()
+ async Task Append2()
{
+ switch (conflictAfter)
+ {
+ case nameof(TestInMemoryEventRepository.LockedSemaphore):
+ Assert.True(TestEventRepository.LockedSemaphore.Wait(_semaphoreWaitTimeout));
+ break;
+
+ case nameof(TestInMemoryEventRepository.AppendedSemaphore):
+ Assert.True(TestEventRepository.AppendedSemaphore.Wait(_semaphoreWaitTimeout));
+ break;
+
+ case nameof(TestInMemoryEventRepository.UnlockedSemaphore):
+ Assert.True(TestEventRepository.UnlockedSemaphore.Wait(_semaphoreWaitTimeout));
+ break;
+
+ default:
+ throw new ApplicationException($"unexpected value conflictAfter: '{conflictAfter}'");
+ }
await EventRepository.AppendEventsAsync(nameof(TestRoundAggregate), "1", events_2!);
}
+ async Task AppendInParallel()
+ {
+ var task1 = Task.Run(Append1);
+ var task2 = Task.Run(Append2);
+ await Task.WhenAll(task1, task2);
+ }
+ void WaitForConflict()
+ {
+ Assert.True(TestEventRepository.ConflictedSemaphore.Wait(_semaphoreWaitTimeout));
+ }
+ switch (conflictAfter)
+ {
+ case nameof(TestInMemoryEventRepository.LockedSemaphore):
+ TestEventRepository.LockedCallback = WaitForConflict;
+ break;
+
+ case nameof(TestInMemoryEventRepository.AppendedSemaphore):
+ TestEventRepository.AppendedCallback = WaitForConflict;
+ break;
+
+ case nameof(TestInMemoryEventRepository.UnlockedSemaphore):
+ TestEventRepository.UnlockedCallback = WaitForConflict;
+ break;
+
+ default:
+ throw new ApplicationException($"unexpected value conflictAfter: '{conflictAfter}'");
+ }
// Assert
- await Assert.ThrowsAsync<OptimisticConcurrencyException>(() =>
+ await Assert.ThrowsAsync<OptimisticConcurrencyException>(AppendInParallel);
+ Assert.True((await EventRepository.ListEventsAsync(nameof(TestRoundAggregate), "1"))
+ .Cast<TestWrappedEvent>().SequenceEqual(events_1));
+ Assert.True((await EventRepository.ListAggregateIdsAsync(nameof(TestRoundAggregate)))
+ .SequenceEqual(new[] { "1" }));
+ }
+
+ [Theory]
+ [InlineData(nameof(TestInMemoryEventRepository.LockedSemaphore))]
+ [InlineData(nameof(TestInMemoryEventRepository.AppendedSemaphore))]
+ [InlineData(nameof(TestInMemoryEventRepository.UnlockedSemaphore))]
+ public async Task AppendEvents_CriticalAppendConflicts_Async(string conflictAfter)
{
- var task1 = Task.Run(Action1);
- var task2 = Task.Run(async () =>
+ // Arrange
+ var events_1 = new[] { new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a") };
+ var events_2 = new[] { new TestWrappedEvent(3, "b"), new TestWrappedEvent(4, "b") };
+
+ // Act
+ async Task Append1()
{
- // wait until task1 enters critical section so we have a conflict
- await TestEventRepository.LockedSemaphoreToRelease.WaitAsync();
- await Action2();
- });
- return Task.WhenAll(task1, task2);
- });
+ await EventRepository.AppendEventsAsync(nameof(TestRoundAggregate), "1", events_1!);
+ }
+ async Task Append2()
+ {
+ switch (conflictAfter)
+ {
+ case nameof(TestInMemoryEventRepository.LockedSemaphore):
+ Assert.True(TestEventRepository.LockedSemaphore.Wait(_semaphoreWaitTimeout));
+ break;
+
+ case nameof(TestInMemoryEventRepository.AppendedSemaphore):
+ Assert.True(TestEventRepository.AppendedSemaphore.Wait(_semaphoreWaitTimeout));
+ break;
+
+ case nameof(TestInMemoryEventRepository.UnlockedSemaphore):
+ Assert.True(TestEventRepository.UnlockedSemaphore.Wait(_semaphoreWaitTimeout));
+ break;
+
+ default:
+ throw new ApplicationException($"unexpected value conflictAfter: '{conflictAfter}'");
+ }
+ await EventRepository.AppendEventsAsync(nameof(TestRoundAggregate), "1", events_2!);
+ }
+ async Task AppendInParallel()
+ {
+ var task1 = Task.Run(Append1);
+ var task2 = Task.Run(Append2);
+ await Task.WhenAll(task1, task2);
+ }
+ void WaitForConflict()
+ {
+ Assert.True(TestEventRepository.ConflictedSemaphore.Wait(_semaphoreWaitTimeout));
+ }
+ switch (conflictAfter)
+ {
+ case nameof(TestInMemoryEventRepository.LockedSemaphore):
+ TestEventRepository.LockedCallback = WaitForConflict;
+ break;
+
+ case nameof(TestInMemoryEventRepository.AppendedSemaphore):
+ TestEventRepository.AppendedCallback = WaitForConflict;
+ break;
+
+ case nameof(TestInMemoryEventRepository.UnlockedSemaphore):
+ // do nothing
+ break;
+
+ default:
+ throw new ApplicationException($"unexpected value conflictAfter: '{conflictAfter}'");
+ }
+
+ // Assert
+ if (conflictAfter == nameof(TestInMemoryEventRepository.UnlockedSemaphore))
+ { // no conflict
+ await AppendInParallel();
+ }
+ else
+ { // conflict
+ await Assert.ThrowsAsync<OptimisticConcurrencyException>(AppendInParallel);
+ // retry after optimistic concurrency conflict
+ TestEventRepository.LockedCallback = null;
+ TestEventRepository.AppendedCallback = null;
+ TestEventRepository.UnlockedCallback = null;
+ await EventRepository.AppendEventsAsync(nameof(TestRoundAggregate), "1", events_2!);
+ }
Assert.True((await EventRepository.ListEventsAsync(nameof(TestRoundAggregate), "1"))
- .Cast<TestWrappedEvent>().SequenceEqual(events_1));
+ .Cast<TestWrappedEvent>().SequenceEqual(events_1.Concat(events_2)));
Assert.True((await EventRepository.ListAggregateIdsAsync(nameof(TestRoundAggregate)))
.SequenceEqual(new[] { "1" }));
}

MODIFY
WalletWasabi.Tests/UnitTests/EventSourcing/TestDomain/TestInMemoryEventRepository.cs
WalletWasabi.Tests/UnitTests/EventSourcing/TestDomain/TestInMemoryEventRepository.cs
@@ -13,41 +13,61 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing.TestDomain
{
protected ITestOutputHelper Output { get; init; }
- public SemaphoreSlim LockedSemaphoreToRelease { get; } = new SemaphoreSlim(0);
- public SemaphoreSlim AppendedSemaphoreToRelease { get; } = new SemaphoreSlim(0);
- public SemaphoreSlim UnlockedSemaphoreToRelease { get; } = new SemaphoreSlim(0);
+ public SemaphoreSlim ValidatedSemaphore { get; } = new SemaphoreSlim(0);
+ public SemaphoreSlim ConflictedSemaphore { get; } = new SemaphoreSlim(0);
+ public SemaphoreSlim LockedSemaphore { get; } = new SemaphoreSlim(0);
+ public SemaphoreSlim AppendedSemaphore { get; } = new SemaphoreSlim(0);
+ public SemaphoreSlim UnlockedSemaphore { get; } = new SemaphoreSlim(0);
- public SemaphoreSlim LockedSemaphoreToWait { get; } = new SemaphoreSlim(0);
- public SemaphoreSlim AppendedSemaphoreToWait { get; } = new SemaphoreSlim(0);
- public SemaphoreSlim UnlockedSemaphoreToWait { get; } = new SemaphoreSlim(0);
+ public Action? ValidatedCallback { get; set; }
+ public Action? ConflictedCallback { get; set; }
+ public Action? LockedCallback { get; set; }
+ public Action? AppendedCallback { get; set; }
+ public Action? UnlockedCallback { get; set; }
public TestInMemoryEventRepository(ITestOutputHelper output)
{
Output = output;
}
+ protected override void Validated()
+ {
+ base.Validated();
+ Output.WriteLine(nameof(Validated));
+ ValidatedSemaphore.Release();
+ ValidatedCallback?.Invoke();
+ }
+
+ protected override void Conflicted()
+ {
+ base.Conflicted();
+ Output.WriteLine(nameof(Conflicted));
+ ConflictedSemaphore.Release();
+ ConflictedCallback?.Invoke();
+ }
+
protected override void Locked()
{
base.Locked();
- Output.WriteLine("Locked");
- LockedSemaphoreToRelease.Release();
- LockedSemaphoreToWait.Wait();
+ Output.WriteLine(nameof(Locked));
+ LockedSemaphore.Release();
+ LockedCallback?.Invoke();
}
protected override void Appended()
{
base.Appended();
- Output.WriteLine("Appended");
- AppendedSemaphoreToRelease.Release();
- AppendedSemaphoreToWait.Wait();
+ Output.WriteLine(nameof(Appended));
+ AppendedSemaphore.Release();
+ AppendedCallback?.Invoke();
}
protected override void Unlocked()
{
base.Unlocked();
- Output.WriteLine("Unlocked");
- UnlockedSemaphoreToRelease.Release();
- UnlockedSemaphoreToWait.Wait();
+ Output.WriteLine(nameof(Unlocked));
+ UnlockedSemaphore.Release();
+ UnlockedCallback?.Invoke();
}
}
}

MODIFY
WalletWasabi/EventSourcing/InMemoryEventRepository.cs
WalletWasabi/EventSourcing/InMemoryEventRepository.cs
@@ -71,16 +71,18 @@ namespace WalletWasabi.EventSourcing
{
throw new ArgumentException($"invalid firstSequenceId (gap in sequence ids) expected: '{concurrentSequenceId + 1}' given: '{firstSequenceId}'", nameof(wrappedEvents));
}
+ Validated(); // no action
// atomically detect conflict and replace lastSequenceId and lock to ensure strong order in eventsBatches
if (!aggregateEventsBatches.TryUpdate(id, (lastSequenceId, true, eventsBatches), (firstSequenceId - 1, false, eventsBatches)))
{
+ Conflicted(); // no action
throw new OptimisticConcurrencyException($"Conflict while commiting events. Retry command. aggregate: '{aggregateType}' id: '{id}'");
}
try
{
- Locked();
+ Locked(); // no action
eventsBatches.Enqueue(wrappedEventsList);
- Appended();
+ Appended(); // no action
}
finally
{
@@ -91,7 +93,7 @@ namespace WalletWasabi.EventSourcing
// TODO: convert into Debug.Assert ???
throw new ApplicationException("unexpected failure 89#");
}
- Unlocked();
+ Unlocked(); // no action
}
// if it is a first event for given aggregate
@@ -172,6 +174,18 @@ namespace WalletWasabi.EventSourcing
}
}
+ // for parallel critical section testing
+ [Conditional("DEBUG")]
+ protected virtual void Validated()
+ {
+ }
+
+ // for parallel critical section testing
+ [Conditional("DEBUG")]
+ protected virtual void Conflicted()
+ {
+ }
+
// for parallel critical section testing
[Conditional("DEBUG")]
protected virtual void Locked()

create parallel test for critical section conflict in InMemoryEventRepository