Fix live-tail Load New Events not rendering the new row until reload

src/EventLogExpert.Runtime/EventLog/FilteringEffects.cs

@@ -4,7 +4,6 @@
 using EventLogExpert.Eventing.Common.Channels;
 using EventLogExpert.Eventing.Common.EventLogs;
 using EventLogExpert.Eventing.Common.Events;
-using EventLogExpert.Filtering.Compilation;
 using EventLogExpert.Filtering.Evaluation;
 using EventLogExpert.Logging.Abstractions;
 using EventLogExpert.Runtime.FilterProgress;
@@ -21,7 +20,6 @@ internal sealed class FilteringEffects(
     IState<EventLogState> eventLogState,
     IState<RawEventStoreState> rawEventStore,
     IState<LogTableState> logTableState,
-    IFilterService filterService,
     [FromKeyedServices(LogCategories.EventLog)] ITraceLogger logger,
     LogCloseCoordinator closeCoordinator,
     EventLogConcurrencyState concurrencyState,
@@ -31,53 +29,30 @@ internal sealed class FilteringEffects(
     private readonly EventLogConcurrencyState _concurrencyState = concurrencyState;
     private readonly TimeSpan _convergenceDelay = convergenceDelay ?? TimeSpan.FromMilliseconds(250);
     private readonly IState<EventLogState> _eventLogState = eventLogState;
-    private readonly IFilterService _filterService = filterService;
-    private readonly ITraceLogger _logger = logger;
     private readonly IState<LogTableState> _logTableState = logTableState;
+    private readonly ITraceLogger _logger = logger;
     private readonly IState<RawEventStoreState> _rawEventStore = rawEventStore;
 
     [EffectMethod]
     public Task HandleAddEvent(AddEventAction action, IDispatcher dispatcher)
     {
-        if (!_eventLogState.Value.OpenLogs.TryGetValue(action.NewEvent.OwningLog, out var owningLog))
-        {
-            return Task.CompletedTask;
-        }
-
-        if (_eventLogState.Value.ContinuouslyUpdate)
+        // The non-live-tail buffering is handled atomically by ReduceAddEvent; this effect only drives the
+        // continuously-update live tail (ingest the new event, rebuild its display).
+        if (!_eventLogState.Value.ContinuouslyUpdate ||
+            !_eventLogState.Value.OpenLogs.TryGetValue(action.NewEvent.OwningLog, out var owningLog))
         {
-            // Raw ingest is unconditional even when the filter hides the event; only the display rebuild is filter-gated.
-            dispatcher.Dispatch(new IngestRawEventsAction(
-                new Dictionary<EventLogId, IReadOnlyList<ResolvedEvent>> { [owningLog.Id] = [action.NewEvent] },
-                RawIngestMode.Prepend));
-
-            // Skip the rebuild when the new event is hidden: the view is unchanged, so avoid the version churn.
-            var filteredNew = _filterService.GetFilteredEvents(
-                [action.NewEvent],
-                _eventLogState.Value.AppliedFilter);
-
-            if (filteredNew.Count > 0 &&
-                _rawEventStore.Value.ByLog.TryGetValue(owningLog.Id, out var store))
-            {
-                var view = DisplayViewBuilder.Build(
-                    store, owningLog.Id, _eventLogState.Value.AppliedFilter, _logTableState.Value.SortContext);
-
-                dispatcher.Dispatch(new AppendTableEventsAction(owningLog.Id) { View = view });
-            }
-
             return Task.CompletedTask;
         }
 
-        var updatedBuffer = new List<ResolvedEvent>(_eventLogState.Value.NewEventBuffer.Count + 1)
+        var newEventsByLog = new Dictionary<EventLogId, IReadOnlyList<ResolvedEvent>>
         {
-            action.NewEvent
+            [owningLog.Id] = [action.NewEvent]
         };
 
-        updatedBuffer.AddRange(_eventLogState.Value.NewEventBuffer);
-
-        var full = updatedBuffer.Count >= EventLogState.MaxNewEvents;
-
-        dispatcher.Dispatch(new EventBufferedAction(updatedBuffer.AsReadOnly(), full));
+        // Ingest before the rebuild: the filter-gated build runs in the continuation so it reads the post-ingest store
+        // (inlining here would lag the tail by one event). Live tail owns no buffer.
+        dispatcher.Dispatch(new IngestRawEventsAction(newEventsByLog, RawIngestMode.Prepend));
+        dispatcher.Dispatch(new RebuildDisplayViewsAction(newEventsByLog, BufferEntriesToConsume: null));
 
         return Task.CompletedTask;
     }
@@ -198,8 +173,7 @@ public Task HandleSetContinuouslyUpdate(SetContinuouslyUpdateAction action, IDis
     {
         if (action.ContinuouslyUpdate)
         {
-            LogReloadEffects.ProcessNewEventBuffer(
-                _eventLogState.Value, _rawEventStore, _logTableState, dispatcher, _filterService);
+            LogReloadEffects.ProcessNewEventBuffer(_eventLogState.Value, dispatcher);
         }
 
         return Task.CompletedTask;

src/EventLogExpert.Runtime/EventLog/LogReloadEffects.cs

@@ -75,12 +75,44 @@ public Task HandleLoadNewEvents(IDispatcher dispatcher)
         return Task.CompletedTask;
     }
 
-    internal static void ProcessNewEventBuffer(
-        EventLogState state,
-        IState<RawEventStoreState> rawEventStore,
-        IState<LogTableState> logTableState,
-        IDispatcher dispatcher,
-        IFilterService filterService)
+    [EffectMethod]
+    public Task HandleRebuildDisplayViews(RebuildDisplayViewsAction action, IDispatcher dispatcher)
+    {
+        // Continuation of the IngestRawEventsAction dispatched just before it, so these reads see the post-ingest store. Do
+        // not inline back into the producer effect: a same-effect read would see the stale pre-ingest store (the fixed bug).
+        var raw = _rawEventStore.Value.ByLog;
+        var filter = _eventLogState.Value.AppliedFilter;
+        var context = _logTableState.Value.SortContext;
+        var viewsByLog = new Dictionary<EventLogId, EventColumnView>(action.NewEventsByLog.Count);
+
+        foreach (var (logId, newEvents) in action.NewEventsByLog)
+        {
+            // Existence check before filter work: skip a log a concurrent close dropped without filtering it, and rebuild
+            // only when a new event survives the filter so a fully hidden batch doesn't churn the view.
+            if (raw.TryGetValue(logId, out var store) &&
+                _filterService.GetFilteredEvents(newEvents, filter).Count > 0)
+            {
+                viewsByLog[logId] = DisplayViewBuilder.Build(store, logId, filter, context);
+            }
+        }
+
+        if (viewsByLog.Count > 0)
+        {
+            dispatcher.Dispatch(new AppendTableEventsBatchAction { ViewsByLog = viewsByLog });
+        }
+
+        // Consume only after a successful rebuild (unreachable if it threw above), so a build failure preserves the count.
+        // Consuming the captured snapshot - not a blanket clear - keeps a mid-flush event; skip the dispatch when nothing
+        // was captured (an all-filtered rebuild still consumes its non-empty snapshot).
+        if (action.BufferEntriesToConsume is { Count: > 0 } bufferEntriesToConsume)
+        {
+            dispatcher.Dispatch(new NewEventBufferConsumedAction(bufferEntriesToConsume));
+        }
+
+        return Task.CompletedTask;
+    }
+
+    internal static void ProcessNewEventBuffer(EventLogState state, IDispatcher dispatcher)
     {
         var grouped = new Dictionary<EventLogId, List<ResolvedEvent>>();
 
@@ -106,29 +138,9 @@ internal static void ProcessNewEventBuffer(
             dispatcher.Dispatch(new IngestRawEventsAction(rawByLog, RawIngestMode.Prepend));
         }
 
-        var postIngestRaw = rawEventStore.Value.ByLog;
-        var context = logTableState.Value.SortContext;
-        var filter = state.AppliedFilter;
-        var viewsByLog = new Dictionary<EventLogId, EventColumnView>(grouped.Count);
-
-        foreach (var (logId, events) in grouped)
-        {
-            // Rebuild a log's display only when a buffered event survives the filter, so a fully hidden batch doesn't
-            // churn the view or its version.
-            var filtered = filterService.GetFilteredEvents(events, filter);
-
-            if (filtered.Count > 0 && postIngestRaw.TryGetValue(logId, out var store))
-            {
-                viewsByLog[logId] = DisplayViewBuilder.Build(store, logId, filter, context);
-            }
-        }
-
-        if (viewsByLog.Count > 0)
-        {
-            dispatcher.Dispatch(new AppendTableEventsBatchAction { ViewsByLog = viewsByLog });
-        }
-
-        dispatcher.Dispatch(new EventBufferedAction([], false));
+        // Continuation runs after the ingest, so it reads the post-ingest store and consumes exactly this snapshot by
+        // identity. Atomic reducer buffering (ReduceAddEvent) keeps an event buffered concurrently with the flush alive.
+        dispatcher.Dispatch(new RebuildDisplayViewsAction(rawByLog, BufferEntriesToConsume: state.NewEventBuffer));
     }
 
     private static void RestoreSelection(
@@ -165,7 +177,4 @@ private static void RestoreSelection(
         SelectionEntry? focused = focusEntry ?? (restored.Count > 0 ? restored[^1] : null);
         dispatcher.Dispatch(new SetSelectedEventsAction(restored, focused));
     }
-
-    private void ProcessNewEventBuffer(EventLogState state, IDispatcher dispatcher) =>
-        ProcessNewEventBuffer(state, _rawEventStore, _logTableState, dispatcher, _filterService);
 }

src/EventLogExpert.Runtime/EventLog/NewEventBufferConsumedAction.cs

@@ -0,0 +1,10 @@
+// // Copyright (c) Microsoft Corporation.
+// // Licensed under the MIT License.
+
+using EventLogExpert.Eventing.Common.Events;
+
+namespace EventLogExpert.Runtime.EventLog;
+
+// Removes only the buffer entries captured for a completed flush, by reference identity, so an event a watcher buffers
+// mid-flush survives (a blanket clear would drop it).
+internal sealed record NewEventBufferConsumedAction(IReadOnlyList<ResolvedEvent> ConsumedEvents);

src/EventLogExpert.Runtime/EventLog/RebuildDisplayViewsAction.cs

@@ -0,0 +1,14 @@
+// // Copyright (c) Microsoft Corporation.
+// // Licensed under the MIT License.
+
+using EventLogExpert.Eventing.Common.EventLogs;
+using EventLogExpert.Eventing.Common.Events;
+
+namespace EventLogExpert.Runtime.EventLog;
+
+// Effect-only continuation dispatched after IngestRawEventsAction so the rebuild reads the post-ingest store.
+// BufferEntriesToConsume is the flush's captured buffer snapshot to remove on a successful rebuild; null on the
+// live-tail path (no buffer to consume).
+internal sealed record RebuildDisplayViewsAction(
+    IReadOnlyDictionary<EventLogId, IReadOnlyList<ResolvedEvent>> NewEventsByLog,
+    IReadOnlyList<ResolvedEvent>? BufferEntriesToConsume);

src/EventLogExpert.Runtime/EventLog/Reducers.cs

@@ -18,6 +18,27 @@ internal sealed class Reducers
     private static readonly ImmutableDictionary<string, ImmutableHashSet<string>> s_emptyNamesByLog =
         ImmutableDictionary<string, ImmutableHashSet<string>>.Empty;
 
+    [ReducerMethod]
+    public static EventLogState ReduceAddEvent(EventLogState state, AddEventAction action)
+    {
+        // Buffer additively in the reducer so concurrent adds and the flush's consume compose against current state (a
+        // stale whole-buffer effect write could clobber them). Continuously-update buffers nothing; HandleAddEvent drives
+        // the live tail.
+        if (state.ContinuouslyUpdate || !state.OpenLogs.ContainsKey(action.NewEvent.OwningLog))
+        {
+            return state;
+        }
+
+        var buffer = new List<ResolvedEvent>(state.NewEventBuffer.Count + 1) { action.NewEvent };
+        buffer.AddRange(state.NewEventBuffer);
+
+        return state with
+        {
+            NewEventBuffer = buffer.AsReadOnly(),
+            NewEventBufferIsFull = buffer.Count >= EventLogState.MaxNewEvents
+        };
+    }
+
     [ReducerMethod]
     public static EventLogState ReduceApplyFilter(EventLogState state, ApplyFilterAction action)
     {
@@ -74,10 +95,6 @@ public static EventLogState ReduceCloseLog(EventLogState state, CloseLogAction a
         };
     }
 
-    [ReducerMethod]
-    public static EventLogState ReduceEventBuffered(EventLogState state, EventBufferedAction action) =>
-        state with { NewEventBuffer = action.UpdatedBuffer, NewEventBufferIsFull = action.IsFull };
-
     [ReducerMethod]
     public static EventLogState ReduceLoadEvents(EventLogState state, LoadEventsAction action)
     {
@@ -132,6 +149,23 @@ public static EventLogState ReduceLoadEventsPartial(EventLogState state, LoadEve
         };
     }
 
+    [ReducerMethod]
+    public static EventLogState ReduceNewEventBufferConsumed(EventLogState state, NewEventBufferConsumedAction action)
+    {
+        if (action.ConsumedEvents.Count == 0) { return state; }
+
+        // Remove only the captured entries by reference identity, so an event a watcher buffered during the flush
+        // (prepended after the snapshot) survives. Mirrors ReduceCloseLog's filtered removal + IsFull recompute.
+        var consumed = new HashSet<object>(action.ConsumedEvents, ReferenceEqualityComparer.Instance);
+        var remaining = state.NewEventBuffer.Where(bufferedEvent => !consumed.Contains(bufferedEvent)).ToList();
+
+        return state with
+        {
+            NewEventBuffer = remaining,
+            NewEventBufferIsFull = remaining.Count >= EventLogState.MaxNewEvents
+        };
+    }
+
     [ReducerMethod]
     public static EventLogState ReduceOpenLog(EventLogState state, OpenLogAction action)
     {