Attempt to make the grammar Nazi happy
MODIFY
WalletWasabi.Tests/UnitTests/EventSourcing/InMemoryEventRepositoryTests.cs
WalletWasabi.Tests/UnitTests/EventSourcing/InMemoryEventRepositoryTests.cs
@@ -19,11 +19,6 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
private const string TestRoundAggregate = "TestRoundAggregate";
private readonly TimeSpan _semaphoreWaitTimeout = TimeSpan.FromSeconds(1);
- private IEventRepository EventRepository { get; init; }
- private TestInMemoryEventRepository TestEventRepository { get; init; }
-
- private ITestOutputHelper Output { get; init; }
-
public InMemoryEventRepositoryTests(ITestOutputHelper output)
{
Output = output;
@@ -31,6 +26,11 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
EventRepository = TestEventRepository;
}
+ private IEventRepository EventRepository { get; init; }
+ private TestInMemoryEventRepository TestEventRepository { get; init; }
+
+ private ITestOutputHelper Output { get; init; }
+
[Fact]
public async Task AppendEvents_Zero_Async()
{
@@ -225,7 +225,6 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
[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") };
@@ -365,7 +364,8 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
else
{ // conflict
await Assert.ThrowsAsync<OptimisticConcurrencyException>(AppendInParallel);
- // retry after optimistic concurrency conflict
+
+ // Retry after optimistic concurrency conflict
TestEventRepository.LockedCallback = null;
TestEventRepository.AppendedCallback = null;
TestEventRepository.UnlockedCallback = null;
@@ -449,8 +449,11 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
public async Task ListEventsAsync_OptionalArguments_Async(long afterSequenceId, int limit)
{
// Arrange
- var events = new[] { new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a"),
- new TestWrappedEvent(3, "b"), new TestWrappedEvent(4, "b") };
+ 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);
// Act
@@ -466,8 +469,11 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing
public async Task ListAggregateIdsAsync_Async()
{
// Arrange
- var events = new[] { new TestWrappedEvent(1, "a"), new TestWrappedEvent(2, "a"),
- new TestWrappedEvent(3, "b"), new TestWrappedEvent(4, "b") };
+ 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);

MODIFY
WalletWasabi.Tests/UnitTests/EventSourcing/TestDomain/TestInMemoryEventRepository.cs
WalletWasabi.Tests/UnitTests/EventSourcing/TestDomain/TestInMemoryEventRepository.cs
@@ -11,6 +11,11 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing.TestDomain
{
public class TestInMemoryEventRepository : InMemoryEventRepository
{
+ public TestInMemoryEventRepository(ITestOutputHelper output)
+ {
+ Output = output;
+ }
+
protected ITestOutputHelper Output { get; init; }
public SemaphoreSlim ValidatedSemaphore { get; } = new SemaphoreSlim(0);
@@ -25,11 +30,6 @@ namespace WalletWasabi.Tests.UnitTests.EventSourcing.TestDomain
public Action? AppendedCallback { get; set; }
public Action? UnlockedCallback { get; set; }
- public TestInMemoryEventRepository(ITestOutputHelper output)
- {
- Output = output;
- }
-
protected override void Validated()
{
base.Validated();

MODIFY
WalletWasabi/EventSourcing/InMemoryEventRepository.cs
WalletWasabi/EventSourcing/InMemoryEventRepository.cs
@@ -52,6 +52,12 @@ namespace WalletWasabi.EventSourcing
)
> _aggregatesIds = new();
+ protected enum WriteLockEnum
+ {
+ Unlocked,
+ WritingLocked,
+ }
+
public Task AppendEventsAsync(
string aggregateType,
string aggregateId,
@@ -90,6 +96,7 @@ namespace WalletWasabi.EventSourcing
// no action
Validated();
+
// Atomically detect conflict and replace lastSequenceId and lock to ensure strong order in eventsBatches.
if (!aggregateEventsBatches.TryUpdate(
key: aggregateId,
@@ -126,14 +133,17 @@ namespace WalletWasabi.EventSourcing
return Task.CompletedTask;
}
- public Task<IReadOnlyList<WrappedEvent>> ListEventsAsync(string aggregateType,
- string aggregateId, long afterSequenceId = 0, int? limit = null)
+ public Task<IReadOnlyList<WrappedEvent>> ListEventsAsync(
+ string aggregateType,
+ string aggregateId,
+ long afterSequenceId = 0,
+ int? limit = null)
{
if (_aggregatesEventsBatches.TryGetValue(aggregateType, out var aggregateEventsBatches) &&
aggregateEventsBatches.TryGetValue(aggregateId, out var value))
{
var result = value.EventsBatches.SelectMany(a => a);
- if (0 < afterSequenceId)
+ if (afterSequenceId > 0)
{
result = result.Where(a => afterSequenceId < a.SequenceId);
}
@@ -146,8 +156,10 @@ namespace WalletWasabi.EventSourcing
return Task.FromResult(EmptyResult);
}
- public Task<IReadOnlyList<string>> ListAggregateIdsAsync(string aggregateType,
- string? afterAggregateId = null, int? limit = null)
+ public Task<IReadOnlyList<string>> ListAggregateIdsAsync(
+ string aggregateType,
+ string? afterAggregateId = null,
+ int? limit = null)
{
if (afterAggregateId != null)
{
@@ -225,11 +237,5 @@ namespace WalletWasabi.EventSourcing
protected virtual void Unlocked()
{
}
-
- protected enum WriteLockEnum
- {
- Unlocked,
- WritingLocked,
- }
}
}

MODIFY
WalletWasabi/Interfaces/EventSourcing/IEventRepository.cs
WalletWasabi/Interfaces/EventSourcing/IEventRepository.cs
@@ -16,12 +16,12 @@ namespace WalletWasabi.Interfaces.EventSourcing
/// <seealso cref="OptimisticConcurrencyException"/> is thrown indicating that
/// command should be retried with freshly loaded events from <see cref="ListEventsAsync"/>.
/// </summary>
- /// <param name="aggregateType">type of aggregate</param>
- /// <param name="aggregateId">id of aggregate</param>
- /// <param name="wrappedEvents">ordered list of events to be persisted</param>
- /// <exception cref="OptimisticConcurrencyException">if there is concurrency conflict ; retry command</exception>
- /// <exception cref="TransientException">transient infrastracture failure</exception>
- /// <exception cref="ArgumentException">invalid input</exception>
+ /// <param name="aggregateType">Type of aggregate</param>
+ /// <param name="aggregateId">Id of aggregate</param>
+ /// <param name="wrappedEvents">Ordered list of events to be persisted</param>
+ /// <exception cref="OptimisticConcurrencyException">If there is concurrency conflict ; retry command</exception>
+ /// <exception cref="TransientException">Transient infrastracture failure</exception>
+ /// <exception cref="ArgumentException">Invalid input</exception>
public Task AppendEventsAsync(
string aggregateType,
string aggregateId,
@@ -31,13 +31,13 @@ namespace WalletWasabi.Interfaces.EventSourcing
/// List strongly ordered events of given aggregate. This is the primary source of truth.
/// Events are used to reconstruct aggregate state before processing command.
/// </summary>
- /// <param name="aggregateType">type of aggregate</param>
- /// <param name="aggregateId">id of aggregate</param>
- /// <param name="afterSequenceId">starts with event after given <seealso cref="WrappedEvent.SequenceId"/>
+ /// <param name="aggregateType">Type of aggregate</param>
+ /// <param name="aggregateId">Id of aggregate</param>
+ /// <param name="afterSequenceId">Starts with event after given <seealso cref="WrappedEvent.SequenceId"/>
/// and lists all following events</param>
- /// <param name="limit">limits the number of returned events</param>
- /// <returns>ordered list of events</returns>
- /// <exception cref="TransientException">transient infrastracture failure</exception>
+ /// <param name="limit">Limits the number of returned events</param>
+ /// <returns>Ordered list of events</returns>
+ /// <exception cref="TransientException">Transient infrastracture failure</exception>
public Task<IReadOnlyList<WrappedEvent>> ListEventsAsync(
string aggregateType,
string aggregateId,
@@ -48,12 +48,12 @@ namespace WalletWasabi.Interfaces.EventSourcing
/// Supplementary method for enumerating all ids for <paramref name="aggregateType"/>
/// in this event repository. Order of ids is not defined can be any artificial.
/// </summary>
- /// <param name="aggregateType">type of aggregate</param>
- /// <param name="afterAggregateId">starts with id right after given id and lists all following ids
+ /// <param name="aggregateType">Type of aggregate</param>
+ /// <param name="afterAggregateId">Starts with id right after given id and lists all following ids
/// in any artificial order</param>
- /// <param name="limit">limits the number of returned events</param>
- /// <returns>unordered list of ids</returns>
- /// <exception cref="TransientException">transient infrastracture failure</exception>
+ /// <param name="limit">Limits the number of returned events</param>
+ /// <returns>Unordered list of ids</returns>
+ /// <exception cref="TransientException">Transient infrastracture failure</exception>
public Task<IReadOnlyList<string>> ListAggregateIdsAsync(
string aggregateType,
string? afterAggregateId = null,

Attempt to make the grammar Nazi happy