Conversion to Channels in place of TPL DataFlow

build/build.csproj

@@ -12,7 +12,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="JasperFx" Version="1.0.0" />
+    <PackageReference Include="JasperFx" Version="1.6.0-alpha-4" />
     <PackageReference Include="Npgsql" Version="8.0.3" />
     <PackageReference Include="Nuke.Common" Version="8.0.0" />
   </ItemGroup>

src/DaemonTests/Resiliency/when_skipping_events_in_daemon.cs

@@ -135,14 +135,14 @@ public async Task see_the_dead_letter_events()
         theSession.Logger = new TestOutputMartenLogger(_output);
         var skipped = await theSession.Query<DeadLetterEvent>().ToListAsync();
 
-        skipped.Where(x => x.ProjectionName == "CollateNames" && x.ShardName == "All")
-            .Select(x => x.EventSequence).OrderBy(x => x)
-            .ShouldHaveTheSameElementsAs(4, 5, 6, 7);
-
         skipped.Where(x => x.ProjectionName == "NamedDocuments" && x.ShardName == "All")
             .Select(x => x.EventSequence).OrderBy(x => x)
             .ShouldHaveTheSameElementsAs(4, 5, 6, 7, 11, 14);
 
+        skipped.Where(x => x.ProjectionName == "CollateNames" && x.ShardName == "All")
+            .Select(x => x.EventSequence).OrderBy(x => x)
+            .ShouldHaveTheSameElementsAs(4, 5, 6, 7);
+
     }
 }
 

src/LinqTestsTypes/LinqTestsTypes.csproj

@@ -14,6 +14,6 @@
 
     <ItemGroup>
       <PackageReference Include="FSharp.Core" Version="9.0.100" />
-      <PackageReference Include="JasperFx" Version="1.0.0" />
+      <PackageReference Include="JasperFx" Version="1.6.0-alpha-4" />
     </ItemGroup>
 </Project>

src/Marten/DocumentStore.EventStore.cs

@@ -206,11 +206,11 @@ public async ValueTask<IProjectionBatch<IDocumentOperations, IQuerySession>> Sta
         var projectionBatch = new ProjectionBatch(session, batch, mode);
         if (range.SequenceFloor == 0)
         {
-            batch.Queue.Post(new InsertProjectionProgress(session.Options.EventGraph, range));
+            await batch.Queue.PostAsync(new InsertProjectionProgress(session.Options.EventGraph, range)).ConfigureAwait(false);
         }
         else
         {
-            batch.Queue.Post(new UpdateProjectionProgress(session.Options.EventGraph, range));
+            await batch.Queue.PostAsync(new UpdateProjectionProgress(session.Options.EventGraph, range)).ConfigureAwait(false);
         }
 
         return projectionBatch;
@@ -285,7 +285,7 @@ async Task ISubscriptionRunner<ISubscription>.ExecuteAsync(ISubscription subscri
         };;
 
         // Mark the progression
