From ace8efee2c6d9c2a367db99042575a78c64e0e09 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 4 Oct 2026 16:49:55 -0500 Subject: [PATCH] Live aggregation stamps a strong-typed aggregate id (#426) AggregateIdentity.TrySetIdentity backfills the stream id onto an aggregate whose Create did not set it, as Marten and Polecat do. It assigned only a value the identity member could hold as-is, and a raw Guid is never a KilnId, so a strong-typed id was skipped and the aggregate came back with a default id from live aggregation, FetchForWriting, FetchManyForWriting and ProjectLatest. The fold was right and the identity was not, silently. Not AOT-specific: it reproduced under the JIT. It now wraps the stream id through StrongTypedId.Wrap when the member is a wrapper around the stream id's own type. A Guid is never turned into a string-backed id or the reverse, the rule FetchForWriting refuses by. strong_typed_aggregate_identity fails 5 of 7 against the old backfill. The inline snapshot never had the bug (it goes through the projection's identity setter), and that test pins it. The AOT smoke asserts both strong-typed aggregates' ids natively. Closes #426 Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 11 +- HANDOFF.md | 2 +- README.md | 2 +- smoke/aot-consumer/Program.cs | 4 +- .../Events/strong_typed_aggregate_identity.cs | 158 ++++++++++++++++++ src/Fisher/Storage/AggregateIdentity.cs | 41 ++++- 6 files changed, 205 insertions(+), 13 deletions(-) create mode 100644 src/Fisher.Tests/Events/strong_typed_aggregate_identity.cs diff --git a/CLAUDE.md b/CLAUDE.md index 4a715e9..ae4f4ee 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1389,8 +1389,15 @@ reloads an inline `Snapshot`. - **The declared id type is checked against the identity member**, because the source generator keys the dispatcher on the member's type and a mismatch would fail at the first event, far from the registration. - - ⚠️ **Live aggregation leaves a strong-typed `Id` at its default** — fisher#426, pre-existing and not - AOT-specific (it reproduces under the JIT on 2.80.1). The smoke asserts the folded state only. + - **Live aggregation stamps a strong-typed `Id` too, since fisher#426.** `AggregateIdentity.TrySetIdentity` + backfills the stream id onto an aggregate whose `Create` did not set it, and it assigned only a value + the member could hold as-is — a raw `Guid` is never a `KilnId`, so every wrapper was skipped and came + back at its default from live aggregation, `FetchForWriting`, `FetchManyForWriting` and `ProjectLatest`. + Not AOT-specific: it reproduced under the JIT. It now wraps through `StrongTypedId.Wrap`, and only a + wrapper around the stream id's own type, which is the rule `FetchForWriting` refuses by. + `strong_typed_aggregate_identity` fails 5 of 7 against the old backfill. The inline snapshot never had + the bug (it goes through the projection's identity setter), and that test pins it. The smoke asserts + the ids natively. - **The smoke consumer references `JasperFx.Events.SourceGenerator` itself**, because a project reference does not carry the analyzer the package bundles. Its version is kept in step by hand. diff --git a/HANDOFF.md b/HANDOFF.md index 330c10d..6fc46c7 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -12,7 +12,7 @@ equivalent for and never will. [CLAUDE.md](CLAUDE.md) has the architecture and the SQLite traps. This document is the compliance scoreboard and the things that are true right now but not obvious from either. -**2598 tests green on net9.0 and net10.0** — 2531 in `Fisher.Tests`, 36 in +**2605 tests green on net9.0 and net10.0** — 2538 in `Fisher.Tests`, 36 in `Fisher.AspNetCore.Tests` and 31 in `Fisher.EntityFrameworkCore.Tests`. 678 of them are shared cross-store compliance tests — 509 event sourcing and 169 document. On JasperFx **2.80.2** / Weasel **9.40.0**. diff --git a/README.md b/README.md index baf66da..679325d 100644 --- a/README.md +++ b/README.md @@ -27,7 +27,7 @@ process**. There is no server to install, nothing to provision, and nothing to k > `Fisher.AspNetCore` and `Fisher.EntityFrameworkCore` companion packages all work and are tested. > > Fisher passes **all 58 suites and 678 tests** it enrolls from `JasperFx.Events.ComplianceTests`, -> the shared cross-store suite Marten and Polecat also enroll in, alongside its own 2,531. +> the shared cross-store suite Marten and Polecat also enroll in, alongside its own 2,538. > > That suite pins **API portability, not behavioural equivalence** — code written against one store > compiles and runs against another. It does not pin that the three behave identically, and they do diff --git a/smoke/aot-consumer/Program.cs b/smoke/aot-consumer/Program.cs index 372bf6a..93306d5 100644 --- a/smoke/aot-consumer/Program.cs +++ b/smoke/aot-consumer/Program.cs @@ -163,7 +163,7 @@ await using (var session = store.LightweightSession()) { var live = await session.Events.AggregateStreamAsync(pod); - Expect(live is { Bed: "garden", Peas: 1 }, "a strong-typed aggregate folds live"); + Expect(live is { Bed: "garden", Peas: 1 } && live.Id == new PodId(pod), "a strong-typed aggregate folds live and carries its id"); var stream = await session.Events.FetchForWriting(new PodId(pod)); Expect(stream.Aggregate is { Peas: 1 } && stream.CurrentVersion == 2, "FetchForWriting by a strong-typed id"); @@ -189,7 +189,7 @@ await using (var session = store.LightweightSession()) { var stream = await session.Events.FetchForWriting(shoal); - Expect(stream.Aggregate is { Fish: 2 }, + Expect(stream.Aggregate is { Fish: 2 } && stream.Aggregate.Id == new ShoalId(shoal), "a live-only strong-typed aggregate folds through its declared identity"); } diff --git a/src/Fisher.Tests/Events/strong_typed_aggregate_identity.cs b/src/Fisher.Tests/Events/strong_typed_aggregate_identity.cs new file mode 100644 index 0000000..34bba3e --- /dev/null +++ b/src/Fisher.Tests/Events/strong_typed_aggregate_identity.cs @@ -0,0 +1,158 @@ +using JasperFx; +using JasperFx.Events; +using JasperFx.Events.Projections; + +namespace Fisher.Tests.Events; + +/// +/// fisher#426 — a live-aggregated aggregate keyed on a strong-typed id gets the stream's identity. +/// +/// +/// +/// Live aggregation never consults the aggregate's own id, so an aggregate whose Create +/// does not set it is backfilled afterwards (AggregateIdentity.TrySetIdentity), as Marten +/// and Polecat do. That backfill assigned the stream id only when the member's type could hold it +/// as-is, and a is never a , so a wrapper was skipped and +/// the aggregate came back with a default id. The fold was right and the identity was not, with +/// nothing to say so. +/// +/// +/// Every path that backfills is covered, because they are separate call sites: live aggregation, +/// FetchForWriting by raw id and by wrapper, FetchManyForWriting and +/// ProjectLatest. The inline snapshot is covered too, because the issue asked whether it +/// was stamped, and it goes through the projection's identity setter rather than the backfill. +/// +/// +public class strong_typed_aggregate_identity +{ + private static CancellationToken Token => TestContext.Current.CancellationToken; + + private static DocumentStore GuidStore(TemporaryDatabase database, Action? configure = null) + => DocumentStore.For(options => + { + options.ConnectionString = database.ConnectionString; + options.AutoCreateSchemaObjects = AutoCreate.All; + configure?.Invoke(options); + }); + + private static async Task StartKilnAsync(DocumentStore store) + { + var id = Guid.NewGuid(); + await using var session = store.LightweightSession(); + session.Events.StartStream(id, new KilnFired(900), new KilnFired(1100)); + await session.SaveChangesAsync(Token); + return id; + } + + [Fact] + public async Task live_aggregation_stamps_a_guid_backed_wrapper() + { + using var database = TemporaryDatabase.Create("strong-live-id"); + await using var store = GuidStore(database); + var id = await StartKilnAsync(store); + + await using var query = store.QuerySession(); + var kiln = await query.Events.AggregateStreamAsync(id, token: Token); + + kiln.ShouldNotBeNull(); + kiln.Firings.ShouldBe(2); + kiln.Id.ShouldBe(new KilnId(id)); + } + + [Fact] + public async Task fetch_for_writing_stamps_a_wrapper_by_raw_id_and_by_wrapper() + { + using var database = TemporaryDatabase.Create("strong-fetch-id"); + await using var store = GuidStore(database); + var id = await StartKilnAsync(store); + + await using var session = store.LightweightSession(); + + (await session.Events.FetchForWriting(id, Token)).Aggregate!.Id.ShouldBe(new KilnId(id)); + (await session.Events.FetchForWriting(new KilnId(id), Token)).Aggregate!.Id + .ShouldBe(new KilnId(id)); + } + + [Fact] + public async Task fetch_many_for_writing_stamps_each_wrapper() + { + using var database = TemporaryDatabase.Create("strong-fetch-many-id"); + await using var store = GuidStore(database); + var first = await StartKilnAsync(store); + var second = await StartKilnAsync(store); + + await using var session = store.LightweightSession(); + var streams = await session.Events.FetchManyForWriting([first, second], Token); + + streams.Select(x => x.Aggregate!.Id).ShouldBe([new KilnId(first), new KilnId(second)]); + } + + [Fact] + public async Task project_latest_stamps_a_wrapper() + { + using var database = TemporaryDatabase.Create("strong-project-latest-id"); + await using var store = GuidStore(database); + var id = await StartKilnAsync(store); + + await using var session = store.LightweightSession(); + session.Events.Append(id, new KilnFired(1200)); + + var projected = await session.Events.ProjectLatest(id, Token); + + projected.ShouldNotBeNull(); + projected.Firings.ShouldBe(3); + projected.Id.ShouldBe(new KilnId(id)); + } + + [Fact] + public async Task live_aggregation_stamps_a_string_backed_wrapper() + { + using var database = TemporaryDatabase.Create("strong-live-key"); + await using var store = GuidStore(database, options => options.Events.StreamIdentity = StreamIdentity.AsString); + + await using (var session = store.LightweightSession()) + { + session.Events.StartStream("kettle-1", new KilnFired(100)); + await session.SaveChangesAsync(Token); + } + + await using var query = store.QuerySession(); + var kettle = await query.Events.AggregateStreamAsync("kettle-1", token: Token); + + kettle!.Id.ShouldBe(new KettleKey("kettle-1")); + } + + /// + /// The inline snapshot never had the bug — the projection's identity setter wraps through the + /// storage — and this pins that, since the issue could not say either way. + /// + [Fact] + public async Task an_inline_snapshot_carries_the_wrapper() + { + using var database = TemporaryDatabase.Create("strong-snapshot-id"); + await using var store = GuidStore(database, + options => options.Projections.Snapshot(SnapshotLifecycle.Inline)); + var id = await StartKilnAsync(store); + + await using var query = store.QuerySession(); + var kiln = await query.LoadAsync(new KilnId(id), Token); + + kiln!.Id.ShouldBe(new KilnId(id)); + } + + /// + /// A wrapper whose identity is set by Create to the stream id ends up with the same value + /// either way. What must not happen is a wrapper around the other identity type being "converted": + /// the backfill only ever assigns a wrapper around the stream id's own type, as the + /// FetchForWriting<T, TId> refusal does. + /// + [Fact] + public void a_wrapper_around_another_type_is_left_alone() + { + var kettle = new Kettle(); + + Fisher.Storage.AggregateIdentity.TrySetIdentity(kettle, Guid.NewGuid()); + + kettle.Id.ShouldBe(default); + } +} diff --git a/src/Fisher/Storage/AggregateIdentity.cs b/src/Fisher/Storage/AggregateIdentity.cs index 9db3c24..9ff0777 100644 --- a/src/Fisher/Storage/AggregateIdentity.cs +++ b/src/Fisher/Storage/AggregateIdentity.cs @@ -115,9 +115,19 @@ internal static Type ResolveIdType(Type aggregateType, StreamIdentity streamIden /// settable identity member of a compatible type. /// /// - /// Live aggregation folds events without ever consulting the aggregate's own id, so an - /// aggregate whose Create method does not set Id would otherwise come back with a - /// default one. Marten and Polecat both backfill it here rather than in the aggregator. + /// + /// Live aggregation folds events without ever consulting the aggregate's own id, so an + /// aggregate whose Create method does not set Id would otherwise come back with a + /// default one. Marten and Polecat both backfill it here rather than in the aggregator. + /// + /// + /// A strong-typed id is wrapped first (fisher#426). The stream id is a raw Guid or + /// string, which a wrapper member cannot hold as-is, so it used to be skipped and the aggregate + /// came back with a default id from live aggregation, FetchForWriting, + /// FetchManyForWriting and ProjectLatest alike. Only a wrapper around the stream + /// id's own type is built, so a Guid is never turned into a string-backed id or the reverse — + /// the rule FetchForWriting<T, TId> refuses by. + /// /// internal static void TrySetIdentity(object aggregate, object streamId) { @@ -125,16 +135,33 @@ internal static void TrySetIdentity(object aggregate, object streamId) switch (member) { - case PropertyInfo { CanWrite: true } property when property.PropertyType.IsInstanceOfType(streamId): - property.SetValue(aggregate, streamId); + case PropertyInfo { CanWrite: true } property when IdentityValueFor(property.PropertyType, streamId) is { } value: + property.SetValue(aggregate, value); break; - case FieldInfo { IsInitOnly: false } field when field.FieldType.IsInstanceOfType(streamId): - field.SetValue(aggregate, streamId); + case FieldInfo { IsInitOnly: false } field when IdentityValueFor(field.FieldType, streamId) is { } value: + field.SetValue(aggregate, value); break; } } + /// + /// The stream id as a value an identity member of can hold, or null. + /// + private static object? IdentityValueFor(Type memberType, object streamId) + { + if (memberType.IsInstanceOfType(streamId)) + { + return streamId; + } + + var target = Nullable.GetUnderlyingType(memberType) ?? memberType; + + return StrongTypedId.TryResolve(target, out var wrapper) && wrapper.SimpleType == streamId.GetType() + ? StrongTypedId.Wrap(wrapper, streamId) + : null; + } + /// /// Whether the aggregate type has opted out of single-stream identity with /// .