chore(communication): low-severity cleanup — eviction counting, shared ingest timeout, split app heartbeat interval, doc notes

docs/requirements/Component-Communication.md

@@ -64,6 +64,8 @@ Both central and site clusters. Each side has communication actors that handle m
 - Alarm state stream messages: `[InstanceUniqueName].[AlarmName]`, state (active/normal), priority, timestamp.
 - Central sends an unsubscribe request via ClusterClient when the debug session ends. The gRPC stream is cancelled. The site's `StreamRelayActor` is stopped and the SiteStreamManager subscription is removed.
 - The stream is session-based and temporary.
+- **Accepted limitation (central-side session locality):** debug sessions are process-local state on the central node hosting the `DebugStreamBridgeActor`. A central **restart or failover drops all active sessions** with no `OnStreamTerminated` signal to the operator — the engineer simply re-establishes the debug session from the UI. Debug streaming is an interactive, transient diagnostic aid, so this is an accepted trade-off rather than a durability gap (arch review 02, U7).
+- **Backpressure is lossy-by-design and now observable:** the per-session `Channel<SiteStreamEvent>` (bounded 1000, `DropOldest`) silently evicts the oldest event when a slow consumer falls behind. The eviction is counted via the channel's `itemDropped` callback and logged at Warning (first eviction, then every 500th) so real event loss on a debug stream is visible instead of silent.
 
 #### Site-Side gRPC Streaming Components
 

src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs

@@ -137,20 +137,18 @@ public class CentralCommunicationActor : ReceiveActor
 
     /// <summary>
     /// Default Ask timeout for routing audit ingest commands to the
-    /// AuditLogIngestActor proxy — 30 s, matching the value of
-    /// <c>SiteStreamGrpcServer.AuditIngestAskTimeout</c> (that constant is private
-    /// to the gRPC server and not reachable here, so it is declared locally). A
-    /// generous window absorbs a slow MS SQL connection without the round-trip
-    /// surfacing as a failure on a healthy site. When the window is exceeded the
-    /// Ask faults and that fault is piped back to the caller as a
-    /// <see cref="Status.Failure"/> (see <see cref="HandleIngestAuditEvents"/>).
-    /// </summary>
-    private static readonly TimeSpan DefaultAuditIngestAskTimeout = TimeSpan.FromSeconds(30);
-
-    /// <summary>
     /// Effective Ask timeout for audit ingest routing. Defaults to
-    /// <see cref="DefaultAuditIngestAskTimeout"/>; overridable via the constructor
+    /// <see cref="Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout"/> (30 s) — the two
-    /// so tests can exercise the timeout/fault path without waiting 30 s.
+    /// audit-ingest transports (gRPC vs ClusterClient) now share one source of truth
+    /// for the timeout. Overridable via the constructor so tests can exercise the
+    /// timeout/fault path without waiting 30 s. When the window is exceeded the Ask
+    /// faults and that fault is piped back to the caller as a
+    /// <see cref="Status.Failure"/> (see <see cref="HandleIngestAuditEvents"/>).
+    ///
+    /// The gRPC and ClusterClient ingest transports differ in fault semantics (the
+    /// gRPC handler surfaces the fault as an RpcException; here it becomes a piped
+    /// Status.Failure). Consolidating the two ingest transports is a scheduled
+    /// follow-up — see arch review 02, Underdeveloped #1.
     /// </summary>
     private readonly TimeSpan _auditIngestAskTimeout;
 
@@ -166,7 +164,7 @@ public class CentralCommunicationActor : ReceiveActor
     /// <param name="siteClientFactory">Factory used to create per-site ClusterClient actors.</param>
     /// <param name="auditIngestAskTimeout">
     /// Optional override for the audit-ingest Ask timeout; defaults to
