fix(communication): release the _activeStreams slot when relay ActorOf throws (review 2026-10-21 03/M-2)
This commit is contained in:
@@ -413,10 +413,28 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
|
|||||||
correlationId, streamKind, n, channelCapacity);
|
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 actorSeq = Interlocked.Increment(ref _actorCounter);
|
||||||
var relayActor = _actorSystem!.ActorOf(
|
IActorRef relayActor;
|
||||||
Props.Create(typeof(Actors.StreamRelayActor), correlationId, channel.Writer),
|
try
|
||||||
$"stream-relay-{correlationId}-{actorSeq}");
|
{
|
||||||
|
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<string, StreamEntry>(correlationId, entry));
|
||||||
|
throw;
|
||||||
|
}
|
||||||
|
|
||||||
// The previous code called _streamSubscriber.Subscribe
|
// The previous code called _streamSubscriber.Subscribe
|
||||||
// OUTSIDE the try block that owns relay-actor cleanup. If Subscribe threw
|
// OUTSIDE the try block that owns relay-actor cleanup. If Subscribe threw
|
||||||
|
|||||||
@@ -478,6 +478,29 @@ public class SiteStreamGrpcServerTests : TestKit
|
|||||||
subscriber.DidNotReceive().RemoveSubscriber(Arg.Any<IActorRef>());
|
subscriber.DidNotReceive().RemoveSubscriber(Arg.Any<IActorRef>());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Review 2026-10-21 03/M-2: if <c>ActorOf</c> throws (the actor system terminating via
|
||||||
|
/// the downed-node watchdog, so <c>_shuttingDown</c> was never set), the
|
||||||
|
/// <c>_activeStreams</c> slot registered just before it must not leak.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[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<IServerStreamWriter<SiteStreamEvent>>();
|
||||||
|
await Assert.ThrowsAnyAsync<Exception>(
|
||||||
|
() => server.SubscribeInstance(MakeRequest("corr-m2"), writer, CreateMockContext()));
|
||||||
|
|
||||||
|
Assert.Equal(0, server.ActiveStreamCount);
|
||||||
|
_subscriber.DidNotReceive().Subscribe(Arg.Any<string>(), Arg.Any<IActorRef>());
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>Verifies that the active subscribers gauge rises to one on connect and returns to zero on cancellation.</summary>
|
/// <summary>Verifies that the active subscribers gauge rises to one on connect and returns to zero on cancellation.</summary>
|
||||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|||||||
Reference in New Issue
Block a user