Resolve Server-031..032 (re-triaged) + Server-038..043
Server-031: re-triaged. The recommended gateway-side "skip-while-command-in-flight" guard is already in place at WorkerClient.HeartbeatLoopAsync via WorkerClientOptions.HeartbeatStuckCeiling (default 75s = 5× HeartbeatGrace). Two regression tests pin the behaviour. Recommendation #1 (decouple worker-side _writeLock) is a Worker-module concern (Worker-017 / Worker-023) and out of scope here. Server-032: re-triaged. Recommendation #2 (rich diagnostic) is already in EnqueueWorkerEventAsync, with #3 (overflow grace) absorbed by the TryWrite → WriteAsync-with-timeout fall-through. Test EnqueueWorkerEvent_WhenChannelFullPastTimeout_FaultsWithRichDiagnostic pins the diagnostic string. Recommendation #1 (prose contract in gateway.md / docs) is deferred — outside this pass's edit scope. Server-038 (Security): EventsHub.SubscribeSession's missing per-session ACL is documented with a TODO(per-session-acl) and a <remarks> block explaining the v1 acceptance (any dashboard role can subscribe to any session — non-secret metadata, redacted value logging). The per-session ACL design lands in a follow-up once a session-scoped role exists. Server-039 (Error handling): HubTokenService.Validate now rejects a deserialized payload where both Name and NameIdentifier are null/empty. New test file HubTokenServiceTests.cs covers the regression and five sanity cases. TDD confirmed. Server-040 (Conventions): MapGroupsToRoles gains a precedence comment explaining "full literal match first, leading-RDN fallback; OrdinalIgnoreCase via DashboardOptions.GroupToRole". Documentation-only. Server-041 (Design adherence): EventStreamService.ProduceEventsAsync wraps the broadcaster.Publish call in try/catch (Exception). The producer loop and gRPC stream are no longer at the mercy of the broadcaster's never-throw discipline. New regression test StreamEventsAsync_WhenDashboardBroadcasterThrows_StillYieldsEventsAndDoesNotFaultSession. Server-042 (Performance): DashboardSnapshotPublisher.ExecuteAsync now mirrors AlarmsHubPublisher's reconnect loop — wraps the await foreach in a while-not-cancelled, catches general exceptions, and Task.Delays 5s before retrying. An internal ctor accepts a shorter delay for the test. New test file DashboardSnapshotPublisherTests.cs covers the throw-then-yield reconnect path and the normal-completion case. Server-043 (Documentation): HubTokenService class XML doc gains a <remarks> describing the singleton lifetime, the two consumer scopes (DashboardHubConnectionFactory scoped, HubTokenAuthenticationHandler transient), and the thread-safety contract. Verification: dotnet build src/ZB.MOM.WW.MxGateway.slnx clean (0 warnings / 0 errors); src/ZB.MOM.WW.MxGateway.Tests 486/486 passing. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
@@ -9,6 +9,7 @@ using ZB.MOM.WW.MxGateway.Server.Grpc;
|
||||
using ZB.MOM.WW.MxGateway.Server.Metrics;
|
||||
using ZB.MOM.WW.MxGateway.Server.Sessions;
|
||||
using ZB.MOM.WW.MxGateway.Server.Workers;
|
||||
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Grpc;
|
||||
|
||||
@@ -260,11 +261,81 @@ public sealed class EventStreamServiceTests
|
||||
Assert.Equal(1, metrics.GetSnapshot().Faults);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Tests-026 regression: <see cref="EventStreamService.StreamEventsAsync"/>
|
||||
/// must mirror every yielded event to the
|
||||
/// <see cref="ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs.IDashboardEventBroadcaster"/>
|
||||
/// seam (the only path that fans events out to dashboard SignalR clients).
|
||||
/// A regression that silently dropped the <c>Publish</c> call — e.g. an
|
||||
/// <c>if</c> accidentally added around it, or the broadcaster ctor
|
||||
/// parameter being removed — would have produced no failing test before
|
||||
/// this fixture existed. The recording fake captures every call and we
|
||||
/// assert one publish per yielded event, with the correct session id and
|
||||
/// preserved <c>WorkerSequence</c>.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task StreamEventsAsync_PublishesEachEventToDashboardBroadcaster()
|
||||
{
|
||||
FakeWorkerClient workerClient = new();
|
||||
GatewaySession session = CreateReadySession(workerClient);
|
||||
RecordingDashboardEventBroadcaster recordingBroadcaster = new();
|
||||
EventStreamService service = CreateService(
|
||||
new FakeSessionManager(session),
|
||||
dashboardEventBroadcaster: recordingBroadcaster);
|
||||
workerClient.Events.Add(CreateWorkerEvent(sequence: 7, MxEventFamily.OnDataChange));
|
||||
workerClient.Events.Add(CreateWorkerEvent(sequence: 8, MxEventFamily.OnWriteComplete));
|
||||
workerClient.CompleteAfterConfiguredEvents = true;
|
||||
|
||||
List<MxEvent> events = await CollectEventsAsync(service, session.SessionId);
|
||||
|
||||
Assert.Equal([7UL, 8UL], events.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
|
||||
IReadOnlyList<DashboardEventCapture> captures = recordingBroadcaster.Captures;
|
||||
Assert.Equal(2, captures.Count);
|
||||
Assert.All(captures, capture => Assert.Equal(session.SessionId, capture.SessionId));
|
||||
Assert.Equal([7UL, 8UL], captures.Select(capture => capture.MxEvent.WorkerSequence).ToArray());
|
||||
Assert.Equal(MxEventFamily.OnDataChange, captures[0].MxEvent.Family);
|
||||
Assert.Equal(MxEventFamily.OnWriteComplete, captures[1].MxEvent.Family);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Server-041 regression: <see cref="EventStreamService"/> must not
|
||||
/// abort the gRPC stream when the dashboard broadcaster throws.
|
||||
/// <c>IDashboardEventBroadcaster.Publish</c> is documented as
|
||||
/// best-effort and never-throw, but the gRPC consumer cannot rely on
|
||||
/// implementation discipline alone — the seam itself swallows the
|
||||
/// fault and logs at debug, mirroring the broadcaster's own
|
||||
/// continuation handler. Without the wrap, the producer loop would
|
||||
/// surface the exception and the client would see a faulted stream
|
||||
/// for a dashboard-mirror failure.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task StreamEventsAsync_WhenDashboardBroadcasterThrows_StillYieldsEventsAndDoesNotFaultSession()
|
||||
{
|
||||
FakeWorkerClient workerClient = new();
|
||||
GatewaySession session = CreateReadySession(workerClient);
|
||||
using GatewayMetrics metrics = new();
|
||||
ThrowingDashboardEventBroadcaster throwingBroadcaster = new();
|
||||
EventStreamService service = CreateService(
|
||||
new FakeSessionManager(session),
|
||||
metrics,
|
||||
dashboardEventBroadcaster: throwingBroadcaster);
|
||||
workerClient.Events.Add(CreateWorkerEvent(sequence: 1, MxEventFamily.OnDataChange));
|
||||
workerClient.Events.Add(CreateWorkerEvent(sequence: 2, MxEventFamily.OnDataChange));
|
||||
workerClient.CompleteAfterConfiguredEvents = true;
|
||||
|
||||
List<MxEvent> events = await CollectEventsAsync(service, session.SessionId);
|
||||
|
||||
Assert.Equal([1UL, 2UL], events.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
|
||||
Assert.Equal(2, throwingBroadcaster.PublishAttempts);
|
||||
Assert.NotEqual(SessionState.Faulted, session.State);
|
||||
}
|
||||
|
||||
private static EventStreamService CreateService(
|
||||
FakeSessionManager sessionManager,
|
||||
GatewayMetrics? metrics = null,
|
||||
int queueCapacity = 8,
|
||||
EventBackpressurePolicy backpressurePolicy = EventBackpressurePolicy.FailFast)
|
||||
EventBackpressurePolicy backpressurePolicy = EventBackpressurePolicy.FailFast,
|
||||
ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs.IDashboardEventBroadcaster? dashboardEventBroadcaster = null)
|
||||
{
|
||||
return new EventStreamService(
|
||||
sessionManager,
|
||||
@@ -278,14 +349,19 @@ public sealed class EventStreamServiceTests
|
||||
}),
|
||||
new MxAccessGrpcMapper(),
|
||||
metrics ?? new GatewayMetrics(),
|
||||
NullDashboardEventBroadcaster.Instance,
|
||||
dashboardEventBroadcaster ?? NullDashboardEventBroadcaster.Instance,
|
||||
NullLogger<EventStreamService>.Instance);
|
||||
}
|
||||
|
||||
private sealed class NullDashboardEventBroadcaster : ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs.IDashboardEventBroadcaster
|
||||
private sealed class ThrowingDashboardEventBroadcaster : ZB.MOM.WW.MxGateway.Server.Dashboard.Hubs.IDashboardEventBroadcaster
|
||||
{
|
||||
public static readonly NullDashboardEventBroadcaster Instance = new();
|
||||
public void Publish(string sessionId, MxEvent mxEvent) { }
|
||||
public int PublishAttempts { get; private set; }
|
||||
|
||||
public void Publish(string sessionId, MxEvent mxEvent)
|
||||
{
|
||||
PublishAttempts++;
|
||||
throw new InvalidOperationException("simulated dashboard broadcaster failure");
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task<List<MxEvent>> CollectEventsAsync(
|
||||
|
||||
Reference in New Issue
Block a user