diff --git a/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewScopeState.cs b/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewScopeState.cs index f9cde260..9b2b96c3 100644 --- a/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewScopeState.cs +++ b/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewScopeState.cs @@ -119,6 +119,8 @@ public bool ScopeEquals(IReadOnlyCollection scopeLogs) return matched == _scopeLogs.Count; } + public void SetCoverage(in LogGeneration key, int coveredCount) => _coverage[key] = coveredCount; + public bool TrySetScope(IReadOnlyCollection scopeLogs, long scopeVersion) { if (scopeVersion < ScopeVersion) { return false; } diff --git a/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewShadowEffects.cs b/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewShadowEffects.cs index 64d5a562..8e96a3c5 100644 --- a/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewShadowEffects.cs +++ b/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewShadowEffects.cs @@ -71,7 +71,7 @@ public Task HandleIngestRawEvents(IngestRawEventsAction action, IDispatcher disp { Sync(); - foreach (EventLogId logId in action.EventsByLog.Keys) { Reconcile(logId); } + foreach (EventLogId logId in action.EventsByLog.Keys) { Reconcile(logId, action.Mode == RawIngestMode.Replace); } }); [EffectMethod(typeof(LoadColumnsCompletedAction))] @@ -82,7 +82,7 @@ public Task HandleLoadEvents(LoadEventsAction action, IDispatcher dispatcher) => Shadow(() => { Sync(); - Reconcile(action.LogData.Id); + Reconcile(action.LogData.Id, isReplace: true); }); [EffectMethod] @@ -90,7 +90,7 @@ public Task HandleLoadEventsPartial(LoadEventsPartialAction action, IDispatcher Shadow(() => { Sync(); - Reconcile(action.LogData.Id); + Reconcile(action.LogData.Id, isReplace: false); }); [EffectMethod(typeof(MoveTabToGroupAction))] @@ -140,11 +140,11 @@ public Task HandleOrderedViewDisplayFaulted(OrderedViewDisplayFaultedAction acti [EffectMethod(typeof(ToggleSortingAction))] public Task HandleToggleSorting(IDispatcher dispatcher) => Shadow(Sync); - private void Reconcile(EventLogId logId) + private void Reconcile(EventLogId logId, bool isReplace) { if (_rawEventStore.Value.ByLog.TryGetValue(logId, out var store)) { - _writer.EnqueueReconcile(logId, store.CreateReader(logId)); + _writer.EnqueueReconcile(logId, store.CreateReader(logId), isReplace); } } diff --git a/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewState.cs b/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewState.cs index 676502ae..f59507d4 100644 --- a/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewState.cs +++ b/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewState.cs @@ -51,6 +51,7 @@ internal sealed class OrderedViewState private OrderedViewSnapshot _current = OrderedViewSnapshot.Empty; private long _generation; private bool _holdIngest; + private bool _liveIndexInvalidated; private ChunkedOrderIndex _index; private Func _predicate = static (_, _) => true; private long _publishVersion; @@ -70,6 +71,8 @@ internal OrderedViewState() public long Generation => Volatile.Read(ref _generation); + public bool LiveIndexInvalidated => _liveIndexInvalidated; + public int RowCount => _scopeState.FreezeCoverage().RowCount; public long ScopeVersion => _scopeState.ScopeVersion; @@ -117,7 +120,7 @@ public bool CanRestampAdopted( if (!_activeGeneration.TryGetValue(logId, out int active) || active != reader.Generation) { return false; } - if (_scopeState.Coverage(new LogGeneration(logId, reader.Generation)) < reader.Count) { return false; } + if (_scopeState.Coverage(new LogGeneration(logId, reader.Generation)) != reader.Count) { return false; } } return true; @@ -138,6 +141,7 @@ public OrderedViewSnapshot Clear() _activeContext = new SortContext(null, false, null, false); _requestedContext = new SortContext(null, false, null, false); _holdIngest = false; + _liveIndexInvalidated = false; _index = new ChunkedOrderIndex(OrderKeyComparerFactory.Create(_activeContext, _liveResolver)); return PublishWith(FreezeReaders()); @@ -163,17 +167,39 @@ public void NotifyRebuildFailed(RebuildRequest request) public OrderedViewSnapshot Publish() => PublishWith(FreezeReaders()); - public bool ReconcileLog(EventLogId logId, IEventColumnReader reader) + public bool ReconcileLog(EventLogId logId, IEventColumnReader reader) => ReconcileLog(logId, reader, out _); + + public bool ReconcileLog(EventLogId logId, IEventColumnReader reader, out bool requiresRebuild) => + ReconcileLog(logId, reader, isReplace: false, out requiresRebuild); + + public bool ReconcileLog(EventLogId logId, IEventColumnReader reader, bool isReplace, out bool requiresRebuild) { - if (!TryAdmitReader(logId, reader, out LogGeneration readerKey, out bool sameCountReplace)) { return false; } + requiresRebuild = false; + + if (!TryAdmitReader(logId, reader, isReplace, out LogGeneration readerKey, out bool contentReplaced)) { return false; } int from = _scopeState.Coverage(readerKey); - _scopeState.AdvanceCoverage(readerKey, reader.Count); + if (contentReplaced) + { + _scopeState.SetCoverage(readerKey, reader.Count); + } + else + { + _scopeState.AdvanceCoverage(readerKey, reader.Count); + } + + bool activeGenerationMatch = + _activeGeneration.TryGetValue(logId, out int active) && active == reader.Generation; + + if (contentReplaced && _adoptedScope.Includes(logId) && activeGenerationMatch && reader.Count < from) + { + _liveIndexInvalidated = true; + } bool mutated = false; - if (_adoptedScope.Includes(logId) && !_holdIngest && IsCurrent(readerKey, _activeGeneration)) + if (_adoptedScope.Includes(logId) && !_holdIngest && !_liveIndexInvalidated && activeGenerationMatch) { for (int index = from; index < reader.Count; index++) { @@ -187,14 +213,14 @@ public bool ReconcileLog(EventLogId logId, IEventColumnReader reader) } } - bool displaysThisGeneration = reader.Count > 0 && - _adoptedScope.Includes(logId) && - _activeGeneration.TryGetValue(logId, out int active) && - active == reader.Generation; + bool displaysThisGeneration = reader.Count > 0 && activeGenerationMatch && _adoptedScope.Includes(logId); + + requiresRebuild = contentReplaced && activeGenerationMatch && + (_adoptedScope.Includes(logId) || _scopeState.Includes(logId)); return mutated || (displaysThisGeneration && !_adoptedInScope.Contains(readerKey)) || - (displaysThisGeneration && sameCountReplace); + (displaysThisGeneration && contentReplaced); } public bool ReconcileScopeReaders(IReadOnlyDictionary scopeReaders) @@ -225,7 +251,7 @@ public void RestoreRequestedFromAdopted() public bool SeedScopeReader(EventLogId logId, IEventColumnReader reader) { - bool admitted = TryAdmitReader(logId, reader, out LogGeneration readerKey, out _); + bool admitted = TryAdmitReader(logId, reader, isReplace: false, out LogGeneration readerKey, out _); if ((admitted || _latestReaders.ContainsKey(readerKey)) && reader.Generation > _requestedGeneration.GetValueOrDefault(logId, int.MinValue)) @@ -266,14 +292,19 @@ public AdoptOutcome TryAdoptRebuild(RebuildRequest request, ChunkedOrderIndex re if (!IsCurrent(key, _requestedGeneration)) { continue; } + if (!commitResolver.TryResolve(new EventLocator(key.LogId, key.Generation, 0), out IEventColumnReader? reader)) + { + continue; + } + int from = request.Coverage.CoverageOf(key); - int to = _scopeState.Coverage(key); + int to = Math.Min(_scopeState.Coverage(key), reader.Count); for (int index = from; index < to; index++) { var locator = new EventLocator(key.LogId, key.Generation, index); - if (request.Predicate(locator, commitResolver.Resolve(locator))) + if (request.Predicate(locator, reader)) { rebuilt.Insert(new OrderKey(locator)); } @@ -298,6 +329,7 @@ public AdoptOutcome TryAdoptRebuild(RebuildRequest request, ChunkedOrderIndex re _predicate = request.Predicate; _activeContext = request.Context; _holdIngest = false; + _liveIndexInvalidated = false; _scopeState.EvictOutOfScope(_adoptedScope, _activeGeneration); EvictGenerationsOutOfScope(); @@ -341,13 +373,20 @@ internal static ChunkedOrderIndex BuildIndex( if (!IsCurrent(key, request.RequestedGeneration)) { continue; } - for (int index = 0; index < covered; index++) + if (!request.BeginResolver.TryResolve(new EventLocator(key.LogId, key.Generation, 0), out IEventColumnReader? reader)) + { + continue; + } + + int limit = Math.Min(covered, reader.Count); + + for (int index = 0; index < limit; index++) { if ((examined++ & CancellationCheckMask) == 0) { cancellationToken.ThrowIfCancellationRequested(); } var locator = new EventLocator(key.LogId, key.Generation, index); - if (request.Predicate(locator, request.BeginResolver.Resolve(locator))) + if (request.Predicate(locator, reader)) { rebuilt.Insert(new OrderKey(locator)); } @@ -448,7 +487,12 @@ private static ChunkedOrderIndex BuildSingleLogBulk( if (!IsCurrent(key, request.RequestedGeneration)) { continue; } - keys.Add((key, covered)); + if (!request.BeginResolver.TryResolve(new EventLocator(key.LogId, key.Generation, 0), out IEventColumnReader? reader)) + { + continue; + } + + keys.Add((key, Math.Min(covered, reader.Count))); } return keys; @@ -634,7 +678,7 @@ private long MeasureTail(RebuildRequest request) if (!IsCurrent(key, _requestedGeneration)) { continue; } - tail += _scopeState.Coverage(key) - request.Coverage.CoverageOf(key); + tail += Math.Max(0, _scopeState.Coverage(key) - request.Coverage.CoverageOf(key)); } return tail; @@ -674,10 +718,10 @@ private OrderedViewSnapshot PublishWith(IReaderResolver frozenResolver) } private bool TryAdmitReader( - EventLogId logId, IEventColumnReader reader, out LogGeneration readerKey, out bool sameCountReplace) + EventLogId logId, IEventColumnReader reader, bool isReplace, out LogGeneration readerKey, out bool contentReplaced) { readerKey = new LogGeneration(logId, reader.Generation); - sameCountReplace = false; + contentReplaced = false; if (!_scopeState.Includes(logId)) { return false; } @@ -687,12 +731,14 @@ private bool TryAdmitReader( if (_latestReaders.TryGetValue(readerKey, out var existing)) { - bool strictlyNewer = reader.Count > existing.Count || - (reader.Count == existing.Count && reader.ContentVersion > existing.ContentVersion); + bool admit = reader.Count > existing.Count || + (reader.ContentVersion > existing.ContentVersion && + (reader.Count >= existing.Count || isReplace)); - if (!strictlyNewer) { return false; } + if (!admit) { return false; } - sameCountReplace = reader.Count == existing.Count; + contentReplaced = reader.ContentVersion > existing.ContentVersion && + (reader.Count <= existing.Count || isReplace); } _latestReaders[readerKey] = reader; diff --git a/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewWriter.cs b/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewWriter.cs index 5a749149..e39cb52d 100644 --- a/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewWriter.cs +++ b/src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewWriter.cs @@ -13,6 +13,7 @@ internal sealed class OrderedViewWriter : IAsyncDisposable { private const int DefaultTailBreachLimit = 3; private const int DefaultTailReplayBudget = 50_000; + private const int InvalidatedRebuildRetryLimit = 3; private readonly Task? _cadence; private readonly Channel _commandChannel; @@ -29,34 +30,24 @@ internal sealed class OrderedViewWriter : IAsyncDisposable private int _adoptedGeneration; private ViewIdentity? _adoptedIdentity; private EventLogId? _adoptedLog; - private int _adoptedScopeLogCount; - private long _adoptedSequence; - private int _buildsStarted; - private (Task Task, CancellationTokenSource Cts)? _currentBuild; private RebuildRequest? _desiredBuild; private bool _dirty; - private bool _faultAnnounced; - private volatile Exception? _faulted; - private long _highestSequence; + private int _invalidatedRebuildRetries; private long _lastUpdateVersion; - private PendingBuild? _pending; private Filter _pendingFilter; private int _pendingRebuilds; - private bool _rebuildRequired; - + private bool _replaceAwaitingRebuild; private bool _seededRowsAwaitingBuild; - private int _sincePublish; - private ImmutableHashSet? _singleLogInScope; private LogGeneration? _singleLogInScopeKey; private int _tailBreaches; @@ -152,7 +143,10 @@ public void EnqueueClear(ViewIdentity identity, long sequence) => public void EnqueueFlush() => _commandChannel.Writer.TryWrite(Command.ForFlush()); public void EnqueueReconcile(EventLogId logId, IEventColumnReader reader) => - _commandChannel.Writer.TryWrite(Command.ForReconcile(logId, reader)); + EnqueueReconcile(logId, reader, isReplace: false); + + public void EnqueueReconcile(EventLogId logId, IEventColumnReader reader, bool isReplace) => + _commandChannel.Writer.TryWrite(Command.ForReconcile(logId, reader, isReplace)); public void EnqueueRemoveLog(EventLogId logId) => _commandChannel.Writer.TryWrite(Command.ForRemoveLog(logId)); @@ -252,6 +246,17 @@ private void CompleteRebuild() _desiredBuild = null; StartRebuild(desired); } + else if (_replaceAwaitingRebuild && _pendingRebuilds == 0) + { + _replaceAwaitingRebuild = false; + ForceRebuild(); + } + else if (_state.LiveIndexInvalidated && _pendingRebuilds == 0 && + _invalidatedRebuildRetries < InvalidatedRebuildRetryLimit) + { + _invalidatedRebuildRetries++; + ForceRebuild(); + } if (_pendingRebuilds != 0 || _pendingDrain.Count <= 0) { @@ -281,10 +286,11 @@ private void Dispatch(in Command command) if (command.Reader is { } reconcileReader) { bool reconciled; + bool requiresRebuild; try { - reconciled = _state.ReconcileLog(command.LogId, reconcileReader); + reconciled = _state.ReconcileLog(command.LogId, reconcileReader, command.IsReplace, out requiresRebuild); } catch (Exception reconcileFault) { @@ -294,12 +300,25 @@ private void Dispatch(in Command command) break; } - if (reconciled) + if (requiresRebuild) + { + if (_state.LiveIndexInvalidated) { _invalidatedRebuildRetries = 0; } + + if (_pendingRebuilds > 0) + { + _replaceAwaitingRebuild = true; + _state.SupersedeInFlight(); + } + else + { + ForceRebuild(); + } + } + else if (reconciled) { _dirty = true; - if (_state.Current.Count == 0) { PublishNow(); } - else if (++_sincePublish >= _publishEvery) { PublishNow(); } + if (_state.Current.Count == 0 || ++_sincePublish >= _publishEvery) { PublishNow(); } } } @@ -327,7 +346,9 @@ private void Dispatch(in Command command) _rebuildRequired = false; _faultAnnounced = false; _seededRowsAwaitingBuild = false; + _replaceAwaitingRebuild = false; _tailBreaches = 0; + _invalidatedRebuildRetries = 0; break; case CommandKind.Adopt: @@ -335,6 +356,11 @@ private void Dispatch(in Command command) { AdoptOutcome outcome = AdoptOutcome.DroppedStale; + // Latch before TryAdoptRebuild clears it on a successful adopt: a coalesced view request may have + // seeded a grow the seed path could not tell apart from a replace, so ANY seeded outcome (adopt + // included) must re-derive the order, not only the non-adopt outcomes. + bool seededThisBuild = _seededRowsAwaitingBuild; + if (command is { Request: { } request, Rebuilt: { } rebuilt }) { outcome = _state.TryAdoptRebuild(request, rebuilt, _tailReplayBudget, _tailBreaches < _tailBreachLimit); @@ -364,6 +390,7 @@ private void Dispatch(in Command command) _rebuildRequired = false; _faultAnnounced = false; _seededRowsAwaitingBuild = false; + _invalidatedRebuildRetries = 0; } } @@ -372,9 +399,12 @@ private void Dispatch(in Command command) _tailBreaches++; RebindAndStart(_state.CaptureScopeReseed()); } - else if (outcome != AdoptOutcome.Adopted && _seededRowsAwaitingBuild) + else if (seededThisBuild) { - // rows a retag seeded onto it, exactly as a throwing one would. + // A coalesced view request seeded rows onto this build. A non-adopt outcome never placed them; + // a successful adopt placed them by tail-replay, which is correct for an append but leaves the + // prior rows stale if the seed was actually a re-resolution (the seed path cannot tell them + // apart). Re-derive from the reseeded readers either way. (AbandonedTail already recaptures.) RequireRebuild(); } } @@ -479,6 +509,7 @@ private void DispatchViewRequest(ViewRequest request) if (!_rebuildRequired && !_seededRowsAwaitingBuild && + !_replaceAwaitingRebuild && _adoptedIdentity is { } adoptedIdentity && request.Identity.CoversSameViewAs(adoptedIdentity) && _state.CanRestampAdopted(request.ScopeLogs, request.ScopeReaders)) @@ -522,8 +553,17 @@ private void FailPendingDrain() _pendingDrain.Clear(); } + private void ForceRebuild() + { + _rebuildRequired = true; + + RebindAndStart(_state.CaptureScopeReseed()); + } + private void PublishNow() { + if (_state.LiveIndexInvalidated) { return; } + _state.Publish(); _dirty = false; _sincePublish = 0; @@ -584,9 +624,7 @@ private void RequireRebuild() { if (_rebuildRequired) { return; } - _rebuildRequired = true; - - RebindAndStart(_state.CaptureScopeReseed()); + ForceRebuild(); } private async Task RunAsync() @@ -685,14 +723,16 @@ private readonly struct Command public ViewRequest? ViewRequest { get; private init; } + public bool IsReplace { get; private init; } + public static Command ForViewRequest(ViewRequest request) => new() { Kind = CommandKind.ViewRequest, ViewRequest = request }; public static Command ForReset(EventLogId logId, int generation) => new() { Kind = CommandKind.Reset, LogId = logId, Generation = generation }; - public static Command ForReconcile(EventLogId logId, IEventColumnReader reader) => - new() { Kind = CommandKind.Reconcile, LogId = logId, Reader = reader }; + public static Command ForReconcile(EventLogId logId, IEventColumnReader reader, bool isReplace) => + new() { Kind = CommandKind.Reconcile, LogId = logId, Reader = reader, IsReplace = isReplace }; public static Command ForRemoveLog(EventLogId logId) => new() { Kind = CommandKind.RemoveLog, LogId = logId }; diff --git a/tests/Unit/EventLogExpert.Runtime.Tests/LogTable/OrderedView/OrderedViewWriterFaultStateTests.cs b/tests/Unit/EventLogExpert.Runtime.Tests/LogTable/OrderedView/OrderedViewWriterFaultStateTests.cs index 4ae31fd1..348638c3 100644 --- a/tests/Unit/EventLogExpert.Runtime.Tests/LogTable/OrderedView/OrderedViewWriterFaultStateTests.cs +++ b/tests/Unit/EventLogExpert.Runtime.Tests/LogTable/OrderedView/OrderedViewWriterFaultStateTests.cs @@ -326,13 +326,20 @@ public async Task ATailReplayThatThrows_DoesNotSuppressPublishing_BecauseItDamag EventLogId logId = EventLogId.Create(); int raised = 0; + using var buildEntered = new ManualResetEventSlim(false); + using var releaseBuild = new ManualResetEventSlim(false); + int gateArmed = 1; + await using var writer = new OrderedViewWriter(publishEvery: 1, publishIntervalMs: 0); writer.Updated += _ => Interlocked.Increment(ref raised); writer.EnqueueReconcile(logId, Reader(logId, count: 3000)); - writer.EnqueueViewRequest(Request(logId, static (locator, _) => locator.Index >= 3000 ? throw new InvalidOperationException("tail") : true)); + writer.EnqueueViewRequest(Request(logId, Predicate)); + Assert.True(buildEntered.Wait(SignalTimeout, TestContext.Current.CancellationToken)); + writer.EnqueueReconcile(logId, Reader(logId, count: 3200)); + releaseBuild.Set(); await writer.DrainAsync().WaitAsync(Timeout, TestContext.Current.CancellationToken); Assert.NotNull(writer.Faulted); @@ -342,7 +349,27 @@ public async Task ATailReplayThatThrows_DoesNotSuppressPublishing_BecauseItDamag writer.EnqueueReconcile(logId, Reader(logId, count: 3400)); await writer.DrainAsync().WaitAsync(Timeout, TestContext.Current.CancellationToken); - Assert.True(Volatile.Read(ref raised) > before, "publishing must continue after a fault that damaged nothing"); + // The recovery publish raises Updated in RaiseUpdateIfAdvanced, which runs AFTER the drain's TrySetResult + // completes the await, so wait for the raise rather than reading the counter the instant the drain returns. + Assert.True( + SpinWait.SpinUntil(() => Volatile.Read(ref raised) > before, SignalTimeout), + "publishing must continue after a fault that damaged nothing"); + + return; + + // Gate the build at its first row so the growing tail (3200) is enqueued BEFORE the build's adopt is queued. + // FIFO delivery then guarantees the tail is covered when the adopt runs, so the failure lands in the tail + // REPLAY (which damaged nothing) rather than in a live insert (which would legitimately require a rebuild). + bool Predicate(EventLocator locator, IEventColumnReader reader) + { + if (locator.Index == 0 && Interlocked.Exchange(ref gateArmed, 0) == 1) + { + buildEntered.Set(); + releaseBuild.Wait(SignalTimeout); + } + + return locator.Index >= 3000 ? throw new InvalidOperationException("tail") : true; + } } [Fact] diff --git a/tests/Unit/EventLogExpert.Runtime.Tests/LogTable/OrderedView/ReconcileLogStaleOrderTests.cs b/tests/Unit/EventLogExpert.Runtime.Tests/LogTable/OrderedView/ReconcileLogStaleOrderTests.cs new file mode 100644 index 00000000..fef26349 --- /dev/null +++ b/tests/Unit/EventLogExpert.Runtime.Tests/LogTable/OrderedView/ReconcileLogStaleOrderTests.cs @@ -0,0 +1,468 @@ +// // Copyright (c) Microsoft Corporation. +// // Licensed under the MIT License. + +using EventLogExpert.Eventing.Common.Channels; +using EventLogExpert.Eventing.Common.EventLogs; +using EventLogExpert.Eventing.Common.Events; +using EventLogExpert.Runtime.LogTable; +using EventLogExpert.Runtime.LogTable.OrderedView; +using EventLogExpert.Runtime.Tests.LogTable.TestSupport; + +namespace EventLogExpert.Runtime.Tests.LogTable.OrderedView; + +public sealed class ReconcileLogStaleOrderTests +{ + private static readonly SortContext s_sourceAscending = new(ColumnName.Source, false, null, false); + + [Fact] + public void ReconcileLog_GrowAppendHigherContentVersion_DoesNotSignalRebuild() + { + EventLogId logId = EventLogId.Create(); + IEventColumnReader original = Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1)); + IEventColumnReader grownAppend = Reader(logId, contentVersion: 1, ("AAA", 0), ("BBB", 1), ("CCC", 2)); + + var state = new OrderedViewState(); + AdoptSourceView(state, [logId], new Dictionary { [logId] = original }); + + state.ReconcileLog(logId, grownAppend, isReplace: false, out bool requiresRebuild); + Assert.False(requiresRebuild); + } + + [Fact] + public void ReconcileLog_GrowReplaceHigherContentVersion_SignalsRebuildAndReorders() + { + EventLogId logId = EventLogId.Create(); + IEventColumnReader original = Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1)); + IEventColumnReader grownReplace = Reader(logId, contentVersion: 1, ("CCC", 0), ("BBB", 1), ("AAA", 2)); + + var state = new OrderedViewState(); + AdoptSourceView(state, [logId], new Dictionary { [logId] = original }); + Assert.Equal(0, state.Current.At(0).Locator.Index); + + Assert.True(state.ReconcileLog(logId, grownReplace, isReplace: true, out bool requiresRebuild)); + Assert.True(requiresRebuild); + + RebuildRequest reseed = state.CaptureScopeReseed(); + Assert.True(state.TryAdoptRebuild(reseed, OrderedViewState.BuildIndex(reseed, CancellationToken.None))); + Assert.Equal(3, state.Current.Count); + Assert.Equal(2, state.Current.At(0).Locator.Index); + Assert.Equal(1, state.Current.At(1).Locator.Index); + Assert.Equal(0, state.Current.At(2).Locator.Index); + } + + [Fact] + public void ReconcileLog_HigherContentVersionWithMoreRows_DoesNotSignalRebuild() + { + EventLogId logId = EventLogId.Create(); + var state = new OrderedViewState(); + + Assert.True(state.ReconcileLog(logId, Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1)))); + + state.ReconcileLog( + logId, + Reader(logId, contentVersion: 1, ("AAA", 0), ("BBB", 1), ("CCC", 2)), + out bool requiresRebuild); + Assert.False(requiresRebuild); + } + + [Fact] + public void ReconcileLog_LowerCountHigherContentVersion_IsDroppedAndDoesNotSignalRebuild() + { + EventLogId logId = EventLogId.Create(); + var state = new OrderedViewState(); + + Assert.True(state.ReconcileLog(logId, Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1), ("CCC", 2)))); + + Assert.False(state.ReconcileLog( + logId, + Reader(logId, contentVersion: 1, ("BBB", 0), ("AAA", 1)), + out bool requiresRebuild)); + Assert.False(requiresRebuild); + } + + [Fact] + public void ReconcileLog_SameCountHigherContentVersion_SignalsRebuildAndReorders() + { + EventLogId logId = EventLogId.Create(); + IEventColumnReader original = Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1)); + IEventColumnReader reresolved = Reader(logId, contentVersion: 1, ("BBB", 0), ("AAA", 1)); + + var state = new OrderedViewState(); + AdoptSourceView(state, [logId], new Dictionary { [logId] = original }); + Assert.Equal(0, state.Current.At(0).Locator.Index); + + Assert.True(state.ReconcileLog(logId, reresolved, out bool requiresRebuild)); + Assert.True(requiresRebuild); + + RebuildRequest reseed = state.CaptureScopeReseed(); + Assert.True(state.TryAdoptRebuild(reseed, OrderedViewState.BuildIndex(reseed, CancellationToken.None))); + Assert.Equal(1, state.Current.At(0).Locator.Index); + } + + [Fact] + public void ReconcileLog_SameCountReplaceInDefaultOpenScope_SignalsRebuild() + { + EventLogId logId = EventLogId.Create(); + var state = new OrderedViewState(); + + Assert.True(state.ReconcileLog(logId, Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1)))); + + Assert.True(state.ReconcileLog( + logId, + Reader(logId, contentVersion: 1, ("BBB", 0), ("AAA", 1)), + out bool requiresRebuild)); + Assert.True(requiresRebuild); + } + + [Fact] + public void ReconcileLog_SameCountReplaceOfLiveInsertedLogNotYetPublished_SignalsRebuild() + { + EventLogId visible = EventLogId.Create(); + EventLogId lateLoader = EventLogId.Create(); + + var state = new OrderedViewState(); + AdoptSourceView( + state, + [visible, lateLoader], + new Dictionary + { + [visible] = Reader(visible, contentVersion: 0, ("MMM", 0), ("NNN", 1)), + [lateLoader] = Reader(lateLoader, contentVersion: 0) + }); + + state.ReconcileLog(lateLoader, Reader(lateLoader, contentVersion: 0, ("AAA", 0), ("BBB", 1))); + + Assert.True(state.ReconcileLog( + lateLoader, + Reader(lateLoader, contentVersion: 1, ("BBB", 0), ("AAA", 1)), + out bool requiresRebuild)); + Assert.True(requiresRebuild); + } + + [Fact] + public void ReconcileLog_SameCountReplaceOfLogEnteringScope_SignalsRebuild() + { + EventLogId adopted = EventLogId.Create(); + EventLogId entering = EventLogId.Create(); + + var state = new OrderedViewState(); + AdoptSourceView( + state, + [adopted], + new Dictionary { [adopted] = Reader(adopted, contentVersion: 0, ("AAA", 0), ("BBB", 1)) }); + + Assert.True(state.TrySetActiveScope([adopted, entering], ViewRequests.NextSequence())); + state.ReconcileLog(entering, Reader(entering, contentVersion: 0, ("CCC", 0), ("DDD", 1))); + + state.ReconcileLog( + entering, + Reader(entering, contentVersion: 1, ("DDD", 0), ("CCC", 1)), + out bool requiresRebuild); + Assert.True(requiresRebuild); + } + + [Fact] + public void ReconcileLog_ShrinkReplace_InvalidatesSignalsRebuildAndReorders() + { + EventLogId logId = EventLogId.Create(); + IEventColumnReader original = Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1), ("CCC", 2)); + IEventColumnReader shrunkReplace = Reader(logId, contentVersion: 1, ("BBB", 0), ("AAA", 1)); + + var state = new OrderedViewState(); + AdoptSourceView(state, [logId], new Dictionary { [logId] = original }); + Assert.Equal(3, state.Current.Count); + + Assert.True(state.ReconcileLog(logId, shrunkReplace, isReplace: true, out bool requiresRebuild)); + Assert.True(requiresRebuild); + Assert.True(state.LiveIndexInvalidated); + + RebuildRequest reseed = state.CaptureScopeReseed(); + Assert.True(state.TryAdoptRebuild(reseed, OrderedViewState.BuildIndex(reseed, CancellationToken.None))); + Assert.False(state.LiveIndexInvalidated); + Assert.Equal(2, state.Current.Count); + Assert.Equal(1, state.Current.At(0).Locator.Index); + Assert.Equal(0, state.Current.At(1).Locator.Index); + } + + [Fact] + public void ReconcileLog_ShrinkToZeroReplace_InvalidatesSignalsRebuildAndEmpties() + { + EventLogId logId = EventLogId.Create(); + + var state = new OrderedViewState(); + AdoptSourceView( + state, + [logId], + new Dictionary { [logId] = Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1)) }); + Assert.Equal(2, state.Current.Count); + + state.ReconcileLog(logId, Reader(logId, contentVersion: 1), isReplace: true, out bool requiresRebuild); + Assert.True(requiresRebuild); + Assert.True(state.LiveIndexInvalidated); + + RebuildRequest reseed = state.CaptureScopeReseed(); + Assert.True(state.TryAdoptRebuild(reseed, OrderedViewState.BuildIndex(reseed, CancellationToken.None))); + Assert.False(state.LiveIndexInvalidated); + Assert.Equal(0, state.Current.Count); + } + + [Fact] + public async Task Writer_CoalescedReorderReplace_RepublishesFullySortedOrder() + { + EventLogId logId = EventLogId.Create(); + + using var parked = new ManualResetEventSlim(false); + using var entered = new ManualResetEventSlim(false); + CancellationToken token = TestContext.Current.CancellationToken; + + await using var writer = new OrderedViewWriter(publishEvery: 1, publishIntervalMs: 0); + + var original = new Dictionary + { + [logId] = Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1)) + }; + + // Adopt a first view (RecordId sort) to establish the log's readers and coverage. + writer.EnqueueViewRequest(ViewRequests.For( + new SortContext(ColumnName.RecordId, false, null, false), + ViewRequests.EmptyFilter, + [logId], + static (_, _) => true, + readers: original)); + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, token); + + // Re-sort by Source with a gated predicate so THAT build (a different view) parks in flight, then coalesce a + // reordered grow-REPLACE onto it. The seed path admits the replacement as an append; the tail-replay leaves + // the prior rows stale, so the corrective rebuild is the only thing that re-sorts them by the replaced content. + writer.EnqueueViewRequest(ViewRequests.For( + s_sourceAscending, + ViewRequests.EmptyFilter, + [logId], + (_, _) => + { + entered.Set(); + parked.Wait(OrderedViewTestTimeouts.Default, token); + + return true; + }, + readers: original)); + + Assert.True(entered.Wait(OrderedViewTestTimeouts.Default, token)); + + var replaced = new Dictionary + { + [logId] = Reader(logId, contentVersion: 1, ("CCC", 0), ("BBB", 1), ("AAA", 2)) + }; + writer.EnqueueViewRequest( + ViewRequests.For(s_sourceAscending, ViewRequests.EmptyFilter, [logId], static (_, _) => true, readers: replaced)); + + parked.Set(); + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, token); + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, token); + + Assert.True( + SpinWait.SpinUntil( + () => writer.Current.Count == 3 && + writer.Current.At(0).Locator.Index == 2 && + writer.Current.At(1).Locator.Index == 1 && + writer.Current.At(2).Locator.Index == 0, + OrderedViewTestTimeouts.Default), + "the corrective rebuild must republish the fully sorted replaced order"); + Assert.Null(writer.Faulted); + } + + [Fact] + public async Task Writer_GrowReplaceHigherContentVersion_RepublishesReorderedSnapshot() + { + EventLogId logId = EventLogId.Create(); + IEventColumnReader original = Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1)); + IEventColumnReader grownReplace = Reader(logId, contentVersion: 1, ("CCC", 0), ("BBB", 1), ("AAA", 2)); + + await using var writer = new OrderedViewWriter(publishEvery: 5000); + + writer.EnqueueReconcile(logId, original); + writer.EnqueueViewRequest(ViewRequests.For(s_sourceAscending, ViewRequests.EmptyFilter, [logId])); + OrderedViewSnapshot adopted = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + Assert.Equal(2, adopted.Count); + Assert.Equal(0, adopted.At(0).Locator.Index); + + writer.EnqueueReconcile(logId, grownReplace, isReplace: true); + OrderedViewSnapshot republished = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + + Assert.Equal(3, republished.Count); + Assert.Equal(2, republished.At(0).Locator.Index); + Assert.Equal(1, republished.At(1).Locator.Index); + Assert.Equal(0, republished.At(2).Locator.Index); + Assert.Null(writer.Faulted); + } + + [Fact] + public async Task Writer_SameCountHigherContentVersion_RepublishesReorderedSnapshot() + { + EventLogId logId = EventLogId.Create(); + IEventColumnReader original = Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1)); + IEventColumnReader reresolved = Reader(logId, contentVersion: 1, ("BBB", 0), ("AAA", 1)); + + await using var writer = new OrderedViewWriter(publishEvery: 5000); + + writer.EnqueueReconcile(logId, original); + writer.EnqueueViewRequest(ViewRequests.For(s_sourceAscending, ViewRequests.EmptyFilter, [logId])); + OrderedViewSnapshot adopted = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + Assert.Equal(0, adopted.At(0).Locator.Index); + + writer.EnqueueReconcile(logId, reresolved); + OrderedViewSnapshot republished = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + + Assert.Equal(1, republished.At(0).Locator.Index); + Assert.Null(writer.Faulted); + } + + [Fact] + public async Task Writer_SecondReplaceDuringInFlightRebuild_FinalOrderReflectsLatestReader() + { + EventLogId logId = EventLogId.Create(); + + using var entered = new ManualResetEventSlim(false); + using var release = new ManualResetEventSlim(false); + int gateArmed = 0; + + bool Predicate(EventLocator locator, IEventColumnReader reader) + { + if (Volatile.Read(ref gateArmed) == 1 && Interlocked.Exchange(ref gateArmed, 0) == 1) + { + entered.Set(); + release.Wait(OrderedViewTestTimeouts.Default); + } + + return true; + } + + await using var writer = new OrderedViewWriter(publishEvery: 5000); + + writer.EnqueueReconcile(logId, Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1), ("CCC", 2))); + writer.EnqueueViewRequest(ViewRequests.For(s_sourceAscending, ViewRequests.EmptyFilter, [logId], Predicate)); + OrderedViewSnapshot adopted = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + Assert.Equal(0, adopted.At(0).Locator.Index); + + Volatile.Write(ref gateArmed, 1); + writer.EnqueueReconcile(logId, Reader(logId, contentVersion: 1, ("CCC", 0), ("BBB", 1), ("AAA", 2))); + Assert.True(entered.Wait(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken)); + + writer.EnqueueReconcile(logId, Reader(logId, contentVersion: 2, ("BBB", 0), ("AAA", 1), ("CCC", 2))); + release.Set(); + + OrderedViewSnapshot final = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + + Assert.Equal(1, final.At(0).Locator.Index); + Assert.Null(writer.Faulted); + } + + [Fact] + public async Task Writer_ShrinkReplace_PublishAndConcurrentAppendDuringRebuildDoNotFault() + { + EventLogId logId = EventLogId.Create(); + EventLogId other = EventLogId.Create(); + + using var entered = new ManualResetEventSlim(false); + using var release = new ManualResetEventSlim(false); + int gateArmed = 0; + + bool Predicate(EventLocator locator, IEventColumnReader reader) + { + if (Volatile.Read(ref gateArmed) == 1 && Interlocked.Exchange(ref gateArmed, 0) == 1) + { + entered.Set(); + release.Wait(OrderedViewTestTimeouts.Default); + } + + return true; + } + + await using var writer = new OrderedViewWriter(publishEvery: 5000); + + writer.EnqueueReconcile(logId, Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1), ("CCC", 2))); + writer.EnqueueReconcile(other, Reader(other, contentVersion: 0, ("DDD", 0), ("EEE", 1))); + writer.EnqueueViewRequest(ViewRequests.For(s_sourceAscending, ViewRequests.EmptyFilter, [logId, other], Predicate)); + OrderedViewSnapshot adopted = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + Assert.Equal(5, adopted.Count); + + Volatile.Write(ref gateArmed, 1); + writer.EnqueueReconcile(logId, Reader(logId, contentVersion: 1, ("AAA", 0)), isReplace: true); + Assert.True(entered.Wait(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken)); + + writer.EnqueueFlush(); + writer.EnqueueReconcile(other, Reader(other, contentVersion: 1, ("DDD", 0), ("EEE", 1), ("FFF", 2))); + + release.Set(); + + OrderedViewSnapshot final = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + + Assert.Null(writer.Faulted); + Assert.Equal(4, final.Count); + } + + [Fact] + public async Task Writer_ShrinkReplace_RepublishesShrunkSnapshotWithoutFault() + { + EventLogId logId = EventLogId.Create(); + + await using var writer = new OrderedViewWriter(publishEvery: 5000); + + writer.EnqueueReconcile(logId, Reader(logId, contentVersion: 0, ("AAA", 0), ("BBB", 1), ("CCC", 2))); + writer.EnqueueViewRequest(ViewRequests.For(s_sourceAscending, ViewRequests.EmptyFilter, [logId])); + OrderedViewSnapshot adopted = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + Assert.Equal(3, adopted.Count); + + writer.EnqueueReconcile(logId, Reader(logId, contentVersion: 1, ("BBB", 0), ("AAA", 1)), isReplace: true); + OrderedViewSnapshot shrunk = + await writer.DrainAsync().WaitAsync(OrderedViewTestTimeouts.Default, TestContext.Current.CancellationToken); + + Assert.Equal(2, shrunk.Count); + Assert.Equal(1, shrunk.At(0).Locator.Index); + Assert.Equal(0, shrunk.At(1).Locator.Index); + Assert.Null(writer.Faulted); + } + + private static void AdoptSourceView( + OrderedViewState state, + IReadOnlyCollection scopeLogs, + IReadOnlyDictionary scopeReaders) + { + Assert.True(state.TrySetActiveScope(scopeLogs, ViewRequests.NextSequence())); + state.ReconcileScopeReaders(scopeReaders); + + RebuildRequest request = state.BeginRebuild(static (_, _) => true, s_sourceAscending); + + Assert.True(state.TryAdoptRebuild(request, OrderedViewState.BuildIndex(request, CancellationToken.None))); + } + + private static IEventColumnReader Reader(EventLogId logId, int contentVersion, params (string Source, long RecordId)[] rows) + { + var clock = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + var events = new List(rows.Length); + + foreach ((string source, long recordId) in rows) + { + events.Add(new ResolvedEvent("Log", LogPathType.Channel) + { + RecordId = recordId, + TimeCreated = clock.AddMilliseconds(recordId), + Id = 1000, + Level = "Information", + Source = source, + LogName = "Channel" + }); + } + + return EventColumnStore.Build(events, generation: 0, contentVersion: contentVersion).CreateReader(logId); + } +}