fix(communication): streaming consumers re-resolve site gRPC addresses on every reopen; spec matches (review 2026-10-21 03/R-2, 03/D-2)

2026-10-22 06:21:33 -04:00
parent 3ed71167b3
commit 90306f0ccb
8 changed files with 438 additions and 34 deletions
@@ -40,8 +40,25 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
    private readonly Action<object> _onEvent;
    private readonly Action _onTerminated;
    private readonly SiteStreamGrpcClientFactory _grpcFactory;
-    private readonly string _grpcNodeAAddress;
-    private readonly string _grpcNodeBAddress;
+    // Mutable: refreshed from _endpointResolver before each reconnect (review 2026-10-21 03/R-2).
+    private string _grpcNodeAAddress;
+    private string _grpcNodeBAddress;
+
+    /// <summary>
+    /// Re-reads the site's current gRPC endpoints before every reconnect, so an address edited
+    /// mid-session is picked up rather than retried against the start-time address. Null (tests,
+    /// legacy ctor) keeps the ctor-supplied addresses for the session's life.
+    /// </summary>
+    private readonly SiteStreamEndpointResolver? _endpointResolver;
+
+    /// <summary>True while a pre-reconnect endpoint lookup is running. Actor-thread only.</summary>
+    private bool _resolveInFlight;
+
+    /// <summary>The endpoint the current/last stream was opened against (for its unsubscribe). Actor-thread only.</summary>
+    private string? _streamEndpoint;
+
+    /// <summary>Upper bound on one pre-reconnect endpoint lookup; on expiry the reconnect uses the known endpoints.</summary>
+    internal static readonly TimeSpan EndpointResolveTimeout = TimeSpan.FromSeconds(10);

    private const int MaxRetries = 3;
    private const string ReconnectTimerKey = "grpc-reconnect";
@@ -178,7 +195,38 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
        SiteStreamGrpcClientFactory grpcFactory,
        string grpcNodeAAddress,
        string grpcNodeBAddress)
+        : this(siteIdentifier, instanceUniqueName, correlationId, centralCommunicationActor, onEvent,
+            onTerminated, grpcFactory, grpcNodeAAddress, grpcNodeBAddress, endpointResolver: null)
    {
+    }
+
+    /// <summary>
+    /// Initializes the bridge with an endpoint resolver consulted before every reconnect
+    /// (review 2026-10-21 03/R-2). The address arguments seed the first open.
+    /// </summary>
+    /// <param name="siteIdentifier">Site identifier for targeting site-addressed messages and logging.</param>
+    /// <param name="instanceUniqueName">Unique name of the instance whose debug stream is being bridged.</param>
+    /// <param name="correlationId">Correlation id for the debug session.</param>
+    /// <param name="centralCommunicationActor">Actor used to forward site-addressed messages to the site.</param>
+    /// <param name="onEvent">Callback invoked on each received debug event.</param>
+    /// <param name="onTerminated">Callback invoked when the stream terminates.</param>
+    /// <param name="grpcFactory">Factory for creating gRPC streaming clients.</param>
+    /// <param name="grpcNodeAAddress">gRPC address of the site's node A at session start.</param>
+    /// <param name="grpcNodeBAddress">gRPC address of the site's node B at session start.</param>
+    /// <param name="endpointResolver">Re-reads the site's endpoints before each reconnect; null keeps the start-time ones.</param>
+    public DebugStreamBridgeActor(
+        string siteIdentifier,
+        string instanceUniqueName,
+        string correlationId,
+        IActorRef centralCommunicationActor,
+        Action<object> onEvent,
+        Action onTerminated,
+        SiteStreamGrpcClientFactory grpcFactory,
+        string grpcNodeAAddress,
+        string grpcNodeBAddress,
+        SiteStreamEndpointResolver? endpointResolver)
+    {
+        _endpointResolver = endpointResolver;
        _siteIdentifier = siteIdentifier;
        _instanceUniqueName = instanceUniqueName;
        _correlationId = correlationId;
@@ -318,7 +366,16 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
        });

        // Scheduled reconnection
-        Receive<ReconnectGrpcStream>(_ => OpenGrpcStream());
+        Receive<ReconnectGrpcStream>(_ => ReopenGrpcStream());
+
+        // Pre-reconnect endpoint lookup finished (review 2026-10-21 03/R-2).
+        Receive<SiteStreamEndpointsResolved>(msg =>
+        {
+            _resolveInFlight = false;
+            if (_stopped) return;
+            ApplyResolvedEndpoints(msg);
+            OpenGrpcStream();
+        });

        // Consumer requests stop
        Receive<StopDebugStream>(_ =>
@@ -574,11 +631,64 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
        base.PostStop();
    }

