Rebuild the ordered view on same-generation content-version re-resolutions

src/EventLogExpert.Runtime/LogTable/OrderedView/OrderedViewScopeState.cs

@@ -119,6 +119,8 @@ public bool ScopeEquals(IReadOnlyCollection<EventLogId> scopeLogs)
         return matched == _scopeLogs.Count;
     }
 
+    public void SetCoverage(in LogGeneration key, int coveredCount) => _coverage[key] = coveredCount;
+
     public bool TrySetScope(IReadOnlyCollection<EventLogId> scopeLogs, long scopeVersion)
     {
         if (scopeVersion < ScopeVersion) { return false; }

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,15 +82,15 @@ public Task HandleLoadEvents(LoadEventsAction action, IDispatcher dispatcher) =>
         Shadow(() =>
         {
             Sync();
-            Reconcile(action.LogData.Id);
+            Reconcile(action.LogData.Id, isReplace: true);
         });
 
     [EffectMethod]
     public Task HandleLoadEventsPartial(LoadEventsPartialAction action, IDispatcher dispatcher) =>
         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);
         }
     }
 

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<EventLocator, IEventColumnReader, bool> _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<EventLogId, IEventColumnReader> 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;