Fix event load exceptions causing rebuilds to finish early

JasperFx · Jul 9, 2024 · 3aa7aa9 · 3aa7aa9
1 parent 499dd03
commit 3aa7aa9
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 7 deletions.
diff --git a/src/Marten/Events/Daemon/Internals/EventLoader.cs b/src/Marten/Events/Daemon/Internals/EventLoader.cs
@@ -1,3 +1,4 @@
+using System;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -14,7 +15,7 @@
 
 namespace Marten.Events.Daemon.Internals;
 
-internal class EventLoader: IEventLoader
+internal sealed class EventLoader: IEventLoader
 {
     private readonly int _aggregateIndex;
     private readonly int _batchSize;
@@ -23,7 +24,6 @@ internal class EventLoader: IEventLoader
     private readonly NpgsqlParameter _floor;
     private readonly IEventStorage _storage;
     private readonly IDocumentStore _store;
-
     public EventLoader(DocumentStore store, MartenDatabase database, AsyncProjectionShard shard, AsyncOptions options) : this(store, database, options, shard.BuildFilters(store).ToArray())
     {
 
@@ -80,6 +80,8 @@ public async Task<EventPage> LoadAsync(EventRequest request,
         _floor.Value = request.Floor;
         _ceiling.Value = request.HighWater;
 
+        var skippedEvents = 0;
+
         await using var reader = await session.ExecuteReaderAsync(_command, token).ConfigureAwait(false);
         while (await reader.ReadAsync(token).ConfigureAwait(false))
         {
@@ -100,7 +102,8 @@ public async Task<EventPage> LoadAsync(EventRequest request,
             {
                 if (request.ErrorOptions.SkipUnknownEvents)
                 {
-                    request.Runtime.Logger.LogWarning("Skipping unknown event type '{EventTypeName}'", e.EventTypeName);
+                    request.Runtime.Logger.EventUnknown(e.EventTypeName);
+                    skippedEvents++;
                 }
                 else
                 {
@@ -112,7 +115,10 @@ public async Task<EventPage> LoadAsync(EventRequest request,
             {
                 if (request.ErrorOptions.SkipSerializationErrors)
                 {
+                    request.Runtime.Logger.EventDeserializationException(e.InnerException!.GetType().Name!, e.Sequence);
+                    request.Runtime.Logger.EventDeserializationExceptionDebug(e);
                     await request.Runtime.RecordDeadLetterEventAsync(e.ToDeadLetterEvent(request.Name)).ConfigureAwait(false);
+                    skippedEvents++;
                 }
                 else
                 {
@@ -122,8 +128,23 @@ public async Task<EventPage> LoadAsync(EventRequest request,
             }
         }
 
-        page.CalculateCeiling(_batchSize, request.HighWater);
+        page.CalculateCeiling(_batchSize, request.HighWater, skippedEvents);
 
         return page;
     }
+
+}
+
+internal static partial class Log
+{
+    [LoggerMessage(LogLevel.Warning, "Skipping unknown event type '{EventTypeName}'")]
+    public static partial void EventUnknown(this ILogger logger, string eventTypeName);
+
+    [LoggerMessage(LogLevel.Warning,"Suppressed Serialization exception of type {ExceptionName} occured whilst loading event at sequence {Sequence}. Enable debug logging or disable SkipSerializationErrors for full stack trace.")]
+    public static partial void EventDeserializationException(this ILogger logger, string exceptionName, long sequence);
+
+    [LoggerMessage(LogLevel.Debug)]
+    public static partial void EventDeserializationExceptionDebug(this ILogger logger, Exception exception);
 }
+
+
diff --git a/src/Marten/Events/Daemon/Internals/EventPage.cs b/src/Marten/Events/Daemon/Internals/EventPage.cs
@@ -13,9 +13,9 @@ public EventPage(long floor)
     public long Floor { get; }
     public long Ceiling { get; private set; }
 
-    public void CalculateCeiling(int batchSize, long highWaterMark)
+    public void CalculateCeiling(int batchSize, long highWaterMark, int skippedEvents)
     {
-        Ceiling = Count == batchSize
+        Ceiling = (Count + skippedEvents) == batchSize
             ? this.Last().Sequence
             : highWaterMark;
     }

diff --git a/src/Marten/Events/Daemon/Internals/SubscriptionAgent.cs b/src/Marten/Events/Daemon/Internals/SubscriptionAgent.cs
@@ -157,7 +157,7 @@ public async Task StartAsync(SubscriptionExecutionRequest request)
     public async Task ReplayAsync(SubscriptionExecutionRequest request, long highWaterMark, TimeSpan timeout)
     {
         Mode = ShardExecutionMode.Rebuild;
-        _rebuild = new TaskCompletionSource();
+        _rebuild = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         _execution.Mode = ShardExecutionMode.Rebuild;
         ErrorOptions = request.ErrorHandling;
         _runtime = request.Runtime;