diff --git a/src/EventLogExpert.Runtime/EventLog/EventBufferedAction.cs b/src/EventLogExpert.Runtime/EventLog/EventBufferedAction.cs deleted file mode 100644 index e09f1810..00000000 --- a/src/EventLogExpert.Runtime/EventLog/EventBufferedAction.cs +++ /dev/null @@ -1,8 +0,0 @@ -// // Copyright (c) Microsoft Corporation. -// // Licensed under the MIT License. - -using EventLogExpert.Eventing.Common.Events; - -namespace EventLogExpert.Runtime.EventLog; - -internal sealed record EventBufferedAction(IReadOnlyList UpdatedBuffer, bool IsFull); diff --git a/src/EventLogExpert.Runtime/EventLog/FilteringEffects.cs b/src/EventLogExpert.Runtime/EventLog/FilteringEffects.cs index 7a092984..d86242db 100644 --- a/src/EventLogExpert.Runtime/EventLog/FilteringEffects.cs +++ b/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, IState rawEventStore, IState 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; - private readonly IFilterService _filterService = filterService; - private readonly ITraceLogger _logger = logger; private readonly IState _logTableState = logTableState; + private readonly ITraceLogger _logger = logger; private readonly IState _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> { [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(_eventLogState.Value.NewEventBuffer.Count + 1) + var newEventsByLog = new Dictionary> { - 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; diff --git a/src/EventLogExpert.Runtime/EventLog/LogReloadEffects.cs b/src/EventLogExpert.Runtime/EventLog/LogReloadEffects.cs index 6d61311d..9b16a4a1 100644 --- a/src/EventLogExpert.Runtime/EventLog/LogReloadEffects.cs +++ b/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 rawEventStore, - IState 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(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>(); @@ -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(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); } diff --git a/src/EventLogExpert.Runtime/EventLog/NewEventBufferConsumedAction.cs b/src/EventLogExpert.Runtime/EventLog/NewEventBufferConsumedAction.cs new file mode 100644 index 00000000..03cd2217 --- /dev/null +++ b/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 ConsumedEvents); diff --git a/src/EventLogExpert.Runtime/EventLog/RebuildDisplayViewsAction.cs b/src/EventLogExpert.Runtime/EventLog/RebuildDisplayViewsAction.cs new file mode 100644 index 00000000..9b929e3f --- /dev/null +++ b/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> NewEventsByLog, + IReadOnlyList? BufferEntriesToConsume); diff --git a/src/EventLogExpert.Runtime/EventLog/Reducers.cs b/src/EventLogExpert.Runtime/EventLog/Reducers.cs index 11bfabd0..58d1d59c 100644 --- a/src/EventLogExpert.Runtime/EventLog/Reducers.cs +++ b/src/EventLogExpert.Runtime/EventLog/Reducers.cs @@ -18,6 +18,27 @@ internal sealed class Reducers private static readonly ImmutableDictionary> s_emptyNamesByLog = ImmutableDictionary>.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(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(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) { diff --git a/tests/Unit/EventLogExpert.Runtime.Tests/EventLog/EffectsTests.cs b/tests/Unit/EventLogExpert.Runtime.Tests/EventLog/EffectsTests.cs index 905c8ffa..aceea402 100644 --- a/tests/Unit/EventLogExpert.Runtime.Tests/EventLog/EffectsTests.cs +++ b/tests/Unit/EventLogExpert.Runtime.Tests/EventLog/EffectsTests.cs @@ -22,87 +22,120 @@ using NSubstitute; using System.Collections.Immutable; using CloseLogAction = EventLogExpert.Runtime.EventLog.CloseLogAction; +using Reducers = EventLogExpert.Runtime.EventLog.Reducers; namespace EventLogExpert.Runtime.Tests.EventLog; public sealed class EffectsTests { [Fact] - public async Task HandleAddEvent_WhenBufferReachesMaxEvents_ShouldSetFullFlag() + public async Task HandleAddEvent_ContinuouslyUpdate_QueueFaithful_AppendedViewContainsNewEvent() { - // Arrange - var existingBuffer = Enumerable.Range(0, EventLogState.MaxNewEvents - 1) - .Select(i => FilterEventBuilder.CreateTestEvent(i, logName: Constants.LogNameTestLog)) - .ToList(); - + // Regression (live-tail one-event lag): drive HandleAddEvent then drain the dispatch queue in Fluxor order. The + // rebuild effect reads the store only AFTER the ingest reducer applies, so the appended view contains the new + // event. Against the pre-fix same-effect read the view was built from the empty pre-ingest store and stayed empty. var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); - var activeLogs = ImmutableDictionary.Empty.Add(Constants.LogNameTestLog, logData); - var (effects, mockDispatcher) = CreateEffects(false, activeLogs, existingBuffer); + var state = new EventLogState + { + ContinuouslyUpdate = true, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)), + NewEventBuffer = [], + AppliedFilter = new Filter(null, []) + }; - var newEvent = FilterEventBuilder.CreateTestEvent(1000, logName: Constants.LogNameTestLog); - var action = new AddEventAction(newEvent); + var rawState = new RawEventStoreState + { + ByLog = ImmutableDictionary.Empty.Add(logData.Id, EventColumnStore.Empty) + }; + + var (effects, mockDispatcher, _) = CreateEffectsWithMutableState(() => state, () => rawState); + var pending = CaptureDispatchQueue(mockDispatcher); + + AppendTableEventsBatchAction? captured = null; + mockDispatcher.When(d => d.Dispatch(Arg.Any())) + .Do(call => captured = call.ArgAt(0)); + + var newEvent = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); // Act - await effects.HandleAddEvent(action, mockDispatcher); + await effects.HandleAddEvent(new AddEventAction(newEvent), mockDispatcher); + await DrainDispatchQueueAsync(pending, effects, mockDispatcher, () => rawState, r => rawState = r); - // Assert - mockDispatcher.Received(1) - .Dispatch(Arg.Is(a => a != null && - a.IsFull == true && a.UpdatedBuffer.Count == EventLogState.MaxNewEvents)); + // Assert: a single batched append whose view for the log contains the just-arrived event. + Assert.NotNull(captured); + Assert.True(captured.ViewsByLog.ContainsKey(logData.Id)); + Assert.Equal(1, captured.ViewsByLog[logData.Id].Count); + Assert.Equal(newEvent.Id, captured.ViewsByLog[logData.Id].EnumerateDetail().First().Id); + mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); } [Fact] - public async Task HandleAddEvent_WhenContinuouslyUpdateFalse_ShouldBufferEvent() + public async Task HandleAddEvent_WhenContinuouslyUpdateFalse_DoesNotDispatch_BufferingIsReducerOnly() { - // Arrange + // Buffering for the non-continuously-update path is now an atomic reducer (ReduceAddEvent), so the effect itself + // dispatches nothing - it only drives the live tail. Buffer contents / full-flag are covered by the reducer tests. var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); var activeLogs = ImmutableDictionary.Empty.Add(Constants.LogNameTestLog, logData); - var (effects, mockDispatcher) = CreateEffects( - false, - activeLogs); + var (effects, mockDispatcher) = CreateEffects(false, activeLogs); var newEvent = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); - var action = new AddEventAction(newEvent); // Act - await effects.HandleAddEvent(action, mockDispatcher); + await effects.HandleAddEvent(new AddEventAction(newEvent), mockDispatcher); // Assert - mockDispatcher.Received(1) - .Dispatch(Arg.Is(a => a != null && - a.UpdatedBuffer.Count == 1 && a.UpdatedBuffer[0] == newEvent)); + mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); } [Fact] - public async Task HandleAddEvent_WhenContinuouslyUpdateTrue_AndEventFilteredOut_ShouldNotDispatchAppend() + public async Task HandleAddEvent_WhenContinuouslyUpdateTrue_AndEventFilteredOut_ShouldNotAppend() { - // Arrange + // Arrange: continuously-update live tail with an active filter that hides the new event. The producer ingests and + // queues a rebuild unconditionally; draining after the ingest reducer, the rebuild effect must skip the append + // because the event is filtered out, and must NOT consume the buffer (live tail passes BufferEntriesToConsume:null). var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); - var activeLogs = ImmutableDictionary.Empty.Add(Constants.LogNameTestLog, logData); - var (effects, mockDispatcher, _, _, mockFilterService) = - CreateEffectsWithServices(true, activeLogs); + var state = new EventLogState + { + ContinuouslyUpdate = true, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)), + NewEventBuffer = [], + AppliedFilter = new Filter(null, []) + }; + + var rawState = new RawEventStoreState + { + ByLog = ImmutableDictionary.Empty.Add(logData.Id, EventColumnStore.Empty) + }; + + var (effects, mockDispatcher, mockFilterService) = CreateEffectsWithMutableState(() => state, () => rawState); mockFilterService.GetFilteredEvents(Arg.Any>(), Arg.Any()) .Returns(new List()); + var pending = CaptureDispatchQueue(mockDispatcher); var newEvent = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); - var action = new AddEventAction(newEvent); // Act - await effects.HandleAddEvent(action, mockDispatcher); + await effects.HandleAddEvent(new AddEventAction(newEvent), mockDispatcher); + await DrainDispatchQueueAsync(pending, effects, mockDispatcher, () => rawState, r => rawState = r); - // Assert + // Assert: no display append (event hidden) and no buffer consume on the live-tail path. + mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); + mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); } [Fact] public async Task HandleAddEvent_WhenContinuouslyUpdateTrue_AndEventFilteredOut_ShouldStillIngestRaw() { - // Arrange: a live-tail event the active filter hides still belongs in the raw store. + // Arrange: a live-tail event the active filter hides still belongs in the raw store. The producer ingests and + // queues the rebuild unconditionally; the filter gate lives in the rebuild effect, not here. var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); var activeLogs = ImmutableDictionary.Empty.Add(Constants.LogNameTestLog, logData); @@ -117,14 +150,17 @@ public async Task HandleAddEvent_WhenContinuouslyUpdateTrue_AndEventFilteredOut_ // Act await effects.HandleAddEvent(new AddEventAction(newEvent), mockDispatcher); - // Assert: raw is ingested unconditionally (Prepend) even though the filtered display append is skipped. + // Assert: raw is ingested unconditionally (Prepend) and a rebuild is queued even though the display append itself + // is filter-gated inside the rebuild effect. mockDispatcher.Received(1).Dispatch(Arg.Is(a => a != null && a.Mode == RawIngestMode.Prepend && a.EventsByLog.ContainsKey(logData.Id))); + mockDispatcher.Received(1).Dispatch(Arg.Is(a => a != null && + a.NewEventsByLog.ContainsKey(logData.Id) && a.BufferEntriesToConsume == null)); mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); } [Fact] - public async Task HandleAddEvent_WhenContinuouslyUpdateTrue_ShouldIngestRawAndAppend() + public async Task HandleAddEvent_WhenContinuouslyUpdateTrue_ShouldIngestRawAndQueueRebuild() { // Arrange var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); @@ -145,26 +181,24 @@ public async Task HandleAddEvent_WhenContinuouslyUpdateTrue_ShouldIngestRawAndAp var (effects, mockDispatcher, _) = CreateEffectsWithMutableState(() => state, () => rawState); - // Fluxor runs the raw-store reducer synchronously on the ingest dispatch; mirror that here so the - // post-ingest store is visible when HandleAddEvent rebuilds the display view over it. - mockDispatcher - .When(d => d.Dispatch(Arg.Any())) - .Do(callInfo => rawState = RawEventStoreReducers.ReduceIngestRawEvents(rawState, callInfo.ArgAt(0))); - var newEvent = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); var action = new AddEventAction(newEvent); // Act await effects.HandleAddEvent(action, mockDispatcher); - // Assert + // Assert: the producer ingests the raw event and queues a rebuild continuation carrying the new event. It does not + // build the display view itself - that runs in the rebuild effect, after the ingest reducer applies. mockDispatcher.Received(1).Dispatch(Arg.Is(a => a != null && a.Mode == RawIngestMode.Prepend && a.EventsByLog.ContainsKey(logData.Id))); - mockDispatcher.Received(1) - .Dispatch(Arg.Is(a => a != null && - a.LogId == logData.Id && a.View != null && a.View.Count == 1 && a.View.EnumerateDetail().First().Id == newEvent.Id)); + mockDispatcher.Received(1).Dispatch(Arg.Is(a => a != null && + a.BufferEntriesToConsume == null && + a.NewEventsByLog.ContainsKey(logData.Id) && + a.NewEventsByLog[logData.Id].Count == 1 && + a.NewEventsByLog[logData.Id][0].Id == newEvent.Id)); + mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); } @@ -843,6 +877,53 @@ public async Task HandleApplyFilter_WhenFilterDoesNotRequireXml_ShouldNotReloadL mockDispatcher.Received(1).Dispatch(Arg.Any()); } + [Fact] + public async Task HandleApplyFilter_WhenFilterRequiresXmlAndLogLacksXml_ShouldCloseAndReopenLog() + { + // Arrange: active log has not been loaded with XML, so it must be re-read. + var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); + var activeLogs = ImmutableDictionary.Empty.Add(Constants.LogNameTestLog, logData); + + var (effects, mockDispatcher, _, _, _) = CreateEffectsWithServices(activeLogs: activeLogs); + + // Route CloseLog to HandleCloseLog so HandleApplyFilter's await on the close-completion + // TCS resolves quickly (otherwise it hits LogCloseTimeout, 30s). Capture the routed + // tasks so any fault in HandleCloseLog surfaces at the end of the test instead of + // being swallowed by the discard. + var closeTasks = new List(); + + mockDispatcher + .When(d => d.Dispatch(Arg.Any())) + .Do(callInfo => + { + closeTasks.Add(effects.HandleCloseLog(callInfo.ArgAt(0), mockDispatcher)); + }); + + var xmlFilter = FilterBuilder.CreateTestFilter(FilterTestConstants.FilterXmlContainsData, isEnabled: true); + var filter = new Filter(null, [xmlFilter]); + var action = new ApplyFilterAction(filter); + + // Act + await effects.HandleApplyFilter(action, mockDispatcher); + + // Assert + Assert.True(filter.RequiresXml); + + mockDispatcher.Received(1) + .Dispatch(Arg.Is(a => a != null && + a.LogName == Constants.LogNameTestLog && a.LogId == logData.Id)); + + mockDispatcher.Received(1) + .Dispatch(Arg.Is(a => a != null && + a.LogName == Constants.LogNameTestLog && a.LogPathType == LogPathType.Channel)); + + // Reload path returns early; no DisplayReady until LoadEvents fires. + mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); + + // Surface any HandleCloseLog faults before exiting the test. + await Task.WhenAll(closeTasks); + } + [Fact] public async Task HandleApplyFilter_WhenFilterRequiresXml_AwaitsCloseCompletionBeforeReturning() { @@ -994,53 +1075,6 @@ public async Task HandleApplyFilter_WhenFilterRequiresXml_ShouldRestoreSelection await Task.WhenAll(closeTasks); } - [Fact] - public async Task HandleApplyFilter_WhenFilterRequiresXmlAndLogLacksXml_ShouldCloseAndReopenLog() - { - // Arrange: active log has not been loaded with XML, so it must be re-read. - var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); - var activeLogs = ImmutableDictionary.Empty.Add(Constants.LogNameTestLog, logData); - - var (effects, mockDispatcher, _, _, _) = CreateEffectsWithServices(activeLogs: activeLogs); - - // Route CloseLog to HandleCloseLog so HandleApplyFilter's await on the close-completion - // TCS resolves quickly (otherwise it hits LogCloseTimeout, 30s). Capture the routed - // tasks so any fault in HandleCloseLog surfaces at the end of the test instead of - // being swallowed by the discard. - var closeTasks = new List(); - - mockDispatcher - .When(d => d.Dispatch(Arg.Any())) - .Do(callInfo => - { - closeTasks.Add(effects.HandleCloseLog(callInfo.ArgAt(0), mockDispatcher)); - }); - - var xmlFilter = FilterBuilder.CreateTestFilter(FilterTestConstants.FilterXmlContainsData, isEnabled: true); - var filter = new Filter(null, [xmlFilter]); - var action = new ApplyFilterAction(filter); - - // Act - await effects.HandleApplyFilter(action, mockDispatcher); - - // Assert - Assert.True(filter.RequiresXml); - - mockDispatcher.Received(1) - .Dispatch(Arg.Is(a => a != null && - a.LogName == Constants.LogNameTestLog && a.LogId == logData.Id)); - - mockDispatcher.Received(1) - .Dispatch(Arg.Is(a => a != null && - a.LogName == Constants.LogNameTestLog && a.LogPathType == LogPathType.Channel)); - - // Reload path returns early; no DisplayReady until LoadEvents fires. - mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); - - // Surface any HandleCloseLog faults before exiting the test. - await Task.WhenAll(closeTasks); - } - [Fact] public async Task HandleApplyFilter_WhenNewerApplyFilterRacesReload_ShouldStillReopenClosedLogs() { @@ -1579,17 +1613,14 @@ public async Task HandleLoadNewEvents_ShouldProcessBufferAndDispatchActions() }; var (effects, mockDispatcher, _) = CreateEffectsWithMutableState(() => state, () => rawState); - - // Mirror Fluxor's synchronous raw-store reducer on the ingest dispatch so the buffered events are - // visible in the store when the batch view is built. - mockDispatcher - .When(d => d.Dispatch(Arg.Any())) - .Do(callInfo => rawState = RawEventStoreReducers.ReduceIngestRawEvents(rawState, callInfo.ArgAt(0))); + var pending = CaptureDispatchQueue(mockDispatcher); // Act await effects.HandleLoadNewEvents(mockDispatcher); + await DrainDispatchQueueAsync(pending, effects, mockDispatcher, () => rawState, r => rawState = r); - // Assert + // Assert: a single batched append carrying both buffered events (built from the post-ingest store), then the + // buffer consume of exactly the captured snapshot. mockDispatcher.Received(1) .Dispatch(Arg.Is(a => a != null && a.ViewsByLog.Count == 1 && @@ -1599,8 +1630,7 @@ public async Task HandleLoadNewEvents_ShouldProcessBufferAndDispatchActions() mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); mockDispatcher.Received(1) - .Dispatch(Arg.Is(a => a != null && - a.UpdatedBuffer.Count == 0 && a.IsFull == false)); + .Dispatch(Arg.Is(a => a != null && a.ConsumedEvents.Count == 2)); } [Fact] @@ -1613,23 +1643,38 @@ public async Task HandleLoadNewEvents_WhenAllEventsFiltered_ShouldNotDispatchApp }; var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); - var activeLogs = ImmutableDictionary.Empty.Add(Constants.LogNameTestLog, logData); - var (effects, mockDispatcher, _, _, mockFilterService) = - CreateEffectsWithServices(activeLogs: activeLogs, newEventBuffer: bufferedEvents); + var state = new EventLogState + { + ContinuouslyUpdate = false, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)), + NewEventBuffer = bufferedEvents, + AppliedFilter = new Filter(null, []) + }; + + var rawState = new RawEventStoreState + { + ByLog = ImmutableDictionary.Empty.Add(logData.Id, EventColumnStore.Empty) + }; + + var (effects, mockDispatcher, mockFilterService) = CreateEffectsWithMutableState(() => state, () => rawState); mockFilterService.GetFilteredEvents(Arg.Any>(), Arg.Any()) .Returns(new List()); + var pending = CaptureDispatchQueue(mockDispatcher); + // Act await effects.HandleLoadNewEvents(mockDispatcher); + await DrainDispatchQueueAsync(pending, effects, mockDispatcher, () => rawState, r => rawState = r); - // Assert + // Assert: every buffered event is filtered out, so no display append - but the captured snapshot is still consumed + // (a successful zero-view rebuild), so the "New Events" count resets. mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); mockDispatcher.Received(1) - .Dispatch(Arg.Is(a => a != null && - a.UpdatedBuffer.Count == 0 && a.IsFull == false)); + .Dispatch(Arg.Is(a => a != null && a.ConsumedEvents.Count == 1)); } [Fact] @@ -1665,12 +1710,7 @@ public async Task HandleLoadNewEvents_WhenBufferSpansMultipleLogs_ShouldGroupInt }; var (effects, mockDispatcher, _) = CreateEffectsWithMutableState(() => state, () => rawState); - - // Mirror Fluxor's synchronous raw-store reducer so the ingest of both logs is visible when the - // single batch view is built. - mockDispatcher - .When(d => d.Dispatch(Arg.Any())) - .Do(callInfo => rawState = RawEventStoreReducers.ReduceIngestRawEvents(rawState, callInfo.ArgAt(0))); + var pending = CaptureDispatchQueue(mockDispatcher); AppendTableEventsBatchAction? captured = null; @@ -1680,6 +1720,7 @@ public async Task HandleLoadNewEvents_WhenBufferSpansMultipleLogs_ShouldGroupInt // Act await effects.HandleLoadNewEvents(mockDispatcher); + await DrainDispatchQueueAsync(pending, effects, mockDispatcher, () => rawState, r => rawState = r); // Assert: a single batched append covering both logs. mockDispatcher.Received(1).Dispatch(Arg.Any()); @@ -1689,6 +1730,50 @@ public async Task HandleLoadNewEvents_WhenBufferSpansMultipleLogs_ShouldGroupInt Assert.Equal(1, captured.ViewsByLog[testLog.Id].Count); } + [Fact] + public async Task HandleLoadNewEvents_WhenRebuildThrows_PreservesBufferByNotClearing() + { + // Clear-on-throw guard: when the rebuild throws (e.g. a broken compiled filter), the continuation must NOT consume + // the buffer, so the count stays instead of clearing with no row shown. + var bufferedEvents = new List + { + FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog) + }; + + var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); + + var state = new EventLogState + { + ContinuouslyUpdate = false, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)), + NewEventBuffer = bufferedEvents, + AppliedFilter = new Filter(null, []) + }; + + var rawState = new RawEventStoreState + { + ByLog = ImmutableDictionary.Empty.Add(logData.Id, EventColumnStore.Empty) + }; + + var (effects, mockDispatcher, mockFilterService) = CreateEffectsWithMutableState(() => state, () => rawState); + + mockFilterService.GetFilteredEvents(Arg.Any>(), Arg.Any()) + .Returns(_ => throw new InvalidOperationException("filter compile failed")); + + var pending = CaptureDispatchQueue(mockDispatcher); + + // Act + await effects.HandleLoadNewEvents(mockDispatcher); + + await Assert.ThrowsAsync(() => + DrainDispatchQueueAsync(pending, effects, mockDispatcher, () => rawState, r => rawState = r)); + + // Assert: the rebuild threw before appending, so neither the append nor the buffer consume fired - the count stays. + mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); + mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); + } + [Fact] public async Task HandleOpenLog_AwaitsInitialClassificationTask_BeforeResolverConstruction() { @@ -2018,40 +2103,40 @@ public async Task HandleOpenLog_ReverseEagerLoad_SeedsWatcherFromNewestBookmark( } [Fact] - public async Task HandleOpenLog_ReverseEagerLoad_WhenReaderInvalid_SurfacesLoadFailureNotEmptyLog() + public async Task HandleOpenLog_ReverseEagerLoad_WhenReadStopsOnError_SurfacesLoadFailureNotFinalLoad() { + const int total = 60; var fakeFactory = new FakeEventLogReaderFactory( - new FakeEventLogReader([], newestBookmark: null) { IsValid = false, OpenErrorCode = 5 }); + new FakeEventLogReader(BuildReverseBatches(total, batchSize: 30), newestBookmark: "NEWEST") + { + LastErrorCode = 5 // a non-null error once the batches run out: a failed read, not a clean end-of-results + }); - var (openLog, dispatcher, watcher) = CreateEagerLoadEffects(fakeFactory); + var (openLog, dispatcher, _) = CreateEagerLoadEffects(fakeFactory); await openLog.HandleOpenLog(new OpenLogAction(Constants.LogNameApplication, LogPathType.Channel), dispatcher); - // A log that fails to open surfaces an error instead of a misleading empty final load, and never seeds the watcher. + // A read that stops on a Win32 error is surfaced as a failure, not dispatched as a successful final load. dispatcher.Received().Dispatch(Arg.Is(a => a != null && a.ResolverStatus.Contains("Error") && a.ResolverStatus.Contains(Constants.LogNameApplication))); Assert.Empty(dispatcher.ReceivedCalls().Select(call => call.GetArguments()[0]).OfType()); - watcher.DidNotReceive().AddLog(Arg.Any(), Arg.Any(), Arg.Any()); } [Fact] - public async Task HandleOpenLog_ReverseEagerLoad_WhenReadStopsOnError_SurfacesLoadFailureNotFinalLoad() + public async Task HandleOpenLog_ReverseEagerLoad_WhenReaderInvalid_SurfacesLoadFailureNotEmptyLog() { - const int total = 60; var fakeFactory = new FakeEventLogReaderFactory( - new FakeEventLogReader(BuildReverseBatches(total, batchSize: 30), newestBookmark: "NEWEST") - { - LastErrorCode = 5 // a non-null error once the batches run out: a failed read, not a clean end-of-results - }); + new FakeEventLogReader([], newestBookmark: null) { IsValid = false, OpenErrorCode = 5 }); - var (openLog, dispatcher, _) = CreateEagerLoadEffects(fakeFactory); + var (openLog, dispatcher, watcher) = CreateEagerLoadEffects(fakeFactory); await openLog.HandleOpenLog(new OpenLogAction(Constants.LogNameApplication, LogPathType.Channel), dispatcher); - // A read that stops on a Win32 error is surfaced as a failure, not dispatched as a successful final load. + // A log that fails to open surfaces an error instead of a misleading empty final load, and never seeds the watcher. dispatcher.Received().Dispatch(Arg.Is(a => a != null && a.ResolverStatus.Contains("Error") && a.ResolverStatus.Contains(Constants.LogNameApplication))); Assert.Empty(dispatcher.ReceivedCalls().Select(call => call.GetArguments()[0]).OfType()); + watcher.DidNotReceive().AddLog(Arg.Any(), Arg.Any(), Arg.Any()); } [Fact] @@ -2177,8 +2262,8 @@ public async Task HandleSetContinuouslyUpdate_WhenTrue_ShouldProcessBuffer() // Act await effects.HandleSetContinuouslyUpdate(action, mockDispatcher); - // Assert: ProcessNewEventBuffer now dispatches a batched append (no DisplayReady). - mockDispatcher.Received(1).Dispatch(Arg.Any()); + // Assert: ProcessNewEventBuffer ingests the buffered events and queues the display rebuild continuation. + mockDispatcher.Received(1).Dispatch(Arg.Any()); mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); } @@ -2289,6 +2374,50 @@ public async Task HandleUpdateTable_WhenNoSortPending_DoesNotRepublish() mockDispatcher.DidNotReceive().Dispatch(Arg.Any()); } + [Fact] + public async Task HandleUpdateTable_WhenSortPendingMultiLog_RepublishesEveryLogUnderRequested() + { + var logA = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); + var logB = new EventLogData("SecondTestLog", LogPathType.Channel); + + var eventsA = new List { FilterEventBuilder.CreateTestEvent(id: 1, source: "A") }; + var eventsB = new List { FilterEventBuilder.CreateTestEvent(id: 2, source: "B") }; + + var rawState = new RawEventStoreState + { + ByLog = ImmutableDictionary.Empty + .Add(logA.Id, EventColumnStore.Build(eventsA, 0, 0)) + .Add(logB.Id, EventColumnStore.Build(eventsB, 0, 0)) + }; + + var eventLogState = new EventLogState + { + ContinuouslyUpdate = false, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logA.Id, LogPathType.Channel)) + .Add("SecondTestLog", new OpenLogInfo(logB.Id, LogPathType.Channel)), + NewEventBuffer = [], + AppliedFilter = new Filter(null, []) + }; + + var logTableState = new LogTableState + { + RequestedOrderBy = ColumnName.Source, + DisplayListVersion = 3 + }; + + var (effects, mockDispatcher, _) = + CreateEffectsWithMutableState(() => eventLogState, () => rawState, logTableState); + + await effects.Filtering.HandleUpdateTable(mockDispatcher); + + var requested = new SortContext(ColumnName.Source, true, null, false); + mockDispatcher.Received(1).Dispatch(Arg.Is(a => a != null && + a.Version == 3 && + a.Views.ContainsKey(logA.Id) && a.Views[logA.Id].HasContext(requested) && + a.Views.ContainsKey(logB.Id) && a.Views[logB.Id].HasContext(requested))); + } + [Fact] public async Task HandleUpdateTable_WhenSortPending_RepublishesUnderRequestedContextAtCapturedVersion() { @@ -2333,47 +2462,212 @@ public async Task HandleUpdateTable_WhenSortPending_RepublishesUnderRequestedCon } [Fact] - public async Task HandleUpdateTable_WhenSortPendingMultiLog_RepublishesEveryLogUnderRequested() + public async Task RebuildDisplayViews_TwoIngestsBeforeRebuilds_FinalViewContainsBothEvents() { - var logA = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); - var logB = new EventLogData("SecondTestLog", LogPathType.Channel); + // Concurrency safety: model two live events whose ingests both land before either rebuild drains (the worst-case + // interleaving). Because each rebuild reconstructs the whole view from the current store rather than appending only + // its own event, every rebuild yields a superset, so neither event is dropped. + var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); - var eventsA = new List { FilterEventBuilder.CreateTestEvent(id: 1, source: "A") }; - var eventsB = new List { FilterEventBuilder.CreateTestEvent(id: 2, source: "B") }; + var state = new EventLogState + { + ContinuouslyUpdate = true, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)), + NewEventBuffer = [], + AppliedFilter = new Filter(null, []) + }; var rawState = new RawEventStoreState { - ByLog = ImmutableDictionary.Empty - .Add(logA.Id, EventColumnStore.Build(eventsA, 0, 0)) - .Add(logB.Id, EventColumnStore.Build(eventsB, 0, 0)) + ByLog = ImmutableDictionary.Empty.Add(logData.Id, EventColumnStore.Empty) }; - var eventLogState = new EventLogState + var (effects, mockDispatcher, _) = CreateEffectsWithMutableState(() => state, () => rawState); + + var eventA = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); + var eventB = FilterEventBuilder.CreateTestEvent(200, logName: Constants.LogNameTestLog); + + IReadOnlyList justA = [eventA]; + IReadOnlyList justB = [eventB]; + var byLogA = new Dictionary> { [logData.Id] = justA }; + var byLogB = new Dictionary> { [logData.Id] = justB }; + + var pending = new Queue(); + + // Both ingests enqueued ahead of both rebuilds - the interleaving where B's rebuild could miss A under a naive + // append-only design. + pending.Enqueue(new IngestRawEventsAction(byLogA, RawIngestMode.Prepend)); + pending.Enqueue(new IngestRawEventsAction(byLogB, RawIngestMode.Prepend)); + pending.Enqueue(new RebuildDisplayViewsAction(byLogA, BufferEntriesToConsume: null)); + pending.Enqueue(new RebuildDisplayViewsAction(byLogB, BufferEntriesToConsume: null)); + + AppendTableEventsBatchAction? lastAppend = null; + mockDispatcher.When(d => d.Dispatch(Arg.Any())) + .Do(call => lastAppend = call.ArgAt(0)); + + // Act + await DrainDispatchQueueAsync(pending, effects, mockDispatcher, () => rawState, r => rawState = r); + + // Assert: the final rebuilt view holds BOTH events. + Assert.NotNull(lastAppend); + Assert.Equal(2, lastAppend.ViewsByLog[logData.Id].Count); + var ids = lastAppend.ViewsByLog[logData.Id].EnumerateDetail().Select(detail => detail.Id).ToHashSet(); + Assert.Contains(eventA.Id, ids); + Assert.Contains(eventB.Id, ids); + } + + [Fact] + public async Task RebuildDisplayViews_WhenEventBuffersDuringFlush_ConsumesOnlySnapshotAndPreservesNewEvent() + { + // Flush-race resurrection ordering (Ingest(A), Rebuild(A), AddEvent(B)): the ordering that resurrected the flushed + // A under the old whole-buffer write. Atomic reducer buffering + identity consume keep B and don't resurrect A. + var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); + var eventA = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); + var eventB = FilterEventBuilder.CreateTestEvent(200, logName: Constants.LogNameTestLog); + + var state = new EventLogState { ContinuouslyUpdate = false, OpenLogs = ImmutableDictionary.Empty - .Add(Constants.LogNameTestLog, new OpenLogInfo(logA.Id, LogPathType.Channel)) - .Add("SecondTestLog", new OpenLogInfo(logB.Id, LogPathType.Channel)), + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)), + NewEventBuffer = [eventA], + AppliedFilter = new Filter(null, []) + }; + + var rawState = new RawEventStoreState + { + ByLog = ImmutableDictionary.Empty.Add(logData.Id, EventColumnStore.Empty) + }; + + var (effects, mockDispatcher, _) = CreateEffectsWithMutableState(() => state, () => rawState); + var pending = CaptureDispatchQueue(mockDispatcher); + + IReadOnlyList snapshot = [eventA]; + var rawByLog = new Dictionary> { [logData.Id] = snapshot }; + + pending.Enqueue(new IngestRawEventsAction(rawByLog, RawIngestMode.Prepend)); + pending.Enqueue(new RebuildDisplayViewsAction(rawByLog, BufferEntriesToConsume: snapshot)); + pending.Enqueue(new AddEventAction(eventB)); + + // Faithful drain that also applies the EventLogState buffer reducers, so the final buffer reflects both the + // mid-flush AddEvent(B) and the flush's targeted consume. + while (pending.Count > 0) + { + switch (pending.Dequeue()) + { + case IngestRawEventsAction ingest: + rawState = RawEventStoreReducers.ReduceIngestRawEvents(rawState, ingest); + break; + case AddEventAction add: + // Reducer runs before the effect in Fluxor; the effect no-ops for the non-live-tail path. + state = Reducers.ReduceAddEvent(state, add); + await effects.HandleAddEvent(add, mockDispatcher); + break; + case RebuildDisplayViewsAction rebuild: + await effects.HandleRebuildDisplayViews(rebuild, mockDispatcher); + break; + case NewEventBufferConsumedAction consumed: + state = Reducers.ReduceNewEventBufferConsumed(state, consumed); + break; + } + } + + // Assert: the flush consumed only A; B - buffered during the flush - survives with the count intact. + Assert.Single(state.NewEventBuffer); + Assert.Same(eventB, state.NewEventBuffer[0]); + } + + [Fact] + public async Task RebuildDisplayViews_WhenLogAbsentFromStore_SkipsFilterAndStillRebuildsOtherLogs() + { + // Guards the reordered guard (raw.TryGetValue before GetFilteredEvents): a log a concurrent close dropped is + // skipped without filtering, so a throwing filter can't abort the batch. Fails under the pre-fix operand order. + var openLog = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); + var closedLog = new EventLogData(Constants.LogNameApplication, LogPathType.Channel); + + var openEvent = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); + var closedEvent = FilterEventBuilder.CreateTestEvent(200, logName: Constants.LogNameApplication); + + var state = new EventLogState + { + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(openLog.Id, LogPathType.Channel)), NewEventBuffer = [], AppliedFilter = new Filter(null, []) }; - var logTableState = new LogTableState + // Only the open log survives in the raw store; the closed log was dropped by a concurrent CloseLog reducer. + var rawState = new RawEventStoreState { - RequestedOrderBy = ColumnName.Source, - DisplayListVersion = 3 + ByLog = ImmutableDictionary.Empty + .Add(openLog.Id, EventColumnStore.Build(new List { openEvent }, 0, 0)) }; - var (effects, mockDispatcher, _) = - CreateEffectsWithMutableState(() => eventLogState, () => rawState, logTableState); + var (effects, mockDispatcher, mockFilterService) = CreateEffectsWithMutableState(() => state, () => rawState); - await effects.Filtering.HandleUpdateTable(mockDispatcher); + // The filter throws if it is ever asked about the closed log's events; the reordered guard must never call it. + mockFilterService.GetFilteredEvents( + Arg.Is>(events => events != null && events.Any(e => e.Id == closedEvent.Id)), + Arg.Any()) + .Returns(_ => throw new InvalidOperationException("filter must not run for a log absent from the store")); - var requested = new SortContext(ColumnName.Source, true, null, false); - mockDispatcher.Received(1).Dispatch(Arg.Is(a => a != null && - a.Version == 3 && - a.Views.ContainsKey(logA.Id) && a.Views[logA.Id].HasContext(requested) && - a.Views.ContainsKey(logB.Id) && a.Views[logB.Id].HasContext(requested))); + var action = new RebuildDisplayViewsAction( + new Dictionary> + { + [closedLog.Id] = [closedEvent], + [openLog.Id] = [openEvent] + }, + BufferEntriesToConsume: new List { openEvent }); + + AppendTableEventsBatchAction? captured = null; + mockDispatcher.When(d => d.Dispatch(Arg.Any())) + .Do(call => captured = call.ArgAt(0)); + + // Act: must not throw despite the closed log's throwing filter. + await effects.HandleRebuildDisplayViews(action, mockDispatcher); + + // Assert: the open log rendered, the closed (absent) log was skipped, and the buffer was consumed. + Assert.NotNull(captured); + Assert.True(captured.ViewsByLog.ContainsKey(openLog.Id)); + Assert.False(captured.ViewsByLog.ContainsKey(closedLog.Id)); + mockDispatcher.Received(1).Dispatch(Arg.Any()); + } + + [Fact] + public void ReduceNewEventBufferConsumed_RemovesOnlyCapturedEntriesByReferenceIdentity() + { + // The flush consume removes exactly the captured snapshot (by reference identity) and leaves any event buffered + // afterward; a blanket clear would drop it. Newest events prepend, so the survivor sits at the head. + var eventA = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); + var eventB = FilterEventBuilder.CreateTestEvent(200, logName: Constants.LogNameTestLog); + + var state = new EventLogState { NewEventBuffer = [eventB, eventA] }; + + var result = Reducers.ReduceNewEventBufferConsumed(state, new NewEventBufferConsumedAction([eventA])); + + Assert.Single(result.NewEventBuffer); + Assert.Same(eventB, result.NewEventBuffer[0]); + Assert.False(result.NewEventBufferIsFull); + } + + [Fact] + public void ReduceNewEventBufferConsumed_UsesReferenceIdentity_NotValueEquality() + { + // Reference identity, not value equality: a double-delivered event yields two value-equal instances; the consume + // must remove only the captured instance and keep the other. + var captured = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); + var duplicate = captured with { }; + Assert.Equal(captured, duplicate); + Assert.NotSame(captured, duplicate); + + var state = new EventLogState { NewEventBuffer = [duplicate, captured] }; + + var result = Reducers.ReduceNewEventBufferConsumed( + state, new NewEventBufferConsumedAction([captured])); + + Assert.Single(result.NewEventBuffer); + Assert.Same(duplicate, result.NewEventBuffer[0]); } [Fact] @@ -2459,7 +2753,6 @@ private static EffectsHarness BuildHarness( eventLogState, rawEventStore, logTableState, - filterService, logger, closeCoordinator, concurrencyState, @@ -2542,6 +2835,15 @@ private static IReadOnlyList BuildReverseBatches(int total, int b return batches; } + // Records dispatches into a FIFO queue instead of letting them take effect, so a test can drain them after the + // producing effect returns - reproducing Fluxor's nested-dispatch ordering (a nested Dispatch is queued, not run inline). + private static Queue CaptureDispatchQueue(IDispatcher dispatcher) + { + var pending = new Queue(); + dispatcher.When(d => d.Dispatch(Arg.Any())).Do(call => pending.Enqueue(call.ArgAt(0))); + return pending; + } + private static (OpenLogEffects openLog, IDispatcher dispatcher, ILogWatcherService watcher) CreateEagerLoadEffects( IEventLogReaderFactory readerFactory, Func? resolveDelayMs = null) @@ -2861,6 +3163,29 @@ private static (EffectsHarness effects, return (effects, mockDispatcher, mockLogWatcherService, mockResolverCache, mockFilterService); } + // Drains the queue faithfully: IngestRawEventsAction applies the real raw-store reducer, RebuildDisplayViewsAction runs + // its real effect (so it sees the post-ingest store). Draining rather than hand-seeding is what catches the stale read. + private static async Task DrainDispatchQueueAsync( + Queue pending, + EffectsHarness effects, + IDispatcher dispatcher, + Func getRaw, + Action setRaw) + { + while (pending.Count > 0) + { + switch (pending.Dequeue()) + { + case IngestRawEventsAction ingest: + setRaw(RawEventStoreReducers.ReduceIngestRawEvents(getRaw(), ingest)); + break; + case RebuildDisplayViewsAction rebuild: + await effects.HandleRebuildDisplayViews(rebuild, dispatcher); + break; + } + } + } + private static IState EmptyRawStore() { var rawStore = Substitute.For>(); @@ -2943,6 +3268,9 @@ public Task HandleLoadEvents(LoadEventsAction action, IDispatcher dispatcher) => public Task HandleOpenLog(OpenLogAction action, IDispatcher dispatcher) => OpenLog.HandleOpenLog(action, dispatcher); + public Task HandleRebuildDisplayViews(RebuildDisplayViewsAction action, IDispatcher dispatcher) => + LogReload.HandleRebuildDisplayViews(action, dispatcher); + public Task HandleSetContinuouslyUpdate(SetContinuouslyUpdateAction action, IDispatcher dispatcher) => Filtering.HandleSetContinuouslyUpdate(action, dispatcher); } diff --git a/tests/Unit/EventLogExpert.Runtime.Tests/EventLog/EventLogStoreTests.cs b/tests/Unit/EventLogExpert.Runtime.Tests/EventLog/EventLogStoreTests.cs index aae5f119..289d06c7 100644 --- a/tests/Unit/EventLogExpert.Runtime.Tests/EventLog/EventLogStoreTests.cs +++ b/tests/Unit/EventLogExpert.Runtime.Tests/EventLog/EventLogStoreTests.cs @@ -20,16 +20,18 @@ public sealed class EventLogStoreTests public void BufferedEventsWorkflow_ShouldHandleCorrectly() { // Arrange - var state = new EventLogState { ContinuouslyUpdate = false }; + var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); - var events = new List + var state = new EventLogState { - FilterEventBuilder.CreateTestEvent(100), - FilterEventBuilder.CreateTestEvent(200) + ContinuouslyUpdate = false, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)) }; - // Act: Buffer events - state = Reducers.ReduceEventBuffered(state, new EventBufferedAction(events, false)); + // Act: buffer two events additively (newest first). + state = Reducers.ReduceAddEvent(state, new AddEventAction(FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog))); + state = Reducers.ReduceAddEvent(state, new AddEventAction(FilterEventBuilder.CreateTestEvent(200, logName: Constants.LogNameTestLog))); // Assert Assert.Equal(2, state.NewEventBuffer.Count); @@ -257,6 +259,96 @@ public void OpenAndCloseLog_ShouldMaintainStateConsistency() Assert.Equal(0, state.OpenLogCount); } + [Fact] + public void ReduceAddEvent_TwoSequentialEvents_BothPreservedNewestFirst() + { + // Arrange + var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); + var state = new EventLogState + { + ContinuouslyUpdate = false, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)) + }; + + var first = FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog); + var second = FilterEventBuilder.CreateTestEvent(200, logName: Constants.LogNameTestLog); + + // Act: each add composes against current state, so the second cannot clobber the first (the pre-fix whole-buffer + // effect write could). + state = Reducers.ReduceAddEvent(state, new AddEventAction(first)); + state = Reducers.ReduceAddEvent(state, new AddEventAction(second)); + + // Assert + Assert.Equal(2, state.NewEventBuffer.Count); + Assert.Same(second, state.NewEventBuffer[0]); + Assert.Same(first, state.NewEventBuffer[1]); + } + + [Fact] + public void ReduceAddEvent_WhenContinuouslyUpdating_DoesNotBuffer() + { + // Arrange + var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); + var state = new EventLogState + { + ContinuouslyUpdate = true, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)) + }; + + // Act + var result = Reducers.ReduceAddEvent( + state, new AddEventAction(FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog))); + + // Assert: buffering is the live-tail effect's job when continuously updating, not the reducer's. + Assert.Same(state, result); + Assert.Empty(result.NewEventBuffer); + } + + [Fact] + public void ReduceAddEvent_WhenLogNotOpen_DoesNotBuffer() + { + // Arrange + var state = new EventLogState { ContinuouslyUpdate = false }; + + // Act + var result = Reducers.ReduceAddEvent( + state, new AddEventAction(FilterEventBuilder.CreateTestEvent(100, logName: Constants.LogNameTestLog))); + + // Assert + Assert.Same(state, result); + Assert.Empty(result.NewEventBuffer); + } + + [Fact] + public void ReduceAddEvent_WhenLogOpenAndNotContinuouslyUpdating_PrependsEventAndRecomputesFull() + { + // Arrange: buffer already at MaxNewEvents - 1 so the next add flips the full flag. + var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); + var existing = Enumerable.Range(0, EventLogState.MaxNewEvents - 1) + .Select(i => FilterEventBuilder.CreateTestEvent(i, logName: Constants.LogNameTestLog)) + .ToList(); + + var state = new EventLogState + { + ContinuouslyUpdate = false, + OpenLogs = ImmutableDictionary.Empty + .Add(Constants.LogNameTestLog, new OpenLogInfo(logData.Id, LogPathType.Channel)), + NewEventBuffer = existing + }; + + var newEvent = FilterEventBuilder.CreateTestEvent(9999, logName: Constants.LogNameTestLog); + + // Act + var result = Reducers.ReduceAddEvent(state, new AddEventAction(newEvent)); + + // Assert: prepended (newest first) and the full flag recomputed. + Assert.Equal(EventLogState.MaxNewEvents, result.NewEventBuffer.Count); + Assert.Same(newEvent, result.NewEventBuffer[0]); + Assert.True(result.NewEventBufferIsFull); + } + [Fact] public void ReduceApplyFilter_WhenFilterChanged_ShouldUpdateFilter() { @@ -423,57 +515,20 @@ public void ReduceCloseLog_ShouldRemoveSpecifiedLog() } [Fact] - public void ReduceEventBuffered_ShouldUpdateBufferAndFullFlag() - { - // Arrange - var state = new EventLogState(); - - var events = new List - { - FilterEventBuilder.CreateTestEvent(100), - FilterEventBuilder.CreateTestEvent(200) - }; - - var action = new EventBufferedAction(events, true); - - // Act - var newState = Reducers.ReduceEventBuffered(state, action); - - // Assert - Assert.Equal(2, newState.NewEventBuffer.Count); - Assert.True(newState.NewEventBufferIsFull); - } - - [Fact] - public void ReduceEventBuffered_WhenNotFull_ShouldSetFullFlagFalse() + public void ReduceLoadEventsPartial_WhenLogIdDoesNotMatch_ShouldReturnStateUnchanged() { // Arrange - var state = new EventLogState { NewEventBufferIsFull = true }; - var events = new List { FilterEventBuilder.CreateTestEvent(100) }; - var action = new EventBufferedAction(events, false); - - // Act - var newState = Reducers.ReduceEventBuffered(state, action); - - // Assert - Assert.False(newState.NewEventBufferIsFull); - } - - [Fact] - public void ReduceLoadEvents_WhenLogIdDoesNotMatch_ShouldReturnStateUnchanged() - { - // Arrange: open a log, then create stale logData with a different ID var state = new EventLogState(); state = Reducers.ReduceOpenLog(state, new OpenLogAction(Constants.LogNameTestLog, LogPathType.Channel)); - // Create stale logData with a new ID (simulating a previous load instance) var staleLogData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); var events = ImmutableArray.Create(FilterEventBuilder.CreateTestEvent(100)); - // Act: stale LoadEvents with mismatched ID - var newState = Reducers.ReduceLoadEvents(state, new LoadEventsAction(staleLogData, events)); + // Act: stale partial with mismatched ID + var newState = Reducers.ReduceLoadEventsPartial(state, + new LoadEventsPartialAction(staleLogData, events)); // Assert: state unchanged, original log preserved with its ID var originalId = state.OpenLogs[Constants.LogNameTestLog].Id; @@ -483,36 +538,36 @@ public void ReduceLoadEvents_WhenLogIdDoesNotMatch_ShouldReturnStateUnchanged() } [Fact] - public void ReduceLoadEvents_WhenLogNotInOpenLogs_ShouldReturnStateUnchanged() + public void ReduceLoadEventsPartial_WhenLogNotInOpenLogs_ShouldReturnStateUnchanged() { // Arrange: no logs open var state = new EventLogState(); var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); var events = ImmutableArray.Create(FilterEventBuilder.CreateTestEvent(100)); - // Act: stale LoadEvents arrives for a closed log - var newState = Reducers.ReduceLoadEvents(state, new LoadEventsAction(logData, events)); + // Act + var newState = Reducers.ReduceLoadEventsPartial(state, + new LoadEventsPartialAction(logData, events)); - // Assert: state unchanged, log NOT resurrected + // Assert Assert.Same(state, newState); - Assert.Equal(0, newState.OpenLogCount); } [Fact] - public void ReduceLoadEventsPartial_WhenLogIdDoesNotMatch_ShouldReturnStateUnchanged() + public void ReduceLoadEvents_WhenLogIdDoesNotMatch_ShouldReturnStateUnchanged() { - // Arrange + // Arrange: open a log, then create stale logData with a different ID var state = new EventLogState(); state = Reducers.ReduceOpenLog(state, new OpenLogAction(Constants.LogNameTestLog, LogPathType.Channel)); + // Create stale logData with a new ID (simulating a previous load instance) var staleLogData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); var events = ImmutableArray.Create(FilterEventBuilder.CreateTestEvent(100)); - // Act: stale partial with mismatched ID - var newState = Reducers.ReduceLoadEventsPartial(state, - new LoadEventsPartialAction(staleLogData, events)); + // Act: stale LoadEvents with mismatched ID + var newState = Reducers.ReduceLoadEvents(state, new LoadEventsAction(staleLogData, events)); // Assert: state unchanged, original log preserved with its ID var originalId = state.OpenLogs[Constants.LogNameTestLog].Id; @@ -522,19 +577,19 @@ public void ReduceLoadEventsPartial_WhenLogIdDoesNotMatch_ShouldReturnStateUncha } [Fact] - public void ReduceLoadEventsPartial_WhenLogNotInOpenLogs_ShouldReturnStateUnchanged() + public void ReduceLoadEvents_WhenLogNotInOpenLogs_ShouldReturnStateUnchanged() { // Arrange: no logs open var state = new EventLogState(); var logData = new EventLogData(Constants.LogNameTestLog, LogPathType.Channel); var events = ImmutableArray.Create(FilterEventBuilder.CreateTestEvent(100)); - // Act - var newState = Reducers.ReduceLoadEventsPartial(state, - new LoadEventsPartialAction(logData, events)); + // Act: stale LoadEvents arrives for a closed log + var newState = Reducers.ReduceLoadEvents(state, new LoadEventsAction(logData, events)); - // Assert + // Assert: state unchanged, log NOT resurrected Assert.Same(state, newState); + Assert.Equal(0, newState.OpenLogCount); } [Fact] @@ -640,36 +695,36 @@ public void ReduceSelectEvent_WhenEventNotSelected_ShouldAddEvent() } [Fact] - public void ReduceSelectEvent_WhenMultiSelect_ShouldAddToExisting() + public void ReduceSelectEvent_WhenMultiSelectAndEventAlreadySelected_ShouldRemoveEvent() { // Arrange - var existing = Entry(0); - var state = new EventLogState { Selection = [existing] }; - var incoming = Entry(1); - var action = new SelectEventAction(incoming, true); + var selection = Entry(0); + var state = new EventLogState { Selection = [selection] }; + var action = new SelectEventAction(selection, true); // Act var newState = Reducers.ReduceSelectEvent(state, action); // Assert - Assert.Equal(2, newState.Selection.Count); - Assert.Contains(existing, newState.Selection); - Assert.Contains(incoming, newState.Selection); + Assert.Empty(newState.Selection); } [Fact] - public void ReduceSelectEvent_WhenMultiSelectAndEventAlreadySelected_ShouldRemoveEvent() + public void ReduceSelectEvent_WhenMultiSelect_ShouldAddToExisting() { // Arrange - var selection = Entry(0); - var state = new EventLogState { Selection = [selection] }; - var action = new SelectEventAction(selection, true); + var existing = Entry(0); + var state = new EventLogState { Selection = [existing] }; + var incoming = Entry(1); + var action = new SelectEventAction(incoming, true); // Act var newState = Reducers.ReduceSelectEvent(state, action); // Assert - Assert.Empty(newState.Selection); + Assert.Equal(2, newState.Selection.Count); + Assert.Contains(existing, newState.Selection); + Assert.Contains(incoming, newState.Selection); } [Fact]