Skip to content

Fix the aggregate version calculation #475

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alexeyzimarev merged 2 commits into from
Nov 23, 2025
Merged

Fix the aggregate version calculation #475

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
af2da67
* Fix the aggregate version calculation
alexeyzimarev Nov 23, 2025
4868104
Check if stream is empty
alexeyzimarev Nov 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/Core/gen/Eventuous.Shared.Generators/TypeMappingsGenerator.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,12 +163,12 @@ static void Generate(SourceProductionContext context, string assemblyName, Immut
sb.AppendLine();
sb.AppendLine($"internal static class {className} {{");
sb.AppendLine(" [ModuleInitializer]");
sb.AppendLine(" internal static void Initialize() => Register(TypeMap.Instance);");
sb.AppendLine(" internal static void Initialize() => RegisterDiscoveredTypes(TypeMap.Instance);");
sb.AppendLine();
sb.AppendLine(" /// <summary>");
sb.AppendLine(" /// Registers all [EventType] types discovered at compile time into the provided mapper.");
sb.AppendLine(" /// </summary>");
sb.AppendLine(" public static void Register(TypeMapper mapper) {");
sb.AppendLine(" public static void RegisterDiscoveredTypes(this TypeMapper mapper) {");

foreach (var m in distinct) {
// Prefer passing the explicit name when we have it, so runtime reflection is avoided
Expand Down
11 changes: 6 additions & 5 deletions src/Core/src/Eventuous.Domain/Aggregate.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,13 +30,13 @@ namespace Eventuous;
/// It is used for optimistic concurrency to check if there were no changes made to the
/// aggregate state between load and save for the current operation.
/// </summary>
public int OriginalVersion => Original.Length - 1;
public long OriginalVersion { get; private set; } = -1;

/// <summary>
/// The current version is set to the original version when the aggregate is loaded from the store.
/// It should increase for each state transition performed within the scope of the current operation.
/// </summary>
public int CurrentVersion => OriginalVersion + Changes.Count;
public long CurrentVersion => OriginalVersion + Changes.Count;

readonly List<object> _changes = [];

Expand Down Expand Up @@ -78,9 +78,10 @@ protected void EnsureExists(Func<Exception>? getException = null) {
return (previous, State);
}

public void Load(IEnumerable<object?> events) {
Original = events.Where(x => x != null).ToArray()!;
State = Original.Aggregate(State, Fold);
public void Load(long version, IEnumerable<object?> events) {
Original = events.Where(x => x != null).ToArray()!;
OriginalVersion = version;
State = Original.Aggregate(State, Fold);

return;

Expand Down
3 changes: 2 additions & 1 deletion src/Core/src/Eventuous.Persistence/AggregateStore/AggregatePersistenceExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,8 @@ public Task<AppendEventsResult> StoreAggregate<TAggregate, TState, TId>(

try {
var events = await eventReader.ReadStream(streamName, StreamReadPosition.Start, failIfNotFound, cancellationToken).NoContext();
aggregate.Load(events.Select(x => x.Payload));
if (events.Length == 0) return aggregate;
aggregate.Load(events[^1].Revision, events.Select(x => x.Payload));
} catch (StreamNotFound) when (!failIfNotFound) {
return aggregate;
} catch (Exception e) {
Expand Down
12 changes: 6 additions & 6 deletions src/Core/src/Eventuous.Persistence/EventStore/TieredEventReader.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,8 @@ public class TieredEventReader(IEventReader hotReader, IEventReader archiveReade
public async Task<StreamEvent[]> ReadEvents(StreamName streamName, StreamReadPosition start, int count, bool failIfNotFound, CancellationToken cancellationToken) {
var hotEvents = await LoadStreamEvents(hotReader, start, count, true).NoContext();

var archivedEvents = hotEvents.Length == 0 || hotEvents[0].Position > start.Value
? await LoadStreamEvents(archiveReader, start, (int)hotEvents[0].Position, !failIfNotFound).NoContext()
var archivedEvents = hotEvents.Length == 0 || hotEvents[0].Revision > start.Value
? await LoadStreamEvents(archiveReader, start, (int)hotEvents[0].Revision, !failIfNotFound).NoContext()
: Enumerable.Empty<StreamEvent>();

return archivedEvents.Select(x => x with { FromArchive = true }).Concat(hotEvents).Distinct(Comparer).ToArray();
Expand All @@ -35,8 +35,8 @@ async Task<StreamEvent[]> LoadStreamEvents(IEventReader reader, StreamReadPositi
public async Task<StreamEvent[]> ReadEventsBackwards(StreamName streamName, StreamReadPosition start, int count, bool failIfNotFound, CancellationToken cancellationToken) {
var hotEvents = await LoadStreamEvents(hotReader, start, count, true).NoContext();

var archivedEvents = hotEvents.Length == 0 || hotEvents[0].Position > start.Value - count
? await LoadStreamEvents(archiveReader, new(hotEvents[0].Position - 1), count - hotEvents.Length, failIfNotFound).NoContext()
var archivedEvents = hotEvents.Length == 0 || hotEvents[0].Revision > start.Value - count
? await LoadStreamEvents(archiveReader, new(hotEvents[0].Revision - 1), count - hotEvents.Length, failIfNotFound).NoContext()
: Enumerable.Empty<StreamEvent>();

return hotEvents.Concat(archivedEvents.Select(x => x with { FromArchive = true })).Distinct(Comparer).ToArray();
Expand All @@ -53,8 +53,8 @@ async Task<StreamEvent[]> LoadStreamEvents(IEventReader reader, StreamReadPositi
static readonly StreamEventPositionComparer Comparer = new();

class StreamEventPositionComparer : IEqualityComparer<StreamEvent> {
public bool Equals(StreamEvent x, StreamEvent y) => x.Position == y.Position;
public bool Equals(StreamEvent x, StreamEvent y) => x.Revision == y.Revision;

public int GetHashCode(StreamEvent obj) => obj.Position.GetHashCode();
public int GetHashCode(StreamEvent obj) => obj.Revision.GetHashCode();
}
}
2 changes: 1 addition & 1 deletion src/Core/src/Eventuous.Persistence/StateStore/StateStoreFunctions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ public async Task<FoldedEventStream<TState>> LoadState<TState>(
try {
var streamEvents = await reader.ReadStream(streamName, StreamReadPosition.Start, failIfNotFound, cancellationToken).NoContext();
var events = streamEvents.Select(x => x.Payload!).ToArray();
var expectedVersion = events.Length == 0 ? ExpectedStreamVersion.NoStream : new(streamEvents.Last().Position);
var expectedVersion = events.Length == 0 ? ExpectedStreamVersion.NoStream : new(streamEvents.Last().Revision);

return (new(streamName, expectedVersion, events));
} catch (StreamNotFound) when (!failIfNotFound) {
Expand Down
4 changes: 2 additions & 2 deletions src/Core/src/Eventuous.Persistence/StreamEvent.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,10 @@

using System.Runtime.InteropServices;

namespace Eventuous;
namespace Eventuous;

[StructLayout(LayoutKind.Auto)]
public record struct NewStreamEvent(Guid Id, object? Payload, Metadata Metadata);

[StructLayout(LayoutKind.Auto)]
public record struct StreamEvent(Guid Id, object? Payload, Metadata Metadata, string ContentType, long Position, bool FromArchive = false);
public record struct StreamEvent(Guid Id, object? Payload, Metadata Metadata, string ContentType, long Revision, bool FromArchive = false);
2 changes: 1 addition & 1 deletion src/Core/test/Eventuous.Tests.Application/ServiceTestBase.Amendments.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ public async Task Should_amend_event_from_command(CancellationToken cancellation
var service = CreateService(amendEvent: AmendEvent);
var cmd = CreateCommand();

await service.Handle(cmd, cancellationToken);
var result = await service.Handle(cmd, cancellationToken);

var stream = await Store.ReadStream(StreamName.For<Booking>(cmd.BookingId), StreamReadPosition.Start, cancellationToken: cancellationToken);
await Assert.That(stream[0].Metadata["userId"]).IsEqualTo(cmd.ImportedBy);
Expand Down
2 changes: 1 addition & 1 deletion src/Experimental/src/Eventuous.ElasticSearch/Store/ElasticEventStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ public async Task<StreamEvent[]> ReadEventsBackwards(StreamName stream, StreamRe
cancellationToken
);

return response.Length == 0 && failIfNotFound ? throw new StreamNotFound(stream) : response.OrderByDescending(x => x.Position).ToArray();
return response.Length == 0 && failIfNotFound ? throw new StreamNotFound(stream) : response.OrderByDescending(x => x.Revision).ToArray();
}

public async Task<bool> StreamExists(StreamName stream, CancellationToken cancellationToken) {
Expand Down
19 changes: 18 additions & 1 deletion src/KurrentDB/test/Eventuous.Tests.KurrentDB/AppServiceTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ public class AppServiceTests {

public AppServiceTests(StoreFixture fixture) {
_fixture = fixture;
_fixture.TypeMapper.AddType<BookingEvents.BookingImported>();
_fixture.TypeMapper.RegisterDiscoveredTypes();
}

BookingService Service { get; set; } = null!;
Expand All @@ -39,6 +39,23 @@ public async Task ProcessAnyForNew(CancellationToken cancellationToken) {
result.ShouldBeEquivalentTo(expected);
}

[Test]
public async Task ProcessNewThenDeleteAndDoItAgain(CancellationToken cancellationToken) {
// This will create a new stream
var cmd = DomainFixture.CreateImportBooking();
await Service.Handle(cmd, cancellationToken);

var streamName = StreamName.For<Booking>(cmd.BookingId);
await _fixture.EventStore.DeleteStream(streamName, ExpectedStreamVersion.Any, cancellationToken);

var handlingResult = await Service.Handle(cmd, cancellationToken);
handlingResult.Success.ShouldBeTrue();

var cancelCmd = new Commands.CancelBooking(new(cmd.BookingId));
var secondResult = await Service.Handle(cancelCmd, cancellationToken);
Comment on lines +53 to +55
Copy link
Contributor

@qodo-free-for-open-source-projects qodo-free-for-open-source-projects bot Nov 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggestion: Reuse existing ID object in test

Suggested change
var cancelCmd = new Commands.CancelBooking(new(cmd.BookingId));
var secondResult = await Service.Handle(cancelCmd, cancellationToken);
var cancelCmd = new Commands.CancelBooking(cmd.BookingId);
var secondResult = await Service.Handle(cancelCmd, cancellationToken);

Copy link
Contributor Author

@alexeyzimarev alexeyzimarev Nov 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This won't work because cmd.BookingId is a string but CancelBooking constructor expects a BookingId instance which has no implicit conversion from string.

Copy link
Contributor

@qodo-free-for-open-source-projects qodo-free-for-open-source-projects bot Nov 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Question:

This won't work because cmd.BookingId is a string but CancelBooking constructor expects a BookingId instance which has no implicit conversion from string.

Answer:

The user has identified a type mismatch issue where the AI suggested using cmd.BookingId (a string) directly in a constructor that expects a BookingId object instance. This is a case where the AI made an incorrect assumption about the code structure.

To help prevent similar issues in future suggestions, you can add specific guidance to the extra_instructions configuration:

[pr_code_suggestions]
extra_instructions = """
- When suggesting code changes involving domain objects or value objects, ensure type compatibility
- Do not assume implicit conversions between primitive types (like string) and domain objects
- If a constructor requires a specific type instance, verify the variable type matches before suggesting its use
"""

However, it's important to note that this represents an inherent limitation of AI code analysis. The model may not always have complete context about custom type systems, domain objects, and their conversion rules in your codebase. While the extra_instructions can help guide the model toward better type awareness, it cannot guarantee the model will always correctly identify type mismatches in complex domain models.

For C# codebases with strong typing and domain-driven design patterns, you might also consider:

[pr_code_suggestions]
extra_instructions = """
- Respect strong typing in C# - do not suggest implicit conversions that don't exist
- When working with value objects or domain entities, verify constructor signatures match the provided arguments
"""

This type of error is relatively rare but can occur when the AI lacks full context about your type system. The suggestions should still be reviewed critically, as recommended in the Understanding AI Code Suggestions documentation.

Relevant Sources:

secondResult.Success.ShouldBeTrue();
}

[After(Test)]
public void Dispose() => _listener.Dispose();
}
2 changes: 1 addition & 1 deletion src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToStream.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,6 @@ async Task<List<BookingImported>> GenerateAndProduceEvents(int count) {
async Task<long> GetStreamPosition(int count) {
var readEvents = await _fixture.IntegrationFixture.EventReader.ReadEvents(_fixture.Stream, StreamReadPosition.Start, count, true, default);

return readEvents.Last().Position;
return readEvents.Last().Revision;
}
}
3 changes: 2 additions & 1 deletion src/Testing/src/Eventuous.Testing/AggregateSpec.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,8 @@ public abstract class AggregateSpec<TAggregate, TState>(AggregateFactoryRegistry
[MemberNotNull(nameof(Instance))]
protected TAggregate Then() {
Instance = CreateInstance();
Instance.Load(GivenEvents());
var events = GivenEvents();
Instance.Load(events.Length - 1, events);
When(Instance);

return Instance;
Expand Down
2 changes: 1 addition & 1 deletion src/Testing/src/Eventuous.Testing/CommandServiceFixture.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -256,7 +256,7 @@ public FixtureResult FullStreamEventsAre(params object[] events) {
/// <returns></returns>
[StackTraceHidden]
public FixtureResult NewStreamEventsAre(params object[] events) {
var stream = _streamEvents.Where(x => x.Position >= _version).Select(x => x.Payload);
var stream = _streamEvents.Where(x => x.Revision >= _version).Select(x => x.Payload);
stream.ShouldBe(events);

return this;
Expand Down
2 changes: 1 addition & 1 deletion src/Testing/src/Eventuous.Testing/InMemoryEventStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ public IEnumerable<StreamEvent> GetEvents(StreamReadPosition from, int count) {

if (count > 0) selected = selected.Take(count);

return selected.Select(x => x.Event with { Position = x.Position });
return selected.Select(x => x.Event with { Revision = x.Position });
}

public IEnumerable<StreamEvent> GetEventsBackwards(StreamReadPosition from, int count) {
Expand Down
Loading