-    /// <see cref="DefaultAuditIngestAskTimeout"/> (30 s). Exists only so tests can
+    /// <see cref="Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout"/> (30 s). Exists only so tests can
     /// exercise the timeout/fault path quickly — production always uses the default.
     /// </param>
     public CentralCommunicationActor(
@@ -176,7 +174,7 @@ public class CentralCommunicationActor : ReceiveActor
     {
         _serviceProvider = serviceProvider;
         _siteClientFactory = siteClientFactory;
-        _auditIngestAskTimeout = auditIngestAskTimeout ?? DefaultAuditIngestAskTimeout;
+        _auditIngestAskTimeout = auditIngestAskTimeout ?? Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout;
 
         // Site address cache loaded from database
         Receive<SiteAddressCacheLoaded>(HandleSiteAddressCacheLoaded);

src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs

@@ -434,12 +434,14 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
     {
         _log.Info("SiteCommunicationActor started for site {0}", _siteId);
 
-        // Schedule periodic heartbeat to central
+        // Schedule periodic heartbeat to central. Uses the application heartbeat
+        // cadence — distinct from the Akka.Remote transport failure-detector
+        // interval — so the two can be tuned independently.
         Timers.StartPeriodicTimer(
             "heartbeat",
             new SendHeartbeat(),
             TimeSpan.FromSeconds(1), // initial delay
-            _options.TransportHeartbeatInterval);
+            _options.ApplicationHeartbeatInterval);
     }
 
     private void HandleRegisterLocalHandler(RegisterLocalHandler msg)

src/ZB.MOM.WW.ScadaBridge.Communication/Actors/StreamRelayActor.cs

@@ -104,9 +104,13 @@ public class StreamRelayActor : ReceiveActor
 
     private void WriteToChannel(SiteStreamEvent protoEvent)
     {
+        // TryWrite on a DropOldest bounded channel never returns false for a "full"
+        // channel — eviction of the oldest item is what happens instead, and it is
+        // counted by the channel's itemDropped callback (SiteStreamGrpcServer). This
+        // branch therefore guards only a completed/closed channel, not backpressure.
         if (!_channelWriter.TryWrite(protoEvent))
         {
-            _log.Warning("Channel full, dropping event for correlation {0}", _correlationId);
+            _log.Debug("Channel closed, dropping event for correlation {0}", _correlationId);
         }
     }
 

src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs

@@ -56,6 +56,14 @@ public class CommunicationOptions
     /// <summary>Akka.Remote transport heartbeat interval.</summary>
     public TimeSpan TransportHeartbeatInterval { get; set; } = TimeSpan.FromSeconds(5);
 
+    /// <summary>
+    /// Application-level site→central heartbeat cadence. Distinct from
+    /// <see cref="TransportHeartbeatInterval"/> (the Akka.Remote failure-detector
+    /// setting) — tuning the transport FD must not silently retune the health
+    /// heartbeat. Default equals the old effective value (5s) — no behavior change.
+    /// </summary>
+    public TimeSpan ApplicationHeartbeatInterval { get; set; } = TimeSpan.FromSeconds(5);
+
     /// <summary>Akka.Remote transport failure detection threshold.</summary>
     public TimeSpan TransportFailureThreshold { get; set; } = TimeSpan.FromSeconds(15);
 

src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs

@@ -42,7 +42,9 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
     // Ask timeout is 30 s. The ingest actor's repo
     // calls are sub-100 ms in steady state; a generous timeout absorbs a slow
     // MSSQL connection without surfacing as a gRPC failure on a healthy site.
-    private static readonly TimeSpan AuditIngestAskTimeout = TimeSpan.FromSeconds(30);
+    // Shared with CentralCommunicationActor (same assembly) so the two audit-ingest
+    // transports use one source of truth for the timeout.
+    internal static readonly TimeSpan AuditIngestAskTimeout = TimeSpan.FromSeconds(30);
     // Audit Log: site-local queue handed in by AkkaHostedService on
     // site roles so the central reconciliation puller's PullAuditEvents RPC
     // can read Pending/Forwarded rows. Null when not wired (e.g. central-only
@@ -240,7 +242,10 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
             existingEntry.Cts.Dispose();
         }
 
-        // Check max concurrent streams after duplicate removal
+        // Check max concurrent streams after duplicate removal.
+        // Deliberately check-then-act: two concurrent subscribes at the boundary can
+        // both pass, overshooting the cap by at most (concurrent subscribers - 1).
+        // Bounded and benign — not worth a lock on this path.
         if (_activeStreams.Count >= _maxConcurrentStreams)
             throw new RpcException(new GrpcStatus(StatusCode.ResourceExhausted, "Max concurrent streams reached"));
 
@@ -253,8 +258,20 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
         var entry = new StreamEntry(streamCts);
         _activeStreams[request.CorrelationId] = entry;
 
+        long dropped = 0;
+        var correlationId = request.CorrelationId;
         var channel = Channel.CreateBounded<SiteStreamEvent>(
-            new BoundedChannelOptions(1000) { FullMode = BoundedChannelFullMode.DropOldest });
+            new BoundedChannelOptions(1000) { FullMode = BoundedChannelFullMode.DropOldest },
+            _ =>
+            {
+                // Lossy-under-backpressure is the debug-view spec; make the real
+                // loss visible: first eviction + every 500th thereafter.
+                var n = Interlocked.Increment(ref dropped);
+                if (n == 1 || n % 500 == 0)
+                    _logger.LogWarning(
+                        "Debug stream {CorrelationId} backpressure: {Dropped} oldest event(s) evicted so far",
+                        correlationId, n);
+            });
 
         var actorSeq = Interlocked.Increment(ref _actorCounter);
         var relayActor = _actorSystem!.ActorOf(

tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsTests.cs

@@ -58,4 +58,15 @@ public class CommunicationOptionsTests
         Assert.Equal(TimeSpan.FromMinutes(5), options.DeploymentTimeout);
         Assert.Equal(TimeSpan.FromSeconds(2), options.TransportHeartbeatInterval);
     }
+
+    [Fact]
+    public void ApplicationHeartbeatInterval_DefaultsTo5s_IndependentOfTransport()
+    {
+        var o = new CommunicationOptions();
+        Assert.Equal(TimeSpan.FromSeconds(5), o.ApplicationHeartbeatInterval);
+        // Retuning the Akka.Remote failure-detector heartbeat must NOT silently
+        // retune the application-level site→central health heartbeat.
+        o.TransportHeartbeatInterval = TimeSpan.FromSeconds(30);
+        Assert.Equal(TimeSpan.FromSeconds(5), o.ApplicationHeartbeatInterval);
+    }
 }