fix(historian): capture while the recording Primary is unreachable (#672)

A SIGKILLed Primary stays an unreachable-but-not-downed member until auto-down, and the redundancy singleton usually dies with it, so the standby's cached snapshot kept naming it Primary and nobody captured for ~30 s. DriverHostActor now holds its own ReachabilityEvent (+MemberUp) subscription, caches the last snapshot, and excludes unreachable peers from the historization peer-is-Primary predicate only, republishing just the capture verdict on a reachability change. Alarm drain and the Primary data-plane verdict are unchanged. Gated by ContinuousHistorization:CaptureWhilePrimaryUnreachable (default true). Tests: message-driven cases with real ReachabilityEvents (incl. a falsification control on the drain/data-plane verdicts) plus a real two-node cluster crash test with downing disabled.
2026-10-11 08:26:39 -04:00
parent db6a53036b
commit 1b2c36eb87
11 changed files with 573 additions and 21 deletions
@@ -55,7 +55,9 @@
    "MaxWriteAttempts": 5,
    "_MaxWriteAttemptsComment": "Gitea #538: consecutive failed drain passes one tag may accumulate — counted only across passes where OTHER tags wrote successfully — before its queued entries are abandoned so the FIFO head can advance.",
    "MinBackoffSeconds": 1,
-    "MaxBackoffSeconds": 30
+    "MaxBackoffSeconds": 30,
+    "CaptureWhilePrimaryUnreachable": true,
+    "_CaptureWhilePrimaryUnreachableComment": "Gitea #672: a standby stood down for its mesh Primary resumes capture as soon as that Primary is reported UNREACHABLE (not after auto-down) — shrinks a crash-failover capture gap to the failure-detector window, at the cost of short duplicate windows. false = pre-#672."
  },
  "AlarmHistorian": {
    "_comment": "Durable store-and-forward alarm sink. Buffers into the node's consolidated LocalDb (see the LocalDb section) so the queue replicates to the redundant pair peer, and drains alarm events to the ServerHistorian gateway's SendEvent path from whichever node holds the Primary role. The downstream connection (endpoint/key/TLS) is sourced from the ServerHistorian section.",
@@ -461,6 +461,30 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
    /// </summary>
    private readonly Func<CommonsNodeId, bool>? _sharesHistorizationScopeWithLocal;

+    /// <summary>Gitea #672: whether an UNREACHABLE recording Primary still stands this node's historization
+    /// capture down (<c>false</c>) or not (<c>true</c>, the default). See <see cref="ComputeHistorizationVerdict"/>.</summary>
+    private readonly bool _captureWhilePrimaryUnreachable;
+
+    /// <summary>Gitea #672: the last redundancy snapshot's nodes, so a reachability change can re-derive the
+    /// historization capture verdict without waiting for the next snapshot — which, when the recording Primary
+    /// crashed, may not come until auto-down (the redundancy singleton usually lived on that Primary).</summary>
+    private IReadOnlyList<NodeRedundancyState>? _lastRedundancySnapshotNodes;
+
+    /// <summary>Gitea #672: peers the cluster failure detector currently reports unreachable (never this node).
+    /// Fed by <see cref="Akka.Cluster.ClusterEvent.ReachabilityEvent"/>; cleared for an address that comes Up again.</summary>
+    private readonly HashSet<CommonsNodeId> _unreachableNodes = new();
+
+    /// <summary>Gitea #672: the unreachable recording Primary this node is currently capturing in place of, or
+    /// null — drives the once-per-transition Information log.</summary>
+    private CommonsNodeId? _capturingForUnreachablePrimary;
+
+    /// <summary>The last historization capture verdict published (null before the first), so the reachability
+    /// path republishes only a CHANGED verdict.</summary>
+    private bool? _lastPublishedHistorizationVerdict;
+
+    /// <summary>Whether this actor holds its own (#672) cluster reachability subscription.</summary>
+    private bool _subscribedToReachabilityEvents;
+
    /// <summary>Gitea #601 — how long an Applied ack waits for the address-space outcome (see
    /// <see cref="DefaultApplyOutcomeTimeout"/>).</summary>
    private readonly TimeSpan _applyOutcomeTimeout;
@@ -664,7 +688,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
        TimeSpan? reprovisionBackoff = null,
        TimeSpan? commsRecoveryAuditDelay = null,
        Func<CommonsNodeId, bool>? sharesAlarmQueueWithLocal = null,
-        TimeSpan? applyOutcomeTimeout = null) =>
+        TimeSpan? applyOutcomeTimeout = null,
+        bool captureWhilePrimaryUnreachable = true) =>
        // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an
        // expression tree. Six IActorRef? parameters and several interface-typed ones mean a
        // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds
@@ -677,7 +702,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
            deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher,
            alarmStateStore, telemetryHub, primaryRoleSource, historizationRoleView,
            sharesHistorizationScopeWithLocal, applyWedgeTimeout, reprovisionBackoff, commsRecoveryAuditDelay,
-            sharesAlarmQueueWithLocal, applyOutcomeTimeout));
+            sharesAlarmQueueWithLocal, applyOutcomeTimeout, captureWhilePrimaryUnreachable));

    /// <summary>Initializes a new DriverHostActor with the specified dependencies.</summary>
    /// <param name="dbFactory">Database context factory for configuration database access.</param>
@@ -745,6 +770,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
    /// PeerAddress), whose partner is its per-cluster mesh partner. Ignored when
    /// <paramref name="replicationPeerHost"/> is set. Null ⇒ no such peer is known and this node never stands its
    /// drain down.</param>
+    /// <param name="captureWhilePrimaryUnreachable">Gitea #672 (<c>ContinuousHistorization:CaptureWhilePrimaryUnreachable</c>,
+    /// default <c>true</c>): when a mesh-partner Primary this node would stand its historization capture down for is
+    /// reported UNREACHABLE by the cluster failure detector, stop deferring to it and capture — rather than waiting
+    /// for auto-down + a fresh redundancy snapshot. Affects ONLY the historization capture verdict; the alarm drain
+    /// and the Primary data-plane verdict are unchanged. <c>false</c> restores the pre-#672 behaviour.</param>
    public DriverHostActor(
        IDbContextFactory<OtOpcUaConfigDbContext>? dbFactory,
        CommonsNodeId localNode,
@@ -776,8 +806,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
        TimeSpan? reprovisionBackoff = null,
        TimeSpan? commsRecoveryAuditDelay = null,
        Func<CommonsNodeId, bool>? sharesAlarmQueueWithLocal = null,
-        TimeSpan? applyOutcomeTimeout = null)
+        TimeSpan? applyOutcomeTimeout = null,
+        bool captureWhilePrimaryUnreachable = true)
    {
+        _captureWhilePrimaryUnreachable = captureWhilePrimaryUnreachable;
        _applyOutcomeTimeout = applyOutcomeTimeout ?? DefaultApplyOutcomeTimeout;
        _sharesAlarmQueueWithLocal = sharesAlarmQueueWithLocal;
        _applyWedgeTimeout = applyWedgeTimeout ?? DefaultApplyWedgeTimeout;
@@ -847,6 +879,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
        // ApplyScriptedAlarms through the same PushDesiredSubscriptions pass.
        SpawnScriptedAlarmHost();
        StartPrimaryVerdictPublishing();
+        StartReachabilityTracking();
        Bootstrap();
    }

@@ -894,8 +927,18 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
    /// <summary>A cluster member changed status: re-publish the Primary verdict now and once more shortly after.
    /// The deferred pass covers the cluster read-view (what <see cref="DriverMemberCount"/> reads) updating on its
    /// own subscriber AFTER this event was delivered here — the two are not ordered.</summary>
-    private void OnDriverMembershipChanged(Akka.Cluster.ClusterEvent.IMemberEvent _)
+    private void OnDriverMembershipChanged(Akka.Cluster.ClusterEvent.IMemberEvent evt)
    {
+        // Gitea #672: a member that comes Up at an address is reachable — the usual case being a restarted
+        // node's new incarnation. A REMOVED member deliberately stays in the set: it never answers again, and
+        // dropping it would re-derive the verdict from a cached snapshot that may still name it Primary,
+        // closing the gate on a dead node until the next snapshot (the very hole #672 closes).
+        if (evt is Akka.Cluster.ClusterEvent.MemberUp up
+            && _unreachableNodes.Remove(ToNodeId(up.Member.Address)))
+        {
+            RepublishHistorizationVerdictIfChanged();
+        }
+
        PublishPrimaryVerdictIfChanged();
        // One keyed timer: a burst of member events collapses into a single deferred pass.
        Timers.StartSingleTimer("republish-primary-verdict", RepublishPrimaryVerdict.Instance, MembershipRepublishDelay);
@@ -919,10 +962,62 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
        PublishPrimaryVerdict();
    }

+    /// <summary>
+    ///     Gitea #672: follow cluster reachability so the historization capture gate stops deferring to a recording
+    ///     Primary the failure detector already considers dead. Its OWN subscription, deliberately not folded into
+    ///     <see cref="StartPrimaryVerdictPublishing"/>, which early-returns on nodes with no primary-role source.
+    ///     <see cref="Akka.Cluster.ClusterEvent.MemberUp"/> rides along so a re-joined incarnation leaves the set.
+    /// </summary>
+    private void StartReachabilityTracking()
+    {
+        if (_historizationRoleView is null || !_captureWhilePrimaryUnreachable) return;
+        try
+        {
+            Akka.Cluster.Cluster.Get(Context.System).Subscribe(
+                Self, Akka.Cluster.ClusterEvent.SubscriptionInitialStateMode.InitialStateAsEvents,
+                typeof(Akka.Cluster.ClusterEvent.ReachabilityEvent),
+                typeof(Akka.Cluster.ClusterEvent.MemberUp));
+            _subscribedToReachabilityEvents = true;
+        }
+        catch (Exception)
+        {
+            // Non-cluster ActorRefProvider — no failure detector to follow; the snapshot path alone decides.
+        }
+    }
+
+    /// <summary>Gitea #672: a peer became unreachable or reachable again. Records it and re-derives ONLY the
+    /// historization capture verdict from the cached snapshot — the alarm drain and the Primary data-plane verdict
+    /// are deliberately untouched (a false allow there is a dual claimant, not a duplicate sample).</summary>
+    private void OnReachabilityChanged(Akka.Cluster.ClusterEvent.ReachabilityEvent evt)
+    {
+        var node = ToNodeId(evt.Member.Address);
+        if (node == _localNode) return;   // this node's own reachability as seen by others is not a peer question
+        var changed = evt is Akka.Cluster.ClusterEvent.UnreachableMember
+            ? _unreachableNodes.Add(node)
+            : _unreachableNodes.Remove(node);
+        if (changed) RepublishHistorizationVerdictIfChanged();
+    }
+
+    /// <summary>Re-derives the historization capture verdict from the cached snapshot and publishes it only when it
+    /// differs from the last one published. No-op before the first snapshot.</summary>
+    private void RepublishHistorizationVerdictIfChanged()
+    {
+        if (_historizationRoleView is null || _lastRedundancySnapshotNodes is null) return;
+        var verdict = ComputeHistorizationVerdict(_lastRedundancySnapshotNodes);
+        if (verdict == _lastPublishedHistorizationVerdict) return;
+        _lastPublishedHistorizationVerdict = verdict;
+        _historizationRoleView.PublishHistorizationRole(verdict);
+    }
+
+    /// <summary>Same host:port projection as <c>ClusterRoleInfo</c> / <c>RedundancyStateActor.ToNodeId</c>, so a
+    /// reachability event's member compares equal to the NodeId a redundancy snapshot names.</summary>
+    private static CommonsNodeId ToNodeId(Akka.Actor.Address address) =>
+        CommonsNodeId.Parse($"{address.Host ?? string.Empty}:{address.Port ?? 0}");
+
    /// <inheritdoc />
    protected override void PostStop()
    {
-        if (_subscribedToMemberEvents)
+        if (_subscribedToMemberEvents || _subscribedToReachabilityEvents)
        {
            try { Akka.Cluster.Cluster.Get(Context.System).Unsubscribe(Self); }
            catch (Exception) { /* system shutting down */ }
@@ -1349,6 +1444,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
        Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
        Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
        Receive<Akka.Cluster.ClusterEvent.IMemberEvent>(OnDriverMembershipChanged);
+        Receive<Akka.Cluster.ClusterEvent.ReachabilityEvent>(OnReachabilityChanged);
        Receive<RepublishPrimaryVerdict>(_ => PublishPrimaryVerdictIfChanged());
        Receive<Akka.Cluster.ClusterEvent.CurrentClusterState>(_ => PublishPrimaryVerdictIfChanged());
        // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
@@ -1410,6 +1506,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
        Receive<RouteNativeAlarmAck>(HandleRouteNativeAlarmAck);
        Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
        Receive<Akka.Cluster.ClusterEvent.IMemberEvent>(OnDriverMembershipChanged);
+        Receive<Akka.Cluster.ClusterEvent.ReachabilityEvent>(OnReachabilityChanged);
        Receive<RepublishPrimaryVerdict>(_ => PublishPrimaryVerdictIfChanged());
        Receive<Akka.Cluster.ClusterEvent.CurrentClusterState>(_ => PublishPrimaryVerdictIfChanged());
        // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
@@ -2208,20 +2305,15 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
            PrimaryGatePolicy.ShouldDrainAlarmHistory(_localRole, queueSharingPeerIsPrimary: peerIsPrimary));

        // Gitea #537: the continuous-historization CAPTURE verdict. Same role rule as the drain gate
-        // above, but a different peer question — historization is not pair-local, it is fleet-wide by
-        // historian tag name, so the only node worth standing down for is one recording the SAME refs:
-        // a mesh partner serving this node's address space. With no way to establish that (a legacy
-        // node carrying no cluster-scoped role, or a test harness), the predicate is null and this
-        // node keeps capturing. Duplicate samples are recoverable; a hole in the trend is not.
-        var historizationPeerIsPrimary =
-            _sharesHistorizationScopeWithLocal is { } sharesScope
-            && msg.Nodes.Any(n => n.Role == RedundancyRole.Primary
-                                  && n.NodeId != _localNode
-                                  && sharesScope(n.NodeId));
-
-        _historizationRoleView?.PublishHistorizationRole(
-            PrimaryGatePolicy.ShouldRecordHistorizedValues(
-                _localRole, addressSpaceSharingPeerIsPrimary: historizationPeerIsPrimary));
+        // above, but a different peer question — see ComputeHistorizationVerdict. Cached (#672) so a
+        // reachability change can re-derive it before the next snapshot arrives.
+        _lastRedundancySnapshotNodes = msg.Nodes;
+        if (_historizationRoleView is not null)
+        {
+            var historizationVerdict = ComputeHistorizationVerdict(msg.Nodes);
+            _lastPublishedHistorizationVerdict = historizationVerdict;
+            _historizationRoleView.PublishHistorizationRole(historizationVerdict);
+        }

        // The DATA-PLANE verdict, deliberately — the same one HandleRouteNodeWrite gates on, not the
        // drain verdict published immediately above. A driver reading this claims an exclusive identity
@@ -2230,6 +2322,66 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
        PublishPrimaryVerdict();
    }

