diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs index 0130f00d..00795858 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs @@ -413,10 +413,28 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase correlationId, streamKind, n, channelCapacity); }); + // ActorOf can throw when the actor system is terminating through a path that never + // set _shuttingDown (the downed-node watchdog). The _activeStreams slot registered + // above would otherwise leak against GrpcMaxConcurrentStreams forever (review + // 2026-10-21 03/M-2), so release it — and the channel — before rethrowing. var actorSeq = Interlocked.Increment(ref _actorCounter); - var relayActor = _actorSystem!.ActorOf( - Props.Create(typeof(Actors.StreamRelayActor), correlationId, channel.Writer), - $"stream-relay-{correlationId}-{actorSeq}"); + IActorRef relayActor; + try + { + relayActor = _actorSystem!.ActorOf( + Props.Create(typeof(Actors.StreamRelayActor), correlationId, channel.Writer), + $"stream-relay-{correlationId}-{actorSeq}"); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "Could not create the relay actor for {Description} (correlation {CorrelationId}); releasing the stream slot.", + description, correlationId); + channel.Writer.TryComplete(); + _activeStreams.TryRemove( + new KeyValuePair(correlationId, entry)); + throw; + } // The previous code called _streamSubscriber.Subscribe // OUTSIDE the try block that owns relay-actor cleanup. If Subscribe threw diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs index d5457507..daca7c56 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs @@ -478,6 +478,29 @@ public class SiteStreamGrpcServerTests : TestKit subscriber.DidNotReceive().RemoveSubscriber(Arg.Any()); } + /// + /// Review 2026-10-21 03/M-2: if ActorOf throws (the actor system terminating via + /// the downed-node watchdog, so _shuttingDown was never set), the + /// _activeStreams slot registered just before it must not leak. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ActorOfThrows_DoesNotLeakActiveStreamSlot() + { + var terminated = ActorSystem.Create("m2-terminated"); + await terminated.Terminate(); + + var server = CreateServer(); + server.SetReady(terminated); + + var writer = Substitute.For>(); + await Assert.ThrowsAnyAsync( + () => server.SubscribeInstance(MakeRequest("corr-m2"), writer, CreateMockContext())); + + Assert.Equal(0, server.ActiveStreamCount); + _subscriber.DidNotReceive().Subscribe(Arg.Any(), Arg.Any()); + } + /// Verifies that the active subscribers gauge rises to one on connect and returns to zero on cancellation. /// A task that represents the asynchronous operation. [Fact]