diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index 783327ab..83daa298 100644 --- a/docs/requirements/Component-Communication.md +++ b/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` (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 diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs index bcdc2a37..9f1ecf51 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs @@ -137,20 +137,18 @@ public class CentralCommunicationActor : ReceiveActor /// /// Default Ask timeout for routing audit ingest commands to the - /// AuditLogIngestActor proxy — 30 s, matching the value of - /// SiteStreamGrpcServer.AuditIngestAskTimeout (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 ). - /// - private static readonly TimeSpan DefaultAuditIngestAskTimeout = TimeSpan.FromSeconds(30); - - /// /// Effective Ask timeout for audit ingest routing. Defaults to - /// ; overridable via the constructor - /// so tests can exercise the timeout/fault path without waiting 30 s. + /// (30 s) — the two + /// 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 ). + /// + /// 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. /// private readonly TimeSpan _auditIngestAskTimeout; @@ -166,7 +164,7 @@ public class CentralCommunicationActor : ReceiveActor /// Factory used to create per-site ClusterClient actors. /// /// Optional override for the audit-ingest Ask timeout; defaults to - /// (30 s). Exists only so tests can + /// (30 s). Exists only so tests can /// exercise the timeout/fault path quickly — production always uses the default. /// 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(HandleSiteAddressCacheLoaded); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs index 9cdfab9d..b3027ac6 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs +++ b/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) diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/StreamRelayActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/StreamRelayActor.cs index 0546296e..7f59aaa2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/StreamRelayActor.cs +++ b/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); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs index f794f9cc..5ae541b9 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs @@ -56,6 +56,14 @@ public class CommunicationOptions /// Akka.Remote transport heartbeat interval. public TimeSpan TransportHeartbeatInterval { get; set; } = TimeSpan.FromSeconds(5); + /// + /// Application-level site→central heartbeat cadence. Distinct from + /// (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. + /// + public TimeSpan ApplicationHeartbeatInterval { get; set; } = TimeSpan.FromSeconds(5); + /// Akka.Remote transport failure detection threshold. public TimeSpan TransportFailureThreshold { get; set; } = TimeSpan.FromSeconds(15); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs index 134e3ba6..fa35f99c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs +++ b/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( - 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( diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsTests.cs index 08773331..1cb4e276 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsTests.cs +++ b/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); + } }