+    /// <summary>
+    /// Reconnects the stream, first re-reading the site's endpoints when a resolver is configured
+    /// (review 2026-10-21 03/R-2), so a session whose site address was edited reconnects to the
+    /// new address instead of burning its retries on the old one. A lookup that fails, times out
+    /// or finds no address reconnects against the endpoints already known.
+    /// </summary>
+    private void ReopenGrpcStream()
+    {
+        if (_stopped) return;
+        if (_endpointResolver is null)
+        {
+            OpenGrpcStream();
+            return;
+        }
+        if (_resolveInFlight) return; // the pending lookup opens when it completes
+
+        _resolveInFlight = true;
+        var resolver = _endpointResolver;
+        Task.Run(async () =>
+        {
+            using var cts = new CancellationTokenSource(EndpointResolveTimeout);
+            try
+            {
+                return new SiteStreamEndpointsResolved(await resolver(cts.Token).ConfigureAwait(false), null);
+            }
+            catch (Exception ex)
+            {
+                return new SiteStreamEndpointsResolved(null, ex);
+            }
+        }).PipeTo(Self);
+    }
+
+    private void ApplyResolvedEndpoints(SiteStreamEndpointsResolved msg)
+    {
+        if (msg.Error is not null)
+        {
+            _log.Warning("Could not re-read gRPC endpoints for site {0} before reconnecting ({1}); using {2} / {3}",
+                _siteIdentifier, msg.Error.Message, _grpcNodeAAddress, _grpcNodeBAddress);
+            return;
+        }
+        if (msg.Endpoints is not { } endpoints
+            || (endpoints.NodeA == _grpcNodeAAddress && endpoints.NodeB == _grpcNodeBAddress))
+        {
+            return;
+        }
+
+        _log.Info("Site {0} gRPC endpoints changed ({1} / {2} -> {3} / {4}); reconnecting the debug stream for {5} to the new address",
+            _siteIdentifier, _grpcNodeAAddress, _grpcNodeBAddress, endpoints.NodeA, endpoints.NodeB, _instanceUniqueName);
+        _grpcNodeAAddress = endpoints.NodeA;
+        _grpcNodeBAddress = endpoints.NodeB;
+    }
+
    private void OpenGrpcStream()
    {
        if (_stopped) return;

        var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
+        _streamEndpoint = endpoint;
        _log.Info("Opening gRPC stream for {0} to {1}", _instanceUniqueName, endpoint);

        _grpcCts?.Cancel();
@@ -637,7 +747,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
        _grpcCts?.Cancel();
        _grpcCts?.Dispose();
        _grpcCts = null;
-        _grpcFactory.TryGet(_siteIdentifier, _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress)
+        _grpcFactory.TryGet(_siteIdentifier, _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress))
            ?.Unsubscribe(_correlationId);

        Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectGrpcStream(), ReconnectDelay);
@@ -668,7 +778,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers
        // alive -- propagates gRPC cancellation to the site so its SiteStreamGrpcServer
        // stops the StreamRelayActor for this correlation ID, rather than leaving a
        // zombie relay actor until TCP RST / keepalive eventually detects the loss.
-        var previousEndpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
+        var previousEndpoint = _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress);
        // TryGet, not GetOrCreate: unsubscribing a failed stream must never open a
        // fresh channel (and, with (site,endpoint) keying, must never touch another
        // session's healthy channel). Absent client => the channel is already gone
@@ -697,7 +807,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers

        // TryGet, not GetOrCreate: teardown must never open a fresh channel just to
        // unsubscribe. Absent client => nothing to cancel.
-        var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
+        var endpoint = _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress);
        _grpcFactory.TryGet(_siteIdentifier, endpoint)?.Unsubscribe(_correlationId);
    }

@@ -53,8 +53,25 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
    private readonly Func<CancellationToken, Task<IReadOnlyList<AlarmStateChanged>>> _seedFn;
    private readonly Action<IReadOnlyList<AlarmStateChanged>, bool> _publish;
    private readonly SiteStreamGrpcClientFactory _grpcFactory;
