Fix stale reads in MongoDB Sink during bulk mode catch-up

Author: einari

Source/Kernel/Core/Namespaces/NamespacesReactor.cs

@@ -3,31 +3,50 @@
 
 using Cratis.Chronicle.Concepts.Events;
 using Cratis.Chronicle.Concepts.EventSequences;
+using Cratis.Chronicle.Concepts.Seeding;
 using Cratis.Chronicle.Observation.Reactors.Kernel;
-
-namespace Cratis.Chronicle.Namespaces;
+using Cratis.Chronicle.Seeding;
 
 #pragma warning disable IDE0060 // Remove unused parameter
 
+namespace Cratis.Chronicle.Namespaces;
+
 /// <summary>
 /// Represents a reactor that handles namespace events.
 /// </summary>
+/// <param name="grainFactory">The <see cref="IGrainFactory"/> for creating grains.</param>
 [Reactor(eventSequence: WellKnownEventSequences.System, systemEventStoreOnly: true)]
-public class NamespacesReactor : Reactor
+public class NamespacesReactor(IGrainFactory grainFactory) : Reactor
 {
     /// <summary>
-    /// Handles the addition of a namespace.
+    /// Handles the addition of a namespace by applying any existing global seed data to it.
     /// </summary>
     /// <param name="event">The event containing the namespace information.</param>
     /// <param name="eventContext">The context of the event.</param>
     /// <returns>Await Task.</returns>
-    public Task Added(NamespaceAdded @event, EventContext eventContext)
+    public async Task Added(NamespaceAdded @event, EventContext eventContext)
     {
-        // TODO: In the future, we need to retrieve global seed data and apply it to the new namespace
-        // When implemented, we will need to use IGrainFactory to:
-        // 1. Get the global seed grain using EventSeedingKey.ForGlobal(@event.EventStore)
-        // 2. Get the namespace-specific seed grain using EventSeedingKey.ForNamespace(@event.EventStore, @event.Namespace)
-        // 3. Retrieve global seed data and apply it to the new namespace
-        return Task.CompletedTask;
+        var globalKey = EventSeedingKey.ForGlobal(@event.EventStore);
+        var globalGrain = grainFactory.GetGrain<IEventSeeding>(globalKey.ToString());
+        var seeds = await globalGrain.GetSeededEvents();
+
+        if (seeds.ByEventSource.Count == 0)
+        {
+            return;
+        }
+
+        var entries = seeds.ByEventSource
+            .SelectMany(kvp => kvp.Value)
+            .GroupBy(e => new { e.EventSourceId, e.EventTypeId, e.Content })
+            .Select(g => g.First())
+            .Select(e => new SeedingEntry(e.EventSourceId, e.EventTypeId, e.Content, e.Tags?.Select(t => new Tag(t))))
+            .ToArray();
+
+        if (entries.Length > 0)
+        {
+            var namespaceKey = EventSeedingKey.ForNamespace(@event.EventStore, @event.Namespace);
+            var nsGrain = grainFactory.GetGrain<IEventSeeding>(namespaceKey.ToString());
+            await nsGrain.Seed(entries);
+        }
     }
 }

Source/Kernel/Core/Seeding/EventSeeding.cs

@@ -10,6 +10,7 @@
 using Cratis.Chronicle.Concepts.Identities;
 using Cratis.Chronicle.Concepts.Seeding;
 using Cratis.Chronicle.EventSequences;
+using Cratis.Chronicle.Namespaces;
 using Cratis.Chronicle.Storage.Seeding;
 using Microsoft.Extensions.Logging;
 using Orleans.Providers;
@@ -60,19 +61,34 @@ public async Task Seed(IEnumerable<SeedingEntry> entries)
                 new Dictionary<EventTypeId, IEnumerable<SeededEventEntry>>(),
                 new Dictionary<EventSourceId, IEnumerable<SeededEventEntry>>());
 
