From d1a06a0298d068dbb94af8a690ae4103750933a6 Mon Sep 17 00:00:00 2001 From: Martin Hans Date: Wed, 29 Apr 2026 15:42:27 +0200 Subject: [PATCH] Fix double-apply of boundary event in stream aggregator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EventRegistrationEventAggregator.AggregateAsync(stream, target, ...) called stream.ListAsync(originalTarget.Version, ...), whose predicate is inclusive on startVersion. The event already applied to the target was fetched again and re-applied — a no-op for idempotent aggregators but a duplication for ones that mutate collections (e.g. List.Add). Split the public overloads onto a private AggregateInternalAsync that takes an explicit fromVersion: 0 for fresh aggregations, target.Version + 1 for continuations. Adds a regression test in Papst.EventStore.Tests using a hand-rolled IEventStream so the test does not depend on storage adapter version semantics. --- .../EventRegistrationEventAggregator.cs | 24 ++- .../EventRegistrationEventAggregatorTests.cs | 160 ++++++++++++++++++ 2 files changed, 180 insertions(+), 4 deletions(-) create mode 100644 tests/Papst.EventStore.Tests/EventRegistrationEventAggregatorTests.cs diff --git a/src/Papst.EventStore.Aggregation.EventRegistration/EventRegistrationEventAggregator.cs b/src/Papst.EventStore.Aggregation.EventRegistration/EventRegistrationEventAggregator.cs index fbd545e..9654d1a 100755 --- a/src/Papst.EventStore.Aggregation.EventRegistration/EventRegistrationEventAggregator.cs +++ b/src/Papst.EventStore.Aggregation.EventRegistration/EventRegistrationEventAggregator.cs @@ -32,10 +32,10 @@ public EventRegistrationEventAggregator( public async Task AggregateAsync(IEventStream stream, ulong targetVersion, CancellationToken cancellationToken) { _logger.CreatingNewEntity(typeof(TEntity).Name, stream.StreamId); - // create the Entity using the StartVersion - 1 because the Aggregator will increment it after applying the first Event - return await AggregateAsync( + return await AggregateInternalAsync( stream, - new TEntity() { Version = 0 }, + new TEntity(), + fromVersion: 0, targetVersion, cancellationToken ).ConfigureAwait(false); @@ -47,6 +47,22 @@ public EventRegistrationEventAggregator( /// public async Task AggregateAsync(IEventStream stream, TEntity originalTarget, ulong targetVersion, CancellationToken cancellationToken) + { + // The caller's target already has events 0..originalTarget.Version applied, so + // start from the next version. ListAsync's predicate is inclusive on startVersion; + // if we passed originalTarget.Version directly the boundary event would be applied + // twice, which is silent for idempotent aggregators but doubles every entry for + // ones that mutate collections (e.g. List.Add). + return await AggregateInternalAsync( + stream, + originalTarget, + fromVersion: originalTarget.Version + 1, + targetVersion, + cancellationToken + ).ConfigureAwait(false); + } + + private async Task AggregateInternalAsync(IEventStream stream, TEntity originalTarget, ulong fromVersion, ulong targetVersion, CancellationToken cancellationToken) { TEntity? target = originalTarget; @@ -61,7 +77,7 @@ public EventRegistrationEventAggregator( stream.Created, stream.Created); - await foreach (var evt in stream.ListAsync(originalTarget.Version, cancellationToken)) + await foreach (var evt in stream.ListAsync(fromVersion, cancellationToken)) { try { diff --git a/tests/Papst.EventStore.Tests/EventRegistrationEventAggregatorTests.cs b/tests/Papst.EventStore.Tests/EventRegistrationEventAggregatorTests.cs new file mode 100644 index 0000000..fe55236 --- /dev/null +++ b/tests/Papst.EventStore.Tests/EventRegistrationEventAggregatorTests.cs @@ -0,0 +1,160 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Newtonsoft.Json.Linq; +using Papst.EventStore.Aggregation; +using Papst.EventStore.Aggregation.EventRegistration; +using Papst.EventStore.Documents; +using Papst.EventStore.EventRegistration; +using Shouldly; +using Xunit; + +namespace Papst.EventStore.Tests; + +/// +/// Regression tests for the boundary-event re-application bug in +/// . Uses a hand-rolled +/// IEventStream so the test does not depend on a specific storage adapter's +/// version-tracking semantics. +/// +public class EventRegistrationEventAggregatorTests +{ + [Fact] + public async Task AggregateAsync_AppliesEachEventOnce_WhenContinuingFromExistingTarget() + { + var aggregator = BuildAggregator(); + var stream = new FakeEventStream(); + stream.Append(new IncrementEvent("a")); + stream.Append(new IncrementEvent("b")); + stream.Append(new IncrementEvent("c")); + + var entity = await aggregator.AggregateAsync(stream, CancellationToken.None); + entity.ShouldNotBeNull(); + entity.Tags.ShouldBe(["a", "b", "c"]); + + stream.Append(new IncrementEvent("d")); + stream.Append(new IncrementEvent("e")); + + entity = await aggregator.AggregateAsync(stream, entity, CancellationToken.None); + + entity.ShouldNotBeNull(); + entity.Tags.ShouldBe(["a", "b", "c", "d", "e"]); + } + + [Fact] + public async Task AggregateAsync_AppliesAllEvents_WhenAggregatingFromScratch() + { + var aggregator = BuildAggregator(); + var stream = new FakeEventStream(); + stream.Append(new IncrementEvent("a")); + stream.Append(new IncrementEvent("b")); + + var entity = await aggregator.AggregateAsync(stream, CancellationToken.None); + + entity.ShouldNotBeNull(); + entity.Tags.ShouldBe(["a", "b"]); + } + + private static IEventStreamAggregator BuildAggregator() + { + var registration = new EventDescriptionEventRegistration(); + registration.AddEvent(new EventAttributeDescriptor(nameof(IncrementEvent), true)); + var typeProvider = new EventRegistrationTypeProvider(NullLogger.Instance, [registration]); + + var services = new ServiceCollection(); + services.AddSingleton(NullLoggerFactory.Instance); + services.AddSingleton(typeof(ILogger<>), typeof(Logger<>)); + services.AddSingleton(typeProvider); + services.AddRegisteredEventAggregation(); + services.AddTransient, IncrementEventAggregator>(); + return services.BuildServiceProvider().GetRequiredService>(); + } + + public class CountingEntity : IEntity + { + public Guid Id { get; set; } + public ulong Version { get; set; } + public List Tags { get; set; } = []; + } + + [EventName(nameof(IncrementEvent))] + public record IncrementEvent(string Tag); + + public class IncrementEventAggregator : EventAggregatorBase + { + public override ValueTask ApplyAsync(IncrementEvent evt, CountingEntity entity, IAggregatorStreamContext ctx) + { + entity.Tags.Add(evt.Tag); + return AsTask(entity); + } + } + + private sealed class FakeEventStream : IEventStream + { + private readonly List _events = []; + + public Guid StreamId { get; } = Guid.NewGuid(); + public ulong Version => _events.Count == 0 ? 0 : _events[^1].Version; + public DateTimeOffset Created { get; } = DateTimeOffset.UtcNow; + public ulong? LatestSnapshotVersion => null; + public EventStreamMetaData MetaData { get; } = new(); + + public void Append(TEvent evt) where TEvent : notnull + { + _events.Add(new EventStreamDocument + { + Id = Guid.NewGuid(), + StreamId = StreamId, + DocumentType = EventStreamDocumentType.Event, + Version = (ulong)_events.Count, + Time = DateTimeOffset.UtcNow, + Name = typeof(TEvent).Name, + Data = JObject.FromObject(evt), + DataType = typeof(TEvent).Name, + TargetType = nameof(CountingEntity), + }); + } + + public Task GetLatestSnapshot(CancellationToken cancellationToken = default) + => Task.FromResult(null); + + public Task AppendAsync(Guid id, TEvent evt, EventStreamMetaData? metaData = null, CancellationToken cancellationToken = default) where TEvent : notnull + => throw new NotSupportedException(); + + public Task AppendSnapshotAsync(Guid id, TEntity entity, EventStreamMetaData? metaData = null, CancellationToken cancellationToken = default) where TEntity : notnull + => throw new NotSupportedException(); + + public Task CreateTransactionalBatchAsync() + => throw new NotSupportedException(); + + public IAsyncEnumerable ListAsync(ulong startVersion = 0u, CancellationToken cancellationToken = default) + => ListAsync(startVersion, Version, cancellationToken); + + public async IAsyncEnumerable ListAsync(ulong startVersion, ulong endVersion, [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + foreach (var doc in _events) + { + if (doc.Version >= startVersion && doc.Version <= endVersion) + { + yield return doc; + } + } + await Task.CompletedTask; + } + + public IAsyncEnumerable ListDescendingAsync(ulong endVersion, ulong startVersion, CancellationToken cancellationToken = default) + => throw new NotSupportedException(); + + public IAsyncEnumerable ListDescendingAsync(ulong endVersion, CancellationToken cancellationToken = default) + => throw new NotSupportedException(); + + public Task UpdateStreamMetaData(EventStreamMetaData metaData, CancellationToken cancellationToken = default) + => Task.CompletedTask; + } +}