Use Immutable list
MODIFY
WalletWasabi.Tests/UnitTests/EventSourcing/InMemoryEventRepositoryTests.cs
WalletWasabi.Tests/UnitTests/EventSourcing/InMemoryEventRepositoryTests.cs
@@ -221,9 +221,7 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
}
[Theory]
- [InlineData(nameof(TestInMemoryEventRepository.LockedSemaphore))]
[InlineData(nameof(TestInMemoryEventRepository.AppendedSemaphore))]
- [InlineData(nameof(TestInMemoryEventRepository.UnlockedSemaphore))]
public async Task AppendEvents_CriticalSectionConflicts_Async(string conflictAfter)
{
// Arrange
@@ -239,18 +237,10 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
{
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}'");
}
@@ -266,23 +256,11 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
{
Assert.True(TestEventRepository.ConflictedSemaphore.Wait(_semaphoreWaitTimeout));
}
- switch (conflictAfter)
+ TestEventRepository.AppendedCallback = conflictAfter switch
{
- 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}'");
- }
+ nameof(TestInMemoryEventRepository.AppendedSemaphore) => WaitForConflict,
+ _ => throw new ApplicationException($"unexpected value conflictAfter: '{conflictAfter}'"),
+ };
// Assert
await Assert.ThrowsAsync<OptimisticConcurrencyException>(AppendInParallel);
@@ -293,9 +271,7 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
}
[Theory]
- [InlineData(nameof(TestInMemoryEventRepository.LockedSemaphore))]
[InlineData(nameof(TestInMemoryEventRepository.AppendedSemaphore))]
- [InlineData(nameof(TestInMemoryEventRepository.UnlockedSemaphore))]
public async Task AppendEvents_CriticalAppendConflicts_Async(string conflictAfter)
{
// Arrange
@@ -311,18 +287,10 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
{
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}'");
}
@@ -334,43 +302,19 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
var task2 = Task.Run(Append2);
await Task.WhenAll(task1, task2);
}
- void WaitForConflict()
+ void WaitForNoConflict()
{
- Assert.True(TestEventRepository.ConflictedSemaphore.Wait(_semaphoreWaitTimeout));
+ Assert.False(TestEventRepository.ConflictedSemaphore.Wait(_semaphoreWaitTimeout));
}
- switch (conflictAfter)
+ TestEventRepository.AppendedCallback = conflictAfter switch
{
- 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}'");
- }
+ nameof(TestInMemoryEventRepository.AppendedSemaphore) => WaitForNoConflict,
+ _ => throw new ApplicationException($"unexpected value conflictAfter: '{conflictAfter}'"),
+ };
- // Assert
- if (conflictAfter == nameof(TestInMemoryEventRepository.UnlockedSemaphore))
- { // no conflict
+ // 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.Concat(events_2)));
Assert.True((await EventRepository.ListAggregateIdsAsync(nameof(TestRoundAggregate)))
@@ -379,9 +323,7 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
[Theory]
[InlineData(nameof(TestInMemoryEventRepository.ValidatedCallback))]
- [InlineData(nameof(TestInMemoryEventRepository.LockedCallback))]
[InlineData(nameof(TestInMemoryEventRepository.AppendedCallback))]
- [InlineData(nameof(TestInMemoryEventRepository.UnlockedCallback))]
public async Task ListEventsAsync_ConflictWithAppending_Async(string listOnCallback)
{
// Arrange
@@ -401,18 +343,10 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
TestEventRepository.ValidatedCallback = ListCallback;
break;
- case nameof(TestInMemoryEventRepository.LockedCallback):
- TestEventRepository.LockedCallback = ListCallback;
- break;
-
case nameof(TestInMemoryEventRepository.AppendedCallback):
TestEventRepository.AppendedCallback = ListCallback;
break;
- case nameof(TestInMemoryEventRepository.UnlockedCallback):
- TestEventRepository.UnlockedCallback = ListCallback;
- break;
-
default:
throw new ApplicationException($"unexpected value listOnCallback: '{listOnCallback}'");
}
@@ -423,7 +357,6 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
switch (listOnCallback)
{
case nameof(TestInMemoryEventRepository.AppendedCallback):
- case nameof(TestInMemoryEventRepository.UnlockedCallback):
expected = expected.Concat(events_2);
break;
}
@@ -483,5 +416,37 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
// Assert
Assert.True(result.SequenceEqual(new[] { "1", "2" }));
}
+
+ [Theory]
+ [InlineData("0", 0)]
+ [InlineData("0", 1)]
+ [InlineData("0", 2)]
+ [InlineData("0", 3)]
+ [InlineData("1", 0)]
+ [InlineData("1", 1)]
+ [InlineData("1", 2)]
+ [InlineData("2", 0)]
+ [InlineData("2", 1)]
+ [InlineData("3", 0)]
+ [InlineData("3", 1)]
+ public async Task ListAggregateIdsAsync_OptionalArguments_Async(string afterAggregateId, int limit)
+ {
+ // Arrange
+ var events = new[]
+ {
+ new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a"),
+ new TestWrappedEvent(3, "b"), new TestWrappedEvent(4, "b")
+ };
+ await EventRepository.AppendEventsAsync(nameof(TestRoundAggregate), "1", events);
+ await EventRepository.AppendEventsAsync(nameof(TestRoundAggregate), "2", events);
+
+ // Act
+ var result = await EventRepository.ListAggregateIdsAsync(
+ nameof(TestRoundAggregate), afterAggregateId, limit);
+
+ // Assert
+ Assert.True(result.Count <= limit);
+ Assert.True(result.All(a => afterAggregateId.CompareTo(a) <= 0));
+ }
}
}

