Improve IEventRepository interface
MODIFY
WalletWasabi/EventSourcing/InMemoryEventRepository.cs
WalletWasabi/EventSourcing/InMemoryEventRepository.cs
@@ -22,27 +22,41 @@ namespace WalletWasabi.EventSourcing
private readonly ConcurrentDictionary<
string /*aggregateType*/,
- ConcurrentDictionary<string /*id*/,
- (long LastSequenceId,
+ ConcurrentDictionary<
+ string /*aggregateId*/,
+ (
+ // SequenceId of the last event of this aggregate
+ long TailSequenceId,
+
+ // locked flag for appending into EventsBatches of this aggregate
bool Locked,
- ConcurrentQueue<IReadOnlyList<WrappedEvent>> EventsBatches) /* sequence of atomically appended event batches */>>
- _aggregatesEventsBatches = new();
+
+ // list of lists of events for atomic insertion of multiple events
+ // in one "database transaction"
+ ConcurrentQueue<IReadOnlyList<WrappedEvent>> EventsBatches
+ )
+ >
+ > _aggregatesEventsBatches = new();
private readonly ConcurrentDictionary<
string /*aggregateType*/,
- (long LastIndex,
+ (
+ // index of the last aggregateId in this aggregateType
+ long TailIndex,
ConcurrentDictionary<
- long, /*index*/
- string /*id*/> Ids)>
- _aggregatesIds = new();
+ // index of aggregateId in this aggregateType
+ long,
+ string /*aggregateId*/> Ids
+ )
+ > _aggregatesIds = new();
public Task AppendEventsAsync(
string aggregateType,
- string id,
+ string aggregateId,
IEnumerable<WrappedEvent> wrappedEvents)
{
Guard.NotNullOrEmpty(nameof(aggregateType), aggregateType);
- Guard.NotNullOrEmpty(nameof(id), id);
+ Guard.NotNullOrEmpty(nameof(aggregateId), aggregateId);
Guard.NotNull(nameof(wrappedEvents), wrappedEvents);
var wrappedEventsList = wrappedEvents.ToList().AsReadOnly();
if (wrappedEventsList.Count <= 0)
@@ -65,18 +79,23 @@ namespace WalletWasabi.EventSourcing
}
var aggregateEventsBatches = _aggregatesEventsBatches.GetOrAdd(aggregateType, _ => new());
- var (concurrentSequenceId, locked, eventsBatches) = aggregateEventsBatches.GetOrAdd(id, _ => (0, false, new()));
+ var (tailSequenceId, locked, eventsBatches) = aggregateEventsBatches.GetOrAdd(aggregateId, _ => (0, false, new()));
- if (concurrentSequenceId + 1 < firstSequenceId)
+ if (tailSequenceId + 1 < firstSequenceId)
{
- throw new ArgumentException($"invalid firstSequenceId (gap in sequence ids) expected: '{concurrentSequenceId + 1}' given: '{firstSequenceId}'", nameof(wrappedEvents));
+ throw new ArgumentException($"invalid firstSequenceId (gap in sequence ids) expected: '{tailSequenceId + 1}' given: '{firstSequenceId}'", nameof(wrappedEvents));
}
- Validated(); // no action
+
+ // no action
+ Validated();
// 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)))
+ if (!aggregateEventsBatches.TryUpdate(
+ key: aggregateId,
+ newValue: (lastSequenceId, true, eventsBatches),
+ comparisonValue: (firstSequenceId - 1, false, eventsBatches)))
{
Conflicted(); // no action
- throw new OptimisticConcurrencyException($"Conflict while commiting events. Retry command. aggregate: '{aggregateType}' id: '{id}'");
+ throw new OptimisticConcurrencyException($"Conflict while commiting events. Retry command. aggregate: '{aggregateType}' id: '{aggregateId}'");
}
try
{
@@ -87,7 +106,10 @@ namespace WalletWasabi.EventSourcing
finally
{
// unlock
- if (!aggregateEventsBatches.TryUpdate(id, (lastSequenceId, false, eventsBatches), (lastSequenceId, true, eventsBatches)))
+ if (!aggregateEventsBatches.TryUpdate(
+ key: aggregateId,
+ newValue: (lastSequenceId, false, eventsBatches),
+ comparisonValue: (lastSequenceId, true, eventsBatches)))
{
#warning
// TODO: convert into Debug.Assert ???
@@ -97,35 +119,33 @@ namespace WalletWasabi.EventSourcing
}
// if it is a first event for given aggregate
- if (concurrentSequenceId == 0)
+ if (tailSequenceId == 0)
{ // add index of aggregate id into the dictionary
- IndexNewAggregateId(aggregateType, id);
+ IndexNewAggregateId(aggregateType, aggregateId);
}
return Task.CompletedTask;
}
- public Task<IReadOnlyList<WrappedEvent>> ListEventsAsync(string aggregateType, string id, long fromSequenceId = 0, int? limit = null)
+ public Task<IReadOnlyList<WrappedEvent>> ListEventsAsync(string aggregateType,
+ string aggregateId, long afterSequenceId = -1)
{
if (_aggregatesEventsBatches.TryGetValue(aggregateType, out var aggregateEventsBatches) &&
- aggregateEventsBatches.TryGetValue(id, out var value))
+ aggregateEventsBatches.TryGetValue(aggregateId, out var value))
{
var result = value.EventsBatches.SelectMany(a => a);
- if (0 < fromSequenceId)
+ if (-1 < afterSequenceId)
{
- result = result.Where(a => fromSequenceId <= a.SequenceId);
- }
- if (limit.HasValue)
- {
- result = result.Take(limit.Value);
+ result = result.Where(a => afterSequenceId < a.SequenceId);
}
return Task.FromResult((IReadOnlyList<WrappedEvent>)result.ToList().AsReadOnly());
}
return Task.FromResult(EmptyResult);
}
- public Task<IReadOnlyList<string>> ListAggregateIdsAsync(string aggregateType, string? fromId = null, int? limit = null)
+ public Task<IReadOnlyList<string>> ListAggregateIdsAsync(string aggregateType,
+ string? afterId = null, int? limit = null)
{
- if (fromId != null)
+ if (afterId != null)
{
throw new NotImplementedException();
}
@@ -135,7 +155,7 @@ namespace WalletWasabi.EventSourcing
}
if (_aggregatesIds.TryGetValue(aggregateType, out var tuple))
{
- var lastIndex = tuple.LastIndex;
+ var lastIndex = tuple.TailIndex;
var ids = tuple.Ids;
var result = new List<string>();
for (var i = 1L; i <= lastIndex; i++)
@@ -164,8 +184,13 @@ namespace WalletWasabi.EventSourcing
{
throw new ApplicationException("live lock detected");
}
- (lastIndex, aggregateIds) = _aggregatesIds.GetOrAdd(aggregateType, _ => new(0, new()));
- } while (!_aggregatesIds.TryUpdate(aggregateType, (lastIndex + 1, aggregateIds), (lastIndex, aggregateIds)));
+ (lastIndex, aggregateIds) = _aggregatesIds.GetOrAdd(aggregateType,
+ _ => new(0, new()));
+ }
+ while (!_aggregatesIds.TryUpdate(
+ key: aggregateType,
+ newValue: (lastIndex + 1, aggregateIds),
+ comparisonValue: (lastIndex, aggregateIds)));
if (!aggregateIds.TryAdd(lastIndex + 1, id))
{
#warning
@@ -174,30 +199,31 @@ namespace WalletWasabi.EventSourcing
}
}
- // for parallel critical section testing
+ // helper for parallel critical section testing in DEBUG build only
[Conditional("DEBUG")]
protected virtual void Validated()
{
}
- // for parallel critical section testing
+ // helper for parallel critical section testing in DEBUG build only
[Conditional("DEBUG")]
protected virtual void Conflicted()
{
}
- // for parallel critical section testing
+ // helper for parallel critical section testing in DEBUG build only
[Conditional("DEBUG")]
protected virtual void Locked()
{
}
- // for parallel critical section testing
+ // 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()
{

MODIFY
WalletWasabi/Interfaces/EventSourcing/IEventRepository.cs
WalletWasabi/Interfaces/EventSourcing/IEventRepository.cs
@@ -14,12 +14,13 @@ namespace WalletWasabi.Interfaces.EventSourcing
/// Atomically persistently appends ordered list of events for given aggregate.
/// In case of duplicate of <seealso cref="WrappedEvent.SequenceId"/>
/// <seealso cref="OptimisticConcurrencyException"/> is thrown indicating that
- /// command should be retried with freshly loaded events.
+ /// 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>
public Task AppendEventsAsync(
string aggregateType,
@@ -27,32 +28,34 @@ namespace WalletWasabi.Interfaces.EventSourcing
IEnumerable<WrappedEvent> wrappedEvents);
/// <summary>
- /// List events for given aggregate. This the primary source of truth.
+ /// 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="fromSequenceId">starts with event with given <seealso cref="WrappedEvent.SequenceId"/>
+ /// <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>
public Task<IReadOnlyList<WrappedEvent>> ListEventsAsync(
string aggregateType,
string aggregateId,
- long fromSequenceId = 0,
- int? limit = null);
+ long afterSequenceId = 0);
/// <summary>
- /// Supplementary method to enumerating all aggregate ids in this event repository.
- /// Order of ids is not defined can be any artificial.
+ /// 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="fromId">starts with given id and lists following in any artificial order</param>
+ /// <param name="afterId">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>
public Task<IReadOnlyList<string>> ListAggregateIdsAsync(
string aggregateType,
- string? fromId = null,
+ string? afterId = null,
int? limit = null);
}
}

Improve IEventRepository interface