fix(historian): probe the gateway on a timer for the health snapshot (lmxopcua#746)

The Gateway historian's ProcessConnectionOpen / EventConnectionOpen were derived from read outcomes only, so a freshly booted node with a healthy gateway reported the historian closed until a client read, and a node that only writes (continuous historization) never flipped it. The data source now runs a ProbeAsync loop (ServerHistorian:Gateway: ProbeInterval, default 30 s, first probe at start-up; 0 = off). Each probe, bounded at 10 s, is recorded as the latest connectivity observation; the latest observation (probe or read) is what the snapshot reports. Probes never touch the query/failure counters. DisposeAsync stops the loop before disposing the client. Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22 in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-10-22 16:33:11 -04:00
parent 0b85c499e3
commit 8dc594d8e4
5 changed files with 233 additions and 10 deletions
@@ -120,6 +120,24 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
    private long _totalFailures;
    private int _consecutiveFailures;

+    /// <summary>
+    ///     The latest connectivity observation — a read outcome or a probe result, whichever came last (null = none yet).
+    ///     <see cref="GetHealthSnapshot"/> reports it as the connection flags.
+    /// </summary>
+    private bool? _lastObservationOpen;
+
+    /// <summary>The probe cadence (<c>ServerHistorian:Gateway:ProbeInterval</c>); zero = no probe loop.</summary>
+    private readonly TimeSpan _probeInterval;
+
+    /// <summary>The probe cadence — exposed so the factory wiring is testable.</summary>
+    internal TimeSpan ProbeInterval => _probeInterval;
+
+    private readonly CancellationTokenSource _probeStop = new();
+    private readonly Task _probeLoop = Task.CompletedTask;
+
+    /// <summary>The most one probe may take before it counts as a failed (closed) observation.</summary>
+    private static readonly TimeSpan ProbeCallBound = TimeSpan.FromSeconds(10);
+
    /// <summary>Creates a gateway-backed historian data source.</summary>
    /// <param name="client">The gateway client seam used for all reads.</param>
    /// <param name="logger">Diagnostic logger; failures are recorded without leaking tag/host detail.</param>
@@ -148,6 +166,11 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
    ///     default <c>false</c>): until historiangw#27 is live-verified they answer Good-status wrong values, so by default they
    ///     are refused <see cref="HistoryAggregateNotSupportedException"/> (<c>BadAggregateNotSupported</c>) and not advertised.
    /// </param>
+    /// <param name="probeInterval">
+    ///     Cadence of the <c>ProbeAsync</c> health loop (<c>ServerHistorian:Gateway:ProbeInterval</c>); the first probe runs
+    ///     at construction. Null / zero (the default here; the factory passes the configured 30 s) = no loop, health from
+    ///     read outcomes only.
+    /// </param>
    /// <param name="eventDrainBudget">
    ///     Stream rows a capped events read consumes (uncollected) past its cap before it abandons the gateway stream
    ///     (<c>ServerHistorian:Gateway:EventDrainBudget</c>). Negative counts as 0 = abandon at the cap.
@@ -155,7 +178,8 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
    public GatewayHistorianDataSource(IHistorianGatewayClient client, ILogger<GatewayHistorianDataSource> logger,
        int maxRawRows = TimeWeightedAggregator.DefaultMaxWindowRows, TimeProvider? timeProvider = null,
        bool atTimeFromRaw = true, int maxPreWindowRows = DefaultMaxPreWindowRows, int maxRawValuesPerRead = 0,
-        int eventDrainBudget = DefaultEventDrainBudget, bool enableNativeTimeWeighted = false)
+        int eventDrainBudget = DefaultEventDrainBudget, bool enableNativeTimeWeighted = false,
+        TimeSpan? probeInterval = null)
    {
        _enableNativeTimeWeighted = enableNativeTimeWeighted;
        _eventDrainBudget = Math.Max(0, eventDrainBudget);
@@ -170,6 +194,53 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
        _time = timeProvider ?? TimeProvider.System;
        _ceiling = new GatewayRowCeiling(_time);
        _tags = new GatewayTagPresenceCache(_client, _time);
+        _probeInterval = probeInterval is { } pi && pi > TimeSpan.Zero ? pi : TimeSpan.Zero;
+        if (_probeInterval > TimeSpan.Zero) _probeLoop = Task.Run(() => ProbeLoopAsync(_probeStop.Token));
+    }
+
+    /// <summary>
+    ///     Probes the gateway once immediately, then every <see cref="_probeInterval"/>, recording each result as the
+    ///     latest connectivity observation. Probes are not history queries: they never touch the query / failure counters.
+    ///     A probe that throws or overruns <see cref="ProbeCallBound"/> is a closed observation; nothing escapes the loop.
+    /// </summary>
+    private async Task ProbeLoopAsync(CancellationToken stop)
+    {
+        using var timer = new PeriodicTimer(_probeInterval, _time);
+        do
+        {
+            bool open;
+            try
+            {
+                using var bound = CancellationTokenSource.CreateLinkedTokenSource(stop);
+                bound.CancelAfter(ProbeCallBound);
+                open = await _client.ProbeAsync(bound.Token).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) when (stop.IsCancellationRequested)
+            {
+                return;
+            }
+            catch (Exception ex)
+            {
+                open = false;
+                _logger.LogDebug("Historian gateway probe failed ({Exception}).", ex.GetType().Name);
+            }
+
+            bool changed;
+            lock (_healthLock)
+            {
+                changed = _lastObservationOpen != open;
+                _lastObservationOpen = open;
+            }
+            if (changed)
+                _logger.LogInformation("Historian gateway probe: the gateway is {State}.", open ? "reachable" : "unreachable");
+        }
+        while (await WaitNextAsync(timer, stop).ConfigureAwait(false));
+    }
+
+    private static async Task<bool> WaitNextAsync(PeriodicTimer timer, CancellationToken stop)
+    {
+        try { return await timer.WaitForNextTickAsync(stop).ConfigureAwait(false); }
+        catch (OperationCanceledException) { return false; }
    }

    /// <inheritdoc />