-    private readonly string _grpcNodeAAddress;
-    private readonly string _grpcNodeBAddress;
+    // Mutable: refreshed from _endpointResolver before each reopen (review 2026-10-21 03/R-2).
+    private string _grpcNodeAAddress;
+    private string _grpcNodeBAddress;
+
+    /// <summary>
+    /// Re-reads the site's current gRPC endpoints before every reopen, so an address edited while
+    /// viewers are attached is picked up instead of reopening against the start-time address
+    /// forever. Null (tests, legacy ctor) keeps the ctor-supplied addresses for the actor's life.
+    /// </summary>
+    private readonly SiteStreamEndpointResolver? _endpointResolver;
+
+    /// <summary>True while a pre-reopen endpoint lookup is running. Actor-thread only.</summary>
+    private bool _resolveInFlight;
+
+    /// <summary>The endpoint the current/last stream was opened against (for its unsubscribe). Actor-thread only.</summary>
+    private string? _streamEndpoint;
+
+    /// <summary>Upper bound on one pre-reopen endpoint lookup; on expiry the reopen uses the known endpoints.</summary>
+    internal static readonly TimeSpan EndpointResolveTimeout = TimeSpan.FromSeconds(10);
    private readonly TimeSpan _reconcileInterval;
    private readonly TimeSpan _publishCoalesce;
    private readonly double _reconcileJitterFraction;
@@ -302,7 +319,47 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        TimeSpan stabilityWindow,
        double reconcileJitterFraction = 0.0,
        TimeSpan? openAttemptTimeout = null)
+        : this(siteIdentifier, correlationId, seedFn, publish, grpcFactory, grpcNodeAAddress, grpcNodeBAddress,
+            reconcileInterval, publishCoalesce, reconnectDelay, stabilityWindow, reconcileJitterFraction,
+            openAttemptTimeout, endpointResolver: null)
    {
+    }
+
+    /// <summary>
+    /// Initializes the aggregator with an endpoint resolver that is consulted before every stream
+    /// reopen (review 2026-10-21 03/R-2). The address arguments seed the first open.
+    /// </summary>
+    /// <param name="siteIdentifier">Site identifier (used for gRPC client keying and logging).</param>
+    /// <param name="correlationId">Correlation id for the site-wide alarm stream subscription.</param>
+    /// <param name="seedFn">Snapshot fan-out returning the whole site's current alarm rows.</param>
+    /// <param name="publish">Publishes a fresh immutable cache snapshot plus the stream-liveness flag.</param>
+    /// <param name="grpcFactory">Factory caching one gRPC client per (site, endpoint).</param>
+    /// <param name="grpcNodeAAddress">gRPC address of the site's node A at start.</param>
+    /// <param name="grpcNodeBAddress">gRPC address of the site's node B at start.</param>
+    /// <param name="reconcileInterval">Periodic reconcile snapshot cadence.</param>
+    /// <param name="publishCoalesce">Publish-coalescing window for live deltas.</param>
+    /// <param name="reconnectDelay">Delay between gRPC reconnection attempts.</param>
+    /// <param name="stabilityWindow">How long a fresh gRPC stream must stay up before its retry budget recovers.</param>
+    /// <param name="reconcileJitterFraction">Fraction of the reconcile interval added as per-tick jitter.</param>
+    /// <param name="openAttemptTimeout">Deadline on one stream open attempt; null takes the default.</param>
+    /// <param name="endpointResolver">Re-reads the site's endpoints before each reopen; null keeps the start-time ones.</param>
+    public SiteAlarmAggregatorActor(
+        string siteIdentifier,
+        string correlationId,
+        Func<CancellationToken, Task<IReadOnlyList<AlarmStateChanged>>> seedFn,
+        Action<IReadOnlyList<AlarmStateChanged>, bool> publish,
+        SiteStreamGrpcClientFactory grpcFactory,
+        string grpcNodeAAddress,
+        string grpcNodeBAddress,
+        TimeSpan reconcileInterval,
+        TimeSpan publishCoalesce,
+        TimeSpan reconnectDelay,
+        TimeSpan stabilityWindow,
+        double reconcileJitterFraction,
+        TimeSpan? openAttemptTimeout,
+        SiteStreamEndpointResolver? endpointResolver)
+    {
+        _endpointResolver = endpointResolver;
        _siteIdentifier = siteIdentifier;
        _correlationId = correlationId;
        _seedFn = seedFn;
@@ -414,7 +471,16 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
            HandleGrpcCompleted();
        });