+    /// <summary>
+    ///     The continuous-historization capture verdict for a snapshot (Gitea #537, #672).
+    /// </summary>
+    /// <remarks>
+    ///     <para>Historization is not pair-local, it is fleet-wide by historian tag name, so the only node worth
+    ///     standing down for is one recording the SAME refs: a mesh partner serving this node's address space. With
+    ///     no way to establish that (a legacy node carrying no cluster-scoped role, or a test harness), the
+    ///     predicate is null and this node keeps capturing. Duplicate samples are recoverable; a hole in the trend
+    ///     is not.</para>
+    ///     <para><b>#672 — an UNREACHABLE recording Primary does not count.</b> A crashed (SIGKILLed) Primary stays a
+    ///     cluster member until auto-down, and the redundancy singleton usually died with it, so the cached snapshot
+    ///     keeps naming it Primary for the whole failure-detector + auto-down window (~25 s) while nobody captures.
+    ///     Excluding unreachable peers shrinks that hole to the failure-detector window alone. The cost — a short
+    ///     duplicate window, and both nodes capturing for the duration of a genuine partition — is the same
+    ///     asymmetric trade #537 already makes. Gated by <c>ContinuousHistorization:CaptureWhilePrimaryUnreachable</c>.</para>
+    /// </remarks>
+    private bool ComputeHistorizationVerdict(IReadOnlyList<NodeRedundancyState> nodes)
+    {
+        var peerIsPrimary = false;
+        CommonsNodeId? unreachablePrimary = null;
+        if (_sharesHistorizationScopeWithLocal is { } sharesScope)
+        {
+            foreach (var n in nodes)
+            {
+                if (n.Role != RedundancyRole.Primary || n.NodeId == _localNode || !sharesScope(n.NodeId)) continue;
+                if (_captureWhilePrimaryUnreachable && _unreachableNodes.Contains(n.NodeId))
+                {
+                    unreachablePrimary ??= n.NodeId;
+                    continue;
+                }
+
+                peerIsPrimary = true;
+            }
+        }
+
+        var verdict = PrimaryGatePolicy.ShouldRecordHistorizedValues(
+            _localRole, addressSpaceSharingPeerIsPrimary: peerIsPrimary);
+
+        // Log only when the unreachable exclusion is what OPENED the gate — i.e. the verdict would have been
+        // "stand down" had that Primary been reachable.
+        var overriding = unreachablePrimary is not null && verdict
+            && !PrimaryGatePolicy.ShouldRecordHistorizedValues(_localRole, addressSpaceSharingPeerIsPrimary: true);
+        if (overriding && _capturingForUnreachablePrimary is null)
+        {
+            _capturingForUnreachablePrimary = unreachablePrimary;
+            _log.Info(
+                "DriverHost {Node}: capturing historized values while recording Primary {Primary} is unreachable (Gitea #672)",
+                _localNode, unreachablePrimary);
+        }
+        else if (!overriding && _capturingForUnreachablePrimary is { } previous)
+        {
+            _capturingForUnreachablePrimary = null;
+            _log.Info(
+                "DriverHost {Node}: no longer capturing in place of unreachable recording Primary {Primary}; capture verdict now {Verdict}",
+                _localNode, previous, verdict ? "capture" : "stand down");
+        }
+
+        return verdict;
+    }
+
    private void Stale()
    {
        Receive<DispatchDeployment>(_ =>
@@ -2251,6 +2403,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
        Receive<ReconnectDriver>(HandleReconnectDriver);
        Receive<RedundancyStateChanged>(OnRedundancyStateChanged);
        Receive<Akka.Cluster.ClusterEvent.IMemberEvent>(OnDriverMembershipChanged);
+        Receive<Akka.Cluster.ClusterEvent.ReachabilityEvent>(OnReachabilityChanged);
        Receive<RepublishPrimaryVerdict>(_ => PublishPrimaryVerdictIfChanged());
        Receive<Akka.Cluster.ClusterEvent.CurrentClusterState>(_ => PublishPrimaryVerdictIfChanged());
        // The redundancy-state topic also carries OpcUaProbeResult (OpcUaPublishActor peer-probes).
@@ -98,6 +98,13 @@ public static class PrimaryGatePolicy
    ///         promoted node takes to see its first snapshot. That is inherent to any non-duplicating
    ///         design; the alternative — both nodes capturing — is what #537 was filed about.
    ///     </para>
+    ///     <para>
+    ///         Gitea #672: on a CRASH the newly promoted node may not see a snapshot until auto-down (the
+    ///         redundancy singleton died with the old Primary), so the caller excludes a Primary the cluster
+    ///         failure detector reports unreachable from <paramref name="addressSpaceSharingPeerIsPrimary"/> —
+    ///         shrinking the gap to the failure-detector window. This method is unchanged by that; it only
+    ///         sees the narrower input.
+    ///     </para>
    /// </remarks>
    /// <param name="localRole">This node's last-known redundancy role, or <c>null</c> when unknown.</param>
    /// <param name="addressSpaceSharingPeerIsPrimary">
@@ -129,6 +129,25 @@ public sealed class ContinuousHistorizationOptions
    /// <summary>Cap (seconds) on the exponential retry backoff after repeated drain failures. Defaults to 30.</summary>
    public double MaxBackoffSeconds { get; init; } = 30;

+    /// <summary>
+    ///     Gitea #672: when <c>true</c> (the default), a redundant node whose capture is stood down for its
+    ///     mesh-partner Primary (Gitea #537) resumes capturing as soon as the cluster failure detector reports
+    ///     that Primary <b>unreachable</b>, instead of waiting for auto-down and a fresh redundancy snapshot.
+    ///     <para>
+    ///         A crashed Primary otherwise leaves a capture hole of failure-detection + auto-down (~25 s on the
+    ///         shipped <c>akka.conf</c>): the redundancy singleton usually died with it, so the standby's last
+    ///         snapshot keeps naming the dead node Primary. With this on, the hole shrinks to the failure-detector
+    ///         window (~10–12 s). The cost is duplicates: during a genuine partition both nodes capture from
+    ///         detection onward, and a briefly-unreachable-but-alive Primary overlaps the standby. SqlServer and
+    ///         QuestDB absorb a duplicate of a source-timestamped sample idempotently; a driver with no source
+    ///         timestamp (e.g. Modbus, stamped with each node's own poll time) writes a duplicate row on every
+    ///         backend, and the Gateway backend stores duplicates. <c>false</c> restores the pre-#672 behaviour.
+    ///         Only the historization capture verdict is affected — never the alarm drain or the Primary data-plane
+    ///         gate.
+    ///     </para>
+    /// </summary>
+    public bool CaptureWhilePrimaryUnreachable { get; init; } = true;
+
    /// <summary>Returns operator-facing misconfiguration warnings for an <c>Enabled</c> recorder
    /// (empty when disabled or correctly configured). Pure — the registration logs each entry.</summary>
    /// <returns>Zero or more human-readable warning messages (never carrying secret values).</returns>
@@ -625,6 +625,9 @@ public static class ServiceCollectionExtensions
                    // verdict, and how it decides which peer is worth standing down for.
                    historizationRoleView: resolver.GetService<IHistorizationRoleView>(),
                    sharesHistorizationScopeWithLocal: historizationScopePeer,
+                    // Gitea #672: stop deferring the capture to a recording Primary the failure detector
+                    // already reports unreachable. Unbound section ⇒ the option's default (true).
+                    captureWhilePrimaryUnreachable: continuousOptions?.CaptureWhilePrimaryUnreachable ?? true,
                    // Review R6 MINOR-4: the passive half of a LocalDb pair has no PeerAddress, so it is told
                    // its queue-sharing peer as its mesh partner instead — otherwise it never stands down.
                    sharesAlarmQueueWithLocal: BuildAlarmQueuePeerPredicate(