Inline side effects, ameliorations for concurrent access to messaging in side effect processing

Author: jeremydmiller

src/Marten/Events/Aggregation/AggregationRuntime.cs

@@ -84,6 +84,13 @@ public AggregationRuntime(IDocumentStore store, IAggregateProjection projection,
     public IEventSlicer<TDoc, TId> Slicer { get; }
     public IDocumentStorage<TDoc, TId> Storage { get; }
 
+    private bool shouldProcessSideEffects(DocumentSessionBase session, ProjectionLifecycle lifecycle)
+    {
+        if (session is ProjectionDocumentSession { Mode: ShardExecutionMode.Continuous }) return true;
+
+        return lifecycle == ProjectionLifecycle.Inline && session.Options.Events.EnableSideEffectsOnInlineProjections;
+    }
+
     public async ValueTask ApplyChangesAsync(DocumentSessionBase session,
         EventSlice<TDoc, TId> slice, CancellationToken cancellation,
         ProjectionLifecycle lifecycle = ProjectionLifecycle.Inline)
@@ -94,7 +101,7 @@ public async ValueTask ApplyChangesAsync(DocumentSessionBase session,
         {
             var operation = Storage.DeleteForId(slice.Id, slice.Tenant.TenantId);
 
-            if (session is ProjectionDocumentSession { Mode: ShardExecutionMode.Continuous })
+            if (shouldProcessSideEffects(session, lifecycle))
             {
                 await processPossibleSideEffects(session, slice).ConfigureAwait(false);
             }
@@ -148,7 +155,7 @@ public async ValueTask ApplyChangesAsync(DocumentSessionBase session,
 
         maybeArchiveStream(session, slice);
 
-        if (session is ProjectionDocumentSession { Mode: ShardExecutionMode.Continuous })
+        if (shouldProcessSideEffects(session, lifecycle))
         {
             // Need to set the aggregate in case it didn't exist upfront
             slice.Aggregate = aggregate;

src/Marten/Events/Aggregation/Rebuilds/AggregatePageHandler.cs

@@ -153,6 +153,8 @@ public async Task ProcessPageAsync(IDaemonRuntime runtime, ShardName shardName,
             ShouldApplyListeners = false
         };
 
+        await batch.InitializeMessageBatch().ConfigureAwait(false);
+
         // Gotta use the current tenant if using conjoined tenancy
         var sessionOptions = SessionOptions.ForDatabase(_session.TenantId, _session.Database);
 

src/Marten/Events/Aggregation/SideEffectCode.cs

@@ -10,7 +10,7 @@ public interface IMessageSink
     ValueTask PublishAsync<T>(T message);
 }
 
-public interface ITenantedMessageSink
+public interface ITenantedMessageSink : IMessageSink
 {
     ValueTask PublishAsync<T>(T message, string tenantId);
 }

src/Marten/Events/Daemon/Internals/GroupedProjectionExecution.cs

@@ -220,6 +220,8 @@ private async Task<ProjectionUpdateBatch> buildBatchAsync(EventRangeGroup group,
                 ShouldApplyListeners = group.Agent.Mode == ShardExecutionMode.Continuous && group.Range.Events.Any()
             };
 
+            await batch.InitializeMessageBatch().ConfigureAwait(false);
+
             // Mark the progression
             batch.Queue.Post(group.Range.BuildProgressionOperation(_store.Events));
 

src/Marten/Events/Daemon/Internals/ProjectionUpdateBatch.cs

@@ -25,7 +25,7 @@ public class ProjectionUpdateBatch: IUpdateBatch, IAsyncDisposable, IDisposable,
     private readonly DaemonSettings _settings;
     private readonly CancellationToken _token;
     private OperationPage? _current;
-    private DocumentSessionBase? _session;
+    private DocumentSessionBase _session;
 
     private IMartenSession Session
     {
@@ -41,6 +41,7 @@ private IMartenSession Session
     internal ProjectionUpdateBatch(DaemonSettings settings,
         DocumentSessionBase? session, ShardExecutionMode mode, CancellationToken token)
     {
+
         _settings = settings;
         _session = session ?? throw new ArgumentNullException(nameof(session));
         _token = token;
@@ -54,6 +55,12 @@ internal ProjectionUpdateBatch(DaemonSettings settings,
         startNewPage(session);
     }
 
+    public async Task InitializeMessageBatch()
+    {
+        _batch = await _session!.Options.Events.MessageOutbox.CreateBatch(_session).ConfigureAwait(false);
+        Listeners.Add(_batch);
+    }
+
     public async Task WaitForCompletion()
     {
         Queue.Complete();
@@ -335,25 +342,8 @@ protected void Dispose(bool disposing)
     }
 
     private IMessageBatch? _batch;
-    private readonly SemaphoreSlim _semaphore = new(1, 1);
-    public async ValueTask<IMessageBatch> CurrentMessageBatch(DocumentSessionBase session)
+    public ValueTask<IMessageBatch> CurrentMessageBatch(DocumentSessionBase session)
     {
-        if (_batch != null) return _batch;
-
-        await _semaphore.WaitAsync(_token).ConfigureAwait(false);
-
-        if (_batch != null) return _batch;
-
-        try
-        {
-            _batch = await _session.Options.Events.MessageOutbox.CreateBatch(session).ConfigureAwait(false);
-            Listeners.Add(_batch);
-
-            return _batch;
-        }
-        finally
-        {
-            _semaphore.Release();
-        }
+        return new ValueTask<IMessageBatch>(_batch);
     }
 }

src/Marten/Events/EventGraph.cs

@@ -115,6 +115,12 @@ internal EventGraph(StoreOptions options)
     /// </summary>
     public bool EnableUniqueIndexOnEventId { get; set; } = false;
 
+    /// <summary>
+    /// Opt into having Marten process "side effects" on aggregation projections (SingleStreamProjection/MultiStreamProjection) while
+    /// running in an Inline lifecycle. Default is false;
+    /// </summary>
+    public bool EnableSideEffectsOnInlineProjections { get; set; } = false;
+
     /// <summary>
     ///     Configure whether event streams are identified with Guid or strings
     /// </summary>

src/Marten/Events/IEventStoreOptions.cs

@@ -106,6 +106,12 @@ bool UseIdentityMapForInlineAggregates
         /// </summary>
         public bool UseMandatoryStreamTypeDeclaration { get; set; }
 
+        /// <summary>
+        /// Opt into having Marten process "side effects" on aggregation projections (SingleStreamProjection/MultiStreamProjection) while
+        /// running in an Inline lifecycle. Default is false;
+        /// </summary>
+        bool EnableSideEffectsOnInlineProjections { get; set; }
+
         /// <summary>
         ///     Register an event type with Marten. This isn't strictly necessary for normal usage,
         ///     but can help Marten with asynchronous projections where Marten hasn't yet encountered

src/Marten/Events/IReadOnlyEventStoreOptions.cs

@@ -83,4 +83,10 @@ public interface IReadOnlyEventStoreOptions
     /// but this will be true in 8.0
     /// </summary>
     bool UseMandatoryStreamTypeDeclaration { get; set; }
+
+    /// <summary>
+    /// Opt into having Marten process "side effects" on aggregation projections (SingleStreamProjection/MultiStreamProjection) while
+    /// running in an Inline lifecycle. Default is false;
+    /// </summary>
+    public bool EnableSideEffectsOnInlineProjections { get; set; }
 }

src/Marten/Events/Projections/ProjectionOptions.cs

@@ -356,11 +356,6 @@ public void Add<TProjection>(
     )
         where TProjection : GeneratedProjection, new()
     {
-        if (lifecycle == ProjectionLifecycle.Live)
-        {
-            throw new InvalidOperationException("The generic overload of Add does not support Live projections, please use the non-generic overload.");
-        }
-
         var projection = new TProjection { Lifecycle = lifecycle };
 
         asyncConfiguration?.Invoke(projection.Options);

src/Marten/Internal/Sessions/DocumentSessionBase.SaveChanges.cs

@@ -104,12 +104,17 @@ public async Task SaveChangesAsync(CancellationToken token = default)
 
         await ExecuteBatchAsync(batch, token).ConfigureAwait(false);
 
+        if (_messageBatch != null)
+        {
+            await _messageBatch.AfterCommitAsync(this, _workTracker, token).ConfigureAwait(false);
+        }
+
         resetDirtyChecking();
 
         EjectPatchedTypes(_workTracker);
         Logger.RecordSavedChanges(this, _workTracker);
 
-        foreach (var listener in Listeners)
+        foreach (var listener in Listeners.ToArray())
         {
             await listener.AfterCommitAsync(this, _workTracker, token).ConfigureAwait(false);
         }