fix(archreview): gateway core P0 remediation (GWC-01/02/03, TST-02, TST-12)

Interlocking changes across the gateway server (shared GatewaySession.cs /
SessionManager.cs), committed together:

- GWC-01 (Critical): alarm monitor now attaches as an internal
  (non-counted) distributor subscriber instead of a second raw drain of the
  single worker event channel; WorkerClient._events -> SingleReader with a
  claimed-once guard so a future dual-consumer regression throws loudly.
- GWC-02 (High): faulted sessions are swept in CloseExpiredLeasesAsync
  (IsFaultedReapable + FaultedReason); new FaultedGraceSeconds (default 0).
- GWC-03 (High): configurable MaxSparseArrayLength (default 1_000_000)
  enforced before allocation.
- TST-02 (High, security): StreamEvents attach now enforces the opening key
  id -> PermissionDenied on owner mismatch.
- TST-12 (Medium): CLAUDE.md retention-defaults sentence corrected.

Verified: NonWindows build clean; targeted tests 135/135 on macOS, plus
WorkerClientTests 18/18 on the Windows host.
This commit is contained in:
Joseph Doherty
2026-07-09 05:51:57 -04:00
parent 31eec41456
commit 20392cf246
30 changed files with 632 additions and 48 deletions
+3 -1
View File
@@ -76,7 +76,7 @@ powershell -ExecutionPolicy Bypass -File scripts/run-client-e2e-tests.ps1
- **Style guides** in `docs/style-guides/` are authoritative. Follow `CSharpStyleGuide.md` for gateway/worker/.NET-client code: file-scoped namespaces, `sealed` by default, `Async` suffix on Task-returning methods, MXAccess-aligned names (`MxStatusProxy`, `ServerHandle`, `ItemHandle`, `HResult`).
- **MXAccess parity is the contract.** Don't "fix" surprising MXAccess behavior (e.g., `WriteSecured` failing before a value-bearing NMX body, distinct `OperationComplete` semantics, invalid-handle exceptions) unless the client explicitly opts into a non-parity mode. The installed MXAccess COM component is the baseline.
- **Don't synthesize events.** The gateway forwards only events the worker emits; it never invents `OperationComplete` from write completion or command replies.
- **One worker per session** (invariant). Multi-subscriber event fan-out and reconnect-with-replay have shipped and are config-gated: `AllowMultipleEventSubscribers` (default `false`) enables fan-out up to `MaxEventSubscribersPerSession` (default `8`); `DetachGraceSeconds` (default `30`) retains a session after its last subscriber drops so clients can reconnect; `ReplayBufferCapacity` / `ReplayRetentionSeconds` control how much event history the replay ring keeps. Default config preserves the original single-subscriber, no-retention behavior. See `docs/DesignDecisions.md` and `docs/Sessions.md`.
- **One worker per session** (invariant). Multi-subscriber event fan-out and reconnect-with-replay have shipped and are config-gated: `AllowMultipleEventSubscribers` (default `false`) enables fan-out up to `MaxEventSubscribersPerSession` (default `8`); `DetachGraceSeconds` (default `30`) retains a session after its last subscriber drops so clients can reconnect; `ReplayBufferCapacity` / `ReplayRetentionSeconds` control how much event history the replay ring keeps. Default config is single-subscriber (`AllowMultipleEventSubscribers` off), but detach-grace and replay retention are **on** by default (`DetachGraceSeconds=30`, `ReplayBufferCapacity=1024`, `ReplayRetentionSeconds=300`): a detached session is retained for 30 s and recent events are buffered for reconnect. See `docs/DesignDecisions.md` and `docs/Sessions.md`.
- **Gateway restart does not reattach orphan workers.** The first version terminates orphaned workers on startup; do not design code paths that assume reattachment.
- **No Blazor UI component libraries.** Dashboard uses local Bootstrap CSS/JS only — do not introduce MudBlazor, Radzen, FluentUI, etc.
- **Don't log secrets or full tag values by default.** API keys, passwords, `WriteSecured` payloads, and `AuthenticateUser` credentials must never reach logs. Value logging is opt-in and redacted.
@@ -121,6 +121,8 @@ External analysis sources referenced by design docs:
Gateway gRPC clients authenticate with an API key in metadata: `authorization: Bearer mxgw_<key-id>_<secret>`. Keys are stored hashed (with a peppered SHA) in a gateway-owned SQLite DB (default `C:\ProgramData\MxGateway\gateway-auth.db`). Scopes (`session`, `invoke`, `event`, `metadata`, `admin`) gate specific RPCs; missing → `Unauthenticated`, insufficient → `PermissionDenied`. The `apikey` subcommand on the server exe manages keys; see `src/ZB.MOM.WW.MxGateway.Server/Security/Authentication/`.
Session event streaming is **owner-scoped**: the API key that opened a session is recorded on the session, and every `StreamEvents` attach/reattach is rejected with `PermissionDenied` unless the caller's key id matches the owner. Possessing the `event` scope and knowing a session id is not sufficient — this closes the reconnect/fan-out trust boundary (detach-grace and replay retention are on by default) so an `event`-scoped key cannot attach to another key's retained session.
Dashboard auth is LDAP-backed (separate from the gRPC API-key model). `/login` binds against `MxGateway:Ldap` and maps the user's LDAP groups to `Admin` or `Viewer` via `MxGateway:Dashboard:GroupToRole`, then issues an HTTP-only secure `__Host-MxGatewayDashboard` cookie. SignalR hubs at `/hubs/{snapshot,alarms,events}` accept either the cookie or a 30-minute bearer minted at `/hubs/token`. `Dashboard:AllowAnonymousLocalhost` bypasses auth on loopback when enabled. `Dashboard:DisableLogin` (default `false`) auto-authenticates every dashboard request — including remote browsers — as `Dashboard:AutoLoginUser` (default `multi-role`) with both Admin and Viewer roles; dev/test only, never enable in production.
## Process / Platform Notes
+11
View File
@@ -85,6 +85,17 @@ delivery. If the requested position precedes the oldest retained event, a
is atomic (no gap, no duplicate). See [Sessions](./Sessions.md) for the full
reconnect and replay protocol.
Decision: event-stream attach is bound to the opening API key. Because the
detach-grace and replay-retention windows are on by default, the reconnect
surface is a trust boundary — a retained session outlives the stream that opened
it. The session records the opening key id (`GatewaySession.OwnerKeyId`) and
every `StreamEvents` attach/reattach is rejected with `PermissionDenied` unless
the caller's key id matches the owner. Gating on the `event` scope alone was
rejected: it would let any `event`-scoped key that learned a session id attach to
another key's retained session and receive its replayed and live data. The check
runs before any subscriber is attached, so a foreign key never touches the
replay ring. Admin-scope override is deferred.
## Event Subscribers
Multi-subscriber fan-out for data-side `StreamEvents` is shipped and
+5 -1
View File
@@ -38,6 +38,7 @@ paths, timeouts, queue sizes, enum values, or protocol values are invalid.
"DefaultLeaseSeconds": 1800,
"LeaseSweepIntervalSeconds": 30,
"DetachGraceSeconds": 30,
"FaultedGraceSeconds": 0,
"AllowMultipleEventSubscribers": false,
"MaxEventSubscribersPerSession": 8,
"WorkerReadyWaitTimeoutMs": 0
@@ -46,7 +47,8 @@ paths, timeouts, queue sizes, enum values, or protocol values are invalid.
"QueueCapacity": 10000,
"BackpressurePolicy": "FailFast",
"ReplayBufferCapacity": 1024,
"ReplayRetentionSeconds": 300
"ReplayRetentionSeconds": 300,
"MaxSparseArrayLength": 1000000
},
"Dashboard": {
"Enabled": true,
@@ -129,6 +131,7 @@ to avoid accidental large allocations from malformed or oversized frames.
| `MxGateway:Sessions:DefaultLeaseSeconds` | `1800` | Initial session lease and refresh duration. Unary client activity extends the lease by this duration. |
| `MxGateway:Sessions:LeaseSweepIntervalSeconds` | `30` | Hosted monitor interval for closing expired leases. Active event-stream subscribers keep a session from expiring while the stream remains attached. |
| `MxGateway:Sessions:DetachGraceSeconds` | `30` | Detach-grace retention window. When positive, a session whose last external (gRPC) event-stream subscriber drops is retained in `Ready` for this many seconds so a client can reconnect; if no external subscriber re-attaches within the window, the lease monitor closes it with `detach-grace-expired`. The internal dashboard mirror does not count as an external subscriber, so a dashboard-only session still enters detach-grace. `0` disables retention and reverts to closing only on normal lease expiry. Must be zero or greater. Reconnect/replay itself is implemented separately (Task 12); this option controls retention and expiry only. The effective close happens within the next sweep cycle after the window elapses — up to `LeaseSweepIntervalSeconds` after expiry. Operators wanting a firm minimum retention bound should set `DetachGraceSeconds` greater than `LeaseSweepIntervalSeconds`. |
| `MxGateway:Sessions:FaultedGraceSeconds` | `0` | Grace window before the lease monitor reaps a faulted session (killing its worker and freeing the slot). A faulted session is permanently unusable yet otherwise pins a session slot and a live x86 worker until its normal lease expires. `0` (the default) reaps it on the next sweep cycle, bounding blast radius; a positive value keeps the faulted session observable via `GetSessionStatus` for that window before it is reclaimed, with the close reason `faulted-reaped`. Must be zero or greater. |
| `MxGateway:Sessions:AllowMultipleEventSubscribers` | `false` | Controls whether multiple `StreamEvents` subscribers may attach to one session. When `false` the session refuses a second subscriber with `AlreadyExists`. Set to `true` to enable fan-out via the `SessionEventDistributor`. |
| `MxGateway:Sessions:MaxEventSubscribersPerSession` | `8` | Maximum number of concurrent `StreamEvents` subscribers per session when `AllowMultipleEventSubscribers` is `true`. Effectively 1 when `AllowMultipleEventSubscribers` is `false`. Must be greater than zero. |
| `MxGateway:Sessions:WorkerReadyWaitTimeoutMs` | `0` | Bounded time, in milliseconds, the gateway will wait for a worker to reach `Ready` when the session is already `Ready` but the worker state has transiently diverged (e.g. `Handshaking` after a heartbeat blip). Applies only to transient worker states; terminal states (`Faulted`/`Closing`/`Closed`/no worker) fail fast immediately regardless of this setting. `0` (the default) disables the wait and preserves the original fail-fast behavior. Must be greater than or equal to zero. |
@@ -143,6 +146,7 @@ All numeric session options must be greater than zero.
| `MxGateway:Events:BackpressurePolicy` | `FailFast` | Per-subscriber event backpressure behavior when a subscriber's bounded event channel overflows. Overflow is isolated to the offending subscriber: it is always disconnected with an `EventQueueOverflow` fault while the session pump and other subscribers keep running. `FailFast` additionally faults the whole session only in the legacy single-subscriber case (the current default mode); with multiple subscribers it degrades to a per-subscriber disconnect so one slow consumer never faults a shared session. `DisconnectSubscriber` disconnects only the slow subscriber in all cases. |
| `MxGateway:Events:ReplayBufferCapacity` | `1024` | Maximum number of events retained per session in the replay ring buffer, used to re-deliver events a returning subscriber missed (reconnect/reattach). The oldest retained event is evicted once this count is exceeded. `0` disables replay retention. |
| `MxGateway:Events:ReplayRetentionSeconds` | `300` | Maximum age, in seconds, of an event retained in the replay ring buffer. Entries older than this are evicted regardless of capacity. `0` disables age-based eviction. |
| `MxGateway:Events:MaxSparseArrayLength` | `1000000` | Maximum `total_length` a sparse-array write (`MxSparseArray`) may declare. A write above this cap is rejected with `InvalidArgument` before the full array is materialized, guarding against a single write forcing a multi-GB allocation. Must be between `1` and `Array.MaxLength`. |
`QueueCapacity` must be greater than zero; it bounds each per-subscriber event
channel fed by the session's single event pump. A slow subscriber overflows only
+9 -1
View File
@@ -72,7 +72,7 @@ private void EnsureSessionCapacity()
}
```
`SessionManager` also defines four close-reason constants — `DefaultCloseReason` (`"client-close"`), `GatewayShutdownReason` (`"gateway-shutdown"`), `LeaseExpiredReason` (`"lease-expired"`), and `DetachGraceExpiredReason` (`"detach-grace-expired"`) — so that the metrics and worker shutdown paths agree on a fixed vocabulary.
`SessionManager` also defines five close-reason constants — `DefaultCloseReason` (`"client-close"`), `GatewayShutdownReason` (`"gateway-shutdown"`), `LeaseExpiredReason` (`"lease-expired"`), `FaultedReason` (`"faulted-reaped"`), and `DetachGraceExpiredReason` (`"detach-grace-expired"`) — so that the metrics and worker shutdown paths agree on a fixed vocabulary.
### SessionRegistry (ISessionRegistry)
@@ -197,6 +197,8 @@ Event streaming uses `AttachEventSubscriber` which returns a disposable lease. W
`FailFast` event backpressure faults the whole session only in single-subscriber mode; in multi-subscriber mode it degrades to a per-subscriber disconnect so one slow consumer never faults a session shared by others. The session passes its mode to the `SessionEventDistributor` at construction, so this decision is made on the fixed mode rather than a live subscriber-count snapshot.
The single worker event channel has exactly one direct reader: the `SessionEventDistributor` pump (`MapWorkerEventsAsync`). Both gateway-owned internal consumers — the dashboard mirror and the central alarm monitor — attach as distributor subscribers rather than draining the worker channel themselves. `GatewaySession.AttachInternalEventSubscriber` mirrors the dashboard-mirror lease (`isInternal: true`): the alarm monitor's `SessionManager.ReadAlarmEventsAsync` registers one so it consumes the same mapped `MxEvent`s the pump fans to every subscriber, without counting against `MaxEventSubscribersPerSession` and without a slow reconcile faulting the session. This is what keeps the alarm feed and the dashboard from splitting the stream between two raw drains (which would silently lose Acknowledge and provider-mode transitions); the worker channel is single-reader and a second `WorkerClient.ReadEventsAsync` consumer throws so a regression fails loudly.
Sessions open with `MxGateway:Sessions:DefaultLeaseSeconds` (default 1800) added to the open timestamp. Unary client activity refreshes the lease by the same duration. `ExtendLease` and `IsLeaseExpired` cooperate with `SessionManager.CloseExpiredLeasesAsync`, which iterates a registry snapshot and closes any session whose lease has expired with `LeaseExpiredReason`. `SessionLeaseMonitorHostedService` runs that sweep every `MxGateway:Sessions:LeaseSweepIntervalSeconds` seconds (default 30).
#### Detach-grace retention
@@ -207,10 +209,16 @@ Mechanically: when the last external subscriber detaches and `DetachGraceSeconds
`DetachGraceSeconds` controls retention and expiry only; the reconnect/replay path that re-attaches a dropped client to a retained session is described in [Reconnect and replay](#reconnect-and-replay).
#### Faulted-session reaping
A session that faults (for example, a slow single-subscriber client that overflows its event queue under the `FailFast` backpressure policy) is left permanently unusable — every command fails `EvaluateReadyUnderLock` — but its worker keeps running and it still pins one of `MxGateway:Sessions:MaxSessions` slots. Rather than waiting up to `DefaultLeaseSeconds` for the lease to expire, `SessionManager.CloseExpiredLeasesAsync` also checks `IsFaultedReapable(now)` and reaps a faulted session through the same `TryBeginCloseIfExpired``CloseSessionCoreAsync` teardown, with the distinct `FaultedReason` (`"faulted-reaped"`). `MarkFaulted` stamps a fault timestamp from the session's `TimeProvider`; `MxGateway:Sessions:FaultedGraceSeconds` (default `0`) bounds how long the faulted session is retained before the next sweep reaps it — `0` reaps it on the next sweep, a positive value keeps it observable via `GetSessionStatus` for that window first. Sweep precedence when several conditions fire at once is lease-expiry, then faulted, then detach-grace.
#### Reconnect and replay
A client that drops mid-stream reconnects by re-issuing `StreamEvents` with `StreamEventsRequest.after_worker_sequence` set to the last `worker_sequence` it observed. A non-zero `after_worker_sequence` means *resume*; `0` means *fresh stream* and behaves exactly as a first-time subscribe — no replay, no sentinel.
**Owner-scoped attach (security control).** `OpenSession` records the caller's API key id on the session as `GatewaySession.OwnerKeyId`. Every `StreamEvents` attach and reattach is owner-checked: `EventStreamService.StreamEventsAsync` compares the caller's key id (threaded from `GatewayGrpcAuthorizationInterceptor` via `MxAccessGatewayService.StreamEvents`) to the session owner and throws `SessionManagerException(PermissionDenied)` — surfaced as gRPC `PermissionDenied` — when they differ, before any subscriber is attached and before any replayed or live event is delivered. Possessing the `event` scope and knowing a session id is not enough. This is what makes the default-on detach-grace and replay-retention windows safe: without the check, any `event`-scoped key that learned a session id could attach to another key's retained session and receive its replayed and live data. A `null` owner (session opened with authentication disabled) matches only a `null` caller key.
On a resume, `EventStreamService.StreamEventsAsync` attaches through `GatewaySession.AttachEventSubscriberWithReplay`, which calls `SessionEventDistributor.RegisterWithReplay`. That method snapshots the session's replay ring for events newer than `after_worker_sequence` **and** registers the live subscriber inside a single `_replayLock` critical section. This atomicity is what makes the replay→live handoff free of gaps and duplicates: the pump appends each event to the replay ring (under `_replayLock`) before fanning it to subscriber channels, so relative to that one critical section every event is either in the replay snapshot or fanned into the freshly-registered live channel — never both observably, never neither.
The handoff is sealed by a watermark. `RegisterWithReplay` returns `LiveResumeSequence` (the highest replayed sequence, or `after_worker_sequence` when nothing was replayed); `EventStreamService` then filters the live channel to events strictly greater than that watermark. An event that was both included in the replay snapshot and — racing the registration — also written to the live channel has `worker_sequence <= LiveResumeSequence`, so the live filter drops it exactly once (no duplicate), while every newer event is delivered (no gap). The same per-item filter governs replayed and live events identically, so a constrained or resuming caller never sees a replayed event it could not have seen live.
+12
View File
@@ -143,6 +143,17 @@ session if the worker faults. Gated by `MxGateway:Alarms:Enabled` — see
`docs/DesignDecisions.md` for why this reverses the v1 single-subscriber rule
for the alarm subsystem.
The monitor consumes its session's events as an **internal distributor
subscriber** (`GatewaySession.AttachInternalEventSubscriber`), not by draining
the worker event channel directly. The single worker event channel therefore has
exactly one reader — the per-session `SessionEventDistributor` pump — which fans
every event to both the dashboard mirror and the alarm feed; the alarm subscriber
is internal (`isInternal: true`), so it is not counted against
`MaxEventSubscribersPerSession` and a slow alarm reconcile can never fault the
session. The worker event channel is single-reader and asserts it (a second
`WorkerClient.ReadEventsAsync` consumer throws), so a regression cannot silently
split the event stream between two drains.
### Alarm providers and failover
The alarm feed has two providers, both implemented worker-side:
@@ -530,6 +541,7 @@ read, all of which contradict MXAccess semantics.
- `element_data_type` that is `Raw` or `Unspecified`
- an element `value` whose kind does not match `element_data_type`
- `total_length` exceeds the gateway-configured maximum array length
(`MxGateway:Events:MaxSparseArrayLength`, default 1,000,000)
An empty `elements` list with a non-zero `total_length` is valid — it writes an
all-defaults array of length `total_length` (explicit reset). A `sparse_array_value`
@@ -225,11 +225,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic
Task reconcileLoop = ReconcileLoopAsync(session.SessionId, linked.Token);
try
{
await foreach (WorkerEvent workerEvent in _sessionManager
.ReadEventsAsync(session.SessionId, linked.Token)
// Consume mapped MxEvents through the session's single distributor pump (as an
// internal, non-counted subscriber) rather than opening a second raw drain of the
// worker event channel — a second drain would split events with the dashboard
// mirror pump and silently lose Acknowledge/mode-change transitions (GWC-01).
await foreach (MxEvent mxEvent in _sessionManager
.ReadAlarmEventsAsync(session.SessionId, linked.Token)
.ConfigureAwait(false))
{
MxEvent? mxEvent = workerEvent.Event;
if (mxEvent is { BodyCase: MxEvent.BodyOneofCase.OnAlarmTransition }
&& mxEvent.OnAlarmTransition is not null)
{
@@ -27,4 +27,14 @@ public sealed class EventOptions
/// bounds the buffer).
/// </summary>
public double ReplayRetentionSeconds { get; init; } = 300;
/// <summary>
/// Gets the maximum <c>total_length</c> a sparse-array write may declare before the
/// gateway rejects it with <c>InvalidArgument</c>, enforced in
/// <see cref="Sessions.SparseArrayExpander"/> before the full array is materialized.
/// Guards against a single write forcing a multi-GB allocation. The default of
/// 1,000,000 elements is far above any realistic MXAccess array write yet well below
/// the frame-size ceiling.
/// </summary>
public int MaxSparseArrayLength { get; init; } = 1_000_000;
}
@@ -181,6 +181,10 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
options.DetachGraceSeconds,
"MxGateway:Sessions:DetachGraceSeconds must be zero or greater (0 disables detach-grace retention).",
builder);
AddIfNegative(
options.FaultedGraceSeconds,
"MxGateway:Sessions:FaultedGraceSeconds must be zero or greater (0 reaps a faulted session on the next sweep).",
builder);
AddIfNegative(
options.WorkerReadyWaitTimeoutMs,
"MxGateway:Sessions:WorkerReadyWaitTimeoutMs must be greater than or equal to zero.",
@@ -214,6 +218,10 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
builder.RequireThat(
options.ReplayRetentionSeconds >= 0,
"MxGateway:Events:ReplayRetentionSeconds must be greater than or equal to zero.");
builder.RequireThat(
options.MaxSparseArrayLength >= 1 && options.MaxSparseArrayLength <= Array.MaxLength,
$"MxGateway:Events:MaxSparseArrayLength must be between 1 and {Array.MaxLength}.");
}
private static void ValidateDashboard(DashboardOptions options, ValidationBuilder builder)
@@ -45,6 +45,18 @@ public sealed class SessionOptions
/// </remarks>
public int DetachGraceSeconds { get; init; } = 30;
/// <summary>
/// Gets the grace period, in seconds, that a faulted session is retained before the
/// lease monitor reaps it (killing its worker and freeing the session slot). A faulted
/// session is otherwise permanently unusable — every command fails the readiness check —
/// yet without this sweep it pins a session slot and a live x86 worker until its normal
/// lease expires (up to <see cref="DefaultLeaseSeconds"/>). A value of <c>0</c> (the
/// default) reaps a faulted session on the next sweep cycle, bounding the blast radius; a
/// positive value keeps the faulted session observable via <c>GetSessionStatus</c> for the
/// grace window before it is reclaimed. Must be greater than or equal to zero.
/// </summary>
public int FaultedGraceSeconds { get; init; }
/// <summary>
/// Gets a value indicating whether multiple event subscribers are allowed per session.
/// </summary>
@@ -48,6 +48,7 @@ public sealed class EventStreamService(
/// </remarks>
public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
string? callerKeyId,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
if (!sessionManager.TryGetSession(request.SessionId, out GatewaySession? session) || session is null)
@@ -57,6 +58,19 @@ public sealed class EventStreamService(
$"Session {request.SessionId} was not found.");
}
// Owner-scoped attach (TST-02, security control): a session's event stream may be
// attached or reattached ONLY by the API key that opened the session. The detach-grace
// and fan-out retention windows are on by default, so without this check any event-scoped
// key that learns a session id could attach to another key's retained session and receive
// its replayed and live data. Ordinal comparison; null owner (session opened with no auth)
// matches only a null caller key.
if (!string.Equals(session.OwnerKeyId, callerKeyId, StringComparison.Ordinal))
{
throw new SessionManagerException(
SessionManagerErrorCode.PermissionDenied,
$"Session {request.SessionId} is owned by a different API key; event-stream attach is owner-scoped.");
}
// No `using` here — subscriber.Dispose() is called exactly once in the finally
// block below, which also disposes the reader. A `using` declaration would add a
// second Dispose on the same path and double-decrement the session subscriber count.
@@ -11,9 +11,16 @@ public interface IEventStreamService
/// Streams events for the specified session to the caller.
/// </summary>
/// <param name="request">Request payload.</param>
/// <param name="callerKeyId">
/// The API key id of the calling client, used to enforce that only the key that opened a
/// session may attach or reattach its event stream. <see langword="null"/> when the call
/// is unauthenticated (e.g. auth disabled), which only matches a session opened with no
/// owner key.
/// </param>
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
/// <returns>The events emitted for the requested session.</returns>
IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
string? callerKeyId,
CancellationToken cancellationToken);
}
@@ -148,7 +148,7 @@ public sealed class MxAccessGatewayService(
{
requestValidator.ValidateStreamEvents(request);
await foreach (MxEvent publicEvent in eventStreamService
.StreamEventsAsync(request, context.CancellationToken)
.StreamEventsAsync(request, identityAccessor.Current?.KeyId, context.CancellationToken)
.WithCancellation(context.CancellationToken)
.ConfigureAwait(false))
{
@@ -931,6 +931,7 @@ public sealed class MxAccessGatewayService(
SessionManagerErrorCode.SessionLimitExceeded => StatusCode.ResourceExhausted,
SessionManagerErrorCode.OpenFailed => StatusCode.Unavailable,
SessionManagerErrorCode.CloseFailed => StatusCode.Unavailable,
SessionManagerErrorCode.PermissionDenied => StatusCode.PermissionDenied,
_ => StatusCode.Unavailable,
};
@@ -23,8 +23,10 @@ public sealed class GatewaySession
private bool _closeStarted;
private int _activeEventSubscriberCount;
private readonly TimeSpan _detachGrace;
private readonly TimeSpan _faultedGrace;
private readonly TimeSpan _workerReadyWaitTimeout;
private DateTimeOffset? _detachedAtUtc;
private DateTimeOffset? _faultedAtUtc;
// True once at least one external subscriber attached SUCCESSFULLY. Detach-grace's
// "last subscriber dropped" stamp (see DetachEventSubscriber) is gated on this so a
// FAILED first attach — which still runs the rollback DetachEventSubscriber from the
@@ -139,6 +141,14 @@ public sealed class GatewaySession
/// When <see langword="null"/> (legacy unit-construction paths that do not exercise Galaxy
/// metadata), addresses pass through unchanged.
/// </param>
/// <param name="faultedGrace">
/// Grace window kept after the session faults before the lease monitor reaps it. When the
/// window is positive the faulted session stays observable via <c>GetSessionStatus</c> for
/// that long before it is reclaimed; <see cref="TimeSpan.Zero"/> (the default) makes the
/// session reapable on the next sweep. The fault timestamp is stamped in
/// <see cref="MarkFaulted"/> using <paramref name="eventStreaming"/>'s clock so the timer
/// is unit-testable.
/// </param>
public GatewaySession(
string sessionId,
string backendName,
@@ -156,7 +166,8 @@ public sealed class GatewaySession
SessionEventStreaming? eventStreaming = null,
TimeSpan detachGrace = default,
TimeSpan workerReadyWaitTimeout = default,
ArrayAddressNormalizer? addressNormalizer = null)
ArrayAddressNormalizer? addressNormalizer = null,
TimeSpan faultedGrace = default)
{
if (string.IsNullOrWhiteSpace(sessionId))
{
@@ -195,6 +206,7 @@ public sealed class GatewaySession
_leaseExpiresAt = openedAt + leaseDuration;
_eventStreaming = eventStreaming ?? SessionEventStreaming.Default;
_detachGrace = detachGrace > TimeSpan.Zero ? detachGrace : TimeSpan.Zero;
_faultedGrace = faultedGrace > TimeSpan.Zero ? faultedGrace : TimeSpan.Zero;
_workerReadyWaitTimeout = workerReadyWaitTimeout > TimeSpan.Zero ? workerReadyWaitTimeout : TimeSpan.Zero;
_addressNormalizer = addressNormalizer;
}
@@ -522,6 +534,33 @@ public sealed class GatewaySession
}
}
/// <summary>
/// Registers a gateway-owned <em>internal</em> (non-counted) distributor subscriber and
/// returns its lease. The lease's <see cref="IEventSubscriberLease.Reader"/> yields the
/// same mapped <see cref="MxEvent"/>s the single distributor pump fans to every
/// subscriber; disposing the lease unregisters it.
/// </summary>
/// <remarks>
/// Used by the central alarm monitor so it consumes events through the one distributor
/// pump instead of opening a second raw drain of the single worker event channel (which
/// would split events between the two readers). Mirrors the dashboard-mirror lease:
/// <c>isInternal: true</c> keeps this subscriber out of the
/// <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
/// disconnects this internal subscriber.
/// </remarks>
/// <returns>The internal subscriber's lease; dispose it to unregister.</returns>
public IEventSubscriberLease AttachInternalEventSubscriber()
{
// 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
// subscriber is always present at pump start, then start the pump if requested.
SessionEventDistributor distributor = EnsureDistributorCreated(out bool startNow);
IEventSubscriberLease lease = distributor.Register(isInternal: true);
StartPumpIfRequested(distributor, startNow);
return lease;
}
private static void StartPumpIfRequested(SessionEventDistributor distributor, bool startNow)
{
if (!startNow)
@@ -724,6 +763,11 @@ public sealed class GatewaySession
_finalFault = reason;
_state = SessionState.Faulted;
// Stamp the fault time once, on the first fault, so the sweeper can apply
// FaultedGraceSeconds. A subsequent MarkFaulted (already-faulted session) keeps the
// original timestamp so the grace window is measured from the first fault.
_faultedAtUtc ??= _eventStreaming.TimeProvider.GetUtcNow();
}
}
@@ -790,6 +834,24 @@ public sealed class GatewaySession
}
}
/// <summary>
/// Determines whether a faulted session is now eligible for reaping by the lease monitor.
/// A faulted session is permanently unusable (every command fails the readiness check),
/// so the sweeper closes it exactly as it closes an expired lease — but no sooner than the
/// configured <c>FaultedGraceSeconds</c> after the fault, so a monitoring client can still
/// observe the fault before the slot is reclaimed. Always returns <see langword="false"/>
/// for a non-faulted session.
/// </summary>
/// <param name="now">Current timestamp for comparison.</param>
/// <returns><see langword="true"/> if the session is faulted and past its fault-grace window; otherwise <see langword="false"/>.</returns>
public bool IsFaultedReapable(DateTimeOffset now)
{
lock (_syncRoot)
{
return IsFaultedReapableCore(now);
}
}
/// <summary>
/// Attaches an event subscriber and returns a lease whose
/// <see cref="IEventSubscriberLease.Reader"/> reads the fanned public
@@ -1053,12 +1115,13 @@ public sealed class GatewaySession
_addressNormalizer?.Normalize(address) ?? address;
// MXAccess writes replace the whole array; expand a sparse value in place so the worker only
// ever receives a whole-array MxValue. No-op for null or non-sparse values.
private static void ExpandValue(MxValue? value)
// ever receives a whole-array MxValue. No-op for null or non-sparse values. The configured
// MxGateway:Events:MaxSparseArrayLength cap is enforced before the full array is allocated.
private void ExpandValue(MxValue? value)
{
if (value is not null)
{
SparseArrayExpander.Expand(value);
SparseArrayExpander.Expand(value, _eventStreaming.EventOptions.MaxSparseArrayLength);
}
}
@@ -1571,7 +1634,7 @@ public sealed class GatewaySession
// Re-verify eligibility atomically. If a subscriber reattached between the sweep's
// eligibility check and this point, neither condition holds and we decline.
bool eligible = IsLeaseExpiredCore(now) || IsDetachGraceExpiredCore(now);
bool eligible = IsLeaseExpiredCore(now) || IsFaultedReapableCore(now) || IsDetachGraceExpiredCore(now);
if (!eligible)
{
alreadyClosing = false;
@@ -1597,6 +1660,12 @@ public sealed class GatewaySession
&& _detachedAtUtc is not null
&& now - _detachedAtUtc.Value >= _detachGrace;
private bool IsFaultedReapableCore(DateTimeOffset now)
=> _state is SessionState.Faulted
&& (_faultedGrace <= TimeSpan.Zero
|| _faultedAtUtc is null
|| now - _faultedAtUtc.Value >= _faultedGrace);
// Final terminal transition; under _syncRoot to keep _state writes single-lock.
// Closed is unconditionally terminal — TransitionTo refuses to overwrite it —
// so we don't need to re-check the precondition here.
@@ -43,6 +43,18 @@ public interface ISessionManager
string sessionId,
CancellationToken cancellationToken);
/// <summary>
/// Reads mapped events for the central alarm monitor by attaching an internal
/// (non-counted) distributor subscriber, so the alarm feed shares the one worker-event
/// pump instead of opening a second raw drain of the single worker event channel.
/// </summary>
/// <param name="sessionId">Identifier of the session.</param>
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
/// <returns>The mapped <see cref="MxEvent"/>s fanned by the session's distributor.</returns>
IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
CancellationToken cancellationToken);
/// <summary>Closes a session and terminates its worker process.</summary>
/// <param name="sessionId">Identifier of the session to close.</param>
/// <param name="cancellationToken">Token to cancel the asynchronous operation.</param>
@@ -1,4 +1,5 @@
using System.Diagnostics.CodeAnalysis;
using System.Runtime.CompilerServices;
using System.Security.Cryptography;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging;
@@ -18,6 +19,7 @@ public sealed class SessionManager : ISessionManager
public const string GatewayShutdownReason = "gateway-shutdown";
public const string LeaseExpiredReason = "lease-expired";
public const string DetachGraceExpiredReason = "detach-grace-expired";
public const string FaultedReason = "faulted-reaped";
private readonly ISessionRegistry _registry;
private readonly ISessionWorkerClientFactory _workerClientFactory;
@@ -189,6 +191,22 @@ public sealed class SessionManager : ISessionManager
return session.ReadEventsAsync(cancellationToken);
}
/// <inheritdoc />
public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
GatewaySession session = GetRequiredSession(sessionId);
using IEventSubscriberLease lease = session.AttachInternalEventSubscriber();
await foreach (MxEvent mxEvent in lease.Reader
.ReadAllAsync(cancellationToken)
.ConfigureAwait(false))
{
yield return mxEvent;
}
}
/// <inheritdoc />
public async Task<SessionCloseResult> CloseSessionAsync(
string sessionId,
@@ -259,22 +277,25 @@ public sealed class SessionManager : ISessionManager
int closedCount = 0;
foreach (GatewaySession session in _registry.Snapshot())
{
// A session is swept when its normal lease has expired OR its detach-grace
// retention window has elapsed (last external subscriber dropped and no client
// reconnected within DetachGraceSeconds). The detach-grace close is the same
// teardown as a lease-expiry close; only the reason differs so operators can tell
// a short reconnect-window expiry from a long idle-lease expiry in logs/metrics.
// Lease-expiry takes PRECEDENCE over detach-grace when both conditions fire
// simultaneously (reason will be lease-expired, not detach-grace-expired).
// A session is swept when its normal lease has expired, it has FAULTED (a faulted
// session is permanently unusable yet otherwise pins a session slot and a live x86
// worker until its DefaultLeaseSeconds lease expires), OR its detach-grace retention
// window has elapsed (last external subscriber dropped and no client reconnected
// within DetachGraceSeconds). All three are the same teardown; only the reason differs
// so operators can tell a fault reap from a short reconnect-window expiry from a long
// idle-lease expiry in logs/metrics. Precedence when several fire simultaneously:
// lease-expired, then faulted, then detach-grace.
//
// TOCTOU note: eligibility is re-verified atomically inside TryBeginCloseIfExpired
// under _syncRoot, so a client that reattaches a subscriber between the check above
// and the close call wins the race and the session is left open and usable.
string? reason = session.IsLeaseExpired(now)
? LeaseExpiredReason
: session.IsDetachGraceExpired(now)
? DetachGraceExpiredReason
: null;
: session.IsFaultedReapable(now)
? FaultedReason
: session.IsDetachGraceExpired(now)
? DetachGraceExpiredReason
: null;
if (reason is null)
{
continue;
@@ -457,7 +478,8 @@ public sealed class SessionManager : ISessionManager
eventStreaming,
TimeSpan.FromSeconds(Math.Max(0, _options.Sessions.DetachGraceSeconds)),
TimeSpan.FromMilliseconds(Math.Max(0, _options.Sessions.WorkerReadyWaitTimeoutMs)),
_addressNormalizer);
_addressNormalizer,
TimeSpan.FromSeconds(Math.Max(0, _options.Sessions.FaultedGraceSeconds)));
}
private static string CreateClientCorrelationId(
@@ -10,4 +10,11 @@ public enum SessionManagerErrorCode
SessionLimitExceeded,
OpenFailed,
CloseFailed,
/// <summary>
/// The caller is not permitted to perform the operation on this session — for example,
/// attaching a <c>StreamEvents</c> stream to a session opened by a different API key.
/// Maps to gRPC <c>PermissionDenied</c>.
/// </summary>
PermissionDenied,
}
@@ -33,13 +33,20 @@ internal static class SparseArrayExpander
/// a sparse array this is a no-op, so callers may invoke it unconditionally.
/// </summary>
/// <param name="value">The value to expand in place.</param>
/// <param name="maxSparseArrayLength">
/// The maximum <c>total_length</c> the sparse array may declare before the write is
/// rejected, enforced before the full array is allocated (see
/// <c>MxGateway:Events:MaxSparseArrayLength</c>). Defaults to <see cref="int.MaxValue"/>
/// so the <see cref="Array.MaxLength"/> backstop is the only bound in test/unit-construction
/// paths that do not thread the configured cap.
/// </param>
/// <exception cref="RpcException">
/// <see cref="StatusCode.InvalidArgument"/> when the sparse payload is invalid: zero
/// total length, an index at or beyond the total length, a duplicate index, an
/// unsupported element type, or an element value whose kind does not match the declared
/// element type.
/// total length, a total length exceeding <paramref name="maxSparseArrayLength"/>, an index
/// at or beyond the total length, a duplicate index, an unsupported element type, or an
/// element value whose kind does not match the declared element type.
/// </exception>
public static void Expand(MxValue value)
public static void Expand(MxValue value, int maxSparseArrayLength = int.MaxValue)
{
ArgumentNullException.ThrowIfNull(value);
@@ -62,6 +69,12 @@ internal static class SparseArrayExpander
throw Invalid($"Sparse array element_data_type '{elementType}' is not a supported scalar element type.");
}
if (totalLength > (uint)maxSparseArrayLength)
{
throw Invalid(
$"Sparse array total_length {totalLength} exceeds the configured maximum {maxSparseArrayLength} (MxGateway:Events:MaxSparseArrayLength).");
}
if (totalLength > (uint)Array.MaxLength)
{
throw Invalid(
@@ -32,6 +32,7 @@ public sealed class WorkerClient : IWorkerClient
private DateTimeOffset _lastHeartbeatAt;
private int? _processId;
private int _eventQueueDepth;
private int _eventsReaderClaimed;
private Task? _readLoopTask;
private Task? _writeLoopTask;
private Task? _heartbeatLoopTask;
@@ -70,7 +71,13 @@ public sealed class WorkerClient : IWorkerClient
_events = Channel.CreateBounded<WorkerEvent>(
new BoundedChannelOptions(_options.EventChannelCapacity)
{
SingleReader = false,
// The worker event channel has exactly ONE consumer: the per-session
// SessionEventDistributor pump. The alarm monitor and dashboard mirror both
// attach to the distributor rather than draining this channel directly, so a
// second concurrent reader would silently split events between the two
// enumerators. SingleReader=true asserts that invariant; ReadEventsAsync adds a
// claimed-once guard so a regression fails loudly instead of losing events.
SingleReader = true,
SingleWriter = true,
FullMode = BoundedChannelFullMode.Wait,
AllowSynchronousContinuations = false,
@@ -224,7 +231,24 @@ public sealed class WorkerClient : IWorkerClient
}
/// <inheritdoc />
public async IAsyncEnumerable<WorkerEvent> ReadEventsAsync(
public IAsyncEnumerable<WorkerEvent> ReadEventsAsync(CancellationToken cancellationToken)
{
// The event channel is SingleReader: only one enumerator may ever drain it, otherwise
// the two readers would each receive a random subset of events. Claim the reader at CALL
// time (not lazily on first MoveNext) and fail loudly on a second consumer rather than
// silently splitting the stream (see GWC-01). The distributor pump is the only intended
// caller; the alarm monitor and dashboard mirror attach to the distributor instead.
if (Interlocked.CompareExchange(ref _eventsReaderClaimed, 1, 0) != 0)
{
throw new InvalidOperationException(
"WorkerClient.ReadEventsAsync was already claimed by another consumer. The worker event "
+ "channel is single-reader; attach to the SessionEventDistributor instead of draining it twice.");
}
return ReadEventsCoreAsync(cancellationToken);
}
private async IAsyncEnumerable<WorkerEvent> ReadEventsCoreAsync(
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
@@ -470,6 +470,20 @@ public sealed class AlarmFailoverEndToEndTests
}
}
/// <inheritdoc />
public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
{
if (workerEvent.Event is not null)
{
yield return workerEvent.Event;
}
}
}
/// <inheritdoc />
public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
{
@@ -783,6 +783,20 @@ public sealed class GatewayAlarmMonitorProviderModeTests
}
}
/// <inheritdoc />
public async IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken))
{
if (workerEvent.Event is not null)
{
yield return workerEvent.Event;
}
}
}
/// <inheritdoc />
public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session)
{
@@ -520,4 +520,32 @@ public sealed class GatewayOptionsValidatorTests
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
Assert.True(result.Succeeded);
}
/// <summary>Verifies a <see cref="EventOptions.MaxSparseArrayLength"/> below one fails validation.</summary>
/// <param name="value">Sparse-array cap under test.</param>
[Theory]
[InlineData(0)]
[InlineData(-1)]
public void Validate_Fails_WhenMaxSparseArrayLengthBelowOne(int value)
{
GatewayOptions options = CloneWithEvents(
ValidOptions(),
new EventOptions { MaxSparseArrayLength = value });
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
Assert.True(result.Failed);
Assert.Contains(
result.Failures!,
f => f.Contains("MxGateway:Events:MaxSparseArrayLength"));
}
/// <summary>Verifies a positive <see cref="EventOptions.MaxSparseArrayLength"/> within range passes validation.</summary>
[Fact]
public void Validate_Succeeds_WhenMaxSparseArrayLengthWithinRange()
{
GatewayOptions options = CloneWithEvents(
ValidOptions(),
new EventOptions { MaxSparseArrayLength = 1 });
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options);
Assert.True(result.Succeeded);
}
}
@@ -281,6 +281,14 @@ public sealed class DashboardSessionAdminServiceTests
throw new NotSupportedException();
}
/// <inheritdoc />
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
CancellationToken cancellationToken)
{
throw new NotSupportedException();
}
/// <inheritdoc />
public Task<SessionCloseResult> CloseSessionAsync(
string sessionId,
@@ -38,6 +38,58 @@ public sealed class EventStreamServiceTests
Assert.Equal(1, metrics.GetSnapshot().StreamDisconnects);
}
/// <summary>
/// TST-02 (owner-scoped attach): the API key that opened a session may attach its event
/// stream — the caller key equals the session owner, so streaming proceeds normally.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task StreamEventsAsync_WhenCallerKeyMatchesOwner_Streams()
{
FakeWorkerClient workerClient = new();
GatewaySession session = CreateReadySession(workerClient, ownerKeyId: "key-owner");
EventStreamService service = CreateService(new FakeSessionManager(session));
workerClient.Events.Add(CreateWorkerEvent(sequence: 5, MxEventFamily.OnDataChange));
workerClient.CompleteAfterConfiguredEvents = true;
List<MxEvent> events = [];
await foreach (MxEvent mxEvent in service
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: "key-owner", CancellationToken.None)
.WithCancellation(CancellationToken.None))
{
events.Add(mxEvent);
}
Assert.Equal([5UL], events.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
}
/// <summary>
/// TST-02 (owner-scoped attach, security control): a caller whose API key differs from
/// the key that opened the session is rejected with a <see cref="SessionManagerErrorCode.PermissionDenied"/>
/// fault before any events are streamed — closing the reconnect/fan-out trust-boundary hole.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task StreamEventsAsync_WhenCallerKeyDiffersFromOwner_ThrowsPermissionDenied()
{
FakeWorkerClient workerClient = new();
GatewaySession session = CreateReadySession(workerClient, ownerKeyId: "key-owner");
EventStreamService service = CreateService(new FakeSessionManager(session));
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(async () =>
{
await foreach (MxEvent _ in service
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: "key-intruder", CancellationToken.None)
.WithCancellation(CancellationToken.None))
{
// No event should be yielded — the owner check runs before the first attach.
}
});
Assert.Equal(SessionManagerErrorCode.PermissionDenied, exception.ErrorCode);
Assert.Equal(0, session.ActiveEventSubscriberCount);
}
/// <summary>Verifies that a second event subscriber is rejected when one is already active.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
@@ -48,13 +100,13 @@ public sealed class EventStreamServiceTests
EventStreamService service = CreateService(new FakeSessionManager(session));
using CancellationTokenSource firstSubscriberCancellation = new();
await using IAsyncEnumerator<MxEvent> firstSubscriber = service
.StreamEventsAsync(CreateRequest(session.SessionId), firstSubscriberCancellation.Token)
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, firstSubscriberCancellation.Token)
.GetAsyncEnumerator(firstSubscriberCancellation.Token);
Task<bool> firstMoveTask = firstSubscriber.MoveNextAsync().AsTask();
await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 1);
await using IAsyncEnumerator<MxEvent> secondSubscriber = service
.StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
@@ -78,7 +130,7 @@ public sealed class EventStreamServiceTests
EventStreamService service = CreateService(new FakeSessionManager(session));
using CancellationTokenSource cancellationTokenSource = new();
await using IAsyncEnumerator<MxEvent> subscriber = service
.StreamEventsAsync(CreateRequest(session.SessionId), cancellationTokenSource.Token)
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, cancellationTokenSource.Token)
.GetAsyncEnumerator(cancellationTokenSource.Token);
Task<bool> moveTask = subscriber.MoveNextAsync().AsTask();
@@ -108,7 +160,7 @@ public sealed class EventStreamServiceTests
workerClient.Events.Add(CreateWorkerEvent(sequence: 3, MxEventFamily.OnDataChange));
workerClient.CompleteAfterConfiguredEvents = true;
await using IAsyncEnumerator<MxEvent> subscriber = service
.StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
Assert.True(await subscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
@@ -142,10 +194,10 @@ public sealed class EventStreamServiceTests
firstWorkerClient.CompleteAfterConfiguredEvents = true;
secondWorkerClient.CompleteAfterConfiguredEvents = true;
await using IAsyncEnumerator<MxEvent> firstSubscriber = service
.StreamEventsAsync(CreateRequest(firstSession.SessionId), CancellationToken.None)
.StreamEventsAsync(CreateRequest(firstSession.SessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
await using IAsyncEnumerator<MxEvent> secondSubscriber = service
.StreamEventsAsync(CreateRequest(secondSession.SessionId), CancellationToken.None)
.StreamEventsAsync(CreateRequest(secondSession.SessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
Assert.True(await firstSubscriber.MoveNextAsync().AsTask().WaitAsync(TestTimeout));
@@ -192,7 +244,7 @@ public sealed class EventStreamServiceTests
workerClient.CompleteAfterConfiguredEvents = true;
await using IAsyncEnumerator<MxEvent> subscriber = service
.StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
// The pump fans 50 events into a subscriber channel with capacity 1 faster than this
@@ -246,7 +298,7 @@ public sealed class EventStreamServiceTests
workerClient.CompleteAfterConfiguredEvents = true;
await using IAsyncEnumerator<MxEvent> subscriber = service
.StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
SessionManagerException exception = await Assert.ThrowsAsync<SessionManagerException>(
@@ -298,7 +350,7 @@ public sealed class EventStreamServiceTests
using GatewayMetrics metrics = new();
EventStreamService service = CreateService(new FakeSessionManager(session), metrics);
await using IAsyncEnumerator<MxEvent> subscriber = service
.StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None)
.StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
@@ -333,7 +385,7 @@ public sealed class EventStreamServiceTests
// Resume after sequence 2: retained window [1..5] covers it — replay 3,4,5 then live.
await using IAsyncEnumerator<MxEvent> resume = service
.StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), CancellationToken.None)
.StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
MxEvent r3 = await ReadNextAsync(resume);
@@ -374,7 +426,7 @@ public sealed class EventStreamServiceTests
// Resume after 1: events 1,2 are below the oldest retained (3) and were evicted, so
// they are unrecoverable => sentinel first, then the retained tail 3,4,5, then live.
await using IAsyncEnumerator<MxEvent> realResume = service
.StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 1), CancellationToken.None)
.StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 1), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
MxEvent sentinel = await ReadNextAsync(realResume);
@@ -419,7 +471,7 @@ public sealed class EventStreamServiceTests
// Resume after 2: replay 3,4 then live 5,6,7. Collect across the boundary and assert
// the full sequence is contiguous with no duplicate and no skip.
await using IAsyncEnumerator<MxEvent> resume = service
.StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), CancellationToken.None)
.StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
List<ulong> collected = [];
@@ -461,7 +513,7 @@ public sealed class EventStreamServiceTests
// reads must be exactly 4 then 5 (no sentinel, no <=3 event); a live tag confirms the
// stream resumed live strictly after 5.
await using IAsyncEnumerator<MxEvent> resume = service
.StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 3), CancellationToken.None)
.StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 3), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
MxEvent first = await ReadNextAsync(resume);
@@ -512,7 +564,7 @@ public sealed class EventStreamServiceTests
int expectedCount)
{
await using IAsyncEnumerator<MxEvent> primer = service
.StreamEventsAsync(CreateRequest(sessionId), CancellationToken.None)
.StreamEventsAsync(CreateRequest(sessionId), callerKeyId: null, CancellationToken.None)
.GetAsyncEnumerator();
for (int i = 0; i < expectedCount; i++)
{
@@ -551,7 +603,7 @@ public sealed class EventStreamServiceTests
{
List<MxEvent> events = [];
await foreach (MxEvent mxEvent in service
.StreamEventsAsync(CreateRequest(sessionId), CancellationToken.None)
.StreamEventsAsync(CreateRequest(sessionId), callerKeyId: null, CancellationToken.None)
.WithCancellation(CancellationToken.None))
{
events.Add(mxEvent);
@@ -575,7 +627,8 @@ public sealed class EventStreamServiceTests
int queueCapacity = 8,
GatewayMetrics? metrics = null,
EventBackpressurePolicy backpressurePolicy = EventBackpressurePolicy.FailFast,
int replayBufferCapacity = 1024)
int replayBufferCapacity = 1024,
string? ownerKeyId = null)
{
// The per-subscriber overflow policy now lives in the session's
// SessionEventDistributor, so the session must share the same metrics sink and
@@ -587,7 +640,7 @@ public sealed class EventStreamServiceTests
"pipe",
"nonce",
"client",
ownerKeyId: null,
ownerKeyId: ownerKeyId,
"client-session",
"client-correlation",
TimeSpan.FromSeconds(30),
@@ -702,6 +755,14 @@ public sealed class EventStreamServiceTests
return _sessions[sessionId].ReadEventsAsync(cancellationToken);
}
/// <inheritdoc />
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
CancellationToken cancellationToken)
{
throw new NotSupportedException();
}
/// <inheritdoc />
public Task<SessionCloseResult> CloseSessionAsync(
string sessionId,
@@ -948,6 +948,14 @@ public sealed class MxAccessGatewayServiceConstraintTests
}
}
/// <inheritdoc />
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
CancellationToken cancellationToken)
{
throw new NotSupportedException();
}
/// <inheritdoc />
public Task<SessionCloseResult> CloseSessionAsync(
string sessionId,
@@ -994,6 +1002,7 @@ public sealed class MxAccessGatewayServiceConstraintTests
/// <inheritdoc />
public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
string? callerKeyId,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
{
foreach (WorkerEvent ev in sessionManager.Events)
@@ -616,6 +616,14 @@ public sealed class MxAccessGatewayServiceTests
}
}
/// <inheritdoc />
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
CancellationToken cancellationToken)
{
throw new NotSupportedException();
}
/// <inheritdoc />
public Task<SessionCloseResult> CloseSessionAsync(
string sessionId,
@@ -653,6 +661,7 @@ public sealed class MxAccessGatewayServiceTests
/// <inheritdoc />
public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
string? callerKeyId,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
sessionManager.RecordReadEventsSessionId(request.SessionId);
@@ -88,7 +88,7 @@ public sealed class GatewaySessionDashboardMirrorTests
Task grpcReader = Task.Run(async () =>
{
await foreach (MxEvent mxEvent in service
.StreamEventsAsync(new StreamEventsRequest { SessionId = session.SessionId }, CancellationToken.None)
.StreamEventsAsync(new StreamEventsRequest { SessionId = session.SessionId }, callerKeyId: null, CancellationToken.None)
.WithCancellation(CancellationToken.None))
{
grpcEvents.Add(mxEvent);
@@ -107,6 +107,65 @@ public sealed class GatewaySessionDashboardMirrorTests
Assert.Equal([1UL, 2UL, 3UL], broadcaster.Captures.Select(capture => capture.MxEvent.WorkerSequence).ToArray());
}
/// <summary>
/// GWC-01 regression: with the internal dashboard mirror active, a second internal
/// subscriber (the alarm monitor's feed, attached via
/// <see cref="GatewaySession.AttachInternalEventSubscriber"/>) receives EVERY event —
/// including the alarm <c>Acknowledge</c> transition — rather than the two consumers
/// each getting a random half of the single worker channel. Before the fix the alarm
/// monitor drained the worker channel directly, so with the dashboard mirror pump also
/// draining it the two split the stream and Acknowledge transitions were silently lost.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task InternalAlarmSubscriber_AndDashboardMirror_BothReceiveEveryEvent()
{
FakeWorkerClient workerClient = new();
workerClient.Events.Add(CreateWorkerEvent(1, MxEventFamily.OnDataChange));
workerClient.Events.Add(CreateAlarmTransitionEvent(2, AlarmTransitionKind.Raise));
workerClient.Events.Add(CreateAlarmTransitionEvent(3, AlarmTransitionKind.Acknowledge));
workerClient.Events.Add(CreateAlarmTransitionEvent(4, AlarmTransitionKind.Clear));
workerClient.CompleteAfterConfiguredEvents = true;
// Hold the finite stream until BOTH internal subscribers (dashboard mirror + alarm feed)
// are registered so neither misses an event before the pump drains.
workerClient.HoldEventsUntilReleased();
RecordingDashboardEventBroadcaster broadcaster = new();
await using GatewaySession session = CreateSession(workerClient, broadcaster);
session.AttachWorkerClient(workerClient);
// MarkReady registers the internal dashboard subscriber and starts the (gated) pump.
session.MarkReady();
// Attach the alarm monitor's internal subscriber and drain it concurrently. Registered
// BEFORE the stream is released, so it is present at pump start alongside the dashboard.
using IEventSubscriberLease alarmLease = session.AttachInternalEventSubscriber();
List<MxEvent> alarmEvents = [];
Task alarmReader = Task.Run(async () =>
{
await foreach (MxEvent mxEvent in alarmLease.Reader.ReadAllAsync(CancellationToken.None))
{
alarmEvents.Add(mxEvent);
}
});
workerClient.ReleaseEvents();
await WaitUntilAsync(() => broadcaster.Captures.Count == 4 && alarmEvents.Count == 4);
await alarmReader.WaitAsync(TestTimeout);
// Both internal consumers see the full, identical, ordered stream — no splitting.
Assert.Equal([1UL, 2UL, 3UL, 4UL], broadcaster.Captures.Select(capture => capture.MxEvent.WorkerSequence).ToArray());
Assert.Equal([1UL, 2UL, 3UL, 4UL], alarmEvents.Select(mxEvent => mxEvent.WorkerSequence).ToArray());
// The Acknowledge transition specifically reaches the alarm feed (the transition the
// pre-fix split silently dropped, leaving clients showing unacked alarms indefinitely).
Assert.Contains(
alarmEvents,
mxEvent => mxEvent.BodyCase == MxEvent.BodyOneofCase.OnAlarmTransition
&& mxEvent.OnAlarmTransition.TransitionKind == AlarmTransitionKind.Acknowledge);
}
/// <summary>
/// Hazard guard: starting the pump at Ready with a fast-completing worker stream
/// and zero subscribers used to drain into nothing and leave a later subscriber hanging.
@@ -222,6 +281,19 @@ public sealed class GatewaySessionDashboardMirrorTests
return new WorkerEvent { Event = mxEvent };
}
private static WorkerEvent CreateAlarmTransitionEvent(ulong sequence, AlarmTransitionKind kind)
{
MxEvent mxEvent = new()
{
SessionId = "session-dashboard-mirror",
Family = MxEventFamily.OnAlarmTransition,
WorkerSequence = sequence,
OnAlarmTransition = new OnAlarmTransitionEvent { TransitionKind = kind },
};
return new WorkerEvent { Event = mxEvent };
}
private static async Task WaitUntilAsync(Func<bool> predicate, [CallerArgumentExpression(nameof(predicate))] string? condition = null)
{
using CancellationTokenSource cancellationTokenSource = new(TestTimeout);
@@ -280,6 +352,11 @@ public sealed class GatewaySessionDashboardMirrorTests
string sessionId,
CancellationToken cancellationToken) => session.ReadEventsAsync(cancellationToken);
/// <inheritdoc />
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
CancellationToken cancellationToken) => throw new NotSupportedException();
/// <inheritdoc />
public Task<SessionCloseResult> CloseSessionAsync(
string sessionId,
@@ -975,6 +975,68 @@ public sealed class SessionManagerTests
Assert.Equal(0, workerClient.ShutdownCount);
}
/// <summary>
/// A faulted session is reaped by the lease sweep (default <c>FaultedGraceSeconds=0</c>)
/// even though its normal lease is still far in the future, tearing down its worker and
/// freeing the slot, while a healthy leased session in the same manager is untouched.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task CloseExpiredLeasesAsync_ReapsFaultedSession()
{
FakeWorkerClient faultedClient = new();
FakeWorkerClient healthyClient = new();
QueueingSessionWorkerClientFactory factory = new(faultedClient, healthyClient);
SessionManager manager = CreateManager(factory);
GatewaySession faultedSession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
GatewaySession healthySession = await manager.OpenSessionAsync(CreateOpenRequest(), "client-2", ownerKeyId: null, CancellationToken.None);
DateTimeOffset now = DateTimeOffset.UtcNow;
// Both leases are far in the future, so only the fault can reap the faulted session.
faultedSession.ExtendLease(now.AddMinutes(30));
healthySession.ExtendLease(now.AddMinutes(30));
faultedSession.MarkFaulted("test fault");
int closedCount = await manager.CloseExpiredLeasesAsync(now, CancellationToken.None);
Assert.Equal(1, closedCount);
Assert.Equal(SessionState.Closed, faultedSession.State);
Assert.Equal(1, faultedClient.ShutdownCount);
Assert.Equal(SessionState.Ready, healthySession.State);
Assert.Equal(0, healthyClient.ShutdownCount);
}
/// <summary>
/// With a positive <c>FaultedGraceSeconds</c>, a faulted session stays observable until
/// the grace window elapses, then the next sweep reaps it.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task CloseExpiredLeasesAsync_RespectsFaultedGraceWindow()
{
FakeWorkerClient workerClient = new();
FakeTimeProvider clock = new(DateTimeOffset.UtcNow);
SessionManager manager = CreateManager(
new FakeSessionWorkerClientFactory(workerClient),
options: CreateOptions(defaultLeaseSeconds: 1800, faultedGraceSeconds: 30),
timeProvider: clock);
GatewaySession session = await manager.OpenSessionAsync(CreateOpenRequest(), "client-1", ownerKeyId: null, CancellationToken.None);
session.MarkFaulted("test fault");
// Before the grace window elapses: the faulted session is retained (still observable).
clock.Advance(TimeSpan.FromSeconds(29));
int closedBefore = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
Assert.Equal(0, closedBefore);
Assert.Equal(SessionState.Faulted, session.State);
// After the grace window elapses: the sweep reaps it.
clock.Advance(TimeSpan.FromSeconds(1));
int closedAfter = await manager.CloseExpiredLeasesAsync(clock.GetUtcNow(), CancellationToken.None);
Assert.Equal(1, closedAfter);
Assert.Equal(SessionState.Closed, session.State);
Assert.Equal(1, workerClient.ShutdownCount);
}
/// <summary>Verifies that shutdown closes all registered sessions.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
@@ -1023,7 +1085,8 @@ public sealed class SessionManagerTests
int maxSessions = 64,
int defaultLeaseSeconds = 1800,
int detachGraceSeconds = 0,
int workerReadyWaitTimeoutMs = 0)
int workerReadyWaitTimeoutMs = 0,
int faultedGraceSeconds = 0)
{
return new GatewayOptions
{
@@ -1034,6 +1097,7 @@ public sealed class SessionManagerTests
DefaultLeaseSeconds = defaultLeaseSeconds,
DetachGraceSeconds = detachGraceSeconds,
WorkerReadyWaitTimeoutMs = workerReadyWaitTimeoutMs,
FaultedGraceSeconds = faultedGraceSeconds,
},
Worker = new WorkerOptions
{
@@ -222,4 +222,35 @@ public sealed class SparseArrayExpanderTests
RpcException ex = Assert.Throws<RpcException>(() => SparseArrayExpander.Expand(value));
Assert.Equal(StatusCode.InvalidArgument, ex.StatusCode);
}
/// <summary>Verifies that a total length above the configured cap throws <see cref="StatusCode.InvalidArgument"/> before the full array is allocated.</summary>
[Fact]
public void Expand_TotalLengthExceedsConfiguredCap_ThrowsBeforeAllocation()
{
// A total_length that would force a huge allocation, but well below Array.MaxLength,
// so only the configured cap can reject it (the Array.MaxLength backstop would not).
MxValue value = SparseValue(MxDataType.Integer, 500_000_000u);
RpcException ex = Assert.Throws<RpcException>(() => SparseArrayExpander.Expand(value, maxSparseArrayLength: 1_000_000));
Assert.Equal(StatusCode.InvalidArgument, ex.StatusCode);
Assert.Contains("MaxSparseArrayLength", ex.Status.Detail, StringComparison.Ordinal);
// The value must be untouched — expansion (allocation) never ran.
Assert.Equal(MxValue.KindOneofCase.SparseArrayValue, value.KindCase);
}
/// <summary>Verifies that a total length at or below the configured cap still expands normally.</summary>
[Fact]
public void Expand_TotalLengthAtConfiguredCap_Expands()
{
MxValue value = SparseValue(
MxDataType.Integer,
4,
(1, new MxValue { Int32Value = 7 }));
SparseArrayExpander.Expand(value, maxSparseArrayLength: 4);
Assert.Equal(MxValue.KindOneofCase.ArrayValue, value.KindCase);
Assert.Equal(new[] { 0, 7, 0, 0 }, value.ArrayValue.Int32Values.Values);
}
}
@@ -122,6 +122,27 @@ public sealed class WorkerClientTests
Assert.Equal(MxEventFamily.OperationComplete, events.Current.Event.Family);
}
/// <summary>
/// The worker event channel is single-reader: a second <see cref="WorkerClient.ReadEventsAsync"/>
/// enumerator must throw rather than silently split events between two consumers (GWC-01).
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task ReadEventsAsync_SecondEnumerator_Throws()
{
await using PipePair pipePair = await PipePair.CreateAsync();
await using WorkerClient client = CreateClient(pipePair);
await CompleteHandshakeAsync(client, pipePair);
using CancellationTokenSource cancellationTokenSource = new(TestTimeout);
// First call claims the single reader.
IAsyncEnumerable<WorkerEvent> first = client.ReadEventsAsync(cancellationTokenSource.Token);
await using IAsyncEnumerator<WorkerEvent> firstEnumerator = first.GetAsyncEnumerator(cancellationTokenSource.Token);
// A second call must fail loudly at call time rather than racing the first for events.
Assert.Throws<InvalidOperationException>(() => client.ReadEventsAsync(cancellationTokenSource.Token));
}
/// <summary>Verifies that the read loop faults the client when the event queue overflows.</summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
@@ -477,6 +477,14 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests
return AsyncEnumerable.Empty<WorkerEvent>();
}
/// <inheritdoc />
public IAsyncEnumerable<MxEvent> ReadAlarmEventsAsync(
string sessionId,
CancellationToken cancellationToken)
{
return AsyncEnumerable.Empty<MxEvent>();
}
/// <inheritdoc />
public Task<SessionCloseResult> CloseSessionAsync(
string sessionId,
@@ -515,6 +523,7 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests
/// <inheritdoc />
public async IAsyncEnumerable<MxEvent> StreamEventsAsync(
StreamEventsRequest request,
string? callerKeyId,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
await Task.CompletedTask;