MODIFY
WalletWasabi.Tests/UnitTests/EventSourcing/TestDomain/TestInMemoryEventRepository.cs
WalletWasabi.Tests/UnitTests/EventSourcing/TestDomain/TestInMemoryEventRepository.cs
@@ -20,15 +20,11 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing.TestDomain
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 Action? ValidatedCallback { get; set; }
public Action? ConflictedCallback { get; set; }
- public Action? LockedCallback { get; set; }
public Action? AppendedCallback { get; set; }
- public Action? UnlockedCallback { get; set; }
protected override void Validated()
{
@@ -46,14 +42,6 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing.TestDomain
ConflictedCallback?.Invoke();
}
- protected override void Locked()
- {
- base.Locked();
- Output.WriteLine(nameof(Locked));
- LockedSemaphore.Release();
- LockedCallback?.Invoke();
- }
-
protected override void Appended()
{
base.Appended();
@@ -61,13 +49,5 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing.TestDomain
AppendedSemaphore.Release();
AppendedCallback?.Invoke();
}
-
- protected override void Unlocked()
- {
- base.Unlocked();
- Output.WriteLine(nameof(Unlocked));
- UnlockedSemaphore.Release();
- UnlockedCallback?.Invoke();
- }
}
}

MODIFY
WalletWasabi/EventSourcing/InMemoryEventRepository.cs
WalletWasabi/EventSourcing/InMemoryEventRepository.cs
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
+using System.Collections.Immutable;
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;
@@ -15,8 +16,8 @@ namespace WalletWasabi.EventSourcing
/// </summary>
public class InMemoryEventRepository : IEventRepository
{
- protected static readonly IReadOnlyList<WrappedEvent> EmptyResult = Array.Empty<WrappedEvent>().ToList().AsReadOnly();
- protected static readonly IReadOnlyList<string> EmptyIds = Array.Empty<string>().ToList().AsReadOnly();
+ protected static readonly IReadOnlyList<WrappedEvent> EmptyResult = ImmutableList<WrappedEvent>.Empty;
+ protected static readonly IReadOnlyList<string> EmptyIds = ImmutableList<string>.Empty;
private readonly ConcurrentDictionary<
// aggregateType
@@ -28,12 +29,8 @@ namespace WalletWasabi.EventSourcing
// SequenceId of the last event of this aggregate
long TailSequenceId,
- // Locked flag for appending into EventsBatches of this aggregate
- WriteLockEnum LockState,
-
- // List of lists of events for atomic insertion of multiple events
- // in one "database transaction"
- ConcurrentQueue<IReadOnlyList<WrappedEvent>> EventsBatches
+ // Ordered list of events
+ ImmutableList<WrappedEvent> Events
)
>
> _aggregatesEventsBatches = new();
@@ -44,20 +41,12 @@ namespace WalletWasabi.EventSourcing
(
// Index of the last aggregateId in this aggregateType
long TailIndex,
- ConcurrentDictionary<
- // Index of aggregateId in this aggregateType
- long,
- // aggregateId
- string> Ids
+
+ // List of aggregate Ids in this aggregateType
+ ImmutableSortedSet<string> Ids
)
> _aggregatesIds = new();
- protected enum WriteLockEnum
- {
- Unlocked,
- WritingLocked,
- }
-
public Task AppendEventsAsync(
string aggregateType,
string aggregateId,
@@ -87,7 +76,7 @@ namespace WalletWasabi.EventSourcing
}
var aggregateEventsBatches = _aggregatesEventsBatches.GetOrAdd(aggregateType, _ => new());
- var (tailSequenceId, locked, eventsBatches) = aggregateEventsBatches.GetOrAdd(aggregateId, _ => (0, WriteLockEnum.Unlocked, new()));
+ var (tailSequenceId, events) = aggregateEventsBatches.GetOrAdd(aggregateId, _ => (0, ImmutableList<WrappedEvent>.Empty));
if (tailSequenceId + 1 < firstSequenceId)
{
@@ -97,33 +86,18 @@ namespace WalletWasabi.EventSourcing
// no action
Validated();
+ var newEvents = events.AddRange(wrappedEventsList);
+
// Atomically detect conflict and replace lastSequenceId and lock to ensure strong order in eventsBatches.
if (!aggregateEventsBatches.TryUpdate(
key: aggregateId,
- newValue: (lastSequenceId, WriteLockEnum.WritingLocked, eventsBatches),
- comparisonValue: (firstSequenceId - 1, WriteLockEnum.Unlocked, eventsBatches)))
+ newValue: (lastSequenceId, newEvents),
+ comparisonValue: (firstSequenceId - 1, events)))
{
Conflicted(); // no action
throw new OptimisticConcurrencyException($"Conflict while commiting events. Retry command. aggregate: '{aggregateType}' id: '{aggregateId}'");
}
- try
- {
- Locked(); // no action
- eventsBatches.Enqueue(wrappedEventsList);
Appended(); // no action
- }
- finally
- {
- // Unlock.
- if (!aggregateEventsBatches.TryUpdate(
- key: aggregateId,
- newValue: (lastSequenceId, WriteLockEnum.Unlocked, eventsBatches),
- comparisonValue: (lastSequenceId, WriteLockEnum.WritingLocked, eventsBatches)))
- {
- throw new AssertionFailedException("Unexpected failure to unlock.");
- }
- Unlocked(); // no action
- }
// If it is a first event for given aggregate.
if (tailSequenceId == 0)
@@ -142,16 +116,27 @@ namespace WalletWasabi.EventSourcing
if (_aggregatesEventsBatches.TryGetValue(aggregateType, out var aggregateEventsBatches) &&
aggregateEventsBatches.TryGetValue(aggregateId, out var value))
{
- var result = value.EventsBatches.SelectMany(a => a);
+ var result = value.Events;
if (afterSequenceId > 0)
{
- result = result.Where(a => afterSequenceId < a.SequenceId);
+ var foundIndex = result.BinarySearch(
+ new WrappedEvent(afterSequenceId),
+ Comparer<WrappedEvent>.Create((a, b) => a.SequenceId.CompareTo(b.SequenceId)));
+ if (foundIndex < 0)
+ {
+ foundIndex = ~foundIndex;
}
- if (limit.HasValue)
+ else
{
- result = result.Take(limit.Value);
+ foundIndex++;
+ }
+ result = result.RemoveRange(0, foundIndex);
}
- return Task.FromResult((IReadOnlyList<WrappedEvent>)result.ToList().AsReadOnly());
+ if (limit.HasValue && limit.Value < result.Count)
+ {
+ result = result.RemoveRange(limit.Value, result.Count - limit.Value);
+ }
+ return Task.FromResult((IReadOnlyList<WrappedEvent>)result);
}
return Task.FromResult(EmptyResult);
}
@@ -161,23 +146,29 @@ namespace WalletWasabi.EventSourcing
string? afterAggregateId = null,
int? limit = null)
{
- if (afterAggregateId != null)
- {
- throw new NotImplementedException();
- }
- limit ??= int.MaxValue;
if (_aggregatesIds.TryGetValue(aggregateType, out var tuple))
{
- var tailIndex = tuple.TailIndex;
var ids = tuple.Ids;
- var result = new List<string>();
- for (var i = 1L; i <= tailIndex && result.Count < limit; i++)
+ var foundIndex = 0;
+ if (afterAggregateId != null)
+ {
+ foundIndex = ids.IndexOf(afterAggregateId);
+ if (foundIndex < 0)
{
- if (!ids.TryGetValue(i, out var id))
+ foundIndex = ~foundIndex;
+ }
+ else
{
- throw new AssertionFailedException($"Unexpected failure to get aggregate id. aggregate type: '{aggregateType}' index: '{i}'");
+ foundIndex++;
}
- result.Add(id);
+ }
+ var result = new List<string>();
+ var afterLastIndex = limit.HasValue ?
+ Math.Min(foundIndex + limit.Value, ids.Count) :
+ ids.Count;
+ for (var i = foundIndex; i < afterLastIndex; i++)
+ {
+ result.Add(ids[i]);
}
return Task.FromResult((IReadOnlyList<string>)result.AsReadOnly());
}
@@ -187,7 +178,8 @@ namespace WalletWasabi.EventSourcing
private void IndexNewAggregateId(string aggregateType, string id)
{
var tailIndex = 0L;
- ConcurrentDictionary<long, string> aggregateIds;
+ ImmutableSortedSet<string> aggregateIds;
+ ImmutableSortedSet<string> newAggregateIds;
var liveLockLimit = 10000;
do
{
@@ -196,16 +188,13 @@ namespace WalletWasabi.EventSourcing
throw new ApplicationException("Live lock detected.");
}
(tailIndex, aggregateIds) = _aggregatesIds.GetOrAdd(aggregateType,
- _ => new(0, new()));
+ _ => new(0, ImmutableSortedSet<string>.Empty));
+ newAggregateIds = aggregateIds.Add(id);
}
while (!_aggregatesIds.TryUpdate(
key: aggregateType,
- newValue: (tailIndex + 1, aggregateIds),
+ newValue: (tailIndex + 1, newAggregateIds),
comparisonValue: (tailIndex, aggregateIds)));
- if (!aggregateIds.TryAdd(tailIndex + 1, id))
- {
- throw new AssertionFailedException("Unexpected failure to add aggregate id to index.");
- }
}
// Helper for parallel critical section testing in DEBUG build only.
@@ -220,22 +209,10 @@ namespace WalletWasabi.EventSourcing
{
}
- // Helper for parallel critical section testing in DEBUG build only.
- [Conditional("DEBUG")]
- protected virtual void Locked()
- {
- }
-
// helper for parallel critical section testing in DEBUG build only.
[Conditional("DEBUG")]
protected virtual void Appended()
{
}
-
- // Helper for parallel critical section testing in DEBUG build only.
- [Conditional("DEBUG")]
- protected virtual void Unlocked()
- {
- }
}
}

Use Immutable list