Ensure destination range is capped within source range (#942)

* Release changes for updating source range cursor

* Ensure destination range is capped within source range

* fix build

* Use two dicts

* Fix tests

* Cleaner code and better comments

* Refactor for less operations

* lint fix

* Test fix

* Test fix

* test fix

---------

Co-authored-by: Toshad Salwekar <toshad.salwekar@cognite.com>

Author: Toshad

Extractor/Streamer.cs

@@ -149,6 +149,12 @@ public virtual async Task EnqueueAsync(IEnumerable<UAEvent> events)
             if (events == null) return;
             await eventQueue.EnqueueAsync(events);
         }
+
+        private struct Ranges
+        {
+            public TimeRange sourceExtractedRange;
+            public TimeRange queuePointsRange;
+        }
         /// <summary>
         /// Push data points to destinations
         /// </summary>
@@ -161,17 +167,26 @@ public async Task PushDataPoints(IEnumerable<IPusher> passingPushers,
             if (!AllowData) return;
 
             var dataPointList = new List<UADataPoint>();
-            var pointRanges = new Dictionary<string, TimeRange>();
+
+            // Track source extracted timestamps and normal timestamps for each node in the current batch to update state later.
+            var ranges = new Dictionary<string, Ranges>();
 
             await foreach (var dp in dataPointQueue.DrainAsync(token))
             {
                 dataPointList.Add(dp);
-                if (!pointRanges.TryGetValue(dp.Id, out var range))
+                if (!ranges.TryGetValue(dp.Id, out var range))
                 {
-                    pointRanges[dp.Id] = new TimeRange(dp.Timestamp, dp.Timestamp);
+                    // Get source extracted range while we are draining, to make sure we don't update the state with a range that exceeds this.
+                    ranges[dp.Id] = new Ranges
+                    {
+                        // Source extracted range will always exist, so default is irrelevant.
+                        sourceExtractedRange = extractor.State.GetNodeState(dp.Id)?.SourceExtractedRange ?? new TimeRange(dp.Timestamp, dp.Timestamp),
+                        queuePointsRange = new TimeRange(dp.Timestamp, dp.Timestamp)
+                    };
                     continue;
                 }
-                pointRanges[dp.Id] = range.Extend(dp.Timestamp, dp.Timestamp);
+                range.queuePointsRange = range.queuePointsRange.Extend(dp.Timestamp, dp.Timestamp);
+                ranges[dp.Id] = range;
             }
 
             var results = await Task.WhenAll(passingPushers.Select(pusher => pusher.PushDataPoints(dataPointList, token)));
@@ -196,6 +211,7 @@ public async Task PushDataPoints(IEnumerable<IPusher> passingPushers,
                 }
                 if (config.FailureBuffer.Enabled && extractor.FailureBuffer != null)
                 {
+                    var pointRanges = ranges.ToDictionary(kvp => kvp.Key, kvp => kvp.Value.queuePointsRange);
                     await extractor.FailureBuffer.WriteDatapoints(dataPointList, pointRanges, token);
                 }
 
@@ -217,10 +233,12 @@ public async Task PushDataPoints(IEnumerable<IPusher> passingPushers,
             {
                 await extractor.FailureBuffer.ReadDatapoints(passingPushers, token);
             }
-            foreach ((string id, var range) in pointRanges)
+            foreach ((string id, var range) in ranges)
             {
                 var state = extractor.State.GetNodeState(id);
-                if (state != null && (extractor.AllowUpdateState || !state.FrontfillEnabled && !state.BackfillEnabled)) state.UpdateDestinationRange(range.First, range.Last);
+                // Make sure we don't update the destination range beyond the source extracted range at the time of reading from queue.
+                var pointRange = range.queuePointsRange.Contract(range.sourceExtractedRange);
+                if (state != null && (extractor.AllowUpdateState || !state.FrontfillEnabled && !state.BackfillEnabled)) state.UpdateDestinationRange(pointRange.First, pointRange.Last);
             }
         }
         /// <summary>
@@ -235,18 +253,19 @@ public async Task PushEvents(IEnumerable<IPusher> passingPushers,
             if (!AllowEvents) return;
 
             var eventList = new List<UAEvent>();
-            var eventRanges = new Dictionary<NodeId, TimeRange>();
+            var ranges = new Dictionary<NodeId, Ranges>();
 
             await foreach (var evt in eventQueue.DrainAsync(token))
             {
                 eventList.Add(evt);
-                if (!eventRanges.TryGetValue(evt.EmittingNode, out var range))
+                if (!ranges.TryGetValue(evt.EmittingNode, out var range))
                 {
-                    eventRanges[evt.EmittingNode] = new TimeRange(evt.Time, evt.Time);
+                    var sourceRange = extractor.State.GetEmitterState(evt.EmittingNode)?.SourceExtractedRange ?? new TimeRange(evt.Time, evt.Time);
+                    ranges[evt.EmittingNode] = new Ranges { sourceExtractedRange = sourceRange, queuePointsRange = new TimeRange(evt.Time, evt.Time) };
                     continue;
                 }
-
-                eventRanges[evt.EmittingNode] = range.Extend(evt.Time, evt.Time);
+                range.queuePointsRange = range.queuePointsRange.Extend(evt.Time, evt.Time);
+                ranges[evt.EmittingNode] = range;
             }
 
             var results = await Task.WhenAll(passingPushers.Select(pusher => pusher.PushEvents(eventList, token)));
@@ -292,10 +311,12 @@ public async Task PushEvents(IEnumerable<IPusher> passingPushers,
             {
                 await extractor.FailureBuffer.ReadEvents(passingPushers, token);
             }
-            foreach (var (id, range) in eventRanges)
+            foreach (var (id, range) in ranges)
             {
                 var state = extractor.State.GetEmitterState(id);
-                if (state != null && (extractor.AllowUpdateState || !state.FrontfillEnabled && !state.BackfillEnabled)) state?.UpdateDestinationRange(range.First, range.Last);
+                // Make sure we don't update the destination range beyond the source extracted range at the time of reading from queue.
+                var eventRange = range.queuePointsRange.Contract(range.sourceExtractedRange);
+                if (state != null && (extractor.AllowUpdateState || !state.FrontfillEnabled && !state.BackfillEnabled)) state?.UpdateDestinationRange(eventRange.First, eventRange.Last);
             }
         }
         /// <summary>

Test/Unit/StreamerTest.cs

@@ -68,6 +68,7 @@ void DummyLooper(CancellationToken token)
             Assert.False(evt.WaitOne(100));
 
             await extractor.Streamer.EnqueueAsync(new UADataPoint(start, "id", -1, StatusCodes.Good));
+            state.UpdateFromStream(start, start);
             Assert.Equal(1, queue.Count);
             await extractor.Streamer.PushDataPoints(new[] { pusher }, Enumerable.Empty<IPusher>(), tester.Source.Token);
 
@@ -100,6 +101,7 @@ void DummyLooper(CancellationToken token)
             Assert.Equal(999_999, queue.Count);
 
             await extractor.Streamer.EnqueueAsync(new UADataPoint(start.AddMilliseconds(3000000), "id", 300, StatusCodes.Good));
+            state.UpdateFromStream(start.AddMilliseconds(3000000), start.AddMilliseconds(3000000));
             Assert.True(evt.WaitOne(10000));
             Assert.Equal(1_000_000, queue.Count);
 
@@ -141,6 +143,7 @@ void DummyLooper(CancellationToken token)
             Assert.False(evt.WaitOne(100));
 
             await extractor.Streamer.EnqueueAsync(new UAEvent { EmittingNode = id, Time = start });
+            state.UpdateFromStream(start, start);
             Assert.Equal(1, queue.Count);
             await extractor.Streamer.PushEvents(new[] { pusher }, Enumerable.Empty<IPusher>(), tester.Source.Token);
 
@@ -175,6 +178,7 @@ await extractor.Streamer.EnqueueAsync(Enumerable.Range(200000, 99999).Select(idx
             Assert.Equal(99999, queue.Count);
 
             await extractor.Streamer.EnqueueAsync(new UAEvent { EmittingNode = id, Time = start.AddMilliseconds(300000) });
+            state.UpdateFromStream(start.AddMilliseconds(300000), start.AddMilliseconds(300000));
             Assert.True(evt.WaitOne(10000));
             Assert.Equal(100000, queue.Count);
 

manifest.yml

@@ -67,6 +67,11 @@ schema:
    - "https://raw.githubusercontent.com/"
 
 versions:
+  "2.40.8":
+    description: Fix rare race condition allowing history range read to increment without pushing events to CDF.
+    changelog:
+      fixed:
+        - Fix rare race condition allowing history range read to increment without pushing events to CDF.
   "2.40.7":
     description: Fix issue causing error handling to fail when encountering mismatched timeseries against data modelling.
     changelog: