fix(GWC-27): gate AttachInternalEventSubscriber on session readiness
AttachInternalEventSubscriber ran EnsureDistributorCreated / Register / StartPumpIfRequested with no state check, unlike AttachEventSubscriber. A premature attach would start the pump against a not-yet-Ready worker; the pump source throws SessionNotReady, PumpAsync completes every subscriber with that error and latches the distributor, and _eventDistributorStarted is never reset — so the session would reach Ready with permanently dead event streaming. Mirror AttachEventSubscriber's gate: check _state/_workerClient.State under _syncRoot and throw SessionManagerException(SessionNotReady) before the distributor is created, keeping the distributor calls outside the lock. Refs: archreview/2026-07-12/remediation/10-gateway-core.md GWC-27
This commit is contained in:
@@ -548,10 +548,37 @@ public sealed class GatewaySession
|
|||||||
/// <c>MaxEventSubscribersPerSession</c> accounting and out of the single-subscriber
|
/// <c>MaxEventSubscribersPerSession</c> accounting and out of the single-subscriber
|
||||||
/// overflow-fault path, so a slow alarm reconcile can never fault the session — it only
|
/// overflow-fault path, so a slow alarm reconcile can never fault the session — it only
|
||||||
/// disconnects this internal subscriber.
|
/// disconnects this internal subscriber.
|
||||||
|
/// <para>
|
||||||
|
/// Gated on readiness exactly like <see cref="AttachEventSubscriber"/>: attaching
|
||||||
|
/// before the session and its worker are <c>Ready</c> throws
|
||||||
|
/// <see cref="SessionManagerException"/> with
|
||||||
|
/// <see cref="SessionManagerErrorCode.SessionNotReady"/>.
|
||||||
|
/// </para>
|
||||||
/// </remarks>
|
/// </remarks>
|
||||||
/// <returns>The internal subscriber's lease; dispose it to unregister.</returns>
|
/// <returns>The internal subscriber's lease; dispose it to unregister.</returns>
|
||||||
|
/// <exception cref="SessionManagerException">
|
||||||
|
/// The session or its worker client is not <c>Ready</c>.
|
||||||
|
/// </exception>
|
||||||
public IEventSubscriberLease AttachInternalEventSubscriber()
|
public IEventSubscriberLease AttachInternalEventSubscriber()
|
||||||
{
|
{
|
||||||
|
// Readiness gate, mirroring AttachEventSubscriber (GWC-27). It must run BEFORE
|
||||||
|
// EnsureDistributorCreated: a premature attach would construct the distributor and start
|
||||||
|
// its pump against a not-yet-Ready worker, the pump source would throw SessionNotReady,
|
||||||
|
// PumpAsync would complete every subscriber with that error and latch _completed, and
|
||||||
|
// _eventDistributorStarted is never reset — so the session would reach Ready with
|
||||||
|
// permanently dead event streaming, silently, for the rest of its life. Failing loudly
|
||||||
|
// here keeps that state unreachable. The check is under _syncRoot and the distributor
|
||||||
|
// calls stay outside it, matching AttachEventSubscriber's lock discipline.
|
||||||
|
lock (_syncRoot)
|
||||||
|
{
|
||||||
|
if (_state != SessionState.Ready || _workerClient?.State != WorkerClientState.Ready)
|
||||||
|
{
|
||||||
|
throw new SessionManagerException(
|
||||||
|
SessionManagerErrorCode.SessionNotReady,
|
||||||
|
$"Session {SessionId} is not ready for event streaming. Current state is {_state}.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Same sequence StartDashboardMirror uses: create the distributor (claiming the pump
|
// Same sequence StartDashboardMirror uses: create the distributor (claiming the pump
|
||||||
// start if we are first), register the internal subscriber BEFORE the pump starts so a
|
// start if we are first), register the internal subscriber BEFORE the pump starts so a
|
||||||
// subscriber is always present at pump start, then start the pump if requested.
|
// subscriber is always present at pump start, then start the pump if requested.
|
||||||
|
|||||||
@@ -668,6 +668,81 @@ public sealed class GatewaySessionTests
|
|||||||
Assert.Equal(SessionState.Ready, session.State);
|
Assert.Equal(SessionState.Ready, session.State);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// GWC-27: <see cref="GatewaySession.AttachInternalEventSubscriber"/> must refuse to
|
||||||
|
/// attach before the session is Ready. Without the gate the attach would construct and
|
||||||
|
/// start the distributor against a not-yet-Ready worker; the pump source throws
|
||||||
|
/// <c>SessionNotReady</c>, every subscriber is completed with that error, and the
|
||||||
|
/// distributor latches — leaving a session that reaches Ready with permanently dead
|
||||||
|
/// event streaming. The second half of the test is the load-bearing one: after the
|
||||||
|
/// failed attach the session still streams live events, proving the distributor was
|
||||||
|
/// never created or started.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task AttachInternalEventSubscriberBeforeReadyThrowsAndDoesNotPoisonDistributor()
|
||||||
|
{
|
||||||
|
FakeWorkerClient workerClient = new();
|
||||||
|
workerClient.Events.Add(new WorkerEvent
|
||||||
|
{
|
||||||
|
Event = new MxEvent { Family = MxEventFamily.OnDataChange, WorkerSequence = 1, OnDataChange = new OnDataChangeEvent() },
|
||||||
|
});
|
||||||
|
workerClient.Events.Add(new WorkerEvent
|
||||||
|
{
|
||||||
|
Event = new MxEvent { Family = MxEventFamily.OnDataChange, WorkerSequence = 2, OnDataChange = new OnDataChangeEvent() },
|
||||||
|
});
|
||||||
|
|
||||||
|
// Constructed but neither worker-attached nor Ready — the premature-attach case.
|
||||||
|
await using GatewaySession session = CreateSession();
|
||||||
|
|
||||||
|
SessionManagerException exception = Assert.Throws<SessionManagerException>(
|
||||||
|
() => session.AttachInternalEventSubscriber());
|
||||||
|
Assert.Equal(SessionManagerErrorCode.SessionNotReady, exception.ErrorCode);
|
||||||
|
|
||||||
|
// Drive the session to Ready and stream: the failed attach must not have poisoned
|
||||||
|
// (or even created) the distributor, so a normal subscriber still receives events.
|
||||||
|
session.AttachWorkerClient(workerClient);
|
||||||
|
session.MarkReady();
|
||||||
|
|
||||||
|
using IEventSubscriberLease lease = session.AttachEventSubscriber(maxSubscribers: 1);
|
||||||
|
List<MxEvent> received = [];
|
||||||
|
using CancellationTokenSource readCts = new(TimeSpan.FromSeconds(5));
|
||||||
|
await foreach (MxEvent mxEvent in lease.Reader.ReadAllAsync(readCts.Token))
|
||||||
|
{
|
||||||
|
received.Add(mxEvent);
|
||||||
|
if (received.Count == 2)
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Assert.Equal([1UL, 2UL], received.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
|
||||||
|
}
|
||||||
|
|
||||||
|
private static GatewaySession CreateSession()
|
||||||
|
{
|
||||||
|
return new GatewaySession(
|
||||||
|
sessionId: "session-test-internal-attach",
|
||||||
|
backendName: "mxaccess",
|
||||||
|
pipeName: "mxaccess-gateway-1-session-test-internal-attach",
|
||||||
|
nonce: "nonce",
|
||||||
|
clientIdentity: "client-1",
|
||||||
|
ownerKeyId: null,
|
||||||
|
clientSessionName: "test-session",
|
||||||
|
clientCorrelationId: "client-correlation-1",
|
||||||
|
commandTimeout: TimeSpan.FromSeconds(5),
|
||||||
|
startupTimeout: TimeSpan.FromSeconds(5),
|
||||||
|
shutdownTimeout: TimeSpan.FromSeconds(5),
|
||||||
|
leaseDuration: TimeSpan.FromMinutes(30),
|
||||||
|
openedAt: DateTimeOffset.UtcNow,
|
||||||
|
eventStreaming: new SessionEventStreaming(
|
||||||
|
new MxAccessGrpcMapper(),
|
||||||
|
new EventOptions { QueueCapacity = 8 },
|
||||||
|
NullLogger<SessionEventDistributor>.Instance,
|
||||||
|
TimeProvider.System,
|
||||||
|
new GatewayMetrics()));
|
||||||
|
}
|
||||||
|
|
||||||
private static GatewaySession CreateReadySessionWithDetachGrace(
|
private static GatewaySession CreateReadySessionWithDetachGrace(
|
||||||
IWorkerClient workerClient,
|
IWorkerClient workerClient,
|
||||||
TimeProvider timeProvider,
|
TimeProvider timeProvider,
|
||||||
@@ -855,6 +930,9 @@ public sealed class GatewaySessionTests
|
|||||||
/// <summary>Gets the count of dispose invocations.</summary>
|
/// <summary>Gets the count of dispose invocations.</summary>
|
||||||
public int DisposeCount { get; private set; }
|
public int DisposeCount { get; private set; }
|
||||||
|
|
||||||
|
/// <summary>Events <see cref="ReadEventsAsync"/> yields, in order, before completing. Empty by default.</summary>
|
||||||
|
public List<WorkerEvent> Events { get; } = [];
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||||
|
|
||||||
@@ -869,7 +947,11 @@ public sealed class GatewaySessionTests
|
|||||||
[EnumeratorCancellation] CancellationToken cancellationToken)
|
[EnumeratorCancellation] CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
await Task.CompletedTask.ConfigureAwait(false);
|
await Task.CompletedTask.ConfigureAwait(false);
|
||||||
yield break;
|
foreach (WorkerEvent workerEvent in Events)
|
||||||
|
{
|
||||||
|
cancellationToken.ThrowIfCancellationRequested();
|
||||||
|
yield return workerEvent;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
|
|||||||
Reference in New Issue
Block a user