From 20392cf246c71fee4bbeb7a226386488c7e9bf1a Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 05:51:57 -0400 Subject: [PATCH] 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. --- CLAUDE.md | 4 +- docs/DesignDecisions.md | 11 +++ docs/GatewayConfiguration.md | 6 +- docs/Sessions.md | 10 +- gateway.md | 12 +++ .../Alarms/GatewayAlarmMonitor.cs | 9 +- .../Configuration/EventOptions.cs | 10 ++ .../Configuration/GatewayOptionsValidator.cs | 8 ++ .../Configuration/SessionOptions.cs | 12 +++ .../Grpc/EventStreamService.cs | 14 +++ .../Grpc/IEventStreamService.cs | 7 ++ .../Grpc/MxAccessGatewayService.cs | 3 +- .../Sessions/GatewaySession.cs | 79 ++++++++++++++- .../Sessions/ISessionManager.cs | 12 +++ .../Sessions/SessionManager.cs | 44 ++++++--- .../Sessions/SessionManagerErrorCode.cs | 7 ++ .../Sessions/SparseArrayExpander.cs | 21 +++- .../Workers/WorkerClient.cs | 28 +++++- .../Alarms/AlarmFailoverEndToEndTests.cs | 14 +++ .../GatewayAlarmMonitorProviderModeTests.cs | 14 +++ .../GatewayOptionsValidatorTests.cs | 28 ++++++ .../DashboardSessionAdminServiceTests.cs | 8 ++ .../Gateway/Grpc/EventStreamServiceTests.cs | 95 +++++++++++++++---- .../MxAccessGatewayServiceConstraintTests.cs | 9 ++ .../Grpc/MxAccessGatewayServiceTests.cs | 9 ++ .../GatewaySessionDashboardMirrorTests.cs | 79 ++++++++++++++- .../Gateway/Sessions/SessionManagerTests.cs | 66 ++++++++++++- .../Sessions/SparseArrayExpanderTests.cs | 31 ++++++ .../Gateway/Workers/WorkerClientTests.cs | 21 ++++ ...atewayGrpcAuthorizationInterceptorTests.cs | 9 ++ 30 files changed, 632 insertions(+), 48 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 156f513..56fd95e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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__`. 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 diff --git a/docs/DesignDecisions.md b/docs/DesignDecisions.md index f3a9e54..93d4896 100644 --- a/docs/DesignDecisions.md +++ b/docs/DesignDecisions.md @@ -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 diff --git a/docs/GatewayConfiguration.md b/docs/GatewayConfiguration.md index f089548..005eb5a 100644 --- a/docs/GatewayConfiguration.md +++ b/docs/GatewayConfiguration.md @@ -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 diff --git a/docs/Sessions.md b/docs/Sessions.md index f9d0fac..77a9db1 100644 --- a/docs/Sessions.md +++ b/docs/Sessions.md @@ -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. diff --git a/gateway.md b/gateway.md index ddad5f7..30267df 100644 --- a/gateway.md +++ b/gateway.md @@ -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` diff --git a/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs b/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs index 568bbb4..1cd41b1 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs @@ -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) { diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/EventOptions.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/EventOptions.cs index 408fb79..b114246 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/EventOptions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/EventOptions.cs @@ -27,4 +27,14 @@ public sealed class EventOptions /// bounds the buffer). /// public double ReplayRetentionSeconds { get; init; } = 300; + + /// + /// Gets the maximum total_length a sparse-array write may declare before the + /// gateway rejects it with InvalidArgument, enforced in + /// 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. + /// + public int MaxSparseArrayLength { get; init; } = 1_000_000; } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs index e83d784..e5b8813 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs @@ -181,6 +181,10 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase= 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) diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/SessionOptions.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/SessionOptions.cs index f95cd55..535b243 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/SessionOptions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/SessionOptions.cs @@ -45,6 +45,18 @@ public sealed class SessionOptions /// public int DetachGraceSeconds { get; init; } = 30; + /// + /// 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 ). A value of 0 (the + /// default) reaps a faulted session on the next sweep cycle, bounding the blast radius; a + /// positive value keeps the faulted session observable via GetSessionStatus for the + /// grace window before it is reclaimed. Must be greater than or equal to zero. + /// + public int FaultedGraceSeconds { get; init; } + /// /// Gets a value indicating whether multiple event subscribers are allowed per session. /// diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs index 19b1a4f..5afce48 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/EventStreamService.cs @@ -48,6 +48,7 @@ public sealed class EventStreamService( /// public async IAsyncEnumerable 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. diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/IEventStreamService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/IEventStreamService.cs index 780e829..7666f84 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/IEventStreamService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/IEventStreamService.cs @@ -11,9 +11,16 @@ public interface IEventStreamService /// Streams events for the specified session to the caller. /// /// Request payload. + /// + /// 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. when the call + /// is unauthenticated (e.g. auth disabled), which only matches a session opened with no + /// owner key. + /// /// Token to cancel the asynchronous operation. /// The events emitted for the requested session. IAsyncEnumerable StreamEventsAsync( StreamEventsRequest request, + string? callerKeyId, CancellationToken cancellationToken); } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs index 67a7ae8..6d6d8d2 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs @@ -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, }; diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs index b5b9b55..ab1e41e 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/GatewaySession.cs @@ -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 (legacy unit-construction paths that do not exercise Galaxy /// metadata), addresses pass through unchanged. /// + /// + /// Grace window kept after the session faults before the lease monitor reaps it. When the + /// window is positive the faulted session stays observable via GetSessionStatus for + /// that long before it is reclaimed; (the default) makes the + /// session reapable on the next sweep. The fault timestamp is stamped in + /// using 's clock so the timer + /// is unit-testable. + /// 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 } } + /// + /// Registers a gateway-owned internal (non-counted) distributor subscriber and + /// returns its lease. The lease's yields the + /// same mapped s the single distributor pump fans to every + /// subscriber; disposing the lease unregisters it. + /// + /// + /// 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: + /// isInternal: true keeps this subscriber out of the + /// MaxEventSubscribersPerSession 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. + /// + /// The internal subscriber's lease; dispose it to unregister. + 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 } } + /// + /// 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 FaultedGraceSeconds after the fault, so a monitoring client can still + /// observe the fault before the slot is reclaimed. Always returns + /// for a non-faulted session. + /// + /// Current timestamp for comparison. + /// if the session is faulted and past its fault-grace window; otherwise . + public bool IsFaultedReapable(DateTimeOffset now) + { + lock (_syncRoot) + { + return IsFaultedReapableCore(now); + } + } + /// /// Attaches an event subscriber and returns a lease whose /// 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. diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs index f9c200e..c9431a1 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/ISessionManager.cs @@ -43,6 +43,18 @@ public interface ISessionManager string sessionId, CancellationToken cancellationToken); + /// + /// 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. + /// + /// Identifier of the session. + /// Token to cancel the asynchronous operation. + /// The mapped s fanned by the session's distributor. + IAsyncEnumerable ReadAlarmEventsAsync( + string sessionId, + CancellationToken cancellationToken); + /// Closes a session and terminates its worker process. /// Identifier of the session to close. /// Token to cancel the asynchronous operation. diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs index a848b99..43ee931 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManager.cs @@ -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); } + /// + public async IAsyncEnumerable 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; + } + } + /// public async Task 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( diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManagerErrorCode.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManagerErrorCode.cs index b1b4235..1175207 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManagerErrorCode.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionManagerErrorCode.cs @@ -10,4 +10,11 @@ public enum SessionManagerErrorCode SessionLimitExceeded, OpenFailed, CloseFailed, + + /// + /// The caller is not permitted to perform the operation on this session — for example, + /// attaching a StreamEvents stream to a session opened by a different API key. + /// Maps to gRPC PermissionDenied. + /// + PermissionDenied, } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SparseArrayExpander.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SparseArrayExpander.cs index c6d7c37..c6006db 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SparseArrayExpander.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SparseArrayExpander.cs @@ -33,13 +33,20 @@ internal static class SparseArrayExpander /// a sparse array this is a no-op, so callers may invoke it unconditionally. /// /// The value to expand in place. + /// + /// The maximum total_length the sparse array may declare before the write is + /// rejected, enforced before the full array is allocated (see + /// MxGateway:Events:MaxSparseArrayLength). Defaults to + /// so the backstop is the only bound in test/unit-construction + /// paths that do not thread the configured cap. + /// /// /// 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 , 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. /// - 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( diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index c2d3947..64b4e42 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -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( 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 } /// - public async IAsyncEnumerable ReadEventsAsync( + public IAsyncEnumerable 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 ReadEventsCoreAsync( [EnumeratorCancellation] CancellationToken cancellationToken) { await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs index f6b869c..e927439 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmFailoverEndToEndTests.cs @@ -470,6 +470,20 @@ public sealed class AlarmFailoverEndToEndTests } } + /// + public async IAsyncEnumerable 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; + } + } + } + /// public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session) { diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs index b3a2ba2..63944d5 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/GatewayAlarmMonitorProviderModeTests.cs @@ -783,6 +783,20 @@ public sealed class GatewayAlarmMonitorProviderModeTests } } + /// + public async IAsyncEnumerable 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; + } + } + } + /// public bool TryGetSession(string sessionId, [MaybeNullWhen(false)] out GatewaySession session) { diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs index a76b85b..e8646eb 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs @@ -520,4 +520,32 @@ public sealed class GatewayOptionsValidatorTests ValidateOptionsResult result = new GatewayOptionsValidator().Validate(null, options); Assert.True(result.Succeeded); } + + /// Verifies a below one fails validation. + /// Sparse-array cap under test. + [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")); + } + + /// Verifies a positive within range passes validation. + [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); + } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs index 126edf1..08d7657 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardSessionAdminServiceTests.cs @@ -281,6 +281,14 @@ public sealed class DashboardSessionAdminServiceTests throw new NotSupportedException(); } + /// + public IAsyncEnumerable ReadAlarmEventsAsync( + string sessionId, + CancellationToken cancellationToken) + { + throw new NotSupportedException(); + } + /// public Task CloseSessionAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs index 01e9b81..e4f7664 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/EventStreamServiceTests.cs @@ -38,6 +38,58 @@ public sealed class EventStreamServiceTests Assert.Equal(1, metrics.GetSnapshot().StreamDisconnects); } + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [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 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()); + } + + /// + /// TST-02 (owner-scoped attach, security control): a caller whose API key differs from + /// the key that opened the session is rejected with a + /// fault before any events are streamed — closing the reconnect/fan-out trust-boundary hole. + /// + /// A task that represents the asynchronous operation. + [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(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); + } + /// Verifies that a second event subscriber is rejected when one is already active. /// A task that represents the asynchronous operation. [Fact] @@ -48,13 +100,13 @@ public sealed class EventStreamServiceTests EventStreamService service = CreateService(new FakeSessionManager(session)); using CancellationTokenSource firstSubscriberCancellation = new(); await using IAsyncEnumerator firstSubscriber = service - .StreamEventsAsync(CreateRequest(session.SessionId), firstSubscriberCancellation.Token) + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, firstSubscriberCancellation.Token) .GetAsyncEnumerator(firstSubscriberCancellation.Token); Task firstMoveTask = firstSubscriber.MoveNextAsync().AsTask(); await WaitUntilAsync(() => session.ActiveEventSubscriberCount == 1); await using IAsyncEnumerator secondSubscriber = service - .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None) + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None) .GetAsyncEnumerator(); SessionManagerException exception = await Assert.ThrowsAsync( @@ -78,7 +130,7 @@ public sealed class EventStreamServiceTests EventStreamService service = CreateService(new FakeSessionManager(session)); using CancellationTokenSource cancellationTokenSource = new(); await using IAsyncEnumerator subscriber = service - .StreamEventsAsync(CreateRequest(session.SessionId), cancellationTokenSource.Token) + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, cancellationTokenSource.Token) .GetAsyncEnumerator(cancellationTokenSource.Token); Task 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 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 firstSubscriber = service - .StreamEventsAsync(CreateRequest(firstSession.SessionId), CancellationToken.None) + .StreamEventsAsync(CreateRequest(firstSession.SessionId), callerKeyId: null, CancellationToken.None) .GetAsyncEnumerator(); await using IAsyncEnumerator 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 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 subscriber = service - .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None) + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None) .GetAsyncEnumerator(); SessionManagerException exception = await Assert.ThrowsAsync( @@ -298,7 +350,7 @@ public sealed class EventStreamServiceTests using GatewayMetrics metrics = new(); EventStreamService service = CreateService(new FakeSessionManager(session), metrics); await using IAsyncEnumerator subscriber = service - .StreamEventsAsync(CreateRequest(session.SessionId), CancellationToken.None) + .StreamEventsAsync(CreateRequest(session.SessionId), callerKeyId: null, CancellationToken.None) .GetAsyncEnumerator(); WorkerClientException exception = await Assert.ThrowsAsync( @@ -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 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 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 resume = service - .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), CancellationToken.None) + .StreamEventsAsync(CreateRequest(session.SessionId, afterWorkerSequence: 2), callerKeyId: null, CancellationToken.None) .GetAsyncEnumerator(); List 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 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 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 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); } + /// + public IAsyncEnumerable ReadAlarmEventsAsync( + string sessionId, + CancellationToken cancellationToken) + { + throw new NotSupportedException(); + } + /// public Task CloseSessionAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs index 24df533..71b0ae2 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceConstraintTests.cs @@ -948,6 +948,14 @@ public sealed class MxAccessGatewayServiceConstraintTests } } + /// + public IAsyncEnumerable ReadAlarmEventsAsync( + string sessionId, + CancellationToken cancellationToken) + { + throw new NotSupportedException(); + } + /// public Task CloseSessionAsync( string sessionId, @@ -994,6 +1002,7 @@ public sealed class MxAccessGatewayServiceConstraintTests /// public async IAsyncEnumerable StreamEventsAsync( StreamEventsRequest request, + string? callerKeyId, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken) { foreach (WorkerEvent ev in sessionManager.Events) diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs index de76a20..a26eeb7 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs @@ -616,6 +616,14 @@ public sealed class MxAccessGatewayServiceTests } } + /// + public IAsyncEnumerable ReadAlarmEventsAsync( + string sessionId, + CancellationToken cancellationToken) + { + throw new NotSupportedException(); + } + /// public Task CloseSessionAsync( string sessionId, @@ -653,6 +661,7 @@ public sealed class MxAccessGatewayServiceTests /// public async IAsyncEnumerable StreamEventsAsync( StreamEventsRequest request, + string? callerKeyId, [EnumeratorCancellation] CancellationToken cancellationToken) { sessionManager.RecordReadEventsSessionId(request.SessionId); diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs index b52292e..b2df718 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/GatewaySessionDashboardMirrorTests.cs @@ -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()); } + /// + /// GWC-01 regression: with the internal dashboard mirror active, a second internal + /// subscriber (the alarm monitor's feed, attached via + /// ) receives EVERY event — + /// including the alarm Acknowledge 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. + /// + /// A task that represents the asynchronous operation. + [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 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); + } + /// /// 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 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); + /// + public IAsyncEnumerable ReadAlarmEventsAsync( + string sessionId, + CancellationToken cancellationToken) => throw new NotSupportedException(); + /// public Task CloseSessionAsync( string sessionId, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs index 60252fa..2a0a1af 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionManagerTests.cs @@ -975,6 +975,68 @@ public sealed class SessionManagerTests Assert.Equal(0, workerClient.ShutdownCount); } + /// + /// A faulted session is reaped by the lease sweep (default FaultedGraceSeconds=0) + /// 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. + /// + /// A task that represents the asynchronous operation. + [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); + } + + /// + /// With a positive FaultedGraceSeconds, a faulted session stays observable until + /// the grace window elapses, then the next sweep reaps it. + /// + /// A task that represents the asynchronous operation. + [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); + } + /// Verifies that shutdown closes all registered sessions. /// A task that represents the asynchronous operation. [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 { diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SparseArrayExpanderTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SparseArrayExpanderTests.cs index dd5449e..2f15957 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SparseArrayExpanderTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SparseArrayExpanderTests.cs @@ -222,4 +222,35 @@ public sealed class SparseArrayExpanderTests RpcException ex = Assert.Throws(() => SparseArrayExpander.Expand(value)); Assert.Equal(StatusCode.InvalidArgument, ex.StatusCode); } + + /// Verifies that a total length above the configured cap throws before the full array is allocated. + [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(() => 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); + } + + /// Verifies that a total length at or below the configured cap still expands normally. + [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); + } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs index cbf1ef3..53411bb 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -122,6 +122,27 @@ public sealed class WorkerClientTests Assert.Equal(MxEventFamily.OperationComplete, events.Current.Event.Family); } + /// + /// The worker event channel is single-reader: a second + /// enumerator must throw rather than silently split events between two consumers (GWC-01). + /// + /// A task that represents the asynchronous operation. + [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 first = client.ReadEventsAsync(cancellationTokenSource.Token); + await using IAsyncEnumerator firstEnumerator = first.GetAsyncEnumerator(cancellationTokenSource.Token); + + // A second call must fail loudly at call time rather than racing the first for events. + Assert.Throws(() => client.ReadEventsAsync(cancellationTokenSource.Token)); + } + /// Verifies that the read loop faults the client when the event queue overflows. /// A task that represents the asynchronous operation. [Fact] diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs index f1dc883..0e66945 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Security/Authorization/GatewayGrpcAuthorizationInterceptorTests.cs @@ -477,6 +477,14 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests return AsyncEnumerable.Empty(); } + /// + public IAsyncEnumerable ReadAlarmEventsAsync( + string sessionId, + CancellationToken cancellationToken) + { + return AsyncEnumerable.Empty(); + } + /// public Task CloseSessionAsync( string sessionId, @@ -515,6 +523,7 @@ public sealed class GatewayGrpcAuthorizationInterceptorTests /// public async IAsyncEnumerable StreamEventsAsync( StreamEventsRequest request, + string? callerKeyId, [EnumeratorCancellation] CancellationToken cancellationToken) { await Task.CompletedTask;