From 561582747eb7544365ef9b161b82cabb49518697 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 19 Aug 2026 14:52:22 -0400 Subject: [PATCH] =?UTF-8?q?fix(centralui):=20stop=20debug-stream=20session?= =?UTF-8?q?s=20abandoned=20mid-start=20=E2=80=94=20dispose-vs-connect=20ra?= =?UTF-8?q?ce=20(arch-review=202026-08-19=20P6-6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A debug-stream session created after the consumer had already gone away was leaked permanently. Both consumers awaited StartStreamAsync (up to 30s) and only then published the session; a circuit teardown / client disconnect inside that window ran the release path first, found nothing registered and stopped nothing. The session stayed registered in the singleton DebugStreamService, whose keepalive renews registered sessions forever (the 5-minute orphan reaper only collects UNregistered ones), pinning the site gRPC subscription, the DebugStreamBridgeActor and the dead consumer's object graph until process restart. Both call sites now publish the session and re-check liveness under a single monitor, so exactly one side tears it down and neither can miss it: - DebugView.Connect/Dispose/Disconnect share a _sessionLock; Dispose sets _disposed and takes the session in one step, Connect publishes only if not disposed and otherwise stops the session it just created. Dispose can run off the renderer's thread, so the Blazor sync context is not relied on. Disconnect and the onTerminated callback go through the same take-once helper, which also makes a repeated Dispose stop-once instead of stop-again. - DebugStreamHub uses the per-connection Items monitor: OnDisconnectedAsync marks the connection gone before releasing, SubscribeInstance publishes only if neither that flag nor ConnectionAborted is set, and UnsubscribeInstance takes the id atomically. DebugStreamService.StartStreamAsync/StopStream become virtual purely as a test seam (no behaviour change, keepalive/orphan contract untouched) so the race can be driven deterministically by a gated stub. Tests: DebugViewConnectDisposeRaceTests (3) + DebugStreamHubSubscribeRaceTests (3). Verified fail-unfixed: with the two source fixes stashed, both dispose-mid-start tests report zero StopStream calls and the DebugView double-dispose test reports two. --- .../Pages/Deployment/DebugView.razor | 68 +++++- .../DebugStreamService.cs | 10 +- .../DebugStreamHub.cs | 62 ++++- .../DebugViewConnectDisposeRaceTests.cs | 211 ++++++++++++++++++ .../DebugStreamHubSubscribeRaceTests.cs | 179 +++++++++++++++ 5 files changed, 517 insertions(+), 13 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Deployment/DebugViewConnectDisposeRaceTests.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/DebugStreamHubSubscribeRaceTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Deployment/DebugView.razor b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Deployment/DebugView.razor index 0051a151..eb1f8247 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Deployment/DebugView.razor +++ b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Deployment/DebugView.razor @@ -343,6 +343,16 @@ } private DebugStreamSession? _session; + + /// + /// Guards the slot together with + /// (arch-review 2026-08-19 P6-6). Connect awaits StartStreamAsync for up + /// to 30s; the circuit can be torn down inside that window, and Dispose may run + /// on a non-UI thread, so publishing the session and observing disposal must be one + /// atomic step. Whoever loses the race is the one that stops the stream. + /// + private readonly object _sessionLock = new(); + private ToastNotification _toast = default!; private string? _initError; @@ -468,7 +478,7 @@ onTerminated: () => { _connected = false; - _session = null; + _ = TakeSession(); // CentralUI-009: skip the toast/render if already disposed. if (_disposed) return; _ = SafeInvokeAsync(() => @@ -490,7 +500,25 @@ return; } - _session = session; + // P6-6: the component may have been disposed while StartStreamAsync was in + // flight. Dispose() would have found _session still null and stopped nothing, + // so this session would stay registered in the singleton DebugStreamService — + // whose keepalive renews registered sessions forever, pinning the site + // subscription, the bridge actor and this dead circuit until process restart. + // Publish and re-check disposal atomically so exactly one side tears it down. + bool abandoned; + lock (_sessionLock) + { + abandoned = _disposed; + if (!abandoned) _session = session; + } + + if (abandoned) + { + DebugStreamService.StopStream(session.SessionId); + _connecting = false; + return; + } // Populate initial state from snapshot _attributeValues.Clear(); @@ -521,12 +549,26 @@ _connecting = false; } + /// + /// Atomically clears the slot and returns what was there, so + /// only the caller that actually took the session stops it (P6-6). + /// + private DebugStreamSession? TakeSession() + { + lock (_sessionLock) + { + var session = _session; + _session = null; + return session; + } + } + private async Task Disconnect() { - if (_session != null) + var session = TakeSession(); + if (session != null) { - DebugStreamService.StopStream(_session.SessionId); - _session = null; + DebugStreamService.StopStream(session.SessionId); } // Clear persisted selection — user explicitly disconnected @@ -725,10 +767,20 @@ { // CentralUI-009: mark disposed first so any in-flight stream callback // sees the flag and no-ops, then stop the stream synchronously. - _disposed = true; - if (_session != null) + // P6-6: the flag is set and the session taken under the same lock Connect + // publishes through, so a Connect still awaiting StartStreamAsync either + // hands the session over here or stops it itself — never neither. + DebugStreamSession? session; + lock (_sessionLock) { - DebugStreamService.StopStream(_session.SessionId); + _disposed = true; + session = _session; + _session = null; + } + + if (session != null) + { + DebugStreamService.StopStream(session.SessionId); } } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs index 2b10cc51..48602795 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs @@ -81,7 +81,12 @@ public class DebugStreamService : IDisposable /// Callback invoked when the stream terminates. /// A cancellation token that can be used to cancel the operation. /// A debug stream session with the initial snapshot. - public async Task StartStreamAsync( + /// + /// Virtual purely as a test seam: the consumers' dispose-vs-start race (P6-6) can only + /// be driven deterministically by a stub that holds the start open. Behaviour and the + /// keepalive/orphan contract are unchanged. + /// + public virtual async Task StartStreamAsync( int instanceId, Action onEvent, Action onTerminated, @@ -209,7 +214,8 @@ public class DebugStreamService : IDisposable /// Stops an active debug stream session. /// /// The session ID of the debug stream to stop. - public void StopStream(string sessionId) + /// Virtual purely as a test seam — see . + public virtual void StopStream(string sessionId) { if (_sessions.TryRemove(sessionId, out var bridgeActor)) { diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/DebugStreamHub.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/DebugStreamHub.cs index 429cda0e..87758fd7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/DebugStreamHub.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/DebugStreamHub.cs @@ -19,6 +19,7 @@ public class DebugStreamHub : Hub private const string SessionIdKey = "DebugStreamSessionId"; private const string RolesKey = "DebugStreamRoles"; private const string PermittedSiteIdsKey = "DebugStreamPermittedSiteIds"; + private const string DisconnectedKey = "DebugStreamDisconnected"; /// /// Pure site-scope authorization check for a debug-stream subscription. @@ -192,7 +193,35 @@ public class DebugStreamHub : Hub CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default); }); - Context.Items[SessionIdKey] = session.SessionId; + // arch-review 2026-08-19 P6-6: the client may have disconnected while + // StartStreamAsync was in flight (up to 30s). OnDisconnectedAsync already ran + // and found no session id to release, so publishing it here would leave the + // session registered in the singleton DebugStreamService — whose keepalive + // renews registered sessions forever, pinning the site subscription and its + // bridge actor until process restart. Publish and re-check liveness under the + // per-connection Items monitor (the same one UnsubscribeInstance takes) so + // exactly one side tears the session down. + var abandoned = false; + lock (Context.Items) + { + if (Context.Items.ContainsKey(DisconnectedKey) || Context.ConnectionAborted.IsCancellationRequested) + { + abandoned = true; + } + else + { + Context.Items[SessionIdKey] = session.SessionId; + } + } + + if (abandoned) + { + _debugStreamService.StopStream(session.SessionId); + _logger.LogInformation( + "DebugStreamHub: {ConnectionId} disconnected while subscribing to instance {InstanceId} — session released", + connectionId, instanceId); + return; + } // Send the initial snapshot await Clients.Caller.SendAsync("OnSnapshot", session.InitialSnapshot); @@ -214,18 +243,45 @@ public class DebugStreamHub : Hub /// A task that represents the asynchronous operation. public Task UnsubscribeInstance() { - if (Context.Items.TryGetValue(SessionIdKey, out var sessionIdObj) && sessionIdObj is string sessionId) + var sessionId = TakeSessionId(); + if (sessionId is not null) { _debugStreamService.StopStream(sessionId); - Context.Items.Remove(SessionIdKey); _logger.LogInformation("DebugStreamHub: {ConnectionId} unsubscribed", Context.ConnectionId); } return Task.CompletedTask; } + /// + /// Atomically removes and returns the session id registered for this connection, so a + /// release path and the late publish in cannot both + /// claim (or both miss) the same session — see P6-6 there. + /// + /// The session id that was registered, or if none was. + private string? TakeSessionId() + { + lock (Context.Items) + { + if (Context.Items.TryGetValue(SessionIdKey, out var sessionIdObj) && sessionIdObj is string sessionId) + { + Context.Items.Remove(SessionIdKey); + return sessionId; + } + } + return null; + } + /// public override async Task OnDisconnectedAsync(Exception? exception) { + // P6-6: mark the connection gone BEFORE releasing, so a SubscribeInstance still + // awaiting StartStreamAsync sees the flag and stops its own session rather than + // registering it against a connection nothing will ever unsubscribe. + lock (Context.Items) + { + Context.Items[DisconnectedKey] = true; + } + await UnsubscribeInstance(); await base.OnDisconnectedAsync(exception); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Deployment/DebugViewConnectDisposeRaceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Deployment/DebugViewConnectDisposeRaceTests.cs new file mode 100644 index 00000000..2bd9f17c --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Deployment/DebugViewConnectDisposeRaceTests.cs @@ -0,0 +1,211 @@ +using System.Reflection; +using System.Security.Claims; +using Bunit; +using Microsoft.AspNetCore.Components.Authorization; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.CentralUI.Auth; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; +using DebugViewPage = ZB.MOM.WW.ScadaBridge.CentralUI.Components.Pages.Deployment.DebugView; + +namespace ZB.MOM.WW.ScadaBridge.CentralUI.Tests.Deployment; + +/// +/// Regression tests for arch-review 2026-08-19 P6-6. Connect() awaits +/// StartStreamAsync — which can take up to 30s — and then publishes the result into +/// _session. If the circuit was torn down inside that window, Dispose() had +/// already run, found _session still null and stopped nothing, so the just-created +/// session stayed registered in the singleton . Its +/// keepalive renews registered sessions forever (the orphan reaper only collects +/// UNregistered ones), so the site subscription, the bridge actor and the dead circuit's +/// object graph were pinned until process restart. The fix publishes the session and +/// re-checks disposal under one lock, so exactly one side tears the session down. +/// +public class DebugViewConnectDisposeRaceTests : BunitContext +{ + /// + /// A whose start can be held open on demand, so the + /// dispose-during-start window is deterministic rather than timing-dependent. + /// + private sealed class GatedDebugStreamService : DebugStreamService + { + private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _entered = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly List _stopped = new(); + + public GatedDebugStreamService( + CommunicationService comms, IServiceProvider provider, SiteStreamGrpcClientFactory factory) + : base(comms, provider, factory, NullLogger.Instance) + { + } + + public string SessionId => "session-under-test"; + + /// Completes once has been entered. + public Task Entered => _entered.Task; + + /// Session ids passed to , in call order. + public IReadOnlyList Stopped + { + get { lock (_stopped) { return _stopped.ToList(); } } + } + + public void ReleaseStart() => _gate.TrySetResult(); + + public override async Task StartStreamAsync( + int instanceId, Action onEvent, Action onTerminated, CancellationToken ct = default) + { + _entered.TrySetResult(); + await _gate.Task; + return new DebugStreamSession( + SessionId, + new DebugViewSnapshot( + "Inst-1", + Array.Empty(), + Array.Empty(), + DateTimeOffset.UtcNow)); + } + + public override void StopStream(string sessionId) + { + lock (_stopped) { _stopped.Add(sessionId); } + } + } + + private sealed class StubAuthStateProvider : AuthenticationStateProvider + { + private readonly AuthenticationState _state; + public StubAuthStateProvider(AuthenticationState state) => _state = state; + public override Task GetAuthenticationStateAsync() + => Task.FromResult(_state); + } + + private (IRenderedComponent Cut, GatedDebugStreamService Stream) RenderPage() + { + JSInterop.Mode = JSRuntimeMode.Loose; + + var repo = Substitute.For(); + var siteRepo = Substitute.For(); + siteRepo.GetAllSitesAsync().Returns(new List()); + Services.AddSingleton(repo); + Services.AddSingleton(siteRepo); + + var comms = new CommunicationService( + Options.Create(new CommunicationOptions()), + NullLogger.Instance); + Services.AddSingleton(comms); + + // An empty throwaway provider — the gated stub never calls into the base start + // path, so nothing is resolved from it. + var stream = new GatedDebugStreamService( + comms, + new ServiceCollection().BuildServiceProvider(), + new SiteStreamGrpcClientFactory(NullLoggerFactory.Instance)); + Services.AddSingleton(stream); + + var identity = new ClaimsIdentity( + new[] { new Claim(ClaimTypes.Name, "deployer") }, "TestCookie"); + var stubAuth = new StubAuthStateProvider( + new AuthenticationState(new ClaimsPrincipal(identity))); + Services.AddSingleton(stubAuth); + Services.AddScoped(_ => new SiteScopeService(stubAuth)); + + return (Render(), stream); + } + + private static void SetField(DebugViewPage page, string name, object value) => + typeof(DebugViewPage).GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)! + .SetValue(page, value); + + private static object? GetField(DebugViewPage page, string name) => + typeof(DebugViewPage).GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(page); + + /// + /// Starts Connect() on the renderer's dispatcher (where it really runs) and hands + /// back its task without awaiting it, so the test can act inside the start window. + /// + private static async Task StartConnectAsync(IRenderedComponent cut) + { + var connect = typeof(DebugViewPage).GetMethod( + "Connect", BindingFlags.Instance | BindingFlags.NonPublic)!; + Task? task = null; + await cut.InvokeAsync(() => { task = (Task)connect.Invoke(cut.Instance, null)!; }); + return task!; + } + + [Fact] + public async Task Connect_DisposedMidStart_StopsTheJustCreatedSession() + { + var (cut, stream) = RenderPage(); + SetField(cut.Instance, "_selectedSiteId", 1); + SetField(cut.Instance, "_selectedInstanceId", 7); + + var connect = await StartConnectAsync(cut); + await stream.Entered.WaitAsync(TimeSpan.FromSeconds(5)); + + // The user navigates away while the start is still in flight. Dispose runs off the + // renderer's thread, exactly as a circuit teardown can. + cut.Instance.Dispose(); + + stream.ReleaseStart(); + await connect.WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.Equal(new[] { stream.SessionId }, stream.Stopped); + Assert.Null(GetField(cut.Instance, "_session")); + } + + [Fact] + public async Task Connect_ThenDispose_StopsTheSessionExactlyOnce() + { + var (cut, stream) = RenderPage(); + SetField(cut.Instance, "_selectedSiteId", 1); + SetField(cut.Instance, "_selectedInstanceId", 7); + + // Normal path: the start completes while the component is very much alive. + stream.ReleaseStart(); + var connect = await StartConnectAsync(cut); + await connect.WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.Empty(stream.Stopped); + Assert.NotNull(GetField(cut.Instance, "_session")); + + cut.Instance.Dispose(); + Assert.Equal(new[] { stream.SessionId }, stream.Stopped); + + // A second Dispose (bUnit tears the component down again at test end) must not + // stop an already-released session a second time. + cut.Instance.Dispose(); + Assert.Equal(new[] { stream.SessionId }, stream.Stopped); + } + + [Fact] + public async Task Disconnect_AfterConnect_StopsTheSessionOnceAndClearsTheSlot() + { + var (cut, stream) = RenderPage(); + SetField(cut.Instance, "_selectedSiteId", 1); + SetField(cut.Instance, "_selectedInstanceId", 7); + + stream.ReleaseStart(); + var connect = await StartConnectAsync(cut); + await connect.WaitAsync(TimeSpan.FromSeconds(10)); + + var disconnect = typeof(DebugViewPage).GetMethod( + "Disconnect", BindingFlags.Instance | BindingFlags.NonPublic)!; + await cut.InvokeAsync(() => (Task)disconnect.Invoke(cut.Instance, null)!); + + Assert.Equal(new[] { stream.SessionId }, stream.Stopped); + Assert.Null(GetField(cut.Instance, "_session")); + + // Dispose after an explicit disconnect has nothing left to stop. + cut.Instance.Dispose(); + Assert.Equal(new[] { stream.SessionId }, stream.Stopped); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/DebugStreamHubSubscribeRaceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/DebugStreamHubSubscribeRaceTests.cs new file mode 100644 index 00000000..30494038 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/DebugStreamHubSubscribeRaceTests.cs @@ -0,0 +1,179 @@ +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Connections.Features; +using Microsoft.AspNetCore.Http.Features; +using Microsoft.AspNetCore.SignalR; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests; + +/// +/// Regression tests for arch-review 2026-08-19 P6-6 on the hub side. SubscribeInstance +/// awaits StartStreamAsync — up to 30s — and only then stores the session id on the +/// connection. A client that disconnects inside that window ran OnDisconnectedAsync +/// while there was still no session id to release, so the session stayed registered in the +/// singleton and its keepalive renewed it forever (the +/// orphan reaper only collects UNregistered sessions). The fix publishes the id and +/// re-checks connection liveness under the per-connection Items monitor, so exactly +/// one side tears the session down. +/// +public class DebugStreamHubSubscribeRaceTests +{ + private sealed class GatedDebugStreamService : DebugStreamService + { + private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _entered = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly List _stopped = new(); + + public GatedDebugStreamService() + : base( + new CommunicationService(Options.Create(new CommunicationOptions()), NullLogger.Instance), + new ServiceCollection().BuildServiceProvider(), + new SiteStreamGrpcClientFactory(NullLoggerFactory.Instance), + NullLogger.Instance) + { + } + + public string SessionId => "session-under-test"; + + /// Completes once has been entered. + public Task Entered => _entered.Task; + + /// Session ids passed to , in call order. + public IReadOnlyList Stopped + { + get { lock (_stopped) { return _stopped.ToList(); } } + } + + public void ReleaseStart() => _gate.TrySetResult(); + + public override async Task StartStreamAsync( + int instanceId, Action onEvent, Action onTerminated, CancellationToken ct = default) + { + _entered.TrySetResult(); + await _gate.Task; + return new DebugStreamSession( + SessionId, + new DebugViewSnapshot( + "Inst-1", + Array.Empty(), + Array.Empty(), + DateTimeOffset.UtcNow)); + } + + public override void StopStream(string sessionId) + { + lock (_stopped) { _stopped.Add(sessionId); } + } + } + + private sealed class TestHubCallerContext : HubCallerContext + { + private readonly FeatureCollection _features = new(); + private readonly CancellationTokenSource _aborted = new(); + + public TestHubCallerContext(HttpContext http) => + _features.Set(new TestHttpContextFeature { HttpContext = http }); + + public override string ConnectionId => "test-conn"; + public override string? UserIdentifier => null; + public override System.Security.Claims.ClaimsPrincipal? User => null; + public override IDictionary Items { get; } = new Dictionary(); + public override IFeatureCollection Features => _features; + public override CancellationToken ConnectionAborted => _aborted.Token; + public override void Abort() => _aborted.Cancel(); + + private sealed class TestHttpContextFeature : IHttpContextFeature { public HttpContext? HttpContext { get; set; } } + } + + private const string SessionIdKey = "DebugStreamSessionId"; + + private static (DebugStreamHub Hub, TestHubCallerContext Context, GatedDebugStreamService Stream) CreateHub() + { + var repo = Substitute.For(); + repo.GetInstanceByIdAsync(7).Returns(new Instance("Inst-1") { Id = 7, SiteId = 1 }); + + var services = new ServiceCollection(); + services.AddSingleton(repo); + var http = new DefaultHttpContext { RequestServices = services.BuildServiceProvider() }; + + var stream = new GatedDebugStreamService(); + var hub = new DebugStreamHub( + stream, Substitute.For>(), NullLogger.Instance) + { + Clients = Substitute.For() + }; + + var context = new TestHubCallerContext(http); + hub.Context = context; + // No roles / no permitted sites on the connection == system-wide scope, so the + // authorization branch is not what these tests are exercising. + return (hub, context, stream); + } + + [Fact] + public async Task SubscribeInstance_ClientDisconnectsMidStart_StopsTheJustCreatedSession() + { + var (hub, context, stream) = CreateHub(); + + var subscribe = hub.SubscribeInstance(7); + await stream.Entered.WaitAsync(TimeSpan.FromSeconds(5)); + + // The client goes away while the start is still in flight — the disconnect path + // finds nothing registered yet. + await hub.OnDisconnectedAsync(null); + Assert.Empty(stream.Stopped); + + stream.ReleaseStart(); + await subscribe.WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.Equal(new[] { stream.SessionId }, stream.Stopped); + Assert.False(context.Items.ContainsKey(SessionIdKey)); + } + + [Fact] + public async Task SubscribeInstance_ConnectionAbortedMidStart_StopsTheJustCreatedSession() + { + // Same window, but the abort token trips without OnDisconnectedAsync having run yet + // (SignalR aborts the connection before dispatching the disconnect handler). + var (hub, context, stream) = CreateHub(); + + var subscribe = hub.SubscribeInstance(7); + await stream.Entered.WaitAsync(TimeSpan.FromSeconds(5)); + + context.Abort(); + stream.ReleaseStart(); + await subscribe.WaitAsync(TimeSpan.FromSeconds(10)); + + Assert.Equal(new[] { stream.SessionId }, stream.Stopped); + Assert.False(context.Items.ContainsKey(SessionIdKey)); + } + + [Fact] + public async Task SubscribeInstance_ThenDisconnect_StopsTheSessionExactlyOnce() + { + var (hub, context, stream) = CreateHub(); + + stream.ReleaseStart(); + await hub.SubscribeInstance(7); + + Assert.Empty(stream.Stopped); + Assert.Equal(stream.SessionId, context.Items[SessionIdKey]); + + await hub.OnDisconnectedAsync(null); + Assert.Equal(new[] { stream.SessionId }, stream.Stopped); + Assert.False(context.Items.ContainsKey(SessionIdKey)); + + // A second release (an explicit unsubscribe racing the disconnect) is a no-op. + await hub.UnsubscribeInstance(); + Assert.Equal(new[] { stream.SessionId }, stream.Stopped); + } +}