-        batch.Queue.Post(range.BuildProgressionOperation(Events));
+        await batch.Queue.PostAsync(range.BuildProgressionOperation(Events)).ConfigureAwait(false);
 
         await using var session = new ProjectionDocumentSession(this, batch,
             new SessionOptions

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

@@ -1,10 +1,9 @@
 using System;
-using System.Collections;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using System.Threading.Tasks.Dataflow;
+using JasperFx.Blocks;
 using JasperFx.Core;
 using JasperFx.Events;
 using JasperFx.Events.Daemon;
@@ -17,7 +16,6 @@
 using Marten.Services;
 
 namespace Marten.Events.Daemon.Internals;
-#nullable enable
 
 /// <summary>
 ///     Incrementally built batch command for projection updates
@@ -26,52 +24,63 @@ public class ProjectionUpdateBatch: IUpdateBatch, IAsyncDisposable, IDisposable,
 {
     private readonly List<Type> _documentTypes = new();
     private readonly List<OperationPage> _pages = new();
+
+    private readonly List<IStorageOperation> _patches = new();
+    private readonly SemaphoreSlim _semaphore = new(1, 1);
     private readonly ProjectionOptions _settings;
     private readonly CancellationToken _token;
+
+    private IMessageBatch? _batch;
     private OperationPage? _current;
     private DocumentSessionBase? _session;
 
-    private IMartenSession Session
-    {
-        get => _session ?? throw new InvalidOperationException("Session already released");
-    }
-
-    public List<IChangeListener> Listeners { get; } = new();
-
-    public ShardExecutionMode Mode { get; }
-
-    public bool ShouldApplyListeners { get; set; }
-
     internal ProjectionUpdateBatch(ProjectionOptions settings,
         DocumentSessionBase? session, ShardExecutionMode mode, CancellationToken token)
     {
         _settings = settings;
         _session = session ?? throw new ArgumentNullException(nameof(session));
         _token = token;
         Mode = mode;
-        Queue = new ActionBlock<IStorageOperation>(processOperation,
-            new ExecutionDataflowBlockOptions
-            {
-                MaxDegreeOfParallelism = 1, EnsureOrdered = true, CancellationToken = token
-            });
+
+        Queue = new Block<IStorageOperation>(processOperationAsync);
 
         startNewPage(session);
     }
 
-    public async Task WaitForCompletion()
+    private IMartenSession Session => _session ?? throw new InvalidOperationException("Session already released");
+
+    public List<IChangeListener> Listeners { get; } = new();
+
+    public ShardExecutionMode Mode { get; }
+
+    public bool ShouldApplyListeners { get; set; }
+
+    // TODO -- make this private
+    public Block<IStorageOperation> Queue { get; }
+
+    public async ValueTask DisposeAsync()
     {
         Queue.Complete();
 
-        await Queue.Completion.ConfigureAwait(false);
-        foreach (var patch in _patches)
+        await Queue.DisposeAsync().ConfigureAwait(false);
+
+        foreach (var page in _pages) page.ReleaseSession();
+
+        if (_session != null)
         {
-            applyOperation(patch);
+            await _session.DisposeAsync().ConfigureAwait(true);
+            _session = null;
         }
-    }
 
-    // TODO -- make this private
-    public ActionBlock<IStorageOperation> Queue { get; }
+        Dispose(false);
+        GC.SuppressFinalize(this);
+    }
 
+    public void Dispose()
+    {
+        Dispose(true);
+        GC.SuppressFinalize(this);
+    }
 
 
     IEnumerable<IDeletion> IUnitOfWork.Deletions()
@@ -205,7 +214,7 @@ public void EjectAll()
         throw new NotSupportedException();
     }
 
-    public void PurgeOperations<T, TId>(TId id) where T : notnull where TId: notnull
+    public void PurgeOperations<T, TId>(TId id) where T : notnull where TId : notnull
     {
         // Do nothing here
     }
@@ -223,7 +232,10 @@ public async Task PostUpdateAsync(IMartenSession session)
         }
 
         var listeners = _settings.AsyncListeners.Concat(Listeners).ToArray();
-        if (listeners.Length == 0) return;
+        if (listeners.Length == 0)
+        {
+            return;
+        }
 
         var unitOfWorkData = new UnitOfWork(_pages.SelectMany(x => x.Operations));
         foreach (var listener in listeners)
@@ -241,7 +253,10 @@ public async Task PreUpdateAsync(IMartenSession session)
         }
 
         var listeners = _settings.AsyncListeners.Concat(Listeners).ToArray();
-        if (listeners.Length == 0) return;
+        if (listeners.Length == 0)
+        {
+            return;
+        }
 
         var unitOfWorkData = new UnitOfWork(_pages.SelectMany(x => x.Operations));
         foreach (var listener in listeners)
@@ -251,30 +266,52 @@ await listener.BeforeCommitAsync((IDocumentSession)session, unitOfWorkData, _tok
         }
     }
 
