Joseph Doherty
561582747e
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.
289 lines
13 KiB
C#
289 lines
13 KiB
C#
using Microsoft.AspNetCore.SignalR;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
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.Security;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.ManagementService;
|
|
|
|
/// <summary>
|
|
/// SignalR hub for real-time debug stream subscriptions.
|
|
/// External consumers (CLI) connect via WebSocket, authenticate with Basic Auth,
|
|
/// and receive streaming attribute value and alarm state changes.
|
|
/// </summary>
|
|
public class DebugStreamHub : Hub
|
|
{
|
|
private const string SessionIdKey = "DebugStreamSessionId";
|
|
private const string RolesKey = "DebugStreamRoles";
|
|
private const string PermittedSiteIdsKey = "DebugStreamPermittedSiteIds";
|
|
private const string DisconnectedKey = "DebugStreamDisconnected";
|
|
|
|
/// <summary>
|
|
/// Pure site-scope authorization check for a debug-stream subscription.
|
|
/// Returns true when the caller may subscribe to a debug stream for an instance
|
|
/// belonging to <paramref name="instanceSiteId"/>.
|
|
/// Administrator role, or an empty <paramref name="permittedSiteIds"/> (system-wide
|
|
/// Deployer), grants access to any site; otherwise the instance's site must be
|
|
/// in the permitted set.
|
|
/// </summary>
|
|
/// <param name="roles">Roles held by the connected user.</param>
|
|
/// <param name="permittedSiteIds">Site ids the user is scoped to; empty means system-wide.</param>
|
|
/// <param name="instanceSiteId">Site id of the instance being subscribed to.</param>
|
|
/// <returns><see langword="true"/> if the user may subscribe to the instance; otherwise <see langword="false"/>.</returns>
|
|
public static bool IsInstanceAccessAllowed(
|
|
IReadOnlyCollection<string> roles,
|
|
IReadOnlyCollection<string> permittedSiteIds,
|
|
int instanceSiteId)
|
|
{
|
|
if (roles.Contains(Roles.Administrator, StringComparer.OrdinalIgnoreCase)) return true;
|
|
if (permittedSiteIds.Count == 0) return true; // system-wide deployment
|
|
return permittedSiteIds.Contains(instanceSiteId.ToString());
|
|
}
|
|
|
|
private readonly DebugStreamService _debugStreamService;
|
|
private readonly IHubContext<DebugStreamHub> _hubContext;
|
|
private readonly ILogger<DebugStreamHub> _logger;
|
|
|
|
/// <summary>
|
|
/// Initializes the hub with its required services.
|
|
/// </summary>
|
|
/// <param name="debugStreamService">Service that manages debug stream subscriptions.</param>
|
|
/// <param name="hubContext">SignalR hub context for pushing events to clients.</param>
|
|
/// <param name="logger">Logger for this hub.</param>
|
|
public DebugStreamHub(
|
|
DebugStreamService debugStreamService,
|
|
IHubContext<DebugStreamHub> hubContext,
|
|
ILogger<DebugStreamHub> logger)
|
|
{
|
|
_debugStreamService = debugStreamService;
|
|
_hubContext = hubContext;
|
|
_logger = logger;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public override async Task OnConnectedAsync()
|
|
{
|
|
var httpContext = Context.GetHttpContext();
|
|
if (httpContext == null)
|
|
{
|
|
_logger.LogWarning("DebugStreamHub connection rejected: no HTTP context");
|
|
Context.Abort();
|
|
return;
|
|
}
|
|
|
|
// Authenticate via the SAME shared Basic→LDAP flow as POST /management and the
|
|
// audit REST (arch-review R2 N1): ManagementAuthenticator applies the DisableLogin
|
|
// bypass, the LoginThrottle lockout check BEFORE the bind, and RecordFailure/
|
|
// RecordSuccess bookkeeping — so hub connection attempts can no longer be used as
|
|
// an unthrottled LDAP-bind oracle, and hub failures feed the shared lockout.
|
|
var outcome = await ManagementAuthenticator.AuthenticateAsync(httpContext);
|
|
if (outcome.User is null)
|
|
{
|
|
_logger.LogWarning("DebugStreamHub connection rejected: authentication failed or throttled");
|
|
Context.Abort();
|
|
return;
|
|
}
|
|
var user = outcome.User;
|
|
|
|
// Role check — Deployer role required (unchanged policy).
|
|
if (!user.Roles.Contains(Roles.Deployer))
|
|
{
|
|
_logger.LogWarning("DebugStreamHub connection rejected: {Username} lacks Deployer role", user.Username);
|
|
Context.Abort();
|
|
return;
|
|
}
|
|
|
|
// Persist the resolved identity on the connection so per-instance site-scope
|
|
// enforcement can be applied to SubscribeInstance calls. PermittedSiteIds is the
|
|
// authenticator's normalized form (empty == system-wide), same as every other surface.
|
|
Context.Items[RolesKey] = user.Roles;
|
|
Context.Items[PermittedSiteIdsKey] = user.PermittedSiteIds;
|
|
|
|
_logger.LogInformation("DebugStreamHub connection established for {Username}", user.Username);
|
|
await base.OnConnectedAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Subscribes to a debug stream for the specified instance.
|
|
/// Sends the initial snapshot immediately, then streams incremental changes.
|
|
/// </summary>
|
|
/// <param name="instanceId">Database id of the instance to subscribe to.</param>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
public async Task SubscribeInstance(int instanceId)
|
|
{
|
|
// Stop any existing subscription for this connection
|
|
await UnsubscribeInstance();
|
|
|
|
var connectionId = Context.ConnectionId;
|
|
|
|
// Per-instance site-scope enforcement: a site-scoped Deployment user must not
|
|
// be able to stream an instance belonging to a site outside their scope.
|
|
var httpContext = Context.GetHttpContext();
|
|
if (httpContext == null)
|
|
{
|
|
_logger.LogWarning("DebugStreamHub: {ConnectionId} subscribe rejected — no HTTP context", connectionId);
|
|
await Clients.Caller.SendAsync("OnStreamTerminated", "Authorization context unavailable.");
|
|
return;
|
|
}
|
|
|
|
var roles = Context.Items.TryGetValue(RolesKey, out var rolesObj) && rolesObj is string[] r
|
|
? r : Array.Empty<string>();
|
|
var permittedSiteIds = Context.Items.TryGetValue(PermittedSiteIdsKey, out var sitesObj) && sitesObj is string[] s
|
|
? s : Array.Empty<string>();
|
|
|
|
var instanceRepo = httpContext.RequestServices.GetRequiredService<ITemplateEngineRepository>();
|
|
var instance = await instanceRepo.GetInstanceByIdAsync(instanceId);
|
|
if (instance == null)
|
|
{
|
|
_logger.LogWarning("DebugStreamHub: {ConnectionId} subscribe rejected — instance {InstanceId} not found",
|
|
connectionId, instanceId);
|
|
await Clients.Caller.SendAsync("OnStreamTerminated", $"Instance {instanceId} not found.");
|
|
return;
|
|
}
|
|
|
|
if (!IsInstanceAccessAllowed(roles, permittedSiteIds, instance.SiteId))
|
|
{
|
|
_logger.LogWarning(
|
|
"DebugStreamHub: {ConnectionId} subscribe to instance {InstanceId} denied — site {SiteId} outside permitted scope",
|
|
connectionId, instanceId, instance.SiteId);
|
|
await Clients.Caller.SendAsync("OnStreamTerminated",
|
|
$"Access denied: instance {instanceId} belongs to a site outside your Deployment scope.");
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
// Use IHubContext for callbacks — the hub instance is transient (disposed after method returns),
|
|
// but IHubContext is a singleton that remains valid for the lifetime of the connection.
|
|
var hubClients = _hubContext.Clients;
|
|
// Capture the logger into a local: the callbacks outlive this (transient)
|
|
// hub instance, and ILogger is a singleton, so this is safe to hold.
|
|
var logger = _logger;
|
|
|
|
var session = await _debugStreamService.StartStreamAsync(
|
|
instanceId,
|
|
onEvent: evt =>
|
|
{
|
|
var send = evt switch
|
|
{
|
|
AttributeValueChanged changed =>
|
|
hubClients.Client(connectionId).SendAsync("OnAttributeChanged", changed),
|
|
AlarmStateChanged changed =>
|
|
hubClients.Client(connectionId).SendAsync("OnAlarmChanged", changed),
|
|
DebugViewSnapshot snapshot =>
|
|
hubClients.Client(connectionId).SendAsync("OnSnapshot", snapshot),
|
|
_ => Task.CompletedTask
|
|
};
|
|
// Fire-and-forget (deliberate — a slow/dead client must not block the
|
|
// stream), but log the push failure rather than swallowing it silently.
|
|
_ = send.ContinueWith(
|
|
t => logger.LogWarning(t.Exception?.GetBaseException(),
|
|
"Debug stream push to connection {ConnectionId} failed", connectionId),
|
|
CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);
|
|
},
|
|
onTerminated: () =>
|
|
{
|
|
_ = hubClients.Client(connectionId).SendAsync("OnStreamTerminated", "Site disconnected")
|
|
.ContinueWith(
|
|
t => logger.LogWarning(t.Exception?.GetBaseException(),
|
|
"Debug stream termination push to connection {ConnectionId} failed", connectionId),
|
|
CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);
|
|
});
|
|
|
|
// 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);
|
|
|
|
_logger.LogInformation("DebugStreamHub: {ConnectionId} subscribed to instance {InstanceId}",
|
|
connectionId, instanceId);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "DebugStreamHub: Failed to subscribe {ConnectionId} to instance {InstanceId}",
|
|
connectionId, instanceId);
|
|
await Clients.Caller.SendAsync("OnStreamTerminated", ex.Message);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Unsubscribes from the current debug stream.
|
|
/// </summary>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
public Task UnsubscribeInstance()
|
|
{
|
|
var sessionId = TakeSessionId();
|
|
if (sessionId is not null)
|
|
{
|
|
_debugStreamService.StopStream(sessionId);
|
|
_logger.LogInformation("DebugStreamHub: {ConnectionId} unsubscribed", Context.ConnectionId);
|
|
}
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Atomically removes and returns the session id registered for this connection, so a
|
|
/// release path and the late publish in <see cref="SubscribeInstance"/> cannot both
|
|
/// claim (or both miss) the same session — see P6-6 there.
|
|
/// </summary>
|
|
/// <returns>The session id that was registered, or <see langword="null"/> if none was.</returns>
|
|
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;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
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);
|
|
}
|
|
}
|