-        Receive<ReconnectAlarmStream>(_ => OpenGrpcStream());
+        Receive<ReconnectAlarmStream>(_ => ReopenGrpcStream());
+
+        // Pre-reopen endpoint lookup finished (review 2026-10-21 03/R-2).
+        Receive<SiteStreamEndpointsResolved>(msg =>
+        {
+            _resolveInFlight = false;
+            if (_stopped) return;
+            ApplyResolvedEndpoints(msg);
+            OpenGrpcStream();
+        });

        // Owning service asks us to stop (last viewer left + linger elapsed).
        Receive<StopSiteAlarmAggregator>(_ =>
@@ -513,7 +579,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
            // Telemetry: a reconcile-driven reopen after the stream was given up is a reconnect.
            ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();
            _seedOnConnect = true;
-            OpenGrpcStream();
+            ReopenGrpcStream();
            reopening = true;
        }

@@ -843,6 +909,66 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers

    // ── gRPC stream lifecycle (mirrors DebugStreamBridgeActor) ──────────────────

+    /// <summary>
+    /// Reopens the stream, first re-reading the site's endpoints when a resolver is configured
+    /// (review 2026-10-21 03/R-2): an address edited while the aggregator runs used to be
+    /// ignored, so every reopen — retries and the reconcile tick's self-heal alike — dialled
+    /// the dead start-time address for as long as viewers stayed attached. The lookup holds the
+    /// open-in-flight guard so a reconcile tick cannot start a second reopen meanwhile; a lookup
+    /// that fails, times out or finds no address reopens against the endpoints already known.
+    /// </summary>
+    private void ReopenGrpcStream()
+    {
+        if (_stopped) return;
+        if (_endpointResolver is null)
+        {
+            OpenGrpcStream();
+            return;
+        }
+        if (_resolveInFlight) return; // the pending lookup opens when it completes
+
+        _resolveInFlight = true;
+        _openInFlight = true;
+        var resolver = _endpointResolver;
+        var lifetime = _lifetimeCts?.Token ?? CancellationToken.None;
+        Task.Run(async () =>
+        {
+            using var cts = CancellationTokenSource.CreateLinkedTokenSource(lifetime);
+            cts.CancelAfter(EndpointResolveTimeout);
+            try
+            {
+                return new SiteStreamEndpointsResolved(await resolver(cts.Token).ConfigureAwait(false), null);
+            }
+            catch (Exception ex)
+            {
+                return new SiteStreamEndpointsResolved(null, ex);
+            }
+        }).PipeTo(Self);
+    }
+
+    private void ApplyResolvedEndpoints(SiteStreamEndpointsResolved msg)
+    {
+        if (msg.Error is not null)
+        {
+            _log.Warning("Could not re-read gRPC endpoints for site {0} before reopening ({1}); using {2} / {3}",
+                _siteIdentifier, msg.Error.Message, _grpcNodeAAddress, _grpcNodeBAddress);
+            return;
+        }
+        if (msg.Endpoints is not { } endpoints)
+        {
+            _log.Warning("Site {0} has no gRPC endpoints configured any more; reopening against {1} / {2}",
+                _siteIdentifier, _grpcNodeAAddress, _grpcNodeBAddress);
+            return;
+        }
+        if (endpoints.NodeA == _grpcNodeAAddress && endpoints.NodeB == _grpcNodeBAddress)
+            return;
+
+        _log.Info("Site {0} gRPC endpoints changed ({1} / {2} -> {3} / {4}); reopening the alarm stream against the new address",
+            _siteIdentifier, _grpcNodeAAddress, _grpcNodeBAddress, endpoints.NodeA, endpoints.NodeB);
+        _grpcNodeAAddress = endpoints.NodeA;
+        _grpcNodeBAddress = endpoints.NodeB;
+    }
+
    private void OpenGrpcStream()
    {
        if (_stopped) return;
@@ -850,6 +976,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        // The stream is not live again until the server actually accepts it — liveness flips
        // on GrpcAlarmStreamConnected, not on the attempt.
        var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
+        _streamEndpoint = endpoint;
        _log.Info("Opening site-alarm gRPC stream for {0} to {1}", _siteIdentifier, endpoint);

        _grpcCts?.Cancel();
@@ -962,7 +1089,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers

        // Unsubscribe the failed stream on the previous endpoint (TryGet, never
        // GetOrCreate) so the site releases its relay actor instead of leaving a zombie.
-        var previousEndpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
+        var previousEndpoint = _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress);
        _grpcFactory.TryGet(_siteIdentifier, previousEndpoint)?.Unsubscribe(_correlationId);

        // Flip to the other node.
@@ -1016,7 +1143,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        _grpcCts?.Dispose();
        _grpcCts = null;

-        var endpoint = _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress;
+        var endpoint = _streamEndpoint ?? (_useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress);
        _grpcFactory.TryGet(_siteIdentifier, endpoint)?.Unsubscribe(_correlationId);
    }