+    public IReadOnlyList<OperationPage> BuildPages(IMartenSession session)
+    {
+        if (_token.IsCancellationRequested)
+        {
+            return Array.Empty<OperationPage>();
+        }
+
+        // Guard against empty batches
+        return _pages.Where(x => x.Operations.Any()).ToList();
+    }
+
+    public async Task WaitForCompletion()
+    {
+        await Queue.WaitForCompletionAsync().ConfigureAwait(false);
+
+        foreach (var patch in _patches) applyOperation(patch);
+    }
+
     private void startNewPage(IMartenSession session)
     {
         if (_token.IsCancellationRequested)
+        {
             return;
+        }
 
         _current = new OperationPage(session);
         _pages.Add(_current);
     }
 
-    private readonly List<IStorageOperation> _patches = new();
-
-    private void processOperation(IStorageOperation operation)
+    private Task processOperationAsync(IStorageOperation operation, CancellationToken _)
     {
         if (_token.IsCancellationRequested)
-            return;
+        {
+            return Task.CompletedTask;
+        }
 
         // If there's one patch, then everything needs to be queued up for later
         if (operation is PatchOperation || _patches.Any())
         {
             _patches.Add(operation);
-            return;
+            return Task.CompletedTask;
         }
 
         applyOperation(operation);
+
+        return Task.CompletedTask;
     }
 
     private void applyOperation(IStorageOperation operation)
@@ -289,45 +326,18 @@ private void applyOperation(IStorageOperation operation)
         }
     }
 
-    public IReadOnlyList<OperationPage> BuildPages(IMartenSession session)
-    {
-        if (_token.IsCancellationRequested)
-        {
-            return Array.Empty<OperationPage>();
-        }
-
-        // Guard against empty batches
-        return _pages.Where(x => x.Operations.Any()).ToList();
-    }
-
-
-    public ValueTask CloseSession() => DisposeAsync();
-
-    public void Dispose()
-    {
-        Dispose(disposing: true);
-        GC.SuppressFinalize(this);
-    }
 
-    public async ValueTask DisposeAsync()
+    public ValueTask CloseSession()
     {
-        Queue.Complete();
-        foreach (var page in _pages) page.ReleaseSession();
-
-        if (_session != null)
-        {
-            await _session.DisposeAsync().ConfigureAwait(true);
-            _session = null;
-        }
-
-        Dispose(disposing: false);
-        GC.SuppressFinalize(this);
+        return DisposeAsync();
     }
 
     protected void Dispose(bool disposing)
     {
         if (!disposing)
+        {
             return;
+        }
 
         Queue.Complete();
 
@@ -338,16 +348,19 @@ protected void Dispose(bool disposing)
         _session = null;
     }
 
-    private IMessageBatch? _batch;
-    private readonly SemaphoreSlim _semaphore = new(1, 1);
-
     public async ValueTask<IMessageBatch> CurrentMessageBatch(DocumentSessionBase session)
     {
-        if (_batch != null) return _batch;
+        if (_batch != null)
+        {
+            return _batch;
+        }
 
         await _semaphore.WaitAsync(_token).ConfigureAwait(false);
 
-        if (_batch != null) return _batch;
+        if (_batch != null)
+        {
+            return _batch;
+        }
 
         try
         {
@@ -361,6 +374,4 @@ public async ValueTask<IMessageBatch> CurrentMessageBatch(DocumentSessionBase se
             _semaphore.Release();
         }
     }
-
-
 }

src/Marten/Internal/Sessions/AutoClosingLifetime.cs

@@ -3,6 +3,7 @@
 using System.Collections.Generic;
 using System.Data;
 using System.Data.Common;
+using System.Diagnostics;
 using System.Linq;
 using System.Runtime.ExceptionServices;
 using System.Threading;
@@ -295,6 +296,15 @@ public async Task ExecuteBatchPagesAsync(IReadOnlyList<OperationPage> pages, Lis
                 foreach (var page in pages)
                 {
                     var batch = page.Compile();
+
+                    foreach (var batchCommand in batch.BatchCommands)
+                    {
+                        if (batchCommand.CommandText.IsEmpty())
+                        {
+                            Debug.WriteLine("what the hell?");
+                        }
+                    }
+
                     batch.Timeout = CommandTimeout;
                     batch.Connection = conn;
                     batch.Transaction = tx;