Do not add state store restoration as an issue for new states (#892)

* Do not add state store restoration as an issue for new states

If there are new states, we will trigger history based on that later.
We were adding and removing a history issue during state restoration.
This caused history to restart after rebrowse, even when unnecessary.
Instead, I'll make two changes:

 - Only add new states to the history reader _after_ restoring state, if
   state restoration is enabled.
 - Add the state restoration history issue only on startup, so that we
   don't run history until we've done the _initial_ state restoration.

This probably needs a rethink, but this is all changing a lot with the
v3 refactor, and I'd prefer to wait with the big refactoring changes
until then. This should fix the issue.

* Try to improve test

* Bump version

Author: einarmo

Extractor/History/HistoryReader.cs

@@ -62,8 +62,13 @@ public sealed class HistoryReader : IDisposable
         public bool CurrentHistoryRunIsBad { get; private set; }
 
 
-        public HistoryReader(ILogger<HistoryReader> log,
-            UAClient uaClient, UAExtractor extractor, TypeManager typeManager, FullConfig config, CancellationToken token)
+        public HistoryReader(
+            ILogger<HistoryReader> log,
+            UAClient uaClient,
+            UAExtractor extractor,
+            TypeManager typeManager,
+            FullConfig config,
+            CancellationToken token)
         {
             this.log = log;
             this.config = config;
@@ -76,7 +81,7 @@ public HistoryReader(ILogger<HistoryReader> log,
             continuationPoints = new BlockingResourceCounter(
                 throttling.MaxNodeParallelism > 0 ? throttling.MaxNodeParallelism : 1_000);
             waiter = new OperationWaiter();
-            state = new State(config);
+            state = new State(config, extractor.StateStorage != null && config.StateStorage.IntervalValue.Value != Timeout.InfiniteTimeSpan);
         }
 
         private async Task RunHistoryBatch(IEnumerable<UAHistoryExtractionState> states, HistoryReadType type)
@@ -235,7 +240,7 @@ public bool RemoveIssue(StateIssue issue)
                 return issues.Remove(issue);
             }
 
-            public State(FullConfig config)
+            public State(FullConfig config, bool hasStateStore)
             {
                 if (config.Subscriptions.Events && config.Events.Enabled && config.Events.History)
                 {
@@ -245,6 +250,10 @@ public State(FullConfig config)
                 {
                     issues.Add(StateIssue.DataPointSubscription);
                 }
+                if (hasStateStore)
+                {
+                    issues.Add(StateIssue.StateRestorationPending);
+                }
                 issues.Add(StateIssue.NodeHierarchyRead);
             }
 

Extractor/UAExtractor.cs

@@ -1022,7 +1022,7 @@ private async Task PushNodes(PusherInput input)
             var newStates = input.Variables
                 .Select(ts => ts.Id)
                 .Distinct()
-                .Select(id => State.GetNodeState(id));
+                .SelectNonNull(id => State.GetNodeState(id));
 
             bool initial = input.Variables.Count() + input.Objects.Count() >= State.NumActiveNodes;
 
@@ -1077,6 +1077,10 @@ private async Task PushNodes(PusherInput input)
             {
                 pushTasks = pushTasks.Append(RestoreExtractionStateWithRetry(newStates));
             }
+            else
+            {
+                historyReader?.AddStates(newStates, State.EmitterStates);
+            }
 
 
             pushTasks = pushTasks.ToList();
@@ -1094,7 +1098,7 @@ private async Task PushNodes(PusherInput input)
                 trackedTimeseres.Inc(input.Variables.Count());
             }
 
-            foreach (var state in newStates.Concat<UAHistoryExtractionState?>(State.EmitterStates))
+            foreach (var state in newStates.Concat<UAHistoryExtractionState>(State.EmitterStates))
             {
                 state?.FinalizeRangeInit();
             }
@@ -1106,10 +1110,8 @@ private async Task PushNodes(PusherInput input)
         /// </summary>
         /// <param name="newStates">Variable states to restore</param>
         /// <returns>Task that completes when state is restored or extractor is cancelled</returns>
-        private async Task RestoreExtractionStateWithRetry(IEnumerable<VariableExtractionState?> newStates)
+        private async Task RestoreExtractionStateWithRetry(IEnumerable<VariableExtractionState> newStates)
         {
-            // Add issue to block history until state restoration completes
-            historyReader?.AddIssue(HistoryReader.StateIssue.StateRestorationPending);
 
             // Retry state restoration forever until it succeeds
             var retryConfig = new RetryUtilConfig
@@ -1138,7 +1140,7 @@ await RetryUtil.RetryAsync(
                     if (Streamer.AllowData)
                     {
                         tasks.Add(StateStorage!.RestoreExtractionState(
-                            newStates.Where(state => state != null && state.FrontfillEnabled).ToDictionary(state => state?.Id!, state => state!),
+                            newStates.Where(state => state.FrontfillEnabled).ToDictionary(state => state.Id),
                             Config.StateStorage.VariableStore,
                             false,
                             Source.Token));
@@ -1156,6 +1158,7 @@ await RetryUtil.RetryAsync(
             );
 
             // Successfully restored state - allow history to proceed
+            historyReader?.AddStates(newStates, State.EmitterStates);
             historyReader?.RemoveIssue(HistoryReader.StateIssue.StateRestorationPending);
             log.LogInformation("Successfully restored extraction state from state storage");
         }
@@ -1209,8 +1212,6 @@ private IEnumerable<Func<CancellationToken, Task>> CreateSubscriptions(IEnumerab
         {
             var states = variables.Select(ts => ts.Id).Distinct().SelectNonNull(id => State.GetNodeState(id));
 
-            historyReader?.AddStates(states, State.EmitterStates);
-
             log.LogInformation("Synchronize {NumNodesToSynch} nodes", variables.Count());
             var tasks = new List<Func<CancellationToken, Task>>();
             // Create tasks to subscribe to nodes, then start history read. We might lose data if history read finished before subscriptions were created.

Test/Integration/NodeExtractionTests.cs

@@ -1090,6 +1090,8 @@ public async Task TestRebrowseFailedWeirdState()
             // This is a rebrowse, and should _not_ set the extractor to uninitialized.
             Assert.True(pusher.Initialized);
 
+            await extractor.WaitForSubscription(SubscriptionName.DataPoints);
+
             // Trigger a datapoint update.
             tester.Server.UpdateNode(tester.Server.Ids.Base.DoubleVar1, 321.123);
 

manifest.yml

@@ -67,6 +67,11 @@ schema:
    - "https://raw.githubusercontent.com/"
 
 versions:
+  "2.40.4":
+    description: Fix regression causing history to restart on rebrowse if state store is enabled.
+    changelog:
+      fixed:
+        - Fix regression causing history to restart on rebrowse if state store is enabled.
   "2.40.3":
     description: Set extractor version as a label value on the opcua_version metric.
     changelog: