Release 9.19.0 — adopt JasperFx 2.35.0, RaiseSideEffects slice-id docs, fix #5039

Directory.Build.props

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
 <Project>
     <PropertyGroup>
-        <Version>9.18.0</Version>
+        <Version>9.19.0</Version>
         <LangVersion>13.0</LangVersion>
         <Authors>Jeremy D. Miller;Babu Annamalai;Jaedyn Tonee</Authors>
         <PackageIconUrl>https://martendb.io/logo.png</PackageIconUrl>

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. -->
-    <PackageVersion Include="JasperFx" Version="2.34.0" />
-    <PackageVersion Include="JasperFx.Events" Version="2.34.0" />
-    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.34.0">
+         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<TDoc> 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. -->
+    <PackageVersion Include="JasperFx" Version="2.35.0" />
+    <PackageVersion Include="JasperFx.Events" Version="2.35.0" />
+    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.35.0">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageVersion>
-    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.34.0" />
+    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.35.0" />
     <PackageVersion Include="Jil" Version="3.0.0-alpha2" />
     <PackageVersion Include="Lamar" Version="7.1.1" />
     <PackageVersion Include="Lamar.Microsoft.DependencyInjection" Version="15.0.0" />

docs/events/projections/side-effects.md

@@ -94,6 +94,90 @@ public partial class TripProjection: SingleStreamProjection<Trip, Guid>
 <sup><a href='https://github.com/JasperFx/marten/blob/master/src/EventSourcingTests/Examples/TripProjectionWithEventMetadata.cs#L31-L98' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_aggregation_using_event_metadata' title='Start of snippet'>anchor</a></sup>
 <!-- endSnippet -->
 
+## Recovering the Slice Identity for Deleted Aggregates <Badge type="tip" text="9.19" />
+
+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`:
+
+<!-- snippet: sample_raise_side_effects_with_slice_id -->
+<a id='snippet-sample_raise_side_effects_with_slice_id'></a>
+```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<TeamRoster, Guid>
+{
+    public TeamRosterProjection()
+    {
+        // All three event types are grouped by TeamId, which becomes the
+        // identity of each slice/aggregate.
+        Identity<MemberJoinedTeam>(x => x.TeamId);
+        Identity<MemberLeftTeam>(x => x.TeamId);
+        Identity<TeamDisbanded>(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<TeamRoster> 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();
+    }
+}
+```
+<sup><a href='https://github.com/JasperFx/marten/blob/master/src/EventSourcingTests/Aggregation/raise_side_effects_with_slice_id.cs#L75-L134' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_raise_side_effects_with_slice_id' title='Start of snippet'>anchor</a></sup>
+<!-- endSnippet -->
+
+The two overloads are fully backwards compatible: the original two-argument
+`RaiseSideEffects(IDocumentOperations, IEventSlice<TDoc>)` 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<TDoc> 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

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;
+
+/// <summary>
+/// Regression test for #5039: registering a secondary store with a <em>generic</em> marker
+/// interface threw <see cref="UriFormatException"/> because the closed generic CLR type name
+/// contains a backtick + arity (e.g. <c>IMartenStoreMarker`1</c>), which is not a valid URI
+/// hostname when composing the <c>marten://</c> subject in <c>SecondaryStoreConfig.Build</c>.
+/// </summary>
+public class Bug_5039_generic_secondary_store_marker
+{
+    public sealed class MyContext;
+    public sealed class OtherContext;
+
+    public interface IMartenStoreMarker<TContext> : IDocumentStore;
+
+    [Fact]
+    public void sanitized_uri_strips_backtick_and_includes_generic_argument()
+    {
+        var subject = SecondaryStoreConfig<IMartenStoreMarker<MyContext>>
+            .SanitizeForUri(typeof(IMartenStoreMarker<MyContext>));
+
+        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<IMartenStoreMarker<OtherContext>>
+            .SanitizeForUri(typeof(IMartenStoreMarker<OtherContext>));
+        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<IMartenStoreMarker<MyContext>>(opts =>
+                {
+                    opts.Connection(ConnectionSource.ConnectionString);
+                    opts.DatabaseSchemaName = "bug5039_ancillary";
+                });
+            })
+            .StartAsync();
+
+        var store = host.Services.GetRequiredService<IMartenStoreMarker<MyContext>>();
+        store.ShouldNotBeNull();
+    }
+}

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<TeamRosterProjection>(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<TeamRoster>(teamId);
+        roster.MemberCount.ShouldBe(2);
+
+        // Disband the team from yet another stream. This triggers DeleteEvent<TeamDisbanded>(),
+        // 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<TeamRoster>(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<NotifyTeamClosed>()
+            .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<TeamRoster, Guid>
+{
+    public TeamRosterProjection()
+    {
+        // All three event types are grouped by TeamId, which becomes the
+        // identity of each slice/aggregate.
+        Identity<MemberJoinedTeam>(x => x.TeamId);
+        Identity<MemberLeftTeam>(x => x.TeamId);
+        Identity<TeamDisbanded>(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<TeamRoster> 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<RecordingSliceIdBatch> Batches = new();
+
+    public ValueTask<IMessageBatch> CreateBatch(DocumentSessionBase session)
+    {
+        var batch = new RecordingSliceIdBatch();
+        Batches.Add(batch);
+        return new ValueTask<IMessageBatch>(batch);
+    }
+}
+
+public record SliceIdTenantMessage(string tenantId, object message);
+
+public class RecordingSliceIdBatch: IMessageBatch
+{
+    public readonly List<SliceIdTenantMessage> 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>(T message, string tenantId)
+    {
+        Messages.Add(new SliceIdTenantMessage(tenantId, message));
+        return new ValueTask();
+    }
+}

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<DocumentStore>().Subject = new Uri("marten://" + typeof(T).Name.ToLowerInvariant());
+        store.As<DocumentStore>().Subject = new Uri("marten://" + SanitizeForUri(typeof(T)));
 
         return store;
     }
+
+    // #5039: a closed generic marker interface (e.g. IMartenStoreMarker<MyContext>) 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();
+    }
 }