diff --git a/Directory.Build.props b/Directory.Build.props index c4b3269a0b..d481e33273 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -1,7 +1,7 @@ - 9.18.0 + 9.19.0 13.0 Jeremy D. Miller;Babu Annamalai;Jaedyn Tonee https://martendb.io/logo.png diff --git a/Directory.Packages.props b/Directory.Packages.props index 3e1723a672..f2cbd6590f 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -128,14 +128,20 @@ JasperFx 2.33.1: jasperfx#550/#551 (marten#5001) — ShardStateTracker.AssignedNodeNumber is stamped onto every published ShardState, so a distribution layer (Wolverine-managed subscription distribution) can drive the extended-progression running_on_node column. Marten's - WriteExtendedProgressionAsync already persists it; this bump makes the tracker seam available. --> - - - + WriteExtendedProgressionAsync already persists it; this bump makes the tracker seam available. + JasperFx 2.35.0: jasperfx#561 — new backwards-compatible virtual overload on the aggregation + projection base, RaiseSideEffects(TOperations operations, TId id, IEventSlice slice). + AggregationRunner now always invokes the 3-arg overload with slice.Id, so a projection can + recover the slice identity even when slice.Snapshot is null (e.g. a deleted MultiStreamProjection + slice) to emit a follow-on event or publish a message. The old 2-arg override still works; the + default 3-arg implementation delegates to it. --> + + + all runtime; build; native; contentfiles; analyzers; buildtransitive - + diff --git a/docs/events/projections/side-effects.md b/docs/events/projections/side-effects.md index 2836ff9193..0b2126937d 100644 --- a/docs/events/projections/side-effects.md +++ b/docs/events/projections/side-effects.md @@ -94,6 +94,90 @@ public partial class TripProjection: SingleStreamProjection snippet source | anchor +## Recovering the Slice Identity for Deleted Aggregates + +The `RaiseSideEffects()` override shown above receives the current `slice.Snapshot`, which is the most +convenient way to read the aggregate identity. That works well right up until the aggregate is _deleted_ in +the same batch that is raising side effects — at that point `slice.Snapshot` is `null` and there is no +document to read the identity from. This bites hardest in a `MultiStreamProjection`, where the slice groups +events drawn from many different streams: `slice.Events().First().StreamId` is one of the _member_ stream +ids, not the grouped aggregate identity, so there was previously no reliable way to recover the key of a +just-deleted slice in order to emit a follow-on event or publish a message. + +Starting with Marten 9.19 (JasperFx.Events 2.35.0) there is a second `RaiseSideEffects()` overload that adds +a strongly-typed `id` parameter carrying the slice identity — and it is populated even when +`slice.Snapshot == null`: + + + +```cs +// A running count of the active members of a team, aggregated across many +// separate per-member event streams by TeamId. +public class TeamRoster +{ + public Guid Id { get; set; } + public int MemberCount { get; set; } +} + +public record MemberJoinedTeam(Guid TeamId); +public record MemberLeftTeam(Guid TeamId); +public record TeamDisbanded(Guid TeamId); + +// The message we want to publish when a team is closed out +public record NotifyTeamClosed(Guid TeamId); + +public partial class TeamRosterProjection: MultiStreamProjection +{ + public TeamRosterProjection() + { + // All three event types are grouped by TeamId, which becomes the + // identity of each slice/aggregate. + Identity(x => x.TeamId); + Identity(x => x.TeamId); + Identity(x => x.TeamId); + } + + public void Apply(TeamRoster roster, MemberJoinedTeam _) => roster.MemberCount++; + public void Apply(TeamRoster roster, MemberLeftTeam _) => roster.MemberCount--; + + // When a team is disbanded the aggregate is deleted, so the snapshot handed to + // RaiseSideEffects below is null. + public bool ShouldDelete(TeamDisbanded _) => true; + + // NEW in JasperFx.Events 2.35.0: the second parameter hands you the identity + // of the current slice even when its snapshot has been deleted in this batch. + public override ValueTask RaiseSideEffects(IDocumentOperations operations, Guid id, + IEventSlice slice) + { + if (slice.Snapshot == null) + { + // The aggregate was deleted (TeamDisbanded -> ShouldDelete(...) == true), + // so there is no snapshot to read the identity from. Before this overload, + // there was no reliable way to recover the team identity here -- the slice + // groups events from many different streams, so slice.Events().First().StreamId + // is a *member* stream id, not the TeamId. The id parameter is the slice + // identity (the TeamId) regardless of whether the snapshot still exists. + slice.PublishMessage(new NotifyTeamClosed(id)); + return new ValueTask(); + } + + // Normal, non-deleted processing still has full access to the current snapshot + // (and, of course, to id)... + + return new ValueTask(); + } +} +``` +snippet source | anchor + + +The two overloads are fully backwards compatible: the original two-argument +`RaiseSideEffects(IDocumentOperations, IEventSlice)` still works exactly as before, and the default +implementation of the new three-argument overload simply delegates to it. Override the new +`RaiseSideEffects(IDocumentOperations operations, TId id, IEventSlice slice)` overload whenever you need +the aggregate identity independently of the snapshot — most commonly to react to a deletion — and keep using +the original overload for everything else. + A couple important facts about this new functionality: - The `RaiseSideEffects()` method is only called during _continuous_ asynchronous projection execution, and will not diff --git a/src/CoreTests/Bugs/Bug_5039_generic_secondary_store_marker.cs b/src/CoreTests/Bugs/Bug_5039_generic_secondary_store_marker.cs new file mode 100644 index 0000000000..e5d44d59b9 --- /dev/null +++ b/src/CoreTests/Bugs/Bug_5039_generic_secondary_store_marker.cs @@ -0,0 +1,68 @@ +using System; +using System.Threading.Tasks; +using Marten; +using Marten.Internal; +using Marten.Testing.Harness; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Xunit; + +namespace CoreTests.Bugs; + +/// +/// Regression test for #5039: registering a secondary store with a generic marker +/// interface threw because the closed generic CLR type name +/// contains a backtick + arity (e.g. IMartenStoreMarker`1), which is not a valid URI +/// hostname when composing the marten:// subject in SecondaryStoreConfig.Build. +/// +public class Bug_5039_generic_secondary_store_marker +{ + public sealed class MyContext; + public sealed class OtherContext; + + public interface IMartenStoreMarker : IDocumentStore; + + [Fact] + public void sanitized_uri_strips_backtick_and_includes_generic_argument() + { + var subject = SecondaryStoreConfig> + .SanitizeForUri(typeof(IMartenStoreMarker)); + + subject.ShouldNotContain("`"); + + // The result must compose into a valid URI + var uri = new Uri("marten://" + subject); + uri.Host.ShouldBe("imartenstoremarker-mycontext"); + + // Distinct closed generics must map to distinct subjects + var other = SecondaryStoreConfig> + .SanitizeForUri(typeof(IMartenStoreMarker)); + other.ShouldNotBe(subject); + } + + [Fact] + public async Task can_register_and_resolve_generic_marker_store() + { + using var host = await Host.CreateDefaultBuilder() + .ConfigureServices(services => + { + services.AddMarten(opts => + { + opts.Connection(ConnectionSource.ConnectionString); + opts.DatabaseSchemaName = "bug5039_primary"; + }); + + // This threw UriFormatException before the fix + services.AddMartenStore>(opts => + { + opts.Connection(ConnectionSource.ConnectionString); + opts.DatabaseSchemaName = "bug5039_ancillary"; + }); + }) + .StartAsync(); + + var store = host.Services.GetRequiredService>(); + store.ShouldNotBeNull(); + } +} diff --git a/src/EventSourcingTests/Aggregation/raise_side_effects_with_slice_id.cs b/src/EventSourcingTests/Aggregation/raise_side_effects_with_slice_id.cs new file mode 100644 index 0000000000..36d1d06375 --- /dev/null +++ b/src/EventSourcingTests/Aggregation/raise_side_effects_with_slice_id.cs @@ -0,0 +1,165 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using JasperFx.Core; +using JasperFx.Events; +using JasperFx.Events.Projections; +using Marten; +using Marten.Events; +using Marten.Events.Aggregation; +using Marten.Events.Projections; +using Marten.Internal.Sessions; +using Marten.Services; +using Marten.Testing.Harness; +using Shouldly; +using Xunit; + +namespace EventSourcingTests.Aggregation; + +public class raise_side_effects_with_slice_id: OneOffConfigurationsContext +{ + [Fact] + public async Task recovers_the_slice_identity_when_the_snapshot_was_deleted() + { + var outbox = new RecordingSliceIdOutbox(); + + StoreOptions(opts => + { + opts.Projections.Add(ProjectionLifecycle.Async); + opts.Events.MessageOutbox = outbox; + }); + + await theStore.Advanced.Clean.DeleteAllDocumentsAsync(); + await theStore.Advanced.Clean.DeleteAllEventDataAsync(); + + var daemon = await theStore.BuildProjectionDaemonAsync(); + await daemon.StartAllAsync(); + + var teamId = Guid.NewGuid(); + + // Each member joins from their OWN event stream. The projection groups all + // of them under the shared TeamId, so the slice identity (TeamId) is not the + // same as any single stream id. + theSession.Events.StartStream(Guid.NewGuid(), new MemberJoinedTeam(teamId)); + theSession.Events.StartStream(Guid.NewGuid(), new MemberJoinedTeam(teamId)); + await theSession.SaveChangesAsync(); + + await daemon.WaitForNonStaleData(30.Seconds()); + + var roster = await theSession.LoadAsync(teamId); + roster.MemberCount.ShouldBe(2); + + // Disband the team from yet another stream. This triggers DeleteEvent(), + // so the snapshot is null inside RaiseSideEffects. + theSession.Events.StartStream(Guid.NewGuid(), new TeamDisbanded(teamId)); + await theSession.SaveChangesAsync(); + + await daemon.WaitForNonStaleData(30.Seconds()); + + (await theSession.LoadAsync(teamId)).ShouldBeNull(); + + // The message was published for the deleted slice using the id parameter, + // even though slice.Snapshot was null. + outbox + .Batches + .SelectMany(x => x.Messages) + .Select(x => x.message) + .OfType() + .Single() + .TeamId.ShouldBe(teamId); + } +} + +#region sample_raise_side_effects_with_slice_id + +// A running count of the active members of a team, aggregated across many +// separate per-member event streams by TeamId. +public class TeamRoster +{ + public Guid Id { get; set; } + public int MemberCount { get; set; } +} + +public record MemberJoinedTeam(Guid TeamId); +public record MemberLeftTeam(Guid TeamId); +public record TeamDisbanded(Guid TeamId); + +// The message we want to publish when a team is closed out +public record NotifyTeamClosed(Guid TeamId); + +public partial class TeamRosterProjection: MultiStreamProjection +{ + public TeamRosterProjection() + { + // All three event types are grouped by TeamId, which becomes the + // identity of each slice/aggregate. + Identity(x => x.TeamId); + Identity(x => x.TeamId); + Identity(x => x.TeamId); + } + + public void Apply(TeamRoster roster, MemberJoinedTeam _) => roster.MemberCount++; + public void Apply(TeamRoster roster, MemberLeftTeam _) => roster.MemberCount--; + + // When a team is disbanded the aggregate is deleted, so the snapshot handed to + // RaiseSideEffects below is null. + public bool ShouldDelete(TeamDisbanded _) => true; + + // NEW in JasperFx.Events 2.35.0: the second parameter hands you the identity + // of the current slice even when its snapshot has been deleted in this batch. + public override ValueTask RaiseSideEffects(IDocumentOperations operations, Guid id, + IEventSlice slice) + { + if (slice.Snapshot == null) + { + // The aggregate was deleted (TeamDisbanded -> ShouldDelete(...) == true), + // so there is no snapshot to read the identity from. Before this overload, + // there was no reliable way to recover the team identity here -- the slice + // groups events from many different streams, so slice.Events().First().StreamId + // is a *member* stream id, not the TeamId. The id parameter is the slice + // identity (the TeamId) regardless of whether the snapshot still exists. + slice.PublishMessage(new NotifyTeamClosed(id)); + return new ValueTask(); + } + + // Normal, non-deleted processing still has full access to the current snapshot + // (and, of course, to id)... + + return new ValueTask(); + } +} + +#endregion + +public class RecordingSliceIdOutbox: IMessageOutbox +{ + public readonly List Batches = new(); + + public ValueTask CreateBatch(DocumentSessionBase session) + { + var batch = new RecordingSliceIdBatch(); + Batches.Add(batch); + return new ValueTask(batch); + } +} + +public record SliceIdTenantMessage(string tenantId, object message); + +public class RecordingSliceIdBatch: IMessageBatch +{ + public readonly List Messages = new(); + + public Task AfterCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token) + => Task.CompletedTask; + + public Task BeforeCommitAsync(IDocumentSession session, IChangeSet commit, CancellationToken token) + => Task.CompletedTask; + + public ValueTask PublishAsync(T message, string tenantId) + { + Messages.Add(new SliceIdTenantMessage(tenantId, message)); + return new ValueTask(); + } +} diff --git a/src/Marten/Internal/SecondaryStoreConfig.cs b/src/Marten/Internal/SecondaryStoreConfig.cs index d6281a780b..bd0f39492d 100644 --- a/src/Marten/Internal/SecondaryStoreConfig.cs +++ b/src/Marten/Internal/SecondaryStoreConfig.cs @@ -88,8 +88,30 @@ public T Build(IServiceProvider provider) // the proxy only needs a forwarding constructor. var storeType = SecondaryStoreProxyFactory.GetOrCreate(typeof(T)); var store = (T)Activator.CreateInstance(storeType, options)!; - store.As().Subject = new Uri("marten://" + typeof(T).Name.ToLowerInvariant()); + store.As().Subject = new Uri("marten://" + SanitizeForUri(typeof(T))); return store; } + + // #5039: a closed generic marker interface (e.g. IMartenStoreMarker) has a + // CLR type name containing a backtick and arity ("IMartenStoreMarker`1"), which is not a + // valid URI hostname and throws UriFormatException. Strip the arity and fold in the + // (sanitized) generic argument names so distinct closed generics still map to distinct URIs. + internal static string SanitizeForUri(Type type) + { + var name = type.Name; + var tick = name.IndexOf('`'); + if (tick >= 0) + { + name = name.Substring(0, tick); + } + + if (type.IsGenericType) + { + var arguments = type.GetGenericArguments().Select(SanitizeForUri); + name = name + "-" + string.Join("-", arguments); + } + + return name.ToLowerInvariant(); + } }