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); + } +}