-        // For global grains, just store the entries without appending
+        // For global grains, store the entries and apply to all existing namespaces
         if (_key.IsGlobal)
         {
+            var newEntries = new List<SeedingEntry>();
             foreach (var entry in entriesList)
             {
                 var tags = entry.Tags?.Select(t => t.Value) ?? [];
                 var seededEntry = new SeededEventEntry(entry.EventSourceId, entry.EventTypeId, entry.Content, tags);
                 if (!IsAlreadySeeded(seededEntry))
                 {
                     TrackSeededEvent(seededEntry);
+                    newEntries.Add(entry);
                 }
             }
             await state.WriteStateAsync();
+
+            if (newEntries.Count > 0)
+            {
+                var namespacesGrain = GrainFactory.GetGrain<INamespaces>(_key.EventStore.Value);
+                var namespaces = await namespacesGrain.GetAll();
+                foreach (var ns in namespaces)
+                {
+                    logger.ApplyingSeedsToNamespace(ns.Value);
+                    var namespaceKey = EventSeedingKey.ForNamespace(_key.EventStore, ns);
+                    var nsGrain = GrainFactory.GetGrain<IEventSeeding>(namespaceKey.ToString());
+                    await nsGrain.Seed(newEntries);
+                }
+            }
         }
         else
         {

Source/Kernel/Core/Seeding/EventSeedingLogging.cs

@@ -19,4 +19,7 @@ internal static partial class EventSeedingLogMessages
 
     [LoggerMessage(LogLevel.Debug, "All events have already been seeded, skipping")]
     internal static partial void AllEventsAlreadySeeded(this ILogger<EventSeeding> logger);
+
+    [LoggerMessage(LogLevel.Debug, "Applying global seeds to namespace '{Namespace}'")]
+    internal static partial void ApplyingSeedsToNamespace(this ILogger<EventSeeding> logger, string @namespace);
 }

Source/Kernel/Storage.MongoDB/Sinks/Sink.cs

@@ -47,6 +47,7 @@ public class Sink(
 
     readonly List<WriteModel<BsonDocument>> _bulkOperations = [];
     readonly Dictionary<int, (Key EventSourceId, EventSequenceNumber SequenceNumber)> _bulkOperationMetadata = [];
+    readonly Dictionary<string, ExpandoObject> _bulkStateCache = [];
     int _currentBulkSize;
     bool _isBulkMode;
 
@@ -59,6 +60,15 @@ public class Sink(
     /// <inheritdoc/>
     public async Task<ExpandoObject?> FindOrDefault(Key key)
     {
+        if (_isBulkMode)
+        {
+            var cacheKey = converter.ToBsonValue(key).ToString()!;
+            if (_bulkStateCache.TryGetValue(cacheKey, out var cachedState))
+            {
+                return cachedState;
+            }
+        }
+
         var collection = Collection;
 
         using var result = await collection.FindAsync(Builders<BsonDocument>.Filter.Eq("_id", converter.ToBsonValue(key)));
@@ -86,6 +96,7 @@ public async Task<IEnumerable<FailedPartition>> ApplyChanges(
             if (_isBulkMode)
             {
                 AddToBulk(new DeleteOneModel<BsonDocument>(filter), key, eventSequenceNumber);
+                _bulkStateCache.Remove(converter.ToBsonValue(key).ToString()!);
                 return await FlushBulkIfNeeded();
             }
 
@@ -110,6 +121,10 @@ public async Task<IEnumerable<FailedPartition>> ApplyChanges(
                 ArrayFilters = converted.ArrayFilters
             };
             AddToBulk(updateModel, key, eventSequenceNumber);
+            if (!changeset.HasJoined())
+            {
+                _bulkStateCache[converter.ToBsonValue(key).ToString()!] = changeset.CurrentState;
+            }
             return await FlushBulkIfNeeded();
         }
 
@@ -130,6 +145,7 @@ public Task BeginBulk()
         _isBulkMode = true;
         _bulkOperations.Clear();
         _bulkOperationMetadata.Clear();
+        _bulkStateCache.Clear();
         _currentBulkSize = 0;
         return Task.CompletedTask;
     }
@@ -144,6 +160,7 @@ public async Task EndBulk()
         _isBulkMode = false;
         _bulkOperations.Clear();
         _bulkOperationMetadata.Clear();
+        _bulkStateCache.Clear();
         _currentBulkSize = 0;
     }