@@ -760,12 +831,13 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
                LastSuccessTime: _lastSuccessUtc,
                LastFailureTime: _lastFailureUtc,
                LastError: _lastError,
-                // No connection probe exists (the old RefreshConnectionStateAsync had no production
-                // caller and was deleted, archreview 06/S-6): "open" is derived from observed reads
-                // alone -- at least one success and no failure since. The gateway is non-clustered to
-                // us, so node fields are null/empty (mirrors the Wonderware client's Finding 010 posture).
-                ProcessConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0,
-                EventConnectionOpen: _totalSuccesses > 0 && _consecutiveFailures == 0,
+                // "Open" is the latest connectivity observation: a read outcome, or a ProbeAsync result from the probe
+                // loop (ServerHistorian:Gateway:ProbeInterval), whichever came last — so a node that has served no
+                // HistoryRead yet (or only writes) still reports a healthy gateway. Without a probe loop it is read
+                // outcomes alone: at least one success and no failure since. The gateway is non-clustered to us, so
+                // node fields are null/empty.
+                ProcessConnectionOpen: _lastObservationOpen ?? false,
+                EventConnectionOpen: _lastObservationOpen ?? false,
                ActiveProcessNode: null,
                ActiveEventNode: null,
                Nodes: []);
@@ -850,6 +922,7 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
        lock (_healthLock)
        {
            _totalQueries++;
+            _lastObservationOpen = success;
            if (success)
            {
                _totalSuccesses++;
@@ -869,7 +942,16 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
    /// <summary>Disposes the underlying gateway client. Prefer <see cref="DisposeAsync"/>.</summary>
    public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();

-    /// <summary>Asynchronously disposes the underlying gateway client.</summary>
+    /// <summary>Stops the probe loop (bounded wait), then asynchronously disposes the underlying gateway client.</summary>
    /// <returns>A task that completes when the client has been disposed.</returns>
-    public ValueTask DisposeAsync() => _client.DisposeAsync();
+    public async ValueTask DisposeAsync()
+    {
+        if (!_probeStop.IsCancellationRequested)
+        {
+            _probeStop.Cancel();
+            // The loop never throws; a probe stuck past its own bound is abandoned rather than awaited further.
+            await Task.WhenAny(_probeLoop, Task.Delay(ProbeCallBound)).ConfigureAwait(false);
+            await _client.DisposeAsync().ConfigureAwait(false);
+        }
+    }
 }
@@ -50,7 +50,9 @@ public static class GatewayHistorian
            // How far a capped events read drains its stream past the cap before abandoning it.
            eventDrainBudget: options.Gateway.EventDrainBudget,
            // AVEVA's native TimeAverage / Total — off until historiangw#27 is live-verified.
-            enableNativeTimeWeighted: options.Gateway.EnableNativeTimeWeighted);
+            enableNativeTimeWeighted: options.Gateway.EnableNativeTimeWeighted,
+            // A slow ProbeAsync timer so the health snapshot reflects the gateway without waiting for a read.
+            probeInterval: options.Gateway.ProbeInterval);
    }